api: add x-continuum-heads to writes

2 files changed, 45 insertions, 26 deletions

diff --git a/src/api/store.rs b/src/api/store.rs
index 7039008..1478d14 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -203,9 +203,22 @@ async fn shallow(
 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`).
+/// hex), on request (`?heads=true`); and on every write, after it: a read with these
+/// (`?at=`) includes the write.
 const HEADS_HEADER: &str = "x-continuum-heads";
 
+fn with_heads(mut response: Response, heads: &[ChangeHash]) -> Response {
+    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);
+    }
+    response
+}
+
 fn mark_source(
     response: Result<Response, ApiError>,
     source: ReadSource,
@@ -213,14 +226,7 @@ fn mark_source(
 ) -> Result<Response, ApiError> {
     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);
-        }
+        response = with_heads(response, &heads);
     }
     if let ReadSource::Stale(epoch) = source
         && let Ok(value) = HeaderValue::from_str(&epoch.to_string())
@@ -250,14 +256,17 @@ pub(crate) async fn put_store_entry(
     let value = body_value(&headers, &body)?;
 
     let written = value.clone();
-    state
+    let ((), heads) = state
         .mutate_doc_for(Some(user.0.clone()), move |doc| {
             store::write_path(doc, &path, &written)
         })
         .await
         .map_err(|e| store_status(e, "Failed to write store value"))?;
 
-    response_for_value(&value, headers.get(ACCEPT))
+    Ok(with_heads(
+        response_for_value(&value, headers.get(ACCEPT))?,
+        &heads,
+    ))
 }
 
 fn etag_for(bytes: &[u8]) -> String {
@@ -386,13 +395,16 @@ pub(crate) async fn post_store_entry(
         return Ok(redirect);
     }
     if let Some(by) = query.increment {
-        let value = state
+        let (value, heads) = state
             .mutate_doc_for(Some(user.0.clone()), move |doc| {
                 store::increment_path(doc, &path, by)
             })
             .await
             .map_err(|e| store_status(e, "Failed to increment a counter"))?;
-        return response_for_value(&JsonValue::from(value), headers.get(ACCEPT));
+        return Ok(with_heads(
+            response_for_value(&JsonValue::from(value), headers.get(ACCEPT))?,
+            &heads,
+        ));
     }
     let index = path.last().and_then(|leaf| store::list_index(leaf));
     let value = body_value(&headers, &body)?;
@@ -402,11 +414,14 @@ pub(crate) async fn post_store_entry(
         })
         .await;
     match inserted {
-        Ok(()) => Ok((
-            StatusCode::CREATED,
-            axum::Json(serde_json::json!({ "index": index })),
-        )
-            .into_response()),
+        Ok(((), heads)) => Ok(with_heads(
+            (
+                StatusCode::CREATED,
+                axum::Json(serde_json::json!({ "index": index })),
+            )
+                .into_response(),
+            &heads,
+        )),
         Err(error) => Err(store_status(error, "Failed to insert into a list")),
     }
 }
@@ -437,14 +452,14 @@ pub(crate) async fn delete_store_entry(
     }
 
     let prune = query.prune;
-    state
+    let ((), heads) = state
         .mutate_doc_for(Some(user.0.clone()), move |doc| {
             store::delete_path(doc, &path, prune)
         })
         .await
         .map_err(|e| store_status(e, "Failed to delete store value"))?;
 
-    Ok(StatusCode::NO_CONTENT.into_response())
+    Ok(with_heads(StatusCode::NO_CONTENT.into_response(), &heads))
 }
 
 /// A path with negative list indexes (`-1` the last item; for an insert, the end): a
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 85955fe..aebc59a 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -418,16 +418,17 @@ impl AppState {
         T: Send + 'static,
         E: From<StoreError> + Send + 'static,
     {
-        self.mutate_doc_for(None, f).await
+        self.mutate_doc_for(None, f).await.map(|(out, _)| out)
     }
 
     /// [`mutate_doc`](Self::mutate_doc) for `user`, who wrote through the API: the
-    /// change names them (`replica::Writer`).
+    /// change names them (`replica::Writer`). With the heads of the document written to
+    /// after the write (sorted): what a read must have to include it.
     pub(crate) async fn mutate_doc_for<T, E>(
         &self,
         user: Option<String>,
         f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
-    ) -> Result<T, E>
+    ) -> Result<(T, Vec<crate::document::ChangeHash>), E>
     where
         T: Send + 'static,
         E: From<StoreError> + Send + 'static,
@@ -451,12 +452,15 @@ impl AppState {
                         node: state.local_id(),
                         user: user.as_deref(),
                     };
-                    return crate::document::replica::transact(
+                    let out = crate::document::replica::transact(
                         &mut doc,
                         state.clock.now_millis(),
                         &writer,
                         f,
-                    );
+                    )?;
+                    let mut heads = doc.get_heads();
+                    heads.sort_unstable();
+                    return Ok((out, heads));
                 }
                 // One API write = one change; on error, its partial writes (e.g. the old
                 // value deleted, the new one not written) are dropped.
@@ -466,7 +470,7 @@ impl AppState {
                 };
                 let result = replica.transact(state.clock.now_millis(), &writer, f);
                 replica.refresh_heads();
-                result
+                result.map(|out| (out, replica.heads().to_vec()))
             }; // lock released before notifying
 
             if result.is_ok() {