diff options
| -rw-r--r-- | docs/api.md | 3 | ||||
| -rw-r--r-- | docs/example.toml | 5 | ||||
| -rw-r--r-- | docs/openapi.yaml | 3 | ||||
| -rw-r--r-- | docs/operations.md | 7 | ||||
| -rw-r--r-- | src/api/store.rs | 22 | ||||
| -rw-r--r-- | src/config/mod.rs | 15 | ||||
| -rw-r--r-- | src/main.rs | 1 | ||||
| -rw-r--r-- | src/node.rs | 3 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 1 | ||||
| -rw-r--r-- | src/state/mod.rs | 3 |
10 files changed, 58 insertions, 5 deletions
diff --git a/docs/api.md b/docs/api.md index e813845..cdc722e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -194,6 +194,9 @@ node or by sync: {"key":"settings/theme","deleted":true} ``` +While nothing changes, a blank line every 30 s (`--stream-heartbeat`) keeps proxies in +front from taking the stream for dead; skip blank lines. + An event carries the value as a read would return it then. Changes in quick succession may come as one event with the latest value: a stream tells the value, not every write. A change that leaves the value as it was sends nothing. The connection stays open; after diff --git a/docs/example.toml b/docs/example.toml index cef5754..00bfed3 100644 --- a/docs/example.toml +++ b/docs/example.toml @@ -68,6 +68,11 @@ htpasswd = "/etc/continuum/htpasswd" # authentication. The proxy must set or clear it on every request. # user-header = "X-Remote-User" +# How often GET /stream sends a blank line while nothing changes, so proxies in front +# keep it open: below their read timeout (nginx's proxy_read_timeout, 60 s by default). +# 0: never. +# stream-heartbeat = "30s" + # Where the reverse proxies whose user-header is believed connect from: IP addresses (or # ranges, as ADDRESS/BITS), and unix for the API's Unix sockets (bind-addr). On a host # with other users, leave out the addresses and have the proxy use a Unix socket: others diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 387a7ac..701cdf9 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -329,7 +329,8 @@ paths: description: | Newline-delimited JSON: one event with the current value, then one whenever the resolved value changes (by a local write or by sync). The connection stays - open. + open; while nothing changes, a blank line every `--stream-heartbeat` (30 s) + keeps proxies from closing it: skip blank lines. responses: '200': description: Event stream. diff --git a/docs/operations.md b/docs/operations.md index da73fce..fae3b6c 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -162,9 +162,10 @@ The names the proxy gives are the users of `--protocol-writers` and of who wrote ([`api.md`](api.md)). What streams (`/stream`, the dashboard's events) says `X-Accel-Buffering: no`, so nginx -passes it on as it comes. The dashboard's events have a keepalive every 15 s; `/stream` -sends nothing while nothing changes, so nginx ends a quiet one after its -`proxy_read_timeout` (60 s): raise that for it, or have clients reconnect. +passes it on as it comes, and says it's alive while nothing changes: the dashboard's +events every 15 s, `/stream` every 30 s (a blank line, `--stream-heartbeat`), within +nginx's `proxy_read_timeout` (60 s). With a proxy that gives up sooner, lower +`--stream-heartbeat` below its timeout. The API may also be served under a path of its own (`location /continuum/ { proxy_pass http://unix:/run/continuum/api.sock:/; }`): its redirects (`Location`), the dashboard's diff --git a/src/api/store.rs b/src/api/store.rs index eac3262..d419d37 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -861,6 +861,7 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo let stream_state = state.clone(); let stream_key = key.clone(); + let heartbeat = state.stream_heartbeat; let output = async_stream::stream! { let (mut last_value, mut conflicted) = initial; @@ -872,9 +873,27 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo return; } } + // When the last line went out: a blank one follows after `heartbeat` without + // another, so proxies in front don't take the stream for dead. + let mut sent = tokio::time::Instant::now(); loop { - let relevant = match changes.recv().await { + let beat = async { + match sent.checked_add(heartbeat).filter(|_| !heartbeat.is_zero()) { + Some(at) => tokio::time::sleep_until(at).await, + None => std::future::pending::<()>().await, + } + }; + let received = tokio::select! { + received = changes.recv() => Some(received), + () = beat => None, + }; + let Some(received) = received else { + sent = tokio::time::Instant::now(); + yield Ok(Bytes::from_static(b"\n")); + continue; + }; + let relevant = match received { // Conflicted paths can change without a patch on their path; re-read always. Ok(set) => conflicted || set.paths.iter().any(|changed| paths_related(changed, &path)), Err(broadcast::error::RecvError::Lagged(skipped)) => { @@ -910,6 +929,7 @@ async fn stream(state: AppState, key: String, path: Vec<String>) -> Result<Respo return; } } + sent = tokio::time::Instant::now(); last_value = next_value; } }; diff --git a/src/config/mod.rs b/src/config/mod.rs index 628f0ff..a6f70aa 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -250,6 +250,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "Where people reach this node's dashboard (http:// or https://, e.g. behind a reverse proxy). Published in the swarm's member list: the other nodes' dashboards open it on a double-click on this node. A build without the dashboard feature warns and publishes none (removing one published before).", ), option( + "stream-heartbeat", + Kind::Duration, + "DURATION", + Some("30s"), + "How often GET /stream sends a blank line while nothing changes, so proxies in front keep it open: below their read timeout (nginx's proxy_read_timeout, 60 s by default). 0: never.", + ), + option( "keepalive", Kind::Duration, "DURATION", @@ -551,6 +558,11 @@ pub(crate) struct Settings { #[serde(default = "default_keepalive", with = "crate::units::duration_option")] pub(crate) keepalive: Duration, #[serde( + default = "default_stream_heartbeat", + with = "crate::units::duration_option" + )] + pub(crate) stream_heartbeat: Duration, + #[serde( default, deserialize_with = "crate::units::optional_duration", serialize_with = "crate::units::serialize_optional_duration" @@ -664,6 +676,9 @@ fn default_document_rewrite() -> Duration { fn default_keepalive() -> Duration { Duration::from_secs(60) } +fn default_stream_heartbeat() -> Duration { + Duration::from_secs(30) +} fn default_neighbours() -> usize { 12 } diff --git a/src/main.rs b/src/main.rs index 49971fb..c3d7458 100644 --- a/src/main.rs +++ b/src/main.rs @@ -222,6 +222,7 @@ async fn main() -> Result<()> { dashboard_url: args.dashboard_url.clone(), electorate_window: args.electorate_window, clock_skew_warn: args.clock_skew_warn, + stream_heartbeat: args.stream_heartbeat, superseded_writes: args.superseded_writes, // At most 4MiB (`Settings::check`). blob_chunk_bytes: args.blob_chunk as u32, diff --git a/src/node.rs b/src/node.rs index c1b0aea..8b18cd4 100644 --- a/src/node.rs +++ b/src/node.rs @@ -49,6 +49,8 @@ pub(crate) struct NodeConfig { pub(crate) dashboard_url: Option<String>, pub(crate) electorate_window: Duration, pub(crate) clock_skew_warn: Duration, + /// `--stream-heartbeat`. + pub(crate) stream_heartbeat: Duration, /// `--superseded-writes`. pub(crate) superseded_writes: crate::remake::Superseded, pub(crate) blob_chunk_bytes: u32, @@ -121,6 +123,7 @@ impl Node { dashboard_url: config.dashboard_url, electorate_window: config.electorate_window, clock_skew_warn: config.clock_skew_warn, + stream_heartbeat: config.stream_heartbeat, superseded_writes: config.superseded_writes, clock_ahead_ms: Default::default(), clock_skew: Default::default(), diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index 808f44f..267768f 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -189,6 +189,7 @@ impl Sim { dashboard_url: None, electorate_window: Duration::from_secs(30 * 24 * 3600), clock_skew_warn: Duration::from_secs(2), + stream_heartbeat: Duration::from_secs(30), superseded_writes: Default::default(), blob_chunk_bytes: 64 * 1024, swarm_key: None, diff --git a/src/state/mod.rs b/src/state/mod.rs index 36b1132..fef6448 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -115,6 +115,9 @@ pub(crate) struct AppState { /// This node's clock is off when it differs from most of its peers' by more than /// this (`--clock-skew-warn`, `swarm::skew`); 0: never. pub(crate) clock_skew_warn: Duration, + /// How often `GET /stream` says it's alive while nothing changes + /// (`--stream-heartbeat`); 0: never. + pub(crate) stream_heartbeat: Duration, /// What a replay does with writes a newer one superseded (`--superseded-writes`). pub(crate) superseded_writes: crate::remake::Superseded, /// How far this node's clock is ahead of its peers' (their median, `swarm::skew`), as |