diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/mod.rs | 72 | ||||
| -rw-r--r-- | src/transport/wire.rs | 30 |
2 files changed, 91 insertions, 11 deletions
diff --git a/src/transport/mod.rs b/src/transport/mod.rs index c109b0c..3bc88e4 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -147,6 +147,9 @@ pub(crate) enum BlobId { Change(#[serde(with = "crate::transport::wire::codec::change_hash")] ChangeHash), /// Bytes by their BLAKE3 hash: e.g. the content of a remake. Content(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]), + /// The changes of a `DeltaSpec`, by its ID (`DeltaSpec::id`). The requests carry + /// the spec, so any node holding its `to` can serve it. + Delta(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]), } impl std::fmt::Display for BlobId { @@ -157,6 +160,10 @@ impl std::fmt::Display for BlobId { write!(f, "content:")?; hash.iter().try_for_each(|byte| write!(f, "{byte:02x}")) } + BlobId::Delta(id) => { + write!(f, "delta:")?; + id.iter().try_for_each(|byte| write!(f, "{byte:02x}")) + } } } } @@ -176,8 +183,12 @@ pub(crate) struct BlobMessage { #[derive(Clone, Serialize, Deserialize)] pub(crate) enum BlobBody { - /// Asks for these chunks; for the manifest if empty. - Request { chunks: Vec<u32> }, + /// Asks for these chunks; for the manifest if empty. `delta`: the spec of a + /// `BlobId::Delta`, which the ID alone doesn't say. + Request { + chunks: Vec<u32>, + delta: Option<DeltaSpec>, + }, /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter. Manifest { size: u64, @@ -198,9 +209,11 @@ pub(crate) enum BlobBody { impl std::fmt::Debug for BlobBody { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - BlobBody::Request { chunks } => { - f.debug_struct("Request").field("chunks", chunks).finish() - } + BlobBody::Request { chunks, delta } => f + .debug_struct("Request") + .field("chunks", chunks) + .field("delta", &delta.is_some()) + .finish(), BlobBody::Manifest { size, chunk_size, @@ -325,14 +338,59 @@ pub(crate) struct SyncEnvelope { /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver /// ignores an epoch older than the newest it has seen in the same session. pub(crate) epoch_seq: u32, - #[serde(with = "crate::transport::wire::codec::sync_message")] - pub(crate) message: SyncPayload, + pub(crate) body: SyncBody, /// The sender was ready when it generated the message: its document is the swarm's, /// not one it is still filling (joining, rejoining). Only such a peer can tell a /// catching-up node that it has everything; an empty document says nothing else. pub(crate) ready: bool, } +/// What a sync envelope carries. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) enum SyncBody { + /// An Automerge sync message. + Message(#[serde(with = "crate::transport::wire::codec::sync_message")] SyncPayload), + /// Instead of a message too large to send in one piece: fetch these changes as a + /// blob (`BlobId::Delta`), then sync on. + Delta(DeltaSpec), +} + +/// The changes in the history of `to` but not of `from`: what a peer at `from` lacks of +/// a document at `to`. Every node holding `to` produces the same bytes for it +/// (`swarm::blob::delta_bytes`). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct DeltaSpec { + /// Sorted. + #[serde(with = "crate::transport::wire::codec::change_hashes")] + pub(crate) from: Vec<ChangeHash>, + /// Sorted. + #[serde(with = "crate::transport::wire::codec::change_hashes")] + pub(crate) to: Vec<ChangeHash>, +} + +impl DeltaSpec { + pub(crate) fn new(mut from: Vec<ChangeHash>, mut to: Vec<ChangeHash>) -> Self { + from.sort_unstable(); + from.dedup(); + to.sort_unstable(); + to.dedup(); + Self { from, to } + } + + /// Names the spec as a blob: SHA-256 over both lists. + pub(crate) fn id(&self) -> [u8; 32] { + let mut hasher = Sha256::new(); + hasher.update(b"continuum/delta/1"); + for heads in [&self.from, &self.to] { + hasher.update((heads.len() as u32).to_be_bytes()); + for head in heads { + hasher.update(head.0); + } + } + hasher.finalize().into() + } +} + /// An Automerge sync message in Automerge's own encoding. Decoded only where it is /// applied (`decode`), after the sender proved to be a member and off the consumer: a /// large message from anyone else costs nothing but its bytes. diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 67d2fb3..009380f 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{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 = 2; +pub(crate) const VERSION: u8 = 3; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -233,6 +233,28 @@ pub(crate) mod codec { } } + /// Change hashes, each a 32-byte binary. + pub(crate) mod change_hashes { + use super::*; + use automerge::ChangeHash; + + pub(crate) fn serialize<S: Serializer>( + value: &[ChangeHash], + s: S, + ) -> Result<S::Ok, S::Error> { + s.collect_seq(value.iter().map(|hash| Bytes::new(&hash.0))) + } + + pub(crate) fn deserialize<'de, D: Deserializer<'de>>( + d: D, + ) -> Result<Vec<ChangeHash>, D::Error> { + Vec::<ByteBuf>::deserialize(d)? + .iter() + .map(|hash| Ok(ChangeHash(fixed(hash, "a change hash")?))) + .collect() + } + } + /// An Automerge sync message, in Automerge's own encoding (left encoded: /// `SyncPayload`). pub(crate) mod sync_message { @@ -416,7 +438,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("020097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("030097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -433,7 +455,7 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "020199a5737761726d91a170{epoch_hex}01c400c2{}05{}", + "030199a5737761726d91a170{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) @@ -448,7 +470,7 @@ mod tests { }); assert_eq!( hex(&encode(&ping).unwrap()), - format!("0203959290c400{epoch_hex}01c3c2") + format!("0303959290c400{epoch_hex}01c3c2") ); } } |