serve stale reads during rejoinHEADmaster

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)