feat(aimx)!: exact-topic subscribes send a late-join snapshot (#438)

aimdb-client/tests/aimx_session.rs

@@ -222,9 +222,8 @@ async fn wildcard_subscribe_fans_in_matching_records() {
             tokio::time::sleep(Duration::from_millis(10)).await;
         }
     });
-    // Drain events until the other matched record fires (the seeded record's
-    // current value may replay first — SingleLatest readers start warm). Every
-    // event must be topic-tagged and inside the pattern.
+    // Drain events until the other matched record fires. Every event must be
+    // topic-tagged and inside the pattern.
     let mut saw_berlin = false;
     for _ in 0..20 {
         match tokio::time::timeout(Duration::from_secs(2), stream.next()).await {
@@ -252,6 +251,81 @@ async fn wildcard_subscribe_fans_in_matching_records() {
     drop(conn);
 }
 
+/// The stream's next update, failing the test after two seconds.
+async fn next(
+    stream: &mut aimdb_core::session::BoxStream<
+        'static,
+        Result<aimdb_client::RecordUpdate, aimdb_client::ClientError>,
+    >,
+) -> aimdb_client::RecordUpdate {
+    tokio::time::timeout(Duration::from_secs(2), stream.next())
+        .await
+        .expect("an update within the timeout")
+        .expect("the stream is open")
+        .expect("not refused")
+}
+
+/// An exact-topic subscribe gets the record's current value as a snapshot,
+/// the same one a wildcard matching only that record gets, and then only
+/// later writes: the value the fresh reader starts on is not sent again.
+/// A ring, or a record with no value yet, gets no snapshot (#438).
+#[tokio::test]
+async fn exact_subscribe_sends_the_current_value_once() {
+    let dir = tempfile::tempdir().unwrap();
+    let sock = dir.path().join("aimdb.sock");
+
+    let config = AimxConfig::uds_default().socket_path(sock.to_str().unwrap());
+    let mut builder = AimDbBuilder::new()
+        .runtime(Arc::new(TokioAdapter))
+        .with_connector(UdsServer::from_config(config));
+    for (key, cfg) in [
+        ("temp.vienna", BufferCfg::SingleLatest),
+        ("empty", BufferCfg::SingleLatest),
+        ("ring", BufferCfg::SpmcRing { capacity: 16 }),
+    ] {
+        builder.configure::<Setting>(key, |reg| {
+            reg.buffer(cfg).with_remote_access();
+        });
+    }
+    let (db, runner) = builder.build().await.expect("build db");
+    tokio::spawn(runner.run());
+    for key in ["temp.vienna", "ring"] {
+        db.set_record_from_json(key, json!({ "level": 1 }))
+            .expect("seed");
+    }
+
+    let conn = AimxConnection::connect(sock.to_str().unwrap())
+        .await
+        .expect("connect");
+    let mut exact = conn.subscribe("temp.vienna").expect("subscribe");
+    let mut wildcard = conn.subscribe("temp.#").expect("subscribe");
+    let mut empty = conn.subscribe("empty").expect("subscribe");
+    let mut ring = conn.subscribe("ring").expect("subscribe");
+    // The server handles requests in order: once this replies, every
+    // subscription above is open and its snapshots are sent.
+    conn.get_record("temp.vienna").await.expect("get");
+
+    let snapshot = next(&mut exact).await;
+    assert_eq!(snapshot.value, Some(json!({ "level": 1 })));
+    assert!(snapshot.snapshot_end, "the only snapshot closes the burst");
+    let matched = next(&mut wildcard).await;
+    assert_eq!(
+        (&matched.topic, &matched.value, matched.ts),
+        (&snapshot.topic, &snapshot.value, snapshot.ts),
+        "exact and wildcard give the same snapshot"
+    );
+
+    for key in ["temp.vienna", "empty", "ring"] {
+        db.set_record_from_json(key, json!({ "level": 2 }))
+            .expect("write");
+    }
+    for stream in [&mut exact, &mut wildcard, &mut empty, &mut ring] {
+        let update = next(stream).await;
+        assert_eq!(update.value, Some(json!({ "level": 2 })), "{update:?}");
+        assert!(!update.snapshot_end, "a live event: {update:?}");
+    }
+}
+
 /// `record.get` on a ring (`SpmcRing`) record has no canonical latest, so it
 /// falls back to draining the connection's cursor for the most-recent value.
 #[tokio::test]

aimdb-core/CHANGELOG.md

@@ -23,6 +23,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   which applies the defaults: `name` `#` and `limit` `DEFAULT_QUERY_LIMIT`
   (1). `QueryHandlerParams::limit` is now `usize`; a custom `QueryHandlerFn`
   that relied on `None` now receives 1.
+- **Wire: an exact-topic AimX subscribe sends a late-join snapshot (#438),**
+  as wildcard subscribes and WebSocket already do: the record's current
+  value as a `snap` frame, then events from `seq` 2. A record with no
+  current value (none written yet, or an `SpmcRing`) sends none. The value a
+  fresh reader starts on is no longer also sent as the first event, on exact
+  and wildcard subscribes alike.
 
 ### Removed
 

aimdb-core/src/session/aimx/dispatch.rs

@@ -73,6 +73,7 @@ impl Dispatch for AimxDispatch {
             db: self.db.clone(),
             config: self.config.clone(),
             drain_readers: HashMap::new(),
+            pending_snapshots: Vec::new(),
         })
     }
 }
