diff --git a/lib/persist/conformance.sh b/lib/persist/conformance.sh index 46aededa..1ce9c1e3 100755 --- a/lib/persist/conformance.sh +++ b/lib/persist/conformance.sh @@ -13,7 +13,7 @@ if [ ! -x "$SX_SERVER" ]; then exit 1 fi -SUITES=(event log kv project subscribe concurrency snapshot compaction durable blob recovery) +SUITES=(event log kv project subscribe concurrency snapshot compaction durable blob view recovery) OUT_JSON="lib/persist/scoreboard.json" OUT_MD="lib/persist/scoreboard.md" @@ -37,6 +37,7 @@ run_suite() { (load "lib/persist/compaction.sx") (load "lib/persist/durable.sx") (load "lib/persist/blob.sx") +(load "lib/persist/view.sx") (load "lib/persist/subscribe.sx") (load "lib/persist/api.sx") (epoch 2) diff --git a/lib/persist/scoreboard.json b/lib/persist/scoreboard.json index c5f2f969..4c3650f0 100644 --- a/lib/persist/scoreboard.json +++ b/lib/persist/scoreboard.json @@ -10,9 +10,10 @@ "compaction": {"pass": 11, "fail": 0}, "durable": {"pass": 15, "fail": 0}, "blob": {"pass": 14, "fail": 0}, + "view": {"pass": 11, "fail": 0}, "recovery": {"pass": 6, "fail": 0} }, - "total_pass": 111, + "total_pass": 122, "total_fail": 0, - "total": 111 + "total": 122 } diff --git a/lib/persist/scoreboard.md b/lib/persist/scoreboard.md index d7f95884..41ad6a35 100644 --- a/lib/persist/scoreboard.md +++ b/lib/persist/scoreboard.md @@ -14,5 +14,6 @@ _Generated by `lib/persist/conformance.sh`_ | compaction | 11 | 0 | 11 | | durable | 15 | 0 | 15 | | blob | 14 | 0 | 14 | +| view | 11 | 0 | 11 | | recovery | 6 | 0 | 6 | -| **Total** | **111** | **0** | **111** | +| **Total** | **122** | **0** | **122** | diff --git a/lib/persist/tests/view.sx b/lib/persist/tests/view.sx new file mode 100644 index 00000000..cbe24c81 --- /dev/null +++ b/lib/persist/tests/view.sx @@ -0,0 +1,105 @@ +; Extension — materialized views: stay current on write, read O(1) via peek. + +(define vw-count (fn (acc e) (+ acc 1))) +(define vw (persist/view "order-count" "orders" vw-count 0)) + +(persist-test "view-name" (persist/view-name vw) "order-count") +(persist-test "view-stream" (persist/view-stream vw) "orders") +(persist-test + "view-value folds the stream" + (let + ((b (persist/open))) + (begin + (persist/append b "orders" "x" 0 {}) + (persist/append b "orders" "x" 0 {}) + (persist/view-value b vw))) + 2) +(persist-test + "view-refresh persists a snapshot that peek then reads" + (let + ((b (persist/open))) + (begin + (persist/append b "orders" "x" 0 {}) + (persist/view-refresh b vw) + (persist/view-peek b vw))) + 1) +(persist-test + "peek lags an un-refreshed tail" + (let + ((b (persist/open))) + (begin + (persist/append b "orders" "x" 0 {}) + (persist/view-refresh b vw) + (persist/append b "orders" "x" 0 {}) + (persist/view-peek b vw))) + 1) +(persist-test + "view-value sees the whole stream even after a stale snapshot" + (let + ((b (persist/open))) + (begin + (persist/append b "orders" "x" 0 {}) + (persist/view-refresh b vw) + (persist/append b "orders" "x" 0 {}) + (persist/view-value b vw))) + 2) +(persist-test + "attached view stays current on publish — peek needs no manual refresh" + (let + ((b (persist/open))) + (let + ((h (persist/view-attach (persist/hub b) vw))) + (begin + (persist/publish h "orders" "x" 0 {}) + (persist/publish h "orders" "x" 0 {}) + (persist/publish h "orders" "x" 0 {}) + (persist/view-peek b vw)))) + 3) +(persist-test + "attached view advances the snapshot seq incrementally" + (let + ((b (persist/open))) + (let + ((h (persist/view-attach (persist/hub b) vw))) + (begin + (persist/publish h "orders" "x" 0 {}) + (persist/publish h "orders" "x" 0 {}) + (persist/project-seq + (persist/snapshot-load b "order-count" 0))))) + 2) +(persist-test + "attach only reacts to its own stream" + (let + ((b (persist/open))) + (let + ((h (persist/view-attach (persist/hub b) vw))) + (begin + (persist/publish h "other" "x" 0 {}) + (persist/view-peek b vw)))) + 0) +(persist-test + "materialized view works on the durable backend" + (let + ((db (persist/mock-durable (persist/mem-backend)))) + (let + ((h (persist/view-attach (persist/hub db) vw))) + (begin + (persist/publish h "orders" "x" 0 {}) + (persist/publish h "orders" "x" 0 {}) + (persist/view-peek db vw)))) + 2) +(persist-test + "view sum over event data" + (let + ((b (persist/open)) + (sumv + (persist/view + "rev" + "sales" + (fn (acc e) (+ acc (get (persist/event-data e) :amt))) + 0))) + (begin + (persist/append b "sales" "sale" 0 {:amt 10}) + (persist/append b "sales" "sale" 1 {:amt 25}) + (persist/view-value b sumv))) + 35) diff --git a/lib/persist/view.sx b/lib/persist/view.sx new file mode 100644 index 00000000..0dbf4c85 --- /dev/null +++ b/lib/persist/view.sx @@ -0,0 +1,49 @@ +; persist/view — a materialized view: the consumer-facing read model. It bundles +; a stream, a fold (step + seed) and a snapshot name. Attached to a hub it +; refreshes incrementally on every publish, so the materialized value stays +; current on write and reads are O(1) snapshot loads (persist/view-peek) instead +; of a full fold. This is what feed indices, mod audit rollups, search counters, +; etc. sit on. Requires: lib/persist/snapshot.sx, lib/persist/subscribe.sx. + +(define persist/view (fn (name stream step seed) {:name name :step step :stream stream :seed seed})) +(define persist/view-name (fn (v) (get v :name))) +(define persist/view-stream (fn (v) (get v :stream))) + +; bring the view's snapshot up to date with the log tail; returns the state +(define + persist/view-refresh + (fn + (b v) + (persist/checkpoint + b + (get v :stream) + (get v :name) + (get v :step) + (get v :seed)))) + +; current materialized value — refreshes first, so never stale +(define + persist/view-value + (fn (b v) (persist/project-value (persist/view-refresh b v)))) + +; O(1) read of the last persisted snapshot value WITHOUT folding the tail. Equal +; to view-value when the view is attached (kept current on every publish); +; otherwise may lag the log by the un-refreshed tail. +(define + persist/view-peek + (fn + (b v) + (persist/project-value + (persist/snapshot-load b (get v :name) (get v :seed))))) + +; attach to a hub: refresh the view on every publish to its stream +(define + persist/view-attach + (fn + (h v) + (begin + (persist/subscribe + h + (persist/view-stream v) + (fn (bk s e) (persist/view-refresh bk v))) + h))) diff --git a/plans/persist-on-sx.md b/plans/persist-on-sx.md index ebbf1e01..8ebb7427 100644 --- a/plans/persist-on-sx.md +++ b/plans/persist-on-sx.md @@ -42,7 +42,7 @@ read models (feeds, indices, audit logs) update incrementally. ## Status (rolling) -`bash lib/persist/conformance.sh` → **111/111** (Phases 1–4 complete) +`bash lib/persist/conformance.sh` → **122/122** (Phases 1–4 complete + extensions) ## Ground rules @@ -144,11 +144,22 @@ suspends and the host resumes it transparently, so the facet code is unaware. Tests prove this by routing the identical request shapes through `persist/serve` over an in-process disk (the mock-IO harness). +## Extensions (post-roadmap) +- [x] `view.sx` — materialized views: bundle stream + fold + snapshot name; + `view-attach` keeps the snapshot current on every publish so `view-peek` is an + O(1) read. The consumer-facing read-model abstraction (feed indices, audit + rollups, search counters). + ## Consumers (post-foundation, not in scope here) feed/-log, flow store, mod/audit, search index, acl grants, identity sessions all become `persist` log or kv. Track each migration in that subsystem's plan. ## Progress log +- **Ext: materialized views (122/122).** `view.sx` — `persist/view` bundles + stream + step + seed + snapshot name; `view-attach` subscribes it to a hub so + every publish refreshes the snapshot incrementally; `view-peek` is then an + O(1) current read (no fold), `view-value` always folds the tail so it's never + stale. 11 tests incl. on durable backend + a sum-over-data view. - **Phase 4c+4d (111/111) — Phase 4 complete, roadmap done.** `recovery.sx` — a 6-test crash/restart integration: an order ledger (event log + subscription kv read model + snapshot + compaction + invoice blob ref) over the durable