From 431e719eb29d79e77b8c2334f6b925ad787227d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sat, 7 Nov 2026 20:28:59 +0100 Subject: [PATCH 1/2] feat(ws)!: late-join snapshots from the record's current value (#434) WebSocket kept a per-(record, topic) cache of the last bytes it sent and served late-join snapshots from it. Snapshots now come from the record's current value, rendered through the link's topic writer and serializer when a client subscribes, as AimX does. - aimdb-core: new `OutboundLatest`, each outbound link's current value rendered on demand, with the same route ids as `OutboundRoutes`. Each `ConnectorLink` carries the renderer next to its route factory. - aimdb-websocket-connector: the snapshot cache, `SnapshotProvider`, `NoSnapshot` and `server/session.rs` are gone; `run_broadcast` only fans out. Behaviour changes: `SpmcRing` records send no snapshot, and a record with a topic writer sends one only on its current topic. Co-Authored-By: Claude Opus 5.5 --- aimdb-core/CHANGELOG.md | 5 + aimdb-core/src/connector.rs | 19 ++- aimdb-core/src/lib.rs | 3 +- aimdb-core/src/outbound/latest.rs | 135 ++++++++++++++++++ aimdb-core/src/outbound/mod.rs | 3 + aimdb-core/src/typed_api.rs | 58 +++++++- aimdb-websocket-connector/CHANGELOG.md | 10 ++ .../src/server/builder.rs | 78 +++------- .../src/server/dispatch.rs | 35 +++-- aimdb-websocket-connector/src/server/mod.rs | 1 - .../src/server/session.rs | 43 ------ aimdb-websocket-connector/tests/e2e.rs | 82 +++++++++-- 12 files changed, 344 insertions(+), 128 deletions(-) create mode 100644 aimdb-core/src/outbound/latest.rs delete mode 100644 aimdb-websocket-connector/src/server/session.rs diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index 4eb905f7..618688a2 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -26,6 +26,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `RecordRegistrar::record_key()` (#426), so extension crates can name per-record state after the record. `aimdb-persistence` uses it. +- **`OutboundLatest` (#434):** each outbound link's current value, rendered on + demand through the link's topic writer and serializer, with the same route + ids as `OutboundRoutes`. Connectors build late-join snapshots from it + instead of caching what they sent. Records without a current value + (`SpmcRing`, or nothing written yet) render nothing. ### Changed (wire) diff --git a/aimdb-core/src/connector.rs b/aimdb-core/src/connector.rs index 2a78ffbf..2b097a2a 100644 --- a/aimdb-core/src/connector.rs +++ b/aimdb-core/src/connector.rs @@ -481,6 +481,13 @@ pub struct ConnectorLink { /// Builds the link's route for `OutboundRoutes`. pub(crate) route_factory: crate::outbound::RouteFactoryFn, + + /// Renders the record's current value as this link would send it, for + /// `OutboundLatest`. + pub(crate) latest_fn: crate::outbound::LatestFn, + + /// Whether a topic writer picks the topic per value. + pub(crate) has_topic_writer: bool, } impl Debug for ConnectorLink { @@ -493,12 +500,20 @@ impl Debug for ConnectorLink { } impl ConnectorLink { - /// Creates a new connector link from a link address and route factory - pub(crate) fn new(url: LinkAddress, route_factory: crate::outbound::RouteFactoryFn) -> Self { + /// Creates a new connector link from a link address, its route factory + /// and its latest-value renderer. + pub(crate) fn new( + url: LinkAddress, + route_factory: crate::outbound::RouteFactoryFn, + latest_fn: crate::outbound::LatestFn, + has_topic_writer: bool, + ) -> Self { Self { url, config: Vec::new(), route_factory, + latest_fn, + has_topic_writer, } } } diff --git a/aimdb-core/src/lib.rs b/aimdb-core/src/lib.rs index 1c83c1fc..44525328 100644 --- a/aimdb-core/src/lib.rs +++ b/aimdb-core/src/lib.rs @@ -191,7 +191,8 @@ pub use connector::{TopicBuf, TopicOverflow, TopicWriter}; // Router exports for connector implementations pub use inbound_dispatch::{InboundDispatch, InboundRouteInfo}; pub use outbound::{ - OutboundMessage, OutboundPayload, OutboundRoutes, RouteId, RouteInfo, RouteStats, + LatestMessage, OutboundLatest, OutboundMessage, OutboundPayload, OutboundRoutes, RouteId, + RouteInfo, RouteStats, }; // Topic grammar for connectors with wildcard subscriptions diff --git a/aimdb-core/src/outbound/latest.rs b/aimdb-core/src/outbound/latest.rs new file mode 100644 index 00000000..fc54ca3a --- /dev/null +++ b/aimdb-core/src/outbound/latest.rs @@ -0,0 +1,135 @@ +//! [`OutboundLatest`]: each outbound link's current value, rendered on demand. +//! +//! A connector that sends late-join snapshots needs "what would this link send +//! for the record's current value". [`OutboundRoutes`](super::OutboundRoutes) +//! is owned by the connector's transport task and only sees new writes, so +//! this is a separate, cheap-to-clone view that reads the record's latest value +//! and runs it through the link's topic writer and serializer. It keeps no +//! state of its own. + +use alloc::string::String; +use alloc::sync::Arc; +use alloc::vec::Vec; + +use super::RouteId; +use crate::builder::AimDb; +use crate::RuntimeContext; + +/// One link's current value, as the link would send it. +pub(crate) struct Rendered { + /// Written by the link's topic writer; `None` for the link's default topic. + pub(crate) topic: Option, + pub(crate) payload: Vec, + pub(crate) metadata: crate::WriteMetadata, +} + +/// Renders a link's current value. `None` if the record has no current value +/// (none written yet, or an `SpmcRing` buffer), or if its topic does not fit +/// or serializing it fails. +pub(crate) type LatestFn = Arc Option + Send + Sync>; + +/// A link's current value, rendered by [`OutboundLatest::render`]. +#[derive(Debug, Clone)] +pub struct LatestMessage { + /// Index of the record the link belongs to (its `RecordId`), as in + /// [`RouteInfo::record_index`](super::RouteInfo::record_index). + pub record_index: usize, + /// The topic the link sends this value on. + pub topic: String, + /// The serialized value. + pub payload: Vec, + /// Metadata of the write that stored the value. + pub metadata: crate::WriteMetadata, +} + +impl LatestMessage { + /// When the value was written, on the instance clock. + pub fn stamp(&self) -> crate::Stamp { + self.metadata.stamp() + } +} + +struct LatestRoute { + record_index: usize, + default_topic: Arc, + has_topic_writer: bool, + render: LatestFn, +} + +/// The current value of every outbound link of one scheme, rendered on +/// demand. Route ids are the same as in an +/// [`OutboundRoutes`](super::OutboundRoutes) built for the same scheme. +/// +/// Only records with a current value render: an `SpmcRing` buffer has none, +/// so its links never do. +#[derive(Clone)] +pub struct OutboundLatest { + db: AimDb, + ctx: RuntimeContext, + routes: Arc<[LatestRoute]>, +} + +// Shared by every session of a connector. +const _: fn() = || { + fn assert_send_sync() {} + assert_send_sync::(); +}; + +impl OutboundLatest { + /// Every outbound link of `scheme`, in route order. + pub fn new(db: &AimDb, scheme: &str) -> Self { + let routes: Vec = db + .outbound_links(scheme) + .map(|(record_index, _, link)| LatestRoute { + record_index, + default_topic: Arc::from(link.url.resource_id()), + has_topic_writer: link.has_topic_writer, + render: link.latest_fn.clone(), + }) + .collect(); + Self { + db: db.clone(), + ctx: db.runtime_ctx(), + routes: routes.into(), + } + } + + /// Number of routes; ids are `0..len()`. + pub fn len(&self) -> usize { + self.routes.len() + } + + /// Whether the scheme has no outbound links. + pub fn is_empty(&self) -> bool { + self.routes.is_empty() + } + + /// Index of the record route `id` belongs to. + pub fn record_index(&self, id: RouteId) -> Option { + self.routes.get(id).map(|r| r.record_index) + } + + /// The topic route `id` always sends on, or `None` if a topic writer + /// picks it per value. Lets a caller skip rendering routes whose topic + /// cannot match. + pub fn fixed_topic(&self, id: RouteId) -> Option<&str> { + let route = self.routes.get(id)?; + (!route.has_topic_writer).then_some(&*route.default_topic) + } + + /// Route `id`'s current value, through its topic writer and serializer. + /// `None` if the record has no current value, or the value's topic does + /// not fit or it fails to serialize (the live path reports those). + pub fn render(&self, id: RouteId) -> Option { + let route = self.routes.get(id)?; + let rendered = (route.render)(&self.db, &self.ctx)?; + Some(LatestMessage { + record_index: route.record_index, + topic: rendered + .topic + .unwrap_or_else(|| String::from(&*route.default_topic)), + payload: rendered.payload, + metadata: rendered.metadata, + }) + } +} diff --git a/aimdb-core/src/outbound/mod.rs b/aimdb-core/src/outbound/mod.rs index 2caa8b52..bda1f329 100644 --- a/aimdb-core/src/outbound/mod.rs +++ b/aimdb-core/src/outbound/mod.rs @@ -2,9 +2,12 @@ pub(crate) mod events; +mod latest; mod ready; mod routes; +pub(crate) use latest::{LatestFn, Rendered}; +pub use latest::{LatestMessage, OutboundLatest}; pub use routes::{OutboundMessage, OutboundPayload, OutboundRoutes, RouteInfo, RouteStats}; pub(crate) use routes::{ PollRoute, RouteFactoryFn, RouteOutcome, RouteParts, SerializeFailure, StagedPayload, diff --git a/aimdb-core/src/typed_api.rs b/aimdb-core/src/typed_api.rs index 5e643ee1..c978e658 100644 --- a/aimdb-core/src/typed_api.rs +++ b/aimdb-core/src/typed_api.rs @@ -1589,9 +1589,63 @@ where }) }; + let topic_writer = link.topic_writer; + let has_topic_writer = topic_writer.is_some(); + + // Renders the record's current value through this link's topic writer + // and serializer, for late-join snapshots (`OutboundLatest`). + let latest_fn: crate::outbound::LatestFn = { + let record_key = self.record_key.clone(); + let topic_writer = topic_writer.clone(); + let serialize = serialize.clone(); + let serialize_into = serialize_into.clone(); + Arc::new(move |db: &AimDb, ctx: &crate::RuntimeContext| { + let sample = db + .inner() + .get_typed_record_by_key::(&record_key) + .ok()? + .latest()? + .into_stamped(); + + let topic = match &topic_writer { + None => None, + Some((capacity, writer)) => { + let mut buf = alloc::vec![0u8; *capacity]; + let mut out = crate::connector::TopicBuf::new(&mut buf); + let written = write_topic(&**writer, &sample.value, &mut out).ok()?; + let len = out.len(); + // `TopicBuf` only ever holds whole `&str`s. + written + .then(|| String::from_utf8(buf[..len].to_vec()).ok()) + .flatten() + } + }; + + let mut scratch = + alloc::vec![0u8; serialize_into.as_ref().map_or(0, |(capacity, _)| *capacity)]; + let payload = match serialize_outbound( + &serialize, + serialize_into.as_ref().map(|(_, f)| f), + ctx, + &sample.value, + &mut scratch, + ) + .ok()? + { + crate::outbound::StagedPayload::Scratch(len) => scratch[..len].to_vec(), + crate::outbound::StagedPayload::Owned(bytes) => bytes, + }; + + Some(crate::outbound::Rendered { + topic, + payload, + metadata: *sample.metadata(), + }) + }) + }; + // Subscribed when `OutboundRoutes` is built. let route_factory: crate::outbound::RouteFactoryFn = { - let topic_writer = link.topic_writer; Arc::new(move |db: &AimDb| { let (topic_capacity, writer) = match &topic_writer { Some((capacity, writer)) => (*capacity, Some(writer.clone())), @@ -1612,7 +1666,7 @@ where // Named `out`, not `link`: `out.config = link.config` must read the // user's link, not the new one. - let mut out = ConnectorLink::new(url, route_factory); + let mut out = ConnectorLink::new(url, route_factory, latest_fn, has_topic_writer); out.config = link.config; // Routes are built later, when a connector builds its `OutboundRoutes`. diff --git a/aimdb-websocket-connector/CHANGELOG.md b/aimdb-websocket-connector/CHANGELOG.md index 75927556..982e9169 100644 --- a/aimdb-websocket-connector/CHANGELOG.md +++ b/aimdb-websocket-connector/CHANGELOG.md @@ -22,6 +22,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **The `record.query` reply drops `total` (#429)** and is `{"records": [...]}`, as on AimX. Use the length of `records` instead. Rows from the `QueryHandlerFn` are filtered by grant directly, without a JSON round trip. +- **Late-join snapshots come from the record's current value (#434),** not + from a cache of what the server last sent. A snapshot is the record's + current value, serialized by its `ws://` link, on the topic that value + produces. Two consequences: + - `SpmcRing` records send no snapshot, as on AimX. Use `SingleLatest` for + records whose subscribers need the current value. + - With a topic writer, a record sends a snapshot only on its current topic, + not on every topic it has published to. + + The per-topic cache, which grew with every distinct topic, is gone. ## [0.4.0] - 2026-11-03 diff --git a/aimdb-websocket-connector/src/server/builder.rs b/aimdb-websocket-connector/src/server/builder.rs index 1a274754..fc228012 100644 --- a/aimdb-websocket-connector/src/server/builder.rs +++ b/aimdb-websocket-connector/src/server/builder.rs @@ -15,27 +15,25 @@ //! ``` use std::{ - collections::HashMap, net::{SocketAddr, ToSocketAddrs}, pin::Pin, - sync::{Arc, Mutex}, + sync::Arc, time::Instant, }; use aimdb_data_contracts::Streamable; -use aimdb_core::{ConnectorBuilder, Dispatch, ExactGrammar, InboundDispatch, OutboundRoutes}; +use aimdb_core::{ + ConnectorBuilder, Dispatch, ExactGrammar, InboundDispatch, OutboundLatest, OutboundRoutes, +}; use axum::Router as AxumRouter; -use aimdb_core::topic_matches; - use super::{ auth::{AuthHandler, DynAuthHandler, NoAuth}, client_manager::ClientManager, dispatch::WsDispatch, http::{build_server_future, ServerState}, registry::StreamableRegistry, - session::{NoSnapshot, Snapshot, SnapshotProvider}, }; // ════════════════════════════════════════════════════════════════════ @@ -145,9 +143,11 @@ impl WebSocketConnectorBuilder { /// Enable or disable late-join snapshots (default: `true`). /// - /// When enabled, a client that subscribes to a topic immediately receives - /// the current value (if one is available) as a `snapshot` message before - /// live `data` pushes start. + /// When enabled, a client that subscribes receives a `snap` frame for each + /// matching `ws://` link before live events start. The snapshot is the + /// record's current value, serialized by the link, on the topic that value + /// produces. A record without a current value (nothing written yet, or an + /// `SpmcRing` buffer) sends none. pub fn with_late_join(mut self, enabled: bool) -> Self { self.late_join = enabled; self @@ -285,18 +285,11 @@ impl ConnectorBuilder for WebSocketConnectorBuilder { // bus checks client permissions against. let outbound = OutboundRoutes::new(db, "ws")?; - // ── Late-join snapshot cache (only when enabled) ────── - let snapshot_map: Option = - self.late_join.then(|| Arc::new(Mutex::new(HashMap::new()))); - // ── Client manager ──────────────────────────────────── let client_mgr = ClientManager::new(self.channel_capacity.max(1)); - // ── Build snapshot provider ────────────────────────── - let snapshot_provider: Arc = match &snapshot_map { - Some(map) => Arc::new(DynMapSnapshot(map.clone())), - None => Arc::new(NoSnapshot), - }; + // ── Late-join snapshots: each link's current value ─── + let latest = OutboundLatest::new(db, "ws"); // ── Schema names for record.list enrichment ─────────── // Core builds each `record.list` row but can't map a `TypeId` to a @@ -317,7 +310,7 @@ impl ConnectorBuilder for WebSocketConnectorBuilder { let dispatch: Arc = Arc::new(WsDispatch { db: db.clone(), client_mgr: client_mgr.clone(), - snapshot_provider, + latest, inbound, schema_by_type, auth: self.auth.clone(), @@ -325,7 +318,7 @@ impl ConnectorBuilder for WebSocketConnectorBuilder { }); // ── Outbound: one broadcast loop pulls records → bus ──── - let broadcast = run_broadcast(outbound, client_mgr.clone(), snapshot_map); + let broadcast = run_broadcast(outbound, client_mgr.clone()); // ── Build Axum server future ────────────────────────── let state = ServerState { @@ -351,51 +344,18 @@ impl ConnectorBuilder for WebSocketConnectorBuilder { } } -/// Shared late-join cache: (RecordId, topic) → last serialized bytes and the -/// stamp of the write they came from. -pub(crate) type SnapshotCache = Arc, aimdb_core::Stamp)>>>; - /// Pull every outbound record update and fan it out to subscribed clients. /// /// The bus carries raw record-value bytes tagged with the topic, the /// record's index and the value's write stamp; each connection's AimX codec -/// applies the `event` envelope (with `ts`) downstream. With late-join on, the latest bytes per (record, topic) are -/// cached for snapshots first. Returns once every outbound route has closed. -async fn run_broadcast( - mut outbound: OutboundRoutes, - client_mgr: ClientManager, - snapshot: Option, -) { +/// applies the `event` envelope (with `ts`) downstream. Returns once every +/// outbound route has closed. +async fn run_broadcast(mut outbound: OutboundRoutes, client_mgr: ClientManager) { while let Some(msg) = outbound.next().await { let index = msg.route.record_index; let ts = msg.stamp(); - let bytes = msg.payload.as_slice(); - if let Some(map) = &snapshot { - map.lock() - .unwrap() - .insert((index, msg.topic.to_string()), (bytes.to_vec(), ts)); - } - client_mgr.broadcast(msg.topic, index, bytes, ts).await; - } -} - -// ════════════════════════════════════════════════════════════════════ -// Dynamic snapshot provider backed by the shared Mutex -// ════════════════════════════════════════════════════════════════════ - -/// Wrapper for SnapshotCache -struct DynMapSnapshot(SnapshotCache); - -impl SnapshotProvider for DynMapSnapshot { - fn snapshots(&self, pattern: &str) -> Vec { - let Ok(map) = self.0.lock() else { - return Vec::new(); - }; - map.iter() - .filter(|((_, topic), _)| topic_matches(pattern, topic)) - .map(|((record_id, topic), (bytes, ts))| { - (*record_id, topic.clone(), bytes.clone(), *ts) - }) - .collect() + client_mgr + .broadcast(msg.topic, index, msg.payload.as_slice(), ts) + .await; } } diff --git a/aimdb-websocket-connector/src/server/dispatch.rs b/aimdb-websocket-connector/src/server/dispatch.rs index 660ef590..91c9d3ed 100644 --- a/aimdb-websocket-connector/src/server/dispatch.rs +++ b/aimdb-websocket-connector/src/server/dispatch.rs @@ -28,7 +28,6 @@ use serde_json::Value; use super::{ auth::{AuthHandler, ClientId, ClientInfo, Permissions, RecordsBits}, client_manager::ClientManager, - session::SnapshotProvider, }; /// The shared WS dispatch — one `Arc` per server. @@ -37,7 +36,8 @@ pub struct WsDispatch { /// `QueryHandlerFn` for `record.query`. pub(crate) db: aimdb_core::builder::AimDb, pub(crate) client_mgr: ClientManager, - pub(crate) snapshot_provider: Arc, + /// Each `ws://` link's current value, for late-join snapshots. + pub(crate) latest: aimdb_core::OutboundLatest, pub(crate) inbound: aimdb_core::InboundDispatch, /// Record `type_id` string → data-contract schema name, used to stamp /// `schema_type` onto the `record.list` rows core hands back. @@ -75,7 +75,7 @@ impl Dispatch for WsDispatch { Box::new(WsSession { db: self.db.clone(), client_mgr: self.client_mgr.clone(), - snapshot_provider: self.snapshot_provider.clone(), + latest: self.latest.clone(), inbound: self.inbound.clone(), schema_by_type: self.schema_by_type.clone(), auth: self.auth.clone(), @@ -89,7 +89,7 @@ impl Dispatch for WsDispatch { struct WsSession { db: aimdb_core::builder::AimDb, client_mgr: ClientManager, - snapshot_provider: Arc, + latest: aimdb_core::OutboundLatest, inbound: aimdb_core::InboundDispatch, schema_by_type: Arc>, auth: Arc, @@ -185,12 +185,27 @@ impl Session for WsSession { return Vec::new(); } - // Filtered by client's record permissions - self.snapshot_provider - .snapshots(topic) - .into_iter() - .filter(|(record_id, _, _, _)| self.info.record_perms.is_allowed(*record_id)) - .map(|(_, topic, bytes, ts)| (topic, Payload::from(bytes.as_slice()), ts)) + // Each link the client may read, rendered from its record's current + // value. A link whose topic is fixed is skipped before rendering when + // that topic cannot match. + let latest = &self.latest; + (0..latest.len()) + .filter(|&id| { + latest + .record_index(id) + .is_some_and(|index| self.info.record_perms.is_allowed(index)) + }) + .filter(|&id| { + latest + .fixed_topic(id) + .is_none_or(|fixed| aimdb_core::topic_matches(topic, fixed)) + }) + .filter_map(|id| latest.render(id)) + .filter(|msg| aimdb_core::topic_matches(topic, &msg.topic)) + .map(|msg| { + let ts = msg.stamp(); + (msg.topic, Payload::from(msg.payload), ts) + }) .collect() } diff --git a/aimdb-websocket-connector/src/server/mod.rs b/aimdb-websocket-connector/src/server/mod.rs index 454a44a0..20c771ac 100644 --- a/aimdb-websocket-connector/src/server/mod.rs +++ b/aimdb-websocket-connector/src/server/mod.rs @@ -14,4 +14,3 @@ pub mod client_manager; pub(crate) mod dispatch; pub(crate) mod http; pub(crate) mod registry; -pub(crate) mod session; diff --git a/aimdb-websocket-connector/src/server/session.rs b/aimdb-websocket-connector/src/server/session.rs deleted file mode 100644 index 01276fc2..00000000 --- a/aimdb-websocket-connector/src/server/session.rs +++ /dev/null @@ -1,43 +0,0 @@ -//! Reusable handler traits for the WebSocket dispatch. -//! -//! The WS server rides `run_session` ([`aimdb_core::session::run_session`]) via -//! [`super::dispatch`]; what lives here is the pluggable application surface the -//! dispatch consumes: [`SnapshotProvider`] supplies the late-join current values -//! for a subscription pattern. -//! -//! `record.query` has no WebSocket-specific hook. It is answered by the -//! `QueryHandlerFn` in the database's Extensions (registered by -//! `aimdb-persistence::with_persistence`), the same handler every transport uses. -//! -//! `record.list` rows are core's [`RecordMetadata`](aimdb_core::remote::RecordMetadata), -//! shared with every other transport; the dispatch only stamps in the schema -//! name core can't resolve. - -// ════════════════════════════════════════════════════════════════════ -// Snapshot provider (late-join) -// ════════════════════════════════════════════════════════════════════ - -/// Provides the current serialized values covered by a subscription pattern for -/// late-join snapshots (one `(record id, topic, value)` tuple per covered pair of record - topic -/// — a wildcard pattern may cover several; an exact topic may match several records). -/// The returned snapshots need record id, different clients may have different -/// read permissions to different records -pub trait SnapshotProvider: Send + Sync + 'static { - /// Return the latest serialized values for every topic matching `pattern`. - /// As record keys now rule the broadcasting instead of topic, - /// return must be tuple of (record index, topic, payload, write stamp) - fn snapshots(&self, pattern: &str) -> Vec; -} - -/// One late-join snapshot: record index, topic, serialized value, and the -/// stamp of the write the value came from (AimX 4.0 `ts`). -pub type Snapshot = (usize, String, Vec, aimdb_core::Stamp); - -/// A snapshot provider that always returns nothing (late-join disabled or no data). -pub struct NoSnapshot; - -impl SnapshotProvider for NoSnapshot { - fn snapshots(&self, _pattern: &str) -> Vec { - Vec::new() - } -} diff --git a/aimdb-websocket-connector/tests/e2e.rs b/aimdb-websocket-connector/tests/e2e.rs index 6fd3ebd5..86dd75ad 100644 --- a/aimdb-websocket-connector/tests/e2e.rs +++ b/aimdb-websocket-connector/tests/e2e.rs @@ -162,6 +162,26 @@ async fn spawn(ws: WebSocketConnector) -> (SocketAddr, Arc) { }) .finish(); }); + // Records with a current value, which late-join snapshots are built from + // (a ring like `inject` has none). Each sends on its own key. + for key in LATEST { + sb.configure::(key, |reg| { + reg.buffer(BufferCfg::SingleLatest) + .with_remote_access() + .link_to(&format!("ws://{key}")) + .with_serializer(|_ctx, v: &Value| Ok(serde_json::to_vec(v).unwrap())) + .finish(); + }); + } + // A current value whose topic its writer picks per value. + sb.configure::("moving", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .with_remote_access() + .link_to("ws://_") + .with_topic_writer(128, InjectTopic) + .with_serializer(|_ctx, m: &Inject| Ok(serde_json::to_vec(&m.payload).unwrap())) + .finish(); + }); let (db, runner) = sb.build().await.expect("build server db"); let db = Arc::new(db); tokio::spawn(runner.run()); @@ -174,6 +194,14 @@ async fn spawn_default() -> (SocketAddr, Arc) { spawn(WebSocketConnector::new().with_late_join(true)).await } +/// The fixture's `SingleLatest` records, each linked to `ws://`. +const LATEST: [&str; 3] = ["sensors.temp", "sensors.humidity", "t"]; + +/// Write `value` to one of the [`LATEST`] records. +fn set(db: &AimDb, key: &str, value: Value) { + db.set_record_from_json(key, value).expect("set"); +} + /// Push `payload` to subscribers of `topic` (one outbound record update). fn inject(db: &AimDb, topic: &str, payload: Value) { db.set_record_from_json("inject", json!({ "topic": topic, "payload": payload })) @@ -345,9 +373,8 @@ async fn server_two_subscriptions_and_unsubscribe() { #[tokio::test] async fn server_late_join_snapshot() { let (addr, db) = spawn_default().await; - // Produce the value first so the late-join cache holds it, then subscribe. - inject(&db, "sensors.temp", json!(99)); - tokio::time::sleep(Duration::from_millis(100)).await; + // The snapshot is the record's current value at subscribe time. + set(&db, "sensors.temp", json!(99)); let mut c = ws_connect(addr).await; ws_send(&mut c, json!({"t":"sub","id":4,"topic":"sensors.temp"})).await; @@ -380,14 +407,13 @@ fn without_ts(mut frame: Value) -> (Value, u64) { #[tokio::test] async fn server_wildcard_late_join_snapshots_per_match() { let (addr, db) = spawn_default().await; - inject(&db, "sensors.temp", json!(1)); - inject(&db, "sensors.humidity", json!(2)); - tokio::time::sleep(Duration::from_millis(100)).await; + set(&db, "sensors.temp", json!(1)); + set(&db, "sensors.humidity", json!(2)); let mut c = ws_connect(addr).await; ws_send(&mut c, json!({"t":"sub","id":9,"topic":"sensors.#"})).await; assert_eq!(ws_recv(&mut c).await, json!({"t":"subscribed","sub":"9"})); - // One snapshot per cached record under the pattern (order is map order), + // One snapshot per record under the pattern (in route order), // numbered 1..=N with only the final one flagged `last` — that flag is what // lets a client close out the burst without waiting for a live event. let mut snaps = Vec::new(); @@ -963,13 +989,49 @@ async fn stalled_client_does_not_block_a_healthy_one() { let _ = stalled.close(None).await; } +/// A ring has no current value, so its links send no snapshot: the first +/// frame after the ack is the next live event. +#[tokio::test] +async fn ring_records_send_no_snapshot() { + let (addr, db) = spawn_default().await; + inject(&db, "ring.x", json!(1)); + tokio::time::sleep(Duration::from_millis(100)).await; + + let mut c = ws_connect(addr).await; + ws_send(&mut c, json!({"t":"sub","id":1,"topic":"ring.x"})).await; + assert_eq!(ws_recv(&mut c).await, json!({"t":"subscribed","sub":"1"})); + inject(&db, "ring.x", json!(2)); + let (event, _ts) = without_ts(ws_recv(&mut c).await); + assert_eq!( + event, + json!({"t":"event","sub":"1","seq":1,"topic":"ring.x","data":2}) + ); +} + +/// With a topic writer, the snapshot is the current value on the topic that +/// value produces; topics it published on earlier get none. +#[tokio::test] +async fn topic_writer_snapshot_is_on_the_current_topic() { + let (addr, db) = spawn_default().await; + set(&db, "moving", json!({"topic": "m.a", "payload": 1})); + set(&db, "moving", json!({"topic": "m.b", "payload": 2})); + + let mut c = ws_connect(addr).await; + ws_send(&mut c, json!({"t":"sub","id":1,"topic":"m.#"})).await; + assert_eq!(ws_recv(&mut c).await, json!({"t":"subscribed","sub":"1"})); + let (snap, _ts) = without_ts(ws_recv(&mut c).await); + assert_eq!( + snap, + json!({"t":"snap","sub":"1","seq":1,"last":true,"topic":"m.b","data":2}) + ); +} + // ── Golden wire frames (locks the exact on-the-wire JSON shape) ────── #[tokio::test] async fn golden_wire_frames() { let (addr, db) = spawn_default().await; - inject(&db, "t", json!(5)); // seed the late-join cache - tokio::time::sleep(Duration::from_millis(100)).await; + set(&db, "t", json!(5)); // the late-join snapshot's value let mut c = ws_connect(addr).await; ws_send(&mut c, json!({"t":"sub","id":1,"topic":"t"})).await; @@ -983,7 +1045,7 @@ async fn golden_wire_frames() { // The event continues the snapshot burst's `seq` (one snapshot, so `seq:2`) // — a single counter over the whole subscription, so loss anywhere in it is // visible as a gap. - inject(&db, "t", json!(42)); + set(&db, "t", json!(42)); let (event, event_ts) = without_ts(ws_recv(&mut c).await); assert_eq!( event, From e15a9eabcd796b4fd5f77568c0143b3c7c0d945f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sat, 7 Nov 2026 20:45:35 +0100 Subject: [PATCH 2/2] perf(core): resolve a snapshot's topic before serializing (#434) `OutboundLatest::render_if(id, accept)` runs the link's topic writer, checks the topic with `accept`, and serializes only on a match. A link with a fixed topic is checked before its value is read. Replaces the separate `fixed_topic` pre-filter, so a WebSocket subscribe never serializes links its pattern does not match. Co-Authored-By: Claude Opus 5.5 --- aimdb-core/src/outbound/latest.rs | 36 +++++--- aimdb-core/src/typed_api.rs | 89 ++++++++++--------- .../src/server/dispatch.rs | 13 +-- aimdb-websocket-connector/tests/e2e.rs | 48 ++++++++++ 4 files changed, 122 insertions(+), 64 deletions(-) diff --git a/aimdb-core/src/outbound/latest.rs b/aimdb-core/src/outbound/latest.rs index fc54ca3a..708483cf 100644 --- a/aimdb-core/src/outbound/latest.rs +++ b/aimdb-core/src/outbound/latest.rs @@ -23,10 +23,16 @@ pub(crate) struct Rendered { pub(crate) metadata: crate::WriteMetadata, } -/// Renders a link's current value. `None` if the record has no current value -/// (none written yet, or an `SpmcRing` buffer), or if its topic does not fit -/// or serializing it fails. -pub(crate) type LatestFn = Arc Option + Send + Sync>; +/// Renders a link's current value. The topic is resolved first and passed to +/// `accept` (`None`: the link's default topic); the value is serialized only +/// if `accept` returns `true`. `None` if the record has no current value (none +/// written yet, or an `SpmcRing` buffer), `accept` rejects the topic, or the +/// topic does not fit or serializing fails. +pub(crate) type LatestFn = Arc< + dyn Fn(&AimDb, &RuntimeContext, &dyn Fn(Option<&str>) -> bool) -> Option + + Send + + Sync, +>; /// A link's current value, rendered by [`OutboundLatest::render`]. #[derive(Debug, Clone)] @@ -109,20 +115,24 @@ impl OutboundLatest { self.routes.get(id).map(|r| r.record_index) } - /// The topic route `id` always sends on, or `None` if a topic writer - /// picks it per value. Lets a caller skip rendering routes whose topic - /// cannot match. - pub fn fixed_topic(&self, id: RouteId) -> Option<&str> { - let route = self.routes.get(id)?; - (!route.has_topic_writer).then_some(&*route.default_topic) - } - /// Route `id`'s current value, through its topic writer and serializer. /// `None` if the record has no current value, or the value's topic does /// not fit or it fails to serialize (the live path reports those). pub fn render(&self, id: RouteId) -> Option { + self.render_if(id, |_| true) + } + + /// [`render`](Self::render), but only if `accept` takes the value's topic. + /// The value is serialized only after `accept` returns `true`, and a link + /// without a topic writer is checked before its value is even read, so + /// rejected routes cost next to nothing. + pub fn render_if(&self, id: RouteId, accept: impl Fn(&str) -> bool) -> Option { let route = self.routes.get(id)?; - let rendered = (route.render)(&self.db, &self.ctx)?; + if !route.has_topic_writer && !accept(&route.default_topic) { + return None; + } + let accept_topic = |topic: Option<&str>| accept(topic.unwrap_or(&route.default_topic)); + let rendered = (route.render)(&self.db, &self.ctx, &accept_topic)?; Some(LatestMessage { record_index: route.record_index, topic: rendered diff --git a/aimdb-core/src/typed_api.rs b/aimdb-core/src/typed_api.rs index c978e658..50a6789c 100644 --- a/aimdb-core/src/typed_api.rs +++ b/aimdb-core/src/typed_api.rs @@ -1599,49 +1599,56 @@ where let topic_writer = topic_writer.clone(); let serialize = serialize.clone(); let serialize_into = serialize_into.clone(); - Arc::new(move |db: &AimDb, ctx: &crate::RuntimeContext| { - let sample = db - .inner() - .get_typed_record_by_key::(&record_key) - .ok()? - .latest()? - .into_stamped(); - - let topic = match &topic_writer { - None => None, - Some((capacity, writer)) => { - let mut buf = alloc::vec![0u8; *capacity]; - let mut out = crate::connector::TopicBuf::new(&mut buf); - let written = write_topic(&**writer, &sample.value, &mut out).ok()?; - let len = out.len(); - // `TopicBuf` only ever holds whole `&str`s. - written - .then(|| String::from_utf8(buf[..len].to_vec()).ok()) - .flatten() + Arc::new( + move |db: &AimDb, + ctx: &crate::RuntimeContext, + accept: &dyn Fn(Option<&str>) -> bool| { + let sample = db + .inner() + .get_typed_record_by_key::(&record_key) + .ok()? + .latest()? + .into_stamped(); + + let topic = match &topic_writer { + None => None, + Some((capacity, writer)) => { + let mut buf = alloc::vec![0u8; *capacity]; + let mut out = crate::connector::TopicBuf::new(&mut buf); + let written = write_topic(&**writer, &sample.value, &mut out).ok()?; + let len = out.len(); + // `TopicBuf` only ever holds whole `&str`s. + written + .then(|| String::from_utf8(buf[..len].to_vec()).ok()) + .flatten() + } + }; + // Serialize only for a topic the caller wants. + if !accept(topic.as_deref()) { + return None; } - }; - - let mut scratch = - alloc::vec![0u8; serialize_into.as_ref().map_or(0, |(capacity, _)| *capacity)]; - let payload = match serialize_outbound( - &serialize, - serialize_into.as_ref().map(|(_, f)| f), - ctx, - &sample.value, - &mut scratch, - ) - .ok()? - { - crate::outbound::StagedPayload::Scratch(len) => scratch[..len].to_vec(), - crate::outbound::StagedPayload::Owned(bytes) => bytes, - }; - Some(crate::outbound::Rendered { - topic, - payload, - metadata: *sample.metadata(), - }) - }) + let mut scratch = alloc::vec![0u8; serialize_into.as_ref().map_or(0, |(capacity, _)| *capacity)]; + let payload = match serialize_outbound( + &serialize, + serialize_into.as_ref().map(|(_, f)| f), + ctx, + &sample.value, + &mut scratch, + ) + .ok()? + { + crate::outbound::StagedPayload::Scratch(len) => scratch[..len].to_vec(), + crate::outbound::StagedPayload::Owned(bytes) => bytes, + }; + + Some(crate::outbound::Rendered { + topic, + payload, + metadata: *sample.metadata(), + }) + }, + ) }; // Subscribed when `OutboundRoutes` is built. diff --git a/aimdb-websocket-connector/src/server/dispatch.rs b/aimdb-websocket-connector/src/server/dispatch.rs index 91c9d3ed..4e08a46b 100644 --- a/aimdb-websocket-connector/src/server/dispatch.rs +++ b/aimdb-websocket-connector/src/server/dispatch.rs @@ -185,9 +185,8 @@ impl Session for WsSession { return Vec::new(); } - // Each link the client may read, rendered from its record's current - // value. A link whose topic is fixed is skipped before rendering when - // that topic cannot match. + // Each link the client may read whose topic matches, rendered from its + // record's current value. Non-matching links are never serialized. let latest = &self.latest; (0..latest.len()) .filter(|&id| { @@ -195,13 +194,7 @@ impl Session for WsSession { .record_index(id) .is_some_and(|index| self.info.record_perms.is_allowed(index)) }) - .filter(|&id| { - latest - .fixed_topic(id) - .is_none_or(|fixed| aimdb_core::topic_matches(topic, fixed)) - }) - .filter_map(|id| latest.render(id)) - .filter(|msg| aimdb_core::topic_matches(topic, &msg.topic)) + .filter_map(|id| latest.render_if(id, |t| aimdb_core::topic_matches(topic, t))) .map(|msg| { let ts = msg.stamp(); (msg.topic, Payload::from(msg.payload), ts) diff --git a/aimdb-websocket-connector/tests/e2e.rs b/aimdb-websocket-connector/tests/e2e.rs index 86dd75ad..c51d52b7 100644 --- a/aimdb-websocket-connector/tests/e2e.rs +++ b/aimdb-websocket-connector/tests/e2e.rs @@ -1026,6 +1026,54 @@ async fn topic_writer_snapshot_is_on_the_current_topic() { ); } +/// A subscribe serializes only the links whose topic matches: the topic is +/// resolved first, then the value is serialized for a match. +#[tokio::test] +async fn snapshots_serialize_only_matching_links() { + static SERIALIZED: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); + + let addr = free_addr(); + let mut sb = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(WebSocketConnector::new().bind(addr).path("/ws")); + sb.configure::("counted", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .with_remote_access() + .link_to("ws://_") + .with_topic_writer(128, InjectTopic) + .with_serializer(|_ctx, m: &Inject| { + SERIALIZED.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(serde_json::to_vec(&m.payload).unwrap()) + }) + .finish(); + }); + let (db, runner) = sb.build().await.expect("build db"); + tokio::spawn(runner.run()); + wait_for_listen(addr).await; + + let mut c = ws_connect(addr).await; + // Subscribe first, so the live send is counted before the late joiner. + ws_send(&mut c, json!({"t":"sub","id":1,"topic":"#"})).await; + ws_recv_tag(&mut c, "subscribed").await; + set(&db, "counted", json!({"topic": "a.x", "payload": 1})); + ws_recv_tag(&mut c, "event").await; + let live = SERIALIZED.load(std::sync::atomic::Ordering::SeqCst); + + // A subscribe whose pattern the current topic does not match. + let mut late = ws_connect(addr).await; + ws_send(&mut late, json!({"t":"sub","id":2,"topic":"b.#"})).await; + ws_recv_tag(&mut late, "subscribed").await; + assert_eq!(SERIALIZED.load(std::sync::atomic::Ordering::SeqCst), live); + + // A matching one serializes the value once more for its snapshot. + ws_send(&mut late, json!({"t":"sub","id":3,"topic":"a.#"})).await; + ws_recv_tag(&mut late, "snap").await; + assert_eq!( + SERIALIZED.load(std::sync::atomic::Ordering::SeqCst), + live + 1 + ); +} + // ── Golden wire frames (locks the exact on-the-wire JSON shape) ────── #[tokio::test]