diff options
| -rw-r--r-- | docs/api.md | 5 | ||||
| -rw-r--r-- | src/swarm/blob.rs | 39 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 63 | ||||
| -rw-r--r-- | src/transport/mod.rs | 31 | ||||
| -rw-r--r-- | 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<String>, HashSet<String>)>, + /// The signatures of deltas this node offered (`DeltaSpec::signatures`), served as + /// content by their hash: the latest `KEPT`. + kept: Mutex<VecDeque<Kept>>, } +/// Bytes kept to serve as content, by their BLAKE3 hash (`Blobs::keep`). +type Kept = ([u8; 32], Arc<Vec<u8>>); + +/// 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<u8>) -> 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<Arc<Vec<u8>>> { + 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<DeltaSpec>) -> 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<ChangeHash>, + /// 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<DeltaSignatures>, +} + +/// 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<ParsedMessage> { 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), |