diff --git a/CHANGELOG.md b/CHANGELOG.md index b5d3e28c..231d2921 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — `record.get` answers at once and no longer drains (#444) + +- **`record.get` is a single, non-destructive read.** `aimdb-mcp`'s + `get_record` drops its 3 s retry, the server drops the get-to-drain fallback, + and a ring answers `unsupported` instead of consuming events. See + [aimdb-core](aimdb-core/CHANGELOG.md), + [aimdb-client](aimdb-client/CHANGELOG.md) and + [aimdb-mcp](tools/aimdb-mcp/CHANGELOG.md). + ### Removed — `examples/weather-mesh-demo` - **The in-repo weather mesh demo is gone.** The fleet behind aimdb.dev lives in diff --git a/Cargo.lock b/Cargo.lock index ab1216c4..0ca5788a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -248,12 +248,17 @@ version = "0.10.0" dependencies = [ "aimdb-client", "aimdb-core", + "aimdb-serial-connector", + "aimdb-tokio-adapter", + "aimdb-uds-connector", "anyhow", "chrono", "clap", + "libc", "once_cell", "serde", "serde_json", + "tempfile", "thiserror 1.0.69", "tokio", "tokio-test", diff --git a/Makefile b/Makefile index 5de300a7..7f42589c 100644 --- a/Makefile +++ b/Makefile @@ -243,6 +243,8 @@ test: test-operational-events test-build-info-link cargo test --package aimdb-cli --features yaml @printf "$(YELLOW) → Testing MCP server$(NC)\n" cargo test --package aimdb-mcp + @printf "$(YELLOW) → Testing MCP server (serial: one session for get and drain)$(NC)\n" + cargo test --package aimdb-mcp --features "transport-serial" --test serial_session @printf "$(YELLOW) → Testing persistence backend$(NC)\n" cargo test --package aimdb-persistence @printf "$(YELLOW) → Testing persistence SQLite backend$(NC)\n" diff --git a/aimdb-client/CHANGELOG.md b/aimdb-client/CHANGELOG.md index 8fcb193c..d9725321 100644 --- a/aimdb-client/CHANGELOG.md +++ b/aimdb-client/CHANGELOG.md @@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- **`get_record` on a ring returns `unsupported` (#444).** A ring has no + current value; use `drain_record` or `subscribe`. `get_record` no longer moves + the connection's drain cursor, so `drain_record` returns every unread event. + `watch::arm` treats `unsupported` like `not_found`: no baseline. + +### Added + +- **`AimxConnection::is_closed()` (#450):** whether the connection's engine has + stopped. A caller keeping a connection open uses it to decide when to + reconnect; an `internal` error reply does not mean the connection is gone. + ### Removed (breaking) - **Obsolete AimX message types (#419).** `protocol::{Request, Response, diff --git a/aimdb-client/src/engine.rs b/aimdb-client/src/engine.rs index 92df4643..54b18389 100644 --- a/aimdb-client/src/engine.rs +++ b/aimdb-client/src/engine.rs @@ -231,6 +231,16 @@ impl AimxConnection { &self.server_info } + /// Whether this connection's engine has stopped: the transport closed or + /// failed. The engine does not reconnect, so every later call fails; a + /// caller that keeps a connection open replaces it. + /// + /// This is the reliable signal: an `internal` error can also be an + /// ordinary server reply on a healthy connection. + pub fn is_closed(&self) -> bool { + self.engine.is_finished() + } + /// List all registered records. pub async fn list_records(&self) -> ClientResult> { let reply = self.call("record.list", null_payload()).await?; @@ -239,12 +249,14 @@ impl AimxConnection { /// Get a record's current value. /// - /// For a `SingleLatest`/state record this is a non-destructive read. A ring - /// (`SpmcRing`) has no canonical latest, so the server returns the most recent - /// value from this connection's drain cursor — which **advances that cursor** - /// (interleaving with [`drain_record`](Self::drain_record)) and is empty until - /// the ring produces a value after the connection first reads it. Prefer - /// [`drain_record`](Self::drain_record) for ring/history records. + /// A single, non-destructive read of the record's snapshot (`SingleLatest`, + /// `Mailbox`). It returns at once and never moves this connection's drain + /// cursor. Server errors: + /// + /// - `not_found`: no such record, or no value yet. + /// - `unsupported`: the record is a ring (`SpmcRing`), which has no current + /// value. Use [`drain_record`](Self::drain_record) for unread events or + /// [`subscribe`](Self::subscribe) for new ones. /// /// The value comes with the stamp of its write; a reply without `ts` is a /// [`ClientError::Protocol`]. @@ -571,6 +583,12 @@ fn rpc_err(e: RpcError) -> ClientError { ClientError::server_error("not_found", "method or record not found", None) } RpcError::Denied => ClientError::server_error("denied", "permission denied", None), + RpcError::Unsupported => ClientError::server_error( + "unsupported", + "not supported for this record (a ring has no current value: \ + use drain_record for unread events or subscribe for new ones)", + None, + ), // `Internal` today, plus any future non-exhaustive variant. _ => ClientError::server_error("internal", "engine/transport failure", None), } diff --git a/aimdb-client/src/watch.rs b/aimdb-client/src/watch.rs index 5ffbd5f5..c577aa6d 100644 --- a/aimdb-client/src/watch.rs +++ b/aimdb-client/src/watch.rs @@ -84,8 +84,13 @@ pub async fn arm(conn: &AimxConnection, record: &str) -> Result Some(current.stamp()), // No value yet (or an unknown record, whose subscribe refusal is - // picked up below). - Err(ClientError::ServerError { code, .. }) if code == "not_found" => None, + // picked up below), or a ring: it has no current value, and its + // subscription starts after the last write, so nothing old can count. + Err(ClientError::ServerError { code, .. }) + if code == "not_found" || code == "unsupported" => + { + None + } Err(e) => return Err(e), }; let stream = match stream.next().now_or_never() { diff --git a/aimdb-client/tests/aimx_session.rs b/aimdb-client/tests/aimx_session.rs index 0987b8a6..6b3ec050 100644 --- a/aimdb-client/tests/aimx_session.rs +++ b/aimdb-client/tests/aimx_session.rs @@ -326,10 +326,11 @@ async fn exact_subscribe_sends_the_current_value_once() { } } -/// `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. +/// `record.get` is one non-destructive read (#444): it answers at once for an +/// empty record, an unknown name and a ring, and never moves the connection's +/// drain cursor, so `record.drain` still returns every unread event. #[tokio::test] -async fn record_get_on_ring_falls_back_to_drain() { +async fn record_get_answers_at_once_and_leaves_drain_intact() { let dir = tempfile::tempdir().unwrap(); let sock = dir.path().join("aimdb.sock"); @@ -341,6 +342,9 @@ async fn record_get_on_ring_falls_back_to_drain() { reg.buffer(BufferCfg::SpmcRing { capacity: 16 }) .with_remote_access(); }); + builder.configure::("state", |reg| { + reg.buffer(BufferCfg::SingleLatest).with_remote_access(); + }); let (db, runner) = builder.build().await.expect("build db"); let db = Arc::new(db); tokio::spawn(runner.run()); @@ -348,28 +352,61 @@ async fn record_get_on_ring_falls_back_to_drain() { let conn = AimxConnection::connect(sock.to_str().unwrap()) .await .expect("connect"); - let producer = db.producer::("stream").expect("producer"); - // First get opens the cursor; a fresh broadcast reader starts at the tail, so - // it sees nothing until a value is produced afterwards. - assert!(conn.get_record("stream").await.is_err()); + // Each miss answers at once (the old MCP retry took 3 s per read). + let code = |r: Result<_, aimdb_client::ClientError>| match r { + Err(aimdb_client::ClientError::ServerError { code, .. }) => code, + other => panic!("expected a server error, got {other:?}"), + }; + let started = std::time::Instant::now(); + assert_eq!(code(conn.get_record("state").await), "not_found", "empty"); + assert_eq!( + code(conn.get_record("missing").await), + "not_found", + "unknown" + ); + assert_eq!(code(conn.get_record("stream").await), "unsupported", "ring"); + assert!( + started.elapsed() < Duration::from_secs(1), + "three misses took {:?}", + started.elapsed() + ); - // Produce after the cursor is open; get now returns the most-recent value. - let mut got = None; - for n in 1..=50u64 { - producer.produce(Reading { n }); - tokio::time::sleep(Duration::from_millis(10)).await; - if let Ok(v) = conn.get_record("stream").await { - got = Some(v); + // Open the drain cursors, then write. + assert_eq!(conn.drain_record("stream").await.expect("drain").count, 0); + assert_eq!(conn.drain_record("state").await.expect("drain").count, 0); + let ring = db.producer::("stream").expect("producer"); + let state = db.producer::("state").expect("producer"); + for n in 1..=3 { + ring.produce(Reading { n }); + } + state.produce(Reading { n: 7 }); + + // Reads in between: the ring still refuses, the state answers, twice. + assert_eq!(code(conn.get_record("stream").await), "unsupported"); + for _ in 0..2 { + let got = conn.get_record("state").await.expect("state get"); + assert_eq!(got.value, serde_json::json!({ "n": 7 })); + } + + // Drain still sees every event the gets did not consume. + let mut drained = Vec::new(); + for _ in 0..50 { + drained.extend(conn.drain_record("stream").await.expect("drain").values); + if drained.len() >= 3 { break; } + tokio::time::sleep(Duration::from_millis(10)).await; } - let got = got.expect("ring get returns a value after producing"); - assert!( - got.value.get("n").is_some(), - "ring get yields a Reading: {}", - got.value - ); + let ns: Vec<_> = drained.iter().map(|v| v.value["n"].clone()).collect(); + assert_eq!(ns, [1, 2, 3]); + let state_drain = conn.drain_record("state").await.expect("drain"); + let ns: Vec<_> = state_drain + .values + .iter() + .map(|v| v.value["n"].clone()) + .collect(); + assert_eq!(ns, [7]); drop(conn); } diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index c6f1c808..b1504856 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -7,6 +7,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed (wire) + +- **`record.get` is a single, non-destructive read (#444).** It returns the + buffer's own snapshot (`SingleLatest`, `Mailbox`) and never touches the + connection's drain cursor, so only `record.drain` advances it. The fallback + that drained a ring to answer `record.get` is gone: it made a point-in-time + read wait for a future write and consumed events a later `record.drain` + should have returned. A ring (`SpmcRing`) now answers with the new error + code `unsupported`; read it with `record.drain` (unread events) or + `record.subscribe` (new ones). An empty `SingleLatest`/`Mailbox` record and + an unknown name answer `not_found` at once. A ring registered with + `buffer_raw()` and no `buffer_cfg()` answers `not_found`, since core cannot + tell it is a ring. + +### Added + +- **`RpcError::Unsupported` (#444),** wire code `"unsupported"`: the method + exists but the target cannot serve it. Older clients decode an unknown code + as `internal`. + ### Removed (breaking) - **Obsolete AimX message types (#419).** `remote::{Request, Response, diff --git a/aimdb-core/src/session/aimx/codec.rs b/aimdb-core/src/session/aimx/codec.rs index 5bcd3d47..52496400 100644 --- a/aimdb-core/src/session/aimx/codec.rs +++ b/aimdb-core/src/session/aimx/codec.rs @@ -147,6 +147,7 @@ fn payload_of(raw: Option<&RawValue>) -> Payload { fn err_code(e: &RpcError) -> &'static str { match e { RpcError::NotFound => "not_found", + RpcError::Unsupported => "unsupported", RpcError::Denied => "denied", RpcError::Internal => "internal", RpcError::VersionMismatch => "version_mismatch", @@ -158,6 +159,7 @@ fn err_code(e: &RpcError) -> &'static str { fn code_err(s: &str) -> RpcError { match s { "not_found" => RpcError::NotFound, + "unsupported" => RpcError::Unsupported, "denied" => RpcError::Denied, "version_mismatch" => RpcError::VersionMismatch, _ => RpcError::Internal, @@ -509,6 +511,7 @@ mod tests { fn reply_err_codes_roundtrip() { for err in [ RpcError::NotFound, + RpcError::Unsupported, RpcError::Denied, RpcError::Internal, RpcError::VersionMismatch, diff --git a/aimdb-core/src/session/aimx/dispatch.rs b/aimdb-core/src/session/aimx/dispatch.rs index 9697a2a8..414a0e9c 100644 --- a/aimdb-core/src/session/aimx/dispatch.rs +++ b/aimdb-core/src/session/aimx/dispatch.rs @@ -319,25 +319,37 @@ impl AimxSession { /// `record.get`: the record's current value and its write stamp, /// `{"value": , "ts": }`. /// - /// A `SingleLatest`/state record exposes a non-destructive canonical latest - /// ([`try_latest_as_json_bytes`](crate::AimDb::try_latest_as_json_bytes)). - /// A ring - /// ([`SpmcRing`](crate::buffer::BufferCfg::SpmcRing)) has none, so we fall back - /// to the connection's drain cursor and return the **most recent** available - /// value. Two consequences of that fallback: it *advances the shared drain - /// cursor* (so `record.get` and `record.drain` interleave on one connection), - /// and it yields `NotFound` until the ring produces a value *after* the cursor - /// is first opened (a fresh broadcast reader starts at the tail). Use - /// `record.drain` for a ring's full backlog. - fn record_get(&mut self, name: &str) -> Result { + /// A single, non-destructive read of the buffer's own snapshot + /// ([`try_latest_as_json_bytes`](crate::AimDb::try_latest_as_json_bytes)): + /// it returns at once and never touches this connection's drain cursor. + /// + /// - `NotFound`: no such record, or a record with a snapshot (`SingleLatest`, + /// `Mailbox`) that holds no value yet. + /// - `Unsupported`: an [`SpmcRing`](crate::buffer::BufferCfg::SpmcRing). A + /// ring is a stream with no snapshot; read it with `record.drain` + /// (unread events) or `record.subscribe` (future events). An application + /// that needs an inspectable current state exposes a `SingleLatest` + /// record next to the stream. + /// + /// The ring check reads the declared [`BufferCfg`](crate::buffer::BufferCfg): + /// a ring registered with `buffer_raw()` and no `buffer_cfg()` reports + /// `NotFound`, because core cannot tell it is a ring. + fn record_get(&self, name: &str) -> Result { if let Some(sample) = self.db.try_latest_as_json_bytes(name) { return Ok(super::value_reply(&sample.value, sample.stamp())); } - // Ring fallback: drain to the newest currently-available value (or NotFound). - self.drain_values(name, usize::MAX)? - .pop() - .map(|sample| to_payload(&stamped_value_json(sample))) - .ok_or(RpcError::NotFound) + let inner = self.db.inner(); + let record = inner + .resolve_str(name) + .and_then(|id| inner.storage(id)) + .ok_or(RpcError::NotFound)?; + // Only the miss path pays for this lookup. + let ring = crate::buffer::BufferCfg::SpmcRing { capacity: 0 }.name(); + if record.buffer_info().0 == ring { + Err(RpcError::Unsupported) + } else { + Err(RpcError::NotFound) + } } /// `record.drain`: return everything accumulated since the previous drain @@ -359,9 +371,7 @@ impl AimxSession { } /// Lazily open (on first call) the per-record drain cursor and read up to - /// `limit` values accumulated since the previous read (oldest-first). Shared - /// by [`record.drain`](Self::record_drain) and [`record.get`](Self::record_get)'s - /// ring fallback, so both read from the same per-connection cursor. + /// `limit` values accumulated since the previous read (oldest-first). fn drain_values(&mut self, name: &str, limit: usize) -> Result>, RpcError> { if !self.drain_readers.contains_key(name) { let id = self diff --git a/aimdb-core/src/session/mod.rs b/aimdb-core/src/session/mod.rs index e713bf38..3f6b0695 100644 --- a/aimdb-core/src/session/mod.rs +++ b/aimdb-core/src/session/mod.rs @@ -270,6 +270,9 @@ pub enum CodecError { pub enum RpcError { /// No such method or topic. NotFound, + /// The method exists but the target cannot serve it, e.g. `record.get` on + /// a ring buffer, which has no current value to read. + Unsupported, /// The caller lacks permission for this operation. Denied, /// The handler failed. diff --git a/aimdb-wasm-adapter/src/ws_bridge.rs b/aimdb-wasm-adapter/src/ws_bridge.rs index b39b4cbd..2cfb7120 100644 --- a/aimdb-wasm-adapter/src/ws_bridge.rs +++ b/aimdb-wasm-adapter/src/ws_bridge.rs @@ -840,6 +840,7 @@ impl WsBridge { fn rpc_err_str(e: &RpcError) -> &'static str { match e { RpcError::NotFound => "not_found", + RpcError::Unsupported => "unsupported (a ring has no current value)", RpcError::Denied => "denied", RpcError::VersionMismatch => { "version_mismatch (server rejected the client protocol version)" diff --git a/examples/remote-access-demo/src/client.rs b/examples/remote-access-demo/src/client.rs index f43d48f1..b91e70c8 100644 --- a/examples/remote-access-demo/src/client.rs +++ b/examples/remote-access-demo/src/client.rs @@ -55,27 +55,13 @@ async fn main() -> Result<(), Box> { } println!(); - // A ring (SpmcRing) has no canonical latest, so the server falls back to - // *draining* this connection's cursor and returning the most recent value. A - // fresh cursor starts at the ring tail, so the first read is empty until a new - // value arrives — we retry over a couple of ticks. (This opens the *same* - // cursor `record.drain` uses below.) - println!("📤 record.get on Temperature (SpmcRing — drains the cursor for the latest)..."); - let mut latest = None; - for _ in 0..10 { - if let Ok(v) = conn.get_record("server::Temperature").await { - latest = Some(v); - break; - } - tokio::time::sleep(Duration::from_millis(500)).await; - } - match latest { - Some(v) => println!( - "🌡️ Most recent Temperature (via drain fallback, written {}):\n{}", - since_start(v.stamp()), - serde_json::to_string_pretty(&v.value)? - ), - None => println!("ℹ️ No reading yet — the ring cursor was still empty."), + // A ring (SpmcRing) is a stream with no current value, so record.get + // reports `unsupported`. Its events are read with record.drain (below) or + // record.subscribe. + println!("📤 record.get on Temperature (SpmcRing — no current value)..."); + match conn.get_record("server::Temperature").await { + Ok(v) => println!("❌ UNEXPECTED value: {}", v.value), + Err(e) => println!("✅ Expected: {e}"), } println!(); @@ -138,14 +124,11 @@ async fn main() -> Result<(), Box> { println!("🧪 Record History (record.drain)"); println!("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n"); - // `record.get` above already opened this connection's Temperature cursor, so - // this drain returns only what's accrued since (they share one cursor). - println!("📤 Drain #1: history since the (already-open) cursor..."); + // The first drain opens this connection's Temperature cursor at the ring + // tail, so it returns nothing yet. + println!("📤 Drain #1: opens the cursor..."); let d1 = conn.drain_record("server::Temperature").await?; - println!( - " Values: {} (cursor shared with the record.get above)\n", - d1.count - ); + println!(" Values: {} (cursor just opened)\n", d1.count); println!("⏳ Waiting 7s for temperature readings to accumulate..."); tokio::time::sleep(Duration::from_secs(7)).await; diff --git a/tools/aimdb-cli/README.md b/tools/aimdb-cli/README.md index 32616b07..f237e73a 100644 --- a/tools/aimdb-cli/README.md +++ b/tools/aimdb-cli/README.md @@ -156,7 +156,9 @@ Options: - `-w, --writable`: Show only writable records #### `record get` -Get the current value of a specific record. +Get the current value of a specific record. A `spmc_ring` record is a stream +with no current value, so `record get` reports `unsupported` for it; use +`watch` to follow its events. ```bash aimdb record get [OPTIONS] diff --git a/tools/aimdb-mcp/CHANGELOG.md b/tools/aimdb-mcp/CHANGELOG.md index a882d070..7cb2e3a2 100644 --- a/tools/aimdb-mcp/CHANGELOG.md +++ b/tools/aimdb-mcp/CHANGELOG.md @@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **`get_record` answers at once (#444).** It no longer retries `not_found` for + up to 3 s, so an empty record or a misspelled name returns immediately + instead of after 3 s. A ring returns `unsupported` (use `drain_record`), and + `get_record` no longer shares `drain_record`'s cursor. +- **`get_record` and `drain_record` share one session per endpoint (#450).** + `get_record` opened a second connection, which over serial failed (the tty + was already open) or sent a new `hello` that reset the board's session and + discarded its drain cursor. Both now use the persistent client, one call at a + time. An error reply (`not_found`, `unsupported`, even `internal`) keeps the + session; the client is replaced only once its engine has stopped. + ### Removed (breaking) - **The architecture agent (#411).** The 16 `state.toml` tools diff --git a/tools/aimdb-mcp/Cargo.toml b/tools/aimdb-mcp/Cargo.toml index 3c93fd85..821b532f 100644 --- a/tools/aimdb-mcp/Cargo.toml +++ b/tools/aimdb-mcp/Cargo.toml @@ -69,3 +69,12 @@ chrono = "0.4" # `__event_capture!` for the test-only `events::capture` module. aimdb-core = { version = "3.0.0", path = "../../aimdb-core", default-features = false } tokio-test = "0.4" +# tests/get_record.rs: a real AimX server over a Unix socket. +aimdb-tokio-adapter = { version = "0.8.0", path = "../../aimdb-tokio-adapter" } +aimdb-uds-connector = { version = "0.2.0", path = "../../aimdb-uds-connector" } +tempfile = "3" +# tests/serial_session.rs: a board (`SerialServer`) behind a pty. +aimdb-serial-connector = { version = "0.2.0", path = "../../aimdb-serial-connector", features = [ + "std", +] } +libc = "0.2" diff --git a/tools/aimdb-mcp/src/connection.rs b/tools/aimdb-mcp/src/connection.rs index 06bb6ca7..fe69a065 100644 --- a/tools/aimdb-mcp/src/connection.rs +++ b/tools/aimdb-mcp/src/connection.rs @@ -81,21 +81,37 @@ impl ConnectionPool { } } - /// Get or create a persistent drain client for a socket path. + /// Get or create the persistent client for an endpoint. /// - /// Drain clients are kept alive across calls so the server-side drain - /// reader accumulates values between invocations. The first drain call - /// on a new connection is a cold start (returns empty); subsequent calls - /// return all values accumulated since the previous drain. + /// One connection per endpoint, kept across calls and shared by + /// `get_record` and `drain_record`; callers take its lock for each call. + /// Keeping it open matters twice: the server's drain cursor lives in the + /// session, so values accumulate between drains, and on serial a second + /// connection either fails (the tty is open) or sends a new `hello` that + /// resets the board's session and discards that cursor. + /// + /// The client is replaced only once its engine has stopped + /// ([`AimxConnection::is_closed`]), never because a call returned an error: + /// `not_found`, `unsupported` and even `internal` are replies on a healthy + /// session. A client busy with a call is returned as is; if its engine has + /// stopped, the next call replaces it. pub async fn get_drain_client( &self, endpoint: &str, ) -> Result>, ClientError> { - let drain_map = self.drain_clients.lock().await; + let mut drain_map = self.drain_clients.lock().await; if let Some(client) = drain_map.get(endpoint) { - debug!("♻️ Reusing persistent drain client for {}", endpoint); - return Ok(Arc::clone(client)); + let closed = client.try_lock().is_ok_and(|c| c.is_closed()); + if !closed { + debug!("♻️ Reusing persistent client for {}", endpoint); + return Ok(Arc::clone(client)); + } + debug!( + "🔁 Persistent client for {} has stopped, reconnecting", + endpoint + ); + drain_map.remove(endpoint); } debug!("🔌 Creating persistent drain client for {}", endpoint); @@ -109,7 +125,9 @@ impl ConnectionPool { let mut drain_map = self.drain_clients.lock().await; // Double-check: another task may have inserted while we were connecting if let Some(existing) = drain_map.get(endpoint) { - return Ok(Arc::clone(existing)); + if !existing.try_lock().is_ok_and(|c| c.is_closed()) { + return Ok(Arc::clone(existing)); + } } drain_map.insert(endpoint.to_string(), Arc::clone(&shared)); Ok(shared) diff --git a/tools/aimdb-mcp/src/server.rs b/tools/aimdb-mcp/src/server.rs index 937d57bb..d552bad4 100644 --- a/tools/aimdb-mcp/src/server.rs +++ b/tools/aimdb-mcp/src/server.rs @@ -211,7 +211,9 @@ impl McpServer { description: "Get the current value of a specific record from an AimDB instance. \ Returns {\"value\": , \"ts\": }, where ts is \ nanoseconds since the AimDB instance started (not wall-clock time; only comparable \ - within one instance lifetime).".to_string(), + within one instance lifetime). Answers at once: not_found when the record does \ + not exist or holds no value yet, unsupported for a ring buffer (spmc_ring), which \ + has no current value; use drain_record for its unread events.".to_string(), input_schema: json!({ "type": "object", "properties": { diff --git a/tools/aimdb-mcp/src/tools/record.rs b/tools/aimdb-mcp/src/tools/record.rs index 740ce3a9..1cb64cd0 100644 --- a/tools/aimdb-mcp/src/tools/record.rs +++ b/tools/aimdb-mcp/src/tools/record.rs @@ -1,10 +1,9 @@ //! Record-related tools (list_records, get_record, set_record) use crate::error::{McpError, McpResult}; -use aimdb_client::{AimxConnection, ClientError}; +use aimdb_client::AimxConnection; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; -use std::time::{Duration, Instant}; use tracing::debug; /// Parameters for list_records tool @@ -162,59 +161,27 @@ pub async fn get_record(args: Option) -> McpResult { endpoint, params.record_name ); - // Reuse the *persistent* connection (the same pool `drain_record` uses) rather - // than a throwaway one. For ring buffers (`SpmcRing`, which has no canonical - // latest), the server's `record.get` falls back to *this connection's* drain - // cursor (see `aimdb_core::session::aimx::dispatch`'s `record_get`). A fresh - // connection per call opens a new cursor at the ring tail every time and always - // reads empty → `not_found`; the persistent connection lets that cursor - // accumulate. The get→drain fallback already lives server-side, so nothing is - // duplicated here — we just stop discarding the connection. + // The persistent client `drain_record` uses: on serial a second connection + // fails (the tty is open) or says `hello` again, which resets the board's + // session and its drain cursors. Sharing is safe because `record.get` never + // touches the drain cursor (the server keeps them apart). let pool = super::connection_pool() .ok_or_else(|| McpError::Internal("Connection pool not initialized".to_string()))?; - let client_arc = pool .get_drain_client(&endpoint) .await .map_err(McpError::Client)?; - let client = client_arc.lock().await; - // A ring's drain cursor opens at the tail, so the first read is empty until a - // value is produced after it opened. Briefly retry on `not_found` so a one-shot - // `get_record` on a ring returns the latest value instead of failing. This adds - // no latency for `single_latest` records (they return immediately via the - // server's canonical-latest path) nor for an already-warm ring cursor. Note: a - // record that simply doesn't exist also reports `not_found`, so a bad name - // costs the full window before erroring. - let deadline = Instant::now() + Duration::from_secs(3); - loop { - let err = match client.get_record(¶ms.record_name).await { - Ok(sample) => { - debug!("✅ Retrieved record '{}'", params.record_name); - return Ok(stamped_json(&sample)); - } - Err(e) => e, - }; - - let cursor_warming = - matches!(&err, ClientError::ServerError { code, .. } if code == "not_found"); - - if cursor_warming && Instant::now() < deadline { - tokio::time::sleep(Duration::from_millis(250)).await; - continue; - } - - // A genuine connection/protocol error (not a warming ring cursor) means the - // persistent client is unhealthy — drop it so the next call reconnects. - if !cursor_warming { - let socket = endpoint.clone(); - let pool = pool.clone(); - tokio::spawn(async move { pool.invalidate_drain_client(&socket).await }); - } - - return Err(McpError::Client(err)); - } + // One `record.get`, no retry: `not_found` (no such record, or no value yet) + // and `unsupported` (a ring) come back at once and leave the session as it + // is. A stopped engine is replaced by `get_drain_client` on the next call. + let sample = client + .get_record(¶ms.record_name) + .await + .map_err(McpError::Client)?; + debug!("✅ Retrieved record '{}'", params.record_name); + Ok(stamped_json(&sample)) } /// Set the value of a writable record @@ -290,8 +257,8 @@ pub async fn drain_record(args: Option) -> McpResult { endpoint, params.record_name ); - // Use persistent drain client so the server-side drain reader persists - // across calls, allowing values to accumulate between drains. + // The persistent client, shared with `get_record`: the server-side drain + // cursor lives in its session, so values accumulate between calls. let pool = super::connection_pool() .ok_or_else(|| McpError::Internal("Connection pool not initialized".to_string()))?; @@ -302,27 +269,17 @@ pub async fn drain_record(args: Option) -> McpResult { let client = client_arc.lock().await; - // Drain record values + // An error reply keeps the session (and its cursors); a stopped engine is + // replaced by `get_drain_client` on the next call. let response = match params.limit { Some(limit) => client .drain_record_with_limit(¶ms.record_name, limit) .await - .map_err(|e| { - // On connection error, invalidate so next call reconnects - let socket = endpoint.clone(); - let pool = pool.clone(); - tokio::spawn(async move { pool.invalidate_drain_client(&socket).await }); - McpError::Client(e) - })?, + .map_err(McpError::Client)?, None => client .drain_record(¶ms.record_name) .await - .map_err(|e| { - let socket = endpoint.clone(); - let pool = pool.clone(); - tokio::spawn(async move { pool.invalidate_drain_client(&socket).await }); - McpError::Client(e) - })?, + .map_err(McpError::Client)?, }; debug!( diff --git a/tools/aimdb-mcp/tests/get_record.rs b/tools/aimdb-mcp/tests/get_record.rs new file mode 100644 index 00000000..35252b1f --- /dev/null +++ b/tools/aimdb-mcp/tests/get_record.rs @@ -0,0 +1,106 @@ +//! `get_record` answers at once and leaves `drain_record` intact (#444). +//! +//! It used to retry `not_found` for up to 3 s, so an empty record or a bad +//! name cost the whole window, and a ring was read through the drain cursor. + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use aimdb_core::buffer::BufferCfg; +use aimdb_core::remote::AimxConfig; +use aimdb_core::AimDbBuilder; +use aimdb_mcp::tools; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +use aimdb_uds_connector::UdsServer; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +struct Reading { + n: u64, +} + +fn args(endpoint: &str, record: &str) -> Option { + Some(json!({ "endpoint": endpoint, "record_name": record })) +} + +#[tokio::test] +async fn get_record_answers_at_once_and_leaves_drain_intact() { + let dir = tempfile::tempdir().unwrap(); + let sock = dir.path().join("aimdb.sock"); + let endpoint = sock.to_str().unwrap().to_string(); + + let config = AimxConfig::uds_default().socket_path(&endpoint); + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(UdsServer::from_config(config)); + builder.configure::("stream", |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 16 }) + .with_remote_access(); + }); + builder.configure::("state", |reg| { + reg.buffer(BufferCfg::SingleLatest).with_remote_access(); + }); + let (db, runner) = builder.build().await.expect("build db"); + tokio::spawn(runner.run()); + tools::init_connection_pool(aimdb_mcp::connection::ConnectionPool::new()); + + // Each miss used to take ~3 s. Allow 1 s for all three together. + let started = Instant::now(); + for (record, code) in [ + ("state", "not_found"), // single_latest, no value yet (the issue) + ("missing", "not_found"), // unknown name + ("stream", "unsupported"), // a ring has no current value + ] { + let err = tools::get_record(args(&endpoint, record)) + .await + .expect_err(record) + .to_string(); + assert!(err.contains(code), "{record}: {err}"); + } + assert!( + started.elapsed() < Duration::from_secs(1), + "three misses took {:?}", + started.elapsed() + ); + + // Open the ring's drain cursor, write, and read the ring with get in + // between: drain still returns every event. + let drain = || { + tools::drain_record(Some( + json!({ "endpoint": endpoint, "record_name": "stream" }), + )) + }; + assert_eq!(drain().await.expect("drain")["count"], 0); + let producer = db.producer::("stream").expect("producer"); + for n in 1..=3 { + producer.produce(Reading { n }); + } + assert!(tools::get_record(args(&endpoint, "stream")).await.is_err()); + + let mut ns = Vec::new(); + for _ in 0..50 { + let page = drain().await.expect("drain"); + ns.extend( + page["values"] + .as_array() + .unwrap() + .iter() + .map(|v| v["value"]["n"].clone()), + ); + if ns.len() >= 3 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(ns, [1, 2, 3]); + + // A state record with a value answers with it, at once. + db.producer::("state") + .unwrap() + .produce(Reading { n: 7 }); + let got = tools::get_record(args(&endpoint, "state")) + .await + .expect("get"); + assert_eq!(got["value"], json!({ "n": 7 })); +} diff --git a/tools/aimdb-mcp/tests/persistent_client.rs b/tools/aimdb-mcp/tests/persistent_client.rs new file mode 100644 index 00000000..79ad6ff3 --- /dev/null +++ b/tools/aimdb-mcp/tests/persistent_client.rs @@ -0,0 +1,110 @@ +//! The MCP keeps one AimX client per endpoint and replaces it only when its +//! engine has stopped, never because a call returned an error. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use aimdb_core::buffer::BufferCfg; +use aimdb_core::remote::AimxConfig; +use aimdb_core::AimDbBuilder; +use aimdb_mcp::connection::ConnectionPool; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +use aimdb_uds_connector::UdsServer; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct State { + n: u64, +} + +/// An AimX server in its own runtime on its own thread. Dropping it shuts that +/// runtime down, which drops every session, as a crashed or rebooted instance +/// would. +struct Server { + stop: Option>, + thread: Option>, +} + +impl Server { + fn start(sock: &Path) -> Self { + let sock: PathBuf = sock.to_path_buf(); + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let (stop, stop_rx) = tokio::sync::oneshot::channel::<()>(); + let thread = std::thread::spawn(move || { + let rt = tokio::runtime::Runtime::new().unwrap(); + rt.block_on(async move { + let _ = std::fs::remove_file(&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)); + builder.configure::("state", |reg| { + reg.buffer(BufferCfg::SingleLatest).with_remote_access(); + }); + let (db, runner) = builder.build().await.expect("build"); + db.producer::("state") + .unwrap() + .produce(State { n: 1 }); + tokio::spawn(runner.run()); + while !sock.exists() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + ready_tx.send(()).unwrap(); + let _ = stop_rx.await; + drop(db); + }); + rt.shutdown_timeout(Duration::from_secs(1)); + }); + ready_rx.recv().unwrap(); + Self { + stop: Some(stop), + thread: Some(thread), + } + } +} + +impl Drop for Server { + fn drop(&mut self) { + let _ = self.stop.take().unwrap().send(()); + self.thread.take().unwrap().join().unwrap(); + } +} + +#[tokio::test] +async fn the_client_is_kept_across_error_replies_and_replaced_once_stopped() { + let dir = tempfile::tempdir().unwrap(); + let sock = dir.path().join("aimdb.sock"); + let endpoint = sock.to_str().unwrap().to_string(); + let pool = ConnectionPool::new(); + + let server = Server::start(&sock); + let first = pool.get_drain_client(&endpoint).await.expect("connect"); + + // Error replies are answers on a healthy session: the client stays. + for missing in ["missing", "also-missing"] { + assert!(first.lock().await.get_record(missing).await.is_err()); + let again = pool.get_drain_client(&endpoint).await.unwrap(); + assert!(Arc::ptr_eq(&first, &again), "kept after an error reply"); + } + assert!(!first.lock().await.is_closed()); + + // The instance goes away: the engine stops. + drop(server); + let mut stopped = false; + for _ in 0..200 { + if first.lock().await.is_closed() { + stopped = true; + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(stopped, "the engine stops when the transport closes"); + + // The instance is back: the next call replaces the stopped client. + let _server = Server::start(&sock); + let next = pool.get_drain_client(&endpoint).await.expect("reconnect"); + assert!(!Arc::ptr_eq(&first, &next), "a stopped client is replaced"); + let got = next.lock().await.get_record("state").await.expect("get"); + assert_eq!(got.value["n"], 1); +} diff --git a/tools/aimdb-mcp/tests/serial_session.rs b/tools/aimdb-mcp/tests/serial_session.rs new file mode 100644 index 00000000..80a5c68f --- /dev/null +++ b/tools/aimdb-mcp/tests/serial_session.rs @@ -0,0 +1,225 @@ +//! `get_record` and `drain_record` share one serial session (#450). +//! +//! A UART never closes, so a board learns of a new client only from a `hello`, +//! and a `hello` resets its session, drain cursors included. If `get_record` +//! opened a second connection, it would either fail (the tty is already open) +//! or say `hello` again and lose the cursor `drain_record` reads. +//! +//! The board is a `SerialServer` behind a pty; `aimdb-mcp` opens the other +//! side as `serial://…`. A relay between the pty and the board counts the +//! `hello`s the board receives. + +#![cfg(all(feature = "transport-serial", target_os = "linux"))] + +use std::fs::File; +use std::io::{Read, Write}; +use std::os::fd::{FromRawFd, OwnedFd}; +use std::sync::atomic::{AtomicUsize, Ordering::SeqCst}; +use std::sync::Arc; +use std::time::Duration; + +use aimdb_core::buffer::BufferCfg; +use aimdb_core::AimDbBuilder; +use aimdb_mcp::tools; +use aimdb_serial_connector::framing::FrameAccumulator; +use aimdb_serial_connector::SerialServer; +use aimdb_tokio_adapter::net::TokioByteStream; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Reading { + n: u64, +} + +/// A pty pair: the master (the board's wire) and the slave's path (the +/// host's tty). The slave fd is held open so the master never sees a +/// hang-up between host opens. +struct Pty { + master: File, + path: String, + _slave: OwnedFd, +} + +fn openpty() -> Pty { + let (mut master, mut slave) = (0, 0); + // SAFETY: `openpty` writes two fresh fds; null name/termios/winsize are + // allowed. + let rc = unsafe { + libc::openpty( + &mut master, + &mut slave, + std::ptr::null_mut(), + std::ptr::null(), + std::ptr::null(), + ) + }; + assert_eq!(rc, 0, "openpty: {}", std::io::Error::last_os_error()); + // SAFETY: both fds were just returned by `openpty` and are owned here. + let (master, slave) = unsafe { (File::from_raw_fd(master), OwnedFd::from_raw_fd(slave)) }; + let path = std::fs::read_link(format!("/proc/self/fd/{}", { + use std::os::fd::AsRawFd; + slave.as_raw_fd() + })) + .unwrap() + .to_string_lossy() + .into_owned(); + Pty { + master, + path, + _slave: slave, + } +} + +/// Relay bytes between the pty master and the board's end of a duplex, +/// counting the `hello` requests on the way to the board. Blocking pty I/O +/// runs on plain threads, so nothing left blocked can stall the runtime. +fn relay(master: File, board_line: tokio::io::DuplexStream, hellos: Arc) { + let (mut line_rd, mut line_wr) = tokio::io::split(board_line); + let mut master_wr = master.try_clone().unwrap(); + let mut master_rd = master; + + // host → board + let (to_board, mut to_board_rx) = tokio::sync::mpsc::unbounded_channel::>(); + std::thread::spawn(move || { + let mut frames = FrameAccumulator::new(); + let mut buf = [0u8; 512]; + while let Ok(n) = master_rd.read(&mut buf) { + if n == 0 { + break; + } + frames.push_bytes(&buf[..n]); + while let Some(frame) = frames.next_frame() { + let Ok(frame) = frame else { continue }; + let msg: Value = serde_json::from_slice(&frame).unwrap_or(Value::Null); + if msg["method"] == "hello" { + hellos.fetch_add(1, SeqCst); + } + } + if to_board.send(buf[..n].to_vec()).is_err() { + break; + } + } + }); + tokio::spawn(async move { + while let Some(bytes) = to_board_rx.recv().await { + if line_wr.write_all(&bytes).await.is_err() { + break; + } + } + }); + + // board → host + let (to_host, to_host_rx) = std::sync::mpsc::channel::>(); + tokio::spawn(async move { + let mut buf = [0u8; 512]; + while let Ok(n) = line_rd.read(&mut buf).await { + if n == 0 || to_host.send(buf[..n].to_vec()).is_err() { + break; + } + } + }); + std::thread::spawn(move || { + while let Ok(bytes) = to_host_rx.recv() { + if master_wr.write_all(&bytes).is_err() { + break; + } + } + }); +} + +#[tokio::test] +async fn get_record_shares_the_drain_session_over_serial() { + // The board: a ring of events and a state record, served over the line. + let (board_end, line_end) = tokio::io::duplex(64 * 1024); + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(SerialServer::new(TokioByteStream(board_end))); + builder.configure::("events", |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 16 }) + .with_remote_access(); + }); + builder.configure::("state", |reg| { + reg.buffer(BufferCfg::SingleLatest).with_remote_access(); + }); + let (board, runner) = builder.build().await.expect("build board"); + tokio::spawn(runner.run()); + board + .producer::("state") + .unwrap() + .produce(Reading { n: 42 }); + + let pty = openpty(); + let hellos = Arc::new(AtomicUsize::new(0)); + relay(pty.master, line_end, hellos.clone()); + + tools::init_connection_pool(aimdb_mcp::connection::ConnectionPool::new()); + let endpoint = format!("serial://{}?baud=115200", pty.path); + let drain = || { + tools::drain_record(Some( + json!({ "endpoint": endpoint, "record_name": "events" }), + )) + }; + let get = + |name: &str| tools::get_record(Some(json!({ "endpoint": endpoint, "record_name": name }))); + + // 1. Open the ring's drain cursor. + let first = drain().await.expect("first drain"); + assert_eq!(first["count"], 0); + + // 2. The board publishes events. + let events = board.producer::("events").unwrap(); + for n in 1..=3 { + events.produce(Reading { n }); + } + + // 3. Read other records on the same session, misses included. + let got = get("state").await.expect("get_record over serial"); + assert_eq!(got["value"], json!({ "n": 42 })); + let err = get("missing") + .await + .expect_err("unknown record") + .to_string(); + assert!(err.contains("not_found"), "{err}"); + let err = get("events").await.expect_err("a ring").to_string(); + assert!(err.contains("unsupported"), "{err}"); + // An error reply from drain keeps the session too. + let err = tools::drain_record(Some( + json!({ "endpoint": endpoint, "record_name": "missing" }), + )) + .await + .expect_err("unknown record") + .to_string(); + assert!(err.contains("not_found"), "{err}"); + // Let any deferred cleanup run before the session is used again. + tokio::time::sleep(Duration::from_millis(50)).await; + + // 4. The cursor survived: the drain returns the queued events. + let mut ns = Vec::new(); + for _ in 0..100 { + let page = drain().await.expect("drain"); + ns.extend( + page["values"] + .as_array() + .unwrap() + .iter() + .map(|v| v["value"]["n"].clone()), + ); + if ns.len() >= 3 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(ns, [1, 2, 3]); + + // 5. One session the whole time, still usable. + let got = get("state").await.expect("get_record after the drains"); + assert_eq!(got["value"], json!({ "n": 42 })); + assert_eq!( + hellos.load(SeqCst), + 1, + "the board received exactly one hello" + ); +}