@@ -85,6 +86,9 @@ struct AimxSession {
     config: Arc<AimxConfig>,
     /// Per-record drain readers, created lazily on first `record.drain`.
     drain_readers: HashMap<String, Box<dyn JsonBufferReader + Send>>,
+    /// Late-join snapshots read by the last `subscribe`, handed to the engine
+    /// by `snapshots`, which it calls right after.
+    pending_snapshots: Vec<(String, Payload, crate::Stamp)>,
 }
 
 /// Internal dispatch result. Most methods still build a JSON tree; hot record
@@ -130,32 +134,23 @@ impl Session for AimxSession {
         // The engine owns the subscription lifecycle and the per-connection cap;
         // AimX has no async authorization, so this is a trivial wrapper.
         Box::pin(async move {
+            self.pending_snapshots.clear();
             if is_wildcard(topic) {
                 return Ok(self.subscribe_wildcard(topic));
             }
             // Exact-topic fast path: resolve one key, untagged updates.
-            let stream = crate::remote::stream::stream_record_updates(&self.db, topic)
-                .map_err(map_db_err)?;
+            let stream = self.record_stream(topic).map_err(map_db_err)?;
             Ok(Box::pin(
                 stream.map(|(data, ts, skipped)| SubUpdate::new(data, ts).with_skipped(skipped)),
             ) as BoxStream<'static, SubUpdate>)
         })
     }
 
-    fn snapshots(&mut self, topic: &str) -> Vec<(String, Payload, crate::Stamp)> {
-        // Late-join state for wildcard subscriptions: the current value of every
-        // matched record that has one. Exact-topic AimX subscribes stay
-        // event-only (unchanged wire for UDS/serial/TCP clients).
-        if !is_wildcard(topic) {
-            return Vec::new();
-        }
-        self.matched_keys(topic)
-            .into_iter()
-            .filter_map(|key| {
-                let sample = self.db.try_latest_as_json(&key)?;
-                Some((key, to_payload(&sample.value), sample.stamp()))
-            })
-            .collect()
+    /// The current value of every record the last `subscribe` covers that has
+    /// one, exact topic or wildcard alike. A record with none (nothing written
+    /// yet, or an `SpmcRing` buffer) sends no snapshot.
+    fn snapshots(&mut self, _topic: &str) -> Vec<(String, Payload, crate::Stamp)> {
+        core::mem::take(&mut self.pending_snapshots)
     }
 
     fn write<'a>(
@@ -194,10 +189,10 @@ impl AimxSession {
     /// One wildcard subscription fans in every matched record's update stream,
     /// each event tagged with the record that fired. Records without remote
     /// access are skipped — the subscription covers what can stream.
-    fn subscribe_wildcard(&self, pattern: &str) -> BoxStream<'static, SubUpdate> {
+    fn subscribe_wildcard(&mut self, pattern: &str) -> BoxStream<'static, SubUpdate> {
         let mut streams: Vec<BoxStream<'static, SubUpdate>> = Vec::new();
         for key in self.matched_keys(pattern) {
-            match crate::remote::stream::stream_record_updates(&self.db, &key) {
+            match self.record_stream(&key) {
                 Ok(stream) => {
                     let tag: Arc<str> = Arc::from(key.as_str());
                     streams.push(Box::pin(stream.map(move |(data, ts, skipped)| {
@@ -218,6 +213,30 @@ impl AimxSession {
         Box::pin(futures_util::stream::select_all(streams))
     }
 
+    /// Opens `key`'s update stream, then reads its current value into
+    /// `pending_snapshots`, in that order. A fresh reader may replay that same
+    /// value (a `SingleLatest` reader starts on it), so the stream drops
+    /// leading updates stamped at or before the snapshot: the client sees the
+    /// value once, as the snapshot. A write after the snapshot is stamped later
+    /// and streams as an event.
+    fn record_stream(
+        &mut self,
+        key: &str,
+    ) -> crate::DbResult<
+        impl futures_util::Stream<Item = (Payload, crate::Stamp, u64)> + Send + 'static,
+    > {
+        let stream = crate::remote::stream::stream_record_updates(&self.db, key)?;
+        let snapshot = self.db.try_latest_as_json_bytes(key).map(|sample| {
+            let ts = sample.stamp();
+            self.pending_snapshots
+                .push((key.to_string(), Payload::from(sample.value), ts));
+            ts
+        });
+        Ok(stream.skip_while(move |(_, ts, _)| {
+            core::future::ready(snapshot.is_some_and(|seen| *ts <= seen))
+        }))
+    }
+
     /// Match the method and produce a tree or already-serialized JSON reply.
     async fn dispatch_call(
         &mut self,

tools/aimdb-cli/CHANGELOG.md

@@ -14,6 +14,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   JSON output is unchanged: it already carries `schema_type` /
   `schema_version`.
 
+### Changed
+
+- **`aimdb watch <record>` starts with the record's current value as a
+  snapshot (#438),** followed by the `📸 snapshot complete` line. The value
+  arrived before too, as a plain first event; it still counts toward
+  `--count`, and `--new` still skips it.
+
 ### Removed (breaking)
 
 - **`aimdb generate` (#411)** and the `aimdb-codegen` dependency. Write the

tools/aimdb-cli/README.md

@@ -202,8 +202,9 @@ Options:
 - `-f, --full`: Show full pretty-printed JSON for each event
 - `--format <table|json>`: `json` writes one NDJSON object per update on
   stdout and every status line on stderr
-- `--new`: Count only updates written after the watch is armed. Without it, a
-  `SingleLatest` record's current value can arrive first and pass as a new one.
+- `--new`: Count only updates written after the watch is armed. Without it, the
+  record's current value arrives first, as a snapshot followed by a
+  `📸 snapshot complete` line, and counts toward `--count`.
 - `--timeout <SECS>`: Give up after SECS. Exits 4 if `--count` updates have not
   arrived by then; with `--count 0` it is a normal end.