diff options
| -rw-r--r-- | docs/api.md | 3 | ||||
| -rw-r--r-- | src/document/file.rs | 28 | ||||
| -rw-r--r-- | src/document/signatures.rs | 259 | ||||
| -rw-r--r-- | src/node.rs | 6 | ||||
| -rw-r--r-- | src/state/mod.rs | 10 | ||||
| -rw-r--r-- | src/state/peer.rs | 5 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 41 | ||||
| -rw-r--r-- | src/transport/mod.rs | 4 | ||||
| -rw-r--r-- | src/transport/wire.rs | 3 |
9 files changed, 314 insertions, 45 deletions
diff --git a/docs/api.md b/docs/api.md index 8f5b89c..cc3cc05 100644 --- a/docs/api.md +++ b/docs/api.md @@ -261,6 +261,9 @@ by the node's next signed one: with `--sign-every 1s` (the default), the node si latest change at most a second after it if nothing signed it before, so every change is vouched for within a second, at one signature a second at most; `--sign-every all` signs every change as it is made, and `--sign-every none` signs only those above. +Signatures travel with sync messages: a node passes on, for each node's writes, the +latest signature it checked, and checks one it receives once it holds the change (a +signature costs about 30 µs to check, and one per node and run vouches for the rest). The signatures a node has are kept in its document file (`--document-file`) with the document. diff --git a/src/document/file.rs b/src/document/file.rs index 33f7934..2d588ee 100644 --- a/src/document/file.rs +++ b/src/document/file.rs @@ -6,8 +6,8 @@ //! each a little-endian `u32` length and that many bytes, the first of them its kind //! (`Record`): first a full save of the document, then increments (the changes after //! the heads written before), and the signatures of its changes -//! (`document::signatures`), those the file holds none of yet, each a change hash -//! (32 bytes) and its signature (64). A record cut short by a crash is dropped when +//! (`document::signatures`), those the file holds none of yet, as +//! `signatures::pack` packs them. A record cut short by a crash is dropped when //! reading. A file of `continuum-document/1` has records of the document alone, //! without their kind: a full save, then increments. //! @@ -46,9 +46,6 @@ enum Record { Signatures = 2, } -/// A signature as a record holds it: the change's hash, then the signature. -const SIGNATURE_BYTES: usize = 32 + 64; - #[derive(Debug, Default, Serialize, Deserialize)] struct Header { /// Hex actor IDs. @@ -146,16 +143,10 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> { (Record::Increment, Some(doc)) => { doc.load_incremental(record).context("an increment")?; } - (Record::Signatures, Some(_)) => { - if record.len() % SIGNATURE_BYTES != 0 { - bail!("signatures of {} bytes", record.len()); - } - for entry in record.chunks_exact(SIGNATURE_BYTES) { - let (hash, signature) = entry.split_at(32); - let hash = ChangeHash(hash.try_into().expect("32 bytes")); - signatures.push((hash, signature.try_into().expect("64 bytes"))); - } - } + (Record::Signatures, Some(_)) => signatures.extend( + super::signatures::unpack(record) + .ok_or_else(|| anyhow!("signatures of {} bytes", record.len()))?, + ), (_, None) => bail!("a record before the full save"), } records = after; @@ -184,12 +175,7 @@ fn signatures_record(signatures: &[(ChangeHash, [u8; 64])]) -> Vec<u8> { if signatures.is_empty() { return Vec::new(); } - let mut bytes = Vec::with_capacity(signatures.len() * SIGNATURE_BYTES); - for (hash, signature) in signatures { - bytes.extend_from_slice(&hash.0); - bytes.extend_from_slice(signature); - } - record(Record::Signatures, &bytes) + record(Record::Signatures, &super::signatures::pack(signatures)) } /// A whole file: header, one full save, and the signatures. diff --git a/src/document/signatures.rs b/src/document/signatures.rs index c8af89a..a6a30f7 100644 --- a/src/document/signatures.rs +++ b/src/document/signatures.rs @@ -11,6 +11,12 @@ //! change left unsigned is vouched for by the node's next signed one, which, as //! `--sign-every` says, is its latest change, signed a while after it if nothing signed //! it before (`SignEvery`), or every change at once. +//! +//! Signatures travel with sync messages (`SyncEnvelope::signatures`): each actor's +//! latest one a node verified, those a peer wasn't sent yet (`newer_than`). A received +//! signature is checked once its change is there (`offer`), and kept if it vouches for +//! more of its actor's changes than any before; one of a change not there yet waits +//! (`pending`, a bounded number). use std::collections::HashMap; @@ -150,6 +156,43 @@ impl<'de> serde::Deserialize<'de> for SignEvery { } } +/// Signatures a sync message carries at most (`newer_than`). +pub(crate) const MAX_PER_MESSAGE: usize = 256; +/// A signature as messages and the document file carry it (`pack`): the change's +/// hash (32 bytes), then the signature (64). +pub(crate) const PACKED: usize = 32 + 64; + +/// `signatures` as bytes, one after another (`PACKED`). +pub(crate) fn pack(signatures: &[(ChangeHash, [u8; 64])]) -> Vec<u8> { + let mut bytes = Vec::with_capacity(signatures.len() * PACKED); + for (hash, signature) in signatures { + bytes.extend_from_slice(&hash.0); + bytes.extend_from_slice(signature); + } + bytes +} + +/// What `pack` made; None if `bytes` aren't a whole number of signatures. +pub(crate) fn unpack(bytes: &[u8]) -> Option<Vec<(ChangeHash, [u8; 64])>> { + if !bytes.len().is_multiple_of(PACKED) { + return None; + } + Some( + bytes + .chunks_exact(PACKED) + .map(|entry| { + let (hash, signature) = entry.split_at(32); + ( + ChangeHash(hash.try_into().expect("32 bytes")), + signature.try_into().expect("64 bytes"), + ) + }) + .collect(), + ) +} +/// Received signatures of changes not there yet, kept at most (`offer`). +const MAX_PENDING: usize = 4096; + /// A document's signatures, by change. #[derive(Debug, Default)] pub(crate) struct Signatures { @@ -159,15 +202,48 @@ pub(crate) struct Signatures { order: Vec<ChangeHash>, /// This node's latest change, if no signature vouches for it yet (`SignEvery`). unsigned: Option<ChangeHash>, + /// By actor: its latest change a signature verified, so its changes up to it. + latest: HashMap<super::ActorId, Latest>, + /// Counts each change of `latest`: a peer is sent those after the last it was + /// (`newer_than`). + stamp: u64, + /// Received signatures of changes the document doesn't hold yet. + pending: HashMap<ChangeHash, [u8; 64]>, +} + +/// An actor's latest change a signature verified. +#[derive(Debug, Clone, Copy)] +struct Latest { + seq: u64, + hash: ChangeHash, + /// `Signatures::stamp` when it became the latest. + stamp: u64, } impl Signatures { - /// Signatures kept before, as `since` gave them (the document file). - pub(crate) fn from_kept(kept: Vec<(ChangeHash, [u8; 64])>) -> Self { + /// Signatures kept before, as `since` gave them (the document file), of `doc`'s + /// changes: each actor's latest checked (`check`), as it vouches for the others. + pub(crate) fn from_kept(doc: &Document, kept: Vec<(ChangeHash, [u8; 64])>) -> Self { let mut signatures = Self::default(); + let mut by_actor: HashMap<super::ActorId, (u64, ChangeHash)> = HashMap::new(); for (hash, signature) in kept { + if let Some(meta) = doc.get_change_meta_by_hash(&hash) { + let entry = by_actor + .entry(meta.actor.into_owned()) + .or_insert((meta.seq, hash)); + if meta.seq > entry.0 { + *entry = (meta.seq, hash); + } + } signatures.insert(hash, signature); } + for (hash, signature) in by_actor + .into_values() + .map(|(_, hash)| (hash, signatures.by_change[&hash])) + .collect::<Vec<_>>() + { + signatures.vouch(doc, hash, &signature); + } signatures } @@ -177,6 +253,101 @@ impl Signatures { } } + /// `signature` of `hash`, if valid (`check`), as its actor's latest verified change + /// where it is later than the one before; whether it is. + fn vouch(&mut self, doc: &Document, hash: ChangeHash, signature: &[u8; 64]) -> bool { + // Not checked where it couldn't vouch for more: a signature costs about 30 µs. + let Some(meta) = doc.get_change_meta_by_hash(&hash) else { + return false; + }; + if self + .latest + .get(meta.actor.as_ref()) + .is_some_and(|latest| latest.seq >= meta.seq) + { + return false; + } + let Some((actor, seq)) = check(doc, &hash, signature) else { + return false; + }; + if self + .latest + .get(&actor) + .is_some_and(|latest| latest.seq >= seq) + { + return false; + } + self.stamp += 1; + self.latest.insert( + actor, + Latest { + seq, + hash, + stamp: self.stamp, + }, + ); + true + } + + /// Received signatures, from a peer: each checked once its change is in `doc`, kept + /// if it vouches for more of its actor's changes than any before; those of changes + /// not there yet wait, a bounded number. Then those waiting whose changes came + /// meanwhile. How many were kept. + pub(crate) fn offer(&mut self, doc: &Document, received: Vec<(ChangeHash, [u8; 64])>) -> usize { + let mut kept = 0; + for (hash, signature) in received { + if doc.get_change_meta_by_hash(&hash).is_none() { + if self.pending.len() < MAX_PENDING { + self.pending.insert(hash, signature); + } + } else if self.vouch(doc, hash, &signature) { + self.insert(hash, signature); + kept += 1; + } + } + let arrived: Vec<ChangeHash> = self + .pending + .keys() + .filter(|hash| doc.get_change_meta_by_hash(hash).is_some()) + .copied() + .collect(); + for hash in arrived { + if let Some(signature) = self.pending.remove(&hash) + && self.vouch(doc, hash, &signature) + { + self.insert(hash, signature); + kept += 1; + } + } + kept + } + + /// Each actor's latest verified signature that became so after `stamp`, oldest first, + /// `MAX_PER_MESSAGE` at most; with the stamp to take up from next time. + pub(crate) fn newer_than(&self, stamp: u64) -> (Vec<(ChangeHash, [u8; 64])>, u64) { + let mut newer: Vec<&Latest> = self + .latest + .values() + .filter(|latest| latest.stamp > stamp) + .collect(); + newer.sort_unstable_by_key(|latest| latest.stamp); + newer.truncate(MAX_PER_MESSAGE); + let next = newer.last().map_or(stamp, |latest| latest.stamp); + let signatures = newer + .into_iter() + .map(|latest| (latest.hash, self.by_change[&latest.hash])) + .collect(); + (signatures, next) + } + + /// Whether a signature vouches for `actor`'s change `seq`. + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn verified(&self, actor: &super::ActorId, seq: u64) -> bool { + self.latest + .get(actor) + .is_some_and(|latest| latest.seq >= seq) + } + /// How many signatures there are. pub(crate) fn count(&self) -> usize { self.order.len() @@ -192,10 +363,17 @@ impl Signatures { .collect() } - /// Signs `hash`, this node's latest change, with `key` and keeps the signature: it - /// vouches for every change of the node's before it too. - pub(crate) fn sign(&mut self, hash: ChangeHash, key: &crate::identity::NodeKey) { - self.insert(hash, key.sign(&message(&hash))); + /// Signs `hash`, this node's latest change in `doc`, with `key` and keeps the + /// signature: it vouches for every change of the node's before it too. + pub(crate) fn sign( + &mut self, + doc: &Document, + hash: ChangeHash, + key: &crate::identity::NodeKey, + ) { + let signature = key.sign(&message(&hash)); + self.insert(hash, signature); + self.vouch(doc, hash, &signature); self.unsigned = None; } @@ -209,12 +387,12 @@ impl Signatures { self.unsigned = Some(hash); } - /// Signs this node's latest change if no signature vouches for it yet; whether it - /// did. - pub(crate) fn sign_latest(&mut self, key: &crate::identity::NodeKey) -> bool { + /// Signs this node's latest change in `doc` if no signature vouches for it yet; + /// whether it did. + pub(crate) fn sign_latest(&mut self, doc: &Document, key: &crate::identity::NodeKey) -> bool { match self.unsigned { Some(hash) => { - self.sign(hash, key); + self.sign(doc, hash, key); true } None => false, @@ -226,7 +404,6 @@ impl Signatures { /// seq, committed by the node its author names (its node ID, whose key it is). None if /// `doc` doesn't hold the change, its author isn't a node ID, or the signature isn't /// that node's over it. -#[cfg_attr(not(test), allow(dead_code))] pub(crate) fn check( doc: &Document, hash: &ChangeHash, @@ -304,25 +481,73 @@ mod tests { #[test] fn the_latest_change_is_signed_later() { let key = NodeKey::from_seed([1; 32]); + let doc = Document::new(); let mut signatures = Signatures::default(); - assert!(!signatures.sign_latest(&key)); + assert!(!signatures.sign_latest(&doc, &key)); let a = ChangeHash([1; 32]); let b = ChangeHash([2; 32]); signatures.left_unsigned(a); - assert!(signatures.sign_latest(&key)); + assert!(signatures.sign_latest(&doc, &key)); assert!(signatures.by_change.contains_key(&a)); - assert!(!signatures.sign_latest(&key)); + assert!(!signatures.sign_latest(&doc, &key)); signatures.left_unsigned(a); - signatures.sign(b, &key); - assert!(!signatures.sign_latest(&key)); + signatures.sign(&doc, b, &key); + assert!(!signatures.sign_latest(&doc, &key)); // Kept in the order they came, each once. assert_eq!(signatures.count(), 2); assert_eq!(signatures.since(1)[0].0, b); assert!(signatures.since(5).is_empty()); - let again = Signatures::from_kept(signatures.since(0)); + let again = Signatures::from_kept(&doc, signatures.since(0)); assert_eq!(again.since(0), signatures.since(0)); } + /// Received signatures: one vouches for its actor's changes up to it, an earlier one + /// adds nothing, one of a change not there yet waits for it, a forged one is + /// dropped; each actor's latest goes on to peers once (`newer_than`). + #[test] + fn received_signatures_vouch_for_their_actors_changes() { + let key = NodeKey::from_seed([1; 32]); + let other = NodeKey::from_seed([2; 32]); + let mut theirs = Document::new(); + write(&mut theirs, &key.id(), &["a"], json!(1)); + let first = theirs.get_heads()[0]; + write(&mut theirs, &key.id(), &["b"], json!(2)); + let second = theirs.get_heads()[0]; + let actor = theirs.get_actor().clone(); + let signed = |hash: ChangeHash| (hash, key.sign(&message(&hash))); + + let mut ours = Document::new(); + ours.load_incremental(&theirs.save_after(&[])).unwrap(); + let mut signatures = Signatures::default(); + assert!(!signatures.verified(&actor, 1)); + assert_eq!(signatures.offer(&ours, vec![signed(second)]), 1); + assert!(signatures.verified(&actor, 1) && signatures.verified(&actor, 2)); + // An earlier one vouches for nothing more; a forged one for nothing. + assert_eq!(signatures.offer(&ours, vec![signed(first)]), 0); + assert_eq!( + signatures.offer(&ours, vec![(second, other.sign(&message(&second)))]), + 0 + ); + let (newer, stamp) = signatures.newer_than(0); + assert_eq!(newer, vec![signed(second)]); + assert!(signatures.newer_than(stamp).0.is_empty()); + + // A change not there yet: its signature waits for it. + write(&mut theirs, &key.id(), &["c"], json!(3)); + let third = theirs.get_heads()[0]; + assert_eq!(signatures.offer(&ours, vec![signed(third)]), 0); + assert!(!signatures.verified(&actor, 3)); + ours.load_incremental(&theirs.save_after(&[second])) + .unwrap(); + assert_eq!(signatures.offer(&ours, Vec::new()), 1); + assert!(signatures.verified(&actor, 3)); + assert_eq!(signatures.newer_than(stamp).0, vec![signed(third)]); + + // Kept in a file and read again: verified as before. + let again = Signatures::from_kept(&ours, signatures.since(0)); + assert!(again.verified(&actor, 3)); + } + /// Protocol data, listed paths (within or above them) and asked-for writes are /// signed; others aren't. #[test] diff --git a/src/node.rs b/src/node.rs index 32a82b2..8cf327e 100644 --- a/src/node.rs +++ b/src/node.rs @@ -113,8 +113,10 @@ impl Node { }; let mut replica = Replica::new(config.initial_epoch, config.initial_doc); - replica.signatures = - crate::document::signatures::Signatures::from_kept(config.initial_signatures); + replica.signatures = crate::document::signatures::Signatures::from_kept( + &replica.doc, + config.initial_signatures, + ); let stamp = replica.stamps(); let state = AppState { doc: Arc::new(crate::document::lock::DocumentLock::new(replica)), diff --git a/src/state/mod.rs b/src/state/mod.rs index 1fc8ea8..ff19124 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -532,7 +532,8 @@ impl AppState { && let [hash] = after[..] { if state.sign_every == SignEvery::Change || sign.wanted(&replica.doc) { - replica.signatures.sign(hash, key); + let replica = &mut *replica; + replica.signatures.sign(&replica.doc, hash, key); } else { replica.signatures.left_unsigned(hash); } @@ -565,8 +566,11 @@ impl AppState { let key = Arc::clone(&key); let _ = tokio::task::spawn_blocking(move || { // Looked at first under the read lock: mostly there is nothing to sign. - state.doc.blocking_read().signatures.pending() - && state.doc.blocking_write().signatures.sign_latest(&key) + state.doc.blocking_read().signatures.pending() && { + let mut replica = state.doc.blocking_write(); + let replica = &mut *replica; + replica.signatures.sign_latest(&replica.doc, &key) + } }) .await; } diff --git a/src/state/peer.rs b/src/state/peer.rs index 7000386..e0b161d 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -98,6 +98,9 @@ pub(crate) struct PeerState { /// Everything about the current Automerge sync with this peer; a reset replaces it. pub(crate) sync: SyncSession, + /// The signatures of changes it was sent, up to this stamp + /// (`Signatures::newer_than`); a reset sends them again. + pub(crate) signatures_sent: u64, pub(crate) liveness: Liveness, pub(crate) added_at: Instant, @@ -300,6 +303,7 @@ impl PeerState { hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX), probe_backoff: timing.probe_backoff(), sync: SyncSession::new(timing.sync_reply_backoff()), + signatures_sent: 0, liveness: Liveness::Unknown, added_at: Instant::now(), last_seen: None, @@ -369,6 +373,7 @@ impl PeerState { let mut reply_backoff = self.sync.reply_backoff; reply_backoff.reset(); self.sync = SyncSession::new(reply_backoff); + self.signatures_sent = 0; // Said over the session that ended. self.reaches = None; self.round_trips = None; diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 2cbfa11..ed3679d 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -146,7 +146,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { move || -> (u64, Result<Option<transport::EncodedMessage>>) { // Label the message with the epoch of the document it was generated // from, read under the same lock. - let (session, epoch, epoch_seq, body, writer, delivered) = { + let (session, epoch, epoch_seq, body, writer, delivered, signatures) = { let replica = doc.blocking_read(); let mut peer_state = psm.blocking_lock(); let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq); @@ -258,6 +258,15 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { _ => {} } } + // With what it carries, the signatures the peer wasn't sent yet. + let signatures = if body.is_some() { + let (signatures, sent) = + replica.signatures.newer_than(peer_state.signatures_sent); + peer_state.signatures_sent = sent; + crate::document::signatures::pack(&signatures) + } else { + Vec::new() + }; ( peer_state.sync.id(), epoch, @@ -265,6 +274,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { body, writer, delivered, + signatures, ) }; let Some(body) = body else { @@ -304,6 +314,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { writer, delivered, sent: clock.now_millis(), + signatures: serde_bytes::ByteBuf::from(signatures), }; let encoded = transport .encode(&peer_owned, &ParsedMessage::Sync(envelope)) @@ -464,6 +475,11 @@ pub(crate) async fn handle_sync( return Ok(()); }; let epoch_label = epoch; + // Signatures of changes (`document::signatures`): a malformed list, or more than any + // sender sends, counts for nothing. + let signatures = crate::document::signatures::unpack(&envelope.signatures) + .filter(|signatures| signatures.len() <= crate::document::signatures::MAX_PER_MESSAGE) + .unwrap_or_default(); let from_ready = envelope.ready; let writer = envelope.writer; let delivered = envelope.delivered; @@ -496,6 +512,19 @@ pub(crate) async fn handle_sync( peer_state.last_sync_rx = Some(Instant::now()); peer_state.sync.answered_by_offer(); } + // Of changes the blob brings: they wait for them (`Signatures::offer`). + if !signatures.is_empty() { + let doc = Arc::clone(&state.doc); + tokio::task::spawn_blocking(move || { + let mut replica = doc.blocking_write(); + if replica.epoch == epoch { + let replica = &mut *replica; + replica.signatures.offer(&replica.doc, signatures); + } + }) + .await + .context("Signatures task panicked")?; + } fetch_delta(state, peer, epoch, spec); return Ok(()); } @@ -601,6 +630,11 @@ pub(crate) async fn handle_sync( stats.asked_known_elsewhere(asked); peer_state.asked_known += (asked.0 + asked.1) as u64; replica.refresh_heads(); + // The signatures it sent, and those waiting for what it brought. + if !signatures.is_empty() || replica.heads() != heads_before.as_slice() { + let replica = &mut *replica; + replica.signatures.offer(&replica.doc, signatures); + } let heads_after = replica.heads().to_vec(); if delivered { // Level with the sender now, which takes the message as delivered: @@ -831,6 +865,11 @@ 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(); + // Signatures waiting for what it brought (`Signatures::offer`). + { + let replica = &mut *replica; + replica.signatures.offer(&replica.doc, Vec::new()); + } 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"); diff --git a/src/transport/mod.rs b/src/transport/mod.rs index f5b2db1..35e1e5f 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -704,6 +704,10 @@ pub(crate) struct SyncEnvelope { /// When the sender generated it, by its clock: the receiver takes how long it took /// to come (`swarm::hop`). pub(crate) sent: crate::clock::UnixMillis, + /// Signatures of changes (`document::signatures`): each actor's latest the sender + /// verified that it didn't send this peer yet, `signatures::MAX_PER_MESSAGE` at + /// most, as `signatures::pack` packs them. + pub(crate) signatures: serde_bytes::ByteBuf, } /// An epoch, as sync messages name it: its generation and the first bytes of its ID, diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 8ead9b0..97f8819 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 = 39; +pub(crate) const VERSION: u8 = 40; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -479,6 +479,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { writer: Some(7), delivered: true, sent: crate::clock::UnixMillis(8), + signatures: serde_bytes::ByteBuf::from(vec![9; 96]), }), ParsedMessage::Status(status.clone()), ParsedMessage::StatusReply(status), |