diff options
| -rw-r--r-- | src/api/store.rs | 40 | ||||
| -rw-r--r-- | src/compaction/catchup.rs | 18 | ||||
| -rw-r--r-- | src/compaction/mod.rs | 4 | ||||
| -rw-r--r-- | src/main.rs | 1 | ||||
| -rw-r--r-- | src/state.rs | 69 |
5 files changed, 118 insertions, 14 deletions
diff --git a/src/api/store.rs b/src/api/store.rs index de5ae9b..e991005 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -17,7 +17,7 @@ use tokio::sync::broadcast; use tracing::{debug, error, instrument}; use super::{parse_key_path, response_for_value, store_status}; -use crate::state::AppState; +use crate::state::{AppState, ReadSource}; use crate::store::{self, StoreError}; use crate::sync::paths_related; @@ -62,8 +62,9 @@ pub(crate) async fn get_store_root( let accept = headers.get(ACCEPT).cloned(); let want_conflicts = query.conflicts; - state - .render( + let (response, source) = state + .render_read( + true, move |doc| { if want_conflicts { store::read_path_with_conflicts(doc, &Vec::new()) @@ -82,7 +83,8 @@ pub(crate) async fn get_store_root( } }, ) - .await + .await?; + mark_source(response, source) } #[utoipa::path( @@ -98,7 +100,7 @@ pub(crate) async fn get_store_root( (status = 400, description = "Invalid store path"), (status = 404, description = "Value does not exist"), (status = 500, description = "Failed to read or serialize the value"), - (status = 503, description = "Node is not ready") + (status = 503, description = "Node is not ready (and not rejoining with a stale view)") ) )] pub(crate) async fn get_store_entry( @@ -107,17 +109,14 @@ pub(crate) async fn get_store_entry( headers: HeaderMap, Query(query): Query<ReadQuery>, ) -> Result<Response, StatusCode> { - if !state.is_ready().await { - return Err(StatusCode::SERVICE_UNAVAILABLE); - } - let path = parse_key_path(&key)?; let accept = headers.get(ACCEPT).cloned(); let want_conflicts = query.conflicts; - state - .render( + let (response, source) = state + .render_read( + false, move |doc| { if want_conflicts { store::read_path_with_conflicts(doc, &path) @@ -136,7 +135,24 @@ pub(crate) async fn get_store_entry( } }, ) - .await + .await?; + mark_source(response, source) +} + +/// Header on reads served from the pre-rejoin document; its value is that epoch. +const STALE_HEADER: &str = "x-continuum-stale"; + +fn mark_source( + response: Result<Response, StatusCode>, + source: ReadSource, +) -> Result<Response, StatusCode> { + let mut response = response?; + if let ReadSource::Stale(epoch) = source { + if let Ok(value) = HeaderValue::from_str(&epoch.to_string()) { + response.headers_mut().insert(STALE_HEADER, value); + } + } + Ok(response) } #[utoipa::path( diff --git a/src/compaction/catchup.rs b/src/compaction/catchup.rs index c71bb57..d8bac74 100644 --- a/src/compaction/catchup.rs +++ b/src/compaction/catchup.rs @@ -4,7 +4,8 @@ //! returning, or one whose own build failed) can't follow through ordinary sync. Once //! an active peer advertises a newer generation, the node //! -//! 1. stops serving requests and keeps its document aside; +//! 1. stops being ready and keeps its document aside; reads are answered from a copy +//! of it, marked stale (`X-Continuum-Stale`), writes get `503`; //! 2. switches to an empty document labelled with the newer epoch: an empty history //! conflicts with nothing, so ordinary sync fills it from peers on that epoch, //! exactly like a new node joining; @@ -16,7 +17,8 @@ //! * within that epoch's replay window they are replayed (state per path, as in a //! switch), otherwise they are quarantined for an operator //! (`/admin/quarantine`), as are changes that can't be told apart; -//! 5. archives the old document and serves requests again. +//! 5. archives the old document and serves requests again. A node that was never +//! ready (a new node) is made ready by bootstrap, which waits for the rejoin. //! //! If no peer completes a sync in time, or the check fails, the node restores its old //! document and epoch and tries again later. @@ -117,6 +119,12 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) { error: None, })); + // Keep answering reads, from a copy of the current document marked stale, before + // the node stops being ready: no window in which reads get 503. + if host.is_node_ready().await { + let copy = host.read_doc(|doc| doc.clone()).await; + host.set_stale_view(Some((local, copy))); + } let was_ready = host.suspend().await; let old = match host .switch_epoch(target, |_live| Ok::<_, Infallible>(AutoCommit::new())) @@ -214,6 +222,7 @@ async fn finish<H: Host>(host: &H) { if was_ready { host.resume().await; } + host.set_stale_view(None); return; } }; @@ -277,6 +286,8 @@ async fn finish<H: Host>(host: &H) { if was_ready { host.resume().await; } + // After resuming, so reads move from the stale view to the live document directly. + host.set_stale_view(None); info!( from = %previous, @@ -386,9 +397,12 @@ async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) { { match never {} } + // A node that was never ready must sync again before bootstrap may finish. + host.reset_initial_sync(); if was_ready { host.resume().await; } + host.set_stale_view(None); let attempts = host .compaction() diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs index 4685b86..6de22c5 100644 --- a/src/compaction/mod.rs +++ b/src/compaction/mod.rs @@ -69,6 +69,10 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// Serves requests again after [`Host::suspend`]. fn resume(&self) -> impl Future<Output = ()> + Send; + /// Serves reads from this document (read-only, marked stale) while the node is not + /// ready; `None` ends it. + fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>); + /// Whether the document caught up with a peer since the last reset. fn initial_sync_done(&self) -> bool; diff --git a/src/main.rs b/src/main.rs index 44f9dd4..c98596b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -161,6 +161,7 @@ async fn main() -> Result<()> { change_events, timing: Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)), compaction: Arc::new(compaction::Compaction::default()), + stale: Default::default(), }; trace!("Application state initialized"); diff --git a/src/state.rs b/src/state.rs index a8b573b..e7c1767 100644 --- a/src/state.rs +++ b/src/state.rs @@ -300,6 +300,10 @@ pub(crate) struct AppState { /// Per-node compaction state (prebuilt snapshots, switch status, archive). pub(crate) compaction: Arc<Compaction>, + + /// While rejoining a newer epoch, the previous document, served read-only (and + /// marked stale) instead of answering 503. + pub(crate) stale: Arc<std::sync::RwLock<Option<StaleView>>>, } impl AppState { @@ -552,6 +556,64 @@ impl AppState { } } +/// A read-only view of the document the node served before it started rejoining. +pub(crate) struct StaleView { + pub(crate) epoch: Epoch, + pub(crate) doc: Arc<AutoCommit>, +} + +/// Where a read was served from. +pub(crate) enum ReadSource { + Live, + /// The pre-rejoin document of this epoch. + Stale(Epoch), +} + +impl AppState { + /// Renders a read for the store API: from the live document when the node is ready, + /// from the stale view while it rejoins, and otherwise `503`, or the live document + /// anyway with `unready_live`. + pub(crate) async fn render_read<T, R>( + &self, + unready_live: bool, + read: impl FnOnce(&AutoCommit) -> T + Send + 'static, + finish: impl FnOnce(T) -> R + Send + 'static, + ) -> Result<(R, ReadSource), axum::http::StatusCode> + where + T: Send + 'static, + R: Send + 'static, + { + if self.is_ready().await { + return Ok((self.render(read, finish).await, ReadSource::Live)); + } + + let stale = self + .stale + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .as_ref() + .map(|view| (view.epoch, Arc::clone(&view.doc))); + let Some((epoch, doc)) = stale else { + if unready_live { + return Ok((self.render(read, finish).await, ReadSource::Live)); + } + return Err(axum::http::StatusCode::SERVICE_UNAVAILABLE); + }; + + let rendered = tokio::task::spawn_blocking(move || finish(read(&doc))) + .await + .expect("document read task panicked"); + Ok((rendered, ReadSource::Stale(epoch))) + } + + fn set_stale(&self, view: Option<StaleView>) { + *self + .stale + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = view; + } +} + pub(crate) fn unix_millis() -> u64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -638,6 +700,13 @@ impl Host for AppState { } } + fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>) { + self.set_stale(view.map(|(epoch, doc)| StaleView { + epoch, + doc: Arc::new(doc), + })); + } + fn initial_sync_done(&self) -> bool { self.initial_sync_done .load(std::sync::atomic::Ordering::Acquire) |