serve stale reads during rejoinHEADmaster

1 files changed, 28 insertions, 12 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(