diff options
| -rw-r--r-- | src/api/store.rs | 32 | ||||
| -rw-r--r-- | src/state.rs | 44 |
2 files changed, 67 insertions, 9 deletions
diff --git a/src/api/store.rs b/src/api/store.rs index 8ea0202..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}, @@ -25,6 +26,10 @@ use crate::swarm::sync::paths_related; 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,9 +68,10 @@ pub(crate) async fn get_store_root( let accept = headers.get(ACCEPT).cloned(); let want_conflicts = query.conflicts; - let (response, source) = state + let (response, source, heads) = state .render_read( true, + query.heads, move |doc| { if want_conflicts { store::read_path_with_conflicts(doc, &Vec::new()) @@ -84,7 +91,7 @@ pub(crate) async fn get_store_root( }, ) .await?; - mark_source(response, source) + mark_source(response, source, heads) } #[utoipa::path( @@ -94,6 +101,7 @@ 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"), @@ -114,9 +122,10 @@ pub(crate) async fn get_store_entry( let accept = headers.get(ACCEPT).cloned(); let want_conflicts = query.conflicts; - let (response, source) = state + let (response, source, heads) = state .render_read( false, + query.heads, move |doc| { if want_conflicts { store::read_path_with_conflicts(doc, &path) @@ -136,17 +145,32 @@ pub(crate) async fn get_store_entry( }, ) .await?; - mark_source(response, source) + 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); diff --git a/src/state.rs b/src/state.rs index 20285a0..42393b6 100644 --- a/src/state.rs +++ b/src/state.rs @@ -7,7 +7,7 @@ use std::sync::Arc; use std::time::Duration; use anyhow::{bail, Result}; -use automerge::{sync, AutoCommit}; +use automerge::{sync, AutoCommit, ChangeHash}; use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock}; use tokio::time::Instant; use tracing::{debug, error, warn}; @@ -573,18 +573,24 @@ 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`. + /// + /// With `with_heads`, a live read also returns the document's heads at the moment of + /// the read (for clients that base later requests on what they read). That needs the + /// write lock, so it is opt-in. Stale reads have no heads. pub(crate) async fn render_read<T, R>( &self, unready_live: bool, + with_heads: bool, read: impl FnOnce(&AutoCommit) -> T + Send + 'static, finish: impl FnOnce(T) -> R + Send + 'static, - ) -> Result<(R, ReadSource), axum::http::StatusCode> + ) -> Result<(R, ReadSource, Option<Vec<ChangeHash>>), 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 (rendered, heads) = self.render_live(with_heads, read, finish).await; + return Ok((rendered, ReadSource::Live, heads)); } let stale = self @@ -595,7 +601,8 @@ impl AppState { .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)); + let (rendered, heads) = self.render_live(with_heads, read, finish).await; + return Ok((rendered, ReadSource::Live, heads)); } return Err(axum::http::StatusCode::SERVICE_UNAVAILABLE); }; @@ -603,7 +610,34 @@ impl AppState { let rendered = tokio::task::spawn_blocking(move || finish(read(&doc))) .await .expect("document read task panicked"); - Ok((rendered, ReadSource::Stale(epoch))) + Ok((rendered, ReadSource::Stale(epoch), None)) + } + + async fn render_live<T, R>( + &self, + with_heads: bool, + read: impl FnOnce(&AutoCommit) -> T + Send + 'static, + finish: impl FnOnce(T) -> R + Send + 'static, + ) -> (R, Option<Vec<ChangeHash>>) + where + T: Send + 'static, + R: Send + 'static, + { + if !with_heads { + return (self.render(read, finish).await, None); + } + // AutoCommit::get_heads takes &mut self (it closes a pending transaction). + let (data, mut heads) = self + .write_doc(move |doc| { + let heads = doc.get_heads(); + (read(doc), heads) + }) + .await; + heads.sort(); + let rendered = tokio::task::spawn_blocking(move || finish(data)) + .await + .expect("document read task panicked"); + (rendered, Some(heads)) } fn set_stale(&self, view: Option<StaleView>) { |