diff --git a/tines/src/ipc.rs b/tines/src/ipc.rs index da7b1e53..cfcbdcd9 100644 --- a/tines/src/ipc.rs +++ b/tines/src/ipc.rs @@ -3,8 +3,9 @@ //! The [`serve`] fn hosts a tines [`Node`] and speaks a tiny //! request/response protocol to each connecting peer. A single instance is the //! control plane for its device; external tools (or another tines Node, in -//! later tasks) drive the node through four operations: `observe`, `belief`, -//! `events`, and `status`. +//! later tasks) drive the node through six operations: `observe`, `belief`, +//! `events`, `status`, and `list` (the fleet console adds addressed `observe` +//! and contact/drop reporting to the `events` response). //! //! # Wire format //! @@ -39,7 +40,7 @@ use std::os::unix::net::{UnixListener, UnixStream}; use std::path::Path; use std::time::Duration; -use tines_core::{Node, NodeEvent, Store, Transport, TrimPolicy}; +use tines_core::{Node, NodeEvent, NodeId, Store, Transport, TrimPolicy}; use tines_store::LogStore; use crate::trust::{adopt_wall_clock, PeerTimeWindow}; @@ -52,6 +53,9 @@ const MAX_FRAME: usize = 1 << 20; // 1 MiB cap on a single JSON frame /// of its accept loop. const POLL_CAP: u64 = 64; +/// Max entries a `List` response returns; `truncated` reports when more exist. +pub const LIST_CAP: usize = 1024; + /// A confirmed delivery handed out by the IPC `events` request. #[derive(serde::Serialize, serde::Deserialize, Debug)] pub struct EventOut { @@ -61,6 +65,50 @@ pub struct EventOut { pub peer: String, } +/// A contact change handed out by the IPC `events` request: a transport +/// `Contact` (`joined: true`) or `Lost` (`joined: false`) for one peer. +#[derive(serde::Serialize, serde::Deserialize, Debug)] +pub struct ContactOut { + /// Hex of the peer's `NodeId`. + pub peer: String, + /// `true` for a new contact, `false` for a loss. + pub joined: bool, +} + +/// A counted drop handed out by the IPC `events` request. +#[derive(serde::Serialize, serde::Deserialize, Debug)] +pub struct DropOut { + /// Hex of the peer the dropped frame came from, or `None` for the + /// overflow slot (which has no address to name). + pub peer: Option, + /// The drop reason, debug-formatted (e.g. `"VersionMismatch { version: 255 }"`). + pub reason: String, + /// The number of frames folded into this counter since the last drain. + pub count: u32, +} + +/// One stored observation handed out by the IPC `list` request. +#[derive(serde::Serialize, serde::Deserialize, Debug)] +pub struct ObservationOut { + /// Hex of the `ObservationId`. + pub id: String, + /// Hex of the observer's `NodeId`. + pub observer: String, + /// Hex of the topic. + pub topic: String, + /// Hex of the payload. + pub payload: String, + /// Hex of the destination, or `None` for a broadcast. + pub to: Option, + /// The addressed lifetime in ms, or `None` for a broadcast. + pub lifetime_ms: Option, + /// How old the observation is, in ms (see `Observation::age_ms`). + pub age_ms: u64, + /// The observer's wall-clock time, ms since the Unix epoch, or `None` + /// when the observer had no trusted clock. + pub observed_at: Option, +} + /// A belief returned by the IPC `belief` request. #[derive(serde::Serialize, serde::Deserialize, Debug)] pub struct BeliefOut { @@ -82,16 +130,26 @@ pub enum Request { topic: String, /// Hex of the observation payload. payload: String, + /// Hex of the destination `NodeId` for an addressed observation. + /// Must be set together with `lifetime_ms`. + #[serde(default)] + to: Option, + /// The addressed lifetime in ms. Must be set together with `to`. + #[serde(default)] + lifetime_ms: Option, }, /// Read the latest belief for `topic`. Belief { /// Hex of the topic bytes to query. topic: String, }, - /// Drain outstanding confirmed deliveries and drop counters. + /// Drain outstanding confirmed deliveries, contact changes, and counted + /// drops. Events, /// Report the node's identity and log depth. Status, + /// List every stored observation (capped at [`LIST_CAP`]). + List, } /// An IPC response. See the module docs for the wire format. @@ -108,10 +166,17 @@ pub enum Response { /// The stored belief, or `None` if the topic is unknown. obs: Option, }, - /// Confirmed deliveries drained from the exchange. + /// Confirmed deliveries, contact changes, and counted drops drained from + /// the exchange. Events { /// The drained confirmed deliveries. events: Vec, + /// The drained contact changes (joined or lost), in order. + #[serde(default)] + contacts: Vec, + /// The drained counted drops. + #[serde(default)] + drops: Vec, }, /// The node's `id`, next tines `next_seq`, and observations `stored`. Status { @@ -122,6 +187,14 @@ pub enum Response { /// Number of observations in the store. stored: u32, }, + /// Every stored observation, up to [`LIST_CAP`] entries. + List { + /// The stored observations. + observations: Vec, + /// `true` when more observations exist than fit in + /// [`LIST_CAP`] entries. + truncated: bool, + }, /// The request was malformed or rejected. Error { /// A human-readable explanation. @@ -147,10 +220,15 @@ pub fn handle( ) -> Response where S: Store + 'static, - T: Transport, + T: Transport, { match req { - Request::Observe { topic, payload } => { + Request::Observe { + topic, + payload, + to, + lifetime_ms, + } => { let topic = match decode_hex(&topic) { Ok(t) => t, Err(e) => return error(e), @@ -159,7 +237,26 @@ where Ok(t) => t, Err(e) => return error(e), }; - match node.observe(&topic, &payload, observed_at, now) { + let result = match (to, lifetime_ms) { + (None, None) => node.observe(&topic, &payload, observed_at, now), + (Some(to), Some(lifetime_ms)) => { + let to = match decode_hex(&to) { + Ok(t) => t, + Err(e) => return error(e), + }; + let to: NodeId = match to.try_into() { + Ok(id) => id, + Err(_) => { + return error("to must be a 16-byte NodeId (32 hex chars)"); + } + }; + node.observe_to(to, lifetime_ms, &topic, &payload, observed_at, now) + } + // An addressed observation needs both a destination and a + // lifetime; half-addressed is almost certainly a client bug. + _ => return error("to and lifetime_ms must be set together"), + }; + match result { Ok(id) => Response::Observe { id: encode_hex(&id), }, @@ -183,30 +280,76 @@ where } } Request::Events => { - let events = node - .drain_events() - .into_iter() - .filter_map(|ev| match ev { - NodeEvent::Delivered { peer, id } => Some(EventOut { + let mut events = Vec::new(); + let mut contacts = Vec::new(); + let mut drops = Vec::new(); + for ev in node.drain_events() { + match ev { + NodeEvent::Delivered { peer, id } => events.push(EventOut { id: encode_hex(&id), peer: encode_hex(&peer), }), - // Dropped frames are surfaced to the operator via stderr - // (they are the whole point of D3), not the `events` - // response, so they filter out here. + NodeEvent::Contact { addr } => contacts.push(ContactOut { + peer: encode_hex(&addr), + joined: true, + }), + NodeEvent::Lost { addr } => contacts.push(ContactOut { + peer: encode_hex(&addr), + joined: false, + }), + // Drops surface here AND on stderr (they are the whole + // point of D3; the stderr line stays for `tines` logs). NodeEvent::Dropped { addr, reason, count, } => { eprintln!("tines: drop: {reason:?} (count={count}, addr={addr:?})"); - None + drops.push(DropOut { + peer: addr.map(|a| encode_hex(&a)), + reason: format!("{reason:?}"), + count, + }); } // `NodeEvent` is non-exhaustive: ignore any future variant. - _ => None, - }) - .collect(); - Response::Events { events } + _ => {} + } + } + Response::Events { + events, + contacts, + drops, + } + } + Request::List => { + let mut ids = Vec::with_capacity(LIST_CAP.min(node.store().len())); + let mut truncated = false; + node.store().for_each_id(&mut |id| { + if ids.len() < LIST_CAP { + ids.push(*id); + } else { + truncated = true; + } + }); + let mut observations = Vec::with_capacity(ids.len()); + for id in &ids { + if let Some(obs) = node.store().get(id, now) { + observations.push(ObservationOut { + id: encode_hex(&obs.id()), + observer: encode_hex(&obs.observer), + topic: encode_hex(&obs.topic), + payload: encode_hex(&obs.payload), + to: obs.dest.as_ref().map(|d| encode_hex(&d.to)), + lifetime_ms: obs.dest.as_ref().map(|d| d.lifetime_ms), + age_ms: obs.age_ms, + observed_at: obs.observed_at, + }); + } + } + Response::List { + observations, + truncated, + } } Request::Status => Response::Status { id: encode_hex(&node.id()), @@ -232,7 +375,7 @@ struct HostState { // [`serve`], fed by the socket read timeouts. fn poll_exchange(node: &mut Node, host: &mut HostState) where - T: Transport + 'static, + T: Transport + 'static, { while node.poll(host.clock.now()) > 0 && host.rounds < POLL_CAP { host.rounds += 1; @@ -308,7 +451,7 @@ pub fn serve( peer_time_window_ms: u64, ) -> std::io::Result<()> where - T: Transport + 'static, + T: Transport + 'static, { let socket_path = socket_path.as_ref().to_path_buf(); // Clear a stale socket file left by a previous run so bind() can't fail. @@ -370,7 +513,7 @@ fn serve_stream( host: &mut HostState, ) -> io::Result<()> where - T: Transport + 'static, + T: Transport + 'static, { loop { match read_request(stream) { diff --git a/tines/src/lib.rs b/tines/src/lib.rs index 1d209200..ba3090cc 100644 --- a/tines/src/lib.rs +++ b/tines/src/lib.rs @@ -15,4 +15,7 @@ pub mod ipc; pub mod trust; pub use clock::HostClock; -pub use ipc::{handle, serve, BeliefOut, EventOut, Request, Response}; +pub use ipc::{ + handle, serve, BeliefOut, ContactOut, DropOut, EventOut, ObservationOut, Request, Response, + LIST_CAP, +}; diff --git a/tines/tests/ipc.rs b/tines/tests/ipc.rs index f86be07b..4a0e542e 100644 --- a/tines/tests/ipc.rs +++ b/tines/tests/ipc.rs @@ -266,6 +266,8 @@ fn observe_and_belief_over_socket() { &Request::Observe { topic: hex_encode(&topic()), payload: hex_encode(&payload()), + to: None, + lifetime_ms: None, }, ); let id = match read_response(&mut stream) { @@ -311,7 +313,7 @@ fn events_over_socket() { rpc(&mut stream, &Request::Events); match read_response(&mut stream) { - Response::Events { events } => { + Response::Events { events, .. } => { assert_eq!(events.len(), 1, "expected one confirmed delivery"); assert_eq!(events[0].id, hex_encode(&x), "event id matches observed id"); assert_eq!(events[0].peer, hex_encode(&peer), "event came from peer"); @@ -329,6 +331,8 @@ fn handle_bad_hex_is_error() { Request::Observe { topic: "z".into(), payload: "ff".into(), + to: None, + lifetime_ms: None, }, None, 0, @@ -348,8 +352,14 @@ fn confirmed_delivery_direct_probe() { node.poll(0); node.poll(0); let evs = node.drain_events(); - assert_eq!(evs.len(), 1, "one confirmed delivery after observe+ack"); + // The mock's first poll surfaces a Contact (now a NodeEvent) ahead of the + // confirmed delivery. + assert_eq!(evs.len(), 2, "contact event then confirmed delivery"); match &evs[0] { + NodeEvent::Contact { addr } => assert_eq!(*addr, [7; 16]), + other => panic!("expected a Contact event, got {other:?}"), + } + match &evs[1] { NodeEvent::Delivered { peer, id } => { assert_eq!(*id, x); assert_eq!(*peer, [7; 16]); @@ -589,6 +599,8 @@ fn f2_propagation_between_instances() { &Request::Observe { topic: hex_encode(&topic()), payload: hex_encode(&payload()), + to: None, + lifetime_ms: None, }, ); let obs = match read_response(&mut stream_a) { @@ -694,6 +706,8 @@ fn b_stamp( &Request::Observe { topic: hex_encode(&topic()), payload: hex_encode(format!("probe-{n}").as_bytes()), + to: None, + lifetime_ms: None, }, ); let id = match read_response(&mut stream) { @@ -895,3 +909,275 @@ fn forged_estimate_does_not_hide_an_honest_one() { fn below_floor_observed_at_is_not_adopted() { assert_not_adopted("floor", 1_000_000_000_000); } + +// --- Widened IPC (fleet console): addressed observe, List, contact/drop events + +/// An addressed `Observe` (both `to` and `lifetime_ms` set) returns the id and +/// the observation shows up in `List` with matching addressing. +#[test] +fn observe_addressed_roundtrip() { + let dir = unique_dir("addressed"); + let mut node = build_node([3; 16], &dir); + let to = [4u8; 16]; + + let resp = handle( + &mut node, + Request::Observe { + topic: hex_encode(&topic()), + payload: hex_encode(&payload()), + to: Some(hex_encode(&to)), + lifetime_ms: Some(60_000), + }, + None, + 0, + ); + let id = match resp { + Response::Observe { id } => id, + other => panic!("expected Observe, got {other:?}"), + }; + + match handle(&mut node, Request::List, None, 0) { + Response::List { + observations, + truncated, + } => { + assert!(!truncated); + assert_eq!(observations.len(), 1); + let o = &observations[0]; + assert_eq!(o.id, id); + assert_eq!(o.observer, hex_encode(&[3u8; 16])); + assert_eq!(o.topic, hex_encode(&topic())); + assert_eq!(o.payload, hex_encode(&payload())); + assert_eq!(o.to.as_deref(), Some(hex_encode(&to).as_str())); + assert_eq!(o.lifetime_ms, Some(60_000)); + } + other => panic!("expected List, got {other:?}"), + } +} + +/// Exactly one of `to` / `lifetime_ms` set is an error, in either direction. +#[test] +fn observe_half_addressed_is_error() { + let dir = unique_dir("half"); + let mut node = build_node([3; 16], &dir); + + for (to, lifetime) in [(Some(hex_encode(&[4u8; 16])), None), (None, Some(1u64))] { + let resp = handle( + &mut node, + Request::Observe { + topic: hex_encode(&topic()), + payload: hex_encode(&payload()), + to, + lifetime_ms: lifetime, + }, + None, + 0, + ); + match resp { + Response::Error { message } => { + assert!( + message.contains("must be set together"), + "unexpected message: {message}" + ); + } + other => panic!("expected Error, got {other:?}"), + } + } +} + +/// `List` returns every stored observation (broadcast and addressed) with +/// correct fields. +#[test] +fn list_returns_all_observations() { + let dir = unique_dir("listall"); + let mut node = build_node([3; 16], &dir); + + let _ = handle( + &mut node, + Request::Observe { + topic: hex_encode(&topic()), + payload: hex_encode(b"broadcast"), + to: None, + lifetime_ms: None, + }, + None, + 0, + ); + let _ = handle( + &mut node, + Request::Observe { + topic: hex_encode(&topic()), + payload: hex_encode(b"addressed"), + to: Some(hex_encode(&[4u8; 16])), + lifetime_ms: Some(1_000), + }, + None, + 0, + ); + + match handle(&mut node, Request::List, None, 0) { + Response::List { + observations, + truncated, + } => { + assert!(!truncated); + assert_eq!(observations.len(), 2, "both observations listed"); + let broadcast = observations + .iter() + .find(|o| o.payload == hex_encode(b"broadcast")) + .expect("broadcast present"); + assert_eq!(broadcast.to, None); + assert_eq!(broadcast.lifetime_ms, None); + assert_eq!(broadcast.topic, hex_encode(&topic())); + assert_eq!(broadcast.observer, hex_encode(&[3u8; 16])); + let addressed = observations + .iter() + .find(|o| o.payload == hex_encode(b"addressed")) + .expect("addressed present"); + assert_eq!( + addressed.to.as_deref(), + Some(hex_encode(&[4u8; 16]).as_str()) + ); + assert_eq!(addressed.lifetime_ms, Some(1_000)); + } + other => panic!("expected List, got {other:?}"), + } +} + +/// `List` caps at `LIST_CAP` entries and reports truncation when more exist. +#[test] +fn list_caps_at_list_cap() { + let dir = unique_dir("listcap"); + let mut node = build_node([3; 16], &dir); + + for _ in 0..(tines::LIST_CAP + 1) { + let resp = handle( + &mut node, + Request::Observe { + topic: hex_encode(&topic()), + payload: hex_encode(b"x"), + to: None, + lifetime_ms: None, + }, + None, + 0, + ); + assert!(matches!(resp, Response::Observe { .. })); + } + + match handle(&mut node, Request::List, None, 0) { + Response::List { + observations, + truncated, + } => { + assert_eq!(observations.len(), tines::LIST_CAP); + assert!(truncated, "more observations exist than fit"); + } + other => panic!("expected List, got {other:?}"), + } +} + +/// The transport's first `Contact` surfaces in `Events.contacts` as +/// `joined: true`, alongside an empty `events` (no deliveries yet). +#[test] +fn events_include_contacts() { + let dir = unique_dir("contacts"); + let socket = dir.join("ipc.sock"); + let peer = [7; 16]; + let mut node = build_node(peer, &dir); + node.poll(0); // the mock's first poll surfaces the Contact + + let serve_path = socket.clone(); + let _handle = + thread::spawn(move || serve(node, serve_path, HostClock::start(), 20, WINDOW_MS).unwrap()); + let mut stream = connect(&socket); + + rpc(&mut stream, &Request::Events); + match read_response(&mut stream) { + Response::Events { + events, + contacts, + drops, + } => { + assert!(events.is_empty(), "no deliveries before any ack"); + assert_eq!(drops.len(), 0, "nothing was dropped"); + assert_eq!(contacts.len(), 1, "one contact surfaced"); + assert_eq!(contacts[0].peer, hex_encode(&peer)); + assert!(contacts[0].joined); + } + other => panic!("expected Events, got {other:?}"), + } +} + +/// `Lost` surfaces as `joined: false`, and a garbage frame surfaces in +/// `drops` with its reason (while the stderr line stays in the service). +#[test] +fn events_include_lost_and_drops() { + let dir = unique_dir("lost"); + let peer = [7; 16]; + let mut node = build_node(peer, &dir); + + node.poll(0); // Contact -> NodeEvent::Contact + node.transport_mut().pending.push_back(Event::Lost(peer)); + node.poll(0); // Lost -> NodeEvent::Lost + node.transport_mut().pending.push_back(Event::Frame( + peer, + FrameBuf::from_slice(&[0xff; 8]).expect("small garbage frame fits"), + )); + node.poll(0); // garbage -> counted drop + + match handle(&mut node, Request::Events, None, 0) { + Response::Events { + events, + contacts, + drops, + } => { + assert!(events.is_empty()); + assert_eq!(contacts.len(), 2, "contact then loss"); + assert!(contacts[0].joined); + assert_eq!(contacts[0].peer, hex_encode(&peer)); + assert!(!contacts[1].joined, "the lost contact is closed"); + assert_eq!(contacts[1].peer, hex_encode(&peer)); + + assert_eq!(drops.len(), 1, "one counted drop"); + assert_eq!(drops[0].peer.as_deref(), Some(hex_encode(&peer).as_str())); + assert!( + drops[0].reason.contains("VersionMismatch"), + "the drop names its reason: {}", + drops[0].reason + ); + assert_eq!(drops[0].count, 1); + } + other => panic!("expected Events, got {other:?}"), + } +} + +/// The existing `events` field still reports confirmed deliveries exactly as +/// before the widening (the mock auto-ack path). +#[test] +fn existing_events_field_unchanged() { + let dir = unique_dir("unchanged"); + let socket = dir.join("ipc.sock"); + let peer = [7; 16]; + let mut node = build_node(peer, &dir); + + let x = node.observe(&topic(), &payload(), None, 0).unwrap(); + node.poll(0); + node.poll(0); + node.poll(0); + + let serve_path = socket.clone(); + let _handle = + thread::spawn(move || serve(node, serve_path, HostClock::start(), 20, WINDOW_MS).unwrap()); + let mut stream = connect(&socket); + + rpc(&mut stream, &Request::Events); + match read_response(&mut stream) { + Response::Events { events, .. } => { + assert_eq!(events.len(), 1, "one confirmed delivery"); + assert_eq!(events[0].id, hex_encode(&x)); + assert_eq!(events[0].peer, hex_encode(&peer)); + } + other => panic!("expected Events, got {other:?}"), + } +}