From 8e8dcbf7f2ad4b00ed66b7961722994d9b1fab21 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sat, 7 Nov 2026 22:01:09 +0100 Subject: [PATCH] feat(aimx)!: exact-topic subscribes send a late-join snapshot (#438) An exact-topic AimX subscribe now sends the record's current value as a `snap` frame, as wildcard subscribes and WebSocket already do. A record with no current value (none written yet, or an `SpmcRing`) sends none. Each record's stream is opened first and its snapshot read right after; the stream then drops leading updates stamped at or before the snapshot. A fresh SingleLatest reader starts on the current value, so without this the value arrived twice: as the snapshot and again as the first event (already the case on wildcard subscribes). Co-Authored-By: Claude Opus 5.5 --- aimdb-client/tests/aimx_session.rs | 80 ++++++++++++++++++++++++- aimdb-core/CHANGELOG.md | 6 ++ aimdb-core/src/session/aimx/dispatch.rs | 55 +++++++++++------ tools/aimdb-cli/CHANGELOG.md | 7 +++ tools/aimdb-cli/README.md | 5 +- 5 files changed, 130 insertions(+), 23 deletions(-) diff --git a/aimdb-client/tests/aimx_session.rs b/aimdb-client/tests/aimx_session.rs index 7a69f395..0987b8a6 100644 --- a/aimdb-client/tests/aimx_session.rs +++ b/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 { + 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::(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] diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index a344f9a8..635c3753 100644 --- a/aimdb-core/CHANGELOG.md +++ b/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 diff --git a/aimdb-core/src/session/aimx/dispatch.rs b/aimdb-core/src/session/aimx/dispatch.rs index e055c3f2..9697a2a8 100644 --- a/aimdb-core/src/session/aimx/dispatch.rs +++ b/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, /// Per-record drain readers, created lazily on first `record.drain`. drain_readers: HashMap>, + /// 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> = 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 = 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 + 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, diff --git a/tools/aimdb-cli/CHANGELOG.md b/tools/aimdb-cli/CHANGELOG.md index 4ed9be0c..9861bb4e 100644 --- a/tools/aimdb-cli/CHANGELOG.md +++ b/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 ` 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 diff --git a/tools/aimdb-cli/README.md b/tools/aimdb-cli/README.md index 737f02f9..32616b07 100644 --- a/tools/aimdb-cli/README.md +++ b/tools/aimdb-cli/README.md @@ -202,8 +202,9 @@ Options: - `-f, --full`: Show full pretty-printed JSON for each event - `--format `: `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 `: Give up after SECS. Exits 4 if `--count` updates have not arrived by then; with `--count 0` it is a normal end.