diff options
Diffstat (limited to 'src/document')
| -rw-r--r-- | src/document/signatures.rs | 61 | ||||
| -rw-r--r-- | src/document/sync.rs | 6 |
2 files changed, 42 insertions, 25 deletions
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<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`). + /// 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)); diff --git a/src/document/sync.rs b/src/document/sync.rs index a7a466b..e9afe06 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -983,11 +983,11 @@ impl SyncSession { self.state.their_heads.is_none() && we_ready && peer_ready == Some(false) } - /// Whether the message just generated (`generate`) carries changes. - pub(crate) fn carrying(&self) -> bool { + /// The changes the message just generated (`generate`, `push`) carries. + pub(crate) fn carried(&self) -> &[crate::document::ChangeHash] { self.sending .as_ref() - .is_some_and(|sending| !sending.sent.is_empty()) + .map_or(&[], |sending| sending.sent.as_slice()) } /// Which session this is: a message generated from it reports its send's outcome |