aboutsummaryrefslogtreecommitdiffstats
path: root/src/state
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-10-31 13:25:11 +0100
committertv <tv@krebsco.de>2026-10-31 13:25:11 +0100
commit694347ad456766004e2792c4b4cda865a8a40386 (patch)
treeebcc97eef8ffd9954e3ecae75bcf5935f5c33d9a /src/state
parent028f80f46784b483215ab7a2952cc54fdd46ea5d (diff)
rename AutoMerge to Document
Diffstat (limited to 'src/state')
-rw-r--r--src/state/host.rs18
-rw-r--r--src/state/mod.rs10
-rw-r--r--src/state/read.rs9
-rw-r--r--src/state/session.rs35
4 files changed, 36 insertions, 36 deletions
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<String> {
+ fn electorate(&self, doc: &Document, now: crate::clock::UnixMillis) -> Vec<String> {
crate::swarm::members::electorate(doc, now.as_u64(), self.electorate_window)
}
fn fetch_change(
&self,
hash: ChangeHash,
- ) -> impl Future<Output = Result<automerge::Change>> + Send {
+ ) -> impl Future<Output = Result<crate::document::Change>> + 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<T, F>(&self, f: F) -> impl Future<Output = T> + 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<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
where
- F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
+ F: FnOnce(&mut Document) -> Result<T, E> + Send + 'static,
T: Send + 'static,
E: Send + 'static,
{
@@ -230,16 +230,16 @@ impl Host for AppState {
&self,
epoch: Epoch,
f: F,
- ) -> impl Future<Output = Result<AutoCommit, E>> + Send
+ ) -> impl Future<Output = Result<Document, E>> + Send
where
- F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static,
+ F: FnOnce(&mut Document) -> Result<Document, E> + Send + 'static,
E: Send + 'static,
{
let state = self.clone();
async move {
let swapped = {
let state = state.clone();
- tokio::task::spawn_blocking(move || -> Result<AutoCommit, E> {
+ tokio::task::spawn_blocking(move || -> Result<Document, E> {
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<DocumentLock>,
/// 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<T, R>(
&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<T>(
&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<T, E>(
&self,
- f: impl FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
+ f: impl FnOnce(&mut Document) -> Result<T, E> + Send + 'static,
) -> Result<T, E>
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<AutoCommit>,
+ pub(crate) doc: Arc<Document>,
}
/// 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<Vec<ChangeHash>>), axum::http::StatusCode>
where
@@ -67,7 +66,7 @@ impl AppState {
async fn render_live<T, R>(
&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<Vec<ChangeHash>>)
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<automerge::ChangeHash>,
+ pub(crate) sent: Vec<crate::document::ChangeHash>,
/// 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<automerge::ChangeHash>, bool),
+ before_generate: (Vec<crate::document::ChangeHash>, bool),
/// The answer we expect to our last sync message(s), if any.
pub(crate) reply: Option<SyncReplyWait>,
@@ -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<automerge::ChangeHash>,
+ pub(crate) sent: Vec<crate::document::ChangeHash>,
}
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<sync::Message> {
@@ -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<automerge::ChangeHash> {
+ pub(crate) fn withdraw_for_delta(&mut self) -> Vec<crate::document::ChangeHash> {
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<automerge::ChangeHash>) {
+ pub(crate) fn pull_from(&mut self, heads: Vec<crate::document::ChangeHash>) {
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<automerge::ChangeHash>,
+ sent: Vec<crate::document::ChangeHash>,
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();