feat: add hostile harness, multi-session driver, and frame header deadline

README.md

@@ -55,7 +55,7 @@ bunshin rns-keygen --key server.rns.key
 bunshin serve --rns --key server.key --rns-key server.rns.key --db mail.db --rns-udp 0.0.0.0:4242
 ```
 
-`rns-keygen` writes the 64-byte Reticulum identity and prints the destination hash; publish `smol+rns://<user>@<hash>` as the server's address. RNS-specific limits (`--rns-max-envelope`, `--rns-fetch-budget`, `--rns-max-links`, `--rns-rate-link-requests`, `--rns-rate-link-bytes`) default to mesh-friendly 32 KiB caps; the UDP interface is the one supported transport interface so far. See [RNS.md](RNS.md).
+`rns-keygen` writes the 64-byte Reticulum identity and prints the destination hash; publish `smol+rns://<user>@<hash>` as the server's address. RNS-specific limits (`--rns-max-envelope`, `--rns-fetch-budget`, `--rns-max-links`, `--rns-rate-link-requests`, `--rns-rate-link-bytes`) default to mesh-friendly 32 KiB caps; `--rns-link-idle` (default 300 s) reaps links whose client went silent, since nothing else times a link out; the UDP interface is the one supported transport interface so far. See [RNS.md](RNS.md).
 
 ## Building
 
@@ -77,6 +77,8 @@ bunshin serve --key server.key --db mail.db --host 0.0.0.0 --port 1961
 
 `serve` accepts `--max-envelope`, `--quota`, `--requests-quota`, `--retention-days`, `--requests-retention-days`, `--max-tokens`, `--invite-token`, `--rate-connections`, `--rate-sends` and `--rate-tokens` to control size limits, the mailbox's two quota tiers, their retention, the accept-token cap, registration gating and abuse control. Run `bunshin serve --help` for defaults.
 
+`--verbose` (or `services.bunshin.verbose` in the NixOS module) raises logging to debug level, which adds server-side metrics on both carriers: per-operation timing, request and response sizes, handshake duration, per-session summaries, and RNS link events. The default info level stays quiet on success.
+
 ## Status
 
 Implements SPEC.md version 1.1 in full: `AUTH`, `RESOLVE`, `SEND`, `FETCH`, `DELETE` and `REGISTER`, including accept tokens and the main/requests tier split, fetch cursors, 32-byte message ids, `REGISTER` proof of possession, and dual-signed key rotation chains.

RNS.md

@@ -126,6 +126,7 @@ One process serves both carriers: shared `Store`, shared config, one systemd uni
 | `--rns-max-links` | 100 | concurrent links — refused past the cap |
 | `--rns-rate-link-requests` | 60 | per link per minute |
 | `--rns-rate-link-bytes` | 1048576 | per link per minute, via `ByteRateLimiter` |
+| `--rns-link-idle` | 300 | seconds of silence before a link is reaped; 0 disables the reaper |
 | `--rns-udp` | `127.0.0.1:4242` | UDP interface to listen on, host[:port] |
 | `--rns-udp-forward` | unset | optional forward target; without it the interface replies to the last datagram's source |
 
@@ -144,8 +145,9 @@ The per-link request limiter reuses `RateLimiter` keyed by the link hex. The per
 ## 9. Remaining open items
 
 1. Cross-transport envelope size: per-transport caps are enforced; a 768 KiB TCP envelope is still delivered whole to an RNS FETCH. Operators should apply the smaller cap mailbox-wide when in doubt (upstream spec 13.7).
-2. Interfaces beyond UDP (RNS-compatible TCP hub/client, I2P): microReticulum ships none; the UDP interface is the first supported one.
-3. The reference `test_interop/run_request_response.sh` harness was not rerun against the shim; its PlatformIO build step is unavailable here. The equivalent interop was exercised directly against `smolmail_rns.py` (§10.6).
+2. Upstream `microReticulum` deadlock: a client dying mid-resource-transfer wedges the single transport loop forever — `Resource::request_next` sends its `RESOURCE_REQ` through `Transport::outbound` from inside the `jobs()` resource tick, and `outbound` busy-waits on `_jobs_running`, a flag `jobs()` itself holds. No announces, no new links, mesh dead until restart. The vendored source is a local fork carrying the fix (same-thread reentrant calls skip the cycle wait); the equivalent report is filed upstream.
+3. Interfaces beyond UDP (RNS-compatible TCP hub/client, I2P): microReticulum ships none; the UDP interface is the first supported one.
+4. The reference `test_interop/run_request_response.sh` harness was not rerun against the shim; its PlatformIO build step is unavailable here. The equivalent interop was exercised directly against `smolmail_rns.py` (§10.6).
 
 Formerly open items resolved during implementation: raw identity bytes exist (`load_private_key`); the shim ABI and run-loop drive pattern are §7.1/§7.2; `ALLOW_ALL` matches the reference; announcing at startup + every 2 h, unconditional even when invite-only (the invite gates REGISTER, not the link); the interface question is §7.4.
 

THIRD_PARTY_NOTICES.md

@@ -1,6 +1,6 @@
 # Third-Party Notices
 
-bunshin incorporates the third-party components listed below. Components are grouped by license, each with the copyright notice that applies to it, followed by the full license text. A component listed under several licenses is multi-licensed under those options; any one of them may be chosen. All components are redistributed unmodified from their published sources.
+bunshin incorporates the third-party components listed below. Components are grouped by license, each with the copyright notice that applies to it, followed by the full license text. A component listed under several licenses is multi-licensed under those options; any one of them may be chosen. Components are redistributed from their published sources, except where a bullet notes a local patch; all other components are unmodified.
 
 The Rust crates listed are all crates resolved by `Cargo.lock`, including transitive dependencies, which are statically linked into the bunshin binary.
 The SQLite C library is bundled in source form by the `libsqlite3-sys` crate and is public domain (see the final section).
@@ -102,7 +102,7 @@ The libraries marked "vendored source" are pinned nix flake inputs; each is iden
 - zerocopy-derive 0.8.59 - Copyright 2023 The Fuchsia Authors
 - zeroize 1.9.0 - Copyright (c) 2018-2026 The RustCrypto Project Developers
 - zeroize_derive 1.5.0 - Copyright (c) 2019-2026 The RustCrypto Project Developers
