diff --git a/database/README.md b/database/README.md index d72820eb1..89b7dc798 100644 --- a/database/README.md +++ b/database/README.md @@ -72,8 +72,12 @@ databases: analytics: url: ${ANALYTICS_URL:postgres://localhost/analytics} pool: { max: 5 } +history_max_entries: 200 # console query history caps, defaults shown +history_max_bytes: 262144 # 0 disables recording ``` +`history_max_entries` / `history_max_bytes` cap the per-database console query history stored on the [`state`](../state) worker — whichever cap hits first, oldest entries are dropped. Applied live. + Set or replace the whole value: ```bash @@ -225,7 +229,7 @@ Stored in the [`state`](https://github.com/iii-hq/workers/tree/main/state) worke | `database::saveQuery` | Save a named query. Saving under an existing name replaces it. | | `database::listSavedQueries` | Saved queries for a database, sorted by name. | | `database::deleteSavedQuery` | Delete by id or by name. | -| `database::history` | Recent queries, newest first. Best effort — recording never blocks or fails a query, so this is a convenience rather than an audit log. For an audit trail bind `database::row-changed`. | +| `database::history` | Recent queries, newest first. Best effort — recording never blocks or fails a query, so this is a convenience rather than an audit log. For an audit trail bind `database::row-changed`. Stored history is capped per database (`history_max_entries` / `history_max_bytes`, defaults 200 entries / 256KB — oldest dropped first, `0` disables) and holds metadata only: SQL text (truncated to 4000 chars), verb, timing, row count — never result rows. An oversized or unreadable stored value is replaced wholesale on the next write. | ## Triggers diff --git a/database/config.yaml.example b/database/config.yaml.example index 68d3d6416..403ccdf18 100644 --- a/database/config.yaml.example +++ b/database/config.yaml.example @@ -33,3 +33,9 @@ databases: # orders: # url: mysql://user:pass@localhost:3306/orders # capture: native + +# Console query history caps (per database, stored on the `state` worker). +# Whichever cap hits first, oldest entries are dropped; 0 disables recording. +# Defaults shown. +# history_max_entries: 200 +# history_max_bytes: 262144 diff --git a/database/src/config.rs b/database/src/config.rs index 357e8dd11..1f5064f39 100644 --- a/database/src/config.rs +++ b/database/src/config.rs @@ -25,6 +25,18 @@ pub struct WorkerConfig { #[serde(default)] #[schemars(schema_with = "databases_schema")] pub databases: HashMap, + /// Maximum entries kept per database in the console query history + /// (`database::history`, stored on the `state` worker). Oldest entries + /// are dropped first. `0` disables history recording. Applied live. + #[serde(default = "default_history_max_entries")] + pub history_max_entries: usize, + /// Maximum JSON-serialized size in bytes of one database's stored query + /// history; oldest entries are dropped until the list fits. The hard + /// guard that keeps history from growing into a value large enough to + /// break the `state` worker's connection. `0` disables history + /// recording. Applied live. + #[serde(default = "default_history_max_bytes")] + pub history_max_bytes: usize, } fn worker_config_example() -> WorkerConfig { @@ -211,6 +223,12 @@ fn default_idle_timeout_ms() -> u64 { fn default_acquire_timeout_ms() -> u64 { 5_000 } +fn default_history_max_entries() -> usize { + 200 +} +fn default_history_max_bytes() -> usize { + 262_144 +} impl Default for WorkerConfig { fn default() -> Self { @@ -231,6 +249,8 @@ impl WorkerConfig { driver: DriverKind::default(), }, )]), + history_max_entries: default_history_max_entries(), + history_max_bytes: default_history_max_bytes(), }) .expect("built-in default config is valid") } @@ -280,7 +300,9 @@ impl WorkerConfig { "acquire_timeout_ms": default_acquire_timeout_ms(), } } - } + }, + "history_max_entries": default_history_max_entries(), + "history_max_bytes": default_history_max_bytes(), }), ); } @@ -569,9 +591,28 @@ mod tests { ); } + for field in ["history_max_entries", "history_max_bytes"] { + assert!( + schema["properties"][field].get("description").is_some(), + "missing description for {field}" + ); + } + assert!(schema.get("example").is_some()); } + #[test] + fn history_caps_default_and_override() { + let d = cfg("databases:\n p:\n url: postgres://u@h/db\n"); + assert_eq!(d.history_max_entries, 200); + assert_eq!(d.history_max_bytes, 262_144); + + let c = cfg("databases:\n p:\n url: postgres://u@h/db\n\ + history_max_entries: 25\nhistory_max_bytes: 4096\n"); + assert_eq!(c.history_max_entries, 25); + assert_eq!(c.history_max_bytes, 4096); + } + /// Regenerate the e2e harness schema fixture when `WorkerConfig` changes: /// `EXPORT_E2E_SCHEMA=1 cargo test -p database export_e2e_schema_fixture -- --ignored` #[test] diff --git a/database/src/handlers/query.rs b/database/src/handlers/query.rs index 411e80549..9c6569a75 100644 --- a/database/src/handlers/query.rs +++ b/database/src/handlers/query.rs @@ -83,6 +83,7 @@ pub async fn handle(state: &AppState, req: QueryReq) -> Result Result { - let key = history_key(db); - let raw = state_get(iii, &key).await?; + let raw = state_get(iii, &history_key(db)).await?; let all: Vec = raw .iter() .filter_map(|v| serde_json::from_value(v.clone()).ok()) .collect(); - // Self-healing trim: the write path only appends, so the stored list is - // capped here once it drifts past the high-water mark. - if all.len() > HISTORY_HIGH_WATER { - let tail: Vec<&HistoryEntry> = all.iter().rev().take(HISTORY_LIMIT).rev().collect(); - let _ = state_set(iii, &key, serde_json::to_value(&tail).unwrap_or(json!([]))).await; - } - - let limit = req - .limit - .unwrap_or(HISTORY_LIMIT) - .clamp(1, HISTORY_HIGH_WATER); + // The stored list is capped on write, so `take` self-bounds. + let limit = req.limit.unwrap_or(HISTORY_LIMIT).max(1); let entries: Vec = all.into_iter().rev().take(limit).collect(); Ok(HistoryResp { count: entries.len(), @@ -267,18 +260,46 @@ pub async fn history( }) } -/// The `state::update` op list for appending one entry. -/// -/// Split out only so a test can assert the discriminator, which is the one -/// detail of this file that cannot be caught at runtime. -fn append_ops(entry: Value) -> Vec { - vec![json!({ "type": "append", "value": entry })] +/// Databases whose stored history could not be read. Their next write skips +/// the read and replaces the value wholesale: merely serving an oversized +/// value can reset the `state` worker's connection, so recovery must never +/// depend on reading the value it is recovering from. +static RESET_PENDING: LazyLock>> = + LazyLock::new(|| Mutex::new(HashSet::new())); + +fn reset_pending(db: &str) -> bool { + RESET_PENDING.lock().is_ok_and(|s| s.contains(db)) +} + +fn mark_reset_pending(db: &str) { + if let Ok(mut s) = RESET_PENDING.lock() { + s.insert(db.to_string()); + } +} + +fn clear_reset_pending(db: &str) { + if let Ok(mut s) = RESET_PENDING.lock() { + s.remove(db); + } } /// Record one run. Fire and forget: never awaited on the query path, and a /// failure is logged rather than surfaced, because losing a history line must /// never fail the query the user actually asked for. -pub fn record(iii: Arc, db: String, sql: &str, duration_ms: u64, row_count: usize) { +/// +/// Each write stores the capped tail of the list: read, append, trim to the +/// configured caps, replace. Last-writer-wins — two concurrent records can +/// drop a line, which the lossy-by-design charter above allows. A stored +/// value that cannot be read (missing worker, or a pre-cap oversized blob) is +/// replaced instead of retried, losing old lines but unwedging `state`. +pub fn record( + iii: Arc, + config: Arc>, + db: String, + sql: &str, + duration_ms: u64, + row_count: usize, +) { let entry = HistoryEntry { sql: truncate(sql), verb: leading_verb(sql), @@ -287,17 +308,62 @@ pub fn record(iii: Arc, db: String, sql: &str, duration_ms: u64, row_ at: now(), }; tokio::spawn(async move { - let payload = json!({ - "scope": SCOPE, - "key": history_key(&db), - "ops": append_ops(serde_json::to_value(&entry).unwrap_or(json!({}))), - }); - if let Err(e) = call(&iii, "state::update", payload).await { - tracing::warn!(error = %e, "history not recorded"); + let (max_entries, max_bytes) = { + let cfg = config.read().await; + (cfg.history_max_entries, cfg.history_max_bytes) + }; + if max_entries == 0 || max_bytes == 0 { + return; + } + let key = history_key(&db); + let mut items = if reset_pending(&db) { + Vec::new() + } else { + match state_get(&iii, &key).await { + Ok(items) => items, + Err(e) => { + mark_reset_pending(&db); + tracing::warn!(error = %e, "history unreadable; next write resets it"); + Vec::new() + } + } + }; + items.push(serde_json::to_value(&entry).unwrap_or(json!({}))); + trim_to_caps(&mut items, max_entries, max_bytes); + match state_set(&iii, &key, Value::Array(items)).await { + Ok(()) => clear_reset_pending(&db), + Err(e) => tracing::warn!(error = %e, "history not recorded"), } }); } +/// Drop oldest entries until the list fits both caps. +/// +/// Byte size is the compact `serde_json` encoding, computed without +/// re-serializing the list per drop: `[]` is 2 bytes, `n` entries cost the +/// 2 brackets plus their summed lengths plus `n − 1` commas. +fn trim_to_caps(entries: &mut Vec, max_entries: usize, max_bytes: usize) { + let lens: Vec = entries + .iter() + .map(|v| serde_json::to_vec(v).map_or(usize::MAX, |b| b.len())) + .collect(); + let mut kept = 0usize; + let mut bytes = 0usize; + for len in lens.iter().rev() { + let with_next = bytes + .saturating_add(*len) + .saturating_add(2) // brackets + .saturating_add(kept); // commas once this entry joins + if kept == max_entries || with_next > max_bytes { + break; + } + bytes += len; + kept += 1; + } + let surplus = entries.len() - kept; + entries.drain(..surplus); +} + fn now() -> String { chrono::Utc::now().to_rfc3339() } @@ -350,19 +416,89 @@ mod tests { assert_eq!(t.chars().count(), MAX_SQL_CHARS + 1); } - #[test] - fn append_op_uses_the_type_discriminator() { - // Locked deliberately: `record` is fire-and-forget, so a wrong op - // shape fails where nobody is looking. This is the only cheap place - // to notice. - let ops = append_ops(json!({"sql": "select 1"})); - assert_eq!(ops[0]["type"], "append"); - assert!(ops[0].get("op").is_none()); - } - #[test] fn keys_are_scoped_per_database() { assert_eq!(saved_key("primary"), "saved:primary"); assert_eq!(history_key("analytics"), "history:analytics"); } + + fn entry(sql: &str) -> Value { + serde_json::to_value(HistoryEntry { + sql: sql.into(), + verb: leading_verb(sql), + duration_ms: Some(12), + row_count: Some(3), + at: "2026-08-06T00:00:00+00:00".into(), + }) + .unwrap() + } + + #[test] + fn trim_keeps_newest_within_entry_cap() { + let mut items: Vec = (0..5).map(|i| entry(&format!("select {i}"))).collect(); + trim_to_caps(&mut items, 3, usize::MAX); + let sqls: Vec<&str> = items.iter().map(|v| v["sql"].as_str().unwrap()).collect(); + assert_eq!(sqls, ["select 2", "select 3", "select 4"]); + } + + #[test] + fn trim_enforces_byte_cap_dropping_oldest() { + let mut items: Vec = (0..10).map(|i| entry(&format!("select {i}"))).collect(); + // One byte short of fitting all ten forces at least one drop. + let cap = serde_json::to_vec(&Value::Array(items.clone())) + .unwrap() + .len() + - 1; + trim_to_caps(&mut items, usize::MAX, cap); + assert!(!items.is_empty() && items.len() < 10); + assert_eq!(items.last().unwrap()["sql"], "select 9"); + assert!(serde_json::to_vec(&Value::Array(items)).unwrap().len() <= cap); + } + + #[test] + fn trim_byte_accounting_matches_serde_exactly() { + // The trim never re-serializes the whole list, so its arithmetic must + // match serde's compact encoding to the byte — including multi-byte + // and escaped content. + let items = vec![ + entry("select 'plain'"), + entry("select 'héllo … ↹'"), + entry("select \"quoted\\backslash\"\n\t"), + ]; + let summed: usize = items + .iter() + .map(|v| serde_json::to_vec(v).unwrap().len()) + .sum(); + let actual = serde_json::to_vec(&Value::Array(items.clone())) + .unwrap() + .len(); + assert_eq!(2 + summed + (items.len() - 1), actual); + assert_eq!(serde_json::to_vec(&Value::Array(vec![])).unwrap().len(), 2); + } + + #[test] + fn oversized_single_entry_yields_empty_history() { + let mut items = vec![entry(&"x".repeat(1_000))]; + trim_to_caps(&mut items, 10, 64); + assert!(items.is_empty()); + } + + #[test] + fn trim_noop_when_within_caps() { + let mut items: Vec = (0..3).map(|i| entry(&format!("select {i}"))).collect(); + let before = items.clone(); + trim_to_caps(&mut items, 200, 262_144); + assert_eq!(items, before); + } + + #[test] + fn reset_flag_marks_and_clears_per_database() { + // Unique names: the flag set is a process-wide static shared by tests. + assert!(!reset_pending("reset-flag-test-a")); + mark_reset_pending("reset-flag-test-a"); + assert!(reset_pending("reset-flag-test-a")); + assert!(!reset_pending("reset-flag-test-b")); + clear_reset_pending("reset-flag-test-a"); + assert!(!reset_pending("reset-flag-test-a")); + } } diff --git a/database/tests/e2e/README.md b/database/tests/e2e/README.md index ccf6fe6ef..f43481fa3 100644 --- a/database/tests/e2e/README.md +++ b/database/tests/e2e/README.md @@ -4,7 +4,9 @@ Self-asserting smoke harness for the `database` worker. Validates the function surface (query / execute / prepareStatement / runStatement / transaction), the **interactive-transaction** surface (begin / transactionQuery / transactionExecute / commit / rollback), -`database::row-changed` delivery, and the side-channel-finalization repros +`database::row-changed` delivery, the query-history caps (via +harness-registered mock `state::*` functions), and the +side-channel-finalization repros from the `/review` of branch `feat/database-and-skills` against real **SQLite**, **PostgreSQL 16**, and **MySQL 8.4** with one @@ -117,6 +119,7 @@ accepted; outside-tx COUNT=1`). | `workers/harness/src/cases-row-changed.ts` | Row-change trigger delivery cases | | `workers/harness/src/cases-native-capture.ts` | `capture: native` cases for postgres (LISTEN/NOTIFY, `pg_native_db`), sqlite (changelog + fs watch, `sqlite_native_db`), and mysql (binlog stream, `mysql_native_db`) — cross-client delivery, no double-fire on own writes, table-less binding rejection, commit/rollback gating through interactive transactions, multi-subscriber fan-out with ops filters across trigger reinstall, bulk-statement coalescing (100 rows = 1 event) | | `mysql-init/grant-replication.sql` | Replication grants for the `iii` mysql user (binlog capture streams as a replica). Applied on first volume init only — on an older volume run `docker compose down -v` once | +| `workers/harness/src/cases-history.ts` | Query-history cap cases (MOT-4372). No state worker runs in this stack, so the harness registers mock `state::get`/`state::set`/`state::update` and observes every history write: entry/byte rotation against the tiny caps seeded in `database-config.ts`, oversized-backlog trim, unreadable-value blind-write self-heal, and a `state::update` regression guard. Sqlite only — the write path is driver-agnostic | | `workers/harness/src/cases-tx-control-bypass.ts` | Side-channel-finalization repros | | `reports/report.json` | Per-case results (latest run) | diff --git a/database/tests/e2e/workers/harness/fixtures/database.schema.json b/database/tests/e2e/workers/harness/fixtures/database.schema.json index 3ab99ff2c..17499e854 100644 --- a/database/tests/e2e/workers/harness/fixtures/database.schema.json +++ b/database/tests/e2e/workers/harness/fixtures/database.schema.json @@ -28,7 +28,7 @@ "$ref": "#/definitions/CaptureMode" } ], - "description": "How `database::row-changed` events are captured for this database. `statements` (default) classifies the SQL this worker executes; `native` makes writes from ANY client — psql, other processes — fire too. Postgres: triggers + LISTEN/NOTIFY. File-backed sqlite: triggers + changelog drained on filesystem wake-up. MySQL: the binlog replication stream (needs REPLICATION SLAVE + REPLICATION CLIENT)." + "description": "How `database::row-changed` events are captured for this database. `statements` (default) classifies the SQL this worker executes; `native` makes writes from ANY client — psql, other processes — fire too. Postgres: triggers + LISTEN/NOTIFY. File-backed sqlite: a trigger-fed changelog drained on filesystem wake-up. MySQL: the binlog replication stream (needs REPLICATION SLAVE + CLIENT)." }, "pool": { "allOf": [ @@ -155,7 +155,9 @@ }, "url": "sqlite:./data/iii.db" } - } + }, + "history_max_bytes": 262144, + "history_max_entries": 200 }, "examples": [ { @@ -173,7 +175,9 @@ }, "url": "sqlite:./data/iii.db" } - } + }, + "history_max_bytes": 262144, + "history_max_entries": 200 } ], "properties": { @@ -197,6 +201,20 @@ ], "minProperties": 1, "type": "object" + }, + "history_max_bytes": { + "default": 262144, + "description": "Maximum JSON-serialized size in bytes of one database's stored query history; oldest entries are dropped until the list fits. The hard guard that keeps history from growing into a value large enough to break the `state` worker's connection. `0` disables history recording. Applied live.", + "format": "uint", + "minimum": 0.0, + "type": "integer" + }, + "history_max_entries": { + "default": 200, + "description": "Maximum entries kept per database in the console query history (`database::history`, stored on the `state` worker). Oldest entries are dropped first. `0` disables history recording. Applied live.", + "format": "uint", + "minimum": 0.0, + "type": "integer" } }, "title": "WorkerConfig", diff --git a/database/tests/e2e/workers/harness/src/cases-history.ts b/database/tests/e2e/workers/harness/src/cases-history.ts new file mode 100644 index 000000000..30b1a0893 --- /dev/null +++ b/database/tests/e2e/workers/harness/src/cases-history.ts @@ -0,0 +1,322 @@ +import type { ISdk } from 'iii-sdk' +import type { TestCase, CaseContext } from './cases.ts' +import { expect, expectEqual } from './cases.ts' + +/** + * Query-history cap cases (MOT-4372). The worker stores console history on + * the `state` worker; uncapped, one value grew to ~8MB — large enough that + * serving it reset the state worker's engine connection and unregistered + * `state::*` for the whole stack. + * + * No real state worker runs in this stack, and that is the point: the + * harness registers `state::get` / `state::set` / `state::update` itself, + * which both observes every write the worker makes and scripts the failure + * modes a real state worker cannot safely reproduce (a value too large to + * serve kills the very connection that would deliver it). + * + * Caps under test come from the seeded e2e config (database-config.ts): + * `history_max_entries: 5`, `history_max_bytes: 8192`. The write path is + * driver-agnostic, so the group runs on sqlite only. + */ + +const MAX_ENTRIES = 5 +const MAX_BYTES = 8192 +const SCOPE = 'database' +const READY_TIMEOUT_MS = 10_000 +const WRITE_TIMEOUT_MS = 5_000 + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)) +const jsonBytes = (v: unknown) => Buffer.byteLength(JSON.stringify(v)) +const historyKey = (driver: string) => `${SCOPE}/history:${driver}` + +interface StoredEntry { + sql: string + verb: string + duration_ms?: number + row_count?: number + at: string +} + +interface KeyRef { + scope: string + key: string +} + +interface MockState { + store: Map + /** `state::get` attempts per key, counted before any scripted throw. */ + gets: Map + /** Every `state::set` attempt in arrival order, including scripted failures. */ + setAttempts: Array<{ key: string; failed: boolean }> + /** `state::update` invocations — must stay 0; history's old uncapped append lived there. */ + updates: number + /** Keys whose `state::get` throws, standing in for a value too large to serve. */ + getThrows: Set + /** Upcoming `state::set` calls to fail before succeeding again. */ + setFailuresRemaining: number + unregister(): void +} + +function registerMockState(iii: ISdk): MockState { + const mock: MockState = { + store: new Map(), + gets: new Map(), + setAttempts: [], + updates: 0, + getThrows: new Set(), + setFailuresRemaining: 0, + unregister: () => {}, + } + const k = (p: KeyRef) => `${p.scope}/${p.key}` + + const getRef = iii.registerFunction( + 'state::get', + async (p: KeyRef) => { + const key = k(p) + mock.gets.set(key, (mock.gets.get(key) ?? 0) + 1) + if (mock.getThrows.has(key)) { + throw new Error('MOCK_UNSERVABLE: value too large to serve') + } + return mock.store.get(key) ?? null + }, + { description: 'Harness mock state::get (history-cap e2e).' }, + ) + const setRef = iii.registerFunction( + 'state::set', + async (p: KeyRef & { value: unknown }) => { + const key = k(p) + const failed = mock.setFailuresRemaining > 0 + mock.setAttempts.push({ key, failed }) + if (failed) { + mock.setFailuresRemaining -= 1 + throw new Error('MOCK_SET_DOWN: scripted state::set failure') + } + mock.store.set(key, p.value) + return null + }, + { description: 'Harness mock state::set (history-cap e2e).' }, + ) + const updateRef = iii.registerFunction( + 'state::update', + async () => { + mock.updates += 1 + throw new Error('MOCK_NO_UPDATE: history must not use state::update') + }, + { description: 'Harness mock state::update — history writes must never land here.' }, + ) + mock.unregister = () => { + getRef.unregister() + setRef.unregister() + updateRef.unregister() + } + return mock +} + +/** Wait until the engine routes `state::*` to the mock, then zero the counters. */ +async function mockReady(ctx: CaseContext, mock: MockState): Promise { + const deadline = Date.now() + READY_TIMEOUT_MS + const probe = { scope: SCOPE, key: 'harness:probe' } + for (;;) { + try { + await ctx.call('state::get', probe) + await ctx.call('state::set', { ...probe, value: [] }) + break + } catch (e) { + if (Date.now() > deadline) throw new Error(`mock state::get/set not reachable: ${e}`) + await sleep(50) + } + } + // `state::update` always throws its own marker; routed once we see it. + for (;;) { + try { + await ctx.call('state::update', { ...probe, ops: [] }) + throw new Error('mock state::update resolved — it must throw') + } catch (e: any) { + const msg = String(e?.message ?? e) + if (msg.includes('must throw')) throw e + if (msg.includes('MOCK_NO_UPDATE')) break + if (Date.now() > deadline) throw new Error(`mock state::update not reachable: ${msg}`) + await sleep(50) + } + } + mock.store.clear() + mock.gets.clear() + mock.setAttempts.length = 0 + mock.updates = 0 +} + +/** + * Run one recorded query and wait for its (fire-and-forget) history write + * attempt to reach the mock — success or scripted failure — so consecutive + * queries never interleave their read-modify-write cycles. + */ +async function recordedQuery(ctx: CaseContext, mock: MockState, sql: string): Promise { + const before = mock.setAttempts.length + await ctx.call('database::query', { db: ctx.driver, sql }) + const deadline = Date.now() + WRITE_TIMEOUT_MS + while (mock.setAttempts.length === before) { + // Name the regression rather than waiting out the timeout: the pre-fix + // worker appended through state::update, which is both uncapped and + // exempt from the state worker's size guard. + if (mock.updates > 0) { + throw new Error('history wrote via state::update (uncapped append) instead of state::set') + } + if (Date.now() > deadline) { + throw new Error(`history write for ${JSON.stringify(sql)} never reached state::set`) + } + await sleep(20) + } +} + +/** + * One recorded query whose write lands successfully. Earlier suites run + * their queries with no `state::*` registered at all, which latches the + * worker's replace-without-reading recovery; a successful write clears the + * latch and leaves the store holding exactly this query's entry. + */ +async function flushHistory(ctx: CaseContext, mock: MockState): Promise { + await recordedQuery(ctx, mock, 'SELECT 0') + const last = mock.setAttempts[mock.setAttempts.length - 1] + expect(last !== undefined && !last.failed, 'flush write should succeed') +} + +function storedHistory(mock: MockState, driver: string): StoredEntry[] { + const v = mock.store.get(historyKey(driver)) + expect(Array.isArray(v), `stored history is an array, got: ${JSON.stringify(v)?.slice(0, 200)}`) + return v as StoredEntry[] +} + +export const HISTORY_CASES: TestCase[] = [ + { + name: 'history entry cap rotates oldest entries out', + applies: ['sqlite_db'], + async run(ctx) { + const mock = registerMockState(ctx.iii) + try { + await mockReady(ctx, mock) + await flushHistory(ctx, mock) + for (let i = 1; i <= 8; i++) { + await recordedQuery(ctx, mock, `SELECT ${100 + i}`) + } + const stored = storedHistory(mock, ctx.driver) + expectEqual(stored.length, MAX_ENTRIES, 'stored entry count == history_max_entries') + expectEqual( + stored.map((e) => e.sql), + ['SELECT 104', 'SELECT 105', 'SELECT 106', 'SELECT 107', 'SELECT 108'], + 'newest entries kept, oldest rotated out', + ) + expect(jsonBytes(stored) <= MAX_BYTES, 'stored value within history_max_bytes') + // Read path through the real engine round-trips the same tail. + const h = await ctx.call('database::history', { db: ctx.driver }) + expectEqual(h.count, MAX_ENTRIES, 'database::history count') + expectEqual(h.entries[0].sql, 'SELECT 108', 'database::history newest first') + expectEqual(mock.updates, 0, 'state::update never used') + } finally { + mock.unregister() + } + }, + }, + { + name: 'history byte cap trims fat entries to fit', + applies: ['sqlite_db'], + async run(ctx) { + const mock = registerMockState(ctx.iii) + try { + await mockReady(ctx, mock) + await flushHistory(ctx, mock) + // ~3KB per entry: two fit under the 8KB byte cap, three never do — + // while the 5-entry cap alone would happily keep them all. + const pad = `/* ${'x'.repeat(3000)} */` + for (let i = 1; i <= 4; i++) { + await recordedQuery(ctx, mock, `SELECT ${200 + i} ${pad}`) + expect( + jsonBytes(storedHistory(mock, ctx.driver)) <= MAX_BYTES, + `write ${i}: stored value within history_max_bytes`, + ) + } + const stored = storedHistory(mock, ctx.driver) + expectEqual(stored.length, 2, 'byte cap binds before the entry cap') + expectEqual( + stored.map((e) => e.sql.slice(0, 10)), + ['SELECT 203', 'SELECT 204'], + 'newest fat entries kept', + ) + expectEqual(mock.updates, 0, 'state::update never used') + } finally { + mock.unregister() + } + }, + }, + { + name: 'oversized stored history is trimmed to caps on the next write', + applies: ['sqlite_db'], + async run(ctx) { + const mock = registerMockState(ctx.iii) + try { + await mockReady(ctx, mock) + await flushHistory(ctx, mock) + // ~500KB of readable pre-cap backlog — what a stack upgraded from + // the uncapped worker wakes up with (short of transport-fatal). + const backlog = Array.from({ length: 500 }, (_, i) => ({ + sql: `SELECT ${i} /* ${'y'.repeat(950)} */`, + verb: 'select', + at: '2026-01-01T00:00:00+00:00', + })) + mock.store.set(historyKey(ctx.driver), backlog) + await recordedQuery(ctx, mock, 'SELECT 999') + const stored = storedHistory(mock, ctx.driver) + expectEqual(stored.length, MAX_ENTRIES, 'backlog trimmed to the entry cap') + expect(jsonBytes(stored) <= MAX_BYTES, 'backlog trimmed to the byte cap') + expectEqual(stored[stored.length - 1]?.sql, 'SELECT 999', 'new entry survives the trim') + expectEqual(mock.updates, 0, 'state::update never used') + } finally { + mock.unregister() + } + }, + }, + { + name: 'unreadable stored history is replaced blind without re-reading', + applies: ['sqlite_db'], + async run(ctx) { + const mock = registerMockState(ctx.iii) + try { + await mockReady(ctx, mock) + await flushHistory(ctx, mock) + const hk = historyKey(ctx.driver) + // Script the outage: reads of this key die (in production the ~8MB + // value resets the connection that would serve it), and the first + // write after the failed read dies with it. + mock.getThrows.add(hk) + mock.setFailuresRemaining = 1 + const getsBefore = mock.gets.get(hk) ?? 0 + + // First write: read fails, worker latches replace-without-reading, + // and the blind write is scripted to fail so the latch survives. + await recordedQuery(ctx, mock, 'SELECT 301') + const firstAttempt = mock.setAttempts[mock.setAttempts.length - 1] + expect(firstAttempt !== undefined && firstAttempt.failed, 'first write hit the scripted set failure') + expectEqual((mock.gets.get(hk) ?? 0) - getsBefore, 1, 'exactly one read attempt so far') + + // Second write: no re-read of the poisoned key — straight to a small + // replacement value. This is the self-heal: the value that broke the + // connection is never round-tripped again. + await recordedQuery(ctx, mock, 'SELECT 302') + expectEqual((mock.gets.get(hk) ?? 0) - getsBefore, 1, 'poisoned key not re-read') + const stored = storedHistory(mock, ctx.driver) + expectEqual( + stored.map((e) => e.sql), + ['SELECT 302'], + 'replacement value holds only the new entry', + ) + expect(jsonBytes(stored) <= MAX_BYTES, 'replacement value within caps') + expectEqual(mock.updates, 0, 'state::update never used') + + // Back to normal reads; leave the worker un-latched for whatever runs next. + mock.getThrows.delete(hk) + await recordedQuery(ctx, mock, 'SELECT 303') + } finally { + mock.unregister() + } + }, + }, +] diff --git a/database/tests/e2e/workers/harness/src/database-config.ts b/database/tests/e2e/workers/harness/src/database-config.ts index e470ef3a6..2d8c26157 100644 --- a/database/tests/e2e/workers/harness/src/database-config.ts +++ b/database/tests/e2e/workers/harness/src/database-config.ts @@ -56,4 +56,8 @@ export const DATABASE_CONFIG_VALUE = { tls: { mode: 'disable' as const }, }, }, + // Tiny query-history caps so cases-history.ts exercises entry/byte + // rotation with a handful of queries. Worker defaults: 200 / 262144. + history_max_entries: 5, + history_max_bytes: 8192, } as const; diff --git a/database/tests/e2e/workers/harness/src/runner.ts b/database/tests/e2e/workers/harness/src/runner.ts index 8d6776573..500dce6bb 100644 --- a/database/tests/e2e/workers/harness/src/runner.ts +++ b/database/tests/e2e/workers/harness/src/runner.ts @@ -11,6 +11,7 @@ import { CONCURRENCY_CASES } from './cases-concurrency.ts' import { TX_CONTROL_BYPASS_CASES } from './cases-tx-control-bypass.ts' import { ROW_CHANGED_CASES } from './cases-row-changed.ts' import { NATIVE_CAPTURE_CASES } from './cases-native-capture.ts' +import { HISTORY_CASES } from './cases-history.ts' interface CaseResult { driver: DriverKey @@ -156,6 +157,7 @@ export class Runner { ...CONCURRENCY_CASES, ...ROW_CHANGED_CASES, ...NATIVE_CAPTURE_CASES, + ...HISTORY_CASES, ]) { if (!matchesDriver(driver, c)) continue record(await this.runCase(driver, c))