From cf544214804f613d4c11e959f1ecebb0a98fe921 Mon Sep 17 00:00:00 2001 From: David J Parrott Date: Fri, 18 Sep 2026 23:48:50 -0400 Subject: [PATCH] Add transactional durable REST invalidation replay --- .github/workflows/ci.yml | 11 +- BACKLOG.md | 5 + build.zig | 20 +- changelog.d/durable-realtime.md | 3 + docs/api.md | 78 ++++++ docs/framework.md | 5 + src/api/meta.zig | 3 + src/api/realtime_backfill.zig | 46 +++- src/api/records.zig | 7 + src/framework.zig | 7 +- src/realtime/durable.zig | 347 +++++++++++++++++++++++++++ src/root.zig | 1 + tests/admin/test_durable_realtime.py | 162 +++++++++++++ 13 files changed, 681 insertions(+), 14 deletions(-) create mode 100644 changelog.d/durable-realtime.md create mode 100644 src/realtime/durable.zig create mode 100644 tests/admin/test_durable_realtime.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 883838f4..f417de97 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -302,7 +302,7 @@ jobs: key: zig-global-${{ runner.os }}-zig0.16.0-${{ hashFiles('vendor/sqlite/sqlite3.c', 'build.zig', 'build.zig.zon') }} restore-keys: zig-global-${{ runner.os }}-zig0.16.0- - name: Unit + PostgreSQL driver tests (-Dpostgres=true -Dvector=true) - run: mise exec zig@0.16.0 -- zig build test --summary all -Dpostgres=true -Dvector=true -Dquery-workbench=true + run: mise exec zig@0.16.0 -- zig build test --summary all -Dpostgres=true -Dvector=true -Dquery-workbench=true -Ddurable-realtime=true - name: PostgreSQL workbench paired gating and live HTTP run: | mise exec zig@0.16.0 -- zig build query-workbench-fixture -Dpostgres=true --prefix zig-out/workbench-off @@ -310,6 +310,10 @@ jobs: bash scripts/check-query-workbench-gating.sh zig-out/workbench-off/bin/query-workbench-fixture zig-out/workbench-on/bin/query-workbench-fixture mise exec python@3.13 -- python -m pip install --quiet pytest ZIGBASE_TEST_QUERY_WORKBENCH_OFF_BINARY="$GITHUB_WORKSPACE/zig-out/workbench-off/bin/query-workbench-fixture" ZIGBASE_TEST_QUERY_WORKBENCH_BINARY="$GITHUB_WORKSPACE/zig-out/workbench-on/bin/query-workbench-fixture" mise exec python@3.13 -- python -m pytest tests/admin/test_query_workbench_postgres.py -q + - name: Durable realtime PostgreSQL cross-instance recovery + run: | + mise exec zig@0.16.0 -- zig build -Dpostgres=true -Ddurable-realtime=true --prefix zig-out/replay-pg + ZIGBASE_TEST_DURABLE_REALTIME_BINARY="$GITHUB_WORKSPACE/zig-out/replay-pg/bin/zigbase" mise exec python@3.13 -- python -m pytest tests/admin/test_durable_realtime.py::test_postgres_cross_instance_replay -q - name: Rails PostgreSQL row-source integration run: | mise exec python@3.13 -- python -m pip install --quiet pytest -r tools/rails/requirements-postgres.txt @@ -935,6 +939,11 @@ jobs: mise exec zig@0.16.0 -- zig build test durable-uploads-fixture check-resumable-contracts -Dresumable-uploads=true -Ddurable-resumable-uploads=true --prefix zig-out/durable-uploads --summary all bash scripts/check-gating.sh --durable-uploads zig-out/resumable/bin/resumable-uploads-fixture zig-out/durable-uploads/bin/durable-uploads-fixture ZIGBASE_TEST_DURABLE_UPLOADS_BINARY="$GITHUB_WORKSPACE/zig-out/durable-uploads/bin/durable-uploads-fixture" mise exec python@3.13 -- python -m pytest tests/admin/test_durable_uploads.py -q -n 2 --dist load --durations=20 + - name: Durable realtime replay regressions + if: matrix.lane == 'coordination' + run: | + mise exec zig@0.16.0 -- zig build -Ddurable-realtime=true -Drealtime-backfill=true --prefix zig-out/durable-realtime + ZIGBASE_TEST_DURABLE_REALTIME_BINARY="$GITHUB_WORKSPACE/zig-out/durable-realtime/bin/zigbase" mise exec python@3.13 -- python -m pytest tests/admin/test_durable_realtime.py -q - name: Opt-in realtime backfill regressions if: matrix.lane == 'coordination' run: | diff --git a/BACKLOG.md b/BACKLOG.md index 4897ced4..9e4e9536 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -185,6 +185,11 @@ product priority and acceptance evidence; it does not create duplicate implement - [x] Exercise benchmark correctness under ReleaseSafe in CI. - [x] Bounded single-process SQLite backfill with current authorization and explicit gap semantics (#412). - [ ] Durable cross-instance replay with explicit resource and retention budgets. + Implemented: opt-in transactional SQLite/PostgreSQL built-in REST invalidations, + shared 4096-entry / 4 MiB / 24-hour retention, commit-ordered checkpoints, + restart/cross-instance recovery, and current per-item authorization. Data/raw-SQL, + hook side-writes and custom-channel capture remain outside this REST scope; + end-to-end replay coverage and configurable per-workload retention remain open. ## Analytics batching (#401) diff --git a/build.zig b/build.zig index 36260411..1469211f 100644 --- a/build.zig +++ b/build.zig @@ -12,6 +12,11 @@ const BuildOptionValues = struct { s3: bool, file_inventory: bool, realtime_backfill: bool, + durable_realtime: bool, + replay_max_entries: u32, + replay_max_bytes: u32, + replay_max_frame_bytes: u32, + replay_retention_seconds: u32, resumable_uploads: bool, image_thumbnails: bool, durable_resumable_uploads: bool, @@ -116,7 +121,15 @@ pub fn build(b: *std.Build) void { // a bare UnknownCommand. const dev_tools = b.option(bool, "dev-tools", "Compile development CLI verbs: init, agents-md, typegen, capabilities, routes, migrate preview, tune, diagnostics (default: on)") orelse true; const file_inventory = b.option(bool, "file-inventory", "Compile read-only local/S3 file inventory reporting (default: off)") orelse false; - const realtime_backfill = b.option(bool, "realtime-backfill", "Compile bounded process-local record invalidation backfill (default: off)") orelse false; + const replay_max_entries = b.option(u32, "replay-max-entries", "Durable journal entry budget (1..65536)") orelse 4096; + const replay_max_bytes = b.option(u32, "replay-max-bytes", "Durable journal frame byte budget (1..1073741824)") orelse 4194304; + const replay_max_frame_bytes = b.option(u32, "replay-max-frame-bytes", "Durable journal per-frame byte budget (1..1048576)") orelse 65536; + const replay_retention_seconds = b.option(u32, "replay-retention-seconds", "Durable journal retention (1..31536000 seconds)") orelse 86400; + if (replay_max_entries == 0 or replay_max_entries > 65536 or replay_max_bytes == 0 or replay_max_bytes > 1073741824 or replay_max_frame_bytes == 0 or replay_max_frame_bytes > 1048576 or replay_max_frame_bytes > replay_max_bytes or replay_retention_seconds == 0 or replay_retention_seconds > 31536000) + @panic("invalid durable replay budgets; frame bytes must not exceed total bytes"); + const durable_realtime = b.option(bool, "durable-realtime", "Compile transactional durable REST record replay (default: off)") orelse false; + const local_backfill = b.option(bool, "realtime-backfill", "Compile bounded process-local record invalidation backfill (default: off)") orelse false; + const realtime_backfill = durable_realtime or local_backfill; const resumable_uploads = b.option(bool, "resumable-uploads", "Compile bounded process-local resumable file uploads (default: off)") orelse false; const image_thumbnails = b.option(bool, "image-thumbnails", "Compile ImageMagick thumbnail integration (default: off)") orelse false; const durable_resumable_uploads = b.option(bool, "durable-resumable-uploads", "Compile opt-in SQLite upload persistence (requires resumable-uploads)") orelse false; @@ -172,6 +185,11 @@ pub fn build(b: *std.Build) void { .dev_tools = dev_tools, .file_inventory = file_inventory, .realtime_backfill = realtime_backfill, + .durable_realtime = durable_realtime, + .replay_max_entries = replay_max_entries, + .replay_max_bytes = replay_max_bytes, + .replay_max_frame_bytes = replay_max_frame_bytes, + .replay_retention_seconds = replay_retention_seconds, .resumable_uploads = resumable_uploads, .image_thumbnails = image_thumbnails, .durable_resumable_uploads = durable_resumable_uploads, diff --git a/changelog.d/durable-realtime.md b/changelog.d/durable-realtime.md new file mode 100644 index 00000000..57347633 --- /dev/null +++ b/changelog.d/durable-realtime.md @@ -0,0 +1,3 @@ +### Features + +- Add opt-in durable REST record invalidation replay on SQLite and PostgreSQL, with transactionally ordered checkpoints surviving restarts and cross-instance reads, bounded shared retention, and current authorization. Journal failures roll back record writes. Enable with `-Ddurable-realtime=true`; raw SQL, Data facade writes and hook side-writes remain outside capture coverage. diff --git a/docs/api.md b/docs/api.md index 044e6f03..8aeda972 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1449,6 +1449,79 @@ rewrites, consulted before the marker. See ## Realtime (WebSocket + SSE) +### Durable record invalidation replay (SQLite and PostgreSQL) + +Build with `-Ddurable-realtime=true` (and `-Dpostgres=true` for PostgreSQL). +This also enables `GET /api/realtime/backfill` with the checkpoint protocol and +current authorization described below, replacing the process-local store with a +transactional database journal. Checkpoints survive application restarts and work +on other instances connected to the same database and PostgreSQL schema. +Cursor positions are encrypted and authenticated with a persistent journal secret +and fresh random nonces; they bind collection identity and rename generation +without exposing other collections' aggregate write positions. + +Built-in REST record create, update and delete operations append invalidations +**inside the record transaction**, including resumable record updates. A journal +failure rolls back the mutation; an oversized delete snapshot also rejects the +mutation. Response serialization, after-hooks and live transport run after commit, +so failure there does not remove the committed invalidation. Create/update entries +retain only record IDs. Deletes retain private at-rest authorization snapshots; +replay never returns these snapshots or historical field values. + +A single metadata row serializes journal positions through commit on both +backends. PostgreSQL sequence allocation cannot run ahead of an earlier uncommitted +writer. This introduces a shared lock across journaled mutations: capacity should +be measured for the application, especially transactions that also perform slow +work. Replay reads a consistent database snapshot, releases that reader, then +rechecks current authorization independently for each item. + +By default the journal has a global ceiling of **4096 entries and 4 MiB of encoded frames**, +with **64 KiB per frame** and **24-hour retention**. Build-time controls are +`-Dreplay-max-entries` (1..65536), `-Dreplay-max-bytes` (1..1073741824), +`-Dreplay-max-frame-bytes` (1..1048576, no greater than total bytes), and +`-Dreplay-retention-seconds` (1..31536000). Successful replay responses report +these effective budgets as `retention: {maxEntries,maxBytes,maxFrameBytes,seconds}`. +All writer replicas must use the same budgets. Expiry is stamped when captured; +changing retention does not recompute existing event expiry. Lowering size budgets +prunes on the next write; lowering frame budgets can reject older larger frames, +requiring a new checkpoint and snapshot. `/api/meta` exposes `realtimeBackfill` +and `durableRealtime` capability booleans, distinguishing the two modes. +These budgets bound +retained logical data, not total database/WAL size or process RSS. Indexes, database +pages and transient serialization/authorization memory add overhead. Each write +prunes expired entries and evicts the oldest prefix until both size budgets fit; +idle expired rows are reclaimed on the next write. Cursors represent the last consumed +sequence: a cursor strictly below the discarded/expired prefix requires reset, while +a cursor equal to its last sequence is still caught up. This also keeps fresh head +checkpoints usable on an idle journal after every retained entry expires. Reads reject stale checkpoints +even before physical cleanup. Clock differences can shorten useful retention; +keep replica clocks synchronized. A backwards clock never revives a pruned cursor. +Pages copy at most 128 frames (8 MiB encoded with default frame budget), plus temporary database and JSON +storage. Admission/concurrency budgets remain application responsibilities. + +Unlike the local store, retention is shared across collections: a busy collection +can cause another collection's checkpoint to require reset. Invalid, expired, +evicted, collection-renamed or wrong-collection checkpoints return the same `409 resetRequired` +envelope below. Obtain a new checkpoint before reloading a snapshot. A cursorless +request does not reserve retention. Snapshot reload and idempotent application of +invalidations are still required; WS/SSE messages do not carry journal cursors. + +Enable this option on **every writer** in the deployment. Writes from older or +non-enabled binaries, raw SQL, migrations, the `Data` facade, hook side-writes, +authentication helpers and custom channels are outside this journal's REST scope. +Changing capture coverage requires clients to discard checkpoints and reload; +the journal cannot detect writes performed by code that bypasses it. This is durable +REST invalidation recovery, not exactly-once delivery or an audit/event-sourcing log. +Database backups must include the journal tables. Restoring or rewinding a database +requires clients to discard checkpoints and reload snapshots; restored history can +reuse earlier journal positions. Before serving a restored database, with all +instances stopped, replace `_replay_state.secret` with a newly generated +cryptographically random 32-byte secret encoded as 64 hexadecimal characters. +This forces old cursors to return `409`, including clients unaware of the restore. Normal database durability settings +apply. Startup creates `_replay_state` and `_replay_events` in SQLite `main` or the +PostgreSQL current schema; internal SQL qualifies that schema so temporary tables +and search-path fallbacks cannot redirect journal operations. + ### Optional record invalidation backfill (SQLite) Build with `-Drealtime-backfill=true` to register @@ -1791,6 +1864,8 @@ full, only the work-count rejection is counted because it is checked first. "oauth2": true, "postgres": false, "queryWorkbench": false, + "realtimeBackfill": false, + "durableRealtime": false, "s3": false, "senders": true, "tenancy": true, @@ -1801,6 +1876,7 @@ full, only the work-count rejection is counted because it is checked first. "health": "/api/health", "state": "/api/state", "realtimeSse": "/api/realtime/sse", + "realtimeBackfill": null, "queryWorkbench": null }, "limits": { @@ -1837,6 +1913,8 @@ binary carries: | `mailWebhook` | The mail bounce/complaint webhook route is mounted. | | `oauth2` | The OAuth2+PKCE auth method route group is mounted. | | `postgres` | The binary was compiled with `-Dpostgres` (the pure-Zig PostgreSQL backend is linked in). | +| `realtimeBackfill` | The backfill route is compiled in (process-local SQLite unless durableRealtime is true). | +| `durableRealtime` | Transactional REST replay is enabled, with database-backed cross-instance checkpoints. | | `s3` | The binary was compiled with `-Ds3` (the S3-compatible storage backend is linked in). | | `senders` | The verified-senders email route group is mounted. | | `tenancy` | Multi-tenancy is configured (`App(.{ .tenancy = ... })`). | diff --git a/docs/framework.md b/docs/framework.md index b75bb893..d9706a1f 100644 --- a/docs/framework.md +++ b/docs/framework.md @@ -6153,6 +6153,11 @@ code to comptime-dead when off, so a build that doesn't need a feature doesn't p | `-Dvector` | off | Opt-in nearest-neighbor `?vector=` KNN search — sqlite-vec on SQLite, pgvector on Postgres. → [docs/search.md](./search.md#vector-search-opt-in) | | `-Dpostgres` | off | Opt-in pure-Zig PostgreSQL wire-protocol backend, alongside the default SQLite one. → [docs/postgres.md](./postgres.md) | | `-Dfile-inventory` | off | Read-only `files inventory` plus offline local/SQLite `files reconcile` (dry-run unless `--apply`). Bounded pages; no HTTP surface. Inventory supports S3 with `-Ds3=true`; reconciliation refuses S3/PostgreSQL/custom storage. The boot-lifetime local-storage maintenance lease remains in default builds. | +| `-Ddurable-realtime` | off | Transactional built-in REST record replay on SQLite/PostgreSQL; implies realtime-backfill. Shared 4096-entry / 4 MiB journal, 64 KiB frame ceiling, 24-hour retention; restart/cross-instance checkpoints and current authorization. Raw SQL/Data/hook side-writes are outside capture scope. See API realtime section. | +| `-Dreplay-max-entries` | 4096 | Durable journal global entry budget, 1..65536. All replicas must agree. | +| `-Dreplay-max-bytes` | 4194304 | Durable journal encoded-frame byte budget, 1..1073741824. | +| `-Dreplay-max-frame-bytes` | 65536 | Maximum individual durable frame bytes, 1..1048576, at most total byte budget. | +| `-Dreplay-retention-seconds` | 86400 | Durable event lifetime stamped on capture, 1..31536000 seconds; physical expiry cleanup occurs on writes. | | `-Drealtime-backfill` | off | Single-process SQLite record invalidation backfill: 16 lazy collection slots, each 256 entries / 64 KiB (1 MiB encoded total), current authorization, explicit reset on gaps or slot replacement. No historical payloads or durable/cross-instance guarantee. See the API realtime section. | | `-Dimage-thumbnails` | off | Named PNG/JPEG/WebP derivatives on built-in local storage via a trusted external ImageMagick executable; configure `.files.thumbnails`. Subprocess support, routes and admission state are excluded when off; no image codec is linked. See [thumbnails](thumbnails.md). | | `-Dresumable-uploads` | off | Principal-bound network-resume for file fields on existing records. Process-local by default, fully buffered, configurable session/byte/chunk/expiry budgets. SQLite restart persistence requires the additional `-Ddurable-resumable-uploads` flag and `.files.resumable.durable = true`; neither mode provides cross-instance durability. See [resumable uploads](resumable-uploads.md). | diff --git a/src/api/meta.zig b/src/api/meta.zig index 71c29bc0..335e5857 100644 --- a/src/api/meta.zig +++ b/src/api/meta.zig @@ -47,6 +47,8 @@ pub fn handle(ctx: *http.RequestCtx) !http.Response { .oauth2 = g.oauth2, .postgres = build_options.postgres, .queryWorkbench = build_options.query_workbench, + .realtimeBackfill = build_options.realtime_backfill, + .durableRealtime = build_options.durable_realtime, .s3 = build_options.s3, .senders = g.senders, .tenancy = g.tenancy, @@ -57,6 +59,7 @@ pub fn handle(ctx: *http.RequestCtx) !http.Response { .health = "/api/health", .state = app.features_public_route, .realtimeSse = "/api/realtime/sse", + .realtimeBackfill = if (build_options.realtime_backfill) "/api/realtime/backfill" else @as(?[]const u8, null), .queryWorkbench = if (build_options.query_workbench) "/api/query-workbench/stats" else @as(?[]const u8, null), }, .limits = .{ .maxUploadSize = app.max_upload_size }, diff --git a/src/api/realtime_backfill.zig b/src/api/realtime_backfill.zig index 8f0cae42..d75cbbaa 100644 --- a/src/api/realtime_backfill.zig +++ b/src/api/realtime_backfill.zig @@ -16,8 +16,8 @@ pub fn get(ctx: *http.RequestCtx) !http.Response { const app = ctx.app orelse return ApiError.notFound().toResponse(a); // LISTEN/NOTIFY is best-effort, not a durable ordered log. Do not imply a // complete backfill on PostgreSQL even when requests stick to one instance. - if (db.poolBackend(app.pool) != .sqlite) return unsupported_backend.toResponse(a); - const store = app.backfill orelse return ApiError.notFound().toResponse(a); + if (!@import("build_options").durable_realtime and db.poolBackend(app.pool) != .sqlite) return unsupported_backend.toResponse(a); + if (!@import("build_options").durable_realtime and app.backfill == null) return ApiError.notFound().toResponse(a); if (ctx.query.len > 4096) return ApiError.badRequest("Backfill query too large.").toResponse(a); const params = try @import("../query/params.zig").parse(a, ctx.query); defer params.deinit(a); @@ -45,16 +45,36 @@ pub fn get(ctx: *http.RequestCtx) !http.Response { if (col == null) return ApiError.notFound().toResponse(a); if (hub.subscribeCheck(app, &conn, ctx.allocator, input.topic) != .ok) return ApiError.withCode(403, .forbidden, "Subscription denied.").toResponse(a); const collection = col.?; + const cursor_scope = if (comptime @import("build_options").durable_realtime) + try std.fmt.allocPrint(a, "{s}:{d}", .{ collection.id, collection.rename_epoch }) + else + collection.id; const checkpoint: ?[]const u8 = if (input.cursor) |cursor| blk: { - if (cursor.len <= collection.id.len or !std.mem.startsWith(u8, cursor, collection.id) or cursor[collection.id.len] != ':') return resetRequired(a); - break :blk cursor[collection.id.len + 1 ..]; + if (cursor.len <= cursor_scope.len or !std.mem.startsWith(u8, cursor, cursor_scope) or cursor[cursor_scope.len] != ':') return resetRequired(a); + break :blk cursor[cursor_scope.len + 1 ..]; } else null; - const page = store.page(ctx.allocator, collection.id, checkpoint, input.limit) catch |err| switch (err) { - error.ResetRequired => return resetRequired(a), - else => return err, + const result = blk: { + if (comptime @import("build_options").durable_realtime) { + var journal_reader = try app.pool.acquireReader(); + defer app.pool.releaseReader(&journal_reader); + const page = @import("../realtime/durable.zig").readPage(ctx.allocator, app.io, &journal_reader, collection.id, cursor_scope, checkpoint, input.limit, @import("../clock.zig").nowUnix(app.io)) catch |err| switch (err) { + error.ResetRequired, error.ReplayFrameTooLarge => return resetRequired(a), + else => return err, + }; + break :blk page; + } else { + const store = app.backfill.?; + const page = store.page(ctx.allocator, collection.id, checkpoint, input.limit) catch |err| switch (err) { + error.ResetRequired => return resetRequired(a), + else => return err, + }; + const entries = try a.alloc(@import("../realtime/durable.zig").Entry, page.items.len); + for (page.items, entries) |source, *dest| dest.* = .{ .frame = source.frame }; + break :blk @import("../realtime/durable.zig").Page{ .items = entries, .position = try store.cursor(a, page), .has_next = page.has_next }; + } }; var items: std.ArrayList(std.json.Value) = .empty; - for (page.items) |e| { + for (result.items) |e| { // Deliberately revalidate identity and reacquire current collection metadata // per item: a revocation committed mid-page discards the entire response. // A request-wide identity/schema snapshot would weaken that guarantee. @@ -64,9 +84,13 @@ pub fn get(ctx: *http.RequestCtx) !http.Response { try items.append(a, value); } } - const position = try store.cursor(a, page); - const next_cursor = try std.fmt.allocPrint(a, "{s}:{s}", .{ collection.id, position }); - return .{ .status = 200, .body = try std.json.Stringify.valueAlloc(a, .{ .items = items.items, .nextCursor = next_cursor, .hasNext = page.has_next, .resetRequired = false }, .{}) }; + const position = result.position; + const next_cursor = try std.fmt.allocPrint(a, "{s}:{s}", .{ cursor_scope, position }); + if (comptime @import("build_options").durable_realtime) { + const durable = @import("../realtime/durable.zig"); + return .{ .status = 200, .body = try std.json.Stringify.valueAlloc(a, .{ .items = items.items, .nextCursor = next_cursor, .hasNext = result.has_next, .resetRequired = false, .retention = .{ .maxEntries = durable.max_entries, .maxBytes = durable.max_bytes, .maxFrameBytes = durable.max_frame_bytes, .seconds = durable.retention_seconds } }, .{}) }; + } + return .{ .status = 200, .body = try std.json.Stringify.valueAlloc(a, .{ .items = items.items, .nextCursor = next_cursor, .hasNext = result.has_next, .resetRequired = false }, .{}) }; } /// The shared delivery path can fail after earlier items were authorized. Return diff --git a/src/api/records.zig b/src/api/records.zig index 04adfa8f..dcc4fa02 100644 --- a/src/api/records.zig +++ b/src/api/records.zig @@ -471,6 +471,7 @@ pub fn create(ctx: *http.RequestCtx) anyerror!http.Response { } } const rid = rec.object.get("id").?.string; + if (comptime @import("build_options").durable_realtime) try @import("../realtime/durable.zig").capture(ctx.allocator.a, app.io, w, col, .create, rid, null); // Bytes already exist; only a successful commit makes their references visible. try w.commit(); committed = true; @@ -749,6 +750,7 @@ fn updateImpl(ctx: *http.RequestCtx, comptime is_resumable: bool, resumable: if return ApiError.withCode(503, .internal, "Upload completion receipt could not be persisted; inspect upload status.").toResponse(ctx.allocator.a); }; }; + if (comptime @import("build_options").durable_realtime) try @import("../realtime/durable.zig").capture(ctx.allocator.a, app.io, w, col, .update, rid, null); try w.commit(); txn_open = false; committed = true; // row is durable — the write-cleanup defer must NOT fire past here @@ -829,6 +831,7 @@ pub fn delete(ctx: *http.RequestCtx) anyerror!http.Response { // create/update path (all compare ciphertext) — plus the cross-instance NOTIFY token (Postgres // only; the wire carries only the token, never the row data). On SQLite with no encrypted // fields this reuses `ex_mut` with no extra read (byte-identical). + const durable_snapshot = if (comptime @import("build_options").durable_realtime) try records.getAtRest(ctx.allocator.a, w, col, rid) else null; const rt = realtime_ws.prepareDelete(ctx.allocator.a, app, w, col, rid, ex_mut); if (!try records.deleteInTxn(ctx.allocator.a, w, col, rid)) { return ApiError.notFound().toResponse(ctx.allocator.a); @@ -841,6 +844,7 @@ pub fn delete(ctx: *http.RequestCtx) anyerror!http.Response { _ = try st.step(); } if (app.files.cleanup) |enqueue| try enqueue(ctx.allocator.a, w, app.io, app.files.cleanup_queue.?, col, rid, existing, null); + if (comptime @import("build_options").durable_realtime) try @import("../realtime/durable.zig").capture(ctx.allocator.a, app.io, w, col, .delete, rid, durable_snapshot); try w.commit(); txn_open = false; @@ -1047,6 +1051,7 @@ const TestEnv = struct { const w = env.pool.acquireWriter(); defer env.pool.releaseWriter(); try migrations.run(w); + if (comptime @import("build_options").durable_realtime) try @import("../realtime/durable.zig").initialize(std.testing.allocator, std.testing.io, w); var setup_arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer setup_arena.deinit(); const sa = setup_arena.allocator(); @@ -2395,6 +2400,7 @@ test "update pre-authorizes tenant ownership BEFORE the before_update hook fires const w = pool.acquireWriter(); defer pool.releaseWriter(); try migrations.run(w); + if (comptime @import("build_options").durable_realtime) try @import("../realtime/durable.zig").initialize(std.testing.allocator, std.testing.io, w); _ = try collections.create(a, std.testing.io, w, .{ .id = "", .name = "users", .type = .auth, .fields = &.{} }); const users_col = (try collections.get(a, w, "users")).?; var ud: std.json.ObjectMap = .empty; @@ -2521,6 +2527,7 @@ const F1Env = struct { const w = env.pool.acquireWriter(); defer env.pool.releaseWriter(); try migrations.run(w); + if (comptime @import("build_options").durable_realtime) try @import("../realtime/durable.zig").initialize(std.testing.allocator, std.testing.io, w); _ = try collections.create(a, std.testing.io, w, .{ .id = "", .name = "users", diff --git a/src/framework.zig b/src/framework.zig index 4ece36e0..842716d5 100644 --- a/src/framework.zig +++ b/src/framework.zig @@ -5145,7 +5145,12 @@ fn bootApp( } else &.{}; errdefer static_files.freeSpaRoots(allocator, holder.spa_roots); - holder.backfill_store = if (comptime build_options.realtime_backfill) try @import("realtime/backfill.zig").Store.create(allocator, io, db.poolBackend(&holder.pool)) else {}; + if (comptime build_options.durable_realtime) { + const writer = holder.pool.acquireWriter(); + defer holder.pool.releaseWriter(); + try @import("realtime/durable.zig").initialize(allocator, io, writer); + } + holder.backfill_store = if (comptime build_options.realtime_backfill) (if (comptime build_options.durable_realtime) null else try @import("realtime/backfill.zig").Store.create(allocator, io, db.poolBackend(&holder.pool))) else {}; errdefer if (comptime build_options.realtime_backfill) if (holder.backfill_store) |store| store.destroy(); holder.admission_state = if (comptime opts.admission_config) |cfg_admission| @import("admission.zig").State.init(io, cfg_admission) else {}; holder.resumable_store = if (comptime build_options.resumable_uploads) try @import("files/resumable.zig").Store.create(allocator, io, opts.files.resumable) else {}; diff --git a/src/realtime/durable.zig b/src/realtime/durable.zig new file mode 100644 index 00000000..19f841ef --- /dev/null +++ b/src/realtime/durable.zig @@ -0,0 +1,347 @@ +//! Transactional REST invalidation journal. Live fanout remains best effort. +const std = @import("std"); +const db = @import("../db.zig"); +const protocol = @import("protocol.zig"); +const schema = @import("../schema.zig"); +const Arena = @import("../request_arena.zig").RequestArena; +const options = @import("build_options"); +pub const max_entries = options.replay_max_entries; +pub const max_bytes = options.replay_max_bytes; +pub const max_frame_bytes = options.replay_max_frame_bytes; +pub const retention_seconds = options.replay_retention_seconds; +pub const Entry = struct { frame: []const u8 }; +pub const Page = struct { items: []const Entry, position: []const u8, has_next: bool }; + +fn prep(a: std.mem.Allocator, conn: *db.Db, sql: []const u8) !db.Stmt { + const pg = db.dbDialect(conn).kind == .postgres; + const prefix = if (pg) blk: { + var ns = try conn.prepare("SELECT pg_catalog.quote_ident(current_schema()), current_schema();"); + defer ns.finalize(); + if (!try ns.step() or ns.isNull(0) or std.mem.startsWith(u8, ns.columnText(1), "pg_")) return error.InvalidReplaySchema; + break :blk try std.fmt.allocPrint(a, "{s}.", .{ns.columnText(0)}); + } else try a.dupe(u8, "main."); + defer a.free(prefix); + // Rewrite only engine-owned placeholders. A quoted PostgreSQL schema may + // legally contain '?' or quotes that the placeholder scanner does not parse. + const numbered = try @import("../sql/param_sink.zig").renumber(a, db.dbDialect(conn), sql); + defer if (pg) a.free(numbered); + const qualified = try std.mem.replaceOwned(u8, a, numbered, "journal.", prefix); + defer a.free(qualified); + const z = try a.dupeZ(u8, qualified); + defer a.free(z); + return conn.prepare(z); +} +fn execute(a: std.mem.Allocator, conn: *db.Db, sql: []const u8) !void { + var s = try prep(a, conn, sql); + defer s.finalize(); + _ = try s.step(); +} +pub fn initialize(a: std.mem.Allocator, io: std.Io, conn: *db.Db) !void { + if (conn.inTransaction()) return error.NestedTransaction; + try conn.beginImmediate(); + errdefer conn.rollback() catch |err| std.log.warn("replay initialization rollback: {s}", .{@errorName(err)}); + if (db.dbDialect(conn).kind == .postgres) try conn.exec("SELECT pg_advisory_xact_lock(20903, 419);"); + try execute(a, conn, "CREATE TABLE IF NOT EXISTS journal._replay_state (id INTEGER PRIMARY KEY, secret TEXT NOT NULL, sequence BIGINT NOT NULL, floor BIGINT NOT NULL);"); + try execute(a, conn, "CREATE TABLE IF NOT EXISTS journal._replay_events (sequence BIGINT PRIMARY KEY, collection TEXT NOT NULL, frame TEXT NOT NULL, bytes BIGINT NOT NULL, expires BIGINT NOT NULL);"); + if (db.dbDialect(conn).kind == .postgres) { + var valid = try conn.prepare("SELECT count(*) FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname=current_schema() AND c.relname IN ('_replay_state','_replay_events') AND c.relkind='r' AND c.relpersistence='p';"); + defer valid.finalize(); + _ = try valid.step(); + if (valid.columnInt(0) != 2) return error.InvalidReplayLedger; + } else { + var valid = try conn.prepare("PRAGMA main.table_list;"); + defer valid.finalize(); + var count: usize = 0; + while (try valid.step()) { + if (!std.mem.eql(u8, valid.columnText(1), "_replay_state") and !std.mem.eql(u8, valid.columnText(1), "_replay_events")) continue; + if (!std.mem.eql(u8, valid.columnText(2), "table")) return error.InvalidReplayLedger; + count += 1; + } + if (count != 2) return error.InvalidReplayLedger; + } + var random_secret: [32]u8 = undefined; + io.random(&random_secret); + const secret = std.fmt.bytesToHex(random_secret, .lower); + var insert = try prep(a, conn, "INSERT INTO journal._replay_state VALUES(1,?1,0,0) ON CONFLICT(id) DO NOTHING;"); + defer insert.finalize(); + try insert.bindText(1, &secret); + _ = try insert.step(); + try conn.commit(); +} + +pub fn capture(a: std.mem.Allocator, io: std.Io, conn: *db.Db, col: schema.Collection, action: protocol.Action, id: []const u8, snapshot: ?std.json.Value) !void { + var record: std.json.ObjectMap = .empty; + defer record.deinit(a); + try record.put(a, "id", .{ .string = id }); + if (action == .delete) try record.put(a, @import("hub.zig").delete_snapshot_key, snapshot orelse return error.MissingDeleteSnapshot); + const frame = try protocol.serializeEvent(a, col.name, action, .{ .object = record }); + defer a.free(frame); + try append(a, conn, col.id, frame, @import("../clock.zig").nowUnix(io)); +} + +/// Caller owns transaction. The metadata row lock is held through its commit, +/// so another process cannot allocate a visible position ahead of this write. +pub fn append(a: std.mem.Allocator, conn: *db.Db, collection: []const u8, frame: []const u8, now: i64) !void { + if (!conn.inTransaction()) return error.TransactionRequired; + if (frame.len > max_frame_bytes) return error.ReplayFrameTooLarge; + if (now < 0 or now > @as(i64, std.math.maxInt(i64)) - retention_seconds) return error.InvalidTime; + try execute(a, conn, "UPDATE journal._replay_state SET sequence=sequence+1 WHERE id=1 AND sequence<9223372036854775807;"); + if (conn.changesCount() != 1) return error.ReplaySequenceExhausted; + var insert = try prep(a, conn, "INSERT INTO journal._replay_events SELECT sequence,?1,?2,?3,?4 FROM journal._replay_state WHERE id=1;"); + defer insert.finalize(); + try insert.bindText(1, collection); + try insert.bindText(2, frame); + try insert.bindInt(3, @intCast(frame.len)); + try insert.bindInt(4, now + retention_seconds); + _ = try insert.step(); + // Bound work by the fixed journal ceiling, independent of application size. + // Expiry follows the sequence prefix even if server clocks move backwards. + var expired = try prep(a, conn, "UPDATE journal._replay_state SET floor=COALESCE((SELECT MAX(sequence) FROM journal._replay_events WHERE expires<=?1),floor) WHERE id=1;"); + defer expired.finalize(); + try expired.bindInt(1, now); + _ = try expired.step(); + try execute(a, conn, "DELETE FROM journal._replay_events WHERE sequence<=(SELECT floor FROM journal._replay_state WHERE id=1);"); + // One bounded reverse window pass chooses the oldest prefix to evict. + // A large frame cannot trigger thousands of repeated COUNT/SUM rescans. + var budget = try prep(a, conn, "UPDATE journal._replay_state SET floor=COALESCE((SELECT MAX(sequence) FROM (SELECT sequence, ROW_NUMBER() OVER (ORDER BY sequence DESC) AS n, SUM(bytes) OVER (ORDER BY sequence DESC) AS b FROM journal._replay_events) AS budget WHERE n>?1 OR b>?2),floor) WHERE id=1;"); + defer budget.finalize(); + try budget.bindInt(1, max_entries); + try budget.bindInt(2, max_bytes); + _ = try budget.step(); + try execute(a, conn, "DELETE FROM journal._replay_events WHERE sequence<=(SELECT floor FROM journal._replay_state WHERE id=1);"); +} + +/// A consistent read transaction protects metadata/rows against concurrent prune. +/// Current authorization is applied by the HTTP layer AFTER releasing this reader. +pub fn readPage(arena: Arena, io: std.Io, conn: *db.Db, collection: []const u8, binding: []const u8, cursor: ?[]const u8, limit: usize, now: i64) !Page { + if (limit == 0 or limit > 128) return error.BadLimit; + if (conn.inTransaction()) return error.NestedTransaction; + if (db.dbDialect(conn).kind == .postgres) try conn.exec("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY;") else try conn.exec("BEGIN;"); + errdefer conn.rollback() catch |err| std.log.warn("replay read rollback: {s}", .{@errorName(err)}); + const a = arena.a; + var state = try prep(a, conn, "SELECT secret,sequence,floor,COALESCE((SELECT MAX(sequence) FROM journal._replay_events WHERE expires<=?1),0) FROM journal._replay_state WHERE id=1;"); + defer state.finalize(); + try state.bindInt(1, now); + if (!try state.step()) return error.ReplayUnavailable; + var key: [32]u8 = undefined; + if (state.columnText(0).len != 64) return error.InvalidReplaySecret; + _ = std.fmt.hexToBytes(&key, state.columnText(0)) catch return error.InvalidReplaySecret; + const head = state.columnInt(1); + const floor = @max(state.columnInt(2), state.columnInt(3)); + state.reset(); + var after = head; + if (cursor) |c| { + after = try openCursor(key, binding, c); + // Cursors mean "after this sequence". Equality has missed no event; + // rejecting it would also invalidate a fresh checkpoint on an idle journal. + if (after < floor or after > head) return error.ResetRequired; + } + var rows = try prep(a, conn, if (db.dbDialect(conn).kind == .postgres) + "SELECT sequence,CASE WHEN octet_length(frame)<=?4 THEN frame ELSE NULL END FROM journal._replay_events WHERE sequence>?1 AND collection=?2 ORDER BY sequence LIMIT ?3;" + else + "SELECT sequence,CASE WHEN length(CAST(frame AS BLOB))<=?4 THEN frame ELSE NULL END FROM journal._replay_events WHERE sequence>?1 AND collection=?2 ORDER BY sequence LIMIT ?3;"); + defer rows.finalize(); + try rows.bindInt(1, after); + try rows.bindText(2, collection); + try rows.bindInt(3, @intCast(limit + 1)); + try rows.bindInt(4, max_frame_bytes); + var items: std.ArrayList(Entry) = .empty; + var next = head; + var more = false; + var last = after; + while (try rows.step()) { + if (items.items.len == limit) { + next = last; + more = true; + break; + } + if (rows.isNull(1)) return error.ReplayFrameTooLarge; + const frame = rows.columnText(1); + if (frame.len > max_frame_bytes) return error.ReplayFrameTooLarge; + try items.append(a, .{ .frame = try a.dupe(u8, frame) }); + last = rows.columnInt(0); + } + rows.reset(); + const position = try sealCursor(a, io, key, binding, next); + try conn.commit(); + return .{ .items = items.items, .position = position, .has_next = more }; +} + +const Cipher = std.crypto.aead.chacha_poly.XChaCha20Poly1305; +const cursor_bytes = Cipher.nonce_length + 8 + Cipher.tag_length; + +fn sealCursor(a: std.mem.Allocator, io: std.Io, key: [32]u8, binding: []const u8, sequence: i64) ![]u8 { + var token: [cursor_bytes]u8 = undefined; + const nonce = token[0..Cipher.nonce_length]; + io.random(nonce); + var plaintext: [8]u8 = undefined; + std.mem.writeInt(i64, &plaintext, sequence, .big); + Cipher.encrypt(token[Cipher.nonce_length..][0..8], token[Cipher.nonce_length + 8 ..][0..Cipher.tag_length], &plaintext, binding, nonce.*, key); + const encoded = std.fmt.bytesToHex(token, .lower); + return a.dupe(u8, &encoded); +} + +fn openCursor(key: [32]u8, binding: []const u8, cursor: []const u8) !i64 { + if (cursor.len != cursor_bytes * 2) return error.ResetRequired; + var token: [cursor_bytes]u8 = undefined; + _ = std.fmt.hexToBytes(&token, cursor) catch return error.ResetRequired; + var plaintext: [8]u8 = undefined; + Cipher.decrypt(&plaintext, token[Cipher.nonce_length..][0..8], token[Cipher.nonce_length + 8 ..][0..Cipher.tag_length].*, binding, token[0..Cipher.nonce_length].*, key) catch return error.ResetRequired; + return std.mem.readInt(i64, &plaintext, .big); +} + +fn page(arena: Arena, conn: *db.Db, collection: []const u8, cursor: ?[]const u8, limit: usize, now: i64) !Page { + return readPage(arena, std.testing.io, conn, collection, collection, cursor, limit, now); +} + +test "journal transaction rollback pagination expiry and hard budgets" { + var conn = try db.Db.openMemory(); + defer conn.close(); + try exercise(&conn); +} + +fn exercise(conn: *db.Db) !void { + const a = std.testing.allocator; + var scratch = std.heap.ArenaAllocator.init(a); + defer scratch.deinit(); + const arena = Arena{ .a = scratch.allocator() }; + try initialize(a, std.testing.io, conn); + const initial = try page(arena, conn, "one", null, 2, 100); + try std.testing.expectError(error.TransactionRequired, append(a, conn, "one", "{}", 100)); + try conn.beginImmediate(); + try append(a, conn, "one", "rolled-back", 100); + try conn.rollback(); + try std.testing.expectEqual(@as(usize, 0), (try page(arena, conn, "one", initial.position, 2, 100)).items.len); + try conn.beginImmediate(); + try append(a, conn, "one", "first", 100); + try append(a, conn, "other", "private", 100); + try append(a, conn, "one", "second", 100); + try conn.commit(); + // Reinitialization preserves both epoch and rows across owner lifetimes. + try initialize(a, std.testing.io, conn); + const first = try page(arena, conn, "one", initial.position, 1, 100); + try std.testing.expect(first.has_next); + try std.testing.expectEqualStrings("first", first.items[0].frame); + const second = try page(arena, conn, "one", first.position, 1, 100); + try std.testing.expect(!second.has_next); + try std.testing.expectEqualStrings("second", second.items[0].frame); + try std.testing.expectError(error.ResetRequired, page(arena, conn, "one", initial.position, 1, 100 + retention_seconds)); + // At the expiry floor, all expired events were already consumed. Both the + // caught-up cursor and a newly issued head checkpoint must remain usable. + const caught_up = try page(arena, conn, "one", second.position, 1, 100 + retention_seconds); + try std.testing.expectEqual(@as(usize, 0), caught_up.items.len); + const fresh = try page(arena, conn, "one", null, 1, 100 + retention_seconds); + try std.testing.expectEqual(@as(usize, 0), (try page(arena, conn, "one", fresh.position, 1, 100 + retention_seconds)).items.len); + const oversized = try a.alloc(u8, max_frame_bytes + 1); + defer a.free(oversized); + try conn.beginImmediate(); + try std.testing.expectError(error.ReplayFrameTooLarge, append(a, conn, "one", oversized, 100)); + try conn.rollback(); + const large = oversized[0..max_frame_bytes]; + @memset(large, 'x'); + for (0..max_bytes / max_frame_bytes + 1) |_| { + try conn.beginImmediate(); + try append(a, conn, "one", large, 100); + try conn.commit(); + } + try std.testing.expectError(error.ResetRequired, page(arena, conn, "one", second.position, 1, 100)); + var size = try prep(a, conn, "SELECT COUNT(*),SUM(bytes) FROM journal._replay_events;"); + defer size.finalize(); + _ = try size.step(); + try std.testing.expect(size.columnInt(0) <= max_entries); + try std.testing.expect(size.columnInt(1) <= max_bytes); +} + +test "pg journal isolates schema and serializes visibility across connections" { + if (comptime !@import("build_options").postgres) return error.SkipZigTest; + const url = std.testing.environ.getPosix("ZIGBASE_PG_TEST_URL") orelse return error.SkipZigTest; + const a = std.testing.allocator; + var first = try db.Db.openPostgres(a, std.testing.io, url); + defer first.close(); + var other = try db.Db.openPostgres(a, std.testing.io, url); + defer other.close(); + const token = try @import("../crypto.zig").genToken(std.testing.io, a, 12); + defer a.free(token); + const name = try std.fmt.allocPrint(a, "replay?\"'{s}", .{token}); + defer a.free(name); + const quoted = try @import("../ddl.zig").quoteIdent(a, name); + defer a.free(quoted); + const create = try std.fmt.allocPrintSentinel(a, "CREATE SCHEMA {s};", .{quoted}, 0); + defer a.free(create); + const drop = try std.fmt.allocPrintSentinel(a, "DROP SCHEMA {s} CASCADE;", .{quoted}, 0); + defer a.free(drop); + const path = try std.fmt.allocPrintSentinel(a, "SET search_path TO {s};", .{quoted}, 0); + defer a.free(path); + try first.exec(create); + defer first.exec(drop) catch |err| std.log.err("replay test cleanup: {s}", .{@errorName(err)}); + try first.exec(path); + try other.exec(path); + try exercise(&first); + var scratch = std.heap.ArenaAllocator.init(a); + defer scratch.deinit(); + const arena = Arena{ .a = scratch.allocator() }; + const checkpoint = try page(arena, &other, "cross", null, 1, 100); + try first.beginImmediate(); + try append(a, &first, "cross", "committed-first", 100); + // Readers never see an uncommitted sequence. A second writer cannot skip it. + try std.testing.expectEqual(@as(usize, 0), (try page(arena, &other, "cross", checkpoint.position, 1, 100)).items.len); + try other.exec("SET lock_timeout='100ms';"); + try other.beginImmediate(); + try std.testing.expectError(error.StepFailed, append(a, &other, "cross", "must-not-commit", 100)); + try std.testing.expectEqualStrings("canceling statement due to lock timeout", other.errMsg()); + try other.rollback(); + try first.commit(); + try other.beginImmediate(); + try append(a, &other, "cross", "committed-second", 100); + try other.commit(); + const replay = try page(arena, &other, "cross", checkpoint.position, 2, 100); + try std.testing.expectEqual(@as(usize, 2), replay.items.len); + try std.testing.expectEqualStrings("committed-first", replay.items[0].frame); + try std.testing.expectEqualStrings("committed-second", replay.items[1].frame); + try first.exec("ALTER TABLE _replay_events SET UNLOGGED;"); + try std.testing.expectError(error.InvalidReplayLedger, initialize(a, std.testing.io, &first)); + try first.exec("ALTER TABLE _replay_events SET LOGGED;"); + try first.exec("CREATE TEMP TABLE _replay_events(sequence INTEGER);"); + // Explicit current-schema qualification ignores temporary shadows. + try initialize(a, std.testing.io, &first); + try std.testing.expectEqual(@as(usize, 2), (try page(arena, &first, "cross", checkpoint.position, 2, 100)).items.len); +} + +test "journal count budget evicts oldest entry and preserves newest prefix" { + const a = std.testing.allocator; + var conn = try db.Db.openMemory(); + defer conn.close(); + try initialize(a, std.testing.io, &conn); + try conn.beginImmediate(); + var fill = try prep(a, &conn, "WITH RECURSIVE n(v) AS (SELECT 1 UNION ALL SELECT v+1 FROM n WHERE v