diff --git a/Cargo.lock b/Cargo.lock index 96621b65..f342e39c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1465,6 +1465,7 @@ name = "tines" version = "0.3.0" dependencies = [ "heapless 0.8.0", + "postcard", "serde", "serde_json", "tines-bt", diff --git a/tines-bt/src/transport.rs b/tines-bt/src/transport.rs index 5c05c029..718db742 100644 --- a/tines-bt/src/transport.rs +++ b/tines-bt/src/transport.rs @@ -54,9 +54,9 @@ use bluer::{ }; use tines_core::consts::{MAX_FRAME, MAX_PEERS}; +use tines_core::frame_sender; use tines_core::observation::NodeId; use tines_core::transport::{Event, FrameBuf, SendError, Transport}; -use tines_core::wire::frame_sender; /// The radio I/O seam for [`BtTransport`]. /// diff --git a/tines-core/src/bloom.rs b/tines-core/src/bloom.rs index 3ae83a60..1b7e2cc6 100644 --- a/tines-core/src/bloom.rs +++ b/tines-core/src/bloom.rs @@ -21,8 +21,8 @@ use crate::observation::ObservationId; /// A fixed-size salted Bloom filter keyed by `ObservationId`. #[derive(Clone, Debug, PartialEq, Eq)] -pub struct Bloom { - pub salt: u32, +pub(crate) struct Bloom { + pub(crate) salt: u32, bits: [u8; consts::SUMMARY_BYTES], /// Bytes in use: a power of two in `1..=SUMMARY_BYTES`. len: usize, diff --git a/tines-core/src/consts.rs b/tines-core/src/consts.rs index 7dd2e4e0..2440581d 100644 --- a/tines-core/src/consts.rs +++ b/tines-core/src/consts.rs @@ -5,7 +5,7 @@ /// Wire envelope version byte prepended to every serialized `Frame`. /// -/// The first byte of every frame on the wire; [`crate::wire::Frame::decode`] +/// The first byte of every frame on the wire; `crate::wire::Frame::decode` /// rejects any frame whose leading byte differs from this /// (`WireError::VersionMismatch`), and [`crate::wire::frame_sender`] skips it. /// `4` is the first value in the `4..=0x7F` range that fits a single-byte @@ -42,7 +42,7 @@ pub const REASSEMBLY_TIMEOUT_MS: u64 = 5000; pub const SEND_QUEUE_CAP: usize = 256; /// Minimum usable UDP frame size in bytes (the BLE floor below which a -/// fragment header leaves no room for payload). [`crate::frag::fragment`] gives +/// fragment header leaves no room for payload). `crate::frag::fragment` gives /// up cleanly with `WireError::TooLarge` for any `mtu` below this rather than /// emitting a truncated frame; the MTU sweep runs from here to [`MAX_FRAME`]. pub const MIN_MTU: usize = 23; diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index e3f08d79..82b09c00 100644 --- a/tines-core/src/exchange.rs +++ b/tines-core/src/exchange.rs @@ -1,6 +1,6 @@ //! Exchange state machine for tines-core. //! -//! [`Exchange`] tracks the peers this node knows about and drives the +//! `Exchange` tracks the peers this node knows about and drives the //! content-defined gossip protocol: acknowledging received observations, //! summarizing what we hold, requesting what peers are missing, and delivering //! observations (fragmenting them across the wire when they exceed the MTU). @@ -13,8 +13,10 @@ use crate::consts; use crate::{ - Bloom, Event, Frame, NodeId, Observation, ObservationId, Reassembler, SendError, Store, Tick, - WireError, + bloom::Bloom, + frag::Reassembler, + wire::{Frame, WireError}, + Event, NodeId, Observation, ObservationId, SendError, Store, Tick, }; use heapless::Vec; @@ -31,7 +33,7 @@ const SUMMARY_OVERHEAD: usize = 25; /// MTU changed). The stream is then abandoned and restarted rather than /// sending bytes the receiver cannot reassemble. #[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct InFlight { +pub(crate) struct InFlight { /// The observation being fragmented. pub id: ObservationId, /// Index of the next fragment to send. @@ -45,8 +47,8 @@ pub struct InFlight { pub len: usize, } -/// Why the [`Exchange`] dropped an inbound frame it could not use; reported -/// via [`NodeEvent::Dropped`]. A peer's [`Exchange`] rejects expired addressed +/// Why the `Exchange` dropped an inbound frame it could not use; reported +/// via [`NodeEvent::Dropped`]. A peer's `Exchange` rejects expired addressed /// observations by policy — those are *not* counted as drops. #[non_exhaustive] #[derive(Clone, Copy, Debug, PartialEq)] @@ -57,7 +59,7 @@ pub enum DropReason { /// The version byte read off the wire. version: u8, }, - /// The frame was not a valid [`Frame`] (truncated, garbage, or empty). + /// The frame was not a valid `Frame` (truncated, garbage, or empty). Malformed, } @@ -91,7 +93,7 @@ pub enum NodeEvent { }, } -/// The [`Exchange`]'s counted-drop table: up to [`consts::DROP_TABLE_CAP`] +/// The `Exchange`'s counted-drop table: up to [`consts::DROP_TABLE_CAP`] /// distinct `(addr, DropReason)` pairs, each with a saturating count, plus a /// single overflow counter. Counters are drained — and reset — through /// [`Exchange::drain_events`]. @@ -163,8 +165,8 @@ impl DropTable { } } -/// A single peer the [`Exchange`] is tracking. -pub struct Peer { +/// A single peer the `Exchange` is tracking. +pub(crate) struct Peer { /// Wire address of the peer. pub addr: A, /// Node id learned from the first frame received from this peer. @@ -266,7 +268,7 @@ struct UndoAck { } /// The exchange state machine, tracking up to [`consts::MAX_PEERS`] peers. -pub struct Exchange { +pub(crate) struct Exchange { me: NodeId, salt: u32, peers: Vec, { consts::MAX_PEERS }>, @@ -763,12 +765,14 @@ impl Exchange { Some((addr, n)) } + // Only reached from alloc-gated tests, so unused under --no-default-features. + #[allow(dead_code)] /// Returns the current peers. pub fn peers(&self) -> &[Peer] { &self.peers } - /// [`Exchange`]s emit [`NodeEvent`]s — confirmed deliveries plus counted + /// `Exchange`s emit [`NodeEvent`]s — confirmed deliveries plus counted /// drops — through `Exchange::drain_events`; `tines` surfaces them to its /// transport, so the drain method is unused inside `tines-core` itself. /// @@ -3103,7 +3107,7 @@ mod tests { } /// A frame that passes the version gate (byte 0 is WIRE_VERSION) but whose - /// body isn't a valid postcard [`Frame`] is a counted malformed drop. + /// body isn't a valid postcard `Frame` is a counted malformed drop. #[test] fn malformed_is_counted() { let mut store = MemStore::new(); diff --git a/tines-core/src/frag.rs b/tines-core/src/frag.rs index 61d66f74..f29ed957 100644 --- a/tines-core/src/frag.rs +++ b/tines-core/src/frag.rs @@ -1,11 +1,14 @@ //! Fragmentation and reassembly of `Observation`s over a byte-oriented link. use crate::consts; -use crate::{Frame, NodeId, Observation, ObservationId, Tick, WireError}; +use crate::{ + wire::{Frame, WireError}, + NodeId, Observation, ObservationId, Tick, +}; use heapless::Vec; /// Header allowance per fragment (bytes), pinned. -pub const FRAG_OVERHEAD: usize = 65; +pub(crate) const FRAG_OVERHEAD: usize = 65; /// The most fragments one observation may be split into: the reassembler /// tracks received indices in a `u64` bitset. @@ -27,9 +30,11 @@ pub(crate) fn plan(len: usize, mtu: usize) -> Option<(usize, usize)> { Some((chunk, count)) } +// Only reached from alloc-gated tests, so unused under --no-default-features. +#[allow(dead_code)] /// Encodes `obs` as one Obs frame if it fits in `mtu`, else as N Frag frames. /// Returns the number of frames; writes each via `emit(frame_bytes)`. -pub fn fragment( +pub(crate) fn fragment( from: NodeId, obs: &Observation, mtu: usize, @@ -74,7 +79,7 @@ pub fn fragment( } /// A single-slot reassembler for inbound Frag streams. -pub struct Reassembler { +pub(crate) struct Reassembler { id: Option, count: u8, received: u64, diff --git a/tines-core/src/lib.rs b/tines-core/src/lib.rs index 49bee3b0..7717c56a 100644 --- a/tines-core/src/lib.rs +++ b/tines-core/src/lib.rs @@ -5,29 +5,27 @@ extern crate alloc; pub mod belief; -pub mod bloom; +mod bloom; pub mod consts; -pub mod exchange; -pub mod frag; +mod exchange; +mod frag; #[cfg(feature = "alloc")] pub mod mem_store; pub mod node; pub mod observation; pub mod store; pub mod transport; -pub mod wire; +mod wire; pub use belief::{BeliefPolicy, LatestWins, SelfFirst}; -pub use bloom::Bloom; -pub use exchange::{DropReason, Exchange, NodeEvent, Peer}; -pub use frag::{fragment, Reassembler}; +pub use exchange::{DropReason, NodeEvent}; pub use node::{Node, ObserveError}; pub use observation::{ Dest, NodeId, Observation, ObservationId, Payload, SigAlgo, Signature, Tick, Topic, }; pub use store::{Store, StoreError, TrimPolicy}; pub use transport::{Event, FrameBuf, SendError, Transport}; -pub use wire::{Frame, WireError}; +pub use wire::frame_sender; #[cfg(feature = "alloc")] pub use mem_store::MemStore; diff --git a/tines-core/src/node.rs b/tines-core/src/node.rs index 03584871..927a9874 100644 --- a/tines-core/src/node.rs +++ b/tines-core/src/node.rs @@ -1,21 +1,22 @@ //! Node facade for tines-core. //! //! [`Node`] is the single entry point that ties the [`Store`], the -//! [`Exchange`] state machine, and a [`Transport`] together behind the two +//! `Exchange` state machine, and a [`Transport`] together behind the two //! operations an application needs: [`Node::observe`] to record an observation, //! and [`Node::poll`] to drive the exchange. //! //! The facade never touches the heap: its only owned buffer, `out`, is a fixed //! `[u8; MAX_FRAME]`. It also panics on nothing — oversized topics/payloads //! become [`ObserveError::TooLarge`] rather than truncating, and malformed -//! transport frames are dropped by the [`Exchange`]. +//! transport frames are dropped by the `Exchange`. use crate::{ - consts, BeliefPolicy, Dest, Exchange, NodeEvent, NodeId, Observation, ObservationId, Payload, - Store, StoreError, Tick, Topic, Transport, TrimPolicy, + consts, exchange::Exchange, BeliefPolicy, Dest, NodeEvent, NodeId, Observation, ObservationId, + Payload, Store, StoreError, Tick, Topic, Transport, TrimPolicy, }; /// Errors returned by [`Node::observe`]. +#[non_exhaustive] #[derive(Debug, PartialEq, Eq)] pub enum ObserveError { /// The topic or payload exceeded `MAX_TOPIC`/`MAX_PAYLOAD`. The @@ -105,7 +106,7 @@ impl Node { /// /// The resulting observation has `dest = Some(Dest { to, lifetime_ms })`, /// so the exchange will deliver it only - /// to `to` (see [`Exchange`]). Returns the observation's content-derived + /// to `to` (see `Exchange`). Returns the observation's content-derived /// id. An oversized topic or payload is rejected with /// [`ObserveError::TooLarge`] rather than truncated; a store rejection is /// surfaced as [`ObserveError::Store`]. Duplicate ids are accepted (the id @@ -143,7 +144,7 @@ impl Node { /// Returns the number of *new* observations stored this round. If the /// transport refuses the frame with `Busy` or `Unreachable`, the exchange /// rolls back and offers the same frame on a later poll; a `TooLarge` - /// frame is dropped (see [`Exchange::on_send_error`]). + /// frame is dropped (see `Exchange::on_send_error`). pub fn poll(&mut self, now: Tick) -> usize { let mut stored = 0usize; while let Some(ev) = self.transport.poll() { @@ -165,7 +166,7 @@ impl Node { stored } - /// Drains the [`NodeEvent`]s the [`Exchange`] has collected: confirmed + /// Drains the [`NodeEvent`]s the `Exchange` has collected: confirmed /// deliveries first (oldest-first), then one [`NodeEvent::Dropped`] per /// counted drop. /// diff --git a/tines-core/src/observation.rs b/tines-core/src/observation.rs index eae682e0..b95ca675 100644 --- a/tines-core/src/observation.rs +++ b/tines-core/src/observation.rs @@ -35,6 +35,11 @@ pub struct Dest { } /// An observation published by a node, with content-derived identity. +/// +/// `#[non_exhaustive]`: a future 1.x may add fields, so build it via the +/// constructors on [`Observation`] rather than a struct literal outside +/// `tines-core`. +#[non_exhaustive] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Observation { pub observer: NodeId, diff --git a/tines-core/src/store.rs b/tines-core/src/store.rs index 40aff969..a87fe463 100644 --- a/tines-core/src/store.rs +++ b/tines-core/src/store.rs @@ -13,6 +13,10 @@ use crate::observation::{Observation, ObservationId, Tick}; /// Rules are applied in the order listed here: per-topic retention, then a /// global age cutoff, then a global count cap. Any field left as `None` is /// ignored. +/// +/// `#[non_exhaustive]`: build via [`TrimPolicy::default`] and set fields, so a +/// future 1.x can add policy knobs without a breaking change. +#[non_exhaustive] #[derive(Clone, Debug, Default)] pub struct TrimPolicy { /// Keep the `N` youngest (lowest current age) observations per exact topic. @@ -24,6 +28,7 @@ pub struct TrimPolicy { } /// Errors returned by [`Store`] operations. +#[non_exhaustive] #[derive(Debug, PartialEq, Eq)] pub enum StoreError { /// The store is full and cannot accept new observations. diff --git a/tines-core/src/transport.rs b/tines-core/src/transport.rs index ab93f396..5800b14b 100644 --- a/tines-core/src/transport.rs +++ b/tines-core/src/transport.rs @@ -15,6 +15,7 @@ pub type FrameBuf = heapless::Vec; /// `Contact`/`Lost`; the `large_enum_variant` lint is intentionally suppressed /// to keep `FrameBuf` off the heap in the `no_std` core. #[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] #[allow(clippy::large_enum_variant)] pub enum Event { /// A peer became reachable. @@ -26,6 +27,7 @@ pub enum Event { } /// Reasons a `Transport::send` call can fail. +#[non_exhaustive] #[derive(Debug, PartialEq, Eq)] pub enum SendError { /// The transport is busy with an in-flight send and cannot accept another. diff --git a/tines-core/src/wire.rs b/tines-core/src/wire.rs index 57c38535..6e08140f 100644 --- a/tines-core/src/wire.rs +++ b/tines-core/src/wire.rs @@ -15,7 +15,7 @@ use serde::{Deserialize, Serialize}; /// large, so the informative `large_enum_variant` lint is suppressed here. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[allow(clippy::large_enum_variant)] -pub enum Frame { +pub(crate) enum Frame { Summary { from: NodeId, salt: u32, @@ -85,7 +85,7 @@ impl Frame { /// Peek the [`NodeId`] a wire frame was sent by, without fully decoding it. /// -/// A [`Frame`] envelope is a `WIRE_VERSION` byte, then an externally-tagged +/// A `Frame` envelope is a `WIRE_VERSION` byte, then an externally-tagged /// postcard enum: `bytes[0]` is the version, `bytes[1]` the variant tag, and /// `from` (the first field) occupies the 16 bytes at `bytes[2..18]`. This skips /// the two leading bytes, reads those 16 bytes, and returns `None` for anything @@ -114,7 +114,7 @@ pub fn frame_sender(bytes: &[u8]) -> Option { /// Errors produced while (de)serializing frames on the wire. #[derive(Debug, PartialEq, Eq)] -pub enum WireError { +pub(crate) enum WireError { /// The output buffer was too small to hold the serialized frame. TooLarge, /// The input bytes were malformed, truncated, or not a valid `Frame`. diff --git a/tines-core/tests/public_api.rs b/tines-core/tests/public_api.rs new file mode 100644 index 00000000..51f7fd3e --- /dev/null +++ b/tines-core/tests/public_api.rs @@ -0,0 +1,26 @@ +//! Compile probe for the tines-core 1.0 public surface (Task 5). +//! +//! Every `use` path here must keep resolving against `tines_core`. A future +//! change that removes any of these items from the public API — or drops a +//! crate-root re-export that Task 5 established — fails to build here, which +//! is exactly what the probe guards against. CI compiles it on the workspace +//! both with and without the default (alloc) features. +#![allow(unused_imports)] + +// Re-exported at the crate root by Task 5's `tines_core::lib`. +use tines_core::consts::WIRE_VERSION; +use tines_core::{ + belief::{BeliefPolicy, LatestWins, SelfFirst}, + node::{Node, ObserveError}, + observation::{ + Dest, NodeId, Observation, ObservationId, Payload, SigAlgo, Signature, Tick, Topic, + }, + store::{Store, StoreError, TrimPolicy}, + transport::{Event, FrameBuf, SendError, Transport}, + DropReason, NodeEvent, +}; +use tines_core::{consts, frame_sender}; + +// `MemStore` lives behind the `alloc` feature. +#[cfg(feature = "alloc")] +use tines_core::MemStore; diff --git a/tines-sim/src/world.rs b/tines-sim/src/world.rs index e081cea6..b9bd8e87 100644 --- a/tines-sim/src/world.rs +++ b/tines-sim/src/world.rs @@ -112,8 +112,8 @@ impl SimWorld { /// A test seam: it lets tests feed arbitrary or malformed byte strings /// straight to a node's exchange, which is where the "never panic on /// untrusted wire input" guarantee lives. The frame is not validated here — - /// a malformed frame is dropped by the receiving node's - /// [`Exchange`](tines_core::Exchange) rather than stored. + /// a malformed frame is dropped by the receiving + /// [`Node`](tines_core::Node) rather than stored. pub fn inject_frame(&mut self, node: usize, peer: usize, bytes: FrameBuf) { self.inboxes[node].push_back(Event::Frame(peer, bytes)); } diff --git a/tines-store/src/log.rs b/tines-store/src/log.rs index 090fe73a..0858c9d4 100644 --- a/tines-store/src/log.rs +++ b/tines-store/src/log.rs @@ -1016,15 +1016,13 @@ mod tests { } fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option) -> Observation { - Observation { - observer, - seq, - age_ms: age, - observed_at: at, - topic: tines_core::Topic::from_slice(b"door/3").unwrap(), - payload: tines_core::Payload::from_slice(b"open").unwrap(), - dest: None, - sig: None, + // `Observation` is `#[non_exhaustive]` outside `tines-core` (Task 5); + // build it via the public constructor instead of a struct literal. + let base = Observation::broadcast(observer, seq, b"door/3", b"open").aged(age); + if let Some(at) = at { + base.with_observed_at(at) + } else { + base } } diff --git a/tines-store/tests/persistence.rs b/tines-store/tests/persistence.rs index 14545891..4921aae3 100644 --- a/tines-store/tests/persistence.rs +++ b/tines-store/tests/persistence.rs @@ -7,21 +7,19 @@ use std::io::Write; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; -use tines_core::{Observation, Payload, Store, Topic}; +use tines_core::{Observation, Store}; use tines_store::LogStore; static COUNTER: AtomicU64 = AtomicU64::new(0); fn obs(observer: [u8; 16], seq: u64, age: u64, at: Option) -> Observation { - Observation { - observer, - seq, - age_ms: age, - observed_at: at, - topic: Topic::from_slice(b"door/3").unwrap(), - payload: Payload::from_slice(b"open").unwrap(), - dest: None, - sig: None, + // `Observation` is `#[non_exhaustive]` outside `tines-core` (Task 5); + // build it via the public constructor instead of a struct literal. + let base = Observation::broadcast(observer, seq, b"door/3", b"open").aged(age); + if let Some(at) = at { + base.with_observed_at(at) + } else { + base } } @@ -335,20 +333,13 @@ mod fleet_fixes_f5 { use super::*; use std::os::unix::fs::MetadataExt; use std::path::Path; - use tines_core::{Observation, Payload, Store, Topic}; + use tines_core::{Observation, Store}; use tines_store::LogStore; fn a_obs(observer: [u8; 16], seq: u64) -> Observation { - Observation { - observer, - seq, - age_ms: 0, - observed_at: Some(10), - topic: Topic::from_slice(b"door/3").unwrap(), - payload: Payload::from_slice(b"open").unwrap(), - dest: None, - sig: None, - } + // `Observation` is `#[non_exhaustive]` outside `tines-core` (Task 5); + // build it via the public constructor instead of a struct literal. + Observation::broadcast(observer, seq, b"door/3", b"open").with_observed_at(10) } /// Flip a byte in the body of the `n`-th record after the header (0 = Boot), diff --git a/tines/Cargo.toml b/tines/Cargo.toml index 762fcd9d..60194bb7 100644 --- a/tines/Cargo.toml +++ b/tines/Cargo.toml @@ -19,3 +19,7 @@ serde_json = "1" [dev-dependencies] heapless = "0.8" +# Only for the ipc mock peer, which reconstructs wire frames from tines-core's +# public `Observation` to classify inbound and synthesize ack/summary frames — +# without reaching into the private `Frame` enum. +postcard = "1" diff --git a/tines/tests/ipc.rs b/tines/tests/ipc.rs index f4e320de..14dc4786 100644 --- a/tines/tests/ipc.rs +++ b/tines/tests/ipc.rs @@ -16,7 +16,8 @@ use heapless::Vec as HeaplessVec; use std::fmt::Write as _; use tines::{handle, serve, HostClock, Request, Response}; use tines_core::{ - consts, Event, Frame, FrameBuf, Node, NodeEvent, NodeId, ObservationId, SendError, Transport, + consts, Event, FrameBuf, Node, NodeEvent, NodeId, Observation, ObservationId, SendError, + Transport, }; use tines_store::LogStore; @@ -56,27 +57,27 @@ impl Transport for MockTransport { fn send(&mut self, to: &NodeId, frame: &[u8]) -> Result<(), SendError> { self.sent.push(frame.to_vec()); - match Frame::decode(frame) { - Ok(Frame::Obs { obs, .. }) => { + match frame_kind(frame) { + Some(OBS_TAG) => { // A well-behaved peer acks each distinct observed broadcast. - if self.acked.insert(obs.id()) { - let ack = ack_frame(*to, obs.id()); - self.pending.push_back(frame_event(*to, &ack)); + if let Ok((_, obs)) = postcard::from_bytes::<(NodeId, Observation)>(&frame[2..]) { + if self.acked.insert(obs.id()) { + self.pending.push_back(ack_event(*to, obs.id())); + } } } - Ok(Frame::Summary { salt, .. }) => { + Some(SUMMARY_TAG) => { // Echo back an empty, full-size Summary so the exchange learns // the peer holds nothing and broadcasts its diff (which we then // ack, firing a NodeEvent::Delivered). - let raw: [u8; consts::SUMMARY_BYTES] = [0u8; consts::SUMMARY_BYTES]; - let bits: HeaplessVec = - HeaplessVec::from_slice(&raw).expect("512 fits SUMMARY_BYTES"); - let reply = Frame::Summary { - from: *to, - salt, - bits, - }; - self.pending.push_back(frame_event(*to, &reply)); + if let Ok((from, salt, _)) = postcard::from_bytes::<( + NodeId, + u32, + HeaplessVec, + )>(&frame[2..]) + { + self.pending.push_back(summary_event(from, salt)); + } } _ => {} } @@ -88,21 +89,51 @@ impl Transport for MockTransport { } } -fn ack_frame(from: NodeId, id: ObservationId) -> Frame { - let ids: HeaplessVec = - HeaplessVec::from_slice(&[id]).expect("one id fits ACK_IDS_CAP"); - Frame::Ack { from, ids } +// Wire-variant tags, mirroring tines-core's private `Frame` enum order +// (Summary=0, Obs=1, Frag=2, Ack=3). tines-core keeps `Frame` internal, so the +// mock peer re-derives these here to classify inbound frames and synthesize the +// acks/summaries the exchange expects from a well-behaved peer. +const SUMMARY_TAG: u8 = 0; +const OBS_TAG: u8 = 1; +const ACK_TAG: u8 = 3; + +/// Returns a wire frame's postcard variant tag, or `None` if `frame` is not a +/// well-formed tines frame (too short, or a wrong leading version byte). +fn frame_kind(frame: &[u8]) -> Option { + (frame.len() >= 2 && frame[0] == consts::WIRE_VERSION).then_some(frame[1]) } -fn frame_event(from: NodeId, frame: &Frame) -> Event { +/// Encode a `Frame`-variant-shaped tuple as wire bytes: the WIRE_VERSION byte, +/// the variant tag, then the postcard payload — mirroring tines-core's +/// internal `Frame::encode`, using only `tines_core`'s public types. +fn encode_frame(from: NodeId, tag: u8, payload: &T) -> Event { let mut buf = [0u8; consts::MAX_FRAME]; - let n = frame.encode(&mut buf).expect("frame fits MAX_FRAME"); + buf[0] = consts::WIRE_VERSION; + buf[1] = tag; + let n = postcard::to_slice(payload, &mut buf[2..]) + .expect("frame fits MAX_FRAME") + .len(); Event::Frame( from, - FrameBuf::from_slice(&buf[..n]).expect("frame fits FrameBuf"), + FrameBuf::from_slice(&buf[..2 + n]).expect("frame fits FrameBuf"), ) } +/// Synthesize the Ack a well-behaved peer sends for one observed id. +fn ack_event(from: NodeId, id: ObservationId) -> Event { + let ids: HeaplessVec = + HeaplessVec::from_slice(&[id]).expect("one id fits ACK_IDS_CAP"); + encode_frame(from, ACK_TAG, &(from, ids)) +} + +/// Synthesize the full-size, all-zero Summary a well-behaved peer echoes for an +/// incoming Summary (telling the exchange the peer holds nothing). +fn summary_event(from: NodeId, salt: u32) -> Event { + let bits: HeaplessVec = + HeaplessVec::from_slice(&[0u8; consts::SUMMARY_BYTES]).expect("512 fits SUMMARY_BYTES"); + encode_frame(from, SUMMARY_TAG, &(from, salt, bits)) +} + fn topic() -> Vec { b"tines/test".to_vec() }