-- microReticulum (vendored source, https://github.com/attermann/microReticulum, rev 40fa628809d5)
+- microReticulum (vendored source, https://github.com/attermann/microReticulum, rev 40fa628809d5; patched locally — same-thread reentrant calls skip the Transport::outbound/inbound jobs-cycle wait, fixing a transport-loop deadlock, see RNS.md §9)
 - microStore (vendored source, https://github.com/attermann/microStore, rev 0f28567fe00a)
 
 ```

flake.lock

@@ -21,18 +21,18 @@
     "microReticulum": {
       "flake": false,
       "locked": {
-        "lastModified": 1784575896,
-        "narHash": "sha256-cqo+BJ3tsmw4fD+SL0D3X9FKCgf+n0VQng2pSIuyK/8=",
-        "owner": "attermann",
-        "repo": "microReticulum",
-        "rev": "40fa628809d57140180c1c833559ab96fec992c1",
-        "type": "github"
+        "lastModified": 1790668417,
+        "narHash": "sha256-BrOkpepwcCfB7o9TOYOQ3SnZfwHVx3WyUTJsIGwAdLk=",
+        "ref": "fix/outbound-reentrancy-deadlock",
+        "rev": "03be60051d40e556933101289c0203d9f3308b62",
+        "revCount": 226,
+        "type": "git",
+        "url": "file:///mnt/data/Projects/code/microReticulum"
       },
       "original": {
-        "owner": "attermann",
-        "repo": "microReticulum",
-        "rev": "40fa628809d57140180c1c833559ab96fec992c1",
-        "type": "github"
+        "ref": "fix/outbound-reentrancy-deadlock",
+        "type": "git",
+        "url": "file:///mnt/data/Projects/code/microReticulum"
       }
     },
     "nixpkgs": {

flake.nix

@@ -9,8 +9,13 @@
     # so every microReticulum FetchContent dependency is vendored as its own
     # flake input and handed over via RNS_<DEP>_SOURCE_DIR. microReticulum
     # is pinned at the commit RNS.md sec 5 verified against.
+    # Local fork of attermann/microReticulum at rev 40fa628809d5 plus a
+    # local patch: Transport::outbound/inbound skip their jobs-cycle wait
+    # for same-thread reentrant calls, which otherwise deadlock the single
+    # transport loop when a stalled resource tick sends a RESOURCE_REQ
+    # from inside jobs() (see RNS.md sec 9).
     microReticulum = {
-      url = "github:attermann/microReticulum/40fa628809d57140180c1c833559ab96fec992c1";
+      url = "git+file:///mnt/data/Projects/code/microReticulum?ref=fix/outbound-reentrancy-deadlock";
       flake = false;
     };
     rns-arduinojson = {
@@ -220,6 +225,15 @@
                 description = "Open the configured TCP port in the firewall.";
               };
 
+              verbose = mkOption {
+                type = types.bool;
+                default = false;
+                description = ''
+                  Enable debug-level logging (--verbose): per-op timing,
+                  session summaries and RNS link events on both carriers.
+                '';
+              };
+
               rns = {
                 enable = mkEnableOption "the RNS carrier alongside TCP (needs an rns-built package)";
 
@@ -263,6 +277,15 @@
                   description = "Max request bytes per minute, per link.";
                 };
 
+                linkIdleSecs = mkOption {
+                  type = types.ints.unsigned;
+                  default = 300;
+                  description = ''
+                    Seconds after which an idle RNS link is reaped, freeing
+                    its session and link slot; 0 disables the reaper.
+                  '';
+                };
+
                 udpListenPort = mkOption {
                   type = types.port;
                   default = 4242;
@@ -328,11 +351,14 @@
                         --rns-max-links ${toString cfg.rns.maxLinks}
                         --rns-rate-link-requests ${toString cfg.rns.rateLinkRequests}
                         --rns-rate-link-bytes ${toString cfg.rns.rateLinkBytes}
+                        --rns-link-idle ${toString cfg.rns.linkIdleSecs}
                         --rns-udp 0.0.0.0:${toString cfg.rns.udpListenPort}
                       )
                     ''}
                     ${lib.optionalString (cfg.rns.enable && cfg.rns.udpForward != null)
                       ''args+=(--rns-udp-forward ${lib.escapeShellArg cfg.rns.udpForward})''}
+                    ${lib.optionalString cfg.verbose
+                      ''args+=(--verbose)''}
                     ${lib.optionalString (cfg.inviteToken != null)
                       ''args+=(--invite-token ${lib.escapeShellArg cfg.inviteToken})''}
                     ${lib.optionalString (cfg.inviteTokenFile != null)

src/channel.rs

@@ -9,7 +9,10 @@ use std::net::TcpStream;
 
 use snow::{Builder, TransportState};
 
-use crate::proto::{ProtocolError, MAX_FRAME, NOISE_PARAMS, NOISE_PAYLOAD, PROLOGUE};
+use crate::proto::{
+    ProtocolError, HEADER_TIMEOUT_SECS, IDLE_TIMEOUT_SECS, MAX_FRAME, NOISE_PARAMS, NOISE_PAYLOAD,
+    PROLOGUE,
+};
 
 /// Runs the Noise_NX responder handshake. The initiator stays anonymous;
 /// only we hold a static key. Returns the transport state and the
@@ -72,6 +75,15 @@ impl Channel {
     }
 
     fn read_noise(&mut self) -> anyhow::Result<Vec<u8>> {
+        // A frame already started must finish quickly; only a client idle
+        // between frames earns the full session idle timeout.
+        let deadline = if self.buf.is_empty() {
+            IDLE_TIMEOUT_SECS
+        } else {
+            HEADER_TIMEOUT_SECS
+        };
+        self.stream
+            .set_read_timeout(Some(std::time::Duration::from_secs(deadline)))?;
         let len = read_u16_len(&mut self.stream)?;
         let mut ciphertext = vec![0u8; len];
         read_exact_into(&mut self.stream, &mut ciphertext)?;

src/harness.rs

@@ -0,0 +1,617 @@
+//! Hostile transport harness (RNS.md sec 9 roster): a client that completes
+//! the real Noise_NX handshake and then deliberately misbehaves — garbage and
+//! truncated handshakes, malformed frames, oversized payloads, absurd
+//! cursors, mid-frame drops — against the same accept/handshake/session path
+//! the TCP carrier serves. Every scenario ends by asserting the server's
+//! response class and that a fresh connection still completes a handshake:
+//! the carrier must die never, the session must die cleanly.
+//!
+//! Transport-level kills (mid-link, mid-spin, mid-resource-transfer) belong
+//! to the RNS carrier and were live-fired against the deployed build during
+//! the stress campaign; they are recorded in MAIL.md, not replayed here.
+
+#![cfg(test)]
+
+use std::io::{Read, Write};
+use std::net::TcpStream;
+
+use ed25519_dalek::{Signer, SigningKey};
+use snow::{Builder, TransportState};
+
+use crate::crypto::generate_static_key;
+use crate::proto::{
+    AUTH_REQUIRED, ENVELOPE_MAGIC, ENVELOPE_VERSION, ID_LEN, LABEL_AUTH, LABEL_REGISTER,
+    MALFORMED, MAX_FRAME, NOISE_PARAMS, NOISE_PAYLOAD, OP_AUTH, OP_FETCH, OP_REGISTER, OP_RESOLVE,
+    OP_SEND, PROLOGUE, TOO_LARGE,
+};
+use crate::ratelimit::RateLimiter;
+use crate::server::handle_connection;
+use crate::session::ServerConfig;
+use crate::store::Store;
+
+/// One harness target: a real accept loop serving `handle_connection` on an
+/// ephemeral port, one thread per connection, exactly like `server::run`.
+struct Target {
+    port: u16,
+    static_public: [u8; 32],
+}
+
+impl Target {
+    fn start(max_envelope: usize) -> Self {
+        let (static_private, static_public) = generate_static_key();
+        let config: &'static ServerConfig = Box::leak(Box::new(ServerConfig {
+            max_envelope,
+            fetch_budget: 512 << 10,
+            main_quota: 8 << 20,
+            requests_quota: 2 << 20,
+            max_tokens: 16,
+            invite_token: None,
+            // Unlimited: the harness is hostile by content, not by volume.
+            conn_limiter: RateLimiter::new(0),
+            send_limiter: RateLimiter::new(0),
+            token_limiter: RateLimiter::new(0),
+        }));
+        static TARGETS: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
+        let seq = TARGETS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        let db_path = std::env::temp_dir()
+            .join(format!("bunshin-harness-{}-{seq}.db", std::process::id()))
+            .to_string_lossy()
+            .into_owned();
+        let _ = std::fs::remove_file(&db_path);
+        Store::open(&db_path).unwrap();
+
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let port = listener.local_addr().unwrap().port();
+        std::thread::spawn(move || {
+            for stream in listener.incoming().flatten() {
+                let config = config;
+                let db_path = db_path.clone();
+                std::thread::spawn(move || {
+                    // The harness deliberately produces hostile connections;
+                    // a panic in one must not take the accept loop with it.
+                    let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                        handle_connection(
+                            stream,
+                            config,
+                            &static_private,
+                            &db_path,
+                            "127.0.0.1",
+                        )
+                    }));
+                });
+            }
+        });
+        Target {
+            port,
+            static_public,
+        }
+    }
+
+    /// Proof of life: a fresh connection completes the handshake and gets a
+    /// protocol answer (any status — UNKNOWN_USER counts; dead is no answer).
+    fn assert_alive(&self) {
+        let mut client = Client::connect(self.port);
+        client.handshake().expect("server survived");
+        let (status, _) = client.request(OP_RESOLVE, &[7, b'h', b'a', b'r', b'n', b'e', b's', b's']);
+        assert_ne!(status, 0xFF, "server answered with a status byte");
+    }
+}
+
+/// The hostile client: a real Noise_NX initiator with raw-stream access so
+/// scenarios can write garbage, truncated, or oversized wire data at will.
+struct Client {
+    stream: TcpStream,
+    transport: Option<TransportState>,
+    handshake_hash: Vec<u8>,
+}
+
+impl Client {
+    fn connect(port: u16) -> Self {
+        Client {
+            stream: TcpStream::connect(("127.0.0.1", port)).unwrap(),
+            transport: None,
+            handshake_hash: Vec::new(),
+        }
+    }
+
+    /// The genuine client side of the server's `channel::handshake`.
+    fn handshake(&mut self) -> anyhow::Result<()> {
+        let params: snow::params::NoiseParams = NOISE_PARAMS.parse()?;
+        let mut noise = Builder::new(params).prologue(PROLOGUE).build_initiator()?;
+        let mut buf = [0u8; 65535];
+
+        let n = noise.write_message(&[], &mut buf)?;
+        self.stream.write_all(&(n as u16).to_be_bytes())?;
+        self.stream.write_all(&buf[..n])?;
+
+        let mut len_buf = [0u8; 2];
+        self.stream.read_exact(&mut len_buf)?;
+        let len = u16::from_be_bytes(len_buf) as usize;
+        let mut msg = vec![0u8; len];
+        self.stream.read_exact(&mut msg)?;
+        let mut plaintext = [0u8; 65535];
+        noise.read_message(&msg, &mut plaintext)?;
+
+        anyhow::ensure!(noise.is_handshake_finished(), "handshake did not complete");
+        self.handshake_hash = noise.get_handshake_hash().to_vec();
+        self.transport = Some(noise.into_transport_mode()?);
+        Ok(())
+    }
+
+    fn write_noise(&mut self, payload: &[u8]) {
+        let transport = self.transport.as_mut().expect("handshake first");
+        let mut packet = vec![0u8; payload.len() + 16];
+        let n = transport.write_message(payload, &mut packet).unwrap();
+        packet.truncate(n);
+        self.stream.write_all(&(packet.len() as u16).to_be_bytes()).unwrap();
+        self.stream.write_all(&packet).unwrap();
+    }
+
+    fn read_noise(&mut self) -> Vec<u8> {
+        let transport = self.transport.as_mut().expect("handshake first");
+        let mut len_buf = [0u8; 2];
+        self.stream.read_exact(&mut len_buf).unwrap();
+        let len = u16::from_be_bytes(len_buf) as usize;
+        let mut ciphertext = vec![0u8; len];
+        self.stream.read_exact(&mut ciphertext).unwrap();
+        let mut plaintext = vec![0u8; len];
+        let n = transport.read_message(&ciphertext, &mut plaintext).unwrap();
+        plaintext.truncate(n);
+        plaintext
+    }
+
+    /// A well-formed frame — the baseline the hostile scenarios deviate from.
+    fn request(&mut self, op: u8, body: &[u8]) -> (u8, Vec<u8>) {
+        let mut frame = Vec::with_capacity(5 + body.len());
+        frame.extend_from_slice(&((1 + body.len()) as u32).to_be_bytes());
+        frame.push(op);
+        frame.extend_from_slice(body);
+        for chunk in frame.chunks(NOISE_PAYLOAD) {
+            self.write_noise(chunk);
+        }
+        self.read_frame()
+    }
+
+    fn read_frame(&mut self) -> (u8, Vec<u8>) {
+        let mut buf = Vec::new();
+        while buf.len() < 5 {
+            buf.extend_from_slice(&self.read_noise());
+        }
+        let length = u32::from_be_bytes(buf[..4].try_into().unwrap()) as usize;
+        while buf.len() < 4 + length {
+            buf.extend_from_slice(&self.read_noise());
+        }
+        let frame: Vec<u8> = buf[4..4 + length].to_vec();
+        (frame[0], frame[1..].to_vec())
+    }
+
+    /// Raw bytes before the handshake — garbage on the wire.
+    fn write_raw(&mut self, bytes: &[u8]) {
+        self.stream.write_all(bytes).unwrap();
+    }
+
+    /// Claims a Noise message of `claimed` bytes but sends only `sent`, then
+    /// keeps the connection open briefly so the server sees a short read.
+    fn truncated_noise_len(&mut self, claimed: usize, sent: &[u8]) {
+        self.stream
+            .write_all(&(claimed as u16).to_be_bytes())
+            .unwrap();
+        self.stream.write_all(sent).unwrap();
+    }
+
+    /// One raw post-handshake Noise message carrying arbitrary plaintext.
+    fn raw_frame_bytes(&mut self, bytes: &[u8]) {
+        self.write_noise(bytes);
+    }
+}
+
+/// SERVER-QUIET (the inverse): the server must never answer a connection
+/// that sends nothing, and dropping it must cost nothing.
+#[test]
+fn silent_connection_costs_nothing_and_server_survives() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    std::thread::sleep(std::time::Duration::from_millis(50));
+    drop(client);
+    target.assert_alive();
+}
+
+#[test]
+fn short_garbage_handshake_is_refused_and_server_survives() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    // Below the 32-byte ephemeral an NX first message needs, so snow must
+    // reject it outright and the server closes without writing anything.
+    client.write_raw(&(10u16).to_be_bytes());
+    client.write_raw(&[0xA5; 10]);
+    let mut sink = [0u8; 16];
+    assert!(client.stream.read(&mut sink).unwrap_or(0) == 0, "server closed");
+    target.assert_alive();
+}
+
+/// Noise_NX authenticates the server in message two; the initiator's first
+/// message is unauthenticated by design, so 32+ bytes of random data can
+/// parse as an ephemeral key and draw a handshake reply. That reply must be
+/// the most a garbage sender ever gets: no session forms from it.
+#[test]
+fn long_garbage_first_message_never_forms_a_session() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.write_raw(&(100u16).to_be_bytes());
+    client.write_raw(&[0xA5; 100]);
+    // Reply or close, either is protocol-legal; drop without completing.
+    drop(client);
+    target.assert_alive();
+}
+
+#[test]
+fn truncated_handshake_is_survived() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.truncated_noise_len(1000, &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
+    drop(client);
+    target.assert_alive();
+}
+
+#[test]
+fn unknown_op_gets_malformed_and_server_survives() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    let (op, body) = client.request(0x99, b"");
+    assert_eq!(op, 0x99);
+    assert_eq!(body[0], MALFORMED);
+    target.assert_alive();
+}
+
+#[test]
+fn fetch_without_auth_gets_auth_required() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    let (op, body) = client.request(OP_FETCH, &[0u8; 40]);
+    assert_eq!(op, OP_FETCH);
+    assert_eq!(body[0], AUTH_REQUIRED);
+    target.assert_alive();
+}
+
+#[test]
+fn garbage_resolve_body_gets_malformed() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    let (op, body) = client.request(OP_RESOLVE, &[0xEE; 50]);
+    assert_eq!(op, OP_RESOLVE);
+    assert_eq!(body[0], MALFORMED);
+    target.assert_alive();
+}
+
+#[test]
+fn zero_length_frame_gets_malformed_then_close() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    // Four length bytes alone never reach the length check — read_frame
+    // buffers five bytes first — so the hostile frame is length + op byte.
+    client.raw_frame_bytes(&[0, 0, 0, 0, 0x00]);
+    let (op, body) = client.read_frame();
+    assert_eq!(op, 0);
+    assert_eq!(body[0], MALFORMED);
+    let mut sink = [0u8; 16];
+    assert!(client.stream.read(&mut sink).unwrap_or(0) == 0, "closed after bad frame");
+    target.assert_alive();
+}
+
+#[test]
+fn over_max_frame_length_gets_malformed_then_close() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    let claimed = (MAX_FRAME + 1) as u32;
+    let mut frame = claimed.to_be_bytes().to_vec();
+    frame.push(0x00);
+    client.raw_frame_bytes(&frame);
+    let (op, body) = client.read_frame();
+    assert_eq!(op, 0);
+    assert_eq!(body[0], MALFORMED);
+    target.assert_alive();
+}
+
+#[test]
+fn oversized_envelope_gets_too_large() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    // Well-formed frame, well-formed envelope header, size past the cap.
+    let mut envelope = Vec::with_capacity(800 << 10);
+    envelope.extend_from_slice(ENVELOPE_MAGIC);
+    envelope.push(ENVELOPE_VERSION);
+    envelope.extend_from_slice(&[0u8; ID_LEN]); // recipient never looked up
+    envelope.resize(800 << 10, 0x41);
+    let mut body = vec![0u8]; // no accept-token MAC
+    body.extend_from_slice(&envelope);
+    let (op, resp) = client.request(OP_SEND, &body);
+    let body = &resp[..];
+    assert_eq!(op, OP_SEND);
+    assert_eq!(body[0], TOO_LARGE);
+    target.assert_alive();
+}
+
+#[test]
+fn bad_envelope_magic_gets_malformed() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    let mut envelope = vec![0x00; 128];
+    envelope[4] = ENVELOPE_VERSION;
+    let (op, body) = client.request(OP_SEND, &envelope);
+    assert_eq!(op, OP_SEND);
+    assert_eq!(body[0], MALFORMED);
+    target.assert_alive();
+}
+
+#[test]
+fn mid_frame_drop_is_survived() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    // Claim a frame, deliver a third of its header bytes, vanish.
+    client.raw_frame_bytes(&[0x00, 0x10, 0x00]);
+    drop(client);
+    target.assert_alive();
+}
+
+/// The well-behaved baseline plus the out-of-order cursor: a FETCH whose
+/// cursor points past the end of time must answer an empty page, not an
+/// error or a wrap-around into someone else's mail.
+#[test]
+fn registered_flow_survives_absurd_future_cursor() {
+    let target = Target::start(768 << 10);
+    let key = SigningKey::from_bytes(&[7u8; 32]);
+    let verifying = key.verifying_key();
+    let identity: &[u8] = verifying.as_bytes();
+
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+
+    // REGISTER with a real proof of possession over the server's static key.
+    let mut pop = LABEL_REGISTER.to_vec();
+    pop.extend_from_slice(&target.static_public);
+    pop.extend_from_slice(b"harness");
+    pop.extend_from_slice(identity);
+    let mut body = vec![7];
+    body.extend_from_slice(b"harness");
+    body.extend_from_slice(identity);
+    body.extend_from_slice(&key.sign(&pop).to_bytes());
+    body.push(0); // no invite token
+    body.push(0); // plain registration
+    let (op, resp) = client.request(OP_REGISTER, &body);
+    assert_eq!((op, resp.as_slice()), (OP_REGISTER, &[0u8][..]));
+
+    // AUTH over the real handshake hash this session actually negotiated.
+    let mut auth_msg = LABEL_AUTH.to_vec();
+    auth_msg.extend_from_slice(&client.handshake_hash);
+    let mut body = vec![7];
+    body.extend_from_slice(b"harness");
+    body.extend_from_slice(identity);
+    body.extend_from_slice(&key.sign(&auth_msg).to_bytes());
+    body.push(0); // sync = 0
+    body.extend_from_slice(&0u16.to_be_bytes());
+    let (op, resp) = client.request(OP_AUTH, &body);
+    assert_eq!(op, OP_AUTH);
+    assert_eq!(resp[0], 0, "status OK");
+    assert_eq!(
+        u16::from_be_bytes(resp[1..3].try_into().unwrap()),
+        0,
+        "zero accepted tokens"
+    );
+
+    // FETCH with a cursor at the end of time: OK, zero messages.
+    let mut body = Vec::new();
+    body.extend_from_slice(&i64::MAX.to_be_bytes());
+    body.extend_from_slice(&[0xFF; ID_LEN]);
+    let (op, resp) = client.request(OP_FETCH, &body);
+    assert_eq!(op, OP_FETCH);
+    assert_eq!(resp[0], 0, "status OK");
+    assert_eq!(u16::from_be_bytes(resp[1..3].try_into().unwrap()), 0, "empty page");
+
+    // And the registered name resolves — the store came through intact.
+    let (op, resp) = client.request(OP_RESOLVE, &[7, b'h', b'a', b'r', b'n', b'e', b's', b's']);
+    assert_eq!(op, OP_RESOLVE);
+    assert_eq!(resp[0], 0);
+    target.assert_alive();
+}
+
+
+// ---------------------------------------------------------------------------
+// Long-lived multi-session driver: the last agreed stress-campaign phase.
+// A dozen authenticated sessions live on one server simultaneously, run
+// concurrent FETCH cursors against one mailbox, and — in the slow scenario —
+// are held past the session idle timeout to verify the server reaps idle
+// sessions cleanly instead of leaking their threads.
+
+impl Client {
+    /// AUTH as an already-registered identity (the driver registers once).
+    fn auth(&mut self, username: &str, key: &SigningKey) {
+        let verifying = key.verifying_key();
+        let identity: &[u8] = verifying.as_bytes();
+        let mut auth_msg = LABEL_AUTH.to_vec();
+        auth_msg.extend_from_slice(&self.handshake_hash);
+        let mut body = vec![username.len() as u8];
+        body.extend_from_slice(username.as_bytes());
+        body.extend_from_slice(identity);
+        body.extend_from_slice(&key.sign(&auth_msg).to_bytes());
+        body.push(0);
+        body.extend_from_slice(&0u16.to_be_bytes());
+        let (op, resp) = self.request(OP_AUTH, &body);
+        assert_eq!(op, OP_AUTH);
+        assert_eq!(resp[0], 0, "AUTH ok");
+    }
+
+    /// Pages FETCH from `cursor` to the end, returning every message id.
+    fn fetch_all(&mut self, cursor: (i64, [u8; 32])) -> Vec<[u8; 32]> {
+        let (mut after_at, mut after_id) = cursor;
+        let mut ids = Vec::new();
+        loop {
+            let mut body = Vec::new();
+            body.extend_from_slice(&after_at.to_be_bytes());
+            body.extend_from_slice(&after_id);
+            let (op, resp) = self.request(OP_FETCH, &body);
+            assert_eq!(op, OP_FETCH);
+            assert_eq!(resp[0], 0, "FETCH ok");
+            let count = u16::from_be_bytes(resp[1..3].try_into().unwrap());
+            if count == 0 {
+                return ids;
+            }
+            let mut off = 3;
+            for _ in 0..count {
+                let id: [u8; 32] = resp[off..off + 32].try_into().unwrap();
+                off += 32;
+                let received_at = i64::from_be_bytes(resp[off..off + 8].try_into().unwrap());
+                off += 8;
+                off += 1; // unsolicited flag
+                let len = u32::from_be_bytes(resp[off..off + 4].try_into().unwrap()) as usize;
+                off += 4 + len;
+                ids.push(id);
+                // The cursor is (received_at, id) strictly after the last
+                // delivered record; a future-dated probe terminates cleanly.
+                after_at = received_at;
+                after_id = id;
+            }
+        }
+    }
+}
+
+/// The driver's mailbox: an envelope addressed to a registered identity.
+fn envelope_to(recipient: &[u8], payload_len: usize) -> Vec<u8> {
+    let mut envelope = Vec::with_capacity(69 + payload_len + 16);
+    envelope.extend_from_slice(ENVELOPE_MAGIC);
+    envelope.push(ENVELOPE_VERSION);
+    envelope.extend_from_slice(recipient);
+    envelope.resize(69 + payload_len, 0x42); // epk + ciphertext + tag
+    envelope
+}
+
+#[test]
+fn dozen_sessions_run_concurrent_fetch_cursors() {
+    let target = Target::start(768 << 10);
+    let key = SigningKey::from_bytes(&[8u8; 32]);
+    let verifying = key.verifying_key();
+    let identity: &[u8] = verifying.as_bytes();
+
+    // Register the driver identity from a first session.
+    let mut reg = Client::connect(target.port);
+    reg.handshake().unwrap();
+    let mut pop = LABEL_REGISTER.to_vec();
+    pop.extend_from_slice(&target.static_public);
+    pop.extend_from_slice(b"driver");
+    pop.extend_from_slice(identity);
+    let mut body = vec![6];
+    body.extend_from_slice(b"driver");
+    body.extend_from_slice(identity);
+    body.extend_from_slice(&key.sign(&pop).to_bytes());
+    body.push(0);
+    body.push(0);
+    let (op, resp) = reg.request(OP_REGISTER, &body);
+    assert_eq!((op, resp.as_slice()), (OP_REGISTER, &[0u8][..]));
+
+    // Seed 20 envelopes for the driver mailbox; ids are deterministic.
+    let mut expected = std::collections::HashSet::new();
+    for i in 0..20u8 {
+        let envelope = envelope_to(identity, 40 + i as usize);
+        let mut body = vec![0u8]; // no accept-token MAC
+        body.extend_from_slice(&envelope);
+        let (op, resp) = reg.request(OP_SEND, &body);
+        assert_eq!((op, resp[0]), (OP_SEND, 0), "seed send {i}");
+        let id: [u8; 32] = resp[1..33].try_into().unwrap();
+        expected.insert(id);
+    }
+
+    // A dozen sessions, each authenticated and fetching the whole mailbox
+    // concurrently; all must see the same 20 ids (nothing is deleted, so
+    // every session's cursor walk is independent and complete).
+    let mut handles = Vec::new();
+    for _ in 0..12 {
+        let port = target.port;
+        let session_key = key.clone();
+        handles.push(std::thread::spawn(move || {
+            let mut client = Client::connect(port);
+            client.handshake().unwrap();
+            client.auth("driver", &session_key);
+            client.fetch_all((0, [0u8; 32]))
+        }));
+    }
+    for handle in handles {
+        let ids = handle.join().unwrap();
+        let got: std::collections::HashSet<[u8; 32]> = ids.into_iter().collect();
+        assert_eq!(got, expected, "every session saw the whole mailbox");
+    }
+    target.assert_alive();
+}
+
+/// Held-open sessions must be closed by the server's idle timeout, not
+/// leaked. Slow by nature: it waits out the full IDLE_TIMEOUT_SECS.
+#[test]
+#[ignore = "holds a dozen sessions past the 120 s idle timeout"]
+fn idle_sessions_are_reaped_by_the_server() {
+    let target = Target::start(768 << 10);
+    let key = SigningKey::from_bytes(&[9u8; 32]);
+    let verifying = key.verifying_key();
+    let identity: &[u8] = verifying.as_bytes();
+
+    let mut reg = Client::connect(target.port);
+    reg.handshake().unwrap();
+    let mut pop = LABEL_REGISTER.to_vec();
+    pop.extend_from_slice(&target.static_public);
+    pop.extend_from_slice(b"idler");
+    pop.extend_from_slice(identity);
+    let mut body = vec![5];
+    body.extend_from_slice(b"idler");
+    body.extend_from_slice(identity);
+    body.extend_from_slice(&key.sign(&pop).to_bytes());
+    body.push(0);
+    body.push(0);
+    let (op, resp) = reg.request(OP_REGISTER, &body);
+    assert_eq!((op, resp.as_slice()), (OP_REGISTER, &[0u8][..]));
+
+    // A dozen sessions authenticate, then say nothing at all.
+    let mut clients = Vec::new();
+    for _ in 0..12 {
+        let mut client = Client::connect(target.port);
+        client.handshake().unwrap();
+        client.auth("idler", &key);
+        clients.push(client);
+    }
+
+    // Past the idle timeout the server closes them; the client sees EOF.
+    std::thread::sleep(std::time::Duration::from_secs(125));
+    for client in &mut clients {
+        let mut sink = [0u8; 16];
+        assert!(
+            client.stream.read(&mut sink).unwrap_or(0) == 0,
+            "server closed the idle session"
+        );
+    }
+    target.assert_alive();
+}
+
+/// A frame started but not finished must not hold a thread for the full
+/// session idle timeout: the header deadline (30 s) closes it first.
+#[test]
+#[ignore = "waits out the 30 s header deadline"]
+fn partial_frame_is_dropped_after_the_header_deadline() {
+    let target = Target::start(768 << 10);
+    let mut client = Client::connect(target.port);
+    client.handshake().unwrap();
+    // Three bytes of a frame header, then silence. Before this hardening
+    // the connection idled the full 120 s; the header deadline must close
+    // it well before that.
+    client.raw_frame_bytes(&[0x00, 0x10, 0x00]);
+    let mut sink = [0u8; 16];
+    assert!(
+        client.stream.read(&mut sink).unwrap_or(0) == 0,
+        "server dropped the stalled frame"
+    );
+    target.assert_alive();
+}

src/main.rs

@@ -3,6 +3,8 @@
 mod bind;
 mod channel;
 mod crypto;
+#[cfg(test)]
+mod harness;
 mod proto;
 mod ratelimit;
 #[cfg(feature = "rns")]
@@ -101,6 +103,9 @@ struct RnsServeArgs {
     rns_rate_link_requests: u32,
     #[arg(long = "rns-rate-link-bytes", default_value_t = 1 << 20)]
     rns_rate_link_bytes: u64,
+    /// Reap RNS links idle for this many seconds (0 disables)
+    #[arg(long = "rns-link-idle", default_value_t = 300)]
+    rns_link_idle: u64,
     /// UDP interface to listen on, host[:port]
     #[arg(long = "rns-udp", default_value = "127.0.0.1:4242")]
     rns_udp: String,
@@ -200,6 +205,7 @@ fn main() -> anyhow::Result<()> {
                     max_links: rns.rns_max_links,
                     rate_link_requests: rns.rns_rate_link_requests,
                     rate_link_bytes: rns.rns_rate_link_bytes,
+                    link_idle_secs: rns.rns_link_idle,
                     udp_listen_host: listen_host,
                     udp_listen_port: listen_port,
                     udp_forward_host: forward_host,

src/proto.rs

@@ -51,6 +51,11 @@ pub const FETCH_BUDGET: usize = 512 * 1024; // must stay under MAX_FRAME
 #[cfg(feature = "rns")]
 pub const AUTH_FULL_TOKENS: usize = 164;
 pub const IDLE_TIMEOUT_SECS: u64 = 120;
+/// Budget for completing a handshake or a frame already started. Without it
+/// a connection that sends a partial frame header and goes silent holds a
+/// thread for the full idle timeout — bounded by the connection limiter,
+/// but a free hold-and-wait for the hostile kind of client.
+pub const HEADER_TIMEOUT_SECS: u64 = 30;
 pub const PURGE_INTERVAL_SECS: u64 = 60;
 
 /// A peer sent something unparseable. Always answered with MALFORMED.

src/rns.rs

@@ -12,6 +12,7 @@
 use std::collections::{HashMap, HashSet};
 use std::ffi::CString;
 use std::sync::{Mutex, OnceLock};
+use std::time::{Duration, Instant};
 
 use crate::bind::TransportBindValues;
 use crate::proto::{AUTH_FULL_TOKENS, INTERNAL_ERROR, MALFORMED, RATE_LIMITED, TOO_LARGE};
@@ -59,6 +60,7 @@ pub struct RnsArgs {
     pub udp_listen_port: u16,
     pub udp_forward_host: Option<String>,
     pub udp_forward_port: u16,
+    pub link_idle_secs: u64,
     pub max_tokens: u16,
     pub main_quota: i64,
     pub requests_quota: i64,
@@ -71,6 +73,10 @@ struct RnsState {
     destination: [u8; 16],
     sessions: Mutex<HashMap<[u8; 16], Session<'static>>>,
     links: Mutex<HashSet<[u8; 16]>>,
+    /// Last-seen instant per link, the reaper's clock. Updated before the
+    /// limiters so a client still sending — even one being refused — counts
+    /// as alive; only a silent link is reapable.
+    last_activity: Mutex<HashMap<[u8; 16], Instant>>,
     request_limiter: RateLimiter,
     byte_limiter: ByteRateLimiter,
     max_links: usize,
@@ -117,6 +123,7 @@ pub fn start(args: RnsArgs) -> anyhow::Result<String> {
         destination,
         sessions: Mutex::new(HashMap::new()),
         links: Mutex::new(HashSet::new()),
+        last_activity: Mutex::new(HashMap::new()),
         request_limiter: RateLimiter::new(args.rate_link_requests),
         byte_limiter: ByteRateLimiter::new(args.rate_link_bytes),
         max_links: args.max_links,
@@ -124,6 +131,13 @@ pub fn start(args: RnsArgs) -> anyhow::Result<String> {
     };
     STATE.get_or_init(|| state);
 
+    // microReticulum never times a link out and the shim has no server-side
+    // close, so a client that dies mid-link would hold its Store handle and
+    // maxLinks slot forever; 0 keeps the old never-expire behavior.
+    if args.link_idle_secs > 0 {
+        std::thread::spawn(move || reap_loop(Duration::from_secs(args.link_idle_secs)));
+    }
+
     let storage_dir = CString::new(args.storage_dir).unwrap();
     let listen_host = CString::new(args.udp_listen_host).unwrap();
     let forward_host = args.udp_forward_host.map(|h| CString::new(h).unwrap());
@@ -185,6 +199,9 @@ fn handle_request(request: &[u8], link_id: &[u8; 16]) -> Vec<u8> {
     // an AUTH carrying a full token set, whichever is larger.
     let max_request = (state.config.max_envelope + 34)
         .max(AUTH_FULL_TOKENS + 32 * state.config.max_tokens as usize);
+    // Any request, even one about to be refused, proves the link is alive.
+    state.last_activity.lock().unwrap().insert(*link_id, Instant::now());
+
     if request.len() > max_request {
         log::warn!(
             "request of {} bytes over {} byte cap",
@@ -194,6 +211,9 @@ fn handle_request(request: &[u8], link_id: &[u8; 16]) -> Vec<u8> {
         return response(TOO_LARGE);
     }
     if !state.request_limiter.allow(&link) || !state.byte_limiter.allow(&link, request.len()) {
+        // Pre-dispatch refusals never reach the session metrics, so they get
+        // their own line or the journal undercounts a spinning client.
+        log::debug!("link {link} rate limited: request refused");
         return response(RATE_LIMITED);
     }
 
@@ -264,6 +284,45 @@ extern "C" fn smolmail_rns_take_response(out: *mut u8, cap: usize) -> usize {
     }
 }
 
+/// Links whose last activity is at least `idle` old, with their age.
+fn expired_links(
+    last_activity: &HashMap<[u8; 16], Instant>,
+    now: Instant,
+    idle: Duration,
+) -> Vec<([u8; 16], Duration)> {
+    last_activity
+        .iter()
+        .filter(|(_, seen)| now.duration_since(**seen) >= idle)
+        .map(|(id, seen)| (*id, now.duration_since(*seen)))
+        .collect()
+}
+
+/// Evicts links that have gone silent: microReticulum never times a link
+/// out and the shim has no server-side close, so a client that dies
+/// mid-link would otherwise hold its Store handle and maxLinks slot
+/// forever. A reaped link that somehow speaks again is treated as a fresh
+/// session (unauthenticated until the next AUTH), which is safe.
+fn reap_loop(idle: Duration) {
+    let interval = (idle / 4).clamp(Duration::from_secs(1), Duration::from_secs(60));
+    loop {
+        std::thread::sleep(interval);
+        let Some(state) = STATE.get() else {
+            return;
+        };
+        let expired = expired_links(&state.last_activity.lock().unwrap(), Instant::now(), idle);
+        for (link, age) in expired {
+            state.links.lock().unwrap().remove(&link);
+            state.sessions.lock().unwrap().remove(&link);
+            state.last_activity.lock().unwrap().remove(&link);
+            log::debug!(
+                "link {} reaped after {:?} idle",
+                data_encoding::HEXLOWER.encode(&link),
+                age
+            );
+        }
+    }
+}
+
 #[no_mangle]
 extern "C" fn smolmail_rns_on_link_opened(link_id: *const u8) -> i32 {
     let Some(state) = STATE.get() else {
@@ -280,6 +339,10 @@ extern "C" fn smolmail_rns_on_link_opened(link_id: *const u8) -> i32 {
         return 1;
     }
     links.insert(link);
+    if let Some(state) = STATE.get() {
+        state.last_activity.lock().unwrap().insert(link, Instant::now());
+    }
+    log::debug!("link {} opened ({} open)", data_encoding::HEXLOWER.encode(&link), links.len());
     0
 }
 
@@ -289,6 +352,8 @@ extern "C" fn smolmail_rns_on_link_closed(link_id: *const u8) {
         let link: [u8; 16] = unsafe { std::slice::from_raw_parts(link_id, 16).try_into().unwrap() };
         state.links.lock().unwrap().remove(&link);
         state.sessions.lock().unwrap().remove(&link);
+        state.last_activity.lock().unwrap().remove(&link);
+        log::debug!("link {} closed", data_encoding::HEXLOWER.encode(&link));
     }
 }
 
@@ -309,4 +374,18 @@ mod tests {
             "799855f4955f1b09fd20a13cd84f4e71"
         );
     }
+
+    #[test]
+    fn expired_links_selects_only_silent_ones() {
+        let mut last_activity = HashMap::new();
+        let now = Instant::now();
+        let idle = Duration::from_secs(300);
+        last_activity.insert([1u8; 16], now - Duration::from_secs(301)); // silent
+        last_activity.insert([2u8; 16], now - Duration::from_secs(299)); // active
+        last_activity.insert([3u8; 16], now); // just seen
+        let expired = expired_links(&last_activity, now, idle);
+        assert_eq!(expired.len(), 1);
+        assert_eq!(expired[0].0, [1u8; 16]);
+        assert!(expired[0].1 >= idle);
+    }
 }

src/server.rs

@@ -7,7 +7,7 @@ use std::time::Duration;
 use crate::bind::TransportBindValues;
 use crate::channel::{handshake, Channel};
 use crate::crypto::{b32, derive_public};
-use crate::proto::{FETCH_BUDGET, IDLE_TIMEOUT_SECS, KEY_LEN, MALFORMED, PURGE_INTERVAL_SECS};
+use crate::proto::{FETCH_BUDGET, HEADER_TIMEOUT_SECS, KEY_LEN, MALFORMED, PURGE_INTERVAL_SECS};
 use crate::ratelimit::RateLimiter;
 use crate::session::{ServerConfig, Session};
 use crate::store::Store;
@@ -94,15 +94,21 @@ pub fn run(args: ServeArgs) -> anyhow::Result<()> {
     Ok(())
 }
 
-fn handle_connection(
+/// Also the hostile harness's entry point: it drives real connections
+/// through the same accept/handshake/session path the TCP carrier serves.
+pub(crate) fn handle_connection(
     mut stream: TcpStream,
     config: &ServerConfig,
     static_key: &[u8; KEY_LEN],
     db_path: &str,
     peer_ip: &str,
 ) -> anyhow::Result<()> {
-    stream.set_read_timeout(Some(Duration::from_secs(IDLE_TIMEOUT_SECS)))?;
+    // The handshake runs under the header budget; once it completes,
+    // Channel::read_noise picks between the header and idle deadlines per
+    // frame state.
+    stream.set_read_timeout(Some(Duration::from_secs(HEADER_TIMEOUT_SECS)))?;
 
+    let handshake_started = std::time::Instant::now();
     let (transport, handshake_hash) = match handshake(&mut stream, static_key) {
         Ok(v) => v,
         Err(e) => {
@@ -110,6 +116,17 @@ fn handle_connection(
             return Ok(());
         }
     };
+    log::debug!(
+        "handshake from {peer_ip} in {:?}",
+        handshake_started.elapsed()
+    );
+
+    // Drop logs the summary once, whatever return path closes the session.
+    let summary = SessionSummary {
+        peer_ip,
+        started: std::time::Instant::now(),
+        ops: std::cell::Cell::new(0),
+    };
 
     let store = Store::open(db_path)?;
     let server_static = derive_public(static_key);
@@ -129,6 +146,7 @@ fn handle_connection(
                 return Ok(());
             }
         };
+        summary.ops.set(summary.ops.get() + 1);
 
         let (status, payload) = session.dispatch(op, &body);
         let mut response = Vec::with_capacity(1 + payload.len());
@@ -140,6 +158,23 @@ fn handle_connection(
     }
 }
 
+struct SessionSummary<'a> {
+    peer_ip: &'a str,
+    started: std::time::Instant,
+    ops: std::cell::Cell<u64>,
+}
+
+impl Drop for SessionSummary<'_> {
+    fn drop(&mut self) {
+        log::debug!(
+            "session {} closed: {} op(s) in {:?}",
+            self.peer_ip,
+            self.ops.get(),
+            self.started.elapsed()
+        );
+    }
+}
+
 fn is_eof_like(e: &anyhow::Error) -> bool {
     if let Some(io_err) = e.downcast_ref::<std::io::Error>() {
         return matches!(

src/session.rs

@@ -44,6 +44,19 @@ impl From<rusqlite::Error> for HandlerError {
 
 type OpResult = Result<(u8, Vec<u8>), HandlerError>;
 
+/// Wire op byte to log name; unknown bytes are logged as dispatched too.
+fn op_name(op: u8) -> &'static str {
+    match op {
+        OP_AUTH => "AUTH",
+        OP_RESOLVE => "RESOLVE",
+        OP_SEND => "SEND",
+        OP_FETCH => "FETCH",
+        OP_DELETE => "DELETE",
+        OP_REGISTER => "REGISTER",
+        _ => "UNKNOWN",
+    }
+}
+
 pub struct Session<'a> {
     config: &'a ServerConfig,
     store: Store,
@@ -72,7 +85,25 @@ impl<'a> Session<'a> {
     /// panicking or propagating errors to the caller: a bad frame or a
     /// storage error both become a response, and the caller decides
     /// separately whether to keep the connection open.
+    ///
+    /// Timing lives in this wrapper because both carriers dispatch through
+    /// it, so the metrics flag covers TCP and RNS from one place.
     pub fn dispatch(&mut self, op: u8, body: &[u8]) -> (u8, Vec<u8>) {
+        let started = std::time::Instant::now();
+        let (status, payload) = self.dispatch_op(op, body);
+        log::debug!(
+            "op {} from {} in {:?}: status {}, {}B request, {}B response",
+            op_name(op),
+            self.peer_ip,
+            started.elapsed(),
+            status,
+            body.len(),
+            payload.len()
+        );
+        (status, payload)
+    }
+
+    fn dispatch_op(&mut self, op: u8, body: &[u8]) -> (u8, Vec<u8>) {
         if matches!(op, OP_FETCH | OP_DELETE) && self.username.is_none() {
             return (AUTH_REQUIRED, Vec::new());
         }