feat(ws)!: late-join snapshots from the record's current value (#434)

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)
 

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

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

aimdb-core/src/outbound/latest.rs

@@ -0,0 +1,145 @@
+//! [`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<String>,
+    pub(crate) payload: Vec<u8>,
+    pub(crate) metadata: crate::WriteMetadata,
+}
+
+/// 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<Rendered>
+        + 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<u8>,
+    /// 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<str>,
+    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<T: Send + Sync>() {}
+    assert_send_sync::<OutboundLatest>();
+};
+
+impl OutboundLatest {
+    /// Every outbound link of `scheme`, in route order.
+    pub fn new(db: &AimDb, scheme: &str) -> Self {
+        let routes: Vec<LatestRoute> = 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<usize> {
+        self.routes.get(id).map(|r| r.record_index)
+    }
+
+    /// 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<LatestMessage> {
+        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<LatestMessage> {
+        let route = self.routes.get(id)?;
+        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
+                .topic
+                .unwrap_or_else(|| String::from(&*route.default_topic)),
+            payload: rendered.payload,
+            metadata: rendered.metadata,
+        })
+    }
+}

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,

aimdb-core/src/typed_api.rs

@@ -1589,9 +1589,70 @@ 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,
+                      accept: &dyn Fn(Option<&str>) -> bool| {
+                    let sample = db
+                        .inner()
+                        .get_typed_record_by_key::<T>(&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(),
+                    })
+                },
+            )
+        };
+
         // 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 +1673,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`.

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