add stream heartbeat

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