From baf9498c69340da18001c7ca2c223ed45fddc249 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 13:25:11 +0100 Subject: rename AutoMerge to Document --- src/api/store.rs | 2 +- src/document/mod.rs | 5 +++++ src/main.rs | 4 ++-- src/node.rs | 4 ++-- src/simtest/mod.rs | 4 ++-- src/state/host.rs | 18 +++++++++--------- src/state/mod.rs | 10 +++++----- src/state/read.rs | 9 ++++----- src/state/session.rs | 35 ++++++++++++++++++----------------- src/swarm/blob.rs | 18 +++++++++--------- src/swarm/bootstrap.rs | 12 ++++++------ src/swarm/members.rs | 34 +++++++++++++++++----------------- src/swarm/peer/forget.rs | 4 ++-- src/swarm/relay.rs | 2 +- src/swarm/state_file.rs | 2 +- src/transport/mod.rs | 3 ++- src/transport/mycelium/mod.rs | 2 +- src/transport/wire.rs | 10 +++++----- 18 files changed, 92 insertions(+), 86 deletions(-) diff --git a/src/api/store.rs b/src/api/store.rs index 274ca70..f121ff6 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -2,7 +2,6 @@ use std::io::Error as IoError; -use automerge::ChangeHash; use axum::{ Extension, body::{Body, Bytes}, @@ -21,6 +20,7 @@ use tracing::{debug, error, instrument, warn}; use super::auth::AuthenticatedUser; use super::{ApiError, media_type, store_status}; +use crate::document::ChangeHash; use crate::document::meta; use crate::document::store::{self, StoreError, paths_related}; use crate::state::{AppState, ReadSource}; diff --git a/src/document/mod.rs b/src/document/mod.rs index b86d1cc..f5faa11 100644 --- a/src/document/mod.rs +++ b/src/document/mod.rs @@ -1,6 +1,11 @@ //! The replicated document: the replica with its epoch, and the store semantics //! (merged-view paths over JSON) the HTTP API and remakes work with. +/// The engine's types, under the names the rest of the crate uses: the document +/// (Automerge's `AutoCommit`), its changes and their hashes, and its writers (actors). +/// Only `document` and `remake` use Automerge itself (`architecture.md`, The document). +pub(crate) use automerge::{ActorId, AutoCommit as Document, Change, ChangeHash}; + pub(crate) mod file; pub(crate) mod lock; pub(crate) mod meta; diff --git a/src/main.rs b/src/main.rs index f5bd5d4..a725174 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,7 +9,6 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use anyhow::{Context, Result}; -use automerge::AutoCommit; use tokio::sync::watch; use tracing::{debug, error, info, warn}; @@ -35,6 +34,7 @@ mod transport; mod units; use api::auth::AuthState; +use document::Document; use document::replica::Epoch; use swarm::bootstrap; use transport::Transport; @@ -148,7 +148,7 @@ async fn main() -> Result<()> { Some((starting.swarm_id, starting.origin)), starting.earlier_actors, ), - None => (AutoCommit::new(), None, None, Vec::new()), + None => (Document::new(), None, None, Vec::new()), }; let mut initial_doc = initial_doc; if let Some(spec) = &args.remake { diff --git a/src/node.rs b/src/node.rs index 42b3c91..f09de5b 100644 --- a/src/node.rs +++ b/src/node.rs @@ -8,13 +8,13 @@ use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::time::Duration; -use automerge::{ActorId, AutoCommit}; use tokio::sync::{RwLock, broadcast, mpsc, watch}; use tracing::trace; use crate::clock::Clock; use crate::document::file::DocumentFile; use crate::document::replica::{Epoch, Replica}; +use crate::document::{ActorId, Document}; use crate::files::Files; use crate::remake; use crate::state::{AppState, NodeLifecycle, Timing}; @@ -37,7 +37,7 @@ pub(crate) struct NodeConfig { /// (simulations only), the node is its transport identity. pub(crate) identity: Option, pub(crate) initial_epoch: Epoch, - pub(crate) initial_doc: AutoCommit, + pub(crate) initial_doc: Document, /// Actors this node wrote with in earlier runs (`--document-file`). pub(crate) earlier_actors: Vec, /// `Timing::from_keepalive(--keepalive, transport.delivery_limit())`. diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index 4dedc4b..9642f28 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -20,12 +20,12 @@ use std::future::Future; use std::sync::Arc; use std::time::Duration; -use automerge::{AutoCommit, ChangeHash}; use serde_json::Value as JsonValue; use tokio::time::Instant; use crate::document::replica::Epoch; use crate::document::store; +use crate::document::{ChangeHash, Document}; use crate::files::Files; use crate::identity::NodeKey; use crate::locks::RwLockExt; @@ -107,7 +107,7 @@ impl Sim { more: Vec::new(), }), initial_epoch: Epoch::GENESIS, - initial_doc: AutoCommit::new(), + initial_doc: Document::new(), earlier_actors: Vec::new(), timing: Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT), neighbours_k: 8, diff --git a/src/state/host.rs b/src/state/host.rs index 912a187..a878381 100644 --- a/src/state/host.rs +++ b/src/state/host.rs @@ -6,7 +6,6 @@ use std::sync::Arc; use std::time::Duration; use anyhow::Result; -use automerge::{AutoCommit, ChangeHash}; use tokio::sync::broadcast; use tokio::time::Instant; use tracing::debug; @@ -14,6 +13,7 @@ use tracing::debug; use super::{AppState, HelloState, NodeLifecycle, StaleView, Standing}; use crate::document::replica::Epoch; use crate::document::store::ChangeSet; +use crate::document::{ChangeHash, Document}; use crate::remake::{Host, PeerSummary, RemakeState}; impl AppState { @@ -86,14 +86,14 @@ impl Host for AppState { crate::swarm::skew::measure(self).await } - fn electorate(&self, doc: &AutoCommit, now: crate::clock::UnixMillis) -> Vec { + fn electorate(&self, doc: &Document, now: crate::clock::UnixMillis) -> Vec { crate::swarm::members::electorate(doc, now.as_u64(), self.electorate_window) } fn fetch_change( &self, hash: ChangeHash, - ) -> impl Future> + Send { + ) -> impl Future> + Send { let state = self.clone(); async move { let peers = state.active_peers_on_local_epoch().await; @@ -184,7 +184,7 @@ impl Host for AppState { } } - fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>) { + fn set_stale_view(&self, view: Option<(Epoch, Document)>) { self.set_stale(view.map(|(epoch, doc)| StaleView { epoch, doc: Arc::new(doc), @@ -209,7 +209,7 @@ impl Host for AppState { fn read_doc(&self, f: F) -> impl Future + Send where - F: FnOnce(&AutoCommit) -> T + Send + 'static, + F: FnOnce(&Document) -> T + Send + 'static, T: Send + 'static, { let state = self.clone(); @@ -218,7 +218,7 @@ impl Host for AppState { fn commit_doc(&self, f: F) -> impl Future> + Send where - F: FnOnce(&mut AutoCommit) -> Result + Send + 'static, + F: FnOnce(&mut Document) -> Result + Send + 'static, T: Send + 'static, E: Send + 'static, { @@ -230,16 +230,16 @@ impl Host for AppState { &self, epoch: Epoch, f: F, - ) -> impl Future> + Send + ) -> impl Future> + Send where - F: FnOnce(&mut AutoCommit) -> Result + Send + 'static, + F: FnOnce(&mut Document) -> Result + Send + 'static, E: Send + 'static, { let state = self.clone(); async move { let swapped = { let state = state.clone(); - tokio::task::spawn_blocking(move || -> Result { + tokio::task::spawn_blocking(move || -> Result { let mut replica = state.doc.blocking_write(); let new_doc = f(&mut replica.doc)?; let old_doc = std::mem::replace(&mut replica.doc, new_doc); diff --git a/src/state/mod.rs b/src/state/mod.rs index 36dd4ad..efa4953 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -35,11 +35,11 @@ use std::sync::atomic::AtomicBool; use std::time::Duration; use anyhow::{Result, bail}; -use automerge::AutoCommit; use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch}; use tokio::time::Instant; use tracing::{debug, error, warn}; +use crate::document::Document; use crate::document::lock::DocumentLock; use crate::document::replica::{Epoch, Stamp}; use crate::document::store::ChangeSet; @@ -59,7 +59,7 @@ pub(crate) enum NodeLifecycle { #[derive(Clone)] pub(crate) struct AppState { - /// The document and its epoch. Derefs to `AutoCommit`. + /// The document and its epoch. Derefs to `Document`. pub(crate) doc: Arc, /// Copy of `doc.epoch` for cheap, lock-free reads (handshakes, keepalives, /peers). @@ -376,7 +376,7 @@ impl AppState { /// Reads the document on the blocking pool; `finish` runs after the lock is released. pub(crate) async fn render( &self, - read: impl FnOnce(&AutoCommit) -> T + Send + 'static, + read: impl FnOnce(&Document) -> T + Send + 'static, finish: impl FnOnce(T) -> R + Send + 'static, ) -> R where @@ -397,7 +397,7 @@ impl AppState { /// Mutates the document on the blocking pool. pub(crate) async fn write_doc( &self, - f: impl FnOnce(&mut AutoCommit) -> T + Send + 'static, + f: impl FnOnce(&mut Document) -> T + Send + 'static, ) -> T where T: Send + 'static, @@ -418,7 +418,7 @@ impl AppState { /// cancelled request can't leave a committed write unannounced. pub(crate) async fn mutate_doc( &self, - f: impl FnOnce(&mut AutoCommit) -> Result + Send + 'static, + f: impl FnOnce(&mut Document) -> Result + Send + 'static, ) -> Result where T: Send + 'static, diff --git a/src/state/read.rs b/src/state/read.rs index f8323a2..a9b3116 100644 --- a/src/state/read.rs +++ b/src/state/read.rs @@ -2,16 +2,15 @@ use std::sync::Arc; -use automerge::{AutoCommit, ChangeHash}; - use super::AppState; use crate::document::replica::Epoch; +use crate::document::{ChangeHash, Document}; use crate::locks::RwLockExt; /// A read-only view of the document the node served before it started rejoining. pub(crate) struct StaleView { pub(crate) epoch: Epoch, - pub(crate) doc: Arc, + pub(crate) doc: Arc, } /// Where a read was served from. @@ -33,7 +32,7 @@ impl AppState { &self, unready_live: bool, with_heads: bool, - read: impl FnOnce(&AutoCommit) -> T + Send + 'static, + read: impl FnOnce(&Document) -> T + Send + 'static, finish: impl FnOnce(T) -> R + Send + 'static, ) -> Result<(R, ReadSource, Option>), axum::http::StatusCode> where @@ -67,7 +66,7 @@ impl AppState { async fn render_live( &self, with_heads: bool, - read: impl FnOnce(&AutoCommit) -> T + Send + 'static, + read: impl FnOnce(&Document) -> T + Send + 'static, finish: impl FnOnce(T) -> R + Send + 'static, ) -> (R, Option>) where diff --git a/src/state/session.rs b/src/state/session.rs index 8885699..98b4dc3 100644 --- a/src/state/session.rs +++ b/src/state/session.rs @@ -9,12 +9,13 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; +use automerge::AutomergeError; use automerge::sync::{self, SyncDoc}; -use automerge::{AutoCommit, AutomergeError}; use tokio::time::Instant; use tracing::debug; use crate::backoff::Backoff; +use crate::document::Document; use crate::swarm::throughput::SyncBytes; /// A sync message being handed to our transport (for Mycelium: the HTTP request is in @@ -22,7 +23,7 @@ use crate::swarm::throughput::SyncBytes; #[derive(Clone)] pub(crate) struct Sending { /// Changes it carries for the first time; taken back if the send fails. - pub(crate) sent: Vec, + pub(crate) sent: Vec, /// A sync pass was requested meanwhile. Once this one is handed over, the peer must /// be enqueued again so the deferred changes go out. pub(crate) regenerate: bool, @@ -63,7 +64,7 @@ pub(crate) struct SyncSession { /// once the peer has the delta and says its heads are ours, never answers it, so the /// peer never learns our heads (a joining node then never finishes its initial /// sync). - before_generate: (Vec, bool), + before_generate: (Vec, bool), /// The answer we expect to our last sync message(s), if any. pub(crate) reply: Option, @@ -81,7 +82,7 @@ pub(crate) struct SyncReplyWait { /// exchanged with the peer by then (`swarm::throughput`). pub(crate) started: (Instant, SyncBytes), /// Changes first sent in the unanswered messages. - pub(crate) sent: Vec, + pub(crate) sent: Vec, } impl SyncSession { @@ -113,7 +114,7 @@ impl SyncSession { /// peer so far (`swarm::throughput`). pub(crate) fn generate( &mut self, - doc: &mut AutoCommit, + doc: &mut Document, now: Instant, bytes: SyncBytes, ) -> Option { @@ -150,7 +151,7 @@ impl SyncSession { /// answer: the caller measures the exchange first (`reply`). pub(crate) fn receive( &mut self, - doc: &mut AutoCommit, + doc: &mut Document, message: sync::Message, ) -> Result<(), AutomergeError> { self.delta_offered = false; @@ -162,7 +163,7 @@ impl SyncSession { /// message being handed over becomes the offer (it carries nothing); the answer is /// still awaited, and if it doesn't come, the offer is made again. Returns the /// peer's heads as far as we know them: where the delta starts. - pub(crate) fn withdraw_for_delta(&mut self) -> Vec { + pub(crate) fn withdraw_for_delta(&mut self) -> Vec { if let Some(sending) = &mut self.sending { let sent = std::mem::take(&mut sending.sent); for hash in &sent { @@ -181,7 +182,7 @@ impl SyncSession { /// Whether `doc` holds everything the peer had when it last told us its heads: they /// are known, not empty (empty heads are "contained" in every document, so they /// prove nothing), and all in `doc`. - pub(crate) fn have_all_theirs(&self, doc: &mut AutoCommit) -> bool { + pub(crate) fn have_all_theirs(&self, doc: &mut Document) -> bool { self.state.their_heads.as_ref().is_some_and(|heads| { !heads.is_empty() && heads @@ -192,7 +193,7 @@ impl SyncSession { /// The peer has heads we lack (its keepalive said so): ask for what they need, /// even though Automerge thinks a message is still in flight. - pub(crate) fn pull_from(&mut self, heads: Vec) { + pub(crate) fn pull_from(&mut self, heads: Vec) { self.state.their_heads = Some(heads); self.state.in_flight = false; } @@ -232,7 +233,7 @@ impl SyncSession { /// For the simulator's dump: the peer's heads as Automerge knows them, and whether /// it thinks a message is in flight. #[cfg(test)] - pub(crate) fn automerge_view(&self) -> (Option<&[automerge::ChangeHash]>, bool) { + pub(crate) fn automerge_view(&self) -> (Option<&[crate::document::ChangeHash]>, bool) { (self.state.their_heads.as_deref(), self.state.in_flight) } @@ -294,7 +295,7 @@ impl SyncSession { /// `bytes` are the sync bytes exchanged with the peer before it. pub(crate) fn expect_reply( &mut self, - sent: Vec, + sent: Vec, now: Instant, bytes: SyncBytes, ) { @@ -372,8 +373,8 @@ mod tests { #[test] fn missing_predecessors_go_at_once() { use automerge::transaction::Transactable; - let mut a = AutoCommit::new(); - let mut b = AutoCommit::new(); + let mut a = Document::new(); + let mut b = Document::new(); let (mut at_a, mut at_b) = (sync::State::new(), sync::State::new()); a.put(automerge::ROOT, "early", 1).unwrap(); // Until in step. @@ -462,10 +463,10 @@ mod tests { let big: String = (0..1024) .map(|_| format!("{:016x}", rng.next_u64())) .collect(); - let mut ours = AutoCommit::new(); + let mut ours = Document::new(); ours.put(ROOT, "big", big).unwrap(); ours.commit(); - let mut theirs = AutoCommit::new(); + let mut theirs = Document::new(); let (mut a, mut b) = (session(), session()); let now = Instant::now(); let bytes = SyncBytes::default(); @@ -508,10 +509,10 @@ mod tests { let big: String = (0..1024) .map(|_| format!("{:016x}", rng.next_u64())) .collect(); - let mut ours = AutoCommit::new(); + let mut ours = Document::new(); ours.put(ROOT, "big", big).unwrap(); ours.commit(); - let mut theirs = AutoCommit::new(); + let mut theirs = Document::new(); let (mut a, mut b) = (session(), session()); let now = Instant::now(); let bytes = SyncBytes::default(); diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index ec99c93..57b02b2 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -28,11 +28,11 @@ use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; use anyhow::{Result, anyhow, bail}; -use automerge::{AutoCommit, ChangeHash}; use tokio::sync::{Semaphore, mpsc}; use tokio::time::Instant; use tracing::{debug, info, warn}; +use crate::document::{ChangeHash, Document}; use crate::locks::MutexExt; use crate::state::AppState; use crate::transport::{BlobBody, BlobId, BlobMessage, DeltaSpec, ParsedMessage}; @@ -586,9 +586,9 @@ pub(crate) async fn fetch_change( state: &AppState, hash: ChangeHash, peers: Vec, -) -> Result { +) -> Result { let bytes = fetch(state, BlobId::Change(hash), peers, None, None).await?; - let change = automerge::Change::from_bytes(bytes) + let change = crate::document::Change::from_bytes(bytes) .map_err(|error| anyhow!("downloaded change {hash} doesn't load: {error}"))?; if change.hash() != hash { bail!("downloaded change has hash {}, not {hash}", change.hash()); @@ -608,9 +608,9 @@ pub(crate) async fn fetch_delta( /// The bytes of delta `spec` from `doc`: the changes in the history of `spec.to` but /// not of `spec.from`, each as Automerge encodes it, sorted by hash and concatenated -/// (`AutoCommit::load_incremental` reads that). The same on every node holding `to`, +/// (`Document::load_incremental` reads that). The same on every node holding `to`, /// whatever else it has. None if `doc` lacks part of `to`. -pub(crate) fn delta_bytes(doc: &mut AutoCommit, spec: &DeltaSpec) -> Option> { +pub(crate) fn delta_bytes(doc: &mut Document, spec: &DeltaSpec) -> Option> { if spec .to .iter() @@ -659,7 +659,7 @@ mod tests { use super::*; use automerge::{ActorId, ROOT, transaction::Transactable}; - fn random_write(rng: &mut crate::testrand::Rng, doc: &mut AutoCommit) { + fn random_write(rng: &mut crate::testrand::Rng, doc: &mut Document) { let key = format!("k{}", rng.below(6)); doc.put(ROOT, key, rng.below(100) as i64).unwrap(); doc.commit(); @@ -677,12 +677,12 @@ mod tests { fn delta_bytes_are_the_same_everywhere_and_complete() { crate::testrand::for_seeds(200, |rng| { let actor = |i: u8| ActorId::from(&[i; 16][..]); - let mut base = AutoCommit::new().with_actor(actor(0)); + let mut base = Document::new().with_actor(actor(0)); for _ in 0..rng.between(0, 5) { random_write(rng, &mut base); } let from = base.get_heads(); - let mut writers: Vec = (1..=rng.between(1, 3)) + let mut writers: Vec = (1..=rng.between(1, 3)) .map(|i| base.fork().with_actor(actor(i as u8))) .collect(); for _ in 0..rng.between(1, 15) { @@ -714,7 +714,7 @@ mod tests { #[test] fn no_delta_without_its_heads() { - let mut doc = AutoCommit::new(); + let mut doc = Document::new(); doc.put(ROOT, "k", 1_i64).unwrap(); doc.commit(); let spec = DeltaSpec::new(Vec::new(), vec![ChangeHash([7; 32])]); diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index a9ab1b2..ec9a355 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -18,7 +18,6 @@ use std::collections::HashSet; use std::path::Path; use anyhow::{Context, Result, bail}; -use automerge::{ActorId, AutoCommit}; use serde_json::Value as JsonValue; use tokio::time::Instant; use tracing::{debug, info, warn}; @@ -28,6 +27,7 @@ use crate::document; use crate::document::meta::{self, SWARM}; use crate::document::replica::Epoch; use crate::document::store; +use crate::document::{ActorId, Document}; use crate::files::{FileKind, Files}; use crate::force::Force; use crate::remake::snapshot; @@ -156,7 +156,7 @@ fn swarm_id_path() -> Vec { } /// The swarm ID the document names, if any. -fn read_swarm_id(doc: &AutoCommit) -> Option { +fn read_swarm_id(doc: &Document) -> Option { match store::read_path(doc, &swarm_id_path()) { Ok(Some(JsonValue::String(id))) if !id.is_empty() => Some(id), _ => None, @@ -184,7 +184,7 @@ impl Origin { /// A document the node starts from, ready at once: on the swarm and epoch it names. pub(crate) struct StartingDocument { /// Under a fresh actor. - pub(crate) doc: AutoCommit, + pub(crate) doc: Document, pub(crate) epoch: Epoch, pub(crate) swarm_id: String, pub(crate) origin: Origin, @@ -281,7 +281,7 @@ pub(crate) fn starting_document( /// The swarm and epoch `doc` is on. fn identify( - mut doc: AutoCommit, + mut doc: Document, origin: Origin, earlier_actors: Vec, ) -> Result { @@ -301,12 +301,12 @@ fn identify( } /// A backup to recover from, as `GET /admin/document` saves it, under a fresh actor. -fn load_backup(path: &Path, files: &Files) -> Result { +fn load_backup(path: &Path, files: &Files) -> Result { let bytes = files .read(FileKind::Backup, path) .with_context(|| format!("Reading the backup {}", path.display()))? .with_context(|| format!("The backup {} doesn't exist", path.display()))?; - Ok(AutoCommit::load(&bytes) + Ok(Document::load(&bytes) .with_context(|| format!("Loading the backup {}", path.display()))? .with_actor(ActorId::random())) } diff --git a/src/swarm/members.rs b/src/swarm/members.rs index 94db4c9..0e27681 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -18,10 +18,10 @@ use std::collections::HashMap; use std::time::Duration; -use automerge::AutoCommit; use serde_json::{Value as JsonValue, json}; use tracing::{debug, info, warn}; +use crate::document::Document; use crate::document::meta; use crate::document::store::{self, StoreError}; use crate::state::AppState; @@ -37,7 +37,7 @@ pub(crate) fn member_path(pk: &str) -> Vec { } /// Public keys of all registered members, sorted. -pub(crate) fn members(doc: &AutoCommit) -> Vec { +pub(crate) fn members(doc: &Document) -> Vec { let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default(); members.sort(); members @@ -67,14 +67,14 @@ fn tier_path(pk: &str) -> Vec { } /// `pk`'s tier; transient unless its entry says stable. -pub(crate) fn tier(doc: &AutoCommit, pk: &str) -> Tier { +pub(crate) fn tier(doc: &Document, pk: &str) -> Tier { match store::read_path(doc, &tier_path(pk)) { Ok(Some(JsonValue::String(tier))) if tier == "stable" => Tier::Stable, _ => Tier::Transient, } } -pub(crate) fn set_tier(doc: &mut AutoCommit, pk: &str, tier: Tier) -> Result<(), StoreError> { +pub(crate) fn set_tier(doc: &mut Document, pk: &str, tier: Tier) -> Result<(), StoreError> { store::write_path(doc, &tier_path(pk), &json!(tier.as_str())) } @@ -85,7 +85,7 @@ fn alias_path(pk: &str) -> Vec { } /// `pk`'s alias, if it gave itself one. -pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option { +pub(crate) fn alias(doc: &Document, pk: &str) -> Option { match store::read_path(doc, &alias_path(pk)) { Ok(Some(JsonValue::String(alias))) => Some(alias), _ => None, @@ -93,14 +93,14 @@ pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option { } /// The aliases of all members that have one. -pub(crate) fn aliases(doc: &AutoCommit) -> HashMap { +pub(crate) fn aliases(doc: &Document) -> HashMap { members(doc) .into_iter() .filter_map(|pk| alias(doc, &pk).map(|alias| (pk, alias))) .collect() } -fn set_alias(doc: &mut AutoCommit, pk: &str, alias: Option<&str>) -> Result<(), StoreError> { +fn set_alias(doc: &mut Document, pk: &str, alias: Option<&str>) -> Result<(), StoreError> { match alias { Some(alias) => store::write_path(doc, &alias_path(pk), &json!(alias)), None => match store::delete_path(doc, &alias_path(pk), false) { @@ -121,7 +121,7 @@ const MAX_ADDRESSES: usize = 8; /// Where `pk` says it can be reached (at most `MAX_ADDRESSES`); none if its entry /// doesn't say. -pub(crate) fn addresses(doc: &AutoCommit, pk: &str) -> Vec { +pub(crate) fn addresses(doc: &Document, pk: &str) -> Vec { match store::read_path(doc, &addresses_path(pk)) { Ok(Some(JsonValue::Array(items))) => items .into_iter() @@ -138,7 +138,7 @@ pub(crate) fn addresses(doc: &AutoCommit, pk: &str) -> Vec { /// What the member list tells this node besides who its neighbours are: where members /// are (`learn_addresses`), what they call themselves (`set_aliases`), and the relays /// listed (`relays`). Done whenever the neighbours are recomputed (`topology::refresh`). -pub(crate) fn apply(state: &AppState, doc: &AutoCommit) { +pub(crate) fn apply(state: &AppState, doc: &Document) { learn_addresses(state, doc); set_aliases(state, aliases(doc)); state.transport.learn_relays(&relays(doc)); @@ -159,7 +159,7 @@ fn set_aliases(state: &AppState, aliases: HashMap) { /// Hands every other member's addresses to the transport (`Transport::learn_addresses`): /// the member list is how a node finds where members it never met are. -pub(crate) fn learn_addresses(state: &AppState, doc: &AutoCommit) { +pub(crate) fn learn_addresses(state: &AppState, doc: &Document) { let me = state.local_id(); for pk in members(doc) { if pk != me { @@ -175,7 +175,7 @@ const MAX_RELAYS: usize = 32; /// addresses (`libp2p:`, with or without the relay's peer ID) of relays that /// aren't members, e.g. nodes that only relay. Every member's transport may /// pick them (`Transport::learn_relays`), after the relays it is configured with. -pub(crate) fn relays(doc: &AutoCommit) -> Vec { +pub(crate) fn relays(doc: &Document) -> Vec { match store::read_path(doc, &meta::path(&[meta::RELAYS])) { Ok(Some(JsonValue::Array(items))) => items .into_iter() @@ -195,7 +195,7 @@ fn alive_path(pk: &str) -> Vec { path } -fn millis_at(doc: &AutoCommit, path: &[String]) -> Option { +fn millis_at(doc: &Document, path: &[String]) -> Option { match store::read_path(doc, path) { Ok(Some(JsonValue::Number(number))) => number.as_u64(), _ => None, @@ -203,7 +203,7 @@ fn millis_at(doc: &AutoCommit, path: &[String]) -> Option { } /// When `pk` last said it is alive (Unix ms): its last renewal, or its registration. -pub(crate) fn alive_at(doc: &AutoCommit, pk: &str) -> Option { +pub(crate) fn alive_at(doc: &Document, pk: &str) -> Option { let mut since = member_path(pk); since.push("since".to_owned()); millis_at(doc, &alive_path(pk)).max(millis_at(doc, &since)) @@ -211,7 +211,7 @@ pub(crate) fn alive_at(doc: &AutoCommit, pk: &str) -> Option { /// The members that said they are alive within `window` before `now_ms` (sorted): the /// electorate of a remake (`remake::coordinator`). -pub(crate) fn electorate(doc: &AutoCommit, now_ms: u64, window: Duration) -> Vec { +pub(crate) fn electorate(doc: &Document, now_ms: u64, window: Duration) -> Vec { let oldest = now_ms.saturating_sub(window.as_millis() as u64); members(doc) .into_iter() @@ -314,7 +314,7 @@ pub(crate) async fn ensure_registered(state: &AppState) { } /// Removes `pk`'s entry (an operator forgot it). Absent is fine. -pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreError> { +pub(crate) fn unregister(doc: &mut Document, pk: &str) -> Result<(), StoreError> { match store::delete_path(doc, &member_path(pk), false) { Err(StoreError::NotFound) => Ok(()), result => result, @@ -329,7 +329,7 @@ mod tests { fn electorate_by_alive_stamp() { let day = 24 * 3600 * 1000; let now = 100 * day; - let mut doc = AutoCommit::new(); + let mut doc = Document::new(); let entries = [ ("fresh", json!({ "since": 1, "alive": now - day })), ("stale", json!({ "since": 1, "alive": now - 8 * day })), @@ -347,7 +347,7 @@ mod tests { #[test] fn listed_relays() { - let mut doc = AutoCommit::new(); + let mut doc = Document::new(); assert!(relays(&doc).is_empty()); let relay = "libp2p:/ip6/fd00::1/tcp/9661"; store::write_path(&mut doc, &meta::path(&[meta::RELAYS]), &json!([relay, 7])).unwrap(); diff --git a/src/swarm/peer/forget.rs b/src/swarm/peer/forget.rs index 6386e3e..3a97677 100644 --- a/src/swarm/peer/forget.rs +++ b/src/swarm/peer/forget.rs @@ -5,11 +5,11 @@ use std::collections::HashMap; use std::time::Duration; -use automerge::AutoCommit; use serde_json::Value as JsonValue; use tokio::time::Instant; use tracing::{info, warn}; +use crate::document::Document; use crate::document::meta; use crate::document::store::{self, StoreError}; use crate::state::{AppState, Standing}; @@ -24,7 +24,7 @@ pub(crate) fn forgotten_path(pk: &str) -> Vec { path } -fn forgotten_peers(doc: &AutoCommit) -> Vec { +fn forgotten_peers(doc: &Document) -> Vec { store::keys_at(doc, &forgotten_dir()).unwrap_or_default() } diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 00043f2..c28c9f1 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -131,7 +131,7 @@ pub(crate) async fn pull_if_behind( // Those of the listed heads we lack. With a digest only a sample is listed; if we // have all of it, a later keepalive lists others. let missing: Vec<_> = { - // AutoCommit's lookups take &mut self. + // Document's lookups take &mut self. let mut replica = state.doc.write().await; remote .heads diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs index a65fa5b..415fb1a 100644 --- a/src/swarm/state_file.rs +++ b/src/swarm/state_file.rs @@ -185,7 +185,7 @@ pub(crate) async fn update(state: &AppState, now: bool) { /// Where to rejoin through: the addresses of members alive within the electorate /// window, stable ones first, of at most `MAX_SEEDS` members, sorted. Changes only when /// members join, leave, move or change tier, so the file rarely does. -fn seeds(doc: &automerge::AutoCommit, me: &str, window: Duration, now: u64) -> Vec { +fn seeds(doc: &crate::document::Document, me: &str, window: Duration, now: u64) -> Vec { let (mut stable, transient): (Vec, Vec) = members::electorate(doc, now, window) .into_iter() .filter(|pk| pk != me) diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 048fe31..c10e0d8 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -1,6 +1,5 @@ use anyhow::Result; use async_trait::async_trait; -use automerge::ChangeHash; use automerge::sync::Message; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -11,6 +10,8 @@ use std::sync::atomic::{AtomicUsize, Ordering}; pub(crate) use crate::document::replica::Epoch; +use crate::document::ChangeHash; + // Used by the transports that don't authenticate their peers themselves (`tor`). #[cfg_attr(not(feature = "tor"), allow(dead_code))] pub(crate) mod challenge; diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index ac55ae9..21c939d 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -657,10 +657,10 @@ pub(crate) enum TransportError { #[cfg(test)] mod tests { use super::*; + use crate::document::ChangeHash; use crate::transport::{ BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, MAX_LISTED_HEADS, Status, }; - use automerge::ChangeHash; /// Mycelium splits messages into chunks of this size. const MYCELIUM_CHUNK: usize = 1300; diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 46cea4c..26794a1 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -203,7 +203,7 @@ pub(crate) mod codec { /// An Automerge change hash. pub(crate) mod change_hash { use super::*; - use automerge::ChangeHash; + use crate::document::ChangeHash; pub(crate) fn serialize( value: &ChangeHash, @@ -242,8 +242,8 @@ pub(crate) mod codec { /// empty), or as a digest (none listed). pub(crate) mod heads { use super::*; + use crate::document::ChangeHash; use crate::transport::{HeadsSummary, MAX_LISTED_HEADS}; - use automerge::ChangeHash; pub(crate) fn serialize( value: &HeadsSummary, @@ -290,7 +290,7 @@ pub(crate) mod codec { /// Change hashes, each a 32-byte binary. pub(crate) mod change_hashes { use super::*; - use automerge::ChangeHash; + use crate::document::ChangeHash; pub(crate) fn serialize( value: &[ChangeHash], @@ -352,12 +352,12 @@ pub(crate) mod codec { /// envelope around it). #[cfg(test)] pub(crate) fn samples() -> Vec { + use crate::document::ChangeHash; use crate::document::replica::Epoch; use crate::transport::{ BlobBody, BlobId, BlobMessage, DeltaSpec, HeadsSummary, HelloReply, HelloRequest, IdentityProof, ListedPeer, Status, SyncBody, SyncEnvelope, }; - use automerge::ChangeHash; let epoch = Epoch { generation: 2, id: [3; 32], @@ -456,12 +456,12 @@ pub(crate) fn samples() -> Vec { #[cfg(test)] mod tests { use super::*; + use crate::document::ChangeHash; use crate::document::replica::Epoch; use crate::transport::{ BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, IdentityProof, ListedPeer, MAX_LISTED_HEADS, Status, }; - use automerge::ChangeHash; /// Every kind decodes to what was encoded: re-encoded, the same bytes. #[test] -- cgit v1.3.1