feat(host): widen service IPC — addressed observe, List, contact/drop events

New public names: ContactOut, DropOut, ObservationOut, LIST_CAP.

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<String>,
+    /// 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<String>,
+    /// The addressed lifetime in ms, or `None` for a broadcast.
+    pub lifetime_ms: Option<u64>,
+    /// 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<u64>,
+}
+
 /// 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<String>,
+        /// The addressed lifetime in ms. Must be set together with `to`.
+        #[serde(default)]
+        lifetime_ms: Option<u64>,
     },
     /// 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<BeliefOut>,
     },
-    /// Confirmed deliveries drained from the exchange.
+    /// Confirmed deliveries, contact changes, and counted drops drained from
+    /// the exchange.
     Events {
         /// The drained confirmed deliveries.
         events: Vec<EventOut>,
+        /// The drained contact changes (joined or lost), in order.
+        #[serde(default)]
+        contacts: Vec<ContactOut>,
+        /// The drained counted drops.
+        #[serde(default)]
+        drops: Vec<DropOut>,
     },
     /// 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<ObservationOut>,
+        /// `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<S, T>(
 ) -> Response
 where
     S: Store + 'static,
-    T: Transport,
+    T: Transport<Addr = NodeId>,
 {
     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<T>(node: &mut Node<LogStore, T>, host: &mut HostState)
 where
-    T: Transport + 'static,
+    T: Transport<Addr = NodeId> + 'static,
 {
     while node.poll(host.clock.now()) > 0 && host.rounds < POLL_CAP {
         host.rounds += 1;
@@ -308,7 +451,7 @@ pub fn serve<T>(
     peer_time_window_ms: u64,
 ) -> std::io::Result<()>
 where
-    T: Transport + 'static,
+    T: Transport<Addr = NodeId> + '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<T>(
     host: &mut HostState,
 ) -> io::Result<()>
 where
-    T: Transport + 'static,
+    T: Transport<Addr = NodeId> + 'static,
 {
     loop {
         match read_request(stream) {

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,
+};

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:?}"),
+    }
+}