diff options
Diffstat (limited to 'src')
| -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 |
6 files changed, 44 insertions, 1 deletions
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 |