diff --git a/docs/memory.md b/docs/memory.md index ebc28a353..8420409b6 100644 --- a/docs/memory.md +++ b/docs/memory.md @@ -316,3 +316,4 @@ For multi-host coordination see [Distributed operations](distributed-operations. - [Dashboard](dashboard.md) — Memory tab - [Dashboard Memory tab — dedicated cognitive-memory graph](reference/dashboard-memory-tab.md) — the live `GET /api/memory/graph` visualization (nodes/edges, per-type filters, node inspection) - [Daemon mode](daemon-mode.md) — when consolidation runs +- [memory-ipc Write-Path Resilience](reference/memory-ipc-write-resilience.md) — the `RemoteCognitiveMemory` client reconnects and idempotently re-sends a framed write on a broken-pipe/EPIPE so large distillation writes are never silently dropped (#4731) diff --git a/docs/reference/memory-ipc-write-resilience.md b/docs/reference/memory-ipc-write-resilience.md new file mode 100644 index 000000000..d53693882 --- /dev/null +++ b/docs/reference/memory-ipc-write-resilience.md @@ -0,0 +1,232 @@ +--- +title: memory-ipc Write-Path Resilience (EPIPE reconnect + retry) +description: How the RemoteCognitiveMemory client survives a broken-pipe / EPIPE mid-write against the daemon's memory socket by reconnecting and idempotently re-sending the framed request, with a strict fail-closed guarantee that no memory write is ever silently dropped. +last_updated: 2026-07-26 +owner: cognitive-memory +doc_type: reference +related: + - ./rpc-wire-protocol.md + - ../memory.md + - ../architecture/cognitive-memory.md + - ../architecture/distillation-semantic-handoff.md +--- + +# memory-ipc Write-Path Resilience (EPIPE reconnect + retry) + +> Shipped in issue [#4731](https://github.com/rysweet/Simard/issues/4731). + +The [`RemoteCognitiveMemory`](./rpc-wire-protocol.md) client — the wire that +`meeting`, `engineer`, and the distiller subprocess use to reach the OODA +daemon's cognitive memory over `{socket_dir}/memory.sock` — is **resilient to a +peer that closes the socket mid-write**. When the daemon drops the connection +while the client is sending a request frame (surfacing as +`write-len: Broken pipe (os error 32)`), the client transparently reconnects +and re-sends the request instead of failing the memory write. + +This closes a recurring, systemic process-health defect in which large +distillation payloads (e.g. a 32-episode consolidation) and OODA cycle +transitions produced clusters of identical +`memory-ipc: connection error: rpc endpoint memory-ipc transport error: +write-len: Broken pipe (os error 32)` failures under load. + +## Guarantee: fail-closed, never silent + +The overriding contract is **no memory write may be silently dropped on EPIPE**: + +- Either the framed request **durably succeeds** after an automatic reconnect + and retry, **or** +- the error is **surfaced** to the caller as + `SimardError::RpcTransportError { endpoint: "memory-ipc", .. }`. + +There is **no silent fallback**, no best-effort drop, and no alternate +transport. A caller that receives `Ok(..)` can trust the write reached the +daemon's authoritative write boundary; a caller that receives `Err(..)` knows +the write did **not** commit and can decide how to react. + +## Scope of the retry + +Retries apply to the **write half only** — the phase where the client is +sending the request and the daemon has not yet committed anything: + +| Frame phase | Retried on EPIPE? | Why | +|-------------|-------------------|-----| +| `write-len` (length prefix) | ✅ yes | Server has not received or committed the request. | +| `write-body` (JSON payload) | ✅ yes | Server has not fully received the request; no commit. | +| `flush` | ✅ yes | Request not yet delivered. | +| `read-len` / `read-body` (response) | ❌ **never** | The server may have **already persisted** the mutation; a blind resend could duplicate it. | + +Read-half failures are propagated immediately as `RpcTransportError` and are +**not** retried. This deliberately sidesteps the double-apply hazard for a +mutation the daemon may already have committed. + +## EPIPE detection + +A failure is treated as a broken pipe when the underlying `std::io::Error` +satisfies **either** condition, evaluated on the raw error **before** it is +stringified into a `SimardError`: + +- `error.kind() == std::io::ErrorKind::BrokenPipe`, **or** +- `error.raw_os_error() == Some(32)` (Linux `EPIPE`). + +This reuses the `raw_os_error()` classification technique already established in +`operator_commands_ooda/daemon/helpers.rs` (which applies the same pattern to +`EMFILE`/`ENFILE`), extended here to the broken-pipe errno. + +## Retry bounds + +The retry loop is bounded by compile-time constants — there are **no +environment variables or config knobs** to amplify it (a DoS-hardening +requirement): + +| Constant | Value | Meaning | +|----------|-------|---------| +| `MAX_ATTEMPTS` | `3` | Total send attempts, including the first. | +| `BACKOFF` | `50 ms` | Fixed sleep between attempts. | + +On exhaustion (all 3 attempts hit a write-half EPIPE), the client returns +`SimardError::RpcTransportError` — it never returns `Ok`. + +Each attempt is still governed by the existing **30-second read/write socket +timeouts**, which are re-applied to every reconnected stream. + +## Reconnect behavior + +On a retryable write-half EPIPE the client: + +1. Emits a `warn` tracing event (reconnecting; carries `attempt`, + `endpoint = "memory-ipc"`, `socket_path`, and the `ErrorKind`/errno — **never + payload bytes**). +2. Opens a **fresh** connection to the **stored, immutable `socket_path`** — it + never re-derives, creates, `chmod`s, or falls back to a more permissive + socket (TOCTOU / socket-redirection defense). +3. Re-applies the 30-second read/write timeouts. +4. Performs an **inline `Ping`/`Pong` handshake** over the new stream and + verifies the response is **exactly `Pong`**. If the handshake returns + anything else (or fails), the reconnect is abandoned and the call returns + `RpcTransportError` **without resending the real payload**. +5. Swaps the new stream into the held `Mutex` guard (`*guard = new_stream`), so + the stale stream is dropped deterministically and concurrent callers observe + the healed connection. +6. Sleeps `BACKOFF`, then re-sends the original serialized request. + +The request is serialized **once** before the loop and the `Mutex` guard is +held **across** the reconnect + backoff, so concurrent memory writes on the +same client are serialized and never reordered. + +### Non-recursive reconnect + +`connect()` originally performed its `Ping` handshake through `call()`, which is +the very method that now reconnects. To avoid a `connect → call → reconnect → +call` recursion, reconnect uses a lower-level `connect_stream(path)` primitive +(connect + apply timeouts, **no** handshake) plus an **inline** `Ping`/`Pong` +exchange on the new stream. Both `connect()` and `reconnect()` share this +primitive. + +## Observability + +All new code paths emit **structured `tracing`** events (plus the existing +OTel wiring). There are **no** `print!` / `println!` / `eprintln!` calls in the +resilience path. + +| Event | Level | Fields | +|-------|-------|--------| +| Reconnecting after write-half EPIPE | `warn` | `attempt`, `endpoint`, `socket_path`, `error_kind`/`errno` | +| Retry attempt about to re-send | `debug` | `attempt`, `endpoint` | +| Terminal failure (attempts exhausted / bad handshake) | `error` | `attempt`, `endpoint`, `socket_path`, `error_kind`/`errno` | + +**Log confidentiality:** tracing events and the `RpcTransportError` message +carry only the attempt number, `endpoint = "memory-ipc"`, `errno`/`ErrorKind`, +and `socket_path`. They **never** include request payload, episode, or +distillation bytes. + +An optional reconnect/retry counter is incremented via +`cognitive_memory::metrics::increment(kind, site)` for dashboards. `increment` +and `cognitive_memory_silent_drop_count` read and write the **same** +`(kind, site)` bucket map, so the resilience path must choose `kind` labels that +keep the accounting semantically clean: + +- A successful reconnect + retry is a *recovery*, not a drop — it uses a + distinct recovery `kind` (e.g. `"epipe_reconnect"`) so a healed write is + **never** miscounted as lost data. +- Exhausting all attempts fails **closed** as a *surfaced* `RpcTransportError`. + That is a loud, observable failure — **not** a silent drop — so it must **not** + reuse the silent-drop `kind` either; if it is counted at all it uses its own + `kind` (e.g. `"epipe_exhausted"`). + +The `cognitive_memory_silent_drop_count` `kind` stays reserved for genuine +silent-drop accounting and is inflated by neither EPIPE recovery nor a loudly +surfaced exhaustion. + +## What did *not* change + +This fix is **additive and non-breaking**: + +- **No wire-format change.** The framing is still a 4-byte big-endian length + prefix followed by JSON. A single logical request is still exactly one frame; + the reconnected request is byte-identical to the original. Mixed old/new + fleets interoperate. +- **No chunking.** Splitting one request across multiple frames was rejected + because it would break the single-frame `read_frame` / `serve_connection` + reader. The existing **8 MiB `MAX_FRAME`** cap already accommodates a + 32-episode distillation payload, so no oversized-frame handling is needed. + `MAX_FRAME` is re-enforced on **every** read, including the post-reconnect + response read. +- **No public-API change.** `RemoteCognitiveMemory::connect`, `call`, and all + `CognitiveMemoryOps` methods keep their signatures. Resilience is internal to + `call()`. +- **`runtime_ipc` is untouched.** `src/runtime_ipc/mod.rs` is a separate + subprocess transport and is out of scope. + +## Affected code + +| File | Change | +|------|--------| +| `src/memory_ipc/mod.rs` | Add `write_frame_raw()` (returns `io::Result<()>` without `ipc_err()` stringification) and `is_broken_pipe(&io::Error) -> bool`; `write_frame` becomes a thin wrapper. | +| `src/memory_ipc/client.rs` | Primary fix: `connect_stream()` reconnect primitive, `reconnect()` with inline `Ping`/`Pong`, and the bounded write-half retry loop inside `call()`. | +| `src/memory_ipc/server.rs` | Verify-only: the accept/write loop already writes whole frames and is not the drop source; no functional change. | + +## Example: transparent recovery + +From a caller's perspective, nothing changes — the write just succeeds even if +the daemon briefly resets the connection mid-frame: + +```rust +let mem = RemoteCognitiveMemory::connect(&socket_path)?; + +// A large distillation write. If the daemon closes the socket mid-write +// (EPIPE), the client reconnects, re-handshakes, and re-sends automatically. +// This returns Ok only if the write durably reached the daemon. +let outcome = mem.remember_fact_gated( + concept, content, confidence, &tags, source_id, &source_episode_ids, pass_id, +)?; +``` + +If every one of the 3 attempts hits an EPIPE, the call fails loudly instead of +dropping the write: + +```text +rpc endpoint 'memory-ipc' transport error: write-len: Broken pipe (os error 32) +``` + +The caller gets an `Err(SimardError::RpcTransportError { .. })` and can retry at +a higher level, back off, or surface the failure — but the write is **never** +silently lost. + +## Regression coverage + +`src/memory_ipc/tests_epipe_resilience_4731.rs` uses a raw `UnixListener` +"malicious server" harness (based on the `tests_transport_roundtrip.rs` pattern) +to force mid-write resets: + +1. **Mid-write reset → durable delivery.** A server that reads a partial frame + then closes the socket forces one EPIPE; the client reconnects and re-sends, + and the payload is delivered intact on the second, healthy connection — no + data loss. +2. **Always-reset → surfaced error.** A server that always resets exhausts the + 3-attempt bound and the client returns `RpcTransportError` within the + attempt/timeout budget — never a silent `Ok`. +3. **Non-`Pong` reconnect → surfaced error, no resend.** If the post-reconnect + handshake returns anything other than `Pong`, the client aborts with + `RpcTransportError` and does **not** resend the real payload. +4. **Log confidentiality.** Emitted logs and the error message contain no + request/payload bytes. diff --git a/mkdocs.yml b/mkdocs.yml index e26b9d50a..5aaae05fe 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -344,6 +344,7 @@ nav: - Runtime Contracts: reference/runtime-contracts.md - Subprocess Prompt Delivery: prompt-delivery.md - RPC Wire Protocol: reference/rpc-wire-protocol.md + - memory-ipc Write-Path Resilience: reference/memory-ipc-write-resilience.md - No-`bridge` Naming Guard: reference/no-bridge-naming-guard.md - Argv-Free Copilot/OODA Invocation: reference/argv-free-copilot-invocation.md - Argv-Free Meeting/Signal Agent Proxy: reference/argv-free-meeting-agent-proxy.md diff --git a/src/memory_ipc/client.rs b/src/memory_ipc/client.rs index 67e73c4c2..fb3d6f6e1 100644 --- a/src/memory_ipc/client.rs +++ b/src/memory_ipc/client.rs @@ -3,16 +3,23 @@ use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; use std::sync::Mutex; +use std::thread; use std::time::Duration; +use tracing::{debug, error, warn}; + use crate::cognitive_memory::CognitiveMemoryOps; +use crate::cognitive_memory::metrics; use crate::error::{SimardError, SimardResult}; use crate::memory_cognitive::{ CognitiveEpisode, CognitiveFact, CognitiveProcedure, CognitiveProspective, CognitiveStatistics, CognitiveWorkingSlot, }; -use super::{FactWriteOutcome, MemoryRequest, MemoryResponse, ipc_err, read_frame, write_frame}; +use super::{ + FactWriteOutcome, MemoryRequest, MemoryResponse, ipc_err, is_broken_pipe, read_frame, + write_frame, write_frame_raw, +}; // ============================================================================ // Client @@ -29,6 +36,32 @@ impl RemoteCognitiveMemory { /// Connect to the daemon's memory socket. Returns an error if the socket /// doesn't exist, the daemon isn't listening, or the handshake fails. pub fn connect(socket_path: &Path) -> SimardResult { + let stream = Self::connect_stream(socket_path)?; + let client = Self { + stream: Mutex::new(stream), + socket_path: socket_path.to_path_buf(), + }; + // Handshake + match client.call(MemoryRequest::Ping)? { + MemoryResponse::Pong => Ok(client), + other => Err(SimardError::RpcSpawnFailed { + endpoint: "memory-ipc-client".into(), + reason: format!("handshake: expected Pong, got {other:?}"), + }), + } + } + + /// Open a fresh timeout-configured stream to `socket_path` WITHOUT a Ping + /// handshake. + /// + /// Factored out so the write-path reconnect ([`Self::reconnect`]) can reuse + /// the exact connect + timeout setup while performing its handshake inline + /// — breaking the `connect() -> call(Ping)` recursion that would otherwise + /// occur if reconnect went back through [`Self::connect`]. The stored, + /// immutable `socket_path` is the only source for reconnects, so a + /// reconnect can never be redirected to a different (possibly more + /// permissive) socket. + fn connect_stream(socket_path: &Path) -> SimardResult { if !socket_path.exists() { return Err(SimardError::RpcSpawnFailed { endpoint: "memory-ipc-client".into(), @@ -42,16 +75,34 @@ impl RemoteCognitiveMemory { // Short timeouts so a wedged daemon doesn't hang meeting forever. let _ = stream.set_read_timeout(Some(Duration::from_secs(30))); let _ = stream.set_write_timeout(Some(Duration::from_secs(30))); - let client = Self { - stream: Mutex::new(stream), - socket_path: socket_path.to_path_buf(), - }; - // Handshake - match client.call(MemoryRequest::Ping)? { - MemoryResponse::Pong => Ok(client), - other => Err(SimardError::RpcSpawnFailed { - endpoint: "memory-ipc-client".into(), - reason: format!("handshake: expected Pong, got {other:?}"), + Ok(stream) + } + + /// Re-establish the connection in place after a mid-write peer reset (issue + /// #4731). Opens a fresh stream via [`Self::connect_stream`], performs an + /// INLINE Ping/Pong handshake on it (never via [`Self::call`], to avoid + /// recursion), and only on a verified `Pong` swaps the new stream into the + /// held guard — deterministically dropping the stale stream. + /// + /// If the handshake returns anything other than `Pong`, the peer is + /// unverified and this returns an error WITHOUT swapping, so the caller + /// must not resend the real payload onto it. + fn reconnect(&self, stream: &mut UnixStream) -> SimardResult<()> { + let mut fresh = Self::connect_stream(&self.socket_path)?; + let ping = + serde_json::to_vec(&MemoryRequest::Ping).map_err(|e| ipc_err("serialize-ping", e))?; + write_frame(&mut fresh, &ping)?; + let resp_bytes = read_frame(&mut fresh)?; + let resp: MemoryResponse = + serde_json::from_slice(&resp_bytes).map_err(|e| ipc_err("parse-pong", e))?; + match resp { + MemoryResponse::Pong => { + *stream = fresh; + Ok(()) + } + other => Err(SimardError::RpcTransportError { + endpoint: "memory-ipc".into(), + reason: format!("reconnect handshake: expected Pong, got {other:?}"), }), } } @@ -62,16 +113,90 @@ impl RemoteCognitiveMemory { } fn call(&self, req: MemoryRequest) -> SimardResult { + // Bounded write-half reconnect+retry (issue #4731). The peer can close + // the socket mid-write under load, so a single large frame write hits + // EPIPE. Because a write-half EPIPE means the server never received or + // committed the request, it is safe to reconnect and idempotently + // re-send. Read-half failures are NEVER retried (the server may have + // already persisted). On exhaustion we surface `RpcTransportError` — + // never a silent `Ok`, never a dropped write. + const MAX_ATTEMPTS: usize = 3; + const BACKOFF: Duration = Duration::from_millis(50); + let bytes = serde_json::to_vec(&req).map_err(|e| ipc_err("serialize-request", e))?; let mut guard = self .stream .lock() .map_err(|e| ipc_err("lock-poisoned", e))?; - write_frame(&mut *guard, &bytes)?; - let resp_bytes = read_frame(&mut *guard)?; - let resp: MemoryResponse = - serde_json::from_slice(&resp_bytes).map_err(|e| ipc_err("parse-response", e))?; - Ok(resp) + + let mut attempt = 1usize; + loop { + match write_frame_raw(&mut *guard, &bytes) { + Ok(()) => { + // Write committed to the peer; read the response. Read-half + // errors propagate as-is (no retry — server may have + // persisted, retrying could duplicate the mutation). + let resp_bytes = read_frame(&mut *guard)?; + let resp: MemoryResponse = serde_json::from_slice(&resp_bytes) + .map_err(|e| ipc_err("parse-response", e))?; + return Ok(resp); + } + Err((phase, e)) => { + let broken = is_broken_pipe(&e); + if !(broken && attempt < MAX_ATTEMPTS) { + // Fail-closed: not a retriable broken pipe, or attempts + // exhausted. Surface the transport error; never drop + // the write silently. Diagnostics carry only transport + // metadata (endpoint, phase, errno) — never payload. + error!( + endpoint = "memory-ipc", + attempt, + max_attempts = MAX_ATTEMPTS, + phase, + error_kind = ?e.kind(), + errno = e.raw_os_error(), + "memory-ipc write failed terminally; surfacing transport error (no silent drop)" + ); + if broken { + metrics::increment("epipe_exhausted", "memory-ipc"); + } + return Err(ipc_err(phase, e)); + } + + warn!( + endpoint = "memory-ipc", + attempt, + max_attempts = MAX_ATTEMPTS, + phase, + error_kind = ?e.kind(), + errno = e.raw_os_error(), + "memory-ipc write hit broken pipe; reconnecting to retry" + ); + metrics::increment("epipe_reconnect", "memory-ipc"); + + // Reconnect + verified handshake, swapping the fresh stream + // into the held guard. A failed/unverified reconnect is + // surfaced immediately and the payload is NOT resent. + if let Err(reconnect_err) = self.reconnect(&mut guard) { + error!( + endpoint = "memory-ipc", + attempt, + "memory-ipc reconnect failed; surfacing transport error (no silent drop)" + ); + return Err(reconnect_err); + } + + debug!( + endpoint = "memory-ipc", + attempt, + next_attempt = attempt + 1, + "memory-ipc reconnected; retrying write after backoff" + ); + thread::sleep(BACKOFF); + attempt += 1; + } + } + } } fn unexpected(name: &str, got: MemoryResponse) -> SimardError { diff --git a/src/memory_ipc/mod.rs b/src/memory_ipc/mod.rs index f7dc9b67e..09f0c5d51 100644 --- a/src/memory_ipc/mod.rs +++ b/src/memory_ipc/mod.rs @@ -15,7 +15,7 @@ //! Fallback: if no daemon is running (socket absent or connect fails), the //! caller should open the DB directly. -use std::io::{Read, Write}; +use std::io::{self, Read, Write}; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -23,6 +23,14 @@ use std::sync::Arc; mod tests_client_isolation; #[cfg(test)] mod tests_default_state_root_1967; +// TDD (RED) for issue #4731: memory-ipc write-path resilience against a +// mid-write peer reset (EPIPE). The client is expected to reconnect and +// idempotently re-send a write-half frame on BrokenPipe/errno-32, bounded to a +// few attempts, surfacing RpcTransportError on exhaustion — never a silent drop. +// Until the bounded reconnect+retry lands in `client::call`, these behavioural +// tests fail (no reconnect ⇒ no durable delivery / no second connection). +#[cfg(test)] +mod tests_epipe_resilience_4731; #[cfg(test)] mod tests_launcher; #[cfg(test)] @@ -370,15 +378,47 @@ pub(crate) fn ipc_err(ctx: &str, e: impl std::fmt::Display) -> SimardError { /// single fact is a few hundred bytes) yet small enough to bound abuse. pub(crate) const MAX_FRAME: usize = 8 * 1024 * 1024; -pub(crate) fn write_frame(w: &mut W, payload: &[u8]) -> SimardResult<()> { - let len = u32::try_from(payload.len()).map_err(|_| SimardError::RpcTransportError { - endpoint: "memory-ipc".into(), - reason: format!("message too large: {} bytes", payload.len()), +/// Whether an `io::Error` is a broken-pipe / `EPIPE` (errno 32). +/// +/// Branch on the RAW `io::Error` here — BEFORE [`ipc_err`] stringifies it — +/// so the client's write-path retry can distinguish a mid-write peer reset +/// (safe to reconnect + resend, the server never committed the request) from +/// any other transport failure. Matches on both the portable +/// [`io::ErrorKind::BrokenPipe`] and the raw `errno == 32`, mirroring the +/// `raw_os_error()` precedent in `operator_commands_ooda/daemon/helpers.rs`. +pub(crate) fn is_broken_pipe(e: &io::Error) -> bool { + e.kind() == io::ErrorKind::BrokenPipe || e.raw_os_error() == Some(32) +} + +/// Write one length-prefixed frame, surfacing the failing **phase** and the +/// raw [`io::Error`] instead of a stringified [`SimardError`]. +/// +/// Keeping the un-stringified `io::Error` lets the client branch on +/// [`is_broken_pipe`] (issue #4731 write-path resilience) before deciding +/// whether the write half is safe to reconnect + resend. [`write_frame`] is a +/// thin wrapper over this that maps `(phase, io::Error)` through [`ipc_err`], +/// so server and all other callers are unchanged. +pub(crate) fn write_frame_raw( + w: &mut W, + payload: &[u8], +) -> Result<(), (&'static str, io::Error)> { + let len = u32::try_from(payload.len()).map_err(|_| { + ( + "write-len", + io::Error::new( + io::ErrorKind::InvalidInput, + format!("message too large: {} bytes", payload.len()), + ), + ) })?; w.write_all(&len.to_be_bytes()) - .map_err(|e| ipc_err("write-len", e))?; - w.write_all(payload).map_err(|e| ipc_err("write-body", e))?; - w.flush().map_err(|e| ipc_err("flush", e)) + .map_err(|e| ("write-len", e))?; + w.write_all(payload).map_err(|e| ("write-body", e))?; + w.flush().map_err(|e| ("flush", e)) +} + +pub(crate) fn write_frame(w: &mut W, payload: &[u8]) -> SimardResult<()> { + write_frame_raw(w, payload).map_err(|(phase, e)| ipc_err(phase, e)) } pub(crate) fn read_frame(r: &mut R) -> SimardResult> { diff --git a/src/memory_ipc/tests_epipe_resilience_4731.rs b/src/memory_ipc/tests_epipe_resilience_4731.rs new file mode 100644 index 000000000..e66e1abfb --- /dev/null +++ b/src/memory_ipc/tests_epipe_resilience_4731.rs @@ -0,0 +1,460 @@ +//! TDD (RED) regression tests for issue #4731 — memory-IPC **write-path +//! resilience** against a mid-write peer reset (EPIPE / `Broken pipe`). +//! +//! ## The defect +//! +//! The memory-ipc RPC client repeatedly failed in production with +//! `memory-ipc: connection error: ... write-len: Broken pipe (os error 32)`, +//! clustered around large distillation payload writes and OODA cycle +//! transitions. The peer closes the socket mid-write under load, so a single +//! large frame write hits `EPIPE` and the whole memory write is lost — a +//! *silent* dropped write, the exact class the launcher's no-silent-fallback +//! rule exists to prevent. +//! +//! ## The contract these tests pin (design-ready requirements) +//! +//! 1. On a **write-half** `BrokenPipe`/`EPIPE` (errno 32), the client +//! [`RemoteCognitiveMemory::call`](super::RemoteCognitiveMemory) must +//! **reconnect** (fresh connect + inline Ping/Pong handshake, timeouts +//! re-applied) and **idempotently re-send** the framed request, bounded to +//! 3 attempts with brief backoff. A large payload that hits a mid-write +//! reset must be **durably delivered** on the reconnect — no data loss, +//! no truncation. *(test `mid_write_reset_reconnects_and_delivers_payload`)* +//! 2. When **every** attempt hits a reset, the client must **surface** a +//! [`SimardError::RpcTransportError`] after exhausting its bounded retries — +//! never a silent `Ok`, never an alternate-transport fallback. +//! *(test `persistent_reset_surfaces_rpc_transport_error_never_silent`)* +//! 3. If the post-reset reconnect handshake does **not** return exactly +//! `Pong`, the client must abort with `RpcTransportError` and must **not** +//! resend the real payload onto an unverified peer. +//! *(test `non_pong_reconnect_aborts_without_resending_payload`)* +//! 4. Retry/exhaustion diagnostics must carry only transport metadata +//! (endpoint, errno/ErrorKind, attempt) and **never** the payload bytes. +//! *(test `surfaced_error_never_leaks_payload_bytes`)* +//! +//! ## Why these fail today (RED) +//! +//! The current `call()` performs a single `write_frame` with no reconnect. So: +//! * (1) returns `Err` instead of the durable `Ok` → RED. +//! * (2)/(3) never open a second connection, so the "a reconnect was +//! attempted" assertion (`connections >= 2`) fails → RED. +//! * (4) additionally asserts a retry was attempted → RED. +//! +//! All four turn GREEN once the bounded write-half reconnect+retry lands. +//! +//! ## Harness +//! +//! A hermetic, scripted **raw `UnixListener` mock server** (no real store, no +//! env, no global state) that drives each accepted connection through a fixed +//! [`ConnScript`]. The `HandshakeThenReset` script completes the client's +//! handshake (so `connect()` succeeds) and then drops the socket, forcing the +//! client's *next* large-frame write to hit `EPIPE`. This reproduces the +//! production "peer closes mid-write under load" failure deterministically. + +use std::io::{self, Read, Write}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; + +use crate::cognitive_memory::CognitiveMemoryOps; +use crate::error::SimardError; + +use super::{MemoryRequest, MemoryResponse, RemoteCognitiveMemory}; + +// --------------------------------------------------------------------------- +// Payload sizing +// --------------------------------------------------------------------------- + +/// A payload large enough to overflow the Unix-socket send buffer, so the +/// client's `write_all` blocks and then observes the peer's mid-write close as +/// `EPIPE` (rather than completing into a kernel buffer). 2 MiB is comfortably +/// under the 8 MiB `MAX_FRAME` cap yet far larger than the default send buffer. +const PAYLOAD_BYTES: usize = 2 * 1024 * 1024; + +/// Sentinel woven into the payload so a confidentiality test can prove the +/// surfaced error never echoes payload bytes. +const PAYLOAD_SENTINEL: &str = "SENTINEL_SECRET_PAYLOAD_MARKER_4731"; + +fn big_episode_content() -> String { + let mut s = String::with_capacity(PAYLOAD_BYTES + PAYLOAD_SENTINEL.len()); + s.push_str(PAYLOAD_SENTINEL); + s.extend(std::iter::repeat_n('x', PAYLOAD_BYTES)); + s +} + +// --------------------------------------------------------------------------- +// Raw framing (4-byte big-endian length prefix + JSON body), matching the +// module wire format, implemented independently here so the mock never depends +// on the code under test for its transport behaviour. +// --------------------------------------------------------------------------- + +fn read_frame_raw(stream: &mut UnixStream) -> io::Result> { + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf)?; + let len = u32::from_be_bytes(len_buf) as usize; + let mut body = vec![0u8; len]; + stream.read_exact(&mut body)?; + Ok(body) +} + +fn write_frame_raw(stream: &mut UnixStream, payload: &[u8]) -> io::Result<()> { + let len = u32::try_from(payload.len()).expect("mock frame within u32"); + stream.write_all(&len.to_be_bytes())?; + stream.write_all(payload)?; + stream.flush() +} + +fn write_response(stream: &mut UnixStream, resp: &MemoryResponse) -> io::Result<()> { + let bytes = serde_json::to_vec(resp).expect("serialize mock response"); + write_frame_raw(stream, &bytes) +} + +// --------------------------------------------------------------------------- +// Scripted mock server +// --------------------------------------------------------------------------- + +/// Per-connection behaviour for the mock server. +#[derive(Clone, Copy)] +enum ConnScript { + /// Complete the client's Ping→Pong handshake, then drop the socket. The + /// client's *subsequent* large-frame write hits a mid-write `EPIPE`. Used + /// both for the `connect()` socket (whose later `store_episode` write is + /// what resets) and for "every attempt resets" scenarios. + HandshakeThenReset, + /// Complete the handshake, then read the full request frame (capturing it + /// for durable-delivery assertions) and answer with `Id("stored-ok")`. + HandshakeThenServe, + /// Answer the reconnect handshake with a **non-`Pong`** frame, then drop. + /// Verifies the client refuses to resend onto an unverified peer. + BadHandshake, +} + +#[derive(Default)] +struct MockState { + /// Number of accepted connections (each reconnect adds one). + connections: usize, + /// Request frames the mock actually *accepted and read in full* on a + /// serving connection. Length 0 means no server ever received the payload. + served_payloads: Vec>, + /// Number of connections that presented a non-`Pong` reconnect handshake. + bad_handshakes: usize, +} + +struct MockServer { + sock: PathBuf, + state: Arc>, + running: Arc, + join: Option>, + _dir: tempfile::TempDir, +} + +impl MockServer { + /// Bind a socket in a fresh `TempDir` and spawn the accept loop. `scripts` + /// are applied to accepted connections in order; any connection beyond the + /// list uses `default` (so "reset forever" scenarios need only a default). + fn spawn(scripts: Vec, default: ConnScript) -> Self { + let dir = tempfile::tempdir().expect("tempdir"); + let sock = dir.path().join("memory.sock"); + let listener = UnixListener::bind(&sock).expect("bind mock socket"); + listener + .set_nonblocking(true) + .expect("nonblocking listener"); + + let state = Arc::new(Mutex::new(MockState::default())); + let running = Arc::new(AtomicBool::new(true)); + + let state_t = Arc::clone(&state); + let running_t = Arc::clone(&running); + let join = thread::Builder::new() + .name("mock-memory-ipc-4731".into()) + .spawn(move || accept_loop(listener, scripts, default, state_t, running_t)) + .expect("spawn mock accept loop"); + + Self { + sock, + state, + running, + join: Some(join), + _dir: dir, + } + } + + fn socket_path(&self) -> &std::path::Path { + &self.sock + } + + fn snapshot(&self) -> (usize, usize, usize) { + let st = self.state.lock().expect("mock state lock"); + (st.connections, st.served_payloads.len(), st.bad_handshakes) + } + + /// The single request frame a serving connection accepted, decoded as a + /// [`MemoryRequest`]. Panics if the mock did not serve exactly one frame. + fn only_served_request(&self) -> MemoryRequest { + let st = self.state.lock().expect("mock state lock"); + assert_eq!( + st.served_payloads.len(), + 1, + "expected exactly one fully-served request frame" + ); + serde_json::from_slice(&st.served_payloads[0]).expect("decode served request") + } +} + +impl Drop for MockServer { + fn drop(&mut self) { + self.running.store(false, Ordering::SeqCst); + if let Some(j) = self.join.take() { + let _ = j.join(); + } + } +} + +fn accept_loop( + listener: UnixListener, + scripts: Vec, + default: ConnScript, + state: Arc>, + running: Arc, +) { + let mut index = 0usize; + while running.load(Ordering::SeqCst) { + match listener.accept() { + Ok((mut stream, _addr)) => { + // Accepted sockets are blocking on Linux regardless of the + // listener's flag, but make the intent explicit. + let _ = stream.set_nonblocking(false); + let script = scripts.get(index).copied().unwrap_or(default); + index += 1; + { + let mut st = state.lock().expect("mock state lock"); + st.connections += 1; + } + // A handler error just ends this connection; the client sees it + // as the reset/close the scenario intends. + let _ = handle_conn(&mut stream, script, &state); + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(5)); + } + Err(_) => break, + } + } +} + +fn handle_conn( + stream: &mut UnixStream, + script: ConnScript, + state: &Arc>, +) -> io::Result<()> { + match script { + ConnScript::HandshakeThenReset => { + handshake_pong(stream)?; + // Drop (return) closes the socket. The client's next large-frame + // write on this connection now hits EPIPE mid-write. + Ok(()) + } + ConnScript::HandshakeThenServe => { + handshake_pong(stream)?; + let frame = read_frame_raw(stream)?; + { + let mut st = state.lock().expect("mock state lock"); + st.served_payloads.push(frame); + } + write_response(stream, &MemoryResponse::Id("stored-ok".into())) + } + ConnScript::BadHandshake => { + // Consume the reconnect Ping, then answer with a NON-Pong frame. + let _ = read_frame_raw(stream)?; + { + let mut st = state.lock().expect("mock state lock"); + st.bad_handshakes += 1; + } + write_response( + stream, + &MemoryResponse::Error("mock-refuses-handshake".into()), + ) + // Return/drop without ever reading a second frame: proves the + // client did NOT resend the payload after a failed handshake. + } + } +} + +/// Read one request frame (expected `Ping`) and answer `Pong`, completing a +/// client handshake so `connect()` / reconnect succeeds. +fn handshake_pong(stream: &mut UnixStream) -> io::Result<()> { + let frame = read_frame_raw(stream)?; + // Best-effort decode; the client only ever sends Ping here, but we don't + // hard-fail the mock on an unexpected shape. + let _req: Result = serde_json::from_slice(&frame); + write_response(stream, &MemoryResponse::Pong) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +/// (1) A large write that hits a mid-write peer reset must be reconnected and +/// re-sent so the payload is **durably delivered** with no loss/truncation. +/// +/// RED today: the current single-shot `call()` returns `Err` on the EPIPE, so +/// `store_episode` never yields the durable `Ok("stored-ok")`. +#[test] +fn mid_write_reset_reconnects_and_delivers_payload() { + // conn1: serves connect() handshake, then resets the store_episode write. + // conn2: serves the reconnected, re-sent request in full. + let mock = MockServer::spawn( + vec![ + ConnScript::HandshakeThenReset, + ConnScript::HandshakeThenServe, + ], + ConnScript::HandshakeThenServe, + ); + let client = + RemoteCognitiveMemory::connect(mock.socket_path()).expect("connect + Ping handshake"); + + let content = big_episode_content(); + let id = client + .store_episode(&content, "distillation", None) + .expect("large write must survive a mid-write reset and be delivered"); + + assert_eq!( + id, "stored-ok", + "the durable response must come from the reconnected server" + ); + + let (connections, served, _bad) = mock.snapshot(); + assert!( + connections >= 2, + "client must reconnect after the mid-write reset (saw {connections} connection(s))" + ); + assert_eq!( + served, 1, + "the payload must be delivered exactly once after reconnect" + ); + + // No data loss: the fully-served frame must carry the entire payload. + match mock.only_served_request() { + MemoryRequest::StoreEpisode { content: got, .. } => { + assert_eq!( + got.len(), + content.len(), + "delivered payload was truncated (data loss)" + ); + assert_eq!(got, content, "delivered payload differs from what was sent"); + } + other => panic!("expected StoreEpisode after reconnect, got {other:?}"), + } +} + +/// (2) When every attempt resets, the client must exhaust its bounded retries +/// and **surface** an `RpcTransportError` — never silently succeed, never +/// silently drop the write. +/// +/// RED today: no reconnect is attempted, so `connections` stays at 1. +#[test] +fn persistent_reset_surfaces_rpc_transport_error_never_silent() { + let mock = MockServer::spawn(vec![], ConnScript::HandshakeThenReset); + let client = + RemoteCognitiveMemory::connect(mock.socket_path()).expect("connect + Ping handshake"); + + let content = big_episode_content(); + let err = client + .store_episode(&content, "distillation", None) + .expect_err("a persistently-resetting peer must surface an error, never a silent Ok"); + + assert!( + matches!(err, SimardError::RpcTransportError { .. }), + "exhaustion must surface RpcTransportError, got: {err:?}" + ); + + let (connections, served, _bad) = mock.snapshot(); + assert!( + connections >= 2, + "client must attempt at least one reconnect before giving up (saw {connections})" + ); + assert_eq!( + served, 0, + "no server ever accepted the payload — it must be surfaced as an error, not dropped silently" + ); +} + +/// (3) If the reconnect handshake does not return exactly `Pong`, the client +/// must abort with `RpcTransportError` and must **not** resend the payload onto +/// the unverified peer. +/// +/// RED today: no reconnect happens, so the BadHandshake connection is never +/// opened (`connections` stays at 1). +#[test] +fn non_pong_reconnect_aborts_without_resending_payload() { + // conn1: serves connect() handshake, resets the write. + // conn2: answers the reconnect handshake with a non-Pong frame. + let mock = MockServer::spawn( + vec![ConnScript::HandshakeThenReset, ConnScript::BadHandshake], + ConnScript::BadHandshake, + ); + let client = + RemoteCognitiveMemory::connect(mock.socket_path()).expect("connect + Ping handshake"); + + let content = big_episode_content(); + let err = client + .store_episode(&content, "distillation", None) + .expect_err("a non-Pong reconnect handshake must surface an error"); + + assert!( + matches!(err, SimardError::RpcTransportError { .. }), + "a failed reconnect handshake must surface RpcTransportError, got: {err:?}" + ); + + let (connections, served, bad) = mock.snapshot(); + assert!( + connections >= 2, + "client must attempt the reconnect (saw {connections} connection(s))" + ); + assert!( + bad >= 1, + "the reconnect handshake must have been exercised and rejected" + ); + assert_eq!( + served, 0, + "the client must NOT resend the payload after an unverified (non-Pong) handshake" + ); +} + +/// (4) The surfaced transport error (and thus any diagnostics derived from it) +/// must never echo the payload bytes. Also asserts a reconnect was attempted so +/// this specifically guards the retry/exhaustion path. +/// +/// RED today: the `connections >= 2` retry assertion fails (no reconnect). +#[test] +fn surfaced_error_never_leaks_payload_bytes() { + let mock = MockServer::spawn(vec![], ConnScript::HandshakeThenReset); + let client = + RemoteCognitiveMemory::connect(mock.socket_path()).expect("connect + Ping handshake"); + + let content = big_episode_content(); + let err = client + .store_episode(&content, "distillation", None) + .expect_err("persistent reset must surface an error"); + + let rendered = err.to_string(); + assert!( + !rendered.contains(PAYLOAD_SENTINEL), + "surfaced error must not leak payload bytes; rendered = {rendered:?}" + ); + assert!( + rendered.len() < 4096, + "surfaced error must be a bounded transport diagnostic, not the payload; len = {}", + rendered.len() + ); + + let (connections, served, _bad) = mock.snapshot(); + assert!( + connections >= 2, + "confidentiality must hold on the retry path — a reconnect must have been attempted (saw {connections})" + ); + assert_eq!(served, 0, "no server accepted the payload"); +}