From 5b13f5e9d83a11b8a60f2d4711065360ba20f70c Mon Sep 17 00:00:00 2001 From: tv Date: Thu, 12 Nov 2026 00:44:56 +0000 Subject: sync: a delta's signatures as a blob of their own A sync message larger than the threshold for its peer (16 chunks, lower on a slow link or after a message to it failed whole) goes as a delta: the peer fetches its changes in chunks, from any node holding them. Its signatures, each writer's latest with changes in it, went in the envelope offering it: up to 65536 of them, 6 MiB, in one message, the size the delta exists to avoid. Nor did an Automerge message count its signatures (one per writer in a whole document pulled) towards the threshold. A delta's signatures are now a blob of their own: the offerer packs and keeps them (Blobs::keep, the latest 64), and the spec names them by BLAKE3 hash and size (DeltaSpec::signatures, None where no signature vouches for its changes). They are served as content (BlobId::Content), so their hash checks them. The receiver fetches them from the offerer alone, as another node holding the changes holds signatures of its own, before the changes, and offers them under the write lock just before judging the changes, so they wait for them as before. The envelope of a delta carries none. An Automerge message's signatures count towards its size: past the threshold, it goes as a delta. Wire version 49. --- docs/api.md | 5 +++- src/swarm/blob.rs | 39 ++++++++++++++++++++++++++++++- src/swarm/sync.rs | 63 ++++++++++++++++++++++++++++++++++++++++++--------- src/transport/mod.rs | 31 +++++++++++++++++++++---- src/transport/wire.rs | 6 ++++- 5 files changed, 125 insertions(+), 19 deletions(-) diff --git a/docs/api.md b/docs/api.md index a4b76b0..2665306 100644 --- a/docs/api.md +++ b/docs/api.md @@ -292,7 +292,10 @@ vouched for as the first time; and a node checks one it receives once it holds t change (a signature costs about 30 µs to check, and one vouches for its node's changes before it). A message carrying no changes carries no signatures; one carrying more than 256 signed changes (a whole document pulled) carries each writer's latest among -them instead, and a delta each writer's latest with changes in it. A writer is a node's +them instead. Its signatures count towards a message's size, past which its changes are +offered as a delta (fetched in chunks, from any node holding them); a delta's +signatures, each writer's latest with changes in it, are a blob of their own, fetched +from the node that offered it before the changes. A writer is a node's run: each start, rejoin or drop makes a fresh one. A node signs the latest change of each of its runs that nothing vouches for as it starts, before anything leaves, and whenever its document is swapped (a drop, a switch, a rejoin): changes it kept diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index 588868f..ec75bb5 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -100,8 +100,18 @@ pub(crate) struct Blobs { /// The peer whose offered delta is being downloaded (`swarm::sync`), and the peers /// whose offers came meanwhile (`start_delta`). deltas: Mutex<(Option, HashSet)>, + /// The signatures of deltas this node offered (`DeltaSpec::signatures`), served as + /// content by their hash: the latest `KEPT`. + kept: Mutex>, } +/// Bytes kept to serve as content, by their BLAKE3 hash (`Blobs::keep`). +type Kept = ([u8; 32], Arc>); + +/// Deltas' signatures kept to serve (`Blobs::keep`): a peer fetches them as it takes +/// the delta, one offered delta at a time. +const KEPT: usize = 64; + /// A sync message larger than this many chunks is sent as a delta blob instead /// (`swarm::sync`): below it, a blob costs more round trips than it saves; above it, a /// lost message costs too much to send again whole, and a slow link may not deliver it @@ -131,7 +141,31 @@ impl Blobs { cache: Mutex::new(None), fetches: Mutex::new(HashMap::new()), deltas: Mutex::new((None, HashSet::new())), + kept: Mutex::new(VecDeque::new()), + } + } + + /// Signatures for a delta this node offers, packed (`signatures::pack`): kept to + /// serve as content, and named for the spec. + pub(crate) fn keep(&self, bytes: Vec) -> crate::transport::DeltaSignatures { + let hash = *blake3::hash(&bytes).as_bytes(); + let size = bytes.len() as u64; + let mut kept = self.kept.lock_or_recover(); + if !kept.iter().any(|(kept, _)| *kept == hash) { + kept.push_back((hash, Arc::new(bytes))); + while kept.len() > KEPT { + kept.pop_front(); + } } + crate::transport::DeltaSignatures { hash, size } + } + + fn kept(&self, hash: &[u8; 32]) -> Option>> { + self.kept + .lock_or_recover() + .iter() + .find(|(kept, _)| kept == hash) + .map(|(_, bytes)| Arc::clone(bytes)) } /// The largest sync message sent as it is (`DELTA_THRESHOLD_CHUNKS`). @@ -293,7 +327,10 @@ async fn prepare(state: &AppState, id: BlobId, delta: Option) -> Opti .flatten() .map(Arc::new)? } - BlobId::Content(hash) => state.remake.content(&hash)?, + BlobId::Content(hash) => state + .remake + .content(&hash) + .or_else(|| state.blobs.kept(&hash))?, }; let chunk_size = state.blobs.chunk_size; let prepared = tokio::task::spawn_blocking(move || { diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 5550c27..528fe47 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -217,7 +217,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { ) }) .flatten(); - let body = if let Some((heads, changes)) = pushed { + // A message's signatures, where finding them was part of deciding it. + let mut carried_signatures = None; + let mut body = if let Some((heads, changes)) = pushed { Some(transport::SyncBody::Changes { heads, changes: changes @@ -239,7 +241,16 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { .sync .generate(&replica.doc, Instant::now(), bytes_before); message.and_then(|payload| { - if payload.0.len() <= delta_threshold { + // Its signatures count towards its size: a whole document + // pulled carries one per writer. + let signatures = replica + .signatures + .for_changes(&replica.doc, peer_state.sync.carried()); + if payload.0.len() + + signatures.len() * crate::document::signatures::PACKED + <= delta_threshold + { + carried_signatures = Some(signatures); return Some(transport::SyncBody::Message(payload)); } // Too large for one message: all or nothing, and on a slow link @@ -300,15 +311,25 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { // With changes, the signatures of those it carries: they go where // changes go, along the trees, so a peer that got these changes // elsewhere got the signatures with them; and changes sent again go - // vouched for as the first time. A delta, of each writer with changes - // in it, its latest. A message carrying no changes takes none. - let signatures = match &body { + // vouched for as the first time. A delta's, of each writer with + // changes in it its latest, go as a blob of their own, from this node + // (`Blobs::keep`): the peer fetches them with the delta. A message + // carrying no changes takes none. + let signatures = match &mut body { Some(transport::SyncBody::Delta(spec)) => { - replica.signatures.for_delta(&replica.doc, &spec.from) + let signatures = replica.signatures.for_delta(&replica.doc, &spec.from); + spec.signatures = (!signatures.is_empty()).then(|| { + signer + .blobs + .keep(crate::document::signatures::pack(&signatures)) + }); + Vec::new() } - Some(_) => replica - .signatures - .for_changes(&replica.doc, peer_state.sync.carried()), + Some(_) => carried_signatures.unwrap_or_else(|| { + replica + .signatures + .for_changes(&replica.doc, peer_state.sync.carried()) + }), None => Vec::new(), }; let signatures = crate::document::signatures::pack(&signatures); @@ -951,6 +972,24 @@ async fn download_and_apply_delta( return Ok(false); } } + // Its signatures first, from the offerer alone: another node holding the changes + // holds signatures of its own, not these. + let signatures = match spec.signatures { + Some(at) => { + let most = (crate::document::signatures::MAX_PER_MESSAGE + * crate::document::signatures::PACKED) as u64; + if at.size > most { + bail!("the offered changes' signatures are larger than any message's"); + } + let bytes = + crate::swarm::blob::fetch_content(state, at.hash, at.size, vec![peer.to_owned()]) + .await + .context("Fetching the offered changes' signatures")?; + crate::document::signatures::unpack(&bytes) + .context("the offered changes' signatures aren't whole")? + } + None => Vec::new(), + }; info!(peer = %peer, to = spec.to.len(), "Fetching changes offered as a blob"); let started = Instant::now(); let bytes = crate::swarm::blob::fetch_delta(state, &spec, peers).await?; @@ -977,10 +1016,12 @@ async fn download_and_apply_delta( } let heads_before = replica.heads().to_vec(); { - // Judged first (`access::screen`), whether this document has a policy yet or - // not. + // Its signatures wait for its changes (`Signatures::offer`), each checked as + // they are judged (`access::screen`), whether this document has a policy yet + // or not. let epoch_seq = replica.epoch_seq; let replica = &mut *replica; + replica.signatures.offer(&replica.doc, signatures); let outcome = crate::access::screen::apply_bytes( &mut replica.doc, epoch, diff --git a/src/transport/mod.rs b/src/transport/mod.rs index fcdf56e..2778ed7 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -734,9 +734,10 @@ pub(crate) struct SyncEnvelope { /// to come (`swarm::hop`). pub(crate) sent: crate::clock::UnixMillis, /// Signatures of changes (`document::signatures`): of the changes the message - /// carries, each writer's latest among them past a few hundred; of a delta, each - /// writer's latest with changes in it (`Signatures::for_changes`, `for_delta`), - /// `signatures::MAX_PER_MESSAGE` at most, as `signatures::pack` packs them. + /// carries, each writer's latest among them past a few hundred + /// (`Signatures::for_changes`), `signatures::MAX_PER_MESSAGE` at most, as + /// `signatures::pack` packs them. A delta's are a blob of their own + /// (`DeltaSpec::signatures`). pub(crate) signatures: serde_bytes::ByteBuf, } @@ -815,6 +816,21 @@ pub(crate) struct DeltaSpec { /// Sorted. #[serde(with = "crate::transport::wire::codec::change_hashes")] pub(crate) to: Vec, + /// The signatures vouching for its changes, where any do: a blob of their own + /// (`BlobId::Content`), from the node that offered it. Each node holding `to` serves + /// the same changes, but its own signatures. + pub(crate) signatures: Option, +} + +/// A delta's signatures as a blob (`DeltaSpec::signatures`): each writer's latest with +/// changes in it (`Signatures::for_delta`), as `signatures::pack` packs them, +/// `signatures::MAX_PER_MESSAGE` at most. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct DeltaSignatures { + /// BLAKE3, as `BlobId::Content` names them. + #[serde(with = "crate::transport::wire::codec::bytes32")] + pub(crate) hash: [u8; 32], + pub(crate) size: u64, } impl DeltaSpec { @@ -823,10 +839,15 @@ impl DeltaSpec { from.dedup(); to.sort_unstable(); to.dedup(); - Self { from, to } + Self { + from, + to, + signatures: None, + } } - /// Names the spec as a blob: SHA-256 over both lists. + /// Names the spec as a blob: SHA-256 over both lists (its changes, the same from + /// every node holding them). pub(crate) fn id(&self) -> [u8; 32] { let mut hasher = Sha256::new(); hasher.update(b"continuum/delta/1"); diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 7fe8983..3b67a74 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 48; +pub(crate) const VERSION: u8 = 49; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -480,6 +480,10 @@ pub(crate) fn samples() -> Vec { body: SyncBody::Delta(DeltaSpec { from: Vec::new(), to: vec![ChangeHash([2; 32])], + signatures: Some(crate::transport::DeltaSignatures { + hash: [3; 32], + size: 96, + }), }), ready: true, writer: Some(7), -- cgit v1.3.1