From ddde687f9ca7ea88b31b230339e1ccf91a4d6f25 Mon Sep 17 00:00:00 2001 From: Tebbens4832 Date: Mon, 13 Jul 2026 16:05:27 -0500 Subject: [PATCH] =?UTF-8?q?NA-0642:=20durable=20store-and-forward=20queue?= =?UTF-8?q?=20=E2=80=94=20embedded=20SQLite,=20retention=20TTL,=20delete-o?= =?UTF-8?q?n-ACKNOWLEDGED-pull=20(D578,=20D-0011)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Store: rusqlite(bundled), WAL + synchronous=FULL, schema v1 (routes keyed by SHA-256 of the route token — raw tokens never persisted; messages: seq FIFO, msg_id non-unique per NA-0275, opaque BLOB body, enqueued_at, leased_until). STORE_PATH is required, fail-closed. Contract (D-0011, per the qsl-protocol D578 design-lock, option B): - legacy GET /v1/pull unchanged byte-for-byte (delete-on-deliver) — the current non-acking qsc client is not stranded - GET /v1/pull?ack=lease leases without deleting (PULL_LEASE_SECS, 60s) - POST /v1/pull/ack deletes only leased copies; idempotent; route-scoped - RETENTION_TTL_SECS (7d default) replaces the 5-min idle-route discard (ROUTE_IDLE_TTL_MS now warn-and-ignore); delivered+acked still forgotten Tests: hard-kill (SIGKILL) restart durability + crash-between-pull-and-ack lease survival; retention expiry + non-vacuity controls; ack semantics; legacy-shape guard; concurrency no-loss/no-dup; token-at-rest privacy. Packaging: relay.env.example (new envs + MAX_QUEUE_DEPTH 256->257 fix), StateDirectory=qsl-server, runbook config section. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 152 ++++++ Cargo.toml | 2 + DECISIONS.md | 8 + packaging/runbook_ubuntu.md | 10 +- packaging/systemd/qsl-server.service | 3 + packaging/systemd/relay.env.example | 16 +- src/lib.rs | 392 +++++++++----- src/main.rs | 348 ++++++++----- src/store.rs | 491 ++++++++++++++++++ tests/config_semantics.rs | 90 +++- tests/na0642_ack_contract.rs | 343 ++++++++++++ tests/na0642_backward_compat.rs | 101 ++++ tests/na0642_concurrency.rs | 201 +++++++ tests/na0642_durability_restart.rs | 273 ++++++++++ ...e_ttl.rs => na0642_retention_lifecycle.rs} | 276 +++++----- tests/na0642_retention_logging.rs | 112 ++++ tests/na0642_store_privacy.rs | 81 +++ tests/qsl_attachments_integration_contract.rs | 27 +- tests/route_lifecycle_ttl_logging.rs | 140 ----- 19 files changed, 2532 insertions(+), 534 deletions(-) create mode 100644 src/store.rs create mode 100644 tests/na0642_ack_contract.rs create mode 100644 tests/na0642_backward_compat.rs create mode 100644 tests/na0642_concurrency.rs create mode 100644 tests/na0642_durability_restart.rs rename tests/{route_lifecycle_ttl.rs => na0642_retention_lifecycle.rs} (51%) create mode 100644 tests/na0642_retention_logging.rs create mode 100644 tests/na0642_store_privacy.rs delete mode 100644 tests/route_lifecycle_ttl_logging.rs diff --git a/Cargo.lock b/Cargo.lock index 54d34e5..1915690 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,18 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -145,6 +157,15 @@ version = "2.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bumpalo" version = "3.19.1" @@ -225,6 +246,35 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "displaydoc" version = "0.2.5" @@ -236,6 +286,18 @@ dependencies = [ "syn", ] +[[package]] +name = "fallible-iterator" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + [[package]] name = "find-msvc-tools" version = "0.1.8" @@ -285,6 +347,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -312,6 +384,24 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", +] + +[[package]] +name = "hashlink" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ba4ff7128dee98c7dc9794b6a411377e1404dba1c97deb8d1a55297bd25d8af" +dependencies = [ + "hashbrown", +] + [[package]] name = "heck" version = "0.5.0" @@ -578,6 +668,17 @@ version = "0.2.180" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc" +[[package]] +name = "libsqlite3-sys" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + [[package]] name = "litemap" version = "0.8.1" @@ -673,6 +774,12 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "pkg-config" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" + [[package]] name = "potential_utf" version = "0.1.4" @@ -707,8 +814,10 @@ dependencies = [ "axum", "clap", "reqwest", + "rusqlite", "serde", "serde_json", + "sha2", "tokio", "tower-http 0.5.2", "tracing", @@ -884,6 +993,20 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rusqlite" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7753b721174eb8ff87a9a0e799e2d7bc3749323e773db92e0984debb00019d6e" +dependencies = [ + "bitflags", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink", + "libsqlite3-sys", + "smallvec", +] + [[package]] name = "rustc-hash" version = "2.1.1" @@ -1003,6 +1126,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -1310,6 +1444,12 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.22" @@ -1363,6 +1503,18 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "vcpkg" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "want" version = "0.3.1" diff --git a/Cargo.toml b/Cargo.toml index aa24ef0..3bbe211 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,8 @@ tokio = { version = "1", features = ["rt-multi-thread", "macros"] } serde = { version = "1", features = ["derive"] } serde_json = "1" uuid = { version = "1", features = ["v4"] } +rusqlite = { version = "0.32", features = ["bundled"] } +sha2 = "0.10" tower-http = { version = "0.5", features = ["trace"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/DECISIONS.md b/DECISIONS.md index 8f656ee..7729039 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -70,3 +70,11 @@ - **Decision:** qsl-server now retires legacy `/v1/push/:channel` and `/v1/pull/:channel?max=N` outright. Canonical header-carried routing on token-free `/v1/push` and `/v1/pull?max=N` is the only supported route-token ingress shape. - **Rationale:** The compatibility window from D-0009 existed only to get supported clients and operator guidance onto the safer header-based posture. That migration is now complete enough that continuing to accept URI-carried route tokens leaves a known passive-leak surface live without adding truthful transport value. - **References:** NA-0012; `src/lib.rs`; `tests/relay_smoke.rs`; `README.md`; `docs/server/DOC-SRV-003_Relay_Inbox_Contract_v1.0.0_DRAFT.md`; `packaging/runbook_ubuntu.md`; `scripts/check_relay_compatibility.sh`; `scripts/verify_remote.sh`; `scripts/aws_update_and_verify.sh`; `scripts/ci/test_relay_deploy_compatibility_guard.sh`; `TRACEABILITY.md` + +- **ID:** D-0011 + - **Status:** Accepted + - **Date:** 2026-07-13 + - **Goals:** G4, G5 + - **Decision:** qsl-server's store-and-forward queue becomes DURABLE (embedded SQLite via rusqlite/bundled, WAL + synchronous=FULL, single-file store at the required `STORE_PATH`; route tokens persisted only as SHA-256 digests; payloads stored verbatim as opaque blobs), and the delivery contract adds an ACKNOWLEDGED-PULL mode per the qsl-protocol D578 design-lock (option B, operator-chosen): `GET /v1/pull?ack=lease` returns messages WITHOUT deleting and leases them for `PULL_LEASE_SECS` (default 60 s); `POST /v1/pull/ack {"ids":[...]}` deletes ONLY leased copies (idempotent; scoped to the route; unleased duplicate copies per the NA-0275 contract survive); un-acked leases expire and the messages reappear. The LEGACY pull (`GET /v1/pull?max=N`, no ack parameter) keeps its exact delete-on-deliver contract and response shape — the current non-acking qsc client is not stranded. The 5-minute idle-route discard (`ROUTE_IDLE_TTL_MS`) is RETIRED (warn-and-ignore) and replaced by an operator-tunable retention TTL for undelivered messages (`RETENTION_TTL_SECS`, default 7 days, ceiling 30 days); delivered+acked messages are still forgotten immediately — the relay is reliable, not an archive. Startup is fail-closed: `STORE_PATH` has no default. + - **Rationale:** The in-memory queue was demo-class: a restart dropped every queued message and idle routes discarded after 5 minutes, unacceptable for the DOC-PROG-003 self-host operator-path (Tier-1, step 1). Delete-on-pull loses a message if the puller crashes between pull and local persistence; the lease model closes that window without wire-semantic or E2EE change (payloads stay opaque, the relay stays blind, nothing precludes future E2EE read receipts as ordinary payloads). Route/pull semantics are a recorded-decision surface per D-0009/D-0010, so the contract change is recorded here; governance authority for the lane lives in qsl-protocol (NA-0642, D578, D-1265). + - **References:** qsl-protocol NA-0642 / QSL-DIR-2026-07-13-578 (D578) / D-1265; D-0009; D-0010; `src/store.rs`; `src/lib.rs`; `src/main.rs`; `packaging/systemd/relay.env.example`; `packaging/systemd/qsl-server.service` (StateDirectory); `packaging/runbook_ubuntu.md`; `tests/na0642_durability_restart.rs`; `tests/na0642_ack_contract.rs`; `tests/na0642_retention_lifecycle.rs`; `tests/na0642_retention_logging.rs`; `tests/na0642_backward_compat.rs`; `tests/na0642_concurrency.rs`; `tests/na0642_store_privacy.rs` diff --git a/packaging/runbook_ubuntu.md b/packaging/runbook_ubuntu.md index 76ba6aa..f5dfc32 100644 --- a/packaging/runbook_ubuntu.md +++ b/packaging/runbook_ubuntu.md @@ -34,7 +34,15 @@ Edit `/etc/qsl-server/relay.env`: - `BIND_ADDR=127.0.0.1` (default safe bind) - `PORT=8080` - `RELAY_TOKEN=` (set a strong token out-of-band; do not commit tokens) -- `MAX_BODY_BYTES`, `MAX_QUEUE_DEPTH` as needed +- `STORE_PATH=/var/lib/qsl-server/relay.db` (REQUIRED since NA-0642 — the durable + SQLite queue; the packaged unit provisions `/var/lib/qsl-server` via + `StateDirectory=qsl-server`. This file IS the relay's data backup unit.) +- `RETENTION_TTL_SECS=604800` (undelivered-message lifetime, default 7 days; + replaces the retired `ROUTE_IDLE_TTL_MS` idle discard — a leftover + `ROUTE_IDLE_TTL_MS` line is warned about and ignored) +- `PULL_LEASE_SECS=60` (visibility timeout for acknowledged pulls, `?ack=lease`) +- `MAX_BODY_BYTES`, `MAX_QUEUE_DEPTH` as needed (template default 257 — the + NA-0598 exact-4-MiB attachment needs 256 chunks + 1 manifest) Apply config: diff --git a/packaging/systemd/qsl-server.service b/packaging/systemd/qsl-server.service index bc9e7b0..4a96fba 100644 --- a/packaging/systemd/qsl-server.service +++ b/packaging/systemd/qsl-server.service @@ -16,6 +16,9 @@ PrivateTmp=true ProtectSystem=strict ProtectHome=true ReadWritePaths=/opt/qsl-server /etc/qsl-server +# NA-0642: creates /var/lib/qsl-server (owned by the service user) and keeps +# it writable under ProtectSystem=strict for the durable STORE_PATH. +StateDirectory=qsl-server CapabilityBoundingSet= AmbientCapabilities= RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX diff --git a/packaging/systemd/relay.env.example b/packaging/systemd/relay.env.example index 2cfcf30..fafd1f7 100644 --- a/packaging/systemd/relay.env.example +++ b/packaging/systemd/relay.env.example @@ -2,7 +2,21 @@ PORT=8080 BIND_ADDR=127.0.0.1 MAX_BODY_BYTES=1048576 -MAX_QUEUE_DEPTH=256 +MAX_QUEUE_DEPTH=257 + +# Durable store-and-forward queue (REQUIRED; NA-0642). The SQLite database +# file holding queued messages — back up this file to back up the relay. +# ':memory:' is accepted only for deliberately ephemeral runs. +STORE_PATH=/var/lib/qsl-server/relay.db + +# Undelivered-message retention in seconds (default 604800 = 7 days, +# ceiling 2592000 = 30 days). Replaces the retired ROUTE_IDLE_TTL_MS +# idle-route discard as the message-lifetime control. +RETENTION_TTL_SECS=604800 + +# Ack-mode pull lease (visibility timeout) in seconds for +# GET /v1/pull?ack=lease (default 60, ceiling 3600). +PULL_LEASE_SECS=60 # Optional bearer token gate for /v1/push and /v1/pull. # Leave empty to disable auth. diff --git a/src/lib.rs b/src/lib.rs index 2274628..6666225 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,45 +7,26 @@ use axum::{ Json, Router, }; use serde::Serialize; +use sha2::{Digest, Sha256}; use std::{ - collections::{HashMap, VecDeque}, + collections::HashMap, sync::{Arc, Mutex}, time::{Duration, Instant}, }; use tracing::info; use uuid::Uuid; -// Named aliases to keep queue types readable (also satisfies clippy::type_complexity). -type QueueMsg = (String, Vec); -type Queue = VecDeque; -type Routes = HashMap; +mod store; -#[derive(Debug)] -struct RouteState { - queue: Queue, - push_rate: PushRateBucket, - last_touched: Instant, -} +pub use store::{ + pull_lease_or_error, retention_ttl_or_error, StoreConfig, SweepStats, MAX_ACK_IDS, + MAX_PULL_LEASE_SECS_CEILING, MAX_RETENTION_TTL_SECS_CEILING, PULL_LEASE_SECS_DEFAULT, + RETENTION_TTL_SECS_DEFAULT, +}; -impl RouteState { - fn new(now: Instant, controls: ResourceControls) -> Self { - Self { - queue: Queue::new(), - push_rate: PushRateBucket::new(now, controls), - last_touched: now, - } - } +use store::{now_unix_secs, AckOutcome, EnqueueOutcome, PullMode, PullOutcome, Store}; - fn touch(&mut self, now: Instant) { - self.last_touched = now; - } - - fn is_expired(&self, controls: ResourceControls, now: Instant) -> bool { - now.checked_duration_since(self.last_touched) - .unwrap_or(Duration::ZERO) - >= controls.route_idle_ttl - } -} +type RateBuckets = HashMap; #[derive(Debug)] struct PushRateBucket { @@ -141,6 +122,9 @@ pub struct ResourceControls { pub max_route_count: usize, pub push_rate_burst: usize, pub push_rate_refill_per_sec: usize, + // Vestigial since NA-0642: message lifetime is now governed by the store's + // retention TTL (StoreConfig::retention_ttl_secs); kept so existing + // constructor call sites remain source-compatible. pub route_idle_ttl: Duration, } @@ -200,14 +184,7 @@ impl ResourceControls { let push_rate_burst = env_limit_or_default("PUSH_RATE_BURST", MAX_PUSH_RATE_BURST_CEILING)?; let push_rate_refill_per_sec = env_refill_or_default("PUSH_RATE_REFILL_PER_SEC", MAX_PUSH_RATE_BURST_CEILING)?; - let route_idle_ttl_ms = - env_limit_or_default("ROUTE_IDLE_TTL_MS", ROUTE_IDLE_TTL_MS_DEFAULT)?; - Self::new_with_route_idle_ttl_ms( - max_route_count, - push_rate_burst, - push_rate_refill_per_sec, - route_idle_ttl_ms, - ) + Self::new(max_route_count, push_rate_burst, push_rate_refill_per_sec) } } @@ -243,9 +220,12 @@ fn push_rate_refill_or_cap(value: usize) -> usize { #[derive(Clone)] pub struct AppState { - // route token -> route state containing queued messages and bounded rate accounting - routes: Arc>, - + // Durable store-and-forward queue (SQLite). Payloads are opaque blobs; + // route tokens are stored only as SHA-256 digests. + store: Store, + // Per-route push rate accounting stays in memory: abuse control, not + // correctness state; resets on restart by design. + push_rates: Arc>, limits: Limits, controls: ResourceControls, relay_token: Option, @@ -276,11 +256,81 @@ impl AppState { controls: ResourceControls, relay_token: Option, ) -> Self { - Self { - routes: Arc::new(Mutex::new(HashMap::new())), + Self::new_with_auth_controls_and_store( + limits, + controls, + relay_token, + StoreConfig::default(), + ) + .expect("ERR_STORE_OPEN_IN_MEMORY") + } + + pub fn new_with_controls_and_store( + limits: Limits, + controls: ResourceControls, + store_cfg: StoreConfig, + ) -> Result { + Self::new_with_auth_controls_and_store( + limits, + controls, + std::env::var("RELAY_TOKEN").ok().filter(|v| !v.is_empty()), + store_cfg, + ) + } + + pub fn new_with_auth_controls_and_store( + limits: Limits, + controls: ResourceControls, + relay_token: Option, + store_cfg: StoreConfig, + ) -> Result { + Ok(Self { + store: Store::open(&store_cfg)?, + push_rates: Arc::new(Mutex::new(HashMap::new())), limits, controls, relay_token, + }) + } + + /// Delete undelivered messages older than the retention TTL. Runs lazily + /// on every push/pull as well; this entry point exists for the periodic + /// background sweep so quiet relays still expire. + pub fn run_retention_sweep(&self) -> SweepStats { + let now = now_unix_secs(); + match self.store.retention_sweep(now) { + Ok(stats) => { + self.log_and_prune_sweep(&stats); + stats + } + Err(e) => { + tracing::error!("event=store_error op=retention_sweep detail={e}"); + SweepStats::default() + } + } + } + + fn log_and_prune_sweep(&self, stats: &SweepStats) { + for (log_id, expired) in &stats.expired_routes { + info!( + "event=retention_expired channel_id={} expired_messages={} ttl_secs={}", + log_id, + expired, + self.store.retention_ttl_secs() + ); + } + if !stats.removed_route_keys.is_empty() { + if let Ok(mut buckets) = self.push_rates.lock() { + for key in &stats.removed_route_keys { + buckets.remove(key); + } + } + } + } + + fn drop_rate_bucket(&self, route_key: &str) { + if let Ok(mut buckets) = self.push_rates.lock() { + buckets.remove(route_key); } } } @@ -304,6 +354,17 @@ struct PullResp { #[derive(serde::Deserialize)] struct PullQuery { max: Option, + ack: Option, +} + +#[derive(serde::Deserialize)] +struct AckReq { + ids: Vec, +} + +#[derive(Serialize)] +struct AckResp { + acked: usize, } const ROUTE_TOKEN_HEADER: &str = "x-qsl-route-token"; @@ -312,6 +373,7 @@ pub fn app(state: AppState) -> Router { Router::new() .route("/v1/push", post(push_message)) .route("/v1/pull", get(pull_message)) + .route("/v1/pull/ack", post(ack_messages)) .with_state(state) } @@ -325,6 +387,17 @@ fn channel_log_id(channel: &str) -> String { format!("{state:016x}") } +fn route_key_for(channel: &str) -> String { + // At-rest key: the raw route token is never persisted; lookups hash the + // header value, so a stolen store file yields no usable routing tokens. + let digest = Sha256::digest(channel.as_bytes()); + let mut out = String::with_capacity(64); + for b in digest { + out.push_str(&format!("{b:02x}")); + } + out +} + fn auth_ok(headers: &HeaderMap, relay_token: Option<&str>) -> bool { match relay_token { None => true, @@ -354,23 +427,18 @@ fn resolve_route_token(headers: &HeaderMap) -> Result { } } -fn expire_idle_routes(routes: &mut Routes, controls: ResourceControls, now: Instant) { - let mut expired = Vec::new(); - routes.retain(|channel, route| { - if route.is_expired(controls, now) { - expired.push((channel_log_id(channel), route.queue.len())); - false - } else { - true +async fn run_store(f: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> Result + Send + 'static, +{ + match tokio::task::spawn_blocking(f).await { + Ok(Ok(v)) => Ok(v), + Ok(Err(e)) => { + tracing::error!("event=store_error detail={e}"); + Err("ERR_STORE") } - }); - for (channel_id, queued_messages) in expired { - info!( - "event=route_expired channel_id={} queued_messages={} ttl_ms={}", - channel_id, - queued_messages, - controls.route_idle_ttl.as_millis() - ); + Err(_) => Err("ERR_STORE"), } } @@ -399,61 +467,92 @@ async fn push_message( .filter(|v| !v.trim().is_empty()) .map(|v| v.to_string()) .unwrap_or_else(|| Uuid::new_v4().to_string()); - let Ok(mut g) = st.routes.lock() else { - return (StatusCode::INTERNAL_SERVER_ERROR, "ERR_LOCK_POISON").into_response(); + let route_key = route_key_for(&channel); + let log_id = channel_log_id(&channel); + let now = now_unix_secs(); + + let status = { + let store = st.store.clone(); + let key = route_key.clone(); + match run_store(move || store.route_status(&key, now)).await { + Ok(v) => v, + Err(code) => return (StatusCode::INTERNAL_SERVER_ERROR, code).into_response(), + } }; + st.log_and_prune_sweep(&status.sweep); - let now = Instant::now(); - expire_idle_routes(&mut g, st.controls, now); - if let Some(route) = g.get_mut(&channel) { - if route.queue.len() >= st.limits.max_queue_depth { - info!( - "event=overloaded queue_depth={} max={}", - route.queue.len(), - st.limits.max_queue_depth - ); - return (StatusCode::TOO_MANY_REQUESTS, "ERR_OVERLOADED").into_response(); - } - if !route.push_rate.try_consume(st.controls, now) { + if status.route_exists && status.depth >= st.limits.max_queue_depth { + info!( + "event=overloaded queue_depth={} max={}", + status.depth, st.limits.max_queue_depth + ); + return (StatusCode::TOO_MANY_REQUESTS, "ERR_OVERLOADED").into_response(); + } + if !status.route_exists && status.live_routes >= st.controls.max_route_count { + info!( + "event=route_cap channel_id={} live_routes={} max={}", + log_id, status.live_routes, st.controls.max_route_count + ); + return (StatusCode::TOO_MANY_REQUESTS, "ERR_ROUTE_CAP").into_response(); + } + + { + let mono_now = Instant::now(); + let Ok(mut buckets) = st.push_rates.lock() else { + return (StatusCode::INTERNAL_SERVER_ERROR, "ERR_LOCK_POISON").into_response(); + }; + let bucket = buckets + .entry(route_key.clone()) + .or_insert_with(|| PushRateBucket::new(mono_now, st.controls)); + if !bucket.try_consume(st.controls, mono_now) { info!( "event=rate_limited channel_id={} burst={} refill_per_sec={}", - channel_log_id(&channel), - st.controls.push_rate_burst, - st.controls.push_rate_refill_per_sec + log_id, st.controls.push_rate_burst, st.controls.push_rate_refill_per_sec ); return (StatusCode::TOO_MANY_REQUESTS, "ERR_RATE_LIMITED").into_response(); } - route.queue.push_back((msg_id.clone(), body.to_vec())); - route.touch(now); - } else { - if g.len() >= st.controls.max_route_count { + } + + let outcome = { + let store = st.store.clone(); + let key = route_key.clone(); + let lid = log_id.clone(); + let mid = msg_id.clone(); + let payload = body.to_vec(); + let max_depth = st.limits.max_queue_depth; + let max_routes = st.controls.max_route_count; + match run_store(move || { + store.enqueue(&key, &lid, &mid, &payload, now, max_depth, max_routes) + }) + .await + { + Ok(v) => v, + Err(code) => return (StatusCode::INTERNAL_SERVER_ERROR, code).into_response(), + } + }; + match outcome { + EnqueueOutcome::Accepted => {} + EnqueueOutcome::Overloaded { depth } => { info!( - "event=route_cap channel_id={} live_routes={} max={}", - channel_log_id(&channel), - g.len(), - st.controls.max_route_count + "event=overloaded queue_depth={} max={}", + depth, st.limits.max_queue_depth ); - return (StatusCode::TOO_MANY_REQUESTS, "ERR_ROUTE_CAP").into_response(); + return (StatusCode::TOO_MANY_REQUESTS, "ERR_OVERLOADED").into_response(); } - let mut route = RouteState::new(now, st.controls); - if !route.push_rate.try_consume(st.controls, now) { + EnqueueOutcome::RouteCap { live_routes } => { info!( - "event=rate_limited channel_id={} burst={} refill_per_sec={}", - channel_log_id(&channel), - st.controls.push_rate_burst, - st.controls.push_rate_refill_per_sec + "event=route_cap channel_id={} live_routes={} max={}", + log_id, live_routes, st.controls.max_route_count ); - return (StatusCode::TOO_MANY_REQUESTS, "ERR_RATE_LIMITED").into_response(); + st.drop_rate_bucket(&route_key); + return (StatusCode::TOO_MANY_REQUESTS, "ERR_ROUTE_CAP").into_response(); } - route.queue.push_back((msg_id.clone(), body.to_vec())); - route.touch(now); - g.insert(channel.clone(), route); } // Never log payload; metadata only. info!( "push channel_id={} id={} bytes={}", - channel_log_id(&channel), + log_id, msg_id, body.len() ); @@ -478,41 +577,90 @@ async fn pull_message( return (StatusCode::BAD_REQUEST, "ERR_BAD_MAX").into_response(); } let max = max.min(st.limits.max_queue_depth); - let Ok(mut g) = st.routes.lock() else { - return (StatusCode::INTERNAL_SERVER_ERROR, "ERR_LOCK_POISON").into_response(); - }; - let now = Instant::now(); - expire_idle_routes(&mut g, st.controls, now); - let Some(route) = g.get_mut(&channel) else { - return StatusCode::NO_CONTENT.into_response(); + let mode = match query.ack.as_deref() { + None => PullMode::Legacy, + Some("lease") => PullMode::Lease, + Some(_) => return (StatusCode::BAD_REQUEST, "ERR_BAD_ACK_MODE").into_response(), }; - if route.queue.is_empty() { - g.remove(&channel); - return StatusCode::NO_CONTENT.into_response(); - } - let mut items = Vec::with_capacity(max); - for _ in 0..max { - if let Some((msg_id, data)) = route.queue.pop_front() { - info!( - "pull channel_id={} id={} bytes={}", - channel_log_id(&channel), - msg_id, - data.len() - ); - items.push(PullItem { id: msg_id, data }); - } else { - break; + let route_key = route_key_for(&channel); + let log_id = channel_log_id(&channel); + let now = now_unix_secs(); + + let outcome: PullOutcome = { + let store = st.store.clone(); + let key = route_key.clone(); + match run_store(move || store.pull(&key, max, now, mode)).await { + Ok(v) => v, + Err(code) => return (StatusCode::INTERNAL_SERVER_ERROR, code).into_response(), } + }; + st.log_and_prune_sweep(&outcome.sweep); + if outcome.route_drained { + st.drop_rate_bucket(&route_key); } - if let Some(route) = g.get_mut(&channel) { - route.touch(now); + if outcome.items.is_empty() { + return StatusCode::NO_CONTENT.into_response(); } - if g.get(&channel).is_some_and(|route| route.queue.is_empty()) { - g.remove(&channel); + let mut items = Vec::with_capacity(outcome.items.len()); + for msg in outcome.items { + info!( + "pull channel_id={} id={} bytes={}", + log_id, + msg.msg_id, + msg.body.len() + ); + items.push(PullItem { + id: msg.msg_id, + data: msg.body, + }); } (StatusCode::OK, Json(PullResp { items })).into_response() } +async fn ack_messages( + State(st): State, + headers: HeaderMap, + body: Bytes, +) -> impl IntoResponse { + if !auth_ok(&headers, st.relay_token.as_deref()) { + return (StatusCode::UNAUTHORIZED, "ERR_UNAUTHORIZED").into_response(); + } + let channel = match resolve_route_token(&headers) { + Ok(v) => v, + Err(code) => return (StatusCode::BAD_REQUEST, code).into_response(), + }; + let Ok(req) = serde_json::from_slice::(&body) else { + return (StatusCode::BAD_REQUEST, "ERR_BAD_ACK_BODY").into_response(); + }; + if req.ids.is_empty() || req.ids.len() > MAX_ACK_IDS { + return (StatusCode::BAD_REQUEST, "ERR_BAD_ACK_IDS").into_response(); + } + let route_key = route_key_for(&channel); + let log_id = channel_log_id(&channel); + let now = now_unix_secs(); + + let outcome: AckOutcome = { + let store = st.store.clone(); + let key = route_key.clone(); + let ids = req.ids; + match run_store(move || store.ack(&key, &ids, now)).await { + Ok(v) => v, + Err(code) => return (StatusCode::INTERNAL_SERVER_ERROR, code).into_response(), + } + }; + st.log_and_prune_sweep(&outcome.sweep); + if outcome.route_drained { + st.drop_rate_bucket(&route_key); + } + info!("ack channel_id={} acked={}", log_id, outcome.acked); + ( + StatusCode::OK, + Json(AckResp { + acked: outcome.acked, + }), + ) + .into_response() +} #[cfg(test)] mod tests { use super::*; diff --git a/src/main.rs b/src/main.rs index 6df4449..f896794 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,13 +1,16 @@ -use std::{env, net::SocketAddr}; +use std::{env, net::SocketAddr, time::Duration}; use clap::Parser; use qsl_server::{ - app, AppState, Limits, ResourceControls, MAX_BODY_BYTES_CEILING, MAX_PUSH_RATE_BURST_CEILING, - MAX_QUEUE_DEPTH_CEILING, MAX_ROUTE_COUNT_CEILING, ROUTE_IDLE_TTL_MS_DEFAULT, + app, pull_lease_or_error, retention_ttl_or_error, AppState, Limits, ResourceControls, + StoreConfig, MAX_BODY_BYTES_CEILING, MAX_PUSH_RATE_BURST_CEILING, MAX_QUEUE_DEPTH_CEILING, + MAX_ROUTE_COUNT_CEILING, PULL_LEASE_SECS_DEFAULT, RETENTION_TTL_SECS_DEFAULT, }; use tokio::net::TcpListener; use tracing::info; +const RETENTION_SWEEP_INTERVAL_SECS: u64 = 60; + #[derive(Parser, Debug)] #[command(name = "qsl-server", version)] struct Cli { @@ -32,9 +35,18 @@ struct Cli { /// Per-route push token refill per second; 0 disables refill (env: PUSH_RATE_REFILL_PER_SEC, default: 257) #[arg(long)] push_rate_refill_per_sec: Option, - /// Route idle TTL in milliseconds (env: ROUTE_IDLE_TTL_MS, default: 300000) + /// DEPRECATED (NA-0642): ignored; message lifetime is RETENTION_TTL_SECS #[arg(long)] route_idle_ttl_ms: Option, + /// SQLite store file for the durable queue; ':memory:' for ephemeral runs (env: STORE_PATH, required) + #[arg(long)] + store_path: Option, + /// Undelivered-message retention in seconds (env: RETENTION_TTL_SECS, default: 604800) + #[arg(long)] + retention_ttl_secs: Option, + /// Ack-mode pull lease (visibility timeout) in seconds (env: PULL_LEASE_SECS, default: 60) + #[arg(long)] + pull_lease_secs: Option, } #[derive(Clone, Debug)] @@ -46,7 +58,11 @@ struct EnvVals { max_route_count: Option, push_rate_burst: Option, push_rate_refill_per_sec: Option, - route_idle_ttl_ms: Option, + // Deprecated: presence is warned about, the value is never parsed. + route_idle_ttl_ms_present: bool, + store_path: Option, + retention_ttl_secs: Option, + pull_lease_secs: Option, } impl EnvVals { @@ -59,7 +75,10 @@ impl EnvVals { max_route_count: env_usize("MAX_ROUTE_COUNT")?, push_rate_burst: env_usize("PUSH_RATE_BURST")?, push_rate_refill_per_sec: env_usize("PUSH_RATE_REFILL_PER_SEC")?, - route_idle_ttl_ms: env_usize("ROUTE_IDLE_TTL_MS")?, + route_idle_ttl_ms_present: env_opt("ROUTE_IDLE_TTL_MS").is_some(), + store_path: env_opt("STORE_PATH"), + retention_ttl_secs: env_usize("RETENTION_TTL_SECS")?, + pull_lease_secs: env_usize("PULL_LEASE_SECS")?, }) } } @@ -70,6 +89,8 @@ struct Config { port: u16, limits: Limits, controls: ResourceControls, + store: StoreConfig, + deprecated_route_idle_ttl: bool, } fn env_opt(name: &str) -> Option { @@ -122,20 +143,39 @@ fn resolve_config(cli: Cli, env: EnvVals) -> Result { .push_rate_refill_per_sec .or(env.push_rate_refill_per_sec) .unwrap_or(MAX_PUSH_RATE_BURST_CEILING); - let route_idle_ttl_ms = cli - .route_idle_ttl_ms - .or(env.route_idle_ttl_ms) - .unwrap_or(ROUTE_IDLE_TTL_MS_DEFAULT); + // Fail-closed: the durable store location must be explicit. ':memory:' + // is accepted only as a deliberate ephemeral choice. + let store_path = cli + .store_path + .or(env.store_path) + .ok_or_else(|| "ERR_INVALID_CONFIG_STORE_PATH".to_string())?; + let retention_ttl_secs = retention_ttl_or_error( + cli.retention_ttl_secs + .or(env.retention_ttl_secs) + .unwrap_or(RETENTION_TTL_SECS_DEFAULT), + )?; + let pull_lease_secs = pull_lease_or_error( + cli.pull_lease_secs + .or(env.pull_lease_secs) + .unwrap_or(PULL_LEASE_SECS_DEFAULT), + )?; + let deprecated_route_idle_ttl = + cli.route_idle_ttl_ms.is_some() || env.route_idle_ttl_ms_present; Ok(Config { bind, port, limits: Limits::new(max_body_bytes, max_queue_depth)?, - controls: ResourceControls::new_with_route_idle_ttl_ms( + controls: ResourceControls::new( max_route_count, push_rate_burst, push_rate_refill_per_sec, - route_idle_ttl_ms, )?, + store: StoreConfig { + path: store_path, + retention_ttl_secs, + pull_lease_secs, + }, + deprecated_route_idle_ttl, }) } @@ -160,6 +200,11 @@ async fn main() { std::process::exit(1); } }; + if cfg.deprecated_route_idle_ttl { + tracing::warn!( + "event=deprecated_config name=ROUTE_IDLE_TTL_MS action=ignored replacement=RETENTION_TTL_SECS" + ); + } let addr: SocketAddr = match format!("{}:{}", cfg.bind, cfg.port).parse() { Ok(v) => v, Err(_) => { @@ -174,8 +219,30 @@ async fn main() { std::process::exit(1); } }; + let addr = match listener.local_addr() { + Ok(v) => v, + Err(_) => { + tracing::error!("ERR_BIND_LISTEN"); + std::process::exit(1); + } + }; - let state = AppState::new_with_controls(cfg.limits, cfg.controls); + let state = match AppState::new_with_controls_and_store(cfg.limits, cfg.controls, cfg.store) { + Ok(v) => v, + Err(code) => { + tracing::error!("{code}"); + std::process::exit(1); + } + }; + let sweeper_state = state.clone(); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(Duration::from_secs(RETENTION_SWEEP_INTERVAL_SECS)); + ticker.tick().await; // first tick fires immediately; skip it + loop { + ticker.tick().await; + sweeper_state.run_retention_sweep(); + } + }); let app = app(state); info!("qsl-server listening on {}", addr); @@ -188,42 +255,63 @@ async fn main() { #[cfg(test)] mod cli_tests { use super::*; + use qsl_server::{MAX_PULL_LEASE_SECS_CEILING, MAX_RETENTION_TTL_SECS_CEILING}; - fn env_vals( - bind: Option<&str>, - port: Option, - max_body_bytes: Option, - max_queue_depth: Option, - ) -> EnvVals { - EnvVals { - bind: bind.map(|v| v.to_string()), - port, - max_body_bytes, - max_queue_depth, + fn base_cli() -> Cli { + Cli { + bind: None, + port: None, + max_body_bytes: None, + max_queue_depth: None, max_route_count: None, push_rate_burst: None, push_rate_refill_per_sec: None, route_idle_ttl_ms: None, + store_path: Some(":memory:".to_string()), + retention_ttl_secs: None, + pull_lease_secs: None, + } + } + + fn base_env() -> EnvVals { + EnvVals { + bind: None, + port: None, + max_body_bytes: None, + max_queue_depth: None, + max_route_count: None, + push_rate_burst: None, + push_rate_refill_per_sec: None, + route_idle_ttl_ms_present: false, + store_path: None, + retention_ttl_secs: None, + pull_lease_secs: None, } } #[test] fn cli_overrides_env() { - let cli = Cli { - bind: None, - port: Some(9000), - max_body_bytes: Some(4096), - max_queue_depth: Some(9), - max_route_count: Some(8), - push_rate_burst: Some(7), - push_rate_refill_per_sec: Some(6), - route_idle_ttl_ms: Some(5_000), - }; - let mut env = env_vals(Some("0.0.0.0"), Some(8080), Some(1024), Some(1)); + let mut cli = base_cli(); + cli.port = Some(9000); + cli.max_body_bytes = Some(4096); + cli.max_queue_depth = Some(9); + cli.max_route_count = Some(8); + cli.push_rate_burst = Some(7); + cli.push_rate_refill_per_sec = Some(6); + cli.store_path = Some("/tmp/cli.db".to_string()); + cli.retention_ttl_secs = Some(120); + cli.pull_lease_secs = Some(30); + let mut env = base_env(); + env.bind = Some("0.0.0.0".to_string()); + env.port = Some(8080); + env.max_body_bytes = Some(1024); + env.max_queue_depth = Some(1); env.max_route_count = Some(2); env.push_rate_burst = Some(3); env.push_rate_refill_per_sec = Some(4); - env.route_idle_ttl_ms = Some(3_000); + env.store_path = Some("/tmp/env.db".to_string()); + env.retention_ttl_secs = Some(999); + env.pull_lease_secs = Some(99); let cfg = resolve_config(cli, env).unwrap(); assert_eq!(cfg.bind, "0.0.0.0"); assert_eq!(cfg.port, 9000); @@ -232,26 +320,28 @@ mod cli_tests { assert_eq!(cfg.controls.max_route_count, 8); assert_eq!(cfg.controls.push_rate_burst, 7); assert_eq!(cfg.controls.push_rate_refill_per_sec, 6); - assert_eq!(cfg.controls.route_idle_ttl.as_millis(), 5_000); + assert_eq!(cfg.store.path, "/tmp/cli.db"); + assert_eq!(cfg.store.retention_ttl_secs, 120); + assert_eq!(cfg.store.pull_lease_secs, 30); } #[test] fn env_overrides_defaults() { let cli = Cli { - bind: None, - port: None, - max_body_bytes: None, - max_queue_depth: None, - max_route_count: None, - push_rate_burst: None, - push_rate_refill_per_sec: None, - route_idle_ttl_ms: None, + store_path: None, + ..base_cli() }; - let mut env = env_vals(Some("0.0.0.0"), Some(7070), Some(2048), Some(7)); + let mut env = base_env(); + env.bind = Some("0.0.0.0".to_string()); + env.port = Some(7070); + env.max_body_bytes = Some(2048); + env.max_queue_depth = Some(7); env.max_route_count = Some(6); env.push_rate_burst = Some(5); env.push_rate_refill_per_sec = Some(4); - env.route_idle_ttl_ms = Some(3_000); + env.store_path = Some("/tmp/env-only.db".to_string()); + env.retention_ttl_secs = Some(3600); + env.pull_lease_secs = Some(120); let cfg = resolve_config(cli, env).unwrap(); assert_eq!(cfg.bind, "0.0.0.0"); assert_eq!(cfg.port, 7070); @@ -260,23 +350,78 @@ mod cli_tests { assert_eq!(cfg.controls.max_route_count, 6); assert_eq!(cfg.controls.push_rate_burst, 5); assert_eq!(cfg.controls.push_rate_refill_per_sec, 4); - assert_eq!(cfg.controls.route_idle_ttl.as_millis(), 3_000); + assert_eq!(cfg.store.path, "/tmp/env-only.db"); + assert_eq!(cfg.store.retention_ttl_secs, 3600); + assert_eq!(cfg.store.pull_lease_secs, 120); } #[test] - fn limits_are_capped() { + fn store_defaults_apply_when_unset() { + let cfg = resolve_config(base_cli(), base_env()).unwrap(); + assert_eq!(cfg.store.path, ":memory:"); + assert_eq!(cfg.store.retention_ttl_secs, RETENTION_TTL_SECS_DEFAULT); + assert_eq!(cfg.store.pull_lease_secs, PULL_LEASE_SECS_DEFAULT); + assert!(!cfg.deprecated_route_idle_ttl); + } + + #[test] + fn missing_store_path_fails_closed() { let cli = Cli { - bind: None, - port: None, - max_body_bytes: Some(MAX_BODY_BYTES_CEILING * 2), - max_queue_depth: Some(MAX_QUEUE_DEPTH_CEILING * 2), - max_route_count: Some(MAX_ROUTE_COUNT_CEILING * 2), - push_rate_burst: Some(MAX_PUSH_RATE_BURST_CEILING * 2), - push_rate_refill_per_sec: Some(MAX_PUSH_RATE_BURST_CEILING * 20), - route_idle_ttl_ms: Some(qsl_server::MAX_ROUTE_IDLE_TTL_MS_CEILING * 2), + store_path: None, + ..base_cli() }; - let env = env_vals(None, None, None, None); - let cfg = resolve_config(cli, env).unwrap(); + let err = resolve_config(cli, base_env()).unwrap_err(); + assert_eq!(err, "ERR_INVALID_CONFIG_STORE_PATH"); + } + + #[test] + fn zero_retention_or_lease_fails_closed() { + let mut cli = base_cli(); + cli.retention_ttl_secs = Some(0); + assert_eq!( + resolve_config(cli, base_env()).unwrap_err(), + "ERR_INVALID_CONFIG_RETENTION_TTL_SECS" + ); + let mut cli = base_cli(); + cli.pull_lease_secs = Some(0); + assert_eq!( + resolve_config(cli, base_env()).unwrap_err(), + "ERR_INVALID_CONFIG_PULL_LEASE_SECS" + ); + } + + #[test] + fn retention_and_lease_are_capped() { + let mut cli = base_cli(); + cli.retention_ttl_secs = Some(MAX_RETENTION_TTL_SECS_CEILING * 2); + cli.pull_lease_secs = Some(MAX_PULL_LEASE_SECS_CEILING * 2); + let cfg = resolve_config(cli, base_env()).unwrap(); + assert_eq!(cfg.store.retention_ttl_secs, MAX_RETENTION_TTL_SECS_CEILING); + assert_eq!(cfg.store.pull_lease_secs, MAX_PULL_LEASE_SECS_CEILING); + } + + #[test] + fn deprecated_route_idle_ttl_is_flagged_not_fatal() { + let mut cli = base_cli(); + cli.route_idle_ttl_ms = Some(5_000); + let cfg = resolve_config(cli, base_env()).unwrap(); + assert!(cfg.deprecated_route_idle_ttl); + + let mut env = base_env(); + env.route_idle_ttl_ms_present = true; + let cfg = resolve_config(base_cli(), env).unwrap(); + assert!(cfg.deprecated_route_idle_ttl); + } + + #[test] + fn limits_are_capped() { + let mut cli = base_cli(); + cli.max_body_bytes = Some(MAX_BODY_BYTES_CEILING * 2); + cli.max_queue_depth = Some(MAX_QUEUE_DEPTH_CEILING * 2); + cli.max_route_count = Some(MAX_ROUTE_COUNT_CEILING * 2); + cli.push_rate_burst = Some(MAX_PUSH_RATE_BURST_CEILING * 2); + cli.push_rate_refill_per_sec = Some(MAX_PUSH_RATE_BURST_CEILING * 20); + let cfg = resolve_config(cli, base_env()).unwrap(); assert_eq!(cfg.limits.max_body_bytes, MAX_BODY_BYTES_CEILING); assert_eq!(cfg.limits.max_queue_depth, MAX_QUEUE_DEPTH_CEILING); assert_eq!(cfg.controls.max_route_count, MAX_ROUTE_COUNT_CEILING); @@ -285,60 +430,27 @@ mod cli_tests { cfg.controls.push_rate_refill_per_sec, qsl_server::MAX_PUSH_RATE_REFILL_PER_SEC_CEILING ); - assert_eq!( - cfg.controls.route_idle_ttl.as_millis(), - qsl_server::MAX_ROUTE_IDLE_TTL_MS_CEILING as u128 - ); } #[test] fn default_bind_is_loopback() { - let cli = Cli { - bind: None, - port: None, - max_body_bytes: None, - max_queue_depth: None, - max_route_count: None, - push_rate_burst: None, - push_rate_refill_per_sec: None, - route_idle_ttl_ms: None, - }; - let env = env_vals(None, None, None, None); - let cfg = resolve_config(cli, env).unwrap(); + let cfg = resolve_config(base_cli(), base_env()).unwrap(); assert_eq!(cfg.bind, "127.0.0.1"); } #[test] fn explicit_public_bind_is_opt_in() { - let cli = Cli { - bind: Some("0.0.0.0".to_string()), - port: None, - max_body_bytes: None, - max_queue_depth: None, - max_route_count: None, - push_rate_burst: None, - push_rate_refill_per_sec: None, - route_idle_ttl_ms: None, - }; - let env = env_vals(None, None, None, None); - let cfg = resolve_config(cli, env).unwrap(); + let mut cli = base_cli(); + cli.bind = Some("0.0.0.0".to_string()); + let cfg = resolve_config(cli, base_env()).unwrap(); assert_eq!(cfg.bind, "0.0.0.0"); } #[test] fn env_bind_can_enable_public_bind() { - let cli = Cli { - bind: None, - port: None, - max_body_bytes: None, - max_queue_depth: None, - max_route_count: None, - push_rate_burst: None, - push_rate_refill_per_sec: None, - route_idle_ttl_ms: None, - }; - let env = env_vals(Some("0.0.0.0"), None, None, None); - let cfg = resolve_config(cli, env).unwrap(); + let mut env = base_env(); + env.bind = Some("0.0.0.0".to_string()); + let cfg = resolve_config(base_cli(), env).unwrap(); assert_eq!(cfg.bind, "0.0.0.0"); } @@ -353,38 +465,16 @@ mod cli_tests { #[test] fn zero_limits_are_rejected() { - let env = env_vals(None, None, Some(0), Some(1)); - let body_err = resolve_config( - Cli { - bind: None, - port: None, - max_body_bytes: None, - max_queue_depth: None, - max_route_count: None, - push_rate_burst: None, - push_rate_refill_per_sec: None, - route_idle_ttl_ms: None, - }, - env, - ) - .unwrap_err(); + let mut env = base_env(); + env.max_body_bytes = Some(0); + env.max_queue_depth = Some(1); + let body_err = resolve_config(base_cli(), env).unwrap_err(); assert_eq!(body_err, "ERR_INVALID_CONFIG_MAX_BODY_BYTES"); - let env = env_vals(None, None, Some(1), Some(0)); - let depth_err = resolve_config( - Cli { - bind: None, - port: None, - max_body_bytes: None, - max_queue_depth: None, - max_route_count: None, - push_rate_burst: None, - push_rate_refill_per_sec: None, - route_idle_ttl_ms: None, - }, - env, - ) - .unwrap_err(); + let mut env = base_env(); + env.max_body_bytes = Some(1); + env.max_queue_depth = Some(0); + let depth_err = resolve_config(base_cli(), env).unwrap_err(); assert_eq!(depth_err, "ERR_INVALID_CONFIG_MAX_QUEUE_DEPTH"); } diff --git a/src/store.rs b/src/store.rs new file mode 100644 index 0000000..7052304 --- /dev/null +++ b/src/store.rs @@ -0,0 +1,491 @@ +use rusqlite::{params, params_from_iter, Connection}; +use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; + +pub const RETENTION_TTL_SECS_DEFAULT: usize = 604_800; // 7 days +pub const MAX_RETENTION_TTL_SECS_CEILING: usize = 2_592_000; // 30 days +pub const PULL_LEASE_SECS_DEFAULT: usize = 60; +pub const MAX_PULL_LEASE_SECS_CEILING: usize = 3_600; + +const SCHEMA_VERSION: i64 = 1; + +// Bounds a single ack request's IN-list; well under SQLite's variable limit. +pub const MAX_ACK_IDS: usize = 4_096; + +/// Durable-store configuration. `path` accepts a filesystem path or the +/// literal `:memory:` for explicitly ephemeral stores (tests, dev runs). +#[derive(Clone, Debug)] +pub struct StoreConfig { + pub path: String, + pub retention_ttl_secs: usize, + pub pull_lease_secs: usize, +} + +impl Default for StoreConfig { + fn default() -> Self { + Self { + path: ":memory:".to_string(), + retention_ttl_secs: RETENTION_TTL_SECS_DEFAULT, + pull_lease_secs: PULL_LEASE_SECS_DEFAULT, + } + } +} + +pub fn retention_ttl_or_error(value: usize) -> Result { + if value == 0 { + return Err("ERR_INVALID_CONFIG_RETENTION_TTL_SECS".to_string()); + } + Ok(value.min(MAX_RETENTION_TTL_SECS_CEILING)) +} + +pub fn pull_lease_or_error(value: usize) -> Result { + if value == 0 { + return Err("ERR_INVALID_CONFIG_PULL_LEASE_SECS".to_string()); + } + Ok(value.min(MAX_PULL_LEASE_SECS_CEILING)) +} + +#[derive(Debug)] +pub(crate) struct StoredMsg { + pub msg_id: String, + pub body: Vec, +} + +#[derive(Debug, Default)] +pub struct SweepStats { + pub expired_messages: usize, + // (redacted channel log id, expired message count) per affected route + pub expired_routes: Vec<(String, usize)>, + // route keys whose route row was removed (for rate-bucket pruning) + pub removed_route_keys: Vec, +} + +#[derive(Debug)] +pub(crate) enum EnqueueOutcome { + Accepted, + Overloaded { depth: usize }, + RouteCap { live_routes: usize }, +} + +#[derive(Debug, Clone, Copy)] +pub(crate) enum PullMode { + // delete-on-deliver, the pre-durability wire contract + Legacy, + // mark in-flight with a visibility deadline; deletion happens on ack + Lease, +} + +#[derive(Debug)] +pub(crate) struct PullOutcome { + pub items: Vec, + pub route_drained: bool, + pub sweep: SweepStats, +} + +#[derive(Debug)] +pub(crate) struct AckOutcome { + pub acked: usize, + pub route_drained: bool, + pub sweep: SweepStats, +} + +#[derive(Debug)] +pub(crate) struct RouteStatus { + pub route_exists: bool, + pub depth: usize, + pub live_routes: usize, + pub sweep: SweepStats, +} + +pub(crate) fn now_unix_secs() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) +} + +#[derive(Clone)] +pub(crate) struct Store { + conn: Arc>, + retention_ttl_secs: i64, + pull_lease_secs: i64, +} + +fn map_err(e: rusqlite::Error) -> String { + format!("ERR_STORE {e}") +} + +impl Store { + pub(crate) fn open(cfg: &StoreConfig) -> Result { + let retention_ttl_secs = retention_ttl_or_error(cfg.retention_ttl_secs)? as i64; + let pull_lease_secs = pull_lease_or_error(cfg.pull_lease_secs)? as i64; + let conn = + Connection::open(&cfg.path).map_err(|_| "ERR_INVALID_CONFIG_STORE_PATH".to_string())?; + // FULL: a 200 on push means the message is fsynced, not merely buffered. + conn.pragma_update(None, "journal_mode", "WAL") + .map_err(map_err)?; + conn.pragma_update(None, "synchronous", "FULL") + .map_err(map_err)?; + conn.pragma_update(None, "foreign_keys", "ON") + .map_err(map_err)?; + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS routes ( + route_key TEXT PRIMARY KEY, + log_id TEXT NOT NULL, + created_at INTEGER NOT NULL, + last_touched INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS messages ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + msg_id TEXT NOT NULL, + route_key TEXT NOT NULL REFERENCES routes(route_key) ON DELETE CASCADE, + body BLOB NOT NULL, + enqueued_at INTEGER NOT NULL, + leased_until INTEGER + ); + CREATE INDEX IF NOT EXISTS idx_messages_route_seq ON messages(route_key, seq); + CREATE INDEX IF NOT EXISTS idx_messages_enqueued ON messages(enqueued_at);", + ) + .map_err(map_err)?; + conn.execute( + "INSERT OR IGNORE INTO meta(key, value) VALUES('schema_version', ?1)", + params![SCHEMA_VERSION.to_string()], + ) + .map_err(map_err)?; + let stored: String = conn + .query_row( + "SELECT value FROM meta WHERE key='schema_version'", + [], + |row| row.get(0), + ) + .map_err(map_err)?; + let stored: i64 = stored + .parse() + .map_err(|_| "ERR_STORE_VERSION".to_string())?; + if stored > SCHEMA_VERSION { + return Err("ERR_STORE_VERSION".to_string()); + } + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + retention_ttl_secs, + pull_lease_secs, + }) + } + + pub(crate) fn retention_ttl_secs(&self) -> i64 { + self.retention_ttl_secs + } + + fn sweep_expired(conn: &Connection, ttl: i64, now: i64) -> Result { + let mut stats = SweepStats::default(); + { + let mut stmt = conn + .prepare_cached( + "SELECT r.log_id, COUNT(m.seq) FROM messages m + JOIN routes r ON r.route_key = m.route_key + WHERE m.enqueued_at + ?1 <= ?2 + GROUP BY m.route_key", + ) + .map_err(map_err)?; + let rows = stmt + .query_map(params![ttl, now], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)) + }) + .map_err(map_err)?; + for row in rows { + let (log_id, n) = row.map_err(map_err)?; + stats.expired_routes.push((log_id, n as usize)); + } + } + stats.expired_messages = conn + .execute( + "DELETE FROM messages WHERE enqueued_at + ?1 <= ?2", + params![ttl, now], + ) + .map_err(map_err)?; + if stats.expired_messages > 0 { + let mut stmt = conn + .prepare_cached( + "SELECT route_key FROM routes r + WHERE NOT EXISTS (SELECT 1 FROM messages m WHERE m.route_key = r.route_key)", + ) + .map_err(map_err)?; + let rows = stmt + .query_map([], |row| row.get::<_, String>(0)) + .map_err(map_err)?; + for row in rows { + stats.removed_route_keys.push(row.map_err(map_err)?); + } + conn.execute( + "DELETE FROM routes WHERE NOT EXISTS + (SELECT 1 FROM messages WHERE messages.route_key = routes.route_key)", + [], + ) + .map_err(map_err)?; + } + Ok(stats) + } + + pub(crate) fn retention_sweep(&self, now: i64) -> Result { + let mut guard = self + .conn + .lock() + .map_err(|_| "ERR_LOCK_POISON".to_string())?; + let tx = guard.transaction().map_err(map_err)?; + let stats = Self::sweep_expired(&tx, self.retention_ttl_secs, now)?; + tx.commit().map_err(map_err)?; + Ok(stats) + } + + pub(crate) fn route_status(&self, route_key: &str, now: i64) -> Result { + let mut guard = self + .conn + .lock() + .map_err(|_| "ERR_LOCK_POISON".to_string())?; + let tx = guard.transaction().map_err(map_err)?; + let sweep = Self::sweep_expired(&tx, self.retention_ttl_secs, now)?; + let route_exists: bool = tx + .query_row( + "SELECT EXISTS(SELECT 1 FROM routes WHERE route_key = ?1)", + params![route_key], + |row| row.get(0), + ) + .map_err(map_err)?; + let depth: i64 = tx + .query_row( + "SELECT COUNT(*) FROM messages WHERE route_key = ?1", + params![route_key], + |row| row.get(0), + ) + .map_err(map_err)?; + let live_routes: i64 = tx + .query_row("SELECT COUNT(*) FROM routes", [], |row| row.get(0)) + .map_err(map_err)?; + tx.commit().map_err(map_err)?; + Ok(RouteStatus { + route_exists, + depth: depth as usize, + live_routes: live_routes as usize, + sweep, + }) + } + + #[allow(clippy::too_many_arguments)] + pub(crate) fn enqueue( + &self, + route_key: &str, + log_id: &str, + msg_id: &str, + body: &[u8], + now: i64, + max_queue_depth: usize, + max_route_count: usize, + ) -> Result { + let mut guard = self + .conn + .lock() + .map_err(|_| "ERR_LOCK_POISON".to_string())?; + let tx = guard.transaction().map_err(map_err)?; + let route_exists: bool = tx + .query_row( + "SELECT EXISTS(SELECT 1 FROM routes WHERE route_key = ?1)", + params![route_key], + |row| row.get(0), + ) + .map_err(map_err)?; + if route_exists { + let depth: i64 = tx + .query_row( + "SELECT COUNT(*) FROM messages WHERE route_key = ?1", + params![route_key], + |row| row.get(0), + ) + .map_err(map_err)?; + if depth as usize >= max_queue_depth { + return Ok(EnqueueOutcome::Overloaded { + depth: depth as usize, + }); + } + } else { + let live_routes: i64 = tx + .query_row("SELECT COUNT(*) FROM routes", [], |row| row.get(0)) + .map_err(map_err)?; + if live_routes as usize >= max_route_count { + return Ok(EnqueueOutcome::RouteCap { + live_routes: live_routes as usize, + }); + } + tx.execute( + "INSERT INTO routes(route_key, log_id, created_at, last_touched) + VALUES(?1, ?2, ?3, ?3)", + params![route_key, log_id, now], + ) + .map_err(map_err)?; + } + tx.execute( + "INSERT INTO messages(msg_id, route_key, body, enqueued_at) + VALUES(?1, ?2, ?3, ?4)", + params![msg_id, route_key, body, now], + ) + .map_err(map_err)?; + tx.execute( + "UPDATE routes SET last_touched = ?2 WHERE route_key = ?1", + params![route_key, now], + ) + .map_err(map_err)?; + tx.commit().map_err(map_err)?; + Ok(EnqueueOutcome::Accepted) + } + + pub(crate) fn pull( + &self, + route_key: &str, + max: usize, + now: i64, + mode: PullMode, + ) -> Result { + let mut guard = self + .conn + .lock() + .map_err(|_| "ERR_LOCK_POISON".to_string())?; + let tx = guard.transaction().map_err(map_err)?; + let sweep = Self::sweep_expired(&tx, self.retention_ttl_secs, now)?; + let mut items = Vec::new(); + let mut seqs: Vec = Vec::new(); + { + let mut stmt = tx + .prepare_cached( + "SELECT seq, msg_id, body FROM messages + WHERE route_key = ?1 AND (leased_until IS NULL OR leased_until <= ?2) + ORDER BY seq LIMIT ?3", + ) + .map_err(map_err)?; + let rows = stmt + .query_map(params![route_key, now, max as i64], |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, String>(1)?, + row.get::<_, Vec>(2)?, + )) + }) + .map_err(map_err)?; + for row in rows { + let (seq, msg_id, body) = row.map_err(map_err)?; + seqs.push(seq); + items.push(StoredMsg { msg_id, body }); + } + } + if !seqs.is_empty() { + let placeholders = vec!["?"; seqs.len()].join(","); + match mode { + PullMode::Legacy => { + let sql = format!("DELETE FROM messages WHERE seq IN ({placeholders})"); + tx.execute(&sql, params_from_iter(seqs.iter())) + .map_err(map_err)?; + } + PullMode::Lease => { + let deadline = now + self.pull_lease_secs; + let sql = format!( + "UPDATE messages SET leased_until = {deadline} WHERE seq IN ({placeholders})" + ); + tx.execute(&sql, params_from_iter(seqs.iter())) + .map_err(map_err)?; + } + } + } + let mut route_drained = false; + if matches!(mode, PullMode::Legacy) { + let remaining: i64 = tx + .query_row( + "SELECT COUNT(*) FROM messages WHERE route_key = ?1", + params![route_key], + |row| row.get(0), + ) + .map_err(map_err)?; + if remaining == 0 { + let removed = tx + .execute( + "DELETE FROM routes WHERE route_key = ?1", + params![route_key], + ) + .map_err(map_err)?; + route_drained = removed > 0; + } else { + tx.execute( + "UPDATE routes SET last_touched = ?2 WHERE route_key = ?1", + params![route_key, now], + ) + .map_err(map_err)?; + } + } else if !seqs.is_empty() { + tx.execute( + "UPDATE routes SET last_touched = ?2 WHERE route_key = ?1", + params![route_key, now], + ) + .map_err(map_err)?; + } + tx.commit().map_err(map_err)?; + Ok(PullOutcome { + items, + route_drained, + sweep, + }) + } + + pub(crate) fn ack( + &self, + route_key: &str, + ids: &[String], + now: i64, + ) -> Result { + let mut guard = self + .conn + .lock() + .map_err(|_| "ERR_LOCK_POISON".to_string())?; + let tx = guard.transaction().map_err(map_err)?; + let sweep = Self::sweep_expired(&tx, self.retention_ttl_secs, now)?; + // Only leased (in-flight or lease-expired-but-undelivered-again) copies + // are deletable: an unleased duplicate copy was never delivered and + // must survive (NA-0275 duplicate-id contract). + let placeholders = vec!["?"; ids.len()].join(","); + let sql = format!( + "DELETE FROM messages WHERE route_key = ?1 AND leased_until IS NOT NULL + AND msg_id IN ({placeholders})" + ); + let mut bind: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(ids.len() + 1); + bind.push(&route_key); + for id in ids { + bind.push(id); + } + let acked = tx.execute(&sql, bind.as_slice()).map_err(map_err)?; + let mut route_drained = false; + if acked > 0 { + let remaining: i64 = tx + .query_row( + "SELECT COUNT(*) FROM messages WHERE route_key = ?1", + params![route_key], + |row| row.get(0), + ) + .map_err(map_err)?; + if remaining == 0 { + let removed = tx + .execute( + "DELETE FROM routes WHERE route_key = ?1", + params![route_key], + ) + .map_err(map_err)?; + route_drained = removed > 0; + } + } + tx.commit().map_err(map_err)?; + Ok(AckOutcome { + acked, + route_drained, + sweep, + }) + } +} diff --git a/tests/config_semantics.rs b/tests/config_semantics.rs index dfc152e..6af869f 100644 --- a/tests/config_semantics.rs +++ b/tests/config_semantics.rs @@ -141,7 +141,9 @@ async fn pull( #[test] fn missing_size_depth_config_uses_safe_defaults() { - assert_config_stays_running(&[]); + // NA-0642: STORE_PATH itself has NO default (fail-closed, see + // store_path_is_required_and_fail_closed); every other knob defaults. + assert_config_stays_running(&[("STORE_PATH", ":memory:")]); let defaults = Limits::default(); assert_eq!(defaults.max_body_bytes, MAX_BODY_BYTES_CEILING); assert_eq!(defaults.max_queue_depth, MAX_QUEUE_DEPTH_CEILING); @@ -160,51 +162,96 @@ fn missing_size_depth_config_uses_safe_defaults() { #[test] fn malformed_or_zero_size_depth_config_fails_closed() { + const STORE: (&str, &str) = ("STORE_PATH", ":memory:"); assert_config_failure( - &[("PORT", "0"), ("MAX_BODY_BYTES", "not-a-size")], + &[("PORT", "0"), STORE, ("MAX_BODY_BYTES", "not-a-size")], "ERR_INVALID_CONFIG_MAX_BODY_BYTES", ); assert_config_failure( - &[("PORT", "0"), ("MAX_BODY_BYTES", "0")], + &[("PORT", "0"), STORE, ("MAX_BODY_BYTES", "0")], "ERR_INVALID_CONFIG_MAX_BODY_BYTES", ); assert_config_failure( - &[("PORT", "0"), ("MAX_QUEUE_DEPTH", "not-a-depth")], + &[("PORT", "0"), STORE, ("MAX_QUEUE_DEPTH", "not-a-depth")], "ERR_INVALID_CONFIG_MAX_QUEUE_DEPTH", ); assert_config_failure( - &[("PORT", "0"), ("MAX_QUEUE_DEPTH", "0")], + &[("PORT", "0"), STORE, ("MAX_QUEUE_DEPTH", "0")], "ERR_INVALID_CONFIG_MAX_QUEUE_DEPTH", ); assert_config_failure( - &[("PORT", "0"), ("MAX_ROUTE_COUNT", "not-a-route-count")], + &[ + ("PORT", "0"), + STORE, + ("MAX_ROUTE_COUNT", "not-a-route-count"), + ], "ERR_INVALID_CONFIG_MAX_ROUTE_COUNT", ); assert_config_failure( - &[("PORT", "0"), ("MAX_ROUTE_COUNT", "0")], + &[("PORT", "0"), STORE, ("MAX_ROUTE_COUNT", "0")], "ERR_INVALID_CONFIG_MAX_ROUTE_COUNT", ); assert_config_failure( - &[("PORT", "0"), ("PUSH_RATE_BURST", "not-a-burst")], + &[("PORT", "0"), STORE, ("PUSH_RATE_BURST", "not-a-burst")], "ERR_INVALID_CONFIG_PUSH_RATE_BURST", ); assert_config_failure( - &[("PORT", "0"), ("PUSH_RATE_BURST", "0")], + &[("PORT", "0"), STORE, ("PUSH_RATE_BURST", "0")], "ERR_INVALID_CONFIG_PUSH_RATE_BURST", ); assert_config_failure( - &[("PORT", "0"), ("PUSH_RATE_REFILL_PER_SEC", "not-a-refill")], + &[ + ("PORT", "0"), + STORE, + ("PUSH_RATE_REFILL_PER_SEC", "not-a-refill"), + ], "ERR_INVALID_CONFIG_PUSH_RATE_REFILL_PER_SEC", ); - assert_config_stays_running(&[("PUSH_RATE_REFILL_PER_SEC", "0")]); + assert_config_stays_running(&[STORE, ("PUSH_RATE_REFILL_PER_SEC", "0")]); assert_config_failure( - &[("PORT", "0"), ("ROUTE_IDLE_TTL_MS", "not-a-ttl")], - "ERR_INVALID_CONFIG_ROUTE_IDLE_TTL_MS", + &[("PORT", "0"), STORE, ("RETENTION_TTL_SECS", "not-a-ttl")], + "ERR_INVALID_CONFIG_RETENTION_TTL_SECS", ); assert_config_failure( - &[("PORT", "0"), ("ROUTE_IDLE_TTL_MS", "0")], - "ERR_INVALID_CONFIG_ROUTE_IDLE_TTL_MS", + &[("PORT", "0"), STORE, ("RETENTION_TTL_SECS", "0")], + "ERR_INVALID_CONFIG_RETENTION_TTL_SECS", ); + assert_config_failure( + &[("PORT", "0"), STORE, ("PULL_LEASE_SECS", "not-a-lease")], + "ERR_INVALID_CONFIG_PULL_LEASE_SECS", + ); + assert_config_failure( + &[("PORT", "0"), STORE, ("PULL_LEASE_SECS", "0")], + "ERR_INVALID_CONFIG_PULL_LEASE_SECS", + ); +} + +#[test] +fn store_path_is_required_and_fail_closed() { + // No STORE_PATH at all: the relay must refuse to start rather than fall + // back to a silent in-memory queue (NA-0642 durability posture). + assert_config_failure(&[("PORT", "0")], "ERR_INVALID_CONFIG_STORE_PATH"); + // Unopenable path: same fail-closed code at startup. + assert_config_failure( + &[ + ("PORT", "0"), + ("STORE_PATH", "/nonexistent-na0642-dir/relay.db"), + ], + "ERR_INVALID_CONFIG_STORE_PATH", + ); +} + +#[test] +fn route_idle_ttl_env_is_deprecated_and_ignored() { + // NA-0642: the idle-route discard is retired; RETENTION_TTL_SECS governs + // message lifetime. A stale relay.env must not brick a restart — any + // ROUTE_IDLE_TTL_MS value (even formerly-invalid ones) is warn-and-ignore. + assert_config_stays_running(&[ + ("STORE_PATH", ":memory:"), + ("ROUTE_IDLE_TTL_MS", "not-a-ttl"), + ]); + assert_config_stays_running(&[("STORE_PATH", ":memory:"), ("ROUTE_IDLE_TTL_MS", "0")]); + assert_config_stays_running(&[("STORE_PATH", ":memory:"), ("ROUTE_IDLE_TTL_MS", "300000")]); } #[test] @@ -214,14 +261,17 @@ fn above_ceiling_size_depth_config_is_capped_explicitly() { let route_above_ceiling = (MAX_ROUTE_COUNT_CEILING + 1).to_string(); let burst_above_ceiling = (MAX_PUSH_RATE_BURST_CEILING + 1).to_string(); let refill_above_ceiling = (MAX_PUSH_RATE_REFILL_PER_SEC_CEILING + 1).to_string(); - let ttl_above_ceiling = (MAX_ROUTE_IDLE_TTL_MS_CEILING + 1).to_string(); + let retention_above_ceiling = (qsl_server::MAX_RETENTION_TTL_SECS_CEILING + 1).to_string(); + let lease_above_ceiling = (qsl_server::MAX_PULL_LEASE_SECS_CEILING + 1).to_string(); assert_config_stays_running(&[ + ("STORE_PATH", ":memory:"), ("MAX_BODY_BYTES", body_above_ceiling.as_str()), ("MAX_QUEUE_DEPTH", depth_above_ceiling.as_str()), ("MAX_ROUTE_COUNT", route_above_ceiling.as_str()), ("PUSH_RATE_BURST", burst_above_ceiling.as_str()), ("PUSH_RATE_REFILL_PER_SEC", refill_above_ceiling.as_str()), - ("ROUTE_IDLE_TTL_MS", ttl_above_ceiling.as_str()), + ("RETENTION_TTL_SECS", retention_above_ceiling.as_str()), + ("PULL_LEASE_SECS", lease_above_ceiling.as_str()), ]); let limits = Limits::new(MAX_BODY_BYTES_CEILING + 1, MAX_QUEUE_DEPTH_CEILING + 1).unwrap(); let controls = ResourceControls::new_with_route_idle_ttl_ms( @@ -249,7 +299,11 @@ fn above_ceiling_size_depth_config_is_capped_explicitly() { fn invalid_port_and_bind_addr_fail_closed() { assert_config_failure(&[("PORT", "not-a-port")], "ERR_INVALID_ENV_PORT"); assert_config_failure( - &[("PORT", "0"), ("BIND_ADDR", "not a socket address")], + &[ + ("PORT", "0"), + ("STORE_PATH", ":memory:"), + ("BIND_ADDR", "not a socket address"), + ], "ERR_BIND_PARSE", ); } diff --git a/tests/na0642_ack_contract.rs b/tests/na0642_ack_contract.rs new file mode 100644 index 0000000..1915943 --- /dev/null +++ b/tests/na0642_ack_contract.rs @@ -0,0 +1,343 @@ +// NA-0642 acknowledged-pull contract (design-lock option B): +// - GET /v1/pull?ack=lease returns messages WITHOUT deleting; each returned +// message is leased (in-flight) until now + PULL_LEASE_SECS. +// - POST /v1/pull/ack {"ids":[...]} deletes ONLY leased copies; unleased +// duplicates (NA-0275 contract) survive. +// - Un-acked leased messages reappear after the lease expires. +// - Legacy pulls (no ack param) keep delete-on-deliver and never see messages +// another pull holds under a live lease. + +use qsl_server::{app, AppState, Limits, ResourceControls, StoreConfig}; +use reqwest::StatusCode as ReqStatus; +use serde::Deserialize; +use std::time::Duration; +use tokio::net::TcpListener; + +const ROUTE_TOKEN_HEADER: &str = "X-QSL-Route-Token"; +const MSG_ID_HEADER: &str = "X-Msg-Id"; + +#[derive(Deserialize)] +struct PullItem { + id: String, + data: Vec, +} + +#[derive(Deserialize)] +struct PullResp { + items: Vec, +} + +#[derive(Deserialize)] +struct AckResp { + acked: usize, +} + +async fn spawn_server(lease_secs: usize) -> (String, tokio::task::JoinHandle<()>) { + let store = StoreConfig { + pull_lease_secs: lease_secs, + ..StoreConfig::default() + }; + let state = AppState::new_with_auth_controls_and_store( + Limits::new(1024 * 1024, 16).unwrap(), + ResourceControls::new(8, 16, 16).unwrap(), + None, + store, + ) + .unwrap_or_else(|e| panic!("{e}")); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .unwrap_or_else(|e| panic!("{e}")); + let addr = listener.local_addr().unwrap_or_else(|e| panic!("{e}")); + let handle = tokio::spawn(async move { + axum::serve(listener, app(state)) + .await + .unwrap_or_else(|e| panic!("{e}")); + }); + (format!("http://{addr}"), handle) +} + +async fn push( + client: &reqwest::Client, + base: &str, + route_token: &str, + msg_id: Option<&str>, + body: impl Into>, +) -> reqwest::Response { + let mut request = client + .post(format!("{base}/v1/push")) + .header(ROUTE_TOKEN_HEADER, route_token) + .body(body.into()); + if let Some(id) = msg_id { + request = request.header(MSG_ID_HEADER, id); + } + request.send().await.unwrap_or_else(|e| panic!("{e}")) +} + +async fn pull_legacy( + client: &reqwest::Client, + base: &str, + route_token: &str, + max: usize, +) -> reqwest::Response { + client + .get(format!("{base}/v1/pull?max={max}")) + .header(ROUTE_TOKEN_HEADER, route_token) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")) +} + +async fn pull_lease( + client: &reqwest::Client, + base: &str, + route_token: &str, + max: usize, +) -> reqwest::Response { + client + .get(format!("{base}/v1/pull?max={max}&ack=lease")) + .header(ROUTE_TOKEN_HEADER, route_token) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")) +} + +async fn ack( + client: &reqwest::Client, + base: &str, + route_token: &str, + ids: &[&str], +) -> reqwest::Response { + client + .post(format!("{base}/v1/pull/ack")) + .header(ROUTE_TOKEN_HEADER, route_token) + .json(&serde_json::json!({ "ids": ids })) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")) +} + +#[tokio::test(flavor = "current_thread")] +async fn lease_pull_does_not_delete_until_ack() { + let (base, handle) = spawn_server(60).await; + let client = reqwest::Client::new(); + let route = "NA0642_ACK_BASIC"; + + let accepted = push( + &client, + &base, + route, + Some("NA0642_ACK_MSG_1"), + b"m1".to_vec(), + ) + .await; + assert_eq!(accepted.status(), ReqStatus::OK); + + let leased = pull_lease(&client, &base, route, 4).await; + assert_eq!(leased.status(), ReqStatus::OK); + let body: PullResp = leased.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(body.items.len(), 1); + assert_eq!(body.items[0].id, "NA0642_ACK_MSG_1"); + assert_eq!(body.items[0].data.as_slice(), b"m1"); + + // In-flight: invisible to both pull modes while the lease is live. + let release = pull_lease(&client, &base, route, 4).await; + assert_eq!(release.status(), ReqStatus::NO_CONTENT); + let legacy = pull_legacy(&client, &base, route, 4).await; + assert_eq!(legacy.status(), ReqStatus::NO_CONTENT); + + let acked = ack(&client, &base, route, &["NA0642_ACK_MSG_1"]).await; + assert_eq!(acked.status(), ReqStatus::OK); + let acked: AckResp = acked.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(acked.acked, 1); + + let after = pull_lease(&client, &base, route, 4).await; + assert_eq!(after.status(), ReqStatus::NO_CONTENT); + + // Idempotent: re-acking a deleted id succeeds with acked=0. + let reack = ack(&client, &base, route, &["NA0642_ACK_MSG_1"]).await; + assert_eq!(reack.status(), ReqStatus::OK); + let reack: AckResp = reack.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(reack.acked, 0); + + handle.abort(); +} + +#[tokio::test(flavor = "current_thread")] +async fn unacked_lease_reappears_after_expiry() { + let (base, handle) = spawn_server(1).await; + let client = reqwest::Client::new(); + let route = "NA0642_ACK_REDELIVERY"; + + let accepted = push( + &client, + &base, + route, + Some("NA0642_REDELIVER_1"), + b"r1".to_vec(), + ) + .await; + assert_eq!(accepted.status(), ReqStatus::OK); + + let leased = pull_lease(&client, &base, route, 1).await; + assert_eq!(leased.status(), ReqStatus::OK); + + // No ack (the puller "crashed"); the message must come back. + tokio::time::sleep(Duration::from_millis(1500)).await; + + let redelivered = pull_lease(&client, &base, route, 1).await; + assert_eq!(redelivered.status(), ReqStatus::OK); + let body: PullResp = redelivered.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(body.items.len(), 1); + assert_eq!(body.items[0].id, "NA0642_REDELIVER_1"); + assert_eq!(body.items[0].data.as_slice(), b"r1"); + + let acked = ack(&client, &base, route, &["NA0642_REDELIVER_1"]).await; + assert_eq!(acked.status(), ReqStatus::OK); + let acked: AckResp = acked.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(acked.acked, 1); + let after = pull_legacy(&client, &base, route, 1).await; + assert_eq!(after.status(), ReqStatus::NO_CONTENT); + + handle.abort(); +} + +#[tokio::test(flavor = "current_thread")] +async fn ack_only_deletes_leased_copies() { + // Never-pulled messages cannot be acked away; with duplicate msg_ids + // (NA-0275: both copies enqueue) only the delivered copy dies. + let (base, handle) = spawn_server(60).await; + let client = reqwest::Client::new(); + let route = "NA0642_ACK_LEASED_ONLY"; + + let first = push( + &client, + &base, + route, + Some("NA0642_DUP_ID"), + b"copy-1".to_vec(), + ) + .await; + assert_eq!(first.status(), ReqStatus::OK); + let second = push( + &client, + &base, + route, + Some("NA0642_DUP_ID"), + b"copy-2".to_vec(), + ) + .await; + assert_eq!(second.status(), ReqStatus::OK); + + // Ack before any pull: nothing is leased, nothing may be deleted. + let premature = ack(&client, &base, route, &["NA0642_DUP_ID"]).await; + assert_eq!(premature.status(), ReqStatus::OK); + let premature: AckResp = premature.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(premature.acked, 0); + + // Lease exactly one copy, then ack the id: only the leased copy dies. + let leased = pull_lease(&client, &base, route, 1).await; + assert_eq!(leased.status(), ReqStatus::OK); + let leased_body: PullResp = leased.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(leased_body.items.len(), 1); + assert_eq!(leased_body.items[0].data.as_slice(), b"copy-1"); + + let acked = ack(&client, &base, route, &["NA0642_DUP_ID"]).await; + assert_eq!(acked.status(), ReqStatus::OK); + let acked: AckResp = acked.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(acked.acked, 1); + + // The undelivered duplicate survives and is still deliverable. + let survivor = pull_legacy(&client, &base, route, 2).await; + assert_eq!(survivor.status(), ReqStatus::OK); + let body: PullResp = survivor.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(body.items.len(), 1); + assert_eq!(body.items[0].id, "NA0642_DUP_ID"); + assert_eq!(body.items[0].data.as_slice(), b"copy-2"); + + handle.abort(); +} + +#[tokio::test(flavor = "current_thread")] +async fn ack_and_mode_inputs_fail_closed() { + let (base, handle) = spawn_server(60).await; + let client = reqwest::Client::new(); + let route = "NA0642_ACK_INPUTS"; + + let bad_mode = client + .get(format!("{base}/v1/pull?max=1&ack=bogus")) + .header(ROUTE_TOKEN_HEADER, route) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(bad_mode.status(), ReqStatus::BAD_REQUEST); + assert_eq!( + bad_mode.text().await.unwrap_or_else(|e| panic!("{e}")), + "ERR_BAD_ACK_MODE" + ); + + let empty_ids = ack(&client, &base, route, &[]).await; + assert_eq!(empty_ids.status(), ReqStatus::BAD_REQUEST); + assert_eq!( + empty_ids.text().await.unwrap_or_else(|e| panic!("{e}")), + "ERR_BAD_ACK_IDS" + ); + + let bad_body = client + .post(format!("{base}/v1/pull/ack")) + .header(ROUTE_TOKEN_HEADER, route) + .body("not-json") + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(bad_body.status(), ReqStatus::BAD_REQUEST); + assert_eq!( + bad_body.text().await.unwrap_or_else(|e| panic!("{e}")), + "ERR_BAD_ACK_BODY" + ); + + let missing_route = client + .post(format!("{base}/v1/pull/ack")) + .json(&serde_json::json!({ "ids": ["x"] })) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(missing_route.status(), ReqStatus::BAD_REQUEST); + assert_eq!( + missing_route.text().await.unwrap_or_else(|e| panic!("{e}")), + "ERR_MISSING_ROUTE_TOKEN" + ); + + handle.abort(); +} + +#[tokio::test(flavor = "current_thread")] +async fn ack_is_scoped_to_the_route() { + // An ack on route B must not delete route A's leased message. + let (base, handle) = spawn_server(60).await; + let client = reqwest::Client::new(); + + let accepted = push( + &client, + &base, + "NA0642_ACK_ROUTE_A", + Some("NA0642_CROSS_ID"), + b"a-copy".to_vec(), + ) + .await; + assert_eq!(accepted.status(), ReqStatus::OK); + let leased = pull_lease(&client, &base, "NA0642_ACK_ROUTE_A", 1).await; + assert_eq!(leased.status(), ReqStatus::OK); + + let cross = ack(&client, &base, "NA0642_ACK_ROUTE_B", &["NA0642_CROSS_ID"]).await; + assert_eq!(cross.status(), ReqStatus::OK); + let cross: AckResp = cross.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(cross.acked, 0); + + // Still deletable by the right route. + let acked = ack(&client, &base, "NA0642_ACK_ROUTE_A", &["NA0642_CROSS_ID"]).await; + assert_eq!(acked.status(), ReqStatus::OK); + let acked: AckResp = acked.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(acked.acked, 1); + + handle.abort(); +} diff --git a/tests/na0642_backward_compat.rs b/tests/na0642_backward_compat.rs new file mode 100644 index 0000000..b673a81 --- /dev/null +++ b/tests/na0642_backward_compat.rs @@ -0,0 +1,101 @@ +// NA-0642 backward-compat guard: the legacy pull contract (no ack parameter) +// must stay byte-identical for the CURRENT non-acking qsc client — same +// delete-on-deliver semantics, same JSON shape with exactly the same fields, +// same 204 behavior. The NA-0640 full-stack e2e relies on this at pin-bump +// time. + +use qsl_server::{app, AppState, Limits}; +use reqwest::StatusCode as ReqStatus; +use serde_json::Value; +use tokio::net::TcpListener; + +const ROUTE_TOKEN_HEADER: &str = "X-QSL-Route-Token"; +const MSG_ID_HEADER: &str = "X-Msg-Id"; + +async fn spawn_server() -> (String, tokio::task::JoinHandle<()>) { + let state = AppState::new_with_auth(Limits::new(1024 * 1024, 8).unwrap(), None); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .unwrap_or_else(|e| panic!("{e}")); + let addr = listener.local_addr().unwrap_or_else(|e| panic!("{e}")); + let handle = tokio::spawn(async move { + axum::serve(listener, app(state)) + .await + .unwrap_or_else(|e| panic!("{e}")); + }); + (format!("http://{addr}"), handle) +} + +#[tokio::test(flavor = "current_thread")] +async fn legacy_pull_shape_and_semantics_are_unchanged() { + let (base, handle) = spawn_server().await; + let client = reqwest::Client::new(); + let route = "NA0642_COMPAT_ROUTE"; + + let push = client + .post(format!("{base}/v1/push")) + .header(ROUTE_TOKEN_HEADER, route) + .header(MSG_ID_HEADER, "NA0642_COMPAT_MSG") + .body(b"compat-payload".to_vec()) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(push.status(), ReqStatus::OK); + // Push response shape: exactly {"id": ...}. + let push_body: Value = push.json().await.unwrap_or_else(|e| panic!("{e}")); + let push_obj = push_body + .as_object() + .unwrap_or_else(|| panic!("not object")); + assert_eq!(push_obj.len(), 1); + assert_eq!( + push_obj.get("id").and_then(Value::as_str), + Some("NA0642_COMPAT_MSG") + ); + + // Legacy pull (no ack parameter): today's exact contract. + let pull = client + .get(format!("{base}/v1/pull?max=2")) + .header(ROUTE_TOKEN_HEADER, route) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(pull.status(), ReqStatus::OK); + let pull_body: Value = pull.json().await.unwrap_or_else(|e| panic!("{e}")); + let pull_obj = pull_body + .as_object() + .unwrap_or_else(|| panic!("not object")); + // Top level: exactly {"items": [...]}. + assert_eq!(pull_obj.len(), 1); + let items = pull_obj + .get("items") + .and_then(Value::as_array) + .unwrap_or_else(|| panic!("items missing")); + assert_eq!(items.len(), 1); + // Item: exactly {"id": ..., "data": [...]} — no new fields leak into the + // legacy response. + let item = items[0].as_object().unwrap_or_else(|| panic!("not object")); + assert_eq!(item.len(), 2); + assert_eq!( + item.get("id").and_then(Value::as_str), + Some("NA0642_COMPAT_MSG") + ); + let data: Vec = item + .get("data") + .and_then(Value::as_array) + .unwrap_or_else(|| panic!("data missing")) + .iter() + .map(|v| v.as_u64().unwrap_or_else(|| panic!("not byte")) as u8) + .collect(); + assert_eq!(data.as_slice(), b"compat-payload"); + + // Delete-on-deliver: a non-acking client drains without any ack. + let drained = client + .get(format!("{base}/v1/pull?max=1")) + .header(ROUTE_TOKEN_HEADER, route) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(drained.status(), ReqStatus::NO_CONTENT); + + handle.abort(); +} diff --git a/tests/na0642_concurrency.rs b/tests/na0642_concurrency.rs new file mode 100644 index 0000000..488041c --- /dev/null +++ b/tests/na0642_concurrency.rs @@ -0,0 +1,201 @@ +// NA-0642 contended-path coverage: the storage layer moved from a process +// mutex over a HashMap to SQLite behind a connection mutex; these tests pin +// the no-loss/no-duplication guarantee under concurrent pushers and pullers +// (previously untested). + +use qsl_server::{app, AppState, Limits, ResourceControls, StoreConfig}; +use reqwest::StatusCode as ReqStatus; +use serde::Deserialize; +use std::collections::HashSet; +use tokio::net::TcpListener; + +const ROUTE_TOKEN_HEADER: &str = "X-QSL-Route-Token"; +const MSG_ID_HEADER: &str = "X-Msg-Id"; + +#[derive(Deserialize)] +struct PullItem { + id: String, +} + +#[derive(Deserialize)] +struct PullResp { + items: Vec, +} + +async fn spawn_server() -> (String, tokio::task::JoinHandle<()>) { + let state = AppState::new_with_auth_controls_and_store( + Limits::new(1024, 257).unwrap(), + ResourceControls::new(8, 257, 4096).unwrap(), + None, + StoreConfig::default(), + ) + .unwrap_or_else(|e| panic!("{e}")); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .unwrap_or_else(|e| panic!("{e}")); + let addr = listener.local_addr().unwrap_or_else(|e| panic!("{e}")); + let handle = tokio::spawn(async move { + axum::serve(listener, app(state)) + .await + .unwrap_or_else(|e| panic!("{e}")); + }); + (format!("http://{addr}"), handle) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_pushes_and_legacy_pulls_lose_and_duplicate_nothing() { + let (base, handle) = spawn_server().await; + let route = "NA0642_CONTENTION_LEGACY"; + const PUSHERS: usize = 8; + const PER_PUSHER: usize = 25; + + let mut push_tasks = Vec::new(); + for p in 0..PUSHERS { + let base = base.clone(); + push_tasks.push(tokio::spawn(async move { + let client = reqwest::Client::new(); + for i in 0..PER_PUSHER { + let id = format!("NA0642_C_{p}_{i}"); + let resp = client + .post(format!("{base}/v1/push")) + .header(ROUTE_TOKEN_HEADER, route) + .header(MSG_ID_HEADER, id.as_str()) + .body(id.clone().into_bytes()) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(resp.status(), ReqStatus::OK, "push {id} failed"); + } + })); + } + for task in push_tasks { + task.await.unwrap_or_else(|e| panic!("{e}")); + } + + // Concurrent legacy pullers drain the route; every message must be + // delivered exactly once across all pullers. + const PULLERS: usize = 4; + let mut pull_tasks = Vec::new(); + for _ in 0..PULLERS { + let base = base.clone(); + pull_tasks.push(tokio::spawn(async move { + let client = reqwest::Client::new(); + let mut got: Vec = Vec::new(); + let mut consecutive_empty = 0; + while consecutive_empty < 3 { + let resp = client + .get(format!("{base}/v1/pull?max=5")) + .header(ROUTE_TOKEN_HEADER, route) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + match resp.status() { + ReqStatus::OK => { + consecutive_empty = 0; + let body: PullResp = resp.json().await.unwrap_or_else(|e| panic!("{e}")); + got.extend(body.items.into_iter().map(|i| i.id)); + } + ReqStatus::NO_CONTENT => consecutive_empty += 1, + other => panic!("unexpected pull status {other}"), + } + } + got + })); + } + let mut all: Vec = Vec::new(); + for task in pull_tasks { + all.extend(task.await.unwrap_or_else(|e| panic!("{e}"))); + } + + let expected: HashSet = (0..PUSHERS) + .flat_map(|p| (0..PER_PUSHER).map(move |i| format!("NA0642_C_{p}_{i}"))) + .collect(); + let delivered: HashSet = all.iter().cloned().collect(); + assert_eq!(all.len(), expected.len(), "duplicate delivery detected"); + assert_eq!(delivered, expected, "lost or foreign messages detected"); + + handle.abort(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_lease_pulls_never_double_deliver_within_the_lease() { + let (base, handle) = spawn_server().await; + let route = "NA0642_CONTENTION_LEASE"; + const TOTAL: usize = 50; + + let client = reqwest::Client::new(); + for i in 0..TOTAL { + let id = format!("NA0642_L_{i}"); + let resp = client + .post(format!("{base}/v1/push")) + .header(ROUTE_TOKEN_HEADER, route) + .header(MSG_ID_HEADER, id.as_str()) + .body(id.clone().into_bytes()) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(resp.status(), ReqStatus::OK); + } + + // Four concurrent lease-pullers race over the same route; within the + // lease window no message may be handed to two pullers. + const PULLERS: usize = 4; + let mut tasks = Vec::new(); + for _ in 0..PULLERS { + let base = base.clone(); + tasks.push(tokio::spawn(async move { + let client = reqwest::Client::new(); + let mut got: Vec = Vec::new(); + loop { + let resp = client + .get(format!("{base}/v1/pull?max=5&ack=lease")) + .header(ROUTE_TOKEN_HEADER, route) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + match resp.status() { + ReqStatus::OK => { + let body: PullResp = resp.json().await.unwrap_or_else(|e| panic!("{e}")); + got.extend(body.items.into_iter().map(|i| i.id)); + } + ReqStatus::NO_CONTENT => break, + other => panic!("unexpected pull status {other}"), + } + } + got + })); + } + let mut all: Vec = Vec::new(); + for task in tasks { + all.extend(task.await.unwrap_or_else(|e| panic!("{e}"))); + } + let unique: HashSet = all.iter().cloned().collect(); + assert_eq!(all.len(), unique.len(), "a message was leased twice"); + assert_eq!(unique.len(), TOTAL, "some messages were never leased"); + + // Ack everything; the route must drain completely. + let ids: Vec = all; + let resp = client + .post(format!("{base}/v1/pull/ack")) + .header(ROUTE_TOKEN_HEADER, route) + .json(&serde_json::json!({ "ids": ids })) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(resp.status(), ReqStatus::OK); + let acked: serde_json::Value = resp.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!( + acked.get("acked").and_then(|v| v.as_u64()), + Some(TOTAL as u64) + ); + + let after = client + .get(format!("{base}/v1/pull?max=1&ack=lease")) + .header(ROUTE_TOKEN_HEADER, route) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(after.status(), ReqStatus::NO_CONTENT); + + handle.abort(); +} diff --git a/tests/na0642_durability_restart.rs b/tests/na0642_durability_restart.rs new file mode 100644 index 0000000..ebd0d1c --- /dev/null +++ b/tests/na0642_durability_restart.rs @@ -0,0 +1,273 @@ +// NA-0642 restart-durability proof, operator-required form: the relay process +// is HARD-KILLED (SIGKILL via Child::kill on unix — no graceful shutdown, no +// flush-on-exit) between the push 200 and the restart, so surviving the +// restart demonstrates the synchronous=FULL fsync guarantee, not a graceful +// flush. Same discipline for the crash-between-pull-and-ack case: the server +// dies mid-lease and the leased message must reappear after lease expiry. + +use serde::Deserialize; +use std::{ + io::{BufRead, BufReader}, + process::{Child, Command, Stdio}, + sync::mpsc, + time::Duration, +}; + +const ROUTE_TOKEN_HEADER: &str = "X-QSL-Route-Token"; +const MSG_ID_HEADER: &str = "X-Msg-Id"; + +#[derive(Deserialize)] +struct PullItem { + id: String, + data: Vec, +} + +#[derive(Deserialize)] +struct PullResp { + items: Vec, +} + +#[derive(Deserialize)] +struct AckResp { + acked: usize, +} + +struct Relay { + child: Child, + base: String, +} + +impl Relay { + fn spawn(store_path: &str, extra_envs: &[(&str, &str)]) -> Self { + let mut command = Command::new(env!("CARGO_BIN_EXE_qsl-server")); + command + .env_clear() + .env("RUST_LOG", "info") + .env("BIND_ADDR", "127.0.0.1") + .env("PORT", "0") + .env("STORE_PATH", store_path) + .stdout(Stdio::piped()) + .stderr(Stdio::null()); + for (name, value) in extra_envs { + command.env(name, value); + } + let mut child = command.spawn().unwrap_or_else(|e| panic!("{e}")); + let stdout = child.stdout.take().unwrap_or_else(|| panic!("no stdout")); + let (tx, rx) = mpsc::channel::(); + // Keep draining stdout for the process lifetime so logging never + // blocks on a full pipe; only the first "listening on" line matters. + std::thread::spawn(move || { + let reader = BufReader::new(stdout); + let mut sent = false; + for line in reader.lines() { + let Ok(line) = line else { break }; + if !sent { + if let Some(idx) = line.find("listening on ") { + let addr = line[idx + "listening on ".len()..].trim().to_string(); + let _ = tx.send(addr); + sent = true; + } + } + } + }); + let addr = rx + .recv_timeout(Duration::from_secs(10)) + .unwrap_or_else(|e| panic!("relay did not report listen address: {e}")); + Self { + child, + base: format!("http://{addr}"), + } + } + + /// SIGKILL — the process gets no chance to flush or shut down cleanly. + fn hard_kill(mut self) { + self.child.kill().unwrap_or_else(|e| panic!("{e}")); + self.child.wait().unwrap_or_else(|e| panic!("{e}")); + } +} + +fn temp_store(tag: &str) -> String { + let dir = std::env::temp_dir().join(format!( + "na0642-{}-{}-{}", + tag, + std::process::id(), + uuid_ish() + )); + std::fs::create_dir_all(&dir).unwrap_or_else(|e| panic!("{e}")); + dir.join("relay.db").to_string_lossy().into_owned() +} + +fn uuid_ish() -> u128 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() +} + +async fn push( + client: &reqwest::Client, + base: &str, + route: &str, + msg_id: &str, + body: Vec, +) -> reqwest::Response { + client + .post(format!("{base}/v1/push")) + .header(ROUTE_TOKEN_HEADER, route) + .header(MSG_ID_HEADER, msg_id) + .body(body) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")) +} + +async fn pull( + client: &reqwest::Client, + base: &str, + route: &str, + max: usize, + lease: bool, +) -> reqwest::Response { + let url = if lease { + format!("{base}/v1/pull?max={max}&ack=lease") + } else { + format!("{base}/v1/pull?max={max}") + }; + client + .get(url) + .header(ROUTE_TOKEN_HEADER, route) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")) +} + +#[tokio::test(flavor = "current_thread")] +async fn pushed_message_survives_hard_kill_and_restart() { + let store = temp_store("restart"); + let route = "NA0642_RESTART_ROUTE"; + let payload: Vec = (0..32_768u32).map(|i| (i % 251) as u8).collect(); + + let relay = Relay::spawn(&store, &[]); + let client = reqwest::Client::new(); + let accepted = push( + &client, + &relay.base, + route, + "NA0642_RESTART_MSG", + payload.clone(), + ) + .await; + assert_eq!(accepted.status(), reqwest::StatusCode::OK); + + // HARD KILL immediately after the 200: no graceful shutdown, no flush. + relay.hard_kill(); + + let relay = Relay::spawn(&store, &[]); + let delivered = pull(&client, &relay.base, route, 2, false).await; + assert_eq!(delivered.status(), reqwest::StatusCode::OK); + let body: PullResp = delivered.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(body.items.len(), 1); + assert_eq!(body.items[0].id, "NA0642_RESTART_MSG"); + assert_eq!(body.items[0].data, payload, "payload not byte-identical"); + + // Delete-on-deliver still holds after the durable round-trip. + let after = pull(&client, &relay.base, route, 1, false).await; + assert_eq!(after.status(), reqwest::StatusCode::NO_CONTENT); + + relay.hard_kill(); + let _ = std::fs::remove_dir_all(std::path::Path::new(&store).parent().unwrap()); +} + +#[tokio::test(flavor = "current_thread")] +async fn leased_message_survives_hard_kill_and_reappears_after_lease_expiry() { + let store = temp_store("lease-crash"); + let route = "NA0642_LEASE_CRASH_ROUTE"; + let lease_env = [("PULL_LEASE_SECS", "5")]; + + let relay = Relay::spawn(&store, &lease_env); + let client = reqwest::Client::new(); + let accepted = push( + &client, + &relay.base, + route, + "NA0642_LEASE_CRASH_MSG", + b"crash-window-payload".to_vec(), + ) + .await; + assert_eq!(accepted.status(), reqwest::StatusCode::OK); + + let leased = pull(&client, &relay.base, route, 1, true).await; + assert_eq!(leased.status(), reqwest::StatusCode::OK); + let leased_body: PullResp = leased.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(leased_body.items.len(), 1); + + // HARD KILL between the acknowledged-pull and the ack: the crash window + // the design exists for. The message must NOT be lost. + relay.hard_kill(); + + let relay = Relay::spawn(&store, &lease_env); + // The lease itself survived the restart: while it is live the message + // stays in-flight and invisible. + let still_leased = pull(&client, &relay.base, route, 1, true).await; + assert_eq!(still_leased.status(), reqwest::StatusCode::NO_CONTENT); + + // After lease expiry the message reappears, byte-identical. + tokio::time::sleep(Duration::from_secs(6)).await; + let redelivered = pull(&client, &relay.base, route, 1, true).await; + assert_eq!(redelivered.status(), reqwest::StatusCode::OK); + let body: PullResp = redelivered.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(body.items.len(), 1); + assert_eq!(body.items[0].id, "NA0642_LEASE_CRASH_MSG"); + assert_eq!(body.items[0].data.as_slice(), b"crash-window-payload"); + + // This time the ack lands: the message is gone for good. + let acked = client + .post(format!("{}/v1/pull/ack", relay.base)) + .header(ROUTE_TOKEN_HEADER, route) + .json(&serde_json::json!({ "ids": ["NA0642_LEASE_CRASH_MSG"] })) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(acked.status(), reqwest::StatusCode::OK); + let acked: AckResp = acked.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(acked.acked, 1); + let drained = pull(&client, &relay.base, route, 1, true).await; + assert_eq!(drained.status(), reqwest::StatusCode::NO_CONTENT); + + relay.hard_kill(); + let _ = std::fs::remove_dir_all(std::path::Path::new(&store).parent().unwrap()); +} + +#[tokio::test(flavor = "current_thread")] +async fn legacy_delivery_does_not_resurrect_after_restart() { + // Negative control: a message delivered via the legacy delete-on-pull + // contract must NOT come back after a hard kill + restart (the store + // must be durable for queued messages, forgetful for delivered ones). + let store = temp_store("no-resurrect"); + let route = "NA0642_NO_RESURRECT_ROUTE"; + + let relay = Relay::spawn(&store, &[]); + let client = reqwest::Client::new(); + let accepted = push( + &client, + &relay.base, + route, + "NA0642_NO_RESURRECT_MSG", + b"delivered-then-gone".to_vec(), + ) + .await; + assert_eq!(accepted.status(), reqwest::StatusCode::OK); + + let delivered = pull(&client, &relay.base, route, 1, false).await; + assert_eq!(delivered.status(), reqwest::StatusCode::OK); + + relay.hard_kill(); + + let relay = Relay::spawn(&store, &[]); + let after = pull(&client, &relay.base, route, 1, false).await; + assert_eq!(after.status(), reqwest::StatusCode::NO_CONTENT); + + relay.hard_kill(); + let _ = std::fs::remove_dir_all(std::path::Path::new(&store).parent().unwrap()); +} diff --git a/tests/route_lifecycle_ttl.rs b/tests/na0642_retention_lifecycle.rs similarity index 51% rename from tests/route_lifecycle_ttl.rs rename to tests/na0642_retention_lifecycle.rs index a44cd00..36c9325 100644 --- a/tests/route_lifecycle_ttl.rs +++ b/tests/na0642_retention_lifecycle.rs @@ -1,4 +1,8 @@ -use qsl_server::{app, AppState, Limits, ResourceControls}; +// NA-0642: message lifetime is governed by the store's retention TTL; the +// 5-minute idle-route discard (NA-0281 route_lifecycle_ttl) is retired. The +// drain-release contracts from NA-0281 are carried forward here unchanged. + +use qsl_server::{app, AppState, Limits, ResourceControls, StoreConfig}; use reqwest::StatusCode as ReqStatus; use serde::Deserialize; use std::time::Duration; @@ -6,8 +10,6 @@ use tokio::net::TcpListener; const ROUTE_TOKEN_HEADER: &str = "X-QSL-Route-Token"; const MSG_ID_HEADER: &str = "X-Msg-Id"; -const TTL_MS: usize = 25; -const TTL_MARGIN: Duration = Duration::from_millis(75); #[derive(Deserialize)] struct PullItem { @@ -20,35 +22,42 @@ struct PullResp { items: Vec, } -async fn spawn_server_with_auth( +fn short_retention() -> StoreConfig { + StoreConfig { + retention_ttl_secs: 1, + ..StoreConfig::default() + } +} + +async fn spawn_server( limits: Limits, controls: ResourceControls, - relay_token: Option<&str>, -) -> (String, tokio::task::JoinHandle<()>) { - let state = - AppState::new_with_auth_and_controls(limits, controls, relay_token.map(str::to_string)); + store: StoreConfig, +) -> (String, AppState, tokio::task::JoinHandle<()>) { + let state = AppState::new_with_auth_controls_and_store(limits, controls, None, store) + .unwrap_or_else(|e| panic!("{e}")); let listener = TcpListener::bind("127.0.0.1:0") .await .unwrap_or_else(|e| panic!("{e}")); let addr = listener.local_addr().unwrap_or_else(|e| panic!("{e}")); + let served = state.clone(); let handle = tokio::spawn(async move { - axum::serve(listener, app(state)) + axum::serve(listener, app(served)) .await .unwrap_or_else(|e| panic!("{e}")); }); - (format!("http://{addr}"), handle) + (format!("http://{addr}"), state, handle) } -fn ttl_controls(max_route_count: usize, push_rate_burst: usize) -> ResourceControls { - ResourceControls::new_with_route_idle_ttl_ms(max_route_count, push_rate_burst, 0, TTL_MS) - .unwrap() +fn rate_controls(max_route_count: usize, push_rate_burst: usize) -> ResourceControls { + ResourceControls::new_with_route_idle_ttl_ms(max_route_count, push_rate_burst, 0, 25) + .unwrap_or_else(|e| panic!("{e}")) } async fn push( client: &reqwest::Client, base: &str, route_token: &str, - auth_token: Option<&str>, msg_id: Option<&str>, body: impl Into>, ) -> reqwest::Response { @@ -56,9 +65,6 @@ async fn push( .post(format!("{base}/v1/push")) .header(ROUTE_TOKEN_HEADER, route_token) .body(body.into()); - if let Some(token) = auth_token { - request = request.header("Authorization", format!("Bearer {token}")); - } if let Some(id) = msg_id { request = request.header(MSG_ID_HEADER, id); } @@ -69,39 +75,40 @@ async fn pull( client: &reqwest::Client, base: &str, route_token: &str, - auth_token: Option<&str>, max: usize, ) -> reqwest::Response { - let mut request = client + client .get(format!("{base}/v1/pull?max={max}")) - .header(ROUTE_TOKEN_HEADER, route_token); - if let Some(token) = auth_token { - request = request.header("Authorization", format!("Bearer {token}")); - } - request.send().await.unwrap_or_else(|e| panic!("{e}")) + .header(ROUTE_TOKEN_HEADER, route_token) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")) } #[tokio::test(flavor = "current_thread")] -async fn unknown_pull_does_not_create_route_slot() { - let (base, handle) = - spawn_server_with_auth(Limits::new(128, 4).unwrap(), ttl_controls(1, 4), None).await; +async fn pull_unknown_route_returns_204_then_push_works() { + let (base, _state, handle) = spawn_server( + Limits::new(128, 4).unwrap(), + rate_controls(4, 4), + StoreConfig::default(), + ) + .await; let client = reqwest::Client::new(); - let unknown = pull(&client, &base, "NA0281_UNKNOWN_PULL", None, 1).await; + let unknown = pull(&client, &base, "NA0281_AFTER_UNKNOWN_PULL", 1).await; assert_eq!(unknown.status(), ReqStatus::NO_CONTENT); let accepted = push( &client, &base, "NA0281_AFTER_UNKNOWN_PULL", - None, Some("NA0281_AFTER_UNKNOWN_PULL_ID"), b"accepted-after-unknown".to_vec(), ) .await; assert_eq!(accepted.status(), ReqStatus::OK); - let delivered = pull(&client, &base, "NA0281_AFTER_UNKNOWN_PULL", None, 1).await; + let delivered = pull(&client, &base, "NA0281_AFTER_UNKNOWN_PULL", 1).await; assert_eq!(delivered.status(), ReqStatus::OK); let body: PullResp = delivered.json().await.unwrap_or_else(|e| panic!("{e}")); assert_eq!(body.items.len(), 1); @@ -113,15 +120,18 @@ async fn unknown_pull_does_not_create_route_slot() { #[tokio::test(flavor = "current_thread")] async fn drain_to_empty_releases_route_slot_and_rate_bucket() { - let (base, handle) = - spawn_server_with_auth(Limits::new(128, 4).unwrap(), ttl_controls(1, 1), None).await; + let (base, _state, handle) = spawn_server( + Limits::new(128, 4).unwrap(), + rate_controls(1, 1), + StoreConfig::default(), + ) + .await; let client = reqwest::Client::new(); let first = push( &client, &base, "NA0281_DRAIN_RATE_ROUTE", - None, Some("NA0281_DRAIN_RATE_FIRST"), b"first".to_vec(), ) @@ -132,7 +142,6 @@ async fn drain_to_empty_releases_route_slot_and_rate_bucket() { &client, &base, "NA0281_DRAIN_RATE_ROUTE", - None, Some("NA0281_DRAIN_RATE_LIMITED"), b"limited".to_vec(), ) @@ -143,21 +152,20 @@ async fn drain_to_empty_releases_route_slot_and_rate_bucket() { "ERR_RATE_LIMITED" ); - let drained = pull(&client, &base, "NA0281_DRAIN_RATE_ROUTE", None, 1).await; + let drained = pull(&client, &base, "NA0281_DRAIN_RATE_ROUTE", 1).await; assert_eq!(drained.status(), ReqStatus::OK); let reused = push( &client, &base, "NA0281_DRAIN_RATE_ROUTE", - None, Some("NA0281_DRAIN_RATE_REUSED"), b"reused".to_vec(), ) .await; assert_eq!(reused.status(), ReqStatus::OK); - let delivered = pull(&client, &base, "NA0281_DRAIN_RATE_ROUTE", None, 1).await; + let delivered = pull(&client, &base, "NA0281_DRAIN_RATE_ROUTE", 1).await; assert_eq!(delivered.status(), ReqStatus::OK); let body: PullResp = delivered.json().await.unwrap_or_else(|e| panic!("{e}")); assert_eq!(body.items.len(), 1); @@ -168,7 +176,6 @@ async fn drain_to_empty_releases_route_slot_and_rate_bucket() { &client, &base, "NA0281_DRAIN_RELEASED_SLOT", - None, Some("NA0281_DRAIN_SLOT_ID"), b"slot".to_vec(), ) @@ -179,177 +186,206 @@ async fn drain_to_empty_releases_route_slot_and_rate_bucket() { } #[tokio::test(flavor = "current_thread")] -async fn idle_route_ttl_releases_capacity() { - let (base, handle) = - spawn_server_with_auth(Limits::new(128, 4).unwrap(), ttl_controls(1, 2), None).await; +async fn retention_ttl_expires_undelivered_messages() { + let (base, _state, handle) = spawn_server( + Limits::new(128, 4).unwrap(), + rate_controls(4, 4), + short_retention(), + ) + .await; let client = reqwest::Client::new(); let stale = push( &client, &base, - "NA0281_TTL_STALE_ROUTE", - None, - Some("NA0281_TTL_STALE_ID"), + "NA0642_RETENTION_ROUTE", + Some("NA0642_RETENTION_STALE_ID"), b"stale".to_vec(), ) .await; assert_eq!(stale.status(), ReqStatus::OK); - tokio::time::sleep(TTL_MARGIN).await; + tokio::time::sleep(Duration::from_millis(1500)).await; + let expired = pull(&client, &base, "NA0642_RETENTION_ROUTE", 1).await; + assert_eq!(expired.status(), ReqStatus::NO_CONTENT); + + // Non-vacuity: the pull path still delivers on the same server after the + // expiry window — the stale message vanished because of retention, not + // because delivery broke. let fresh = push( &client, &base, - "NA0281_TTL_FRESH_ROUTE", - None, - Some("NA0281_TTL_FRESH_ID"), + "NA0642_RETENTION_ROUTE", + Some("NA0642_RETENTION_FRESH_ID"), b"fresh".to_vec(), ) .await; assert_eq!(fresh.status(), ReqStatus::OK); - - let expired = pull(&client, &base, "NA0281_TTL_STALE_ROUTE", None, 1).await; - assert_eq!(expired.status(), ReqStatus::NO_CONTENT); - - let delivered = pull(&client, &base, "NA0281_TTL_FRESH_ROUTE", None, 1).await; + let delivered = pull(&client, &base, "NA0642_RETENTION_ROUTE", 1).await; assert_eq!(delivered.status(), ReqStatus::OK); let body: PullResp = delivered.json().await.unwrap_or_else(|e| panic!("{e}")); assert_eq!(body.items.len(), 1); - assert_eq!(body.items[0].id, "NA0281_TTL_FRESH_ID"); - assert_eq!(body.items[0].data.as_slice(), b"fresh"); + assert_eq!(body.items[0].id, "NA0642_RETENTION_FRESH_ID"); handle.abort(); } #[tokio::test(flavor = "current_thread")] -async fn expired_route_does_not_return_stale_message() { - let (base, handle) = - spawn_server_with_auth(Limits::new(128, 4).unwrap(), ttl_controls(2, 2), None).await; +async fn default_retention_does_not_expire_quickly() { + // Negative control for the expiry test: with the default TTL a message + // must survive the same sleep the short-TTL test uses. + let (base, _state, handle) = spawn_server( + Limits::new(128, 4).unwrap(), + rate_controls(4, 4), + StoreConfig::default(), + ) + .await; let client = reqwest::Client::new(); - let stale = push( + let accepted = push( &client, &base, - "NA0281_TTL_NO_STALE_ROUTE", - None, - Some("NA0281_TTL_NO_STALE_ID"), - b"stale-message".to_vec(), + "NA0642_RETENTION_CONTROL_ROUTE", + Some("NA0642_RETENTION_CONTROL_ID"), + b"survives".to_vec(), ) .await; - assert_eq!(stale.status(), ReqStatus::OK); + assert_eq!(accepted.status(), ReqStatus::OK); - tokio::time::sleep(TTL_MARGIN).await; + tokio::time::sleep(Duration::from_millis(1500)).await; - let expired = pull(&client, &base, "NA0281_TTL_NO_STALE_ROUTE", None, 1).await; - assert_eq!(expired.status(), ReqStatus::NO_CONTENT); + let delivered = pull(&client, &base, "NA0642_RETENTION_CONTROL_ROUTE", 1).await; + assert_eq!(delivered.status(), ReqStatus::OK); + let body: PullResp = delivered.json().await.unwrap_or_else(|e| panic!("{e}")); + assert_eq!(body.items.len(), 1); + assert_eq!(body.items[0].id, "NA0642_RETENTION_CONTROL_ID"); + assert_eq!(body.items[0].data.as_slice(), b"survives"); handle.abort(); } #[tokio::test(flavor = "current_thread")] -async fn expired_route_releases_rate_bucket() { - let (base, handle) = - spawn_server_with_auth(Limits::new(128, 4).unwrap(), ttl_controls(2, 1), None).await; +async fn retention_sweep_entry_point_expires_and_reports() { + // The periodic background sweep uses this entry point; quiet relays must + // expire without any push/pull traffic triggering the lazy path. + let (base, state, handle) = spawn_server( + Limits::new(128, 8).unwrap(), + rate_controls(4, 8), + short_retention(), + ) + .await; let client = reqwest::Client::new(); - let accepted = push( - &client, - &base, - "NA0281_TTL_RATE_ROUTE", - None, - Some("NA0281_TTL_RATE_STALE_ID"), - b"stale".to_vec(), + for id in ["NA0642_SWEEP_A", "NA0642_SWEEP_B"] { + let accepted = push( + &client, + &base, + "NA0642_SWEEP_ROUTE", + Some(id), + b"sweepable".to_vec(), + ) + .await; + assert_eq!(accepted.status(), ReqStatus::OK); + } + + tokio::time::sleep(Duration::from_millis(1500)).await; + + let stats = state.run_retention_sweep(); + assert_eq!(stats.expired_messages, 2); + assert_eq!(stats.expired_routes.len(), 1); + assert_eq!(stats.removed_route_keys.len(), 1); + + let after = pull(&client, &base, "NA0642_SWEEP_ROUTE", 2).await; + assert_eq!(after.status(), ReqStatus::NO_CONTENT); + + handle.abort(); +} + +#[tokio::test(flavor = "current_thread")] +async fn push_after_expiry_does_not_resurrect_old_messages() { + let (base, _state, handle) = spawn_server( + Limits::new(128, 4).unwrap(), + rate_controls(4, 4), + short_retention(), ) .await; - assert_eq!(accepted.status(), ReqStatus::OK); + let client = reqwest::Client::new(); - let limited = push( + let stale = push( &client, &base, - "NA0281_TTL_RATE_ROUTE", - None, - Some("NA0281_TTL_RATE_LIMITED_ID"), - b"limited".to_vec(), + "NA0642_REUSE_ROUTE", + Some("NA0642_REUSE_STALE_ID"), + b"stale".to_vec(), ) .await; - assert_eq!(limited.status(), ReqStatus::TOO_MANY_REQUESTS); + assert_eq!(stale.status(), ReqStatus::OK); - tokio::time::sleep(TTL_MARGIN).await; + tokio::time::sleep(Duration::from_millis(1500)).await; let fresh = push( &client, &base, - "NA0281_TTL_RATE_ROUTE", - None, - Some("NA0281_TTL_RATE_FRESH_ID"), + "NA0642_REUSE_ROUTE", + Some("NA0642_REUSE_FRESH_ID"), b"fresh".to_vec(), ) .await; assert_eq!(fresh.status(), ReqStatus::OK); - let delivered = pull(&client, &base, "NA0281_TTL_RATE_ROUTE", None, 2).await; + let delivered = pull(&client, &base, "NA0642_REUSE_ROUTE", 2).await; assert_eq!(delivered.status(), ReqStatus::OK); let body: PullResp = delivered.json().await.unwrap_or_else(|e| panic!("{e}")); assert_eq!(body.items.len(), 1); - assert_eq!(body.items[0].id, "NA0281_TTL_RATE_FRESH_ID"); + assert_eq!(body.items[0].id, "NA0642_REUSE_FRESH_ID"); assert_eq!(body.items[0].data.as_slice(), b"fresh"); handle.abort(); } #[tokio::test(flavor = "current_thread")] -async fn push_after_expiry_does_not_resurrect_old_messages() { - let (base, handle) = - spawn_server_with_auth(Limits::new(128, 4).unwrap(), ttl_controls(2, 2), None).await; +async fn retention_expiry_releases_route_slot() { + // Replaces the retired idle-TTL capacity release: an expired route frees + // its MAX_ROUTE_COUNT slot via the lazy sweep on the next push. + let (base, _state, handle) = spawn_server( + Limits::new(128, 4).unwrap(), + rate_controls(1, 2), + short_retention(), + ) + .await; let client = reqwest::Client::new(); let stale = push( &client, &base, - "NA0281_TTL_REUSE_ROUTE", - None, - Some("NA0281_TTL_REUSE_STALE_ID"), + "NA0642_SLOT_STALE_ROUTE", + Some("NA0642_SLOT_STALE_ID"), b"stale".to_vec(), ) .await; assert_eq!(stale.status(), ReqStatus::OK); - tokio::time::sleep(TTL_MARGIN).await; + tokio::time::sleep(Duration::from_millis(1500)).await; let fresh = push( &client, &base, - "NA0281_TTL_REUSE_ROUTE", - None, - Some("NA0281_TTL_REUSE_FRESH_ID"), + "NA0642_SLOT_FRESH_ROUTE", + Some("NA0642_SLOT_FRESH_ID"), b"fresh".to_vec(), ) .await; assert_eq!(fresh.status(), ReqStatus::OK); - let delivered = pull(&client, &base, "NA0281_TTL_REUSE_ROUTE", None, 2).await; + let expired = pull(&client, &base, "NA0642_SLOT_STALE_ROUTE", 1).await; + assert_eq!(expired.status(), ReqStatus::NO_CONTENT); + + let delivered = pull(&client, &base, "NA0642_SLOT_FRESH_ROUTE", 1).await; assert_eq!(delivered.status(), ReqStatus::OK); let body: PullResp = delivered.json().await.unwrap_or_else(|e| panic!("{e}")); assert_eq!(body.items.len(), 1); - assert_eq!(body.items[0].id, "NA0281_TTL_REUSE_FRESH_ID"); - assert_eq!(body.items[0].data.as_slice(), b"fresh"); + assert_eq!(body.items[0].id, "NA0642_SLOT_FRESH_ID"); handle.abort(); } - -#[test] -fn route_ttl_config_and_docs_are_explicit() { - let controls = ResourceControls::new_with_route_idle_ttl_ms(4, 4, 0, TTL_MS).unwrap(); - assert_eq!(controls.route_idle_ttl.as_millis(), TTL_MS as u128); - assert!(ResourceControls::new_with_route_idle_ttl_ms(4, 4, 0, 0).is_err()); - - let readme = include_str!("../README.md"); - let inbox_contract = - include_str!("../docs/server/DOC-SRV-003_Relay_Inbox_Contract_v1.0.0_DRAFT.md"); - - for doc in [readme, inbox_contract] { - assert!(doc.contains("ROUTE_IDLE_TTL_MS")); - assert!(doc.contains("ERR_ROUTE_CAP")); - assert!(doc.contains("Time-based idle TTL")); - } -} diff --git a/tests/na0642_retention_logging.rs b/tests/na0642_retention_logging.rs new file mode 100644 index 0000000..cbd9f19 --- /dev/null +++ b/tests/na0642_retention_logging.rs @@ -0,0 +1,112 @@ +// NA-0642: replaces the retired route_lifecycle_ttl_logging redaction test. +// Retention-expiry logs must stay metadata-only: redacted channel id, counts, +// TTL — never the route token, auth token, or payload. + +use qsl_server::{app, AppState, Limits, ResourceControls, StoreConfig}; +use reqwest::StatusCode as ReqStatus; +use std::{ + io::Write, + sync::{Arc, Mutex}, + time::Duration, +}; +use tokio::net::TcpListener; +use tracing::subscriber::set_default; + +const ROUTE_TOKEN_HEADER: &str = "X-QSL-Route-Token"; +const MSG_ID_HEADER: &str = "X-Msg-Id"; + +#[derive(Clone)] +struct SharedWriter(Arc>>); + +impl Write for SharedWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let mut guard = self.0.lock().unwrap_or_else(|e| panic!("{e}")); + guard.extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[tokio::test(flavor = "current_thread")] +async fn retention_cleanup_logs_redact_route_auth_payload() { + let buf = Arc::new(Mutex::new(Vec::new())); + let writer = SharedWriter(buf.clone()); + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::INFO) + .with_ansi(false) + .with_writer(move || writer.clone()) + .finish(); + let _guard = set_default(subscriber); + + let auth_token = "NA0642_AUTH_TOKEN_SENTINEL_MUST_NOT_LEAK"; + let route_token = "NA0642_RETENTION_ROUTE_TOKEN_MUST_NOT_LEAK"; + let payload = b"NA0642_RETENTION_PAYLOAD_MUST_NOT_LEAK".to_vec(); + let controls = ResourceControls::new_with_route_idle_ttl_ms(2, 2, 0, 25).unwrap(); + let store = StoreConfig { + retention_ttl_secs: 1, + ..StoreConfig::default() + }; + let state = AppState::new_with_auth_controls_and_store( + Limits::new(128, 4).unwrap(), + controls, + Some(auth_token.to_string()), + store, + ) + .unwrap_or_else(|e| panic!("{e}")); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .unwrap_or_else(|e| panic!("{e}")); + let addr = listener.local_addr().unwrap_or_else(|e| panic!("{e}")); + let handle = tokio::spawn(async move { + axum::serve(listener, app(state)) + .await + .unwrap_or_else(|e| panic!("{e}")); + }); + let base = format!("http://{addr}"); + let client = reqwest::Client::new(); + + let accepted = client + .post(format!("{base}/v1/push")) + .header(ROUTE_TOKEN_HEADER, route_token) + .header(MSG_ID_HEADER, "NA0642_RETENTION_MSG_ID_NONSECRET_METADATA") + .header("Authorization", format!("Bearer {auth_token}")) + .body(payload) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(accepted.status(), ReqStatus::OK); + + tokio::time::sleep(Duration::from_millis(1500)).await; + + let expired = client + .get(format!("{base}/v1/pull?max=1")) + .header(ROUTE_TOKEN_HEADER, route_token) + .header("Authorization", format!("Bearer {auth_token}")) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(expired.status(), ReqStatus::NO_CONTENT); + + tokio::task::yield_now().await; + handle.abort(); + + let guard = buf.lock().unwrap_or_else(|e| panic!("{e}")); + let logs = String::from_utf8_lossy(&guard); + assert!(logs.contains("event=retention_expired")); + assert!(logs.contains("channel_id=")); + assert!(logs.contains("expired_messages=1")); + assert!(logs.contains("ttl_secs=1")); + + for forbidden in [ + auth_token, + route_token, + "Authorization", + "Bearer", + "NA0642_RETENTION_PAYLOAD_MUST_NOT_LEAK", + ] { + assert!(!logs.contains(forbidden), "logs leaked {forbidden}"); + } +} diff --git a/tests/na0642_store_privacy.rs b/tests/na0642_store_privacy.rs new file mode 100644 index 0000000..e58da44 --- /dev/null +++ b/tests/na0642_store_privacy.rs @@ -0,0 +1,81 @@ +// NA-0642 at-rest posture: raw route tokens are never persisted — the store +// keys routes by SHA-256(route_token). Payload bytes ARE stored verbatim +// (they are opaque E2EE ciphertext by contract), which doubles as the +// non-vacuity control proving we are reading the right file. + +use qsl_server::{app, AppState, Limits, ResourceControls, StoreConfig}; +use reqwest::StatusCode as ReqStatus; +use tokio::net::TcpListener; + +const ROUTE_TOKEN_HEADER: &str = "X-QSL-Route-Token"; + +fn contains_subslice(haystack: &[u8], needle: &[u8]) -> bool { + haystack.windows(needle.len()).any(|w| w == needle) +} + +#[tokio::test(flavor = "current_thread")] +async fn raw_route_token_never_touches_the_store_file() { + let dir = std::env::temp_dir().join(format!("na0642-privacy-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap_or_else(|e| panic!("{e}")); + let db_path = dir.join("relay.db"); + + let route_token = "NA0642_PRIVACY_ROUTE_TOKEN_MUST_NOT_TOUCH_DISK"; + let payload = b"NA0642_PRIVACY_OPAQUE_PAYLOAD_MARKER".to_vec(); + + let store = StoreConfig { + path: db_path.to_string_lossy().into_owned(), + ..StoreConfig::default() + }; + let state = AppState::new_with_auth_controls_and_store( + Limits::new(1024, 8).unwrap(), + ResourceControls::new(4, 8, 8).unwrap(), + None, + store, + ) + .unwrap_or_else(|e| panic!("{e}")); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .unwrap_or_else(|e| panic!("{e}")); + let addr = listener.local_addr().unwrap_or_else(|e| panic!("{e}")); + let handle = tokio::spawn(async move { + axum::serve(listener, app(state)) + .await + .unwrap_or_else(|e| panic!("{e}")); + }); + let base = format!("http://{addr}"); + let client = reqwest::Client::new(); + + let accepted = client + .post(format!("{base}/v1/push")) + .header(ROUTE_TOKEN_HEADER, route_token) + .body(payload.clone()) + .send() + .await + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(accepted.status(), ReqStatus::OK); + handle.abort(); + + // Read every store artifact (main db + WAL + shm if present). + let mut on_disk = Vec::new(); + for suffix in ["", "-wal", "-shm"] { + let path = format!("{}{}", db_path.to_string_lossy(), suffix); + if let Ok(bytes) = std::fs::read(&path) { + on_disk.extend_from_slice(&bytes); + } + } + assert!(!on_disk.is_empty(), "store files missing or empty"); + + // Non-vacuity: the opaque payload IS on disk verbatim (right file, real + // write path)... + assert!( + contains_subslice(&on_disk, &payload), + "payload bytes not found — wrong file or broken write path" + ); + // ...while the raw route token never is. + assert!( + !contains_subslice(&on_disk, route_token.as_bytes()), + "raw route token leaked into the store file" + ); + + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/tests/qsl_attachments_integration_contract.rs b/tests/qsl_attachments_integration_contract.rs index 5c212ac..66f9c34 100644 --- a/tests/qsl_attachments_integration_contract.rs +++ b/tests/qsl_attachments_integration_contract.rs @@ -269,8 +269,29 @@ async fn na0347_quota_rate_retention_purge_and_backup_boundaries_are_bounded() { assert_eq!(second_after_drain.status(), ReqStatus::OK); handle.abort(); - let (base, handle) = - spawn_server_with_auth(Limits::new(64, 2).unwrap(), controls(2, 2, 25), None).await; + // NA-0642: the idle-route discard is retired; the retention TTL is the + // purge boundary. Undelivered messages expire after RETENTION_TTL_SECS. + let retention_store = qsl_server::StoreConfig { + retention_ttl_secs: 1, + ..qsl_server::StoreConfig::default() + }; + let state = qsl_server::AppState::new_with_auth_controls_and_store( + Limits::new(64, 2).unwrap(), + controls(2, 2, 25), + None, + retention_store, + ) + .unwrap_or_else(|e| panic!("{e}")); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .unwrap_or_else(|e| panic!("{e}")); + let addr = listener.local_addr().unwrap_or_else(|e| panic!("{e}")); + let handle = tokio::spawn(async move { + axum::serve(listener, qsl_server::app(state)) + .await + .unwrap_or_else(|e| panic!("{e}")); + }); + let base = format!("http://{addr}"); let stale = push( &client, &base, @@ -281,7 +302,7 @@ async fn na0347_quota_rate_retention_purge_and_backup_boundaries_are_bounded() { ) .await; assert_eq!(stale.status(), ReqStatus::OK); - tokio::time::sleep(Duration::from_millis(75)).await; + tokio::time::sleep(Duration::from_millis(1500)).await; let expired = pull(&client, &base, "NA0347_TTL_STALE_ROUTE", None, 1).await; assert_eq!(expired.status(), ReqStatus::NO_CONTENT); handle.abort(); diff --git a/tests/route_lifecycle_ttl_logging.rs b/tests/route_lifecycle_ttl_logging.rs deleted file mode 100644 index 5050f2b..0000000 --- a/tests/route_lifecycle_ttl_logging.rs +++ /dev/null @@ -1,140 +0,0 @@ -use qsl_server::{app, AppState, Limits, ResourceControls}; -use reqwest::StatusCode as ReqStatus; -use std::{ - io::Write, - sync::{Arc, Mutex}, - time::Duration, -}; -use tokio::net::TcpListener; -use tracing::subscriber::set_default; - -const ROUTE_TOKEN_HEADER: &str = "X-QSL-Route-Token"; -const MSG_ID_HEADER: &str = "X-Msg-Id"; - -#[derive(Clone)] -struct SharedWriter(Arc>>); - -impl Write for SharedWriter { - fn write(&mut self, buf: &[u8]) -> std::io::Result { - let mut guard = self.0.lock().unwrap_or_else(|e| panic!("{e}")); - guard.extend_from_slice(buf); - Ok(buf.len()) - } - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } -} - -async fn spawn_server_with_auth( - limits: Limits, - controls: ResourceControls, - relay_token: Option<&str>, -) -> (String, tokio::task::JoinHandle<()>) { - let state = - AppState::new_with_auth_and_controls(limits, controls, relay_token.map(str::to_string)); - let listener = TcpListener::bind("127.0.0.1:0") - .await - .unwrap_or_else(|e| panic!("{e}")); - let addr = listener.local_addr().unwrap_or_else(|e| panic!("{e}")); - let handle = tokio::spawn(async move { - axum::serve(listener, app(state)) - .await - .unwrap_or_else(|e| panic!("{e}")); - }); - (format!("http://{addr}"), handle) -} - -async fn push( - client: &reqwest::Client, - base: &str, - route_token: &str, - auth_token: Option<&str>, - msg_id: Option<&str>, - body: impl Into>, -) -> reqwest::Response { - let mut request = client - .post(format!("{base}/v1/push")) - .header(ROUTE_TOKEN_HEADER, route_token) - .body(body.into()); - if let Some(token) = auth_token { - request = request.header("Authorization", format!("Bearer {token}")); - } - if let Some(id) = msg_id { - request = request.header(MSG_ID_HEADER, id); - } - request.send().await.unwrap_or_else(|e| panic!("{e}")) -} - -async fn pull( - client: &reqwest::Client, - base: &str, - route_token: &str, - auth_token: Option<&str>, - max: usize, -) -> reqwest::Response { - let mut request = client - .get(format!("{base}/v1/pull?max={max}")) - .header(ROUTE_TOKEN_HEADER, route_token); - if let Some(token) = auth_token { - request = request.header("Authorization", format!("Bearer {token}")); - } - request.send().await.unwrap_or_else(|e| panic!("{e}")) -} - -#[tokio::test(flavor = "current_thread")] -async fn ttl_cleanup_logs_redact_route_auth_payload() { - let buf = Arc::new(Mutex::new(Vec::new())); - let writer = SharedWriter(buf.clone()); - let subscriber = tracing_subscriber::fmt() - .with_max_level(tracing::Level::INFO) - .with_ansi(false) - .with_writer(move || writer.clone()) - .finish(); - let _guard = set_default(subscriber); - - let auth_token = "NA0281_AUTH_TOKEN_SENTINEL_MUST_NOT_LEAK"; - let route_token = "NA0281_TTL_ROUTE_TOKEN_MUST_NOT_LEAK"; - let payload = b"NA0281_TTL_PAYLOAD_MUST_NOT_LEAK".to_vec(); - let controls = ResourceControls::new_with_route_idle_ttl_ms(2, 2, 0, 25).unwrap(); - let (base, handle) = - spawn_server_with_auth(Limits::new(128, 4).unwrap(), controls, Some(auth_token)).await; - let client = reqwest::Client::new(); - - let accepted = push( - &client, - &base, - route_token, - Some(auth_token), - Some("NA0281_TTL_MSG_ID_NONSECRET_METADATA"), - payload, - ) - .await; - assert_eq!(accepted.status(), ReqStatus::OK); - - tokio::time::sleep(Duration::from_millis(75)).await; - - let expired = pull(&client, &base, route_token, Some(auth_token), 1).await; - assert_eq!(expired.status(), ReqStatus::NO_CONTENT); - - tokio::task::yield_now().await; - handle.abort(); - - let guard = buf.lock().unwrap_or_else(|e| panic!("{e}")); - let logs = String::from_utf8_lossy(&guard); - assert!(logs.contains("event=route_expired")); - assert!(logs.contains("channel_id=")); - assert!(logs.contains("queued_messages=1")); - assert!(logs.contains("ttl_ms=25")); - assert!(logs.contains("NA0281_TTL_MSG_ID_NONSECRET_METADATA")); - - for forbidden in [ - auth_token, - route_token, - "Authorization", - "Bearer", - "NA0281_TTL_PAYLOAD_MUST_NOT_LEAK", - ] { - assert!(!logs.contains(forbidden), "logs leaked {forbidden}"); - } -}