diff options
| -rw-r--r-- | docs/api.md | 15 | ||||
| -rw-r--r-- | src/document/signatures.rs | 137 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 12 | ||||
| -rw-r--r-- | src/transport/mod.rs | 5 | ||||
| -rw-r--r-- | src/transport/wire.rs | 2 |
5 files changed, 118 insertions, 53 deletions
diff --git a/docs/api.md b/docs/api.md index 3fb99be..a4b76b0 100644 --- a/docs/api.md +++ b/docs/api.md @@ -291,15 +291,18 @@ peer was sent before, so changes sent again (a message lost on the way) arrive vouched for as the first time; and a node checks one it receives once it holds the 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 4096 signed changes (a whole document pulled), and a delta, carry each node run's -latest instead. 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 unsigned, never sent. +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 +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 +unsigned, never sent. Two cases still arrive unvouched, and wait under a policy requiring signatures: a change vouched for only by a later change of its node that the rules refused (nodes take it, -then can't pass the proof on, [`future-work.md`](future-work.md)); and, in one transfer, -the changes of an epoch with more than 4096 node runs writing. +then can't pass the proof on); and, in one transfer, the changes of runs past the +65536 a message carries signatures of at most, in an epoch with more runs writing than +that since it began. A remake starts the count again. The signatures a node has are kept in its document file (`--document-file`) with the document. diff --git a/src/document/signatures.rs b/src/document/signatures.rs index e5c9e69..b62e545 100644 --- a/src/document/signatures.rs +++ b/src/document/signatures.rs @@ -21,7 +21,7 @@ //! more of its actor's changes than any before; one of a change not there yet waits //! (`pending`, a bounded number: the oldest go). -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use super::{ChangeHash, Document}; @@ -148,11 +148,17 @@ impl<'de> serde::Deserialize<'de> for SignEvery { } } -/// Signatures a sync message carries at most (`for_changes`, `latest_each`), as many -/// as wait for their changes (`MAX_PENDING`): 384 KiB of them. A whole document pulled, -/// or a delta, carries each actor's latest, and an actor is a node's run: an epoch with -/// more runs writing than this leaves some unvouched for in such a transfer. -pub(crate) const MAX_PER_MESSAGE: usize = MAX_PENDING; +/// Signatures a sync message carries at most, and as many wait for their changes +/// (`MAX_PENDING`): 6 MiB of them. A message carries one per writer it holds changes of +/// at most (`for_changes`, `for_delta`), and a writer is a node's run (each start, rejoin +/// or drop makes a fresh actor) since the epoch began: a transfer of an epoch with more +/// runs writing than this, a whole document pulled or a delta, leaves the rest unvouched +/// for, and a node taking it under a policy requiring signatures waits for them. A remake +/// starts the count again. +pub(crate) const MAX_PER_MESSAGE: usize = 1 << 16; +/// Past this many signed changes among those a message carries, it carries each +/// writer's latest among them instead (`for_changes`): they vouch for the rest. +const EACH_WRITER_PAST: 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; @@ -186,7 +192,8 @@ pub(crate) fn unpack(bytes: &[u8]) -> Option<Vec<(ChangeHash, [u8; 64])>> { ) } /// Received signatures of changes not there yet, kept at most (`offer`): the oldest go. -const MAX_PENDING: usize = 4096; +/// As many as a message carries, so a delta's (whose changes come after it) all wait. +const MAX_PENDING: usize = MAX_PER_MESSAGE; /// Received signatures of changes the document doesn't hold yet, `MAX_PENDING` at most: /// the oldest go, so junk a peer sent once (signatures of changes that never come) @@ -237,8 +244,6 @@ 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`: which became so last (`latest_each`). - stamp: u64, /// Received signatures of changes the document doesn't hold yet. pending: Pending, } @@ -248,8 +253,6 @@ pub(crate) struct Signatures { struct Latest { seq: u64, hash: ChangeHash, - /// `Signatures::stamp` when it became the latest. - stamp: u64, } impl Signatures { @@ -325,15 +328,7 @@ impl Signatures { { return false; } - self.stamp += 1; - self.latest.insert( - actor, - Latest { - seq, - hash, - stamp: self.stamp, - }, - ); + self.latest.insert(actor, Latest { seq, hash }); true } @@ -368,24 +363,40 @@ impl Signatures { kept } - /// 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 + /// The signatures to send with a delta from `from` (`SyncBody::Delta`): of each + /// writer with changes in it, its latest verified, vouching for its changes before + /// it; `MAX_PER_MESSAGE` at most. The delta's changes come after it, so these wait + /// for them at the peer (`offer`). + pub(crate) fn for_delta( + &self, + doc: &Document, + from: &[ChangeHash], + ) -> Vec<(ChangeHash, [u8; 64])> { + let writers: HashSet<super::ActorId> = doc + .get_changes_meta(from) .into_iter() + .map(|meta| meta.actor.into_owned()) + .collect(); + writers + .iter() + .filter_map(|writer| self.latest.get(writer)) .map(|latest| (latest.hash, self.by_change[&latest.hash])) + .take(MAX_PER_MESSAGE) .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 + /// signed, each vouching for its writer'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])> { + /// repaired) arrive vouched for as the first time. Past `EACH_WRITER_PAST` (a whole + /// document pulled, a message carrying the batches of many writers), each writer's + /// latest among them, which vouches for the rest: one per writer, + /// `MAX_PER_MESSAGE` at most. + pub(crate) fn for_changes( + &self, + doc: &Document, + changes: &[ChangeHash], + ) -> Vec<(ChangeHash, [u8; 64])> { let signed: Vec<(ChangeHash, [u8; 64])> = changes .iter() .filter_map(|hash| { @@ -394,10 +405,28 @@ impl Signatures { .map(|signature| (*hash, *signature)) }) .collect(); - if signed.len() > MAX_PER_MESSAGE { - return self.latest_each(); + if signed.len() <= EACH_WRITER_PAST { + return signed; } - signed + let signed: HashMap<ChangeHash, [u8; 64]> = signed.into_iter().collect(); + // Each signed one's writer and place among its changes: the metadata of the + // document's changes, read once (a message this large carries most of them). + let mut latest: HashMap<super::ActorId, (u64, ChangeHash)> = HashMap::new(); + for meta in doc.get_changes_meta(&[]) { + if signed.contains_key(&meta.hash) { + let at = latest + .entry(meta.actor.into_owned()) + .or_insert((meta.seq, meta.hash)); + if meta.seq > at.0 { + *at = (meta.seq, meta.hash); + } + } + } + latest + .into_values() + .map(|(_, hash)| (hash, signed[&hash])) + .take(MAX_PER_MESSAGE) + .collect() } /// Received signatures waiting for their changes (`offer`) checked against @@ -424,13 +453,11 @@ impl Signatures { .get(actor) .is_none_or(|latest| latest.seq < change.seq()) { - self.stamp += 1; self.latest.insert( actor.clone(), Latest { seq: change.seq(), hash, - stamp: self.stamp, }, ); } @@ -588,6 +615,35 @@ mod tests { .unwrap(); } + /// Past `EACH_WRITER_PAST` signed changes, a message carries each writer's latest + /// among them; a delta, each writer's with changes in it. + #[test] + fn each_writers_latest_past_many() { + let key = NodeKey::from_seed([1; 32]); + let mut doc = Document::new(); + let mut signatures = Signatures::default(); + let mut hashes = Vec::new(); + for i in 0..=EACH_WRITER_PAST { + write(&mut doc, &key.id(), &["n"], json!(i)); + let hash = doc.get_heads()[0]; + signatures.sign(&doc, hash, &key); + hashes.push(hash); + } + let last = *hashes.last().unwrap(); + assert_eq!(signatures.for_changes(&doc, &hashes).len(), 1); + assert_eq!(signatures.for_changes(&doc, &hashes)[0].0, last); + // Fewer than that: each signed one. + assert_eq!(signatures.for_changes(&doc, &hashes[..10]).len(), 10); + // Another writer's changes alone in a delta: only its latest. + let mut theirs = doc.fork().with_actor(ActorId::random()); + write(&mut theirs, &key.id(), &["m"], json!(1)); + let other = theirs.get_heads()[0]; + signatures.sign(&theirs, other, &key); + let delta = signatures.for_delta(&theirs, &[last]); + assert_eq!(delta.len(), 1); + assert_eq!(delta[0].0, other); + } + /// A listed signed path is a pattern: `*` takes any one key, within and above. #[test] fn listed_paths_take_a_star() { @@ -718,13 +774,14 @@ mod tests { signatures.offer(&ours, vec![(second, other.sign(&message(&second)))]), 0 ); - assert_eq!(signatures.latest_each(), vec![signed(second)]); + assert_eq!(signatures.for_delta(&ours, &[]), vec![signed(second)]); + assert!(signatures.for_delta(&ours, &[second]).is_empty()); // Sent with the changes they are of: the signed one's alone. assert_eq!( - signatures.for_changes(&[first, second]), + signatures.for_changes(&ours, &[first, second]), vec![signed(second)] ); - assert!(signatures.for_changes(&[first]).is_empty()); + assert!(signatures.for_changes(&ours, &[first]).is_empty()); // A change not there yet: its signature waits for it. write(&mut theirs, &key.id(), &["c"], json!(3)); @@ -735,7 +792,7 @@ mod tests { .unwrap(); assert_eq!(signatures.offer(&ours, Vec::new()), 1); assert!(signatures.verified(&actor, 3)); - assert_eq!(signatures.latest_each(), vec![signed(third)]); + assert_eq!(signatures.for_delta(&ours, &[]), 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/swarm/sync.rs b/src/swarm/sync.rs index e3f6055..3a3d92c 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -300,11 +300,15 @@ 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 any writer's changes, - // each writer's latest. A message carrying no changes takes none. + // 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 { - Some(transport::SyncBody::Delta(_)) => replica.signatures.latest_each(), - Some(_) => replica.signatures.for_changes(peer_state.sync.carried()), + Some(transport::SyncBody::Delta(spec)) => { + replica.signatures.for_delta(&replica.doc, &spec.from) + } + Some(_) => replica + .signatures + .for_changes(&replica.doc, peer_state.sync.carried()), None => Vec::new(), }; let signatures = crate::document::signatures::pack(&signatures); diff --git a/src/transport/mod.rs b/src/transport/mod.rs index cbb6734..fcdf56e 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -734,8 +734,9 @@ 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, or each actor's latest for a delta or past `signatures::MAX_PER_MESSAGE` - /// (`Signatures::for_changes`, `latest_each`), as `signatures::pack` packs them. + /// 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. pub(crate) signatures: serde_bytes::ByteBuf, } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 6784f8e..7fe8983 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 = 47; +pub(crate) const VERSION: u8 = 48; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { |