summaryrefslogtreecommitdiffstats
path: root/src/remake
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake')
-rw-r--r--src/remake/api.rs10
-rw-r--r--src/remake/kind.rs229
-rw-r--r--src/remake/mod.rs10
3 files changed, 200 insertions, 49 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs
index 988bb34..51cbe07 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -24,9 +24,15 @@
//! `?force=true` accepts `settleSecs` below the dormancy threshold and `switchDelaySecs`
//! below one keepalive interval: for tests and debugging, not for real swarms.
//!
-//! `kind` is `compact` (no spec) or `replace`, whose spec is
+//! `kind` is `compact` (no spec), `replace`, whose spec is
//! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }`
-//! (see `kind::Replace`). Reports leave the replacement document out (`contentBytes`).
+//! (see `kind::Replace`), or `import`, whose spec is
+//! `{ "document": "<base64 of an Automerge document>", "rules": […] }` (see
+//! `kind::Import`; `GET /admin/document` exports one). Reports leave the document out
+//! (`contentBytes`). Requests may be up to 64 MiB.
+//!
+//! Restoring a backup, keeping writes made since at their paths:
+//! `base64 -w0 backup.automerge | jq -Rs '{kind: "import", spec: {document: ., rules: [{from: "", to: ""}]}}' | curl --data-binary @- -H 'Content-Type: application/json' …/admin/remakes`
use std::time::Duration;
diff --git a/src/remake/kind.rs b/src/remake/kind.rs
index 7d1300f..685fa28 100644
--- a/src/remake/kind.rs
+++ b/src/remake/kind.rs
@@ -10,13 +10,15 @@
//! * where a write made in the old epoch after the base goes in the new one
//! ([`RemakeKind::translate`]).
//!
-//! Kinds: [`Compact`] (same content, no history) and [`Replace`] (an operator's
-//! document, with rules for where later writes go).
+//! Kinds: [`Compact`] (same content, no history), [`Replace`] (an operator's document
+//! in store form, with rules for where later writes go) and [`Import`] (the same with
+//! an Automerge document, e.g. one exported with `GET /admin/document`).
//!
//! Protocol data under `_continuum` is not a kind's business: the generic part carries
//! it forward and replays it unchanged.
use automerge::AutoCommit;
+use base64::engine::{Engine, general_purpose::STANDARD as BASE64};
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
@@ -75,6 +77,7 @@ pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> {
match name {
"compact" => Some(&Compact),
"replace" => Some(&Replace),
+ "import" => Some(&Import),
_ => None,
}
}
@@ -196,33 +199,9 @@ impl RemakeKind for Replace {
));
}
- let mut rules = Vec::with_capacity(spec.rules.len());
- for rule in spec.rules {
- let from = rule.from.segments()?;
- let to = match (rule.to, rule.drop) {
- (Some(to), false) => Some(to.segments()?),
- (None, true) => None,
- _ => {
- return Err(format!(
- "rule for {from:?} needs exactly one of \"to\" and \"drop\""
- ));
- }
- };
- if is_reserved(&from) || to.as_deref().is_some_and(is_reserved) {
- return Err(format!(
- "rules must not touch the reserved key {META_ROOT:?}"
- ));
- }
- if rules.iter().any(|existing: &Rule| existing.from == from) {
- return Err(format!("two rules for {from:?}"));
- }
- rules.push(Rule { from, to });
- }
-
Ok(Prepared {
content: canonical(spec.document),
- translation: serde_json::to_value(Rules { rules })
- .map_err(|error| error.to_string())?,
+ translation: prepare_rules(spec.rules)?,
})
}
@@ -246,31 +225,132 @@ impl RemakeKind for Replace {
}
fn translate(&self, path: &[String], translation: &JsonValue) -> Translation {
- let Ok(Rules { rules }) = serde_json::from_value::<Rules>(translation.clone()) else {
- return Translation::Quarantine("unreadable replacement rules".to_owned());
- };
- let Some(rule) = rules
- .iter()
- .filter(|rule| path.starts_with(&rule.from))
- .max_by_key(|rule| rule.from.len())
- else {
- return Translation::Quarantine("no replacement rule covers this path".to_owned());
+ translate_by_rules(path, translation)
+ }
+}
+
+/// Validates rules from a request; the translation to store.
+fn prepare_rules(specs: Vec<RuleSpec>) -> Result<JsonValue, String> {
+ let mut rules = Vec::with_capacity(specs.len());
+ for rule in specs {
+ let from = rule.from.segments()?;
+ let to = match (rule.to, rule.drop) {
+ (Some(to), false) => Some(to.segments()?),
+ (None, true) => None,
+ _ => {
+ return Err(format!(
+ "rule for {from:?} needs exactly one of \"to\" and \"drop\""
+ ));
+ }
};
- match &rule.to {
- None => Translation::Drop,
- Some(to) => {
- let mut target = to.clone();
- target.extend_from_slice(&path[rule.from.len()..]);
- if target.as_slice() == path {
- Translation::Same
- } else {
- Translation::To(target)
- }
+ if is_reserved(&from) || to.as_deref().is_some_and(is_reserved) {
+ return Err(format!(
+ "rules must not touch the reserved key {META_ROOT:?}"
+ ));
+ }
+ if rules.iter().any(|existing: &Rule| existing.from == from) {
+ return Err(format!("two rules for {from:?}"));
+ }
+ rules.push(Rule { from, to });
+ }
+ serde_json::to_value(Rules { rules }).map_err(|error| error.to_string())
+}
+
+/// Where `path` goes under the stored rules: the longest matching `from` wins.
+fn translate_by_rules(path: &[String], translation: &JsonValue) -> Translation {
+ let Ok(Rules { rules }) = serde_json::from_value::<Rules>(translation.clone()) else {
+ return Translation::Quarantine("unreadable replacement rules".to_owned());
+ };
+ let Some(rule) = rules
+ .iter()
+ .filter(|rule| path.starts_with(&rule.from))
+ .max_by_key(|rule| rule.from.len())
+ else {
+ return Translation::Quarantine("no replacement rule covers this path".to_owned());
+ };
+ match &rule.to {
+ None => Translation::Drop,
+ Some(to) => {
+ let mut target = to.clone();
+ target.extend_from_slice(&path[rule.from.len()..]);
+ if target.as_slice() == path {
+ Translation::Same
+ } else {
+ Translation::To(target)
}
}
}
}
+/// Import: like [`Replace`], but the new document is an Automerge document (bytes, as
+/// saved), so Automerge types the store's JSON can't express survive: text, counters,
+/// bytes, timestamps, floats. Its `_continuum` (protocol data of whatever swarm it came
+/// from) is left out, as is its history: the content is the merged view.
+///
+/// Spec: `{ "document": "<base64 of the saved document>", "rules": […] }`, rules as for
+/// [`Replace`]. To restore a backup of this swarm with every later write kept where it
+/// was: `"rules": [{"from": "", "to": ""}]`.
+pub(crate) struct Import;
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ImportSpec {
+ document: String,
+ #[serde(default)]
+ rules: Vec<RuleSpec>,
+}
+
+/// The document in `content` (base64 of a saved Automerge document).
+fn load_content(content: &JsonValue) -> Result<AutoCommit, String> {
+ let JsonValue::String(encoded) = content else {
+ return Err("the document must be a base64 string".to_owned());
+ };
+ let bytes = BASE64
+ .decode(encoded.trim())
+ .map_err(|error| format!("the document is not valid base64: {error}"))?;
+ AutoCommit::load(&bytes)
+ .map_err(|error| format!("the document is not an Automerge document: {error}"))
+}
+
+impl RemakeKind for Import {
+ fn name(&self) -> &'static str {
+ "import"
+ }
+
+ fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> {
+ let spec: ImportSpec =
+ serde_json::from_value(spec.clone()).map_err(|error| error.to_string())?;
+ let imported = load_content(&JsonValue::String(spec.document))?;
+
+ // Keep only what goes into the new epoch: the merged view without protocol data
+ // and history. Usually much smaller than what was uploaded, and it is what gets
+ // replicated with the proposal.
+ let mut content = AutoCommit::new();
+ store::materialize_into(&imported, &mut content, &is_reserved)
+ .map_err(|error| format!("the document can't be read: {error}"))?;
+
+ Ok(Prepared {
+ content: JsonValue::String(BASE64.encode(content.save())),
+ translation: prepare_rules(spec.rules)?,
+ })
+ }
+
+ fn build(
+ &self,
+ _old_at_base: &AutoCommit,
+ content: &JsonValue,
+ target: &mut AutoCommit,
+ ) -> Result<(), StoreError> {
+ let imported = load_content(content).map_err(|reason| anyhow::anyhow!(reason))?;
+ // Deterministic: the same bytes give the same view, copied in a fixed order.
+ store::materialize_into(&imported, target, &is_reserved)
+ }
+
+ fn translate(&self, path: &[String], translation: &JsonValue) -> Translation {
+ translate_by_rules(path, translation)
+ }
+}
+
/// The translations of several consecutive remakes, applied in order: a node that
/// rejoins several epochs later carries its writes through every remake in between.
#[derive(Default, Clone)]
@@ -448,6 +528,63 @@ mod tests {
}
#[test]
+ fn import_keeps_types_and_drops_protocol_data() {
+ use automerge::{ObjType, ROOT, ReadDoc, transaction::Transactable};
+ use serde_json::json;
+
+ let mut source = AutoCommit::new();
+ let text = source.put_object(ROOT, "notes", ObjType::Text).unwrap();
+ source.splice_text(&text, 0, 0, "hello").unwrap();
+ source
+ .put(ROOT, "hits", automerge::ScalarValue::counter(3))
+ .unwrap();
+ let meta = source.put_object(ROOT, META_ROOT, ObjType::Map).unwrap();
+ source.put(&meta, "members", "elsewhere").unwrap();
+ let encoded = BASE64.encode(source.save());
+
+ assert!(
+ Import
+ .propose(&json!({ "document": "not base64!" }))
+ .is_err()
+ );
+ assert!(
+ Import
+ .propose(&json!({ "document": BASE64.encode(b"junk") }))
+ .is_err()
+ );
+
+ let prepared = Import
+ .propose(&json!({ "document": encoded, "rules": [{ "from": "", "to": "" }] }))
+ .unwrap();
+ let build = || {
+ let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..]));
+ Import
+ .build(&AutoCommit::new(), &prepared.content, &mut target)
+ .unwrap();
+ target.commit_with(automerge::transaction::CommitOptions::default().with_time(0));
+ target
+ };
+ let mut a = build();
+ let mut b = build();
+ assert_eq!(a.get_heads(), b.get_heads());
+
+ assert!(a.get(ROOT, META_ROOT).unwrap().is_none());
+ let (notes, notes_id) = a.get(ROOT, "notes").unwrap().unwrap();
+ assert!(matches!(notes, automerge::Value::Object(ObjType::Text)));
+ assert_eq!(a.text(&notes_id).unwrap(), "hello");
+ assert!(matches!(
+ a.get(ROOT, "hits").unwrap().unwrap().0,
+ automerge::Value::Scalar(ref s) if matches!(s.as_ref(), automerge::ScalarValue::Counter(_))
+ ));
+
+ // Writes after the base keep their paths.
+ assert_eq!(
+ Translator::single(&Import, prepared.translation).translate(&path(&["x", "y"])),
+ Translation::Same
+ );
+ }
+
+ #[test]
fn replace_builds_the_document_deterministically() {
use serde_json::json;
let build = |document: serde_json::Value| {
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 2429ad7..6a9c6f9 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -38,6 +38,7 @@ use std::time::Duration;
use automerge::AutoCommit;
use axum::{
Router,
+ extract::DefaultBodyLimit,
routing::{get, post},
};
use tokio::sync::broadcast;
@@ -122,13 +123,20 @@ pub(crate) struct PeerSummary {
}
/// Admin routes. Merged into the authenticated part of the node's router.
+/// Largest remake request (bytes). The proposal carries the new document to every
+/// node in one sync message, which Mycelium must deliver within its deadline, so
+/// documents near this size are already a stretch.
+const REMAKE_REQUEST_LIMIT: usize = 64 * 1024 * 1024;
+
pub(crate) fn router<H: Host>() -> Router<H> {
Router::new()
.route(
"/admin/remakes",
get(api::list::<H>)
.post(api::start::<H>)
- .delete(api::abort_current::<H>),
+ .delete(api::abort_current::<H>)
+ // A replace or import request carries a whole document.
+ .layer(DefaultBodyLimit::max(REMAKE_REQUEST_LIMIT)),
)
// The original endpoint, kept as a shortcut for a compacting remake.
.route("/admin/compaction", post(api::start_compaction::<H>))
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 21:38:03 +0000