diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/mod.rs | 77 | ||||
| -rw-r--r-- | src/transport/mycelium.rs | 146 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 2 |
3 files changed, 225 insertions, 0 deletions
diff --git a/src/transport/mod.rs b/src/transport/mod.rs index b2f5c9d..c96b053 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -100,6 +100,80 @@ pub enum ParsedMessage { Sync(SyncEnvelope), Ping(Keepalive), Pong(Keepalive), + Blob(BlobMessage), +} + +/// Content-addressed bytes that any node holding them can serve, in chunks +/// (`swarm::blob`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum BlobId { + /// An Automerge change, by its hash: e.g. an epoch's snapshot change. + Change(ChangeHash), +} + +impl std::fmt::Display for BlobId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BlobId::Change(hash) => write!(f, "change:{hash}"), + } + } +} + +/// A blob request or one of its answers. Each is a message of its own, so no message +/// is larger than one chunk. +#[derive(Debug, Clone)] +pub struct BlobMessage { + /// The sender's epoch (every message carries it). + pub epoch: Epoch, + pub id: BlobId, + pub body: BlobBody, +} + +#[derive(Clone)] +pub enum BlobBody { + /// Asks for these chunks; for the manifest if empty. + Request { + chunks: Vec<u32>, + }, + /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter. + Manifest { + size: u64, + chunk_size: u32, + hashes: Vec<[u8; 32]>, + }, + Chunk { + index: u32, + data: bytes::Bytes, + }, + /// The sender doesn't have the blob. + Missing, +} + +/// Sizes instead of contents: logged messages stay short. +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::Manifest { + size, + chunk_size, + hashes, + } => f + .debug_struct("Manifest") + .field("size", size) + .field("chunk_size", chunk_size) + .field("chunks", &hashes.len()) + .finish(), + BlobBody::Chunk { index, data } => f + .debug_struct("Chunk") + .field("index", index) + .field("bytes", &data.len()) + .finish(), + BlobBody::Missing => f.write_str("Missing"), + } + } } /// What a message is, without its contents. @@ -110,6 +184,7 @@ pub enum MessageKind { Sync, Ping, Pong, + Blob, } impl ParsedMessage { @@ -120,6 +195,7 @@ impl ParsedMessage { ParsedMessage::Sync(_) => MessageKind::Sync, ParsedMessage::Ping(_) => MessageKind::Ping, ParsedMessage::Pong(_) => MessageKind::Pong, + ParsedMessage::Blob(_) => MessageKind::Blob, } } @@ -130,6 +206,7 @@ impl ParsedMessage { ParsedMessage::HelloReply(reply) => reply.epoch, ParsedMessage::Sync(envelope) => envelope.epoch, ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch, + ParsedMessage::Blob(blob) => blob.epoch, } } } diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index cc7f25f..a8ab104 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -19,6 +19,7 @@ use tokio::process::{Child, Command}; use url::Url; use crate::transport::EncodedMessage; +use crate::transport::{BlobBody, BlobId, BlobMessage}; use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope}; use crate::transport::{HeadsSummary, Keepalive, MAX_LISTED_HEADS}; use crate::transport::{MessageHandler, ParsedMessage, Transport}; @@ -31,6 +32,7 @@ const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply"; const TOPIC_SYNC: &str = "continuum.v1.sync"; const TOPIC_PING: &str = "continuum.v1.ping"; const TOPIC_PONG: &str = "continuum.v1.pong"; +const TOPIC_BLOB: &str = "continuum.v1.blob"; const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25); @@ -747,6 +749,7 @@ fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, T } TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)), TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)), + TOPIC_BLOB => Ok(ParsedMessage::Blob(bytes_to_blob(&bytes)?)), t => Err(TransportError::UnknownTopic(t.to_string())), } } @@ -774,6 +777,7 @@ fn format_message( ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?), ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?), ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?), + ParsedMessage::Blob(blob) => (TOPIC_BLOB, blob_to_bytes(blob)?), }; let outbound = OutboundMessage { dst: Destination { @@ -1021,6 +1025,107 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr }) } +// --- BlobMessage --- +#[derive(Serialize, Deserialize)] +enum BlobIdShadow { + Change(ByteBuf), +} + +#[derive(Serialize, Deserialize)] +enum BlobBodyShadow { + Request { + chunks: Vec<u32>, + }, + Manifest { + size: u64, + chunk_size: u32, + hashes: Vec<ByteBuf>, + }, + Chunk { + index: u32, + data: ByteBuf, + }, + Missing, +} + +#[derive(Serialize, Deserialize)] +struct BlobShadow { + epoch: EpochShadow, + id: BlobIdShadow, + body: BlobBodyShadow, +} + +fn blob_to_bytes(msg: &BlobMessage) -> Result<Vec<u8>, rmp_serde::encode::Error> { + let id = match msg.id { + BlobId::Change(hash) => BlobIdShadow::Change(ByteBuf::from(hash.0.to_vec())), + }; + let body = match &msg.body { + BlobBody::Request { chunks } => BlobBodyShadow::Request { + chunks: chunks.clone(), + }, + BlobBody::Manifest { + size, + chunk_size, + hashes, + } => BlobBodyShadow::Manifest { + size: *size, + chunk_size: *chunk_size, + hashes: hashes + .iter() + .map(|hash| ByteBuf::from(hash.to_vec())) + .collect(), + }, + BlobBody::Chunk { index, data } => BlobBodyShadow::Chunk { + index: *index, + data: ByteBuf::from(data.to_vec()), + }, + BlobBody::Missing => BlobBodyShadow::Missing, + }; + rmp_serde::to_vec(&BlobShadow { + epoch: epoch_to_shadow(&msg.epoch), + id, + body, + }) +} + +fn hash32(bytes: &[u8], what: &str) -> Result<[u8; 32], TransportError> { + bytes + .try_into() + .map_err(|_| TransportError::InvalidPayload(format!("{what} must be 32 bytes"))) +} + +fn bytes_to_blob(bytes: &[u8]) -> Result<BlobMessage, TransportError> { + let shadow: BlobShadow = rmp_serde::from_slice(bytes)?; + let id = match shadow.id { + BlobIdShadow::Change(hash) => BlobId::Change(ChangeHash(hash32(&hash, "a change hash")?)), + }; + let body = match shadow.body { + BlobBodyShadow::Request { chunks } => BlobBody::Request { chunks }, + BlobBodyShadow::Manifest { + size, + chunk_size, + hashes, + } => BlobBody::Manifest { + size, + chunk_size, + hashes: hashes + .iter() + .map(|hash| hash32(hash, "a chunk hash")) + .collect::<Result<_, _>>()?, + }, + BlobBodyShadow::Chunk { index, data } => BlobBody::Chunk { + index, + data: data.into_vec().into(), + }, + BlobBodyShadow::Missing => BlobBody::Missing, + }; + Ok(BlobMessage { + epoch: epoch_from_shadow(shadow.epoch)?, + id, + body, + }) +} + #[derive(Error, Debug)] pub enum TransportError { #[error("Mycelium delivered a message without a sender public key")] @@ -1087,6 +1192,47 @@ mod tests { } #[test] + fn blob_messages_round_trip() { + let epoch = Epoch { + generation: 3, + id: [7; 32], + }; + let id = BlobId::Change(ChangeHash([9; 32])); + let chunk = BlobMessage { + epoch, + id, + body: BlobBody::Chunk { + index: 5, + data: vec![1u8; 64 * 1024].into(), + }, + }; + let bytes = blob_to_bytes(&chunk).unwrap(); + // A 64 KiB chunk costs only a few bytes of framing. + assert!(bytes.len() < 64 * 1024 + 128, "{} bytes", bytes.len()); + let decoded = bytes_to_blob(&bytes).unwrap(); + assert_eq!(decoded.id, id); + assert_eq!(decoded.epoch, epoch); + assert!( + matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024) + ); + + let manifest = BlobMessage { + epoch, + id, + body: BlobBody::Manifest { + size: 100, + chunk_size: 64, + hashes: vec![[1; 32], [2; 32]], + }, + }; + let decoded = bytes_to_blob(&blob_to_bytes(&manifest).unwrap()).unwrap(); + assert!(matches!( + decoded.body, + BlobBody::Manifest { size: 100, chunk_size: 64, ref hashes } if hashes.len() == 2 + )); + } + + #[test] fn listed_heads_equal_their_digest() { let heads: Vec<_> = (0..3u8).map(|n| ChangeHash([n; 32])).collect(); let listed = HeadsSummary::of(heads.clone()); diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index bb93d48..9dcd91a 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -39,6 +39,7 @@ pub struct ByKind { pub sync: Counter, pub ping: Counter, pub pong: Counter, + pub blob: Counter, } impl ByKind { @@ -49,6 +50,7 @@ impl ByKind { MessageKind::Sync => &mut self.sync, MessageKind::Ping => &mut self.ping, MessageKind::Pong => &mut self.pong, + MessageKind::Blob => &mut self.blob, } } } |