From c13cee08dedeb8e73560f9bf9f7154c9fa089e2d Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 04:00:59 +0100 Subject: replace shadow types by wire codec --- src/transport/mod.rs | 38 ++++++++++++++++++++++++++------------ 1 file changed, 26 insertions(+), 12 deletions(-) (limited to 'src/transport/mod.rs') diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 35e3499..14c5ee4 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -2,6 +2,7 @@ use anyhow::Result; use async_trait::async_trait; use automerge::ChangeHash; use automerge::sync::Message; +use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::boxed::Box; use std::future::Future; @@ -14,6 +15,7 @@ pub mod mycelium; #[cfg(test)] pub(crate) mod sim; pub mod traffic; +pub(crate) mod wire; /// Random identifier of one run of a node (from start to exit). pub type RunId = [u8; 16]; @@ -83,10 +85,12 @@ fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] { } /// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct Keepalive { /// Only comparable between nodes on the same epoch. + #[serde(with = "crate::transport::wire::codec::heads")] pub heads: HeadsSummary, + #[serde(with = "crate::transport::wire::codec::epoch")] pub epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. pub epoch_seq: u32, @@ -109,12 +113,12 @@ pub enum ParsedMessage { /// Content-addressed bytes that any node holding them can serve, in chunks /// (`swarm::blob`). -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum BlobId { /// An Automerge change, by its hash: e.g. an epoch's snapshot change. - Change(ChangeHash), + Change(#[serde(with = "crate::transport::wire::codec::change_hash")] ChangeHash), /// Bytes by their BLAKE3 hash: e.g. the content of a remake. - Content([u8; 32]), + Content(#[serde(with = "crate::transport::wire::codec::bytes32")] [u8; 32]), } impl std::fmt::Display for BlobId { @@ -131,9 +135,10 @@ impl std::fmt::Display for BlobId { /// 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)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct BlobMessage { /// The sender's epoch (every message carries it). + #[serde(with = "crate::transport::wire::codec::epoch")] pub epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. pub epoch_seq: u32, @@ -141,20 +146,20 @@ pub struct BlobMessage { pub body: BlobBody, } -#[derive(Clone)] +#[derive(Clone, Serialize, Deserialize)] pub enum BlobBody { /// Asks for these chunks; for the manifest if empty. - Request { - chunks: Vec, - }, + Request { chunks: Vec }, /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter. Manifest { size: u64, chunk_size: u32, + #[serde(with = "crate::transport::wire::codec::hashes32")] hashes: Vec<[u8; 32]>, }, Chunk { index: u32, + #[serde(with = "crate::transport::wire::codec::shared_bytes")] data: bytes::Bytes, }, /// The sender doesn't have the blob. @@ -234,52 +239,61 @@ impl ParsedMessage { } } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct HelloRequest { /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in /// another swarm doesn't answer. pub swarm_id: Option, + #[serde(with = "crate::transport::wire::codec::epoch")] pub epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. pub epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. + #[serde(with = "serde_bytes")] pub auth: Vec, /// The sender is ready (same meaning as `SyncEnvelope::ready`). pub ready: bool, /// Random per run of the sender; the reply echoes it. + #[serde(with = "crate::transport::wire::codec::run_id")] pub run_id: RunId, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct HelloReply { pub swarm_id: String, pub peers: Vec, + #[serde(with = "crate::transport::wire::codec::epoch")] pub epoch: Epoch, /// See `SyncEnvelope::epoch_seq`. pub epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. + #[serde(with = "serde_bytes")] pub auth: Vec, /// The sender is ready. pub ready: bool, /// The `run_id` of the HELLO answered. A reply to a HELLO from an earlier run of the /// recipient (delivered late, or queued while it was down) is ignored. + #[serde(with = "crate::transport::wire::codec::run_id")] pub run_id: RunId, /// How long before this HELLO the sender had last heard from the recipient, if ever /// (in milliseconds): tells a restarted node how long it was away. pub away_ms: Option, /// The sender's own run (as its HELLOs carry it): a new one means it restarted. + #[serde(with = "crate::transport::wire::codec::run_id")] pub sender_run_id: RunId, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct SyncEnvelope { pub swarm_id: String, /// Epoch of the document the message was generated from. + #[serde(with = "crate::transport::wire::codec::epoch")] pub epoch: Epoch, /// How many times the sender changed epoch in its current run, when it sent this /// (`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 epoch_seq: u32, + #[serde(with = "crate::transport::wire::codec::sync_message")] pub message: Message, /// 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 -- cgit v1.3.1