From b33f4235fe16b8160afb60d0bd743f4c81d9d0ab Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 8 Nov 2026 23:03:42 +0000 Subject: sync: a message carries the signatures of its changes A sync message carried the signatures its peer wasn't sent yet, by a cursor per peer. Changes sent again, as after a message taken as delivered and lost on the way, went without them: the peer was past them. Under a policy requiring signatures they waited until the writer's next signed change, or the next session. A message now carries the signatures of the changes it carries (Signatures::for_changes), whatever the peer was sent before: changes sent again arrive vouched for as the first time. Past 256 (a message carrying the batches of many writers), each writer's latest instead, as a delta carries. The cursor goes, and with it resetting it after a failed send or a new session. --- src/document/signatures.rs | 61 +++++++++++++++++++++++++++++----------------- 1 file changed, 39 insertions(+), 22 deletions(-) (limited to 'src/document/signatures.rs') diff --git a/src/document/signatures.rs b/src/document/signatures.rs index f2809fe..3001690 100644 --- a/src/document/signatures.rs +++ b/src/document/signatures.rs @@ -15,8 +15,8 @@ //! `--sign-every` says (`SignEvery`), is its latest change, signed as it leaves in a //! sync message (`sent`), or every change at once (`all`). //! -//! 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 +//! Signatures travel with the changes they are of, in the sync messages that carry them +//! (`SyncEnvelope::signatures`, `for_changes`). 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: the oldest go). @@ -147,7 +147,7 @@ impl<'de> serde::Deserialize<'de> for SignEvery { } } -/// Signatures a sync message carries at most (`newer_than`). +/// Signatures a sync message carries at most (`for_changes`, `latest_each`). 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). @@ -233,8 +233,7 @@ pub(crate) struct Signatures { unsigned: Option, /// By actor: its latest change a signature verified, so its changes up to it. latest: HashMap, - /// Counts each change of `latest`: a peer is sent those after the last it was - /// (`newer_than`). + /// Counts each change of `latest`: which became so last (`latest_each`). stamp: u64, /// Received signatures of changes the document doesn't hold yet. pending: Pending, @@ -365,22 +364,36 @@ impl Signatures { 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 + /// Each actor's latest verified signature, those that became so last first, + /// `MAX_PER_MESSAGE` at most: for changes of any actor, as a delta carries. + pub(crate) fn latest_each(&self) -> Vec<(ChangeHash, [u8; 64])> { + let mut latest: Vec<&Latest> = self.latest.values().collect(); + latest.sort_unstable_by_key(|latest| std::cmp::Reverse(latest.stamp)); + latest.truncate(MAX_PER_MESSAGE); + latest .into_iter() .map(|latest| (latest.hash, self.by_change[&latest.hash])) + .collect() + } + + /// The signatures to send with `changes`, those a message carries: of each of them + /// signed, each vouching for its actor's changes before it among them. Whatever the + /// peer was sent before, so changes sent again (a message lost on the way, then + /// repaired) arrive vouched for as the first time. Past `MAX_PER_MESSAGE` (a message + /// carrying the batches of many writers), each actor's latest instead. + pub(crate) fn for_changes(&self, changes: &[ChangeHash]) -> Vec<(ChangeHash, [u8; 64])> { + let signed: Vec<(ChangeHash, [u8; 64])> = changes + .iter() + .filter_map(|hash| { + self.by_change + .get(hash) + .map(|signature| (*hash, *signature)) + }) .collect(); - (signatures, next) + if signed.len() > MAX_PER_MESSAGE { + return self.latest_each(); + } + signed } /// Received signatures waiting for their changes (`offer`) checked against @@ -679,9 +692,13 @@ mod tests { 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()); + assert_eq!(signatures.latest_each(), vec![signed(second)]); + // Sent with the changes they are of: the signed one's alone. + assert_eq!( + signatures.for_changes(&[first, second]), + vec![signed(second)] + ); + assert!(signatures.for_changes(&[first]).is_empty()); // A change not there yet: its signature waits for it. write(&mut theirs, &key.id(), &["c"], json!(3)); @@ -692,7 +709,7 @@ mod tests { .unwrap(); assert_eq!(signatures.offer(&ours, Vec::new()), 1); assert!(signatures.verified(&actor, 3)); - assert_eq!(signatures.newer_than(stamp).0, vec![signed(third)]); + assert_eq!(signatures.latest_each(), vec![signed(third)]); // Kept in a file and read again: verified as before. let again = Signatures::from_kept(&ours, signatures.since(0)); -- cgit v1.3.1