diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 90fa173..47389e9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,8 +61,13 @@ jobs: - run: npm run contract:check - run: npm test - run: npm run test:analysis-integration + - name: Configure bounded RAM Redis fixture + run: | + docker exec '${{ job.services.redis.id }}' redis-cli CONFIG SET maxmemory 96mb + docker exec '${{ job.services.redis.id }}' redis-cli CONFIG SET maxmemory-policy noeviction - run: npm run test:ingest-integration - run: npm run build + - run: npm run test:ram-operational frontend: if: github.event_name != 'schedule' diff --git a/FIX-PLAN.md b/FIX-PLAN.md index 2811961..4431f6b 100644 --- a/FIX-PLAN.md +++ b/FIX-PLAN.md @@ -1,146 +1,76 @@ -# UKMesh ingestion repair plan — 2026-10-11 +# UKMesh ingestion review repair plan — 2026-10-11 -## Current authorization/read wave +Specification: `/home/ben/ukmesh-fix-context-20261011/SOL-REVIEW.md`. +The full F01–F17 mission remains active. This current plan supersedes the +historical handoff tables, which remain available in git history. -The working tree adds F01 generation fences to SQL aggregates, WS initial -cache delivery and bounded live batches. F12/F13 parity now includes the -requested SQL page plus aggregates for captured overlap hashes, final paging -after merge, canonical aliases, stable-ID counts, representative ranking and -millisecond/tie ordering. The real fixture also found an F05 interaction: -natural-identity replay suppression must apply only to entries without a -stable observation ID; distinct stable IDs are arbitrated by the ledger. +## Completed sequence -CoreScope public mode uses a dedicated revocable SQLite projection and private -ingestor replay journal. The read-only server checks upstream authority before -and after API reads and before WS fanout. Generation changes rebuild copied -measurement/derived data and replace caches. Existing untracked/mixed data is -preserved and initialization refused. The filtered credentialed HTTP live -source removes the public companion's raw Redis credential; PostgreSQL's -committed-order ledger remains required. This stays within the review's -authorization, identity and handoff design. +1. Repaired #122 F03 observation identity, F04 unsupported N>1 configuration, + F14 metric registration and F15 one-deadline accounting. Perf head + `8968078` is pushed/remote-verified; full 418-test suite/typecheck and both + current CI workflows pass. Merged perf into RAM, retaining the stack. +2. Repaired #125 SQL lock/ledger mutation safety; retry-stable admission UUIDs; + per-entry retained accounting/handoff; writer supervision, decode/quarantine, + read counts/contracts and shutdown/drain. Added real refused-write, + ambiguous admission, snapshot-before-commit/expiry and group-loss gates. +3. Added historical migration 059 and its atomic coverage fence. CoreScope + committed-order completeness, cursor barriers, stable-ID derived state, + actual crash/SQLite failure and shared-schema restart repair are committed + locally. CoreScope `07f05cd` additionally fences/reconstructs copied public + data, preserves participant provenance and uses a filtered HTTP live source. + Full ingestor/server suites pass; three commits await coordinator relay. +4. RAM public/read checkpoint `c683acd` is pushed. Both workflows + `38180468916` / `38180473812` passed every job. Its strict source-labelled + image gate passed exactly 10/10, zero skips. Current policy now governs + SQL aggregates, WS caches and bounded live batches; ready WS regression + covers 800/800 allowed frames and revocation/unavailability. -Focused and real regressions pass, including the ready WS 800-message gate -and the paired crash gate through the projection. RAM typecheck and full -backend tests pass 476/476 (202.440s); integration passes 7 PG + 2 Redis + -1 WS test, with zero skips. CoreScope server passes (103.965s); its final -ingestor suite remains running. After committing/pushing and green CI, continue -F09/F14 loaded headroom/accounting, F15 held runtime shutdown and F16 actual -principal isolation/drain, then final source-labelled images and whole-stack -performance. The overall mission remains active. +## Current operational/read wave -Specification: `/home/ben/ukmesh-fix-context-20261011/SOL-REVIEW.md`, especially findings F01–F17, §§4, 7–9. This plan records implementation order and will be updated as evidence changes. The independent review takes precedence over the earlier fix checklist. +F09 startup/admission reserve 32 MiB of Redis recovery headroom; encoded +quarantine has an exact 500-record/16-KiB cap. F14 missing-group/null-memory +handling clears stale gauges and actual HTTP scrape checks known loaded states. +F15/F16 compiled processes hold real PG persistence across SIGTERM, recover the +original UUID, refuse quarantine rollback, wait through actual 120-second +retention and prove empty stream/PEL/reservation before durable mode. Actual +writer/public/quarantine Redis principals are isolated. CI owns those fixtures. +F13 preserves decrypted-summary fallback and exact JSONB summary extraction; +F01 delivery rechecks generation after asynchronous alias/summary enrichment. +New regressions reproduce old failures. Final full backend suite passes +479/479 (193.952s), typecheck/build pass, full ingest integration passes +7 PG + 3 Redis + 1 WS, and the strict actual ACL/runtime gate passes exactly +4+1 tests with zero skips (runtime 152.153s). Loaded resident-memory refusal +and all ten Compose RAM/public-source settings pass. Next commit/push and CI. -## Boundaries and sequence +## Remaining work, in order -Production and running services remain untouched. No deployment, service restart, PR merge, or CoreScope push. The bench rig's `FINAL-CHAIN-DONE` marker was observed at 17:39:52 UTC; isolated bench verification began afterward. Durable ingestion stays the default; RAM ingestion remains opt-in. Review scratch files and briefs are committed on both branches for handoff. +1. Commit/push the locally verified operational/read RAM wave and watch + its new full CI to actual success. Existing green CI + is evidence only for its matching revision. +2. Build a uniquely labelled final RAM image; pin compiled fixture sources + to that label and require exact nonzero test totals with zero skips. +3. Use newly owned isolated services/processes for a whole-stack campaign; + preserve original services and all other run directories. Replace the + old processed/XLEN drain oracle with exact expected identities, every + side-effect total, actual pending/PEL and completion barriers. Wait for + readers to be ready and stop/join each one at the recorded endpoint. +4. Audit actual paired CoreScope source/server privacy transitions after + rematerialization, late commit/offline/cursor loss and restricted public + credentials. Measure backend and whole-stack CPU/memory, Redis AOF, + PG WAL/checkpoints and physical I/O over the same completed window; + compare final-code durable/RAM workloads with repetitions. +5. Refresh both existing PR finding maps, hashes, CI links, evidence and + limitations. Keep #122 open/non-draft and #125 draft. Finalize FIX-REPORT + and the three-commit CoreScope relay patch, then verify remote UK heads, + clean local CoreScope commits and green current CI before completion. -1. On `perf/mqtt-ingest-20261011` (#122), add failing regressions, repair F03/F04 and its F14/F15 items, run TypeScript and the full backend suite, commit and push, then watch CI to success. Keep the N=1 supported default and gate N>1 pending pinned broker qualification. -2. Merge the pushed perf head into `experiment/ram-first-ingest-20261011` (#125), preserving the stack and recording the merge in both PR descriptions. Repair shared SQL safety first, then identity/privacy/handoff, then writer/read/recovery/operational contracts. Add a regression for each finding before its fix. Run the full backend and targeted integration gates, commit and push, and watch CI to success. -3. In the local CoreScope `feat/redis-live-source` branch, repair the dependent privacy, completeness, cursor, and idempotency contracts. Add failing Go regressions first, run `go test ./...` from `cmd/ingestor`, and commit clean changes for relay without pushing. -4. After the bench campaign settles, build a uniquely tagged image from the repaired source and run the isolated integration harnesses and identity-checked final-head campaign. Preserve the other campaign's logs. Update PR bodies with finding→fix map, evidence and limitations. Write `FIX-REPORT.md` with hashes, CI URLs, per-finding evidence and confirmation that production was not changed. +## Boundaries -## Finding-by-finding implementation - -| Finding | Branch | Intended repair and regression | -|---|---|---| -| F01 | #125 + CoreScope | One current-authority public-read contract with batch policy/generation/rematerialization checks, fail-closed behavior and a revocation barrier. Apply to REST merge, WS initial/live and both CoreScope paths; test observer/source/prefix flips during pending and commit, stale/unavailable authority and network scope. Never use frozen `vo` as permission. | -| F02 | #125 | Parenthesize the hourly lock gate and audit every mutation CTE. Acquire advisory lock in a separate statement on one pooled client inside an explicit Read Committed transaction, or enforce equivalent identity ledger. Concurrent PG refusal must change no table. | -| F03 | #122 | Define canonical observation/replay identity including the distinctions used by claim arbitration; use it in restart probe and SQL matching. Test lost PUBACK, older same-hash sighting, hop/path, observer and scope differences, including historical side effects. | -| F04 | #122 | Explicitly gate N>1 shared subscriptions pending restart/resize qualification while retaining N=1. Test unsupported configuration fails safely and default subscription/reconnect remains supported. | -| F05 | #125 | Create a retry-stable observation ID before the first ambiguous Redis write; atomic bounded admission ledger returns the original record on retry. Preserve distinct same-millisecond sightings and enforce identity at PG insertion. Test lost Redis reply and same-ms distinct path/hop. | -| F06 | #125 | Replace cumulative markers with per-entry commit state or a proven contiguous prefix. Keep retained overlap readable, retain from commit time and dedupe against PG by stable observation ID. Test admission during commit, old PEL hole and snapshot-before-commit handoff. | -| F07 | #125 + CoreScope | Provide a durable committed-order/outbox completeness source, or a formally bounded overlap backed by stable identity; ensure late PG commits remain discoverable after Redis retention/offline periods. Test out-of-order commit, offline beyond retention and cursor loss. | -| F08 | CoreScope | Wait for the active job and all submitted writes before either source saves a cursor; propagate write failures and preserve the last safe cursor. Test blocked last job, DB failure and both source orders. | -| F09 | #125 | Atomic one-time per-entry cleanup/accounting, safe reconciliation and separate pending/retained budgets including PEL/quarantine headroom below shared Redis maxmemory. Test interrupted cleanup, replay and concurrent admission/reconciliation. | -| F10 | #125 | Supervise/retry startup; heal missing group, periodically and boundedly reclaim foreign PEL, clear failed singleton state and expose writer progress/readiness. Test NOGROUP, startup failure and young foreign PEL aging into reclaim. | -| F11 | #125 | Use real ioredis `XADD ... MAXLEN ~` for quarantine; validate common decoder on all delivery paths; preserve source until quarantine succeeds; isolate poison rows. Test real client, malformed own/reclaimed PEL and permanent SQL poison. | -| F12 | #125 | Aggregate/dedupe observations by stable ID with PG identity overlap, canonical observer aliases and correct RX/TX counts. Test multiple sightings by the same and new observers during handoff. | -| F13 | #125 | Match SQL network-family scope, aliases, five-minute window, representative fields, full/slim DTOs, and distinct-hash paging. Cover pending detail/channel-history behavior or document explicit pending support. Test side-by-side SQL/RAM fixtures. | -| F14 | #122 + #125 | Register follow-up metrics in the served registry. Separate pending/retained/PEL age/depth, use minimum admitted time, skip zero-capacity new reads, bound reclaim and count actual inserted/deleted work. Test metric scrape and each queue edge case. | -| F15 | #122 + #125 | Start one MQTT shutdown deadline before disconnect. Join RAM writer loop and active write, serialize drain, return explicit complete/incomplete with PEL/outstanding state, close owned clients. Test held persist and failed flush. | -| F16 | #125 | Wire Compose RAM variables, add `docs/ram-first-ingestion` and a verified standalone drain/rollback procedure plus readiness. Test Compose render and backlog before mode rollback. | -| F17 | CoreScope | Preserve upstream observation ID; apply derived state once for a newly accepted observation and define enrichment on duplicate source. Test one Redis+PG observation in both arrival orders and single observer count increment. | - -## Cross-cutting acceptance gates - -- §7 snapshot caveat: concurrent transaction test that schedules lock release after competing statement start; replay and lock refusal must produce no duplicate packets, paths, hourly or region effects. -- §4 eleven-item verification: use the review's SQL/Redis evidence fixtures and extend the named integration harnesses for refused writes, ambiguous admission, marker gap, privacy flips, real quarantine, group loss and shutdown join. -- §5 methodology: correct the below-minimum byte-limit test; compare expected distinct observation identities and all side-effect totals; stop/join each reader; measure complete windows, Redis AOF and PG WAL/physical I/O, and distinguish backend from whole-stack CPU/RAM. Report unresolved limits honestly. -- Each finding needs a regression that fails on reviewed code and passes on repaired code. Every push gets `npx tsc --noEmit`, full `npm test` in `backend/`, focused suites, and successful CI. CoreScope gets `go test ./...` in `cmd/ingestor`. - -## Current state - -### Successor execution — 2026-10-11 - -Remote checkpoints match local heads: perf `8968078`, RAM `e0bddc1`. -Perf CI `38174016916` and `38174021349` concluded success; the two -`e0bddc1` RAM workflows are running. Next implement F07 historical ledger -coverage in a new transactional migration (do not change migration 058's -shipped checksum). Take the same writer lock in a separate command before -backfill, assign missing historical UUIDs, preserve existing identities, -and publish a coverage fence only with the committed backfill. CoreScope -must require that fence for ledger completeness. Regressions cover old and -compressed packets, rollback/retry, contended writer ordering, late commits -and cursor reset. Then continue the outstanding F01/F06/F08/F09/F12–F16 -gates above; no architecture substitution or production changes. - -F06 now captures bounded Redis overlap before the REST/WS SQL query begins, -then applies current authorization after SQL and stable-ID deduplication. -Its unit regression failed at the checkpoint; the real PG/Redis gate holds -a Repeatable Read snapshot, commits A while admitting B, expires retained A, -and proves A remains readable and B appears on the next read. F07 migration -059 and its interrupted/compressed/contended backfill regression pass. -CoreScope's real paired-source crash gate exposed F17 restart corruption: -shared schema setup recreated the broad legacy index after the upstream-ID -migration, potentially deleting distinct observations. Schema reapplication -now preserves upstream IDs and startup repairs actual missing/legacy indexes -even when an earlier migration marker says done. Both source orders pass -active-last-job process crash, SQLite failure, cursor hold and replay/count -checks. Full ingestor/server/shared-schema suites pass. CoreScope is committed -locally at `ff9eb3b` (not pushed); both unrelayed commits are exported in the -owned bench verification evidence directory. RAM `6ed4296` is pushed; its -source-labelled image passed all eight compiled PostgreSQL/Redis regression -tests with zero skips. Both code-head CI workflows `38176440656` and -`38176443793` concluded success in all four jobs. -Fresh full local gates also pass with two CPUs per suite to bound fixture -startup contention: perf 418/418, RAM 469/469, and both typechecks. -Existing bench services and production remain untouched. Next prioritize F01 -copied-data revocation, then the F09/F12–F16 operational/read gates and the ready -WebSocket gap. Preserve the new fixtures and logs in `verification-sol-20261011`; -Docker cannot bind the tool session's private `/tmp` files directly. - -For F01, also test the SQL snapshot's aggregate observation/observer fields -against a newer authorization generation. The current merge reclassifies the -representative packet, so a privacy transition after SQL must not preserve -older aggregate contributions from other observers or paths. Carry and check -the SQL snapshot generation as part of the current-authority contract. - -### Owner handoff at 20:55 UTC - -The owner requested a stop for a GPT-6.1 Sol successor. **Do not treat the repair mission as complete.** The perf branch is pushed at `89680782c3c6a0e517640c9e02fca59f5c8a0bee` (`8968078` archives `TASK-BRIEF.md` and two review patch scripts); its repair code remains at `cbf78a8`. The RAM branch is pushed at `d1aa6ebb8405e9172a74f1f83a42782de7ecde25` (`8cace8e` archives the RAM brief, SQL capture helper and patch scripts; `d1aa6eb` merges the new perf head). Its ingestion code was bench-tested at `022b7c8`, and its later drain-only repair is `f70fe05`. Both worktrees are clean at this handoff. The local CoreScope commit `e9afb3a8a17f8b65d1463985df18984c8f8b86f4` is **not pushed**; relay it with `git format-patch -1 e9afb3a` after review. - -Local pre-push gates pass: perf `npx tsc --noEmit` and backend `npm test` 418/418 (`/tmp/ukmesh-perf-handoff-npm-test.log`); RAM typecheck and backend 468/468 (`/tmp/ukmesh-ram-handoff-npm-test.log`). Earlier full Go suites in CoreScope `cmd/ingestor` and `cmd/server` pass. Code-head CI `38154922672` (#122) and `38160316617` (#125) succeeded. At 20:55 UTC the new handoff-head perf runs `38174016916` and `38174021349`, RAM runs `38174191342` and `38174195183`, and preceding RAM runs `38173522331` and `38173526291` had not all concluded. Check their actual conclusions; do not assume success. Any documentation push after this snapshot triggers a newer CI run. - -Next, verify the latest remote heads and green CI, then update both existing PR descriptions to the final hashes/evidence while leaving #122 open non-draft and #125 draft. `/tmp/ukmesh-pr125-body.md` is a prepared #125 draft with `FINAL_HEAD` and `FINAL_CI` placeholders. The remaining release work is specified per finding in the handoff table in `FIX-REPORT.md`, especially F01 copied-data revocation, F07 pre-ledger backfill, F06 snapshot/retention, F08 two-source crash, F09 retained-budget reconciliation, F13 parity, F14 load semantics, F15 held-write runtime shutdown, F16 raw-stream ACL isolation, and the ready-reader WebSocket gap. No production system was changed. - -| Finding | Status | Successor's next step | -|---|---|---| -| F01 | In progress | Implement immediate revocation for copied CoreScope data; test privacy flips after copy and rerun the post-rematerialization public audit. | -| F02 | Complete | Verify latest CI, then retain its real-PG lock/replay gate. | -| F03 | Complete | Verify latest CI; supported replay identity is covered. | -| F04 | Complete for N=1 | Keep N>1 gated unless separately qualified across broker restart/resize. | -| F05 | Complete | Verify latest CI and retain lost-reply/same-ms regressions. | -| F06 | In progress | Add snapshot-before-commit and retention-expiry real PG/Redis handoff test with a concurrent admission. | -| F07 | In progress | Implement/test pre-migration ledger backfill, then prove offline, late-commit and cursor-loss completeness. | -| F08 | In progress | Run a real paired-source SQLite crash/failure test proving cursor barriers for the active last job. | -| F09 | In progress | Test startup/retained byte reconciliation and Redis memory headroom through replay and expiry. | -| F10 | Complete | Verify latest CI and retain real NOGROUP recovery gate. | -| F11 | Complete | Verify latest CI and retain real-ioredis poison/quarantine gate. | -| F12 | In progress | Compare merged same/multi-observer counts and representative selection to real SQL across handoff. | -| F13 | In progress | Extend SQL/RAM parity to paging, sampling, representative fields and all supported scope/alias variants. | -| F14 | In progress | Assert every served metric's semantics under known loaded, zero-capacity and retained states. | -| F15 | In progress | Inject held persist during runtime SIGTERM and repair the bench post-restart drain oracle. | -| F16 | In progress | Prove CoreScope raw-stream Redis ACL isolation and rerun final drain/rollback on that configuration. | -| F17 | Complete at code/regression scope | Validate paired-source crash operation under F08. | - -No top-level finding is wholly not started; the named integration, backfill and ACL gates above are not yet implemented or proven. The detailed evidence and limitations for each finding are in `FIX-REPORT.md`. +Production remains untouched. No deployments, existing service restarts, +PR merges/approvals or CoreScope pushes. No subagents. The original bench +`FINAL-CHAIN-DONE` marker was observed at 17:39:52 UTC. Owned test processes +and containers began afterward. Durable is the default; RAM is opt-in; +N>1 remains gated. Current-authority failure is deliberate public +unavailability, and already delivered data cannot be recalled. Preserve +old untracked CoreScope DBs; initialize a new dedicated public projection. diff --git a/FIX-REPORT.md b/FIX-REPORT.md index 8d78bbc..28e874b 100644 --- a/FIX-REPORT.md +++ b/FIX-REPORT.md @@ -1,293 +1,129 @@ -# UKMesh ingestion review repairs — working report (2026-10-11) +# UKMesh ingestion review repair report — 2026-10-11 -Specification: `/home/ben/ukmesh-fix-context-20261011/SOL-REVIEW.md`. This is a working record; **the full repair objective is not yet achieved**. Status below reflects inspected code and completed verification only. +The F01–F17 repair mission remains active. This report replaces the earlier +handoff status tables. Code fixes and completed evidence are recorded below; +the final whole-stack campaign and final-head CI are still required. The +specification is `/home/ben/ukmesh-fix-context-20261011/SOL-REVIEW.md`. -## Current F01/F05/F12/F13/F16 working-tree wave +## Branch and validation checkpoint -The RAM documentation checkpoint `c98c2f53e2921bed017fa1d0ecc2016f20a5c374` -is pushed; its workflows -[38177795874](https://github.com/gadgethd/ukmesh/actions/runs/38177795874) and -[38177798421](https://github.com/gadgethd/ukmesh/actions/runs/38177798421) -both concluded success. The runtime changes described next are newer and -await their own commit/CI; earlier historical evidence below is unchanged. +- Perf/#122: `89680782c3c6a0e517640c9e02fca59f5c8a0bee`, pushed and verified + against the remote. Typecheck and full backend suite pass **418/418** + (138.715s). Both workflows [38174016916](https://github.com/gadgethd/ukmesh/actions/runs/38174016916) + and [38174021349](https://github.com/gadgethd/ukmesh/actions/runs/38174021349) + concluded success in every job. N=1 is supported; N>1 is explicitly gated. +- RAM/#125: public-read code checkpoint + `c683acdec8572e6b6dbc5ba07ccdfcbb062f10ce`, pushed and remote-verified. + Both workflows [38180468916](https://github.com/gadgethd/ukmesh/actions/runs/38180468916) + and [38180473812](https://github.com/gadgethd/ukmesh/actions/runs/38180473812) + concluded success in every job. Its typecheck, **476/476** full backend + suite, **7 PG + 2 Redis + 1 WS** integration tests, and exact compiled + image gate **10/10**, zero skips, passed. +- New RAM operational/read wave: all local gates pass, awaiting commit/push + and its own CI. Final typecheck/build and **479/479** backend tests pass + (193.952s), including the alert receiver. Full ingest integration passes + **7 PostgreSQL + 3 Redis + 1 WS**, zero skips. The strict owned-ACL gate + passes exactly **4 operational + 1 compiled-process test**, zero skips; + runtime suite 152.153s. Loaded resident-memory refusal, real HTTP gauges, + held SQL/SIGTERM recovery, quarantine refusal, actual retention expiry, + standalone drain and durable rollback are all verified. The final Compose + render passes all ten RAM/public-source settings, without outputting secrets. +- CoreScope local head: `07f05cd5562ac27478f4a46b5dd13d157e661162`, clean, + committed and **not pushed**. Full `cmd/ingestor` passes (234.242s), + including real PG/Redis integration; full `cmd/server` passes (103.965s). + Shared-schema tests passed in the preceding unchanged schema wave. -F01 SQL aggregates now carry the snapshot generation and fail closed when -current authority differs. WS initial caches recheck generation before delivery. -MQTT and WS live events use bounded batches with one current authority query, -including active-payload byte reservations and unavailable/overflow metrics. -The actual ready WebSocket test reproduced the old privacy bypass, then passed -observer/path revocation, authority failure, cached reconnect refusal and -**800/800** allowed deliveries. This proves that isolated ready-reader gate, -not whole-stack MQTT performance. +Evidence below is in the newly owned directory +`/home/ben/ukmesh-bench/mqtt-ingest-perf-20261011/verification-sol-20261011/evidence/`. +The image `mqttbench-backend:sol-public-c683acd-20261011` carries source label +`c683acd` and configuration digest +`sha256:d96f200660518a8453fd494513193e0814cfb4bee9a4bfee10aa2e1a44094adc`. +`ukmesh-public-c683acd-image-integrity-pinned.log` proves exactly ten passing, +zero-skipped tests against shipped modules (88.480s). The test source is pinned +with `git show` to the image label. Missing mounts or zero-test exits cannot +pass. A newer operational image is still required. -F12/F13 real SQL/RAM comparisons cover family and unrelated network scopes, -alias/canonical/other observers, full/slim fields, RX/TX and repeated sightings, -representative replacement, five-minute filtering, message rows and page limits. -SQL includes aggregates for captured overlap hashes so a pending representative -moving onto page one retains older committed observations. Dates retain -milliseconds; ties use observation IDs and packet hashes. The fixture exposed -and repaired F05's stable-ID/natural-dedupe interaction: two distinct IDs with -identical fields remain separate observations. Legacy entries without IDs -retain the natural-identity fallback. A genuine older Repeatable Read snapshot -is rejected after the authority generation changes. +## Findings and regression evidence -CoreScope's uncommitted F01/F16 implementation requires a dedicated SQLite -public projection and private ingestor provenance journal. On a policy change -it fences readiness, removes revoked journal records and reconstructs copied -and derived measurements. The server stays read-only and checks authority -before and after buffered API reads and before bounded WS fanout; cache/index -generations are replaced. Old untracked data is preserved and fenced instead -of silently purged. Duplicate-source enrichment preserves participant evidence. -Its actual PostgreSQL-authority/SQLite regression covers observer, source and -all one/two/three-byte prefix revocation across the UK family while preserving -an unrelated network; the final full-suite result is pending. +“Implemented” describes code/regression scope; final release gates are listed +separately. Every finding has a regression that reproduced the reviewed defect. +Later boundary defects also received new failing regressions. -F16 adds `/internal/ingest/public/{policy,entries}`, disabled unless a bearer -secret of at least 32 characters is configured. Every captured read filters -against current authority; private entries advance only the opaque cursor. -CoreScope public mode accepts this HTTP source and rejects raw Redis URL/DSN -configuration, requiring the paired PG ledger. Tests cover credentials, limits, -cursor ordering, unavailable authority and projection crash/replay barriers. -Operational principal isolation, loaded headroom, final drain/rollback and -whole-stack performance remain required. - -Owned evidence is under -`verification-sol-20261011/evidence/`: `ukmesh-ws-authority-baseline-ready.log` -and `ukmesh-ws-authority-after.log`, `ukmesh-read-paging-before.log` / -`ukmesh-read-paging-stable-id.log`, `ukmesh-f13-ties-before.log`, -`core-f01-copied-before.log`, `core-f01-untracked-node-before-valid.log`, -`core-f01-enrichment-before.log`, and `core-public-paired-final2.log`. -The final full backend run passes **476/476** in **202.440s**, including the -alert receiver; typecheck passes. The complete integration command passes -**7 PostgreSQL + 2 Redis + 1 WS tests**, zero skips, after giving the legacy -empty-schema fixture its own database. Evidence is -`ukmesh-public-final-full-tests.log` / `ukmesh-public-final-full-integration.log`. -CoreScope's server suite passes in **103.965s**; its final ingestor gate is -still running. The expanded image harness requires exactly ten passing tests. -Production and the original bench services remain unchanged. - -## Branches and gates - -**Successor progress, 2026-10-11.** The authoritative remote checkpoints were -perf `8968078` and RAM `e0bddc1`, both matching their local worktrees. Perf -workflows `38174016916` / `38174021349` and RAM checkpoint workflows -`38174696402` / `38174699642` have now concluded **success**. - -The next repair wave implements F06 snapshot ordering and F07 historical -coverage and adds a real F08/F17 crash/restart gate. Its UKMesh typecheck and -full backend suite pass **469/469** (`/tmp/ukmesh-sol-ram-final-tests.log`). -The complete ingest integration command passes **6 PostgreSQL + 2 Redis tests** -using a newly created disposable PostgreSQL database and an independently -created Redis fixture (`/tmp/ukmesh-sol-full-ingest-integration.log`). -Focused CoreScope F07/F08/F17 tests pass with real PostgreSQL/Redis/SQLite -(`/tmp/core-sol-f08-integration-after.log`). Full successor Go suites pass: -ingestor **222.972s**, server **116.673s**, shared schema **0.530s**. Logs are -`/tmp/core-sol-full-ingestor.log`, `/tmp/core-sol-full-server.log` and -`/tmp/core-sol-full-dbschema.log`. - -RAM `6ed4296e988f1ec69640d02443f551d37c619400` is pushed and verified by -`git ls-remote`. Its workflows -[38176440656](https://github.com/gadgethd/ukmesh/actions/runs/38176440656) and -[38176443793](https://github.com/gadgethd/ukmesh/actions/runs/38176443793) -both concluded **success in all four jobs**, including full Backend and -Workers/Compose/E2E/empty-volume gates. Both PR descriptions are updated; #122 remains -open/non-draft and #125 remains open/draft. #122's title now describes the -supported N=1 repair instead of obsolete multi-connection performance. - -CoreScope is locally committed at `ff9eb3b93c9addbb29feb9debcc12e6cb99d5c8d`. -Both `e9afb3a` and `ff9eb3b` await relay; CoreScope was not pushed. Export with -`git -C /home/ben/corescope-ukmesh/src format-patch --stdout 90c104eab51618894a3b69ab521829f39313e261..ff9eb3b93c9addbb29feb9debcc12e6cb99d5c8d`. -The ready patch is `/tmp/corescope-sol-relay-20261011.patch` (2383 lines). -Evidence logs and the patch are also preserved in the new owned -`/home/ben/ukmesh-bench/mqtt-ingest-perf-20261011/verification-sol-20261011/evidence/` -directory so tool-session temporary files are not the sole handoff artifacts. - -Both uniquely tagged verification images built successfully: RAM -`mqttbench-backend:sol-handoff-6ed4296-20261011` has image ID -`sha256:563424ed79ac51abc2eb70cf8f2080bec9961e6c0622f1948233bfeaf92b0859` -and revision label `6ed4296`; durable -`mqttbench-backend:sol-durable-8968078-20261011` has image ID -`sha256:f501da065bf98e219c54d335dfad6703d430b51e6437d49a63140341dc72a637` -and label `8968078`. Existing bench services were unchanged. The completed -compiled-regression gate against the RAM image was launched by the new owned -`verification-sol-20261011/run-image-integrity.sh` under the bench directory. -It mounts only tests, executes shipped modules and requires eight passing, -zero-skipped tests; exit-zero/zero-tests cannot pass. -The `6ed4296` image passed **8/8**, zero skips, in **72.190s**, including real -snapshot/expiry, historical/compressed/contended backfill, refused-write, -replay and quarantine cases. Its evidence is -`logs/sol-image-integrity-shared-path-6ed4296-20261011.log` under the bench -directory (`IMAGE-INTEGRITY-END code=0 complete=true`). - -F06's new unit failed on the checkpoint (`/tmp/ukmesh-f06-before.log`), then -passed with `readRecentWithPending`. The real fixture captures Redis before -starting SQL, holds a pre-commit Repeatable Read snapshot, commits A while -admitting B, expires retained A from Redis, and proves A still appears exactly -once. A later request sees A and B. REST recent packets and both WS initial -packet/message paths use this ordering; authorization runs after SQL. - -F07's regression failed at the checkpoint (`/tmp/ukmesh-f07-before.log`). New -migration 059 backfills pre-ledger history under the writer transaction lock, -assigns missing UUIDs in compressed history once, preserves existing IDs and -publishes a coverage fence atomically. An injected migration-ledger failure -rolls back UUIDs and coverage; retry after a contended writer commit succeeds. -CoreScope requires the fence. Its real PG-to-SQLite test rejects incomplete -coverage, ingests historical and older late commits, and safely replays after -cursor-file loss. History already deleted by PG retention remains unavailable. - -The F08 paired-source gate reads actual PostgreSQL ledger and Redis entries, -kills a disposable ingestor test process while its last submitted write is -active with an empty queue, reopens the same SQLite/WAL, injects a SQLite -observation failure and proves both cursor files retain the last safe state. -Successful retry completes both cursors with two observations and exactly two -observer count increments, in both source orders. It exposed a further F17 -restart defect: shared schema reapplication rebuilt the broad legacy identity -index and could collapse distinct upstream IDs. The new shared-schema -regression reproduced that deletion (`/tmp/core-sol-f17-restart-before.log`); -schema setup now defers upstream identity indexes to the ingestor migration, -and startup checks/repairs the actual index state despite a done marker. - -The first local backend suite failed a real-time 50ms slow-mode assertion -under CPU load; its clock is now deterministic and the full suite passes. -Concurrent fresh suite reruns then hit the existing two-second worker-fixture -startup timeout on both branches. Complete suites are being rerun with two -CPUs per suite (same test files/assertions) to bound test-process contention. -Both complete reruns passed: perf **418/418** in **138.715s**, RAM **469/469** -in **210.111s**, with typechecks and the alert receiver gate passing. Logs are -`evidence/perf-8968078-two-cpu-tests.log` and -`evidence/ram-6ed4296-two-cpu-tests.log` in the owned verification directory. -The first full PG invocation used an already initialized bench database and -failed its final empty-schema fixture; the successful rerun used a new owned -database. One intentionally stopped diagnostic CoreScope test left an owned -fixture schema; that exact schema was removed. No production data or services -were involved. Preliminary image-fixture packaging had missing assets or zero -tests; neither is accepted as a pass. Docker's daemon does not share the tool -session's `/tmp`, so fixtures now use the shared, newly owned bench verification -directory. The corrected image run's exact eight-test result is verified above. - -Remaining mission work: F01 copied-data revocation and generation transitions; -F09/F14 memory/accounting and metric semantics; F12/F13 SQL read parity; -F15 held-write runtime shutdown; F16 ACL/public-source isolation; the ready -WebSocket reader gap; final source-labelled builds, integrity/performance -gates, final CI, PR descriptions, relay commit evidence and report audit. -The predecessor handoff below is historical and is superseded for F06–F08/F17 -by this successor evidence. The full objective is still active. - -| Work | Current evidence | -|---|---| -| UKMesh #122 | Repair code `cbf78a8a6c8d0f11da200eed0bcf9f4161c5dd2f` and handoff commit `89680782c3c6a0e517640c9e02fca59f5c8a0bee` are pushed to `perf/mqtt-ingest-20261011`. The handoff commit archives `TASK-BRIEF.md` and two patch scripts; runtime code is unchanged. Local `npx tsc --noEmit` and backend `npm test` 418/418 pass at this head (`/tmp/ukmesh-perf-handoff-npm-test.log`); earlier 9/9 focused F03/F04/F14/F15 tests passed. [Code-head CI 38154922672](https://github.com/gadgethd/ukmesh/actions/runs/38154922672) succeeded in all four jobs. New handoff-head CI status is below. | -| UKMesh #125 | Ingestion repairs through `022b7c8a238f2265f219cbe59cea606525c8806f` and drain fix `f70fe058bff5ef28d89ddea3f271cfcf66a311b2` are pushed. Handoff commit `8cace8e46f8e157729dfe00e9e29a1eb1dc35bbd` archives the brief, SQL capture helper and patch scripts; merge `d1aa6ebb8405e9172a74f1f83a42782de7ecde25` carries the new perf head, preserving the stack. Local TypeScript and backend `npm test` 468/468 pass at this head (`/tmp/ukmesh-ram-handoff-npm-test.log`); contract, build and focused checks passed on the same runtime code. [Earlier code-head CI 38160316617](https://github.com/gadgethd/ukmesh/actions/runs/38160316617) succeeded in all four jobs, including real-PG, real-Redis, E2E and empty-volume Compose. New handoff-head CI status is below. | -| CoreScope #1 | Local `feat/redis-live-source` commit `e9afb3a` contains F01/F07/F08/F17 repairs. Focused regressions, full `cmd/ingestor` Go suite (150.648s) and full `cmd/server` Go suite (104.999s) pass. CoreScope has not been pushed; credentials are unavailable on this host. | -| Bench | `FINAL-CHAIN-DONE` observed at 17:39:52 UTC. The ingestion matrix used exact code heads `022b7c8` and `cbf78a8`; the standalone drain entrypoint was rebuilt at `f70fe05`. Crash replay (120/120), ambiguous commit replay (120/120), concurrent count and byte admission, public-to-private pending/commit boundary, and all six matched-window scenarios pass their database identity gates. The first default-broker saturation run lost messages; a sized-broker retest passed. Standalone drain, quarantine refusal and durable rollback passed. | - -## Finding map - -| Finding | Status | Code and regression evidence | +| Finding | Status | Repair and evidence | |---|---|---| -| F01 | Partial | #125 fresh bulk policy/generation/rematerialization check in `packetBatch.ts`, `liveRead.ts`, `client.ts`, `ws/privacy.ts`; stale-cache regressions pass. CoreScope PG/Redis sources now recheck current authority per batch and fail closed, but already copied derived data has no immediate revocation barrier. | -| F02 | Implemented; real PG lock/replay regression passes | Separate same-client transaction lock and ledger-derived mutations in `packetBatch.ts`; CI `38159431225` passed the contended-lock/replay test. | -| F03 | Fixed on #122 and merged into #125 | `backend/src/mqtt/observationIdentity.ts`, `client.ts`, `db/packetBatch.ts`; unit and real-PG regression passed on #122 CI. | -| F04 | Fixed on #122 and merged into #125 | `backend/src/mqtt/ingestConfig.ts` rejects N>1 before connect; regression passes. | -| F05 | Implemented for UKMesh; final-head gates pass | Delivery-key replay ledger in `ram/liveStore.ts`, stable observation ID and PG ledger; lost-reply and same-ms unit/PG integration regressions added. Real-PG and real-Redis CI tests pass. Final-head crash and ambiguous SQL-commit replay pass; a deterministic real-Redis XADD followed by a lost reply recovers the original observation ID with stream length one. | -| F06 | Implemented; real snapshot/expiry regression passes | Per-entry commit state, retained overlap and stable-ID dedupe plus `readRecentWithPending` before-SQL capture on REST/WS. `F06 real PG snapshot and Redis expiry preserve the captured handoff` covers concurrent admission and retention expiry. Final image gate remains. | -| F07 | Implemented; historical/late-commit/cursor-loss regressions pass | Migration 059 adds atomic compressed-history backfill and coverage fence without changing 058. CoreScope requires coverage and follows `seq > cursor`. `F07 historical and compressed observations enter the complete ledger atomically` and `TestF07RealLedgerHistoricalCoverageLateCommitAndCursorLoss` pass. Final image/paired offline gate remains. | -| F08 | Implemented; real paired crash/failure and full Go suites pass | Both CoreScope sources use `finishSourceBatch`; `TestF08RealPairedSourcesCrashFailureAndRestartBothOrders` proves active-last-job process crash, same-SQLite/WAL restart, actual observation insert failure, held cursors and successful replay. Full successor ingestor/server/schema suites pass at `ff9eb3b`. | -| F09 | Partial | One-time atomic cleanup, bounded reconciliation and retained accounting in `ram/liveStore.ts`, `writer.ts`; replay, cleanup and real-Redis lost-counter reconciliation tests pass. Final-head concurrent byte admission passes at the enforced 1 MiB minimum. Startup/retention accounting in the final-head bench remains. | -| F10 | Implemented; group-loss integration passes | Supervised startup, NOGROUP repair, periodic bounded foreign PEL reclaim, and stale-PEL/progress readiness in `index.ts`, `ram/writer.ts`, `writerRuntime.ts`; unit regressions and a final-head real-Redis group-destruction/recovery gate pass. | -| F11 | Implemented; final-head fault gate passes | Real ioredis XADD quarantine, common validated decode, and rowwise permanent SQL poison isolation in `ram/writer.ts`, `protocol.ts`; malformed PEL, poison-neighbor and real-Redis quarantine tests pass. A final-head simulated permanent SQLSTATE 23514 in a two-row batch quarantined only the poison row through real ioredis, committed its valid neighbor in PG, and left PEL zero. | -| F12 | Implemented, awaiting integration | Stable-ID aggregate and PG overlap in `ram/liveRead.ts`; same-observer sighting regression passes. | -| F13 | Partial | `ram/liveRead.ts` now matches network-family scope, canonical aliases, recent windows, representative preference and slim DTO; `TestF13 RAM recent rows honor family scope, aliases, window, and slim DTO` passes. A corrected final-head fixture persisted the exact admitted stream entry and matched full and slim pending DTOs to real PostgreSQL recent-query DTOs. Detail/long history are explicitly PG-only; wider paging and field combinations remain unproven. | -| F14 | Partial | #122 metric registration fixed. #125 separates pending/retained/PEL gauges, byte budget and minimum admitted-age ordering; commit batch histogram now counts actual inserted PG rows. A final-head live metric scrape confirms all those series are served under load. Final load validation remains. | -| F15 | Implemented, boundary still partly unit-tested | #122 deadline fixed. #125 shares one shutdown deadline, joins writer loop, detects PEL on drain and closes clients; `drainLifecycle.test.ts` holds an active persist past shutdown. Isolated runtime SIGTERM logged graceful completion in about two seconds and its 300/300 exact identities, side effects, queue, PEL and bytes passed a manual post-restart gate. The bench's stock wait script missed this because `processed` has no zero-valued series after restart; no held active persist was injected into that runtime run. | -| F16 | Partial | Compose variables and CI render assertion, `docs/ram-first-ingestion.md`, and `node dist/ram/drain.js` added. The drain waits for Redis readiness, reuses the ready client for final byte reconciliation, closes path-resolution workers, and refuses a nonempty quarantine as well as nonempty stream/PEL/byte reservation. Isolated standalone drain, quarantine refusal and durable rollback passed on `f70fe05`. Raw-stream Redis ACL isolation remains. | -| F17 | Implemented including restart repair; full Go suites pass | CoreScope preserves upstream UUID and increments observer count once. `ff9eb3b` prevents schema reapply from rebuilding the broad legacy index and deleting distinct IDs, and repairs actual missing/legacy index state despite done migration markers. `TestF17UpstreamIdentitySurvivesSchemaReapply`, both-order identity/count and real crash/restart tests pass. | +| F01 | Implemented; final paired public audit remains | Fresh completed policy/generation fences govern SQL aggregates, captured RAM, WS initial caches and bounded MQTT/WS live batches. SQL snapshot generation is never permission after a change. The final fence follows alias/summary waits. Tests: `F01 old SQL aggregate contributions fail closed across a policy generation change`, `F01 alias lookup cannot deliver a captured generation after authority changes`, actual ready WS observer/path flip, unavailable authority, cached reconnect and **800/800** allowed messages. CoreScope uses a dedicated revocable SQLite projection and private provenance journal; the read-only server fences REST/WS and replaces generation caches. Actual PG/SQLite tests revoke copied observer/source/all 1/2/3-byte prefixes across UK families while preserving unrelated networks; held REST and actual WS tests pass. | +| F02 | Implemented and real-PG verified | Same-client explicit transaction acquires the writer lock in a separate statement, then stable-ID ledger arbitrates every mutation. `F02 contended lock and replay never change historical tables` proves refusal changes no packet/path/hourly/daily/node/sighting/region table and replay changes none. This also addresses §7's lock/snapshot caveat. | +| F03 | Fixed on #122, merged into #125 | Canonical observation identity distinguishes observer, path, hop, scope, direction and raw frame; restart probes use the same identity. Named F03 identity/probe and real-PG regressions pass. | +| F04 | Fixed for supported N=1 | Unsupported shared-subscription N>1 configuration is rejected before connecting. `F04 shared-subscription N>1 is gated until restart and resize recovery is qualified` passes. No multicore recovery claim is made. | +| F05 | Implemented and real PG/Redis verified | Admission creates a stable UUID before Redis's ambiguous boundary; the bounded delivery ledger returns the stored entry on retry. PG's identity ledger preserves distinct stable IDs even with identical natural fields; natural fallback is restricted to legacy entries without IDs. Lost Redis reply, distinct same-ms identities and the wider SQL fixture pass. | +| F06 | Implemented and real PG/Redis verified | Per-entry commit timestamps retain payload/reservation for 120 seconds after commit. REST/WS capture bounded Redis overlap before SQL and deduplicate by stable UUID. `F06 real PG snapshot and Redis expiry preserve the captured handoff` holds an older Repeatable Read snapshot, commits A while admitting B, deletes expired A, and proves captured A plus later B remain discoverable exactly once. | +| F07 | Implemented; final offline paired audit remains | Migration 059 backfills historical/compressed observations under the same writer lock, preserves/assigns UUIDs and commits a coverage fence atomically. CoreScope requires coverage and uses committed-order `seq`, replaying retained history after cursor loss and generation change. Historical, interrupted migration, contended writer, older late commit and lost-cursor regressions pass. Deleted upstream retention is outside recoverable history. | +| F08 | Implemented and actual crash/SQLite failure verified | Both sources wait for tracked batch completion including the active last job; errors leave cursors behind. `TestF08RealPairedSourcesCrashFailureAndRestartBothOrders` reads real PG/Redis, kills an owned child at that boundary, reopens the same SQLite/WAL, injects observation failure and proves safe cursor hold/replay in both orders through the public projection. | +| F09 | Implemented and actual resident-memory verified | Atomic final deletion releases reservation once; bounded Redis-side reconciliation serializes with admission/cleanup. Retained records and ambiguous delivery copies remain reserved. Startup/admission require bounded noeviction Redis and 32 MiB recovery headroom. Real concurrent admission/reconciliation, lost counter, retained replay and lost cleanup reply pass. Encoded quarantine is capped at 16 KiB and exactly 500 records; actual physical allocation fits the separate headroom. Unsafe configuration regression fails on `c683acd`; valid-budget admission also refuses when other resident keys consume the recovery space. | +| F10 | Implemented and real group-loss verified | Supervised startup retries before MQTT intake; failed state is cleared. NOGROUP repairs the group, young foreign PEL is periodically/boundedly reclaimed, and commit/stall failures affect MQTT readiness. F10 group-loss and aging foreign-PEL regressions plus the historical compiled fault gate pass. | +| F11 | Implemented and real-ioredis verified | Real XADD/trim quarantine precedes source cleanup; all read paths share validated decode. Permanent SQL poison is isolated so its valid neighbor commits. Quarantine failure retains source. Real malformed/quarantine, own/reclaimed PEL and SQLSTATE 23514 poison-neighbor tests pass; escaping no longer inflates encoded evidence above its cap. | +| F12 | Implemented and real SQL/RAM verified | UUIDs distinguish overlap from new observations, including repeated sightings by one observer. Canonical observer arrays and RX/TX counts match real SQL across handoff, alias/family scopes and paging. | +| F13 | Implemented and wide real SQL/RAM verified | Bounded whole-stream capture, SQL aggregates for overlap hashes, representative rank/UUID ties, millisecond order, final distinct-hash paging, windows, full/slim/message fields, alias/family scopes and decrypted/JSON summary extraction match SQL. New fallback regression failed before repair. Packet detail and long channel history are explicitly PG views until commit; documented pending lists do not promise those views. | +| F14 | Implemented and loaded HTTP-scrape verified | Follow-up metrics use the served registry. Pending/retained/PEL/physical/reserved gauges are separate; minimum admitted age, zero-capacity read skip and bounded reclaim are tested. Loaded-state and missing-stream gauges are asserted through the actual HTTP metrics server. Commit histogram counts actual inserted observations. Missing-group/null-memory regression reproduced stale gauges before repair. | +| F15 | Implemented; corrected campaign oracle remains | MQTT uses one deadline beginning before disconnect. RAM joins the active loop before draining and returns incomplete on deadline/PEL/failure; owned clients close. Unit held-write regressions and actual compiled SIGTERM gate pass: a held INSERT causes exit 1 at two seconds, then recovery waits for release and commits the original UUID with every DB side effect exactly once. The stock old bench oracle is unsuitable; the new campaign must use independent identity/state checks. | +| F16 | Implemented; final whole-stack configuration audit remains | Compose passes RAM settings and the filtered bearer HTTP token; durable stays default. Public CoreScope receives no raw Redis URL/DSN and requires the paired PG ledger/projection. Actual Redis principals deny public/operator raw stream access, separate quarantine review, and HTTP revocation filtering passes. The compiled runtime uses the restricted writer principal and proves quarantine refusal, real retention sweep, zero stream/PEL/reservation and standalone drain before durable rollback. Final Compose render passes; whole-stack rerun remains. | +| F17 | Implemented and actual restart/both-order verified | CoreScope preserves upstream UUID, enriches duplicates without incrementing derived counts and retains original privacy participants. Schema reapplication preserves distinct upstream identities and repairs actual legacy/missing indexes despite migration markers. F17 schema, both-order identity/count and paired crash/replay tests pass. | -## Successor handoff — 2026-10-11 20:55 UTC +## Recent evidence and relay -The owner requested that a GPT-6.1 Sol successor continue the **full** F01–F17 mission. The statuses below are review requirements, not a claim that #125 is merge-ready. “Complete” means the finding's code and named regression are in place; cross-cutting release gates remain listed separately. No top-level finding is wholly not started, but several required integration and operational gates are. +- Read/security before/after logs: `ukmesh-ws-authority-baseline-ready.log`, + `ukmesh-ws-authority-after.log`, `ukmesh-f01-alias-before-valid.log`, + `ukmesh-summary-fence-focused2.log`, `ukmesh-f13-summary-before.log`, + `ukmesh-f13-summary-final.log`, and the earlier paging/tie failure logs. +- Resource/operational logs: `ukmesh-f09-headroom-before.log`, + `ukmesh-f09-quarantine-before.log`, `ukmesh-f14-loaded-before-valid.log`, + `ukmesh-f14-loaded-after.log`, `ukmesh-operational-final-runtime.log`, + `ukmesh-operational-final-full-tests2.log`, + `ukmesh-operational-final-runtime2.log` and + `ukmesh-operational-final-ingest-integration2.log`. +- CoreScope logs: `core-f01-copied-before.log`, + `core-f01-untracked-node-before-valid.log`, `core-f01-enrichment-before.log`, + `core-public-paired-final2.log`, `core-public-final-ingestor2.log`, + `core-public-final-server2.log`. -| Finding | Handoff status | Exact next step if in progress | -|---|---|---| -| F01 | **In progress** | Add an immediate public revocation barrier for **already copied** CoreScope SQLite-derived rows, then test observer/source/prefix privacy flips after copy and during pending/commit with available and unavailable policy authority. Rerun the synthetic post-flip audit after rematerialization completes; its current zero-result audit is inconclusive. | -| F02 | **Complete** | — | -| F03 | **Complete** | — | -| F04 | **Complete** for supported N=1; N>1 is gated | — | -| F05 | **Complete** | — | -| F06 | **In progress** | Add a deterministic PG-snapshot-before-commit and Redis-retention-expiry integration test with a new admission during commit; compare exact observation IDs across the PG/RAM handoff. | -| F07 | **In progress** | Implement and test a safe historical backfill for packets written before migration 058. Prove the paired CoreScope follower reads those rows and late commits across Redis retention/offline periods, including cursor loss. Do not enable Redis-only completeness. | -| F08 | **In progress** | Run a real paired PostgreSQL/Redis-to-SQLite crash/restart test that blocks the last job, injects a write failure, and proves both saved cursors remain behind every incomplete write. | -| F09 | **In progress** | Reconcile byte reservations through startup, retained expiry, replay and interruption against real Redis; prove pending, retained, PEL and quarantine headroom under the configured Redis memory limit. | -| F10 | **Complete** | — | -| F11 | **Complete** | — | -| F12 | **In progress** | Compare RAM/PG merged counts and representative selection to SQL for repeated sightings from the same and multiple observers across a real handoff, including distinct stable IDs. | -| F13 | **In progress** | Extend the real-SQL parity fixture beyond the passed full/slim recent DTO case to paging, sampling, all representative fields, aliases, network-family combinations and documented PG-only detail/history. | -| F14 | **In progress** | Run a controlled loaded metric scrape with known pending/retained/PEL/byte and zero-capacity states, and assert registered series and actual batch/drop semantics at each boundary. | -| F15 | **In progress** | Inject a held active persist into the isolated runtime SIGTERM gate and prove deadline, incomplete result and no lost identities. Fix the bench `wait-drain.sh` post-restart `processed`-series oracle before relying on its `DRAINED` marker. | -| F16 | **In progress** | Prove raw-stream Redis ACL isolation for the CoreScope reader, then repeat Compose render and the already-passed standalone drain/quarantine refusal/durable rollback gate on the resulting final configuration. | -| F17 | **Complete** at code/regression scope | —; real paired-source crash operation is tracked under F08. | +Three clean local CoreScope commits await coordinator relay, in this order: -**Handoff commits and push state.** Perf `89680782c3c6a0e517640c9e02fca59f5c8a0bee` adds the task brief and two review patch scripts. RAM `8cace8e46f8e157729dfe00e9e29a1eb1dc35bbd` adds the Sol brief, RAM patch scripts and SQL capture helper; `d1aa6ebb8405e9172a74f1f83a42782de7ecde25` merges the new perf head. Both UKMesh branches were pushed and locally clean before this report edit. The CoreScope remote `feat/redis-live-source` is at `90c104eab51618894a3b69ab521829f39313e261`; **its only unpushed local commit is `e9afb3a8a17f8b65d1463985df18984c8f8b86f4`**: paired PG/Redis completion barriers, current privacy checks for new batches, committed-order cursor, stable upstream identities and deduped observer counts. Export with `git -C /home/ben/corescope-ukmesh/src format-patch -1 e9afb3a8a17f8b65d1463985df18984c8f8b86f4`; the coordinator must relay-push it. No CoreScope worktree changes are uncommitted. +1. `e9afb3a8a17f8b65d1463985df18984c8f8b86f4` — paired source completion, + current policy checks, committed-order cursor and upstream identity. +2. `ff9eb3b93c9addbb29feb9debcc12e6cb99d5c8d` — historical coverage, + real paired crash/replay and shared-schema identity repair. +3. `07f05cd5562ac27478f4a46b5dd13d157e661162` — copied-data revocation, + provenance reconstruction, filtered live source and fenced public server. -**CI snapshot, 20:55 UTC.** Earlier repair-code runs [#122 38154922672](https://github.com/gadgethd/ukmesh/actions/runs/38154922672) and [#125 38160316617](https://github.com/gadgethd/ukmesh/actions/runs/38160316617) succeeded in all four jobs. New perf handoff-head runs [38174016916](https://github.com/gadgethd/ukmesh/actions/runs/38174016916) and [38174021349](https://github.com/gadgethd/ukmesh/actions/runs/38174021349) were in progress. RAM runs [38173522331](https://github.com/gadgethd/ukmesh/actions/runs/38173522331) and [38173526291](https://github.com/gadgethd/ukmesh/actions/runs/38173526291) were still in progress; the newer RAM handoff-head runs [38174191342](https://github.com/gadgethd/ukmesh/actions/runs/38174191342) and [38174195183](https://github.com/gadgethd/ukmesh/actions/runs/38174195183) were pending. These are **not recorded as green**. Re-query the latest branch heads and runs after this documentation push, then watch the newest full CI to conclusion. Local pre-push gates passed: perf typecheck and 418/418 backend tests (`/tmp/ukmesh-perf-handoff-npm-test.log`); RAM typecheck and 468/468 (`/tmp/ukmesh-ram-handoff-npm-test.log`). Earlier CoreScope full `cmd/ingestor` and `cmd/server` Go suites passed. +The ready three-commit patch is +`evidence/corescope-sol-relay-20261011.patch` (4681 lines). Recreate it with: -**Open handoff threads.** The PR #122 body still cites the repair code head `cbf78a8`; PR #125 still cites `022b7c8` and says the benchmark is running. After final CI, update both existing PR descriptions to the pushed heads and evidence, preserving #122 open/non-draft and #125 draft. `/tmp/ukmesh-pr125-body.md` has a prepared #125 description with `FINAL_HEAD` and `FINAL_CI` placeholders. No PR was merged, approved or deployed. The isolated bench evidence and limitations follow below: ready RAM burst WebSocket reader received 2/800, the low-rate RAM PG follower received 300/360 in its bounded window, the default-broker saturation run lost 315/3,100 before a sized-broker retest passed, and privacy selective filtering after flip remains inconclusive while rematerialization jobs are pending. +```sh +git -C /home/ben/corescope-ukmesh/src format-patch --stdout 90c104eab51618894a3b69ab521829f39313e261..07f05cd5562ac27478f4a46b5dd13d157e661162 +``` -## Regression index +## Remaining acceptance and limitations -| Finding | Named regression or gate | -|---|---| -| F01 | `F01 a privacy flip overrides the loaded public cache for RAM and PG rows`; `F01 known private observer or source suppresses a stale visibility flag`; CoreScope `TestF01CorePublicPolicyOverridesFrozenRedisVisibility`; isolated pending/commit flip. | -| F02 | `F02 contended lock and replay never change historical tables` (real PostgreSQL); `F02 refuses a contended writer before any mutation and gates the hourly predicate`. | -| F03 | `F03 canonical observation key distinguishes hop, path, direction, observer and scope`; `F03 restart probe checks the same identity`; real PostgreSQL #122 CI case. | -| F04 | `F04 shared-subscription N>1 is gated until restart and resize recovery is qualified`. | -| F05 | `F05 lost Redis reply recovers the original observation ID without a second XADD`; `F05 two same-millisecond sightings with distinct hop/path and IDs both persist`; real Redis admission replay and final-head ambiguous SQL-commit replay. | -| F06 | `F06 retained overlap and newer admission remain readable across a commit marker`; `F06 an old pending hole remains discoverable`; `F06 retention starts at commit time`; final-head marker-gap gate passed. | -| F07 | `F07 ledger cursor follows completed writes even when the second observation is older` (real PostgreSQL); CoreScope `TestF07LedgerCursorUsesCommitOrderRatherThanObservedTime`, `TestF07RedisTailRequiresPairedCompletenessSource`. | -| F08 | CoreScope `TestF08TrackedJobReportsActiveCompletionAndWriteFailure`, `TestF08SourceBatchWaitsForLastJobAndReturnsWriteFailure`, historical handler SQLite/observation failure cases. | -| F09 | `F09 interrupted cleanup releases bytes exactly once on retry`; real Redis F05/F09 admission replay and atomic byte reconciliation; final-head concurrent byte cap. | -| F10 | `F10 destroyed group is repaired before another admitted entry is stranded`; `F10 young foreign PEL is reclaimed after it ages without a restart`; final-head NOGROUP gate passed. | -| F11 | `F11 rejects invalid UUIDs, SQL integers, JSON and path arrays before persistence`; `F11 malformed entries survive quarantine failure`; `F11 a permanent SQL poison row is quarantined while a valid neighbor commits`; real ioredis quarantine CI. | -| F12 | `F12 merge deduplicates committed overlap by ID and counts a new same-observer sighting`. | -| F13 | `F13 RAM recent rows honor family scope, aliases, window, and slim DTO`; corrected final-head real-SQL full/slim parity fixture passed. | -| F14 | `F14 follow-up inflight and drop metrics appear in the served registry`; `F14 pending age uses admitted time and skips committed overlap`; `F14 an older reclaimed entry behind a young entry triggers the age flush`. | -| F15 | `F15 one MQTT shutdown deadline includes a blocked disconnect`; `F15 shutdown joins the active persist before attempting to drain`; `F15 an active write past the deadline reports an incomplete shutdown`; `F15 drain reports an uncommitted PEL`. | -| F16 | CI `Validate Compose configuration` RAM render assertion and empty-volume smoke; final-head standalone drain, quarantine-refusal and durable rollback gate passed on `f70fe05`. | -| F17 | CoreScope `TestF17TwoSourceObservationIdentityAndCountInBothOrders`, `TestF17HistoricalSourcesWaitForIdentityIndexMigration`. | +Commit/push the locally verified RAM wave, watch both new CI +workflows to actual success, build a source-labelled image and rerun the strict +compiled gates. Then run an independently isolated whole-stack campaign with +actual CoreScope processes, ready/joined readers, exact observation and +side-effect checks, complete measurement windows, Redis AOF, PG WAL and physical +I/O. Compare durable/RAM using the same final code and repeated workloads; +report overhead as well as latency. Update both existing PR descriptions while +preserving #122 open/non-draft and #125 open/draft. Do not merge or deploy. -The §7 advisory-lock snapshot caveat is addressed in code by a separate-statement same-client transaction and unique identity ledger; its real-PG lock/replay regression passed in CI. The §4 eleven-item verification and §5 methodology repairs remain incomplete. Current tests do not establish immediate CoreScope revocation, exact RAM/SQL read parity, or whole-stack performance. +RAM remains opt-in: ACK precedes PG commit and a Redis loss can lose acknowledged +telemetry. Current policy unavailability/rematerialization deliberately makes +public reads unavailable. Previously delivered/network-in-flight bytes cannot +be recalled. Old untracked/mixed CoreScope data is preserved and fenced; migrate +to a new dedicated projection backed by retained PG history. Best-effort +post-commit enrichment is bounded but not transactional/exactly once. N>1 +remains gated. No whole-stack savings or three-repeat performance claim exists +for the new code yet. The historical measurements below remain evidence only +for their explicitly older revisions. -## Review §4 issue check - -| Prior issue | Current-head evidence and remaining gate | -|---|---| -| 1. Commit before ACK double count | F02 contended-lock/replay PG test passes; F05 stable admission identity and ledger tests pass locally. Final-head ambiguous SQL-commit replay passes 120/120 with 60 paths. The duplicate-publish run acknowledged 1,440 main deliveries but persisted exactly 1,200 distinct main observations, with duplicate-key count zero, max row multiplicity one, and one path per observation. Exact lost Redis admission reply remains separate. | -| 2. Atomic limits | Lua count/byte admission and missing-counter reconciliation have unit and real-Redis CI tests. Final-head concurrent count admission passed 180/180 at cap 180, with 220/220 refusals; byte admission passed 564 entries and 936 refusals under the 1 MiB bound. In a 3,000-message burst with a 2,000-entry RAM cap, the default isolated Mosquitto queue logged outgoing drops: only 2,785/3,100 warmup-plus-main identities reached PG, despite all 3,000 main publisher broker ACKs. With a bench-only `max_queued_messages 10000`, the identical burst passed 3,100/3,100 exact identities, all historical totals, 407 explicit queue-full refusals, no duplicate rows, and zero final queue/PEL/reserved bytes. Broker queue sizing is a deployment requirement; the failed run remains evidence. | -| 3. Queue age | Pending, retained and PEL are separate; the flush trigger uses the earliest admitted entry. The final-head batch-500 live scrape shows pending=203, retained=1,500, PEL=203, reserved bytes=4,138,800 and oldest pending age=20.261 seconds. | -| 4. Privacy | Fresh policy and rematerialization fences cover UKMesh public reads and new CoreScope batches. Final-head isolated observer flip test passed: admitted public, hidden by pending read after flip, committed private. The sustained synthetic privacy run passed 1,600/1,600 PG identities, but its post-flip API/WS audit is inconclusive for selective observer filtering: two rematerialization jobs remained pending, so both public probes saw zero records after the flip (fail closed). Immediate CoreScope revocation remains unresolved. | -| 5. Marker gaps | Per-entry commit state replaces cumulative marker authority; gap and retained-overlap unit tests pass. The final-head admit-during-commit gate observed both rows in the live overlay before and after the first commit, with the second still pending. | -| 6. Multi-observer aggregate | Stable observation IDs prevent observer-membership dedupe errors; alias/window/DTO tests and a corrected real-SQL full/slim parity fixture pass. The isolated three-observer fanout persisted 2,700/2,700 main and 100/100 warmup identities with no duplicate rows. | -| 7. Message type | WS initial message merge limits pending rows to GroupText and scans the bounded whole stream. Long REST channel history remains PG-only. | -| 8. Background work | Post-commit hooks remain best effort and bounded. A crash after PG commit may omit a hook; it is not an exactly-once side effect. | -| 9. Quarantine adapter | Writer uses real ioredis XADD, validates every read path, and acknowledges poison only after quarantine. Real-Redis CI passes; the final-head SQLSTATE 23514 poison-neighbor gate quarantined one row through real ioredis, committed the valid PG neighbor and left PEL zero. | -| 10. Byte drift | Atomic final deletion and Redis-side reconciliation cover the reservation counter. Real-Redis CI passes; startup/restart accounting gate remains. | -| 11. Handoff | Retained PG overlap and committed-order ledger cover late commits after migration. End-to-end two-source offline, retention and cursor-file-loss gates remain. | - -## CoreScope relay-push record - -Local commit `e9afb3a8a17f8b65d1463985df18984c8f8b86f4` on `feat/redis-live-source` is ready to export with `git format-patch -1 e9afb3a8a17f8b65d1463985df18984c8f8b86f4`. It changes only `cmd/ingestor`: paired PG/Redis sources hold cursors through completed SQLite jobs, follow the UKMesh committed-order ledger, recheck current privacy policy for new batches, preserve upstream observation identity, and dedupe observer counts. Migration to the two identity indexes is asynchronous and gates the sources until ready. Both full Go suites pass. The coordinator must relay-push this local commit; it was not pushed here. - -## Existing test observations - -- #122 initial full-suite run: 417/418, with a timing failure in the unrelated `path-beta/workerPool` abort test. Focused rerun: 3/3; final full suite: 418/418. Log: `/tmp/ukmesh-perf-fix-npm-test-final.log`. -- #122 CI run `38154922672` completed successfully in Backend, Frontend, Workers and Compose, and Secret scan. -- #125 final code head `022b7c8` final-head bench integration log: `/home/ben/ukmesh-bench/mqtt-ingest-perf-20261011/logs/integration-20261011-174207.log`, exit 0. Crash replay and ambiguous commit each persisted 120/120 distinct observations and 60/60 paths; replay PEL was zero. Atomic count and byte admissions passed. Isolated privacy boundary log: `/tmp/ukmesh-final-privacy-boundary-rerun.log`, exit 0. -- The drain-only code head `f70fe05` passed `npx tsc --noEmit`, OpenAPI contract check, build and the full backend `npm test` (468/468). Full-suite log: `/tmp/ukmesh-final-head-f70fe05-npm-test.log`. Focused lifecycle and worker-pool checks passed 5/5: `/tmp/ukmesh-final-head-f70fe05-focused.log`. -- Live final-head metric scrape: `/home/ben/ukmesh-bench/mqtt-ingest-perf-20261011/logs/final-head-metric-scrape-b500-20261011-1843.txt`. The same scrape reports 2,180 dropped best-effort post-persist follow-ups. This is a measured nontransactional enrichment limitation, not a missing packet/path/hourly/region database row. -- Isolated shutdown manual gate: `/tmp/ukmesh-s9-manual-gate.log`. PostgreSQL packets=300, Redis stream=0, PEL=0 and all queue gauges/byte reservation=0 after SIGTERM and restart; a separate exact-identity audit passed 300/300. The stock bench `wait-drain.sh` did not emit `DRAINED` because the post-restart `processed` counter had no zero-valued series, so its wrapper exit is a benchmark-oracle failure, not evidence of uncommitted observations. -- Final-head fault gates: `/home/ben/ukmesh-bench/mqtt-ingest-perf-20261011/logs/final-head-fault-gates-20261011-195949.log` records lost admission reply, marker gap, NOGROUP repair and SQL poison passes. Its first SQL parity invocation failed because the bench fixture supplied `PacketBatchInput` fields to an `InsertPacketInput` API and persisted a separately timed row. The corrected fixture uses public input and persists the exact admitted entry; `/home/ben/ukmesh-bench/mqtt-ingest-perf-20261011/logs/final-head-sql-parity-corrected-20261011-2001.log` passes full and slim DTO comparison. -- Standalone drain first exposed two real exit-path defects: a Redis command raced connection readiness, and two path-resolution worker threads held the CLI process open after `RAM_DRAIN_COMPLETE`. `f70fe05` connects explicitly, retains the ready Redis client across writer shutdown for the final byte check, and closes that worker pool. On the exact `f70fe05` bench image, `/home/ben/ukmesh-bench/mqtt-ingest-perf-20261011/logs/final-head-drain-20261011-204138.log` records exit-zero drain, refusal of one quarantined entry, zero final stream/quarantine/bytes and a ready durable-mode backend. The negative invocation output is `/home/ben/ukmesh-bench/mqtt-ingest-perf-20261011/logs/final-head-drain-quarantine-negative.log`. - -## Final-code isolated bench matrix +## Historical bench measurements — earlier code only Both Docker images carry the source revision label (`cbf78a8` durable, `022b7c8` RAM). The isolated final-chain marker preceded these builds and runs. The strict gate compares publisher distinct `(packet hash, observer)` observations against packet, path, hourly and, except for the privacy case, region deltas. From the sustained durable run onward, it also saves the exact PostgreSQL identity set for a separate equality audit. Reader containers are stopped before the next scenario's reset. WAL bytes are from `pg_stat_wal`; Redis AOF size is compared only if no rewrite occurred. CPU and memory below cover backend, Redis and TimescaleDB; broker, publisher, reader probes and CoreScope are not included. @@ -307,7 +143,7 @@ Both Docker images carry the source revision label (`cbf78a8` durable, `022b7c8` These numbers are resource observations for isolated completed windows, not whole-stack or production savings. †The normal scenario runner measures CPU through its final drain. RAM retains committed entries for about 120 seconds, so its run can include a longer post-publisher period than durable mode; the sweep's CPU totals alone are not an equal-duration comparison. ‡The stock bench runner timed out at 240 seconds with 100 already committed entries still in retention. A separate post-timeout wait then observed stream=0 and PEL=0 before the next scenario; its first after-snapshot and CPU window ended at the timeout. The bench-only resumed runner uses a 420-second drain wait for larger batches. The original reader probes start asynchronously; a two-second burst can begin before WebSocket connection, so their burst WebSocket coverage is not a delivery guarantee. The clean-warmup rerun below captures publisher-end and all-observations-committed CPU/WAL/AOF snapshots for both variants, then reports retained cleanup separately. -### Joined-reader clean-window comparison +### Historical joined-reader comparison The clean runner waits for all four reader probes to signal ready before publishing. For RAM, it first waits until all 100 warmup observations are committed with PEL zero, then trims only those retained warmup stream entries while the isolated backend is paused. The comparison endpoint is all expected observations committed, with RAM PEL zero; retained cleanup is measured afterward. CPU includes backend, Redis and TimescaleDB, and excludes broker, publishers and reader containers. WAL comes from `pg_stat_wal`; AOF is the Redis persistence delta. Each row is a single run, so these are workload observations rather than repeatable savings estimates. @@ -324,4 +160,11 @@ The six-run log is `/home/ben/ukmesh-bench/mqtt-ingest-perf-20261011/logs/clean- ## Operational boundary -No production database, Redis, broker, service, configuration, deployment or running process has been modified. No PR has been merged or approved. Only the isolated `mqttbench` rig was used after its `FINAL-CHAIN-DONE` marker. CoreScope has not been pushed. +Production is untouched: no production DB, Redis, broker, configuration, +service, process or deployment was modified. No existing service was restarted, +and no PR was merged or approved. The original campaign's `FINAL-CHAIN-DONE` +was personally observed at 17:39:52 UTC before isolated verification began. +New fixtures use disposable owned DBs/keys/containers/processes and the owned +`verification-sol-20261011` directory; other run directories remain read-only. +One old owned WS diagnostic process was stopped after its evidence was captured. +CoreScope has not been pushed. The full mission remains active. diff --git a/backend/package.json b/backend/package.json index d04bfe0..215279c 100644 --- a/backend/package.json +++ b/backend/package.json @@ -22,6 +22,7 @@ "test": "node --import tsx --test $(find src -name '*.test.ts' ! -name '*.integration.test.ts' -print) && bash ../scripts/test-alert-receiver.sh", "test:analysis-integration": "node --import tsx --test src/analysis/runState.integration.test.ts", "test:ingest-integration": "timeout 300s node --import tsx --test src/db/packetBatch.integration.test.ts src/db/recentRead.integration.test.ts && timeout 90s node --import tsx --test src/ram/ram.integration.test.ts && timeout 60s node --import tsx --test src/ws/publicAuthorization.integration.test.ts", + "test:ram-operational": "timeout 300s node ../scripts/test-ram-redis-acl.mjs", "typecheck": "tsc --noEmit", "contract:generate": "tsx src/tools/generateOpenApi.ts", "contract:check": "tsx src/tools/generateOpenApi.ts --check", diff --git a/backend/src/db/packetBatch.integration.test.ts b/backend/src/db/packetBatch.integration.test.ts index 50fff54..a8221c5 100644 --- a/backend/src/db/packetBatch.integration.test.ts +++ b/backend/src/db/packetBatch.integration.test.ts @@ -26,7 +26,7 @@ import { import { processNextPrivacyRematerialization } from '../workers/privacy-rematerialization.js'; import { packetAlreadyPersisted } from '../mqtt/observationIdentity.js'; import { admitPacketToRam, configureRamStore } from '../ram/liveStore.js'; -import { readRecentWithPending } from '../ram/liveRead.js'; +import { readRecentWithPending, resolvePendingSummaries } from '../ram/liveRead.js'; import { createRamWriter } from '../ram/writer.js'; import { getRamStreamKey } from '../ram/protocol.js'; @@ -356,14 +356,16 @@ test('F06 real PG snapshot and Redis expiry preserve the captured handoff', { await snapshot.query('ROLLBACK'); snapshot.release(); } - }, { limit: 10, network: 'ukmesh', resolveAliases: async () => new Map() }); + }, { limit: 10, network: 'ukmesh', resolveAliases: async () => new Map(), + resolveSummaries: entries=>resolvePendingSummaries(entries,(sql,params)=>pool.query(sql,params)) }); assert.deepEqual(merged.map((row) => row.packet_hash), [input.packetHash]); const later = await readRecentWithPending(async () => (await pool.query(` SELECT packet_hash, time, network, ARRAY[observation_id::text] AS _observation_ids, (SELECT generation::text FROM public_visibility_state WHERE singleton) AS _policy_generation, 0 AS _representative_rank, 1::int AS rx_count, 0::int AS tx_count FROM packets`)).rows, - { limit: 10, network: 'ukmesh', resolveAliases: async () => new Map() }); + { limit: 10, network: 'ukmesh', resolveAliases: async () => new Map(), + resolveSummaries: entries=>resolvePendingSummaries(entries,(sql,params)=>pool.query(sql,params)) }); assert.deepEqual(new Set(later.map((row) => row.packet_hash)), new Set([input.packetHash, '61'.repeat(32)])); const durableIds = (await pool.query('SELECT observation_id::text FROM packets')).rows.map((row) => row.observation_id); assert.deepEqual(durableIds, [first.observationId]); diff --git a/backend/src/db/recentRead.integration.test.ts b/backend/src/db/recentRead.integration.test.ts index abb7aef..f12e12e 100644 --- a/backend/src/db/recentRead.integration.test.ts +++ b/backend/src/db/recentRead.integration.test.ts @@ -8,7 +8,7 @@ import { runMigrations } from './migrations.js'; import { configurePacketBatch, persistEntriesCore, type QueryFn } from './packetBatch.js'; import { admitPacketToRam, configureRamStore, readPendingEntries } from '../ram/liveStore.js'; import type { InsertPacketInput } from './packetInput.js'; -import { readRecentWithPending, mergePendingPacketsIntoRecentRows } from '../ram/liveRead.js'; +import { readRecentWithPending, mergePendingPacketsIntoRecentRows, resolvePendingSummaries, type PendingSummary } from '../ram/liveRead.js'; import { getRamStreamKey } from '../ram/protocol.js'; import { processNextPrivacyRematerialization } from '../workers/privacy-rematerialization.js'; @@ -74,6 +74,7 @@ test('F01/F05/F12/F13 real SQL and RAM contracts agree across scoped paging, sta (source_node_id, canonical_node_id, confidence, reason) VALUES ($1, $2, 'high', 'fixture')`, [alias, canonical]); while (await processNextPrivacyRematerialization(pool, { chunkPauseMs: 0 })) { /* finish fixture policy */ } const aliases = async () => new Map([[alias, canonical]]); + const summaries = (entries: PendingSummary[]) => resolvePendingSummaries(entries,query); const observed = Date.now(); const base = { packetHash: '80'.repeat(32), rxNodeId: alias, topic: `meshcore/ZZZ/${alias}/packets`, network: 'northeast', rawHex: '00', packetType: 5, pathHashes: [] as string[], @@ -91,13 +92,18 @@ test('F01/F05/F12/F13 real SQL and RAM contracts agree across scoped paging, sta await admit({ ...base, packetHash: '81'.repeat(32), time: new Date(observed - 3_500) }), await admit({ ...base, packetHash: '82'.repeat(32), network: 'othernet' }), await admit({ ...base, packetHash: '83'.repeat(32), time: new Date(observed - 301_000) }), + await admit({ ...base, packetHash: '84'.repeat(32), payload: {}, time:new Date(observed-2_500) }), + await admit({ ...base, packetHash: '85'.repeat(32), payload: {_summary:42}, time:new Date(observed-2_600) }), + await admit({ ...base, packetHash: '86'.repeat(32), payload: {_summary:{z:false,a:'text'}}, time:new Date(observed-2_700) }), ]; + await pool.query('INSERT INTO packet_decryptions(packet_hash,decrypted,summary) VALUES ($1,$2,$3)', + ['84'.repeat(32),{},'decrypted fallback summary']); const merged = new Map(); for (const scope of [undefined, 'ukmesh', 'northeast', 'teesside', 'othernet']) { for (const observer of [undefined, alias, canonical, other]) { for (const fields of ['full', 'slim'] as const) { for (const limit of [1, 5]) { - const opts = { limit, network: scope, observer, fields, resolveAliases: aliases, nowMs: observed }; + const opts = { limit, network: scope, observer, fields, resolveAliases: aliases, resolveSummaries:summaries, nowMs: observed }; merged.set(JSON.stringify(opts), json(await readRecentWithPending( (hashes) => getRecentPackets(limit, scope, observer, fields, undefined, query, hashes), opts))); } @@ -105,7 +111,7 @@ test('F01/F05/F12/F13 real SQL and RAM contracts agree across scoped paging, sta } } const messages = json(await readRecentWithPending((hashes) => getRecentMessages(5, 'ukmesh', undefined, undefined, query, hashes), - { limit: 5, network: 'ukmesh', rowKind: 'messages', packetTypes: [5], resolveAliases: aliases, nowMs: observed })); + { limit: 5, network: 'ukmesh', rowKind: 'messages', packetTypes: [5], resolveAliases: aliases, resolveSummaries:summaries, nowMs: observed })); await persistEntriesCore(pending, query, true, transaction); await redis.del(key); // compare against the complete SQL projection, without overlap for (const [serialized, expected] of merged) { @@ -122,7 +128,7 @@ test('F01/F05/F12/F13 real SQL and RAM contracts agree across scoped paging, sta await snapshot.query('BEGIN ISOLATION LEVEL REPEATABLE READ'); const before = await getRecentPackets(5, 'ukmesh', undefined, 'full', undefined, (sql, params) => snapshot.query(sql, params)); - assert.equal(before[0]?.rx_count, 3); + assert.equal(before.find(row=>row.packet_hash==='80'.repeat(32))?.rx_count, 3); await pool.query('UPDATE public_visibility_state SET generation = generation + 1 WHERE singleton'); await pool.query(`UPDATE packet_visibility_materialization_state SET visibility_generation = (SELECT generation FROM public_visibility_state WHERE singleton) WHERE singleton`); diff --git a/backend/src/ram/liveRead.ts b/backend/src/ram/liveRead.ts index e95b437..c268b4d 100644 --- a/backend/src/ram/liveRead.ts +++ b/backend/src/ram/liveRead.ts @@ -20,12 +20,13 @@ * remain available. */ import { readPendingEntries, type PendingStreamEntry } from './liveStore.js'; -import { loadCurrentPublicPrivacyPolicy } from '../db/packetBatch.js'; +import { loadCurrentPublicPrivacyPolicy, type QueryFn } from '../db/packetBatch.js'; import { query } from '../db/index.js'; import { networkMatchesScope } from '../networks.js'; import { getRamIngestMode, type RamPendingEntry } from './protocol.js'; export type RecentRow = Record; +export type PendingSummary = { packet_hash: string; payload: { _summary: unknown } }; export type MergeOptions = { limit: number; @@ -38,6 +39,8 @@ export type MergeOptions = { nowMs?: number; /** Supplied by tests; production resolves stored node aliases in one PG query. */ resolveAliases?: (nodeIds: string[]) => Promise>; + /** Match SQL JSONB text extraction and packet-decryption fallback in one query. */ + resolveSummaries?: (entries: PendingSummary[]) => Promise>; /** Restrict pending contributions to these packet types (e.g. [5] for message history). */ packetTypes?: number[]; rowKind?: 'packets' | 'messages'; @@ -95,6 +98,14 @@ async function resolveCanonicalAliases(nodeIds: string[]): Promise [row.source_node_id, row.canonical_node_id])); } +export async function resolvePendingSummaries(entries: PendingSummary[], readQuery: QueryFn = query): Promise> { + const result = await readQuery(` + SELECT p.packet_hash, COALESCE(p.payload->>'_summary', pd.summary) AS summary + FROM jsonb_to_recordset($1::jsonb) AS p(packet_hash text, payload jsonb) + LEFT JOIN packet_decryptions pd ON pd.packet_hash = p.packet_hash`, [JSON.stringify(entries)]); + return new Map(result.rows.map(row => [String(row.packet_hash), row.summary === null ? null : String(row.summary)])); +} + function canonical(value: string | null, aliases: Map): string | null { return value ? aliases.get(value) ?? value : null; } @@ -153,6 +164,7 @@ function buildMergedRow(group: EntryWithTx[], includePayload: boolean, fields: ' ...(includePayload || rowKind === 'messages' ? { payload } : {}), _representative_rank: representativeRank(representative, rowKind), _representative_id: representative.entry.observationId, + _pending_summary: payload?.['_summary'] ?? null, }; } @@ -207,6 +219,7 @@ function stripInternalIds(rows: T[]): T[] { if (row && typeof row === 'object') delete (row as RecentRow)['_policy_generation']; if (row && typeof row === 'object') delete (row as RecentRow)['_representative_rank']; if (row && typeof row === 'object') delete (row as RecentRow)['_representative_id']; + if (row && typeof row === 'object') delete (row as RecentRow)['_pending_summary']; } return rows; } @@ -234,6 +247,7 @@ export async function mergePendingPacketsIntoRecentRows( ): Promise { if (getRamIngestMode() !== 'ram' || opts.limit <= 0) return stripInternalIds(rows); try { + const pending = opts.pendingSnapshot ?? await readPendingEntries(opts.limit); const policy = await loadCurrentPublicPrivacyPolicy(); if (!policy) return []; const base = (rows as unknown as RecentRow[]).filter((row) => @@ -247,7 +261,6 @@ export async function mergePendingPacketsIntoRecentRows( // The store scans the bounded stream, then this layer filters and pages // distinct hashes. Sampling only the newest N observations would hide a // matching hash behind a large observer fanout or other packet types. - const pending = opts.pendingSnapshot ?? await readPendingEntries(opts.limit); if (pending.length === 0) return stripInternalIds(base.slice(0, opts.limit) as unknown as T[]); const cutoff = (opts.nowMs ?? Date.now()) - (opts.windowMs ?? (opts.packetTypes ? 86_400_000 : 300_000)); const eligible = pending.filter(({ entry }) => { @@ -308,11 +321,23 @@ export async function mergePendingPacketsIntoRecentRows( merged.push(buildMergedRow(group, opts.includePayload === true, opts.fields ?? 'full', aliases, opts.rowKind ?? 'packets')); } const combined = [...base, ...merged]; + const summaries = combined.filter(row => Object.hasOwn(row, '_pending_summary') && typeof row['_pending_summary'] !== 'string') + .map(row => ({ packet_hash: String(row['packet_hash']), payload: { _summary: row['_pending_summary'] } })); + if (summaries.length) { + const resolved = await (opts.resolveSummaries ?? resolvePendingSummaries)(summaries); + for (const row of combined) if (Object.hasOwn(row, '_pending_summary') && typeof row['_pending_summary'] !== 'string') { + row['summary'] = resolved.get(String(row['packet_hash'])) ?? null; + } + } + // Alias and summary queries may wait while authorization changes. Their + // captured generation is never permission to deliver after those waits. + const current = await loadCurrentPublicPrivacyPolicy(); + if (!current || current.currentGeneration !== policy.currentGeneration) return []; combined.sort((left, right) => timeMs(right['time']) - timeMs(left['time']) || String(left['packet_hash']).localeCompare(String(right['packet_hash']))); return stripInternalIds(combined.slice(0, opts.limit) as unknown as T[]); } catch (error) { - console.error('[ram] live merge failed (returning database rows only)', (error as Error).message); + console.error('[ram] live merge failed (public read unavailable)', (error as Error).message); return []; } } diff --git a/backend/src/ram/liveStore.test.ts b/backend/src/ram/liveStore.test.ts index cd71d05..d19a232 100644 --- a/backend/src/ram/liveStore.test.ts +++ b/backend/src/ram/liveStore.test.ts @@ -25,6 +25,9 @@ const mergePendingPacketsIntoRecentRows = (rows: T[], opts: MergeOptions): Pr mergeRecentRows(rows, { windowMs: Number.MAX_SAFE_INTEGER, resolveAliases: async () => new Map(), + resolveSummaries: async entries => new Map(entries.map(entry => [entry.packet_hash, + entry.payload._summary === null ? null : typeof entry.payload._summary === 'string' + ? entry.payload._summary : JSON.stringify(entry.payload._summary)])), ...opts, }); @@ -572,3 +575,13 @@ test('F13 representative ties use stable identity and paging preserves milliseco assert.equal(merged[0]?.rssi,-25,'equal-rank, equal-time representative chooses greatest stable ID'); assert.equal('_representative_id' in merged[0]!,false); }); + +test('F01 alias lookup cannot deliver a captured generation after authority changes', async () => { + let generation=7; + configurePacketBatch(async()=>({rows:[{generation,materialized_generation:generation,remat_active:false,prefixes:[]}]})); + configureRamStore(new FakeStore()); + const entry=entryFor('86'.repeat(32)); + const rows=await mergePendingPacketsIntoRecentRows([], {limit:1,pendingSnapshot:[{id:'401-0',entry}], + resolveAliases:async()=>{generation=8;return new Map();}}); + assert.deepEqual(rows,[],'the final delivery fence must follow all asynchronous enrichment'); +}); diff --git a/backend/src/ram/liveStore.ts b/backend/src/ram/liveStore.ts index 42cc764..c0a5996 100644 --- a/backend/src/ram/liveStore.ts +++ b/backend/src/ram/liveStore.ts @@ -36,6 +36,7 @@ import { getRamQueueMaxPackets, getRamStreamKey, RAM_ENTRY_OVERHEAD_BYTES, + RAM_REDIS_HEADROOM_BYTES, type RamPendingEntry, } from './protocol.js'; import { @@ -67,7 +68,7 @@ export interface RamStoreRedis { max: string, ...args: (string | number)[] ): Promise>; - memory(subcommand: string, key: string): Promise; + memory(subcommand: string, key: string): Promise; get(key: string): Promise; eval(script: string, numKeys: number, ...args: (string | number)[]): Promise; hmget(key: string, ...fields: string[]): Promise>; @@ -91,6 +92,13 @@ if (len > 0 and not recorded) or (recorded and (not tonumber(recorded) or tonumb local used = tonumber(recorded or '0') local add = tonumber(ARGV[3]) if used + add > tonumber(ARGV[2]) then return -2 end +local memory = redis.call('INFO', 'memory') +local maxmemory = tonumber(string.match(memory, 'maxmemory:(%d+)')) or 0 +local resident = tonumber(string.match(memory, 'used_memory:(%d+)')) +local policy = string.match(memory, 'maxmemory_policy:([^%s]+)') +local headroom = tonumber(ARGV[7]) +if not resident or maxmemory == 0 or policy ~= 'noeviction' or tonumber(ARGV[2]) + headroom > maxmemory + or resident + add + headroom >= maxmemory then return -4 end local id = redis.call('XADD', KEYS[1], '*', ARGV[4], ARGV[5]) redis.call('INCRBY', KEYS[2], add) redis.call('SET', KEYS[3], ARGV[5]) @@ -138,7 +146,7 @@ export type RamAdmissionResult = { }; export class RamAdmissionRefusedError extends Error { - readonly reason: 'queue_full' | 'queue_bytes' | 'redis_unavailable'; + readonly reason: 'queue_full' | 'queue_bytes' | 'redis_unavailable' | 'redis_headroom'; constructor(reason: RamAdmissionRefusedError['reason'], cause?: unknown) { super(`RAM_INGEST_REFUSED:${reason}${cause ? `:${(cause as Error).message ?? String(cause)}` : ''}`); @@ -223,9 +231,12 @@ export async function refreshPendingStats(): Promise<{ pending: number; pel: num client.memory('USAGE', key).catch(() => cachedBytes), client.get(getRamQueueBytesKey()), client.xrange(key, '-', '+', 'COUNT', getRamQueueMaxPackets()), - client.xpending(key, getRamConsumerGroup()), + client.xpending(key, getRamConsumerGroup()).catch((error: unknown) => { + if ((error as Error).message?.includes('NOGROUP')) return [0, null, null, []]; + throw error; + }), ]); - cachedBytes = typeof mem === 'number' && Number.isFinite(mem) ? mem : cachedBytes; + cachedBytes = mem === null ? 0 : typeof mem === 'number' && Number.isFinite(mem) ? mem : cachedBytes; ramQueueDepth.set(len); ramQueueBytes.set(cachedBytes); ramQueueReservedBytes.set(Math.max(0, Number(reserved ?? 0) || 0)); @@ -304,6 +315,7 @@ export async function admitPacketToRam( ADMIT_SCRIPT, 3, key, getRamQueueBytesKey(), getRamAdmissionKey(deliveryKey), maxPackets, maxBytes, entryBytes, entryFieldName(), value, delivery?.redelivered ? 1 : 0, + RAM_REDIS_HEADROOM_BYTES, ); let result = await runAdmission(); if (result === -3) { @@ -318,6 +330,10 @@ export async function admitPacketToRam( ramAdmissionsTotal.inc({ outcome: 'queue_bytes' }); throw new RamAdmissionRefusedError('queue_bytes'); } + if (result === -4) { + ramAdmissionsTotal.inc({ outcome: 'redis_headroom' }); + throw new RamAdmissionRefusedError('redis_headroom'); + } if (!Array.isArray(result) || result.length !== 2 || typeof result[1] !== 'string') { throw new RamAdmissionRefusedError('redis_unavailable', new Error('unexpected eval reply')); } diff --git a/backend/src/ram/memory.test.ts b/backend/src/ram/memory.test.ts new file mode 100644 index 0000000..37cb3d9 --- /dev/null +++ b/backend/src/ram/memory.test.ts @@ -0,0 +1,12 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { validateRamRedisHeadroom } from './memory.js'; + +test('F09 startup refuses unlimited, evicting, full or undersized Redis', () => { + const info = (max: number, used: number, policy='noeviction') => + `used_memory:${used}\r\nmaxmemory:${max}\r\nmaxmemory_policy:${policy}\r\n`; + assert.doesNotThrow(()=>validateRamRedisHeadroom(info(96*1048576,1048576),32*1048576)); + for (const value of [info(0,0),info(48*1048576,0),info(96*1048576,80*1048576),info(96*1048576,0,'allkeys-lru'),'']) { + assert.throws(()=>validateRamRedisHeadroom(value,32*1048576),/RAM_REDIS_HEADROOM_UNSAFE/); + } +}); diff --git a/backend/src/ram/memory.ts b/backend/src/ram/memory.ts new file mode 100644 index 0000000..519b2a6 --- /dev/null +++ b/backend/src/ram/memory.ts @@ -0,0 +1,12 @@ +import { getRamQueueMaxBytes, RAM_REDIS_HEADROOM_BYTES } from './protocol.js'; + +export function validateRamRedisHeadroom(info: string, queueBudget = getRamQueueMaxBytes()): void { + const number = (name: string) => Number(info.match(new RegExp(`(?:^|\\n)${name}:(\\d+)`))?.[1]); + const max = number('maxmemory'), used = number('used_memory'); + const policy = info.match(/(?:^|\n)maxmemory_policy:([^\s]+)/)?.[1]; + if (!Number.isSafeInteger(max) || max <= 0 || !Number.isSafeInteger(used) || used < 0 + || policy !== 'noeviction' || queueBudget + RAM_REDIS_HEADROOM_BYTES > max + || used + RAM_REDIS_HEADROOM_BYTES >= max) { + throw new Error('RAM_REDIS_HEADROOM_UNSAFE: bounded noeviction Redis and queue/recovery headroom required'); + } +} diff --git a/backend/src/ram/operational.integration.test.ts b/backend/src/ram/operational.integration.test.ts new file mode 100644 index 0000000..c623193 --- /dev/null +++ b/backend/src/ram/operational.integration.test.ts @@ -0,0 +1,146 @@ +import assert from 'node:assert/strict'; +import http from 'node:http'; +import { once } from 'node:events'; +import test from 'node:test'; +import { pathToFileURL } from 'node:url'; +import express from 'express'; +import { Redis } from 'ioredis'; +import { configurePacketBatch } from '../db/packetBatch.js'; +import { createRamWriter } from './writer.js'; +import { accountedRamEntryBytes, decodeRamStreamValue, RAM_REDIS_HEADROOM_BYTES, + RAM_QUARANTINE_MAX_ENTRIES, RAM_QUARANTINE_ENTRY_MAX_BYTES } from './protocol.js'; +import { PrivatePrefixCache } from '../privacy/privatePrefixCache.js'; +import { createPublicRamSource } from './publicSource.js'; + +const urls = ['TEST_REDIS_ACL_URL','TEST_REDIS_ADMIN_URL','TEST_REDIS_PUBLIC_URL','TEST_REDIS_OPERATOR_URL'].map(name=>process.env[name]); +const ready = urls.every(url=>url&&['127.0.0.1','localhost'].includes(new URL(url).hostname)); +const packet = (i:number)=>({packetHash:i.toString(16).padStart(64,'0'),rxNodeId:'CD'.repeat(32), + topic:`meshcore/ZZZ/${'CD'.repeat(32)}/packets`,rawHex:'DEADBEEF',packetType:4,network:'ukmesh',payload:{data:'x'.repeat(8_000)}}); + +async function fixture(t: test.TestContext, budget:number) { + const clients=urls.map(url=>new Redis(url!,{lazyConnect:true,enableReadyCheck:false,enableOfflineQueue:false,maxRetriesPerRequest:1})); + await Promise.all(clients.map(async client=>{await client.connect();await client.ping();})); + const [writer,admin,publicClient,operator]=clients as [Redis,Redis,Redis,Redis]; + const old=Object.fromEntries(['MQTT_INGEST_MODE','MQTT_RAM_STREAM_KEY','MQTT_RAM_QUEUE_MAX_BYTES','MQTT_RAM_QUEUE_MAX_PACKETS'].map(name=>[name,process.env[name]])); + process.env['MQTT_INGEST_MODE']='ram'; + const key=`meshcore:acl:${process.pid}:${Date.now()}`; + process.env['MQTT_RAM_STREAM_KEY']=key; + process.env['MQTT_RAM_QUEUE_MAX_BYTES']=String(budget); + process.env['MQTT_RAM_QUEUE_MAX_PACKETS']='1000'; + const source=process.env['TEST_RAM_SOURCE_DIR']; + const store = source ? await import(pathToFileURL(source+'/ram/liveStore.ts').href) : await import('./liveStore.js'); + const batch = source ? await import(pathToFileURL(source+'/db/packetBatch.ts').href) : {configurePacketBatch}; + batch.configurePacketBatch(async()=>({rows:[{generation:'1',materialized_generation:'1',remat_active:false,prefixes:[]}]})); + store.configureRamStore(writer); + t.after(async()=>{ + store.configureRamStore(null);batch.configurePacketBatch(null); + const keys=await admin.keys(key+'*');if(keys.length)await admin.del(...keys); + await Promise.all(clients.map(client=>client.quit())); + for(const [name,value] of Object.entries(old)){if(value===undefined)delete process.env[name];else process.env[name]=value;} + }); + return {writer,admin,publicClient,operator,key,store}; +} + +test('F09 actual Redis refuses unsafe headroom before admitting an observation', {skip:ready?false:'owned loopback ACL Redis fixture required'}, async t=>{ + const {writer,admin,key,store}=await fixture(t,32*1048576); + await assert.rejects(store.admitPacketToRam(packet(1)),/redis_headroom/); + assert.equal(await writer.xlen(key),0); + process.env['MQTT_RAM_QUEUE_MAX_BYTES']=String(1048576); + await admin.set(key+':other-private-load','x'.repeat(18*1048576)); + await assert.rejects(store.admitPacketToRam(packet(2)),/redis_headroom/, + 'a valid application budget also refuses admission when other resident keys consume recovery space'); + assert.equal(await writer.xlen(key),0); +}); + +test('F09 concurrent reconciliation, retained replay and lost cleanup reply conserve reservations', {skip:ready?false:'owned loopback ACL Redis fixture required'}, async t=>{ + const {writer,admin,key,store}=await fixture(t,1048576); + const outcomes=await Promise.allSettled([ + ...Array.from({length:200},(_,i)=>store.admitPacketToRam(packet(i))), + ...Array.from({length:8},()=>store.reconcileRamQueueBytes()), + ]); + const admitted=outcomes.slice(0,200).filter(result=>result.status==='fulfilled').length; + assert.ok(admitted>0&&admitted<200,'supported byte budget must refuse some admissions'); + const reserved=async()=>{ + const rows=await writer.xrange(key,'-','+'); + const expected=rows.reduce((sum,row)=>sum+accountedRamEntryBytes(row[1][1]!),0); + assert.equal(Number(await writer.get(key+':bytes')),expected); + assert.ok(expected<=1048576); + return {rows,expected}; + }; + const before=await reserved(); + await admin.del(key+':bytes'); // crash/lost counter before writer startup + await store.reconcileRamQueueBytes(); + await reserved(); + const keys=await admin.keys(key+'*'); + const physical=(await Promise.all(keys.map(key=>admin.memory('USAGE',key)))).reduce((sum,value)=>sum+Number(value??0),0); + assert.ok(physical)=>{ + const result=await target.eval(...args); + if(lostReply&&String(args[0]).includes("local at = tonumber(redis.call('HGET'")){lostReply=false;throw new Error('lost cleanup reply after Redis commit');} + return result; + }; + const value=Reflect.get(target,property);return typeof value==='function'?value.bind(target):value; + }}); + const identities=new Set(); + const running=createRamWriter({redis:proxy,persist:async entries=>{let novel=0;for(const entry of entries){if(!identities.has(entry.observationId!)){identities.add(entry.observationId!);novel++;}}return novel;}, + now:()=>now,sleep:async()=>undefined,batchSize:()=>2000,maxAgeMs:()=>1,streamKey:()=>key,group:()=> 'operational',consumer:()=> 'owned',statsTick:null,markHealthy:()=>undefined,onCommitted:null,log:()=>undefined}); + await running.start();await running.tickOnce(); + assert.equal(identities.size,admitted); + assert.equal((await writer.xpending(key,'operational'))[0],0,'all persisted records are acknowledged'); + assert.equal(await writer.hlen(key+':committed'),admitted,'each committed record has its retention timestamp'); + assert.equal((await reserved()).expected,before.expected,'commit retains the payload reservation'); + now+=120_001; + await running.sweepRetained(); + assert.equal(lostReply,false,'lost-reply boundary was exercised'); + await reserved(); + await running.sweepRetained(); + await store.reconcileRamQueueBytes(); + assert.equal((await reserved()).rows.length,0); + assert.equal((await writer.xpending(key,'operational'))[0],0); +}); + +test('F16 real Redis principals isolate raw/quarantine while public HTTP rechecks authorization', {skip:ready?false:'owned loopback ACL Redis fixture required'}, async t=>{ + const {writer,publicClient,operator,key,store}=await fixture(t,1048576); + await store.admitPacketToRam({...packet(1),payload:{}}); + for(const client of [publicClient,operator]){ + await assert.rejects(client.xrange(key,'-','+'),/NOPERM/); + await assert.rejects(client.xread('STREAMS',key,'0-0'),/NOPERM/); + } + await writer.xadd(key+':quarantine','*','q','private-evidence'); + await assert.rejects(publicClient.xrange(key+':quarantine','-','+'),/NOPERM/); + assert.equal((await operator.xrange(key+':quarantine','-','+')).length,1); + const policy=new PrivatePrefixCache();policy.replace(1,[]); + const app=express();app.use('/public',createPublicRamSource({token:()=> 'owned-fixture-token-with-at-least-32-characters',loadPolicy:async()=>policy, + read:async()=>(await writer.xrange(key,'-','+')).map(([id,fields])=>{const decoded=decodeRamStreamValue(fields[1]);assert.ok(decoded&&decoded.kind==='entry');return {id,entry:decoded.entry};})})); + const server=http.createServer(app);server.listen(0,'127.0.0.1');await once(server,'listening'); + t.after(()=>new Promise((resolve,reject)=>server.close(error=>error?reject(error):resolve()))); + const address=server.address();assert.ok(address&&typeof address!=='string'); + const response=()=>fetch(`http://127.0.0.1:${address.port}/public/entries`,{headers:{authorization:'Bearer owned-fixture-token-with-at-least-32-characters'}}); + assert.equal((await (await response()).json()).entries.length,1); + policy.replace(2,[{node_id:'CD'.repeat(32),network:'teesside',prefix_size_bytes:1,prefix:'CD'}]); + const revoked=await (await response()).json(); + assert.deepEqual(revoked.entries,[]);assert.notEqual(revoked.nextCursor,'0-0'); +}); + +test('F09/F11 real quarantine has an exact count and encoded-byte cap below reserved headroom', {skip:ready?false:'owned loopback ACL Redis fixture required'}, async t=>{ + const {writer,admin,key,store}=await fixture(t,16*1048576); + // Model existing corrupted records, which admission validation would refuse. + const pipeline=writer.pipeline(); + for(let i=0;ierror===null)); + await store.reconcileRamQueueBytes(); + const running=createRamWriter({redis:writer,persist:async()=>{throw new Error('malformed rows must never reach PG');}, + now:()=>Date.now(),sleep:async()=>undefined,batchSize:()=>1000,maxAgeMs:()=>1,streamKey:()=>key, + group:()=> 'quarantine-bound',consumer:()=> 'owned',statsTick:null,markHealthy:()=>undefined,onCommitted:null,log:()=>undefined}); + await running.start();await running.tickOnce(); + const evidence=await writer.xrange(key+':quarantine','-','+'); + assert.equal(evidence.length,RAM_QUARANTINE_MAX_ENTRIES); + assert.ok(evidence.every(([,fields])=>Buffer.byteLength(fields[1]!)<=RAM_QUARANTINE_ENTRY_MAX_BYTES)); + assert.equal(await writer.xlen(key),0);assert.equal((await writer.xpending(key,'quarantine-bound'))[0],0); + assert.equal(await store.reconcileRamQueueBytes(),0); + const keys=await admin.keys(key+'*'); + const physical=(await Promise.all(keys.map(key=>admin.memory('USAGE',key)))).reduce((sum,value)=>sum+Number(value??0),0); + assert.ok(physical ({ pathHashes: ['ABCD'], pathHashSizeBytes: 2, network: 'ukmesh', }); +test('F14 real loaded and empty queue gauges match pending, retained, PEL and bytes', { + skip: localRedis ? false : 'TEST_REDIS_URL must point to a loopback fixture', +}, async (t) => { + const client = await fixture(t); + const key = getRamStreamKey(), group = getRamConsumerGroup(); + await client.xgroup('CREATE', key, group, '0', 'MKSTREAM'); + const metrics=startMetricsServer(0);await once(metrics,'listening'); + t.after(closeMetricsServer); + const address=metrics.address();assert.ok(address&&typeof address!=='string'); + const gauge = async (suffix: string) => { + const response=await fetch(`http://127.0.0.1:${address.port}/metrics`); + assert.equal(response.status,200); + const metric=(await response.text()).match(new RegExp(`^meshcore_ram_ingest_${suffix} ([0-9.e+-]+)$`,'m')); + assert.ok(metric,`${suffix} must be exposed by the actual HTTP metrics server`); + return Number(metric[1]); + }; + await Promise.all(Array.from({length:6},(_,i)=>admitPacketToRam({...packet(),packetHash:i.toString(16).padStart(64,'0')}))); + const rows = await client.xrange(key,'-','+'); + await client.xreadgroup('GROUP',group,'loaded-metrics','COUNT',2,'STREAMS',key,'>'); + await client.hset(`${key}:committed`,rows[2]![0],String(Date.now()),rows[3]![0],String(Date.now())); + await refreshPendingStats(); + assert.equal(await gauge('queue_depth'),6); + assert.equal(await gauge('pending_depth'),4); + assert.equal(await gauge('retained_depth'),2); + assert.equal(await gauge('pel_depth'),2); + assert.ok(Number(await gauge('queue_bytes'))>0); + assert.equal(await gauge('queue_reserved_bytes'),rows.reduce((n,row)=>n+accountedRamEntryBytes(row[1][1]!),0)); + await client.xack(key,group,...rows.map(([id])=>id)); + await client.xdel(key,...rows.map(([id])=>id)); + await client.del(key,`${key}:committed`); + await reconcileRamQueueBytes(); + await refreshPendingStats(); + for(const suffix of ['queue_depth','pending_depth','retained_depth','pel_depth','queue_bytes','queue_reserved_bytes','oldest_pending_age_seconds','oldest_pel_age_seconds']) { + assert.equal(await gauge(suffix),0,`empty ${suffix} must not retain its loaded value`); + } +}); + test('real Redis F05/F09 admission replay and atomic byte reconciliation', { skip: localRedis ? false : 'TEST_REDIS_URL must point to a loopback fixture', }, async (t) => { diff --git a/backend/src/ram/runtime.integration.test.ts b/backend/src/ram/runtime.integration.test.ts new file mode 100644 index 0000000..ba0011a --- /dev/null +++ b/backend/src/ram/runtime.integration.test.ts @@ -0,0 +1,160 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import path from 'node:path'; +import os from 'node:os'; +import net from 'node:net'; +import { spawn, type ChildProcess } from 'node:child_process'; +import { once } from 'node:events'; +import test from 'node:test'; +import pg from 'pg'; +import { Redis } from 'ioredis'; +import { runMigrations } from '../db/migrations.js'; +import { configurePacketBatch } from '../db/packetBatch.js'; +import { admitPacketToRam, configureRamStore } from './liveStore.js'; +import { accountedRamEntryBytes } from './protocol.js'; + +const pgURL=process.env['TEST_INGEST_DATABASE_URL'], redisURL=process.env['TEST_REDIS_URL']; +const local=(url?:string)=>url&&['localhost','127.0.0.1'].includes(new URL(url).hostname); +const sleep=(ms:number)=>new Promise(resolve=>setTimeout(resolve,ms)); +async function until(check:()=>Promise|boolean, timeout=20_000, label='boundary') { + const end=Date.now()+timeout; + while(Date.now()((resolve,reject)=>server.close(error=>error?reject(error):resolve())); + return address.port; +} + +test('F15/F16 actual SIGTERM joins held PostgreSQL persistence, recovers identities and drains before durable rollback', { + skip:local(pgURL)&&local(redisURL)?false:'owned loopback PostgreSQL and Redis fixtures required', timeout:230_000, +},async t=>{ + const admin=new pg.Pool({connectionString:pgURL,max:1}); + const name=`meshcore_runtime_${process.pid}_${Date.now()}`; + await admin.query(`CREATE DATABASE "${name}"`); + const url=new URL(pgURL!);url.pathname='/'+name; + const pool=new pg.Pool({connectionString:url.toString(),max:3}); + const redis=new Redis(redisURL!,{lazyConnect:true,enableOfflineQueue:false}); + const key=`meshcore:acl:runtime:${name}`,group='owned-runtime'; + const files=fs.mkdtempSync(path.join(os.tmpdir(),'meshcore-runtime-')); + fs.writeFileSync(path.join(files,'acl'),'user backend\ntopic readwrite meshcore/#\n',{mode:0o600}); + const children:ChildProcess[]=[]; + const old=Object.fromEntries(['MQTT_INGEST_MODE','MQTT_RAM_STREAM_KEY','MQTT_RAM_QUEUE_MAX_BYTES'].map(name=>[name,process.env[name]])); + let holder:pg.PoolClient|undefined; + t.after(async()=>{ + for(const child of children)if(child.exitCode===null&&child.signalCode===null){child.kill('SIGKILL');await once(child,'exit').catch(()=>undefined);} + if(holder){await holder.query('ROLLBACK').catch(()=>undefined);holder.release();} + configureRamStore(null);configurePacketBatch(null); + const keys=await redis.keys(key+'*');if(keys.length)await redis.del(...keys); + await redis.quit();await pool.end(); + await admin.query(`DROP DATABASE IF EXISTS "${name}" WITH (FORCE)`);await admin.end(); + fs.rmSync(files,{recursive:true,force:true}); + for(const [name,value] of Object.entries(old)){if(value===undefined)delete process.env[name];else process.env[name]=value;} + }); + await redis.connect(); + await pool.query(fs.readFileSync(new URL('../db/schema/base.sql',import.meta.url),'utf8')); + await runMigrations(pool); + await pool.query(`CREATE FUNCTION hold_owned_packet() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN PERFORM pg_advisory_xact_lock(918204,17); RETURN NEW; END $$; + CREATE TRIGGER hold_owned_packet BEFORE INSERT ON packets FOR EACH ROW EXECUTE FUNCTION hold_owned_packet()`); + holder=await pool.connect();await holder.query('BEGIN');await holder.query('SELECT pg_advisory_xact_lock(918204,17)'); + process.env['MQTT_INGEST_MODE']='ram';process.env['MQTT_RAM_STREAM_KEY']=key; + process.env['MQTT_RAM_QUEUE_MAX_BYTES']=String(1048576); + configureRamStore(redis);configurePacketBatch((sql,params)=>pool.query(sql,params)); + const accepted=await admitPacketToRam({packetHash:'F1'.repeat(32),rxNodeId:'CD'.repeat(32),srcNodeId:'AB'.repeat(32), + topic:`meshcore/ZZZ/${'CD'.repeat(32)}/packets`,network:name,rawHex:'00',packetType:5,routeType:0,hopCount:2, + pathHashes:['ABCD','BEEF'],pathHashSizeBytes:2,payload:{_summary:'owned-runtime-fixture'}}); + const initial=await redis.xrange(key,'-','+'); + const bytes=accountedRamEntryBytes(initial[0]![1][1]!); + // Simulate a crash that lost only the companion counter before startup. + await redis.del(key+':bytes'); + const evidence=process.env['TEST_EVIDENCE_DIR']; + const launch=async(mode:'ram'|'durable'|'drain',deadline=7_000)=>{ + const httpPort=await port(),metricsPort=await port(); + const applicationName=`${name}-process-${children.length+1}`; + const env={...process.env,DATABASE_URL:url.toString(),OWNER_DATABASE_URL:url.toString(), + DATABASE_SKIP_SCHEMA_INIT:'1',DATABASE_APPLICATION_NAME:applicationName,DATABASE_STATEMENT_TIMEOUT_MS:'8000', + REDIS_URL:process.env['TEST_REDIS_RUNTIME_URL']??redisURL!,REDIS_PASSWORD:'',MQTT_INGEST_ENABLED:'0',WS_ENABLED:'0', + MQTT_INGEST_MODE:mode==='drain'?'ram':mode,MQTT_RAM_STREAM_KEY:key,MQTT_RAM_CONSUMER_GROUP:group, + MQTT_RAM_CONSUMER_NAME:'owned-runtime',MQTT_RAM_BATCH_SIZE:'1',MQTT_RAM_MAX_AGE_MS:'1000', + MQTT_RAM_QUEUE_MAX_BYTES:String(1048576),MQTT_RAM_DRAIN_DEADLINE_MS:'180000', + OWNER_AUTHORIZATION_MODE:'shadow',OWNER_ACL_MODE:'shadow', + MOSQUITTO_ACL_PATH:path.join(files,'acl'),OWNER_ACL_SHADOW_PATH:path.join(files,'shadow'), + PATH_RESOLVE_WORKERS:'1', + PORT:String(httpPort),METRICS_PORT:String(metricsPort),SHUTDOWN_DEADLINE_MS:String(deadline)}; + const compiled=process.env['TEST_BACKEND_ENTRY']??path.resolve('dist/index.js'); + assert.ok(fs.existsSync(compiled),'build the actual backend before running the runtime gate'); + const entry=mode==='drain'?path.join(path.dirname(compiled),'ram/drain.js'):compiled; + const child=spawn(process.execPath,[entry],{env,stdio:['ignore','pipe','pipe']}); + children.push(child);let output=''; + let log:fs.WriteStream|undefined; + if(evidence)log=fs.createWriteStream(path.join(evidence,`${name}-${mode}-${children.length}.log`)); + child.stdout!.on('data',chunk=>{output+=chunk;log?.write(chunk);}); + child.stderr!.on('data',chunk=>{output+=chunk;log?.write(chunk);}); + const done=new Promise<{code:number|null,signal:NodeJS.Signals|null}>(resolve=>child.once('exit',(code,signal)=>{log?.end();resolve({code,signal});})); + return {child,done,output:()=>output,httpPort,metricsPort,applicationName}; + }; + const counts=async()=>(await pool.query(`SELECT + (SELECT COUNT(*)::int FROM packets WHERE network=$1) packets, + (SELECT COUNT(*)::int FROM ingest_observation_ledger WHERE network=$1) ledger, + (SELECT COUNT(*)::int FROM packet_paths WHERE network=$1) paths, + (SELECT COALESCE(SUM(packet_count),0)::int FROM packet_hourly_stats WHERE network=$1) hourly, + (SELECT COUNT(*)::int FROM packet_daily_stats WHERE network=$1) daily, + (SELECT COUNT(*)::int FROM observer_region_packet_sightings WHERE network=$1) regions`,[name])).rows[0]; + const blocked=async(applicationName:string)=>Number((await pool.query(`SELECT COUNT(*) n FROM pg_stat_activity + WHERE datname=$1 AND application_name=$2 AND wait_event='advisory'`,[name,applicationName])).rows[0].n)>0; + const first=await launch('ram',2_000); + await until(()=>blocked(first.applicationName),25_000,'held real INSERT'); + assert.equal(Number(await redis.get(key+':bytes')),bytes,'writer startup reconciles the lost reservation'); + assert.equal((await redis.xpending(key,group))[0],1); + const loadedMetrics=await (await fetch(`http://127.0.0.1:${first.metricsPort}/metrics`)).text(); + for(const [suffix,value] of [['pending_depth',1],['retained_depth',0],['pel_depth',1],['queue_reserved_bytes',bytes]]) { + assert.match(loadedMetrics,new RegExp(`^meshcore_ram_ingest_${suffix} ${value}$`,'m')); + } + assert.deepEqual(await counts(),{packets:0,ledger:0,paths:0,hourly:0,daily:0,regions:0}); + const signalAt=Date.now();first.child.kill('SIGTERM'); + const failed=await Promise.race([first.done,sleep(6_000).then(()=>{throw new Error('SIGTERM exceeded the shared deadline');})]); + assert.equal(failed.code,1);assert.equal(failed.signal,null); + assert.ok(Date.now()-signalAt<5_000); + assert.doesNotMatch(first.output(),/graceful shutdown complete/); + assert.match(first.output(),/graceful shutdown (?:failed|timed out)|RAM_DRAIN_TIMEOUT/); + assert.deepEqual(await counts(),{packets:0,ledger:0,paths:0,hourly:0,daily:0,regions:0}); + assert.equal((await redis.xpending(key,group))[0],1); + assert.equal(Number(await redis.get(key+':bytes')),bytes); + await until(async()=>!await blocked(first.applicationName),15_000,'aborted first database connection'); + const recovered=await launch('ram'); + await until(()=>blocked(recovered.applicationName),25_000,'held replay INSERT'); + recovered.child.kill('SIGTERM');await sleep(250); + assert.equal(recovered.child.exitCode,null,'successful drain waits for the actual write'); + assert.deepEqual(await counts(),{packets:0,ledger:0,paths:0,hourly:0,daily:0,regions:0}); + await holder.query('COMMIT');holder.release();holder=undefined; + const success=await recovered.done; + assert.equal(success.code,0,recovered.output());assert.match(recovered.output(),/graceful shutdown complete/); + const expected={packets:1,ledger:1,paths:1,hourly:1,daily:1,regions:1}; + assert.deepEqual(await counts(),expected); + assert.equal((await pool.query('SELECT observation_id::text FROM packets WHERE network=$1',[name])).rows[0].observation_id,accepted.observationId); + assert.equal((await redis.xpending(key,group))[0],0); + assert.equal(await redis.xlen(key),1,'committed overlap remains reserved until expiry'); + assert.equal(Number(await redis.get(key+':bytes')),bytes); + // Quarantine blocks rollback even when the source is empty. After review, + // only its separate operator may clear it and rerun the production drainer. + await redis.xadd(key+':quarantine','*','q','owned-operator-review'); + const drain=await launch('drain'); + const refused=await drain.done; + assert.equal(refused.code,1,drain.output());assert.match(drain.output(),/quarantined entries require review/); + assert.doesNotMatch(drain.output(),/RAM_DRAIN_COMPLETE/); + assert.equal(await redis.xlen(key),0);assert.equal((await redis.xpending(key,group))[0],0); + assert.equal(Number(await redis.get(key+':bytes')),0); + await redis.del(key+':quarantine'); + const drained=await launch('drain'); + assert.equal((await drained.done).code,0,drained.output());assert.match(drained.output(),/RAM_DRAIN_COMPLETE/); + const durable=await launch('durable'); + await until(()=>durable.output().includes('[app] listening'),25_000,'durable rollback listener'); + assert.doesNotMatch(durable.output(),/\[ram\] writer started/); + const scrape=await fetch(`http://127.0.0.1:${durable.metricsPort}/metrics`);assert.equal(scrape.status,200); + const metrics=await scrape.text();assert.match(metrics,/meshcore_mqtt_post_persist_inflight/); + durable.child.kill('SIGTERM');assert.equal((await durable.done).code,0,durable.output()); + assert.deepEqual(await counts(),expected,'rollback preserves every observation and side effect exactly once'); +}); diff --git a/backend/src/ram/writer.test.ts b/backend/src/ram/writer.test.ts index ebf47c3..4d76c45 100644 --- a/backend/src/ram/writer.test.ts +++ b/backend/src/ram/writer.test.ts @@ -65,6 +65,11 @@ class FakeRedis { } quarantined: string[] = []; + async xtrim(_key: string, _strategy: string, threshold: number): Promise { + const removed = Math.max(0,this.quarantined.length-threshold); + this.quarantined=this.quarantined.slice(-threshold); + return removed; + } async xadd(_key: string, idOrCommand: string, ...args: (string | number)[]): Promise { if (idOrCommand === 'MAXLEN') { if (this.quarantineFails) throw new Error('quarantine unavailable'); @@ -613,6 +618,19 @@ test('unparseable stream values are quarantined, acknowledged and deleted', asyn assert.equal(record['raw'], 'not-json'); }); +test('F09 quarantine caps encoded evidence bytes including JSON escaping', async () => { + const fake = new FakeRedis(); + fake.entries.push({id:'1700000600000-0',fields:['e','\u0000'.repeat(16_000)]}); + const {writer} = makeWriter(fake,()=>1_700_000_600_000,async()=>undefined,{batchSize:1}); + await writer.start(); + await writer.tickOnce(); + assert.equal(fake.quarantined.length,1); + assert.ok(Buffer.byteLength(fake.quarantined[0]!)<=16_384,'encoded evidence exceeds the reserved per-record bound'); + const value=JSON.parse(fake.quarantined[0]!); + assert.equal(value.bytes,16_000); + assert.equal(value.sha256.length,64); +}); + test('drain flushes everything that remains', async () => { const fake = new FakeRedis(); const now = () => 1_700_000_700_000; diff --git a/backend/src/ram/writer.ts b/backend/src/ram/writer.ts index 5f3d162..1dbf59c 100644 --- a/backend/src/ram/writer.ts +++ b/backend/src/ram/writer.ts @@ -32,6 +32,8 @@ import { getRamCommittedKey, getRamQuarantineKey, getRamQueueBytesKey, + RAM_QUARANTINE_MAX_ENTRIES, + RAM_QUARANTINE_ENTRY_MAX_BYTES, type RamPendingEntry, } from './protocol.js'; import { @@ -62,6 +64,7 @@ export interface RamWriterRedis { xack(key: string, group: string, ...ids: string[]): Promise; xdel(key: string, ...ids: string[]): Promise; xadd(key: string, ...args: (string | number)[]): Promise; + xtrim(key: string, strategy: string, threshold: number): Promise; xautoclaim( key: string, group: string, @@ -130,8 +133,8 @@ const SWEEP_INTERVAL_MS = 10_000; const SWEEP_SCAN_LIMIT = 2_000; const PEL_LIST_LIMIT = 5_000; /** Bounded recovery evidence for undecodable entries. */ -const QUARANTINE_MAX_LEN = 500; -const QUARANTINE_VALUE_CAP = 65_536; +const QUARANTINE_MAX_LEN = RAM_QUARANTINE_MAX_ENTRIES; +const QUARANTINE_VALUE_CAP = RAM_QUARANTINE_ENTRY_MAX_BYTES; const FAILURE_BACKOFF_START_MS = 5_000; const FAILURE_BACKOFF_MAX_MS = 30_000; const RECLAIM_INTERVAL_MS = 5_000; @@ -247,6 +250,12 @@ export function createRamWriter(deps: RamWriterDeps): RamWriter { async function quarantineMalformed(id: string, value: string, reason = 'undecodable', deliveryKey?: string): Promise { const bytes = Buffer.byteLength(value); const digest = createHash('sha256').update(value).digest('hex'); + const record = {src:id,reason,at:deps.now(),sha256:digest,bytes,raw:bytes<=QUARANTINE_VALUE_CAP ? value : ''}; + let encoded = JSON.stringify(record); + if (Buffer.byteLength(encoded)>QUARANTINE_VALUE_CAP) { + record.raw = ''; + encoded = JSON.stringify(record); + } try { await deps.redis.xadd( getRamQuarantineKey(), @@ -255,15 +264,12 @@ export function createRamWriter(deps: RamWriterDeps): RamWriter { QUARANTINE_MAX_LEN, '*', 'q', - JSON.stringify({ - src: id, - reason, - at: deps.now(), - sha256: digest, - bytes, - raw: bytes <= QUARANTINE_VALUE_CAP ? value : '', - }), + encoded, ); + // Approximate XADD trimming is followed by an exact cap. Overshoot is + // transient and one encoded record per concurrent writer; source + // cleanup happens only after both commands succeed. + await deps.redis.xtrim(getRamQuarantineKey(), 'MAXLEN', QUARANTINE_MAX_LEN); ramQuarantinedTotal.inc(); deps.log(`[ram] quarantined ${reason} stream entry id=${id} bytes=${bytes} sha256=${digest}`); } catch (error) { diff --git a/backend/src/ram/writerRuntime.ts b/backend/src/ram/writerRuntime.ts index 5bb0ca2..6d5830f 100644 --- a/backend/src/ram/writerRuntime.ts +++ b/backend/src/ram/writerRuntime.ts @@ -21,6 +21,7 @@ import { } from './protocol.js'; import { createRamWriter, type RamWriterRedis } from './writer.js'; import { joinAndDrainRamWriter } from './drainLifecycle.js'; +import { validateRamRedisHeadroom } from './memory.js'; let singleton: ReturnType | null = null; let writerLoop: Promise | null = null; @@ -47,6 +48,8 @@ export async function startRamWriter(): Promise { if (singleton) return; await reconcileRamQueueBytes(); const redis = productionRedis(); + try { validateRamRedisHeadroom(await redis.info('memory')); } + catch (error) { redis.disconnect(); throw error; } const writer = createRamWriter({ redis: redis as unknown as RamWriterRedis, persist: async (entries, idempotent) => { diff --git a/docs/ram-first-ingestion.md b/docs/ram-first-ingestion.md index b33da82..a37c701 100644 --- a/docs/ram-first-ingestion.md +++ b/docs/ram-first-ingestion.md @@ -4,8 +4,8 @@ RAM mode is opt-in. Durable mode remains the default and acknowledges QoS 1 deliveries after PostgreSQL persistence. RAM mode acknowledges after a bounded Redis Stream admission; the writer commits each observation later. A Redis loss before PostgreSQL commit can therefore lose already acknowledged telemetry. -Keep the Redis instance dedicated to this queue, persistent (AOF), and on a -`noeviction` policy. Do not expose the raw stream to public API credentials; +Keep the Redis queue private, persistent (AOF), and on a bounded +`noeviction` instance. Do not expose the raw stream to public API credentials; only the backend and a separate quarantine operator should have raw stream access. Public companions use the credentialed filtered HTTP source below. @@ -60,9 +60,16 @@ Set `MQTT_INGEST_MODE=ram` in the Compose `.env`. The standard configuration before a maintenance change: ```sh -docker compose config --format json | jq '.services.backend.environment | with_entries(select(.key | startswith("MQTT_RAM_") or . == "MQTT_INGEST_MODE"))' +export COMPOSE_FILE=docker-compose.yml:docker-compose.live.yml +docker compose config --format json | jq '.services.backend.environment | with_entries(select(.key | (startswith("MQTT_RAM_") and . != "MQTT_RAM_PUBLIC_READ_TOKEN") or . == "MQTT_INGEST_MODE"))' ``` +Keep the deployment's established Compose file selection. The example uses +the split-host live overlay; the base file alone does not define its external +data-plane services. CI renders the base plus `docker-compose.ci.yml` as its +self-contained test stack. The drain commands below use the same selected +deployment files. Neither render command changes a running service. + Use a canary or isolated fixture first. Monitor `/readyz`, `ram_pending_depth`, `ram_retained_depth`, `ram_pel_depth`, the oldest pending age, admission refusals, commit failures, PostgreSQL disk/WAL and Redis @@ -72,9 +79,32 @@ when current privacy authority cannot be established. Queue age is a healthy writer flush trigger, not a completion guarantee during an outage. The queue has a hard count bound and conservative byte reservation, including -retained entries and ambiguous admission delivery records. Size Redis -`maxmemory` above the queue budget plus AOF, PEL, quarantine and other key -headroom. Do not use Redis eviction to make room for acknowledged packets. +retained entries and ambiguous admission delivery records. RAM startup and +each new admission require a finite Redis `maxmemory`, `noeviction`, and at +least **32 MiB** beyond the configured queue budget. Admission also checks +current `used_memory` atomically with reservation, leaving that 32 MiB free. +Unsafe startup is retried before MQTT admission starts; unsafe new admission +reports `redis_headroom` and receives no PUBACK. A replay returns its original +stored identity even if capacity is now unavailable. Quarantine is exactly +capped at 500 records and 16 KiB per encoded record, including JSON escaping; +oversized raw values retain their byte length and SHA-256 instead of content. +The separate headroom covers quarantine, PEL, commit metadata and allocator +slack. Application reservations are not a measurement of process RSS, AOF +buffers, other keys or the container's physical memory. Size the container and +storage above those additional costs and measure them in the completed window. +Do not use Redis eviction to make room for acknowledged packets. + +For a dedicated stream principal, the required Redis permissions are +`+@connection +@stream +@string +@hash +del +eval +evalsha +info +memory`, +restricted to the configured stream and its `:*` bookkeeping keys. Add only +the normal backend's separately required queue/cache permissions if sharing +its existing Redis instance. The quarantine operator needs `XRANGE`, `XLEN`, +`DEL` and `MEMORY` on `:quarantine`, without access to the raw stream +or delivery ledger. Disable the default user or remove its unauthenticated +access; keep credentials in private ACL files. The public CoreScope companion +uses only the filtered HTTP token and its restricted PostgreSQL reader, with +no raw Redis credential. These principals are exercised against real Redis +by `npm run test:ram-operational`; the test owns its container and credentials. ## Drain and return to durable mode diff --git a/scripts/test-ram-redis-acl.mjs b/scripts/test-ram-redis-acl.mjs new file mode 100644 index 0000000..33fd672 --- /dev/null +++ b/scripts/test-ram-redis-acl.mjs @@ -0,0 +1,71 @@ +#!/usr/bin/env node +// Own one bounded Redis instance and four principals for the RAM operational +// gates. Credentials stay in memory/hashed ACL files and never enter argv. +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import {fileURLToPath} from 'node:url'; +import {randomBytes,createHash} from 'node:crypto'; +import {spawn,spawnSync} from 'node:child_process'; +import {Redis} from '../backend/node_modules/ioredis/built/index.js'; + +const backend=path.resolve(path.dirname(fileURLToPath(import.meta.url)),'../backend'); +const database=process.env.TEST_INGEST_DATABASE_URL; +if(!database||!['127.0.0.1','localhost'].includes(new URL(database).hostname)) { + throw new Error('TEST_INGEST_DATABASE_URL must be an owned loopback fixture; skipped runtime gates cannot pass'); +} +const root=process.env.TEST_RAM_ACL_FIXTURE_ROOT??os.tmpdir(); +fs.mkdirSync(root,{recursive:true}); +const fixture=fs.mkdtempSync(path.join(root,'ram-acl-')); +fs.chmodSync(fixture,0o700); +fs.mkdirSync(path.join(fixture,'data'),{mode:0o700}); +const name=`ram-acl-test-${process.pid}-${Date.now()}`; +const image=process.env.TEST_RAM_ACL_IMAGE??'redis@sha256:8b81dd37ff027bec4e516d41acfbe9fe2460070dc6d4a4570a2ac5b9d59df065'; +const passwords=Object.fromEntries(['admin','writer','public','operator'].map(user=>[user,randomBytes(32).toString('hex')])); +const auth=user=>'#'+createHash('sha256').update(passwords[user]).digest('hex'); +const acl=`user default off +user admin on ${auth('admin')} ~* &* +@all +user writer on ${auth('writer')} ~meshcore:acl:* &meshcore:live +@connection +@stream +@string +@hash +del +eval +evalsha +info +memory +user public on ${auth('public')} +ping +user operator on ${auth('operator')} ~meshcore:acl:*:quarantine +@connection +xrange +xlen +del +memory +`; +fs.writeFileSync(path.join(fixture,'users.acl'),acl,{mode:0o600}); +const docker=(args)=>{ + const result=spawnSync('docker',args,{encoding:'utf8'}); + if(result.status!==0)throw new Error(`Owned Docker fixture failed: ${result.stderr.trim()}`); + return result.stdout.trim(); +}; +let created=false; +try{ + const user=os.userInfo(); + docker(['run','-d','--rm','--name',name,'--memory','80m','--cpus','1','--user',`${user.uid}:${user.gid}`, + '-p','127.0.0.1::6379','-v',`${fixture}:/fixture:ro`,'-v',`${fixture}/data:/data`, + image,'redis-server','--dir','/data','--maxmemory','48mb','--maxmemory-policy','noeviction', + '--appendonly','yes','--appendfsync','everysec','--aclfile','/fixture/users.acl']); + created=true; + const port=docker(['inspect',name,'--format','{{(index (index .NetworkSettings.Ports "6379/tcp") 0).HostPort}}']); + const url=user=>`redis://${user}:${passwords[user]}@127.0.0.1:${port}/0`; + const admin=new Redis(url('admin'),{enableReadyCheck:false,maxRetriesPerRequest:1}); + admin.on('error',()=>undefined); + await admin.ping();await admin.quit(); + const env={...process.env,TEST_REDIS_ACL_URL:url('writer'),TEST_REDIS_ADMIN_URL:url('admin'), + TEST_REDIS_PUBLIC_URL:url('public'),TEST_REDIS_OPERATOR_URL:url('operator'), + TEST_REDIS_URL:url('admin'),TEST_REDIS_RUNTIME_URL:url('writer')}; + console.log('Owned ACL fixture: Redis maxmemory=48MiB, container limit=80MiB, noeviction, AOF, four isolated principals'); + for(const [file,expected] of [['src/ram/operational.integration.test.ts',4],['src/ram/runtime.integration.test.ts',1]]){ + let transcript=''; + const code=await new Promise((resolve,reject)=>{ + const child=spawn(process.execPath,['--import','tsx','--test',file],{cwd:backend,env,stdio:['ignore','pipe','pipe']}); + child.stdout.on('data',chunk=>{transcript+=chunk;process.stdout.write(chunk);}); + child.stderr.pipe(process.stderr); + child.once('error',reject);child.once('exit',(code,signal)=>resolve(signal?1:code)); + }); + const complete=code===0&&new RegExp(`^# tests ${expected}$`,'m').test(transcript) + &&new RegExp(`^# pass ${expected}$`,'m').test(transcript) + &&/^# skipped 0$/m.test(transcript)&&/^# fail 0$/m.test(transcript); + if(!complete)throw new Error(`Operational gate failed or incomplete: ${file}`); + } +}finally{ + if(created)docker(['stop','--time','2',name]); + fs.rmSync(fixture,{recursive:true,force:true}); +}