aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/api/mod.rs2
-rw-r--r--src/api/store.rs62
-rw-r--r--src/document/at.rs249
-rw-r--r--src/document/mod.rs1
-rw-r--r--src/document/store.rs3
5 files changed, 308 insertions, 9 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs
index b1f865a..8ce855e 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -204,7 +204,7 @@ pub(crate) fn store_status(error: StoreError, context: &str) -> ApiError {
StoreError::Conflict | StoreError::NotAList | StoreError::PastTheEnd { .. } => {
StatusCode::CONFLICT
}
- StoreError::NotFound => StatusCode::NOT_FOUND,
+ StoreError::NotFound | StoreError::UnknownHeads => StatusCode::NOT_FOUND,
StoreError::Rejoining => StatusCode::SERVICE_UNAVAILABLE,
other => {
error!(error = %format!("{other:#}"), "{context}");
diff --git a/src/api/store.rs b/src/api/store.rs
index 82cac30..8b09487 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -38,6 +38,43 @@ pub(crate) struct ReadQuery {
/// level at a time.
#[serde(default)]
shallow: bool,
+ /// Read as the document was at these heads (comma-separated hex, as
+ /// `x-continuum-heads` gives them), not as it is now (`document::at`).
+ at: Option<String>,
+}
+
+impl ReadQuery {
+ /// The heads `at` names, if any; `400` if it doesn't name them.
+ fn at(&self) -> Result<Option<Vec<ChangeHash>>, ApiError> {
+ let Some(at) = &self.at else {
+ return Ok(None);
+ };
+ at.split(',')
+ .map(|head| head.trim().parse::<ChangeHash>())
+ .collect::<Result<Vec<_>, _>>()
+ .map(Some)
+ .map_err(|_| ApiError(StatusCode::BAD_REQUEST, Some("at: not heads".to_owned())))
+ }
+}
+
+/// Reads `doc` with `read`, as it is or, with `at`, as it was at those heads
+/// (`UnknownHeads` if it lacks one).
+macro_rules! read_at {
+ ($doc:expr, $at:expr, |$d:ident| $read:expr) => {
+ match $at {
+ None => {
+ let $d = $doc;
+ $read
+ }
+ Some(heads) => match crate::document::at::At::new($doc, heads) {
+ Some(at) => {
+ let $d = &at;
+ $read
+ }
+ None => Err(StoreError::UnknownHeads),
+ },
+ }
+ };
}
#[derive(Debug, Deserialize)]
@@ -72,7 +109,10 @@ pub(crate) async fn get_store_entry(
RawQuery(raw_query): RawQuery,
) -> Result<Response, ApiError> {
let path = parse_key_path(&key)?;
- if let Some(redirect) = to_positive(&state, &path, false, raw_query.as_deref()).await? {
+ let at = query.at()?;
+ if let Some(redirect) =
+ to_positive(&state, &path, false, at.as_deref(), raw_query.as_deref()).await?
+ {
return Ok(redirect);
}
read(&state, path, &headers, &query).await
@@ -93,18 +133,19 @@ async fn read(
}
let accept = headers.get(ACCEPT).cloned();
let want_conflicts = query.conflicts;
+ let at = query.at()?;
let (response, source, heads) = state
.render_read(
root,
query.heads,
move |doc| {
- if want_conflicts {
+ read_at!(doc, at.as_deref(), |doc| if want_conflicts {
store::read_path_with_conflicts(doc, &path)
.map(|found| found.map(|(value, conflicts)| (value, Some(conflicts))))
} else {
store::read_path(doc, &path).map(|found| found.map(|value| (value, None)))
- }
+ })
},
move |result| {
let found = result.map_err(|e| store_status(e, "Failed to read store path"))?;
@@ -134,11 +175,12 @@ async fn shallow(
if query.conflicts {
return Err(StatusCode::BAD_REQUEST.into());
}
+ let at = query.at()?;
let (response, source, heads) = state
.render_read(
unready_live,
query.heads,
- move |doc| store::read_shallow(doc, &path),
+ move |doc| read_at!(doc, at.as_deref(), |doc| store::read_shallow(doc, &path)),
move |result| {
let value = result
.map_err(|e| store_status(e, "Failed to read store path"))?
@@ -195,7 +237,7 @@ pub(crate) async fn put_store_entry(
}
let path = parse_write_path(&state, &user, &key, "PUT")?;
- if let Some(redirect) = to_positive(&state, &path, false, raw_query.as_deref()).await? {
+ if let Some(redirect) = to_positive(&state, &path, false, None, raw_query.as_deref()).await? {
return Ok(redirect);
}
let value = body_value(&headers, &body)?;
@@ -326,7 +368,7 @@ pub(crate) async fn post_store_entry(
return Err(StatusCode::SERVICE_UNAVAILABLE.into());
}
let path = parse_write_path(&state, &user, &key, "POST")?;
- if let Some(redirect) = to_positive(&state, &path, true, raw_query.as_deref()).await? {
+ if let Some(redirect) = to_positive(&state, &path, true, None, raw_query.as_deref()).await? {
return Ok(redirect);
}
let index = path.last().and_then(|leaf| store::list_index(leaf));
@@ -367,7 +409,7 @@ pub(crate) async fn delete_store_entry(
return Err(StatusCode::BAD_REQUEST.into());
}
- if let Some(redirect) = to_positive(&state, &path, false, raw_query.as_deref()).await? {
+ if let Some(redirect) = to_positive(&state, &path, false, None, raw_query.as_deref()).await? {
return Ok(redirect);
}
@@ -389,6 +431,7 @@ async fn to_positive(
state: &AppState,
path: &[String],
insertion: bool,
+ at: Option<&[ChangeHash]>,
query: Option<&str>,
) -> Result<Option<Response>, ApiError> {
if !path.iter().any(|segment| segment.starts_with('-')) {
@@ -396,7 +439,10 @@ async fn to_positive(
}
let positive = {
let replica = state.doc.read().await;
- store::positive_path(&replica.doc, path, insertion)
+ // Counted from the end of the list as it was then, for a read at heads.
+ read_at!(&replica.doc, at, |doc| store::positive_path(
+ doc, path, insertion
+ ))
}
.map_err(|e| store_status(e, "Failed to resolve a negative index"))?;
let Some(positive) = positive else {
diff --git a/src/document/at.rs b/src/document/at.rs
new file mode 100644
index 0000000..d318f17
--- /dev/null
+++ b/src/document/at.rs
@@ -0,0 +1,249 @@
+//! The document as it was at given heads: every read of `ReadDoc` made at them (its
+//! `*_at` variant), so whatever reads a document (`store`) reads it as it was then.
+
+use std::ops::RangeBounds;
+
+use automerge::iter::{DocIter, Keys, ListRange, MapRange, Spans, Values};
+use automerge::marks::{Mark, MarkSet};
+use automerge::{
+ AutomergeError, Change, ChangeHash, Cursor, CursorPosition, MoveCursor, ObjId, ObjType,
+ Parents, Prop, ReadDoc, Stats, TextEncoding, Value, hydrate,
+};
+
+use super::Document;
+
+pub(crate) struct At<'a> {
+ doc: &'a Document,
+ heads: Vec<ChangeHash>,
+}
+
+impl<'a> At<'a> {
+ /// `doc` at `heads`; None if it lacks one of them (it hasn't synced it yet, or
+ /// never will: of another epoch, or no change at all).
+ pub(crate) fn new(doc: &'a Document, heads: &[ChangeHash]) -> Option<Self> {
+ heads
+ .iter()
+ .all(|head| doc.get_change_by_hash(head).is_some())
+ .then(|| Self {
+ doc,
+ heads: heads.to_vec(),
+ })
+ }
+
+ fn or_ours<'h>(&'h self, heads: Option<&'h [ChangeHash]>) -> Option<&'h [ChangeHash]> {
+ Some(heads.unwrap_or(&self.heads))
+ }
+}
+
+impl ReadDoc for At<'_> {
+ fn parents<O: AsRef<ObjId>>(&self, obj: O) -> Result<Parents<'_>, AutomergeError> {
+ self.doc.parents_at(obj, &self.heads)
+ }
+
+ fn parents_at<O: AsRef<ObjId>>(
+ &self,
+ obj: O,
+ heads: &[ChangeHash],
+ ) -> Result<Parents<'_>, AutomergeError> {
+ self.doc.parents_at(obj, heads)
+ }
+
+ fn keys<O: AsRef<ObjId>>(&self, obj: O) -> Keys<'_> {
+ self.doc.keys_at(obj, &self.heads)
+ }
+
+ fn keys_at<O: AsRef<ObjId>>(&self, obj: O, heads: &[ChangeHash]) -> Keys<'_> {
+ self.doc.keys_at(obj, heads)
+ }
+
+ fn iter_at<O: AsRef<ObjId>>(&self, obj: O, heads: Option<&[ChangeHash]>) -> DocIter<'_> {
+ self.doc.iter_at(obj, self.or_ours(heads))
+ }
+
+ fn map_range<'b, O: AsRef<ObjId>, R: RangeBounds<String> + 'b>(
+ &'b self,
+ obj: O,
+ range: R,
+ ) -> MapRange<'b> {
+ self.doc.map_range_at(obj, range, &self.heads)
+ }
+
+ fn map_range_at<'b, O: AsRef<ObjId>, R: RangeBounds<String> + 'b>(
+ &'b self,
+ obj: O,
+ range: R,
+ heads: &[ChangeHash],
+ ) -> MapRange<'b> {
+ self.doc.map_range_at(obj, range, heads)
+ }
+
+ fn list_range<O: AsRef<ObjId>, R: RangeBounds<usize>>(
+ &self,
+ obj: O,
+ range: R,
+ ) -> ListRange<'_> {
+ self.doc.list_range_at(obj, range, &self.heads)
+ }
+
+ fn list_range_at<O: AsRef<ObjId>, R: RangeBounds<usize>>(
+ &self,
+ obj: O,
+ range: R,
+ heads: &[ChangeHash],
+ ) -> ListRange<'_> {
+ self.doc.list_range_at(obj, range, heads)
+ }
+
+ fn values<O: AsRef<ObjId>>(&self, obj: O) -> Values<'_> {
+ self.doc.values_at(obj, &self.heads)
+ }
+
+ fn values_at<O: AsRef<ObjId>>(&self, obj: O, heads: &[ChangeHash]) -> Values<'_> {
+ self.doc.values_at(obj, heads)
+ }
+
+ fn length<O: AsRef<ObjId>>(&self, obj: O) -> usize {
+ self.doc.length_at(obj, &self.heads)
+ }
+
+ fn length_at<O: AsRef<ObjId>>(&self, obj: O, heads: &[ChangeHash]) -> usize {
+ self.doc.length_at(obj, heads)
+ }
+
+ fn object_type<O: AsRef<ObjId>>(&self, obj: O) -> Result<ObjType, AutomergeError> {
+ // An object's type is fixed when it is made.
+ self.doc.object_type(obj)
+ }
+
+ fn marks<O: AsRef<ObjId>>(&self, obj: O) -> Result<Vec<Mark>, AutomergeError> {
+ self.doc.marks_at(obj, &self.heads)
+ }
+
+ fn marks_at<O: AsRef<ObjId>>(
+ &self,
+ obj: O,
+ heads: &[ChangeHash],
+ ) -> Result<Vec<Mark>, AutomergeError> {
+ self.doc.marks_at(obj, heads)
+ }
+
+ fn get_marks<O: AsRef<ObjId>>(
+ &self,
+ obj: O,
+ index: usize,
+ heads: Option<&[ChangeHash]>,
+ ) -> Result<MarkSet, AutomergeError> {
+ self.doc.get_marks(obj, index, self.or_ours(heads))
+ }
+
+ fn text<O: AsRef<ObjId>>(&self, obj: O) -> Result<String, AutomergeError> {
+ self.doc.text_at(obj, &self.heads)
+ }
+
+ fn text_at<O: AsRef<ObjId>>(
+ &self,
+ obj: O,
+ heads: &[ChangeHash],
+ ) -> Result<String, AutomergeError> {
+ self.doc.text_at(obj, heads)
+ }
+
+ fn spans<O: AsRef<ObjId>>(&self, obj: O) -> Result<Spans<'_>, AutomergeError> {
+ self.doc.spans_at(obj, &self.heads)
+ }
+
+ fn spans_at<O: AsRef<ObjId>>(
+ &self,
+ obj: O,
+ heads: &[ChangeHash],
+ ) -> Result<Spans<'_>, AutomergeError> {
+ self.doc.spans_at(obj, heads)
+ }
+
+ fn get_cursor<O: AsRef<ObjId>, I: Into<CursorPosition>>(
+ &self,
+ obj: O,
+ position: I,
+ at: Option<&[ChangeHash]>,
+ ) -> Result<Cursor, AutomergeError> {
+ self.doc.get_cursor(obj, position, self.or_ours(at))
+ }
+
+ fn get_cursor_moving<O: AsRef<ObjId>, I: Into<CursorPosition>>(
+ &self,
+ obj: O,
+ position: I,
+ at: Option<&[ChangeHash]>,
+ move_cursor: MoveCursor,
+ ) -> Result<Cursor, AutomergeError> {
+ self.doc
+ .get_cursor_moving(obj, position, self.or_ours(at), move_cursor)
+ }
+
+ fn get_cursor_position<O: AsRef<ObjId>>(
+ &self,
+ obj: O,
+ cursor: &Cursor,
+ at: Option<&[ChangeHash]>,
+ ) -> Result<usize, AutomergeError> {
+ self.doc.get_cursor_position(obj, cursor, self.or_ours(at))
+ }
+
+ fn get<O: AsRef<ObjId>, P: Into<Prop>>(
+ &self,
+ obj: O,
+ prop: P,
+ ) -> Result<Option<(Value<'_>, ObjId)>, AutomergeError> {
+ self.doc.get_at(obj, prop, &self.heads)
+ }
+
+ fn get_at<O: AsRef<ObjId>, P: Into<Prop>>(
+ &self,
+ obj: O,
+ prop: P,
+ heads: &[ChangeHash],
+ ) -> Result<Option<(Value<'_>, ObjId)>, AutomergeError> {
+ self.doc.get_at(obj, prop, heads)
+ }
+
+ fn hydrate<O: AsRef<ObjId>>(
+ &self,
+ obj: O,
+ heads: Option<&[ChangeHash]>,
+ ) -> Result<hydrate::Value, AutomergeError> {
+ ReadDoc::hydrate(self.doc, obj, self.or_ours(heads))
+ }
+
+ fn get_all<O: AsRef<ObjId>, P: Into<Prop>>(
+ &self,
+ obj: O,
+ prop: P,
+ ) -> Result<Vec<(Value<'_>, ObjId)>, AutomergeError> {
+ self.doc.get_all_at(obj, prop, &self.heads)
+ }
+
+ fn get_all_at<O: AsRef<ObjId>, P: Into<Prop>>(
+ &self,
+ obj: O,
+ prop: P,
+ heads: &[ChangeHash],
+ ) -> Result<Vec<(Value<'_>, ObjId)>, AutomergeError> {
+ self.doc.get_all_at(obj, prop, heads)
+ }
+
+ fn get_missing_deps(&self, heads: &[ChangeHash]) -> Vec<ChangeHash> {
+ self.doc.get_missing_deps(heads)
+ }
+
+ fn get_change_by_hash(&self, hash: &ChangeHash) -> Option<Change> {
+ self.doc.get_change_by_hash(hash)
+ }
+
+ fn stats(&self) -> Stats {
+ self.doc.stats()
+ }
+
+ fn text_encoding(&self) -> TextEncoding {
+ self.doc.text_encoding()
+ }
+}
diff --git a/src/document/mod.rs b/src/document/mod.rs
index fa1fa77..b9a8b6b 100644
--- a/src/document/mod.rs
+++ b/src/document/mod.rs
@@ -34,6 +34,7 @@ pub(crate) fn actor(doc: &Document) -> ActorId {
doc.get_actor().clone()
}
+pub(crate) mod at;
pub(crate) mod changes;
pub(crate) mod file;
pub(crate) mod lock;
diff --git a/src/document/store.rs b/src/document/store.rs
index c2c0622..c16767b 100644
--- a/src/document/store.rs
+++ b/src/document/store.rs
@@ -94,6 +94,9 @@ pub(crate) enum StoreError {
/// An index past a list's end (`length` items).
#[error("index {index} is past the list's end ({length} items)")]
PastTheEnd { index: usize, length: usize },
+ /// A read at heads the document doesn't have (`at::At`).
+ #[error("this node doesn't have these heads (yet)")]
+ UnknownHeads,
/// The node is rejoining a newer epoch: no writes until it is done
/// (`Replica::writable`).
#[error("the node is rejoining a newer epoch; writes are refused until it is done")]