diff options
| author | tv <tv@krebsco.de> | 2026-10-31 19:53:27 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-10-31 19:53:27 +0100 |
| commit | 9c12bf2e8815dc9b42be62f1e1fd2932dd02938b (patch) | |
| tree | 09cd985fae9142a247a2285c755f9d002a2c7b30 /src/swarm | |
| parent | f4b0a7c5159d418e22682d2bd7d413fb3587ca7f (diff) | |
use Automerge as Document instead of AutoCommit
This allows a lot of operations to drop mut.
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/blob.rs | 5 | ||||
| -rw-r--r-- | src/swarm/bootstrap.rs | 5 | ||||
| -rw-r--r-- | src/swarm/changes.rs | 7 | ||||
| -rw-r--r-- | src/swarm/document_file.rs | 4 | ||||
| -rw-r--r-- | src/swarm/members.rs | 10 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 4 | ||||
| -rw-r--r-- | src/swarm/stats.rs | 4 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 10 |
8 files changed, 23 insertions, 26 deletions
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index a136217..3f66d50 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -217,7 +217,7 @@ async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Opti let spec = delta.filter(|spec| spec.id() == delta_id)?; let doc = Arc::clone(&state.doc); tokio::task::spawn_blocking(move || { - changes::delta_bytes(&mut doc.blocking_write().doc, &spec.from, &spec.to) + changes::delta_bytes(&doc.blocking_read().doc, &spec.from, &spec.to) }) .await .ok() @@ -226,9 +226,8 @@ async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Opti } BlobId::Change(hash) => { let doc = Arc::clone(&state.doc); - // Looking a change up needs the document mutably. tokio::task::spawn_blocking(move || { - changes::change_bytes(&mut doc.blocking_write().doc, &hash) + changes::change_bytes(&doc.blocking_read().doc, &hash) }) .await .ok() diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index d2bbe24..848201e 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -281,7 +281,7 @@ pub(crate) fn starting_document( /// The swarm and epoch `doc` is on. fn identify( - mut doc: Document, + doc: Document, origin: Origin, earlier_actors: Vec<ActorId>, ) -> Result<StartingDocument> { @@ -289,8 +289,7 @@ fn identify( let Some(swarm_id) = read_swarm_id(&doc) else { bail!("{what} names no swarm (no _continuum/swarm); it predates swarm IDs in documents"); }; - let epoch = - snapshot::epoch_of(&mut doc).map_err(|reason| anyhow::anyhow!("{what}: {reason}"))?; + let epoch = snapshot::epoch_of(&doc).map_err(|reason| anyhow::anyhow!("{what}: {reason}"))?; Ok(StartingDocument { doc, epoch, diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs index 3a50769..7f3c2e9 100644 --- a/src/swarm/changes.rs +++ b/src/swarm/changes.rs @@ -21,9 +21,8 @@ pub(crate) async fn run_change_publisher(state: AppState) { let before = last_heads.clone(); let doc = Arc::clone(&state.doc); - // Looking changes up needs the document mutably. let (heads, paths, conflicts) = tokio::task::spawn_blocking(move || { - let mut replica = doc.blocking_write(); + let replica = doc.blocking_read(); let heads = replica.heads().to_vec(); if heads == before { return (heads, Vec::new(), Vec::new()); @@ -31,13 +30,13 @@ pub(crate) async fn run_change_publisher(state: AppState) { // After an epoch switch the previous heads belong to the old document: // report the root as changed, which every stream is related to. - if !changes::missing(&mut replica.doc, &before).is_empty() { + if !changes::missing(&replica.doc, &before).is_empty() { return (heads, vec![Vec::new()], Vec::new()); } let mut paths = Vec::new(); let mut conflicts = Vec::new(); - for changed in store::changed(&mut replica.doc, &before, &heads, InLists::Index) { + for changed in store::changed(&replica.doc, &before, &heads, InLists::Index) { if changed.conflict { conflicts.push(changed.path.clone()); } diff --git a/src/swarm/document_file.rs b/src/swarm/document_file.rs index 435ff32..6a516a1 100644 --- a/src/swarm/document_file.rs +++ b/src/swarm/document_file.rs @@ -41,13 +41,13 @@ pub(crate) async fn flush(state: &AppState) { } Due::Append { epoch, since } => { let (heads, increment) = { - let mut replica = state.doc.write().await; + let replica = state.doc.read().await; // Switched since: the next flush rewrites the file instead. if replica.epoch != epoch { return; } let heads = replica.heads().to_vec(); - (heads, crate::document::save_since(&mut replica.doc, &since)) + (heads, crate::document::save_since(&replica.doc, &since)) }; file.append(since, heads, increment).await; } diff --git a/src/swarm/members.rs b/src/swarm/members.rs index 9885750..b3e2084 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -23,7 +23,7 @@ use tracing::{debug, info, warn}; use crate::document::meta; use crate::document::store::{self, StoreError}; -use crate::document::{Document, Transactable}; +use crate::document::{Document, ReadDoc, Transactable}; use crate::state::AppState; pub(crate) fn members_dir() -> Vec<String> { @@ -37,7 +37,7 @@ pub(crate) fn member_path(pk: &str) -> Vec<String> { } /// Public keys of all registered members, sorted. -pub(crate) fn members(doc: &Document) -> Vec<String> { +pub(crate) fn members(doc: &impl ReadDoc) -> Vec<String> { let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default(); members.sort(); members @@ -199,7 +199,7 @@ fn alive_path(pk: &str) -> Vec<String> { path } -fn millis_at(doc: &Document, path: &[String]) -> Option<u64> { +fn millis_at(doc: &impl ReadDoc, path: &[String]) -> Option<u64> { match store::read_path(doc, path) { Ok(Some(JsonValue::Number(number))) => number.as_u64(), _ => None, @@ -207,7 +207,7 @@ fn millis_at(doc: &Document, path: &[String]) -> Option<u64> { } /// When `pk` last said it is alive (Unix ms): its last renewal, or its registration. -pub(crate) fn alive_at(doc: &Document, pk: &str) -> Option<u64> { +pub(crate) fn alive_at(doc: &impl ReadDoc, pk: &str) -> Option<u64> { let mut since = member_path(pk); since.push("since".to_owned()); millis_at(doc, &alive_path(pk)).max(millis_at(doc, &since)) @@ -215,7 +215,7 @@ pub(crate) fn alive_at(doc: &Document, pk: &str) -> Option<u64> { /// 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: &Document, now_ms: u64, window: Duration) -> Vec<String> { +pub(crate) fn electorate(doc: &impl ReadDoc, now_ms: u64, window: Duration) -> Vec<String> { let oldest = now_ms.saturating_sub(window.as_millis() as u64); members(doc) .into_iter() diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index d11c3e0..1736cec 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -131,8 +131,8 @@ 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 = { - let mut replica = state.doc.write().await; - crate::document::changes::missing(&mut replica.doc, remote.heads.listed()) + let replica = state.doc.read().await; + crate::document::changes::missing(&replica.doc, remote.heads.listed()) }; if missing.is_empty() { return false; diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index 2398b09..04f41b2 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -278,8 +278,8 @@ pub(crate) fn measure_saved_size(state: &AppState) { let doc = std::sync::Arc::clone(&state.doc); // Both steps take time proportional to the document: off the async threads. let measured = match tokio::task::spawn_blocking(move || { - let mut copy = doc.blocking_read().doc.clone(); - crate::document::save(&mut copy).len() as u64 + let copy = doc.blocking_read().doc.clone(); + crate::document::save(©).len() as u64 }) .await { diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 35bc4c7..46bdef6 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -136,7 +136,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { // Label the message with the epoch of the document it was generated // from, read under the same lock. let (session, epoch, epoch_seq, body) = { - let mut replica = doc.blocking_write(); + let replica = doc.blocking_read(); let mut peer_state = psm.blocking_lock(); let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq); // Under the same locks as the epoch label: the message belongs to @@ -144,7 +144,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let message = peer_state .sync - .generate(&mut replica.doc, Instant::now(), bytes_before); + .generate(&replica.doc, Instant::now(), bytes_before); let body = message.map(|payload| { if payload.0.len() <= delta_threshold { return transport::SyncBody::Message(payload); @@ -390,7 +390,7 @@ pub(crate) async fn handle_sync( // doesn't count; the writer's still does. let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms; let delays: Vec<i64> = - crate::document::changes::commit_times(&mut replica.doc, &heads_before) + crate::document::changes::commit_times(&replica.doc, &heads_before) .into_iter() .map(|time| now_ms - time) .collect(); @@ -403,7 +403,7 @@ pub(crate) async fn handle_sync( // catching up has them, and so does a ready peer whose document is empty // (a misconfigured second swarm, or a genesis node in its first seconds, // before it registered itself). - check_caught_up && from_ready && peer_state.sync.have_all_theirs(&mut replica.doc), + check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc), changed, ))) }) @@ -501,7 +501,7 @@ async fn download_and_apply_delta( let heads_before = replica.heads().to_vec(); crate::document::changes::load_delta(&mut replica.doc, &bytes)?; replica.refresh_heads(); - let missing = crate::document::changes::missing(&mut replica.doc, &spec.to).len(); + let missing = crate::document::changes::missing(&replica.doc, &spec.to).len(); if missing > 0 { bail!("{missing} of the offered heads are still missing after loading"); } |
