1 files changed, 54 insertions, 14 deletions

diff --git a/src/api/store.rs b/src/api/store.rs
index de5ae9b..409a530 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -2,6 +2,7 @@
 
 use std::io::Error as IoError;
 
+use automerge::ChangeHash;
 use axum::{
     body::{Body, Bytes},
     extract::{Path, Query, State},
@@ -17,14 +18,18 @@ 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::store::{self, StoreError};
-use crate::sync::paths_related;
+use crate::document::store::{self, StoreError};
+use crate::state::{AppState, ReadSource};
+use crate::swarm::sync::paths_related;
 
 #[derive(Debug, Default, Deserialize)]
 pub(crate) struct ReadQuery {
     #[serde(default)]
     conflicts: bool,
+    /// Return the document's heads at the time of the read (`x-continuum-heads`), e.g.
+    /// to base a replacing remake on exactly what was read.
+    #[serde(default)]
+    heads: bool,
 }
 
 #[derive(Debug, Deserialize)]
@@ -47,6 +52,7 @@ fn conflicts_response(
     tag = "Store",
     params(
         ("conflicts" = Option<bool>, Query, description = "Wrap the value and list every conflict resolved on the way"),
+        ("heads" = Option<bool>, Query, description = "Return the document heads at the time of the read in x-continuum-heads"),
     ),
     responses(
         (status = 200, description = "Returns the complete store as JSON or text, depending on Accept"),
@@ -62,8 +68,10 @@ pub(crate) async fn get_store_root(
     let accept = headers.get(ACCEPT).cloned();
     let want_conflicts = query.conflicts;
 
-    state
-        .render(
+    let (response, source, heads) = state
+        .render_read(
+            true,
+            query.heads,
             move |doc| {
                 if want_conflicts {
                     store::read_path_with_conflicts(doc, &Vec::new())
@@ -82,7 +90,8 @@ pub(crate) async fn get_store_root(
                 }
             },
         )
-        .await
+        .await?;
+    mark_source(response, source, heads)
 }
 
 #[utoipa::path(
@@ -92,13 +101,14 @@ pub(crate) async fn get_store_root(
     params(
         ("key" = String, Path, description = "Slash-separated store path, for example settings/theme"),
         ("conflicts" = Option<bool>, Query, description = "Wrap the value and list every conflict resolved on the way"),
+        ("heads" = Option<bool>, Query, description = "Return the document heads at the time of the read in x-continuum-heads"),
     ),
     responses(
         (status = 200, description = "Returns the value as JSON or text, depending on Accept"),
         (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 +117,15 @@ 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, heads) = state
+        .render_read(
+            false,
+            query.heads,
             move |doc| {
                 if want_conflicts {
                     store::read_path_with_conflicts(doc, &path)
@@ -136,7 +144,39 @@ pub(crate) async fn get_store_entry(
                 }
             },
         )
-        .await
+        .await?;
+    mark_source(response, source, heads)
+}
+
+/// Header on reads served from the pre-rejoin document; its value is that epoch.
+const STALE_HEADER: &str = "x-continuum-stale";
+
+/// Header with the document's heads at the time of the read (sorted, comma-separated
+/// hex), on request (`?heads=true`).
+const HEADS_HEADER: &str = "x-continuum-heads";
+
+fn mark_source(
+    response: Result<Response, StatusCode>,
+    source: ReadSource,
+    heads: Option<Vec<ChangeHash>>,
+) -> Result<Response, StatusCode> {
+    let mut response = response?;
+    if let Some(heads) = heads {
+        let joined = heads
+            .iter()
+            .map(ToString::to_string)
+            .collect::<Vec<_>>()
+            .join(",");
+        if let Ok(value) = HeaderValue::from_str(&joined) {
+            response.headers_mut().insert(HEADS_HEADER, value);
+        }
+    }
+    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(