From 2439933cf2e3fbcda812874ec3829770e0959897 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Tue, 3 Nov 2026 19:40:48 +0000 Subject: [PATCH] feat(chord): live queries over an object's own tables (self.query) self.query(table, q, limit) runs before the object goes live and again in the commit step of every commit that touched the table, publishing only the diff; a reload swaps the object. The tutorial and 59-chat use it. --- packages/chord/CHANGELOG.md | 1 + packages/chord/docs/benchmarks.md | 22 ++ packages/chord/docs/decisions.md | 24 +- packages/chord/docs/design.md | 30 +- .../chord/docs/guide-plugins-hosts/book.md | 19 +- .../chord/docs/guide-plugins-hosts/index.html | 14 +- packages/chord/docs/plugin-authoring.md | 26 +- packages/chord/docs/slices.md | 13 +- packages/chord/docs/tutorial/book.md | 83 +++-- packages/chord/docs/tutorial/index.html | 84 +++-- packages/chord/src/documents/transaction.ts | 14 + packages/chord/src/host/README.md | 29 +- packages/chord/src/host/attempt.ts | 68 +++- packages/chord/src/host/host.ts | 5 + packages/chord/src/host/object.ts | 9 + packages/chord/src/host/staging.ts | 12 + packages/chord/src/store/README.md | 10 +- packages/chord/src/store/live-query.ts | 301 ++++++++++++++++ packages/chord/src/store/table-handle.ts | 8 +- packages/chord/test/bench/live-query.ts | 92 +++++ packages/chord/test/bench/store.ts | 2 + .../examples/59-chat/plugins/chat/worker.ts | 34 +- .../61-tutorial/plugins/lists/worker.ts | 35 +- packages/chord/test/store/live-query.test.ts | 335 ++++++++++++++++++ packages/chord/test/support.ts | 10 +- 25 files changed, 1113 insertions(+), 167 deletions(-) create mode 100644 packages/chord/src/store/live-query.ts create mode 100644 packages/chord/test/bench/live-query.ts create mode 100644 packages/chord/test/store/live-query.test.ts diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 5e17d46e6..de47a4729 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -18,6 +18,7 @@ - Chord v2's root exports `invalid(reason, message, details?)` and the new `forbidden(message, details?)`, the `ChordError`s a plugin throws for arguments or params it refuses (`invalid/args`, `invalid/params`) and for a principal it refuses (`forbidden`); an app's API re-exports them, and the caller gets the same `{ code, reason }` as from Chord's own checks. - Chord v2's React bindings gain `usePageState(key, initial)`, `useState`'s pair with the value kept by the page's host under `key` instead of the component: a component of a reloaded web plugin is a new function and remounts (its `useState` starts over), and page state is what survives (a half-typed message, a scroll position); any value, never on the wire, gone with the page. `TuiScope` gains `pageState(key, initial)` on the same per-host store, its changes calling the scope's `changed`. - Chord v2 reports mistakes that waited silently through `onError` (decision "Mistakes that wait are reported at once"): `unbound_long { waitingFor, error, since }`, a request waiting unbound (a type or registry not served here, not at its version, a registry whose first `compose` threw) once after 5 s, on one timer per host (it keeps waiting until its `readyTimeout`); `registry_undeclared { registry, plugins }`, a registry with contributions and no provider at a publication (`load`, a reload's commit) or when a host-lifetime object's contributions enter, naming the contributing plugins, again only once they change; `transaction_cancelled { object, error }`, a document change or transaction that did not commit because the context it was started with ended first (a root whose `fn` returned, a caller that cancelled, its object's end), besides rejecting. +- Chord v2 gains live queries: `self.query(table, q, limit)` in the factory of a table's owner is a watchable of the query's rows (at most `limit`, in its order), usable as a state member and a derive source. Chord runs it before the object goes live and again inside the commit step of every commit of the object that writes the table, publishing only the diff in that commit's batch; a commit writing no table with a live query runs nothing (`chord.query.runs` counts the runs). Only the object's own tables; the limit is required (`invalid/args` at declaration). A run that fails (a row the code cannot read) is a failed member, reported once as the new `HostError` `query_failed { object, table, error }`, and heals at the next good run. A live query is no native, so an object showing table rows live swaps on reload instead of restarting, as reading them in `self.open` made it. Each touching commit runs the query once (100 rows: about 60 µs, 80 µs watched): not for tables written hundreds of times a second. ### Changed diff --git a/packages/chord/docs/benchmarks.md b/packages/chord/docs/benchmarks.md index 1a18a3bb2..4f5003322 100644 --- a/packages/chord/docs/benchmarks.md +++ b/packages/chord/docs/benchmarks.md @@ -972,3 +972,25 @@ Fix round 1: the bench itself now builds with the app's API and the kinds (its p L5), and every package file's real path is checked against the plugin's `node_modules`. CPU 8, three runs: cold build 261.1-270.2 ms, rebuild 13.6-14.4 ms (the first two runs as before, the third with a slower cold build on a busy machine). + +## Live queries (2026-11-03) + +`test/bench/live-query.ts`, Node 24.4.1, CPU 10, a SQLite `:memory:` store, medians and p95 of seven bursts of 2,000 +commits, three runs (equal to 0.5 µs). Each commit appends one row of a chat table (an int key, a short author, an +80-character text, a number) through a method call; the live query is the newest 100 rows, so every commit shifts the +window by one row. + +| One-row commit through a call | Median | p95 | +| --- | --- | --- | +| no live query | 8.5-8.7 µs | 10.3-10.6 µs | +| a live query of 100 rows, unwatched | 66.9-67.3 µs | 71.5-72.0 µs | +| a live query of 100 rows, watched (one insert and one removal published) | 86.3-86.5 µs | 94.4-95.1 µs | +| a commit of another table beside the watched query | 7.4-7.5 µs | 7.9-8.0 µs | + +A touching commit costs about 58 µs more for the run (the `SELECT` of 100 rows in the step, most of it node:sqlite +building the row objects; rows whose stored columns did not change are not decoded again but reused) and about 19 µs +more for the diff when watched. A commit of another table costs nothing extra (one map lookup per written table). +The diff depends on the reuse: with every row decoded anew, `diffRevisions` compared 100 fresh rows against the +previous 100 by content and the watched commit took 885 µs; with unchanged rows the same objects it compares them by +identity. "There may be dragons": at 100 rows a table written more than a few thousand times a second spends most of +its time in the query. diff --git a/packages/chord/docs/decisions.md b/packages/chord/docs/decisions.md index f8e5dabb7..919814eb6 100644 --- a/packages/chord/docs/decisions.md +++ b/packages/chord/docs/decisions.md @@ -2476,17 +2476,19 @@ validation and copies the v2 kernel added around Delta. or SWC transform in every web build, fragile when hooks change). Open: a live view of table rows (the tutorial's and 59-chat's `self.open` read at activation, which makes every reload a restart): to explain first. - **Live queries over an object's own tables (owner, 2026-11-03: "if self.query is easy enough, and we document it as - 'there may be dragons' then i guess it's ok"; resolves the tutorial's gap 2):** `self.query(table, q)` in a factory - returns a `Watchable` of the query's rows, usable as a state member and a derive source. Chord runs the query before - the object goes live (like loading its documents: no `self.open`, so a reload swaps the object), and again in the - publication step of every commit of the object that touched that table, diffing the new rows against the previous - ones (ops to watchers and holds, as for any member). Only the object's own tables (only it writes them, so its - commits are the exact invalidation); the query is fixed for the attempt (params decide it); it must have a `limit`. - Cost: one query per touching commit, bounded by the limit; documented with a warning ("there may be dragons": not - for tables written hundreds of times a second). Incremental maintenance of the result (an ordered, limited query - with row-level changes kept in memory, re-running only when a removal empties the window or a bulk write hides - which rows changed) is carry-over for when a case needs it. Rejected: a window document written beside the table - (rows stored twice; app code mirroring every update and delete). + 'there may be dragons' then i guess it's ok"; resolves the tutorial's gap 2):** `self.query(table, q, limit)` in a + factory returns a `Watchable` of the query's rows, usable as a state member and a derive source. Chord runs the + query before the object goes live (like loading its documents: no `self.open`, so a reload swaps the object), and + again in the publication step of every commit of the object that touched that table, diffing the new rows against + the previous ones (ops to watchers and holds, as for any member). Only the object's own tables (only it writes them, + so its commits are the exact invalidation); the query is fixed for the attempt (params decide it); a limit is + required (the third argument, as `page` takes it: the query language keeps the limit an argument of the runner; + refused `invalid/args` at declaration; build, 2026-11-03). Cost: one query per touching commit, bounded by the + limit; documented with a warning ("there may be dragons": not for tables written hundreds of times a second). + Incremental maintenance of the result (an ordered, limited query with row-level changes kept in memory, re-running + only when a removal empties the window or a bulk write hides which rows changed) is carry-over for when a case + needs it. Rejected: a window document written beside the table (rows stored twice; app code mirroring every update + and delete). - **A plugin's npm dependencies are bundled; external only by opt-in (owner, 2026-10-29: "explicit opt in + (a)"):** a plugin's `dependencies` are installed into its own `node_modules` (`npm install --ignore-scripts`) and bundled into each facet that imports them (host packages, its `peerDependencies`, stay external and are passed in). On Node a diff --git a/packages/chord/docs/design.md b/packages/chord/docs/design.md index e327e2ff0..254784587 100644 --- a/packages/chord/docs/design.md +++ b/packages/chord/docs/design.md @@ -57,6 +57,7 @@ versions and adapters · 28 building: slices · 29 telemetry · 30 cost model · | reference | what `get` returns; keeps its object alive; live and hydrated when awaited | | document | a named JSON value an object owns, written in transactions | | derive | a value computed from watchables | +| live query | a bounded query of an object's own table, kept current by its commits (`self.query`, section 11) | | native | an OS-level value opened with `self.open`; only in objects | | routine | standing behaviour of an object: `self.on(source, fn)`, `self.onLive(fn)` | | job | owned work a call starts that outlives it: intent, run, finish; optional state document per run | @@ -1195,6 +1196,13 @@ host.serve(DraftStore, (_, self) => { // local, empty params: may }, ctx); await edits.get({ id }, ctx); // outside: committed state, ctx last ``` +- **Live queries in the commit step (owner, 2026-11-03; section 11):** a transaction that writes a table with live + queries (`self.query`) stages one more step write after its own writes: it reads each such query on the step's + backend transaction (so it sees the commit's rows, one `SELECT` counted in `chord.store.statements` and + `chord.query.runs`), writes nothing and records nothing in the outbox; the rows are adopted with the documents in the + commit's batch. A refused step adopts nothing. At activation the queries run right after the tables are prepared, + before the references are bound. A transaction writing no table with a live query stages nothing (one map lookup per + written table). - **The outbox (owner, 2026-10-26):** optional, `openStore(backend, ctx, { outbox: true })` (an omitted option keeps the persisted setting; switching needs an explicit `true` or `false`), persisted in the store's metadata (turning it on starts sync from an archive at the current `lastSeq`); a table in the store's backend, @@ -1251,11 +1259,28 @@ host.serve(DraftStore, (_, self) => { // local, empty params: may | Durable's views | the conversation view, task graph and Durable documents | `Watchable`s through Durable's watches; feed derives and state members | | paged method | large or queryable collections | a method returning pages (chat history) | | table | a plugin's own records with queries (owner, 2026-10-21) | `self.table`, the store's query DSL | +| live query | a bounded window of the object's own table, live (owner, 2026-11-03) | `self.query(table, q, limit)` | | plugin database | derived data only: full-text search, caches (rebuilt, never synced) | a native opened by its owner | App state lives in Chord documents; only transcript-bound data (entries, documents that fork and rewind with a conversation) lives in Durable. +**Live queries** (decision "Live queries over an object's own tables", owner, 2026-11-03): `self.query(table, q, +limit)` in the factory of the table's owner returns a `Watchable` of `q`'s rows (at most `limit`, in its order): a +state member, a derive source, an `on` source. Chord runs it before the object goes live (with its documents and +tables, section 10) and again in the commit step of every commit of the object whose staged writes touch that table; +the rows are adopted in the commit's batch and published as `diffRevisions` ops (W1), unchanged rows kept the same +objects so the diff compares them by identity. Rules: only the object's own tables (only it writes them, so its +commits are the exact invalidation: a commit not writing the table runs nothing); the query is fixed for the attempt +(params decide it); **a limit is required** (`invalid/args` at declaration; the store's query language keeps the limit +an argument of the runner, as `page` has it). A run that fails (a row the code cannot read) is a failed member, as a +throwing derive (one failure call, `query_failed` once per episode, `[["r", rows]]` at the next good run); the object +stays live. It is no native, so an object with live queries swaps on reload (section 21). Cost: one query per +touching commit, bounded by the limit (100 rows: about 60 µs unwatched, 80 µs watched, `benchmarks.md`); **there may +be dragons** for tables written hundreds of times a second. Incremental maintenance of the result (row-level changes +kept in memory, a re-run only when a removal empties the window or a bulk write hides which rows changed) is left for +when a case needs it. + ### Watchables (owner, 2026-10-21; log "Watchable and one watch API", W1-W4) There is no `replicatedState` in v2. Everything observable is one of two kinds, sharing one lifecycle: @@ -1921,7 +1946,7 @@ type ReloadReport = { | --- | --- | | its type is removed | **ended**: as an end (natives close `end`, callers of running calls rejected `released`, handlers drained, in-memory documents gone, persisted ones kept); its holders' requests are not ended: they wait unbound `not_served` (§7) | | failed, parked, activating or waiting for its backoff | **retried** on the new code (`ref.retry()`'s rule, whatever broke it); an activating attempt is sealed and its natives close `reload` | - | live, neither its live attempt nor the new factory opens natives, and no table of the new factory is newer than the store's | **swapped** in place (below) | + | live, neither its live attempt nor the new factory opens natives, and no table of the new factory is newer than the store's | **swapped** in place (below); live queries are no natives (section 11): an object showing table rows through `self.query` swaps, one reading them in `self.open` restarts | | live, otherwise | **restarted**: new calls park (it shows `loading`), its contributions leave, running calls drain (bounded by `closeTimeout`; a call outliving the drain is cut off by the seal fence), the attempt is sealed (in-memory job runs end `aborted { reload }`, persisted ones are stopped and resumed, R3), natives close `reload`, the new attempt activates after that teardown (documents stay; eager table migrations run as at activation) | | any of the above whose new code waits for a dependency | **parked** (decision 4): no overlap of old and new code; reported with the cause | @@ -1942,7 +1967,8 @@ type ReloadReport = { variant lacks fails `invalid/member`; §7); exposed derives are matched by their member names, internal ones are rebuilt; in-memory documents migrate through `migrate` when the new code declares a higher version; the contributions are replaced in one batch; `self.writable` starts at its new initial value (decision 2); `retryOn` - sources are re-declared. **Routines hand over per object:** the old attempt's routines are aborted; the new ones + sources are re-declared; the new attempt's live queries run at its activation (the old attempt's are re-run by the + commits of its remaining calls until it is sealed, then dropped). **Routines hand over per object:** the old attempt's routines are aborted; the new ones start once the old routine calls finished, bounded by `closeTimeout` (a stuck one is reported `drain_timeout` and the whole old attempt is sealed then, audit A12: one seal token per attempt); calls never wait. - **Natives are not kept across reloads by name:** natives have no line, transactions or schema, so old and new code diff --git a/packages/chord/docs/guide-plugins-hosts/book.md b/packages/chord/docs/guide-plugins-hosts/book.md index 1a6802c0a..10c73df42 100644 --- a/packages/chord/docs/guide-plugins-hosts/book.md +++ b/packages/chord/docs/guide-plugins-hosts/book.md @@ -75,7 +75,7 @@ addressable hosts; an **address** (`Session({ id: "p1" })`) names one. An **obje on its own host only) or `remote` (used from any host). An **object** is one instance of a type, per params. A **facet** is a plugin's declarations for one host kind. A **plugin** is a directory with a `package.json`. A **reference** is what you get when you ask for an object (`host.get`, `self.get`, `ctx.get`). A **document** is data an object -stores, changed in **transactions**; a **table** holds many records with queries, in the host's **store**; a **derive** is a value computed from others; a **native** is an outside +stores, changed in **transactions**; a **table** holds many records with queries, in the host's **store**, and a **live query** keeps some of its rows current; a **derive** is a value computed from others; a **native** is an outside resource an object opens and closes. A **registry** is an extension point: one plugin defines it, others contribute. A **rollout** applies a build to the hosts in order. @@ -180,6 +180,20 @@ Persisted data lives only in `local` objects with empty params (one per host and on who asked or with which params; an object with params (a view per project) reads from such a store. Writes go through the same transactions as documents: a note row and a document can change in one commit. +A table has no watch of its own. To show rows live (the newest notes in a tab), the object declares a **live query** +over its own table, with a limit: + +```ts +const newest = self.query(rows, recent, 50); // the 50 newest rows, a value like a document's +return { newest, add, page }; // a member: tabs watch it and get only the changes +``` + +Chord runs the query before the object goes live, and again in every commit of the object that writes `rows`, and +sends what changed. Only the object's own tables: it is their only writer, so its commits are exactly when the rows +can change, and a commit that writes no such table costs nothing. Each writing commit runs the query once more, at +most the limit's rows: right for data people edit, wrong for a table written hundreds of times a second (there may be +dragons; keep what the tab shows in a document the object writes with each change instead). The limit is required. + A **derive** is a value computed from other values, recomputed whenever they change. It is how an object offers something it does not store: @@ -212,7 +226,8 @@ self.onLive(async (ctx) => { // a routine start ``` Natives are why some objects restart when their plugin changes: an open connection cannot be moved to new code, so -Chord closes it and opens it again. Objects without natives are swapped in place. +Chord closes it and opens it again. Objects without natives are swapped in place. A first read of a table is no +native: declare it as a live query, not with `self.open`, or the object restarts at every change of its plugin. **Await what you start in a root or a handler.** What is started with a context ends with it. Code outside objects gets its `ctx` from `host.root(fn, ctx)`, and that scope seals when `fn` returns; a handler's context ends when its diff --git a/packages/chord/docs/guide-plugins-hosts/index.html b/packages/chord/docs/guide-plugins-hosts/index.html index 349975217..a69f6570c 100644 --- a/packages/chord/docs/guide-plugins-hosts/index.html +++ b/packages/chord/docs/guide-plugins-hosts/index.html @@ -169,7 +169,7 @@ addressable hosts; an address (Session({ id: "p1&quo on its own host only) or remote (used from any host). An object is one instance of a type, per params. A facet is a plugin's declarations for one host kind. A plugin is a directory with a package.json. A reference is what you get when you ask for an object (host.get, self.get, ctx.get). A document is data an object -stores, changed in transactions; a table holds many records with queries, in the host's store; a derive is a value computed from others; a native is an outside +stores, changed in transactions; a table holds many records with queries, in the host's store, and a live query keeps some of its rows current; a derive is a value computed from others; a native is an outside resource an object opens and closes. A registry is an extension point: one plugin defines it, others contribute. A rollout applies a build to the hosts in order.

@@ -258,6 +258,15 @@ host.serve(NoteStore, (_para

Persisted data lives only in local objects with empty params (one per host and type), so its address never depends on who asked or with which params; an object with params (a view per project) reads from such a store. Writes go through the same transactions as documents: a note row and a document can change in one commit.

+

A table has no watch of its own. To show rows live (the newest notes in a tab), the object declares a live query +over its own table, with a limit:

+
const newest = self.query(rows, recent, 50);             // the 50 newest rows, a value like a document's
+return { newest, add, page };                            // a member: tabs watch it and get only the changes
+

Chord runs the query before the object goes live, and again in every commit of the object that writes rows, and +sends what changed. Only the object's own tables: it is their only writer, so its commits are exactly when the rows +can change, and a commit that writes no such table costs nothing. Each writing commit runs the query once more, at +most the limit's rows: right for data people edit, wrong for a table written hundreds of times a second (there may be +dragons; keep what the tab shows in a document the object writes with each change instead). The limit is required.

A derive is a value computed from other values, recomputed whenever they change. It is how an object offers something it does not store:

const count = self.derive([doc], ([notes]) => notes.items.length);   // 0, 1, 2, … as notes are added
@@ -277,7 +286,8 @@ self.onLive(async (ctx) =>
 	for await (const file of inbox.value.files(ctx.abortSignal)) await importNote(file, ctx);
 });

Natives are why some objects restart when their plugin changes: an open connection cannot be moved to new code, so -Chord closes it and opens it again. Objects without natives are swapped in place.

+Chord closes it and opens it again. Objects without natives are swapped in place. A first read of a table is no +native: declare it as a live query, not with self.open, or the object restarts at every change of its plugin.

Await what you start in a root or a handler. What is started with a context ends with it. Code outside objects gets its ctx from host.root(fn, ctx), and that scope seals when fn returns; a handler's context ends when its caller cancels or its object ends. A change still waiting when its context ends is cancelled before it commits:

diff --git a/packages/chord/docs/plugin-authoring.md b/packages/chord/docs/plugin-authoring.md index 1598dcbe1..ebf6e3271 100644 --- a/packages/chord/docs/plugin-authoring.md +++ b/packages/chord/docs/plugin-authoring.md @@ -64,6 +64,27 @@ A facet only declares (`host.serve`, `self.contribute`, `host.references`), sync happens in objects: state in documents and tables, standing behaviour in a `local` object with empty params (it lives as long as the host). Names are prefixed by the plugin (`bob.stats`). +## Tables shown live + +A table has no watch. To show its rows live, the object owning it declares a live query with a limit: + +```ts +host.serve(Store, (_params, self) => { // a local type with empty params: it may have tables + const items = self.table(Items); + const newest = self.query(items, query(Items).orderBy((i) => i.id, "desc"), 50); + return { items: newest, add: async (row, ctx) => self.tx(async (tx) => tx.put(items, row), ctx) }; +}); +``` + +- Chord runs it before the object goes live and again in each commit of the object that writes the table; watchers get + only what changed. A commit writing no such table runs nothing. +- Only the object's own tables, the query fixed for the object's attempt (decide it from params), the limit required + (`invalid/args` without one). +- One query per writing commit, at most `limit` rows (for 100 rows about 60 µs more per commit, 80 µs when watched; + `benchmarks.md`): there may be dragons for a table written hundreds of times a second. Keep a document for those. +- Never read a table in `self.open` to fill a window: that makes a native, and the object restarts at every reload of + its plugin instead of swapping in place. A live query is no native. + ## What a plugin may import | Import | Resolves to | @@ -209,5 +230,6 @@ of plugins it has no files of, it declares again, as `Notes` above. 3. Imports: the app's API at one version, own files, same-root contract files by relative path, other repositories' by package name pinned in `dependencies`. 4. Contracts prefixed with the plugin's name; a version bump for every change of shape. -5. Files of one kind named `*..ts(x)`. -6. A test with `testPlugin` and stubs for every name `t.unserved()` lists. +5. Live rows of a table through `self.query(table, q, limit)`, never a read in `self.open`. +6. Files of one kind named `*..ts(x)`. +7. A test with `testPlugin` and stubs for every name `t.unserved()` lists. diff --git a/packages/chord/docs/slices.md b/packages/chord/docs/slices.md index 98cb152c5..1c0636a4e 100644 --- a/packages/chord/docs/slices.md +++ b/packages/chord/docs/slices.md @@ -44,6 +44,8 @@ slice spec's section citations are for orientation, not a reading limit (owner, - **Slice 11** (committed 2026-11-03, `slice-11.md`): `src/surfaces/` (scripts objects, registered scripts, the connection loop, clients, themes), `src/web/`, `src/react/`, `src/tui/`, `src/node/surfaces/`, `src/node/tui/` (the launcher and its child-process sandbox); examples 56-59. +- **Live queries** (built 2026-11-03, decision "Live queries over an object's own tables"): `self.query` in + `src/store/live-query.ts`; examples 59 and 61 use it; `test/bench/live-query.ts`. - **Chord's shared code** (upstream pi): `src/watch.ts` (`Watch`, `NativeWatch`, `WatchHandle`, `WatchEnd`, `NativeWatchable`), Delta (`track`, `diffRevisions`, `applyImmutable`, encoders, `PendingOps`, `appendText`), context, JSON, output, SQLite drivers. @@ -136,6 +138,7 @@ pattern, no code. Decided cuts are noted where they apply. | 11 | `Watchable`, watches (Chord's `Watch`), the batch (W1), reentrancy | 1 | | 11 | `NativeWatchable`, `self.writable` | 8 | | 11 | Durable's views as sources (`self.follow` over Durable's `observeXXX`) | 6b | +| 11 | live queries (`self.query(table, q, limit)`: run before live, re-run per touching commit, diffed) | after 11 | | 12 | `self.derive`, failures per member, lazy when unwatched, `mapShared` | 1 | | 12 | container-width benchmark (R24) | 9 | | 13 | `self.open`, close reasons, order, restart patterns (R16) | 2 (restarts on reload: 7b) | @@ -193,12 +196,14 @@ deleted (the slice's spec and code are the record). Moved here from `carry-over. - **`rebuild` of tables** (key and column-type changes): `evolveTable` has no edit for them; build one only when a real case needs it. (Slice 6a's second spec review; out of slice 7b's scope, `slice-7b.md` §12.) +### Later: live queries (when a case needs it) + +- **Incremental maintenance of a live query's result** (decision "Live queries over an object's own tables"): an + ordered, limited result kept from row-level changes in memory, re-running only when a removal empties the window + or a bulk write hides which rows changed; today every touching commit re-runs the query. (Live queries build.) + ### After slice 11 -- **Live queries** (decision "Live queries over an object's own tables"): `self.query(table, q)`, re-run per - touching commit and diffed; tests (activation before live, swap on reload, a commit not touching the table costs - nothing, the limit required, cost guard); the tutorial and 59-chat use it instead of `self.open`; the guide's - section with the warning. Later, when a case needs it: incremental maintenance of an ordered, limited result. - **Messages and languages in the app kit** (decision "Messages and languages in the app kit", proposal `docs/i18n/`): `defineMessages`, `t.message`/`t.render`, `MessagesOf`, the `Translations` registry with fallbacks and conflicts, the locale preference per principal (default from the device, live switch, `lang`/`dir` on diff --git a/packages/chord/docs/tutorial/book.md b/packages/chord/docs/tutorial/book.md index 762ae170d..66ee21b52 100644 --- a/packages/chord/docs/tutorial/book.md +++ b/packages/chord/docs/tutorial/book.md @@ -70,8 +70,8 @@ without one (a **client**: a page, a terminal). A **link** connects two hosts. A the app and stamped by links. An **object** is state and methods at one identity; its **contract** (`local` for its own host only, `remote` for any host) names its **params** (which instance) and members. A remote contract's **variants** are the member sets different principals get. A **factory** declares an object's parts: **documents** and **tables** -(stored data, changed in **transactions**), **derives** (values computed from others), **natives** (things opened and -closed, such as a file or a database cursor). A **reference** is what asking for an object returns. A **facet** is a +(stored data, changed in **transactions**), **live queries** (a table's rows Chord keeps current), **derives** +(values computed from others), **natives** (things opened and closed, such as a file or a database cursor). A **reference** is what asking for an object returns. A **facet** is a plugin's code for one host kind; a **registry** is an extension point plugins **contribute** to. A **rollout** applies new plugin code to the hosts in order. @@ -285,7 +285,7 @@ each start. It sits in the URL's fragment, which browsers never send, so it neve it and receives a session cookie. A command-line client on the same machine reads `owner-token` instead (mode 0600, chapter III). Nothing is the owner because it comes from loopback: any local process could. -# A List on Its Own Worker {#worker lede="A host kind, a remote contract, a worker facet with a table and a live window, and the principal that adds a todo."} +# A List on Its Own Worker {#worker lede="A host kind, a remote contract, a worker facet with a table and a live query, and the principal that adds a todo."} Each list gets its own process, a **worker**, with its own store: a list that misbehaves cannot take the others down, and an idle list costs nothing. This chapter declares where lists live, what a list offers, and the worker code that @@ -376,7 +376,7 @@ const Items = defineTable("todo.items", { /** The newest todos first. */ const NEWEST = query(Items).orderBy((item) => item.id, "desc"); -/** How many todos the live window holds. */ +/** How many todos the live query holds. */ const WINDOW = 50; ``` @@ -406,40 +406,38 @@ of references. Its factory declares its parts and returns its members: ```ts host.serve(Store, (_params, self) => { const items = self.table(Items); - // The newest todos, oldest first, and the next todo's number: read once when the store goes live, then kept here. - const window = self.writable([]); - let next = 1; - self.open( - async (ctx) => { - const { items: newest } = await items.page(NEWEST, WINDOW, ctx); - window.set([...newest].reverse()); - next = (newest[0]?.id ?? 0) + 1; - }, - () => {}, - ); + // The newest todos, live: read before the store goes live, read again after every commit writing the table. + const newest = self.query(items, NEWEST, WINDOW); return { - items: self.derive([window], ([value]) => value), + // Oldest first, as the list shows them. + items: self.derive([newest], ([rows]) => [...rows].reverse()), ``` A **factory** is synchronous and only declares; Chord activates the object afterwards. `self.table` declares the -table. A table has no watch, so the store keeps a **live window** itself: `self.writable` holds the newest 50 todos, -filled by `self.open`, which runs during activation, and exposed through `self.derive` (a writable is an in-process -value, not a state member; a derive over it is). `self.open` is meant for **natives**, things opened and closed; the -window's first read is one, and chapter VIII shows its price. +table. `self.query(items, NEWEST, WINDOW)` declares a **live query** over it: the rows of `NEWEST`, at most 50, as a +value the object can expose or derive from. Chord runs it before the store goes live, and again after every commit +of the store that writes `items`, and sends watchers only what changed. The store's derive turns the rows around, +oldest first, for the list. + +A live query reads only its own object's tables: only the store writes them, so its own commits are exactly the +moments the rows can change, and a commit that writes no table with a query costs nothing. The limit is required: +each such commit runs the query once more, at most `WINDOW` rows. That is cheap for a list a person edits and wrong +for a table written hundreds of times a second (design §11, "there may be dragons"). Writes go through **transactions**: `self.tx(fn, ctx)` runs `fn` with `tx`, and what it stages is committed at once -when `fn` returns, or not at all if it throws. The window follows after the commit; the store is the table's only -writer, so it knows every change: +when `fn` returns, or not at all if it throws. The live query follows in the same commit; the store keeps nothing +else in memory, not even the next todo's number, which it reads in the transaction: ```ts add: async ({ text, by }, ctx) => { // Refused as Chord refuses arguments that fail their schema: the caller gets `invalid/args`. if (text === "") throw invalid("args", "a todo needs text"); - const item = { id: next++, text, done: false, by, at: Date.now() }; - await self.tx(async (tx) => tx.put(items, item), ctx); - // In number order: two adds may commit in either order. - window.set([...window.value, item].sort((a, b) => a.id - b.id).slice(-WINDOW)); + await self.tx(async (tx) => { + // The number after the newest todo's: the store's transactions run one after another. + const { items: last } = await tx.page(items, NEWEST, 1); + tx.put(items, { id: (last[0]?.id ?? 0) + 1, text, done: false, by, at: Date.now() }); + }, ctx); return null; ``` @@ -1057,7 +1055,7 @@ console.log(" ada adds ' ':", refused); ```text 8. lists changed, rolled out to: server, list, web, tui - on list|groceries: todo.store restarted, todo.todos(groceries) parked + on list|groceries: todo.store swapped, todo.todos(groceries) parked ada adds ' tea ': [ ] tea (ada) ada adds ' ': { code: 'invalid', reason: 'args' } ``` @@ -1074,9 +1072,21 @@ and rolled it out host by host. On each host Chord stages the new code (it runs | its new code waits for a dependency | **parked** until the dependency is live | | its type was removed | **ended**; its holders wait until something serves it again | -The store opens a native (the window's first read), so it restarted and read its window again from the table; the -views waited for it (parked), then went on with the new code. Ada's reference never noticed more than a moment of -`loading`, and her next add was trimmed. Her blank one is now empty after trimming, so the store refused it with +The store opens no natives, and neither does its new factory: its live query is not a native, Chord runs it. So the +store was **swapped**, not restarted. The sequence on the list's worker: + +1. The store's new attempt starts beside the old one: the same object, instance, table and line. Calls arriving now + wait at a gate; calls already running finish on the old code. +2. The new attempt runs its live query (the same read as at the first activation) and goes live; the gate opens, + and references move to its members by name, each with one frame from its last value. +3. The old attempt is sealed once its calls settled; its query leaves with it, so later commits run only the new one. + +The views' new attempts need the store, which was still between steps 1 and 2 when they checked it, so they +**parked** for that moment and went live with the new code right after. Ada's reference waited a moment, and her next add was trimmed. +Had the store kept its window as a one-time read in `self.open`, it would count as holding a native and **restart** +at every reload of its plugin: running calls drained, natives closed and opened again, a new instance. + +Ada's blank todo is now empty after trimming, so the store refused it with `invalid("args", …)` (chapter II): her client got `{ code: "invalid", reason: "args" }`, Chord's own error, which crossed the worker's link and the server's as it was. Had the edit broken the plugin, that plugin alone would have kept its old code and the report would name the error. @@ -1364,14 +1374,11 @@ a tutorial. `"~kind": "Integer"`, which pi-schema's `int()` lacks, so the key of an `int` key column is typed `string`: `tx.get(items, { id })` with a number fails `tsc`. The store here uses `updateWhere` with `eq` instead of `tx.update`. -- **A one-time read at activation is a native.** The live window over a table needs a first read before the object is - live; `self.open` is the only place, so the store counts as holding a native and restarts at every reload of its - plugin, parking its views, though it holds nothing open. Tables have no watch, so every table-backed live view - repeats this pattern (example 59's chat does too). - -Three more, found the same way, are fixed: a reloaded web plugin lost its own React state (now `usePageState`, +Four more, found the same way, are fixed: a reloaded web plugin lost its own React state (now `usePageState`, chapter VIII), a plugin's `contract.ts` could not be imported outside builds (the API is now a workspace package, -chapter I), and plugins could not throw Chord's `invalid` (Chord exports `invalid` and `forbidden`, chapter II). +chapter I), plugins could not throw Chord's `invalid` (Chord exports `invalid` and `forbidden`, chapter II), and a +live window over a table needed a one-time read in `self.open`, which made the store restart at every reload of its +plugin (now `self.query`, a live query Chord runs, chapters II and VIII). # The App's Files {.appendix} @@ -1399,7 +1406,7 @@ chapter I), and plugins could not throw Chord's `invalid` (Chord exports `invali | `startWorker({ api, kinds, onError })` | I | a worker's entry | | `app.ownerLink()`, `app.ownerTokenFile` | I, III | the owner's sign-in, for a browser and for local clients | | `hostKind(name, params)`, `remote(name, { version, params, hostAddress, variants })`, `local(name, { version, params, members })` | II | host kinds and contracts | -| `host.serve(Type, factory)`; factory `self`: `table`, `writable`, `open`, `derive`, `tx`, `get`, `contribute` | II | an object's parts | +| `host.serve(Type, factory)`; factory `self`: `table`, `query`, `derive`, `tx`, `get`, `contribute` | II | an object's parts | | `host.root(fn, ctx, { principal })`, `ctx.get(Type, params)` | II | code outside objects, as a principal | | `invalid(reason, message, details?)`, `forbidden(message, details?)` | II, VIII | Chord's errors, thrown by a plugin | | `self.contribute(Http, { routes })` | III | routes, denied by default | diff --git a/packages/chord/docs/tutorial/index.html b/packages/chord/docs/tutorial/index.html index 9f0f5cae9..39d4c9793 100644 --- a/packages/chord/docs/tutorial/index.html +++ b/packages/chord/docs/tutorial/index.html @@ -181,8 +181,8 @@ without one (a client: a page, a terminal). A linkobject is state and methods at one identity; its contract (local for its own host only, remote for any host) names its params (which instance) and members. A remote contract's variants are the member sets different principals get. A factory declares an object's parts: documents and tables -(stored data, changed in transactions), derives (values computed from others), natives (things opened and -closed, such as a file or a database cursor). A reference is what asking for an object returns. A facet is a +(stored data, changed in transactions), live queries (a table's rows Chord keeps current), derives +(values computed from others), natives (things opened and closed, such as a file or a database cursor). A reference is what asking for an object returns. A facet is a plugin's code for one host kind; a registry is an extension point plugins contribute to. A rollout applies new plugin code to the hosts in order.

@@ -353,7 +353,7 @@ it and receives a session cookie. A command-line client on the same machine read chapter III). Nothing is the owner because it comes from loopback: any local process could.

-
II

A List on Its Own Worker

A host kind, a remote contract, a worker facet with a table and a live window, and the principal that adds a todo.

+
II

A List on Its Own Worker

A host kind, a remote contract, a worker facet with a table and a live query, and the principal that adds a todo.

Each list gets its own process, a worker, with its own store: a list that misbehaves cannot take the others down, and an idle list costs nothing. This chapter declares where lists live, what a list offers, and the worker code that keeps its todos.

@@ -425,7 +425,7 @@ worker's store:

/** The newest todos first. */ const NEWEST = query(Items).orderBy((item) => item.id, "desc"); -/** How many todos the live window holds. */ +/** How many todos the live query holds. */ const WINDOW = 50;

A table holds many records with queries and pages; it lives in an object, never in a facet. Persisted data lives only in local objects with empty params (one per host and type), so its address never depends on who asked (design §9, @@ -447,35 +447,32 @@ of references. Its factory declares its parts and returns its members:

	host.serve(Store, (_params, self) => {
 		const items = self.table(Items);
-		// The newest todos, oldest first, and the next todo's number: read once when the store goes live, then kept here.
-		const window = self.writable<Item[]>([]);
-		let next = 1;
-		self.open(
-			async (ctx) => {
-				const { items: newest } = await items.page(NEWEST, WINDOW, ctx);
-				window.set([...newest].reverse());
-				next = (newest[0]?.id ?? 0) + 1;
-			},
-			() => {},
-		);
+		// The newest todos, live: read before the store goes live, read again after every commit writing the table.
+		const newest = self.query(items, NEWEST, WINDOW);
 		return {
-			items: self.derive([window], ([value]) => value),
+ // Oldest first, as the list shows them. + items: self.derive([newest], ([rows]) => [...rows].reverse()),

A factory is synchronous and only declares; Chord activates the object afterwards. self.table declares the -table. A table has no watch, so the store keeps a live window itself: self.writable holds the newest 50 todos, -filled by self.open, which runs during activation, and exposed through self.derive (a writable is an in-process -value, not a state member; a derive over it is). self.open is meant for natives, things opened and closed; the -window's first read is one, and chapter VIII shows its price.

+table. self.query(items, NEWEST, WINDOW) declares a live query over it: the rows of NEWEST, at most 50, as a +value the object can expose or derive from. Chord runs it before the store goes live, and again after every commit +of the store that writes items, and sends watchers only what changed. The store's derive turns the rows around, +oldest first, for the list.

+

A live query reads only its own object's tables: only the store writes them, so its own commits are exactly the +moments the rows can change, and a commit that writes no table with a query costs nothing. The limit is required: +each such commit runs the query once more, at most WINDOW rows. That is cheap for a list a person edits and wrong +for a table written hundreds of times a second (design §11, "there may be dragons").

Writes go through transactions: self.tx(fn, ctx) runs fn with tx, and what it stages is committed at once -when fn returns, or not at all if it throws. The window follows after the commit; the store is the table's only -writer, so it knows every change:

+when fn returns, or not at all if it throws. The live query follows in the same commit; the store keeps nothing +else in memory, not even the next todo's number, which it reads in the transaction:

			add: async ({ text, by }, ctx) => {
 				// Refused as Chord refuses arguments that fail their schema: the caller gets `invalid/args`.
 				if (text === "") throw invalid("args", "a todo needs text");
-				const item = { id: next++, text, done: false, by, at: Date.now() };
-				await self.tx(async (tx) => tx.put(items, item), ctx);
-				// In number order: two adds may commit in either order.
-				window.set([...window.value, item].sort((a, b) => a.id - b.id).slice(-WINDOW));
+				await self.tx(async (tx) => {
+					// The number after the newest todo's: the store's transactions run one after another.
+					const { items: last } = await tx.page(items, NEWEST, 1);
+					tx.put(items, { id: (last[0]?.id ?? 0) + 1, text, done: false, by, at: Date.now() });
+				}, ctx);
 				return null;

A todo without text is refused with invalid("args", …), from the API: the error crosses every link as it is, so a caller on a page or in a terminal gets { code: "invalid", reason: "args" }, as for arguments that fail the @@ -959,7 +956,7 @@ console.log(" ada adds ' tea ': console.log(" ada adds ' ':", refused); 

8. lists changed, rolled out to: server, list, web, tui
-   on list|groceries: todo.store restarted, todo.todos(groceries) parked
+   on list|groceries: todo.store swapped, todo.todos(groceries) parked
    ada adds '  tea  ': [ ] tea (ada)
    ada adds ' ': { code: 'invalid', reason: 'args' }

The watcher marked @todo/lists dirty when the file changed; applyChanges rebuilt it (one esbuild call per facet) @@ -990,9 +987,20 @@ and rolled it out host by host. On each host Chord stages the new code (it runs ended; its holders wait until something serves it again -

The store opens a native (the window's first read), so it restarted and read its window again from the table; the -views waited for it (parked), then went on with the new code. Ada's reference never noticed more than a moment of -loading, and her next add was trimmed. Her blank one is now empty after trimming, so the store refused it with +

The store opens no natives, and neither does its new factory: its live query is not a native, Chord runs it. So the +store was swapped, not restarted. The sequence on the list's worker:

+
    +
  1. The store's new attempt starts beside the old one: the same object, instance, table and line. Calls arriving now +wait at a gate; calls already running finish on the old code.
    2. +
  2. The new attempt runs its live query (the same read as at the first activation) and goes live; the gate opens, +and references move to its members by name, each with one frame from its last value.
    3. +
  3. The old attempt is sealed once its calls settled; its query leaves with it, so later commits run only the new one.
    4. +
+

The views' new attempts need the store, which was still between steps 1 and 2 when they checked it, so they +parked for that moment and went live with the new code right after. Ada's reference waited a moment, and her next add was trimmed. +Had the store kept its window as a one-time read in self.open, it would count as holding a native and restart +at every reload of its plugin: running calls drained, natives closed and opened again, a new instance.

+

Ada's blank todo is now empty after trimming, so the store refused it with invalid("args", …) (chapter II): her client got { code: "invalid", reason: "args" }, Chord's own error, which crossed the worker's link and the server's as it was. Had the edit broken the plugin, that plugin alone would have kept its old code and the report would name the error.

@@ -1254,15 +1262,13 @@ a tutorial.

  • Integer table keys do not type-check. KeyOf in src/store/table.ts still tests TypeBox's "~kind": "Integer", which pi-schema's int() lacks, so the key of an int key column is typed string: tx.get(items, { id }) with a number fails tsc. The store here uses updateWhere with eq instead of -tx.update.
    • -
  • A one-time read at activation is a native. The live window over a table needs a first read before the object is -live; self.open is the only place, so the store counts as holding a native and restarts at every reload of its -plugin, parking its views, though it holds nothing open. Tables have no watch, so every table-backed live view -repeats this pattern (example 59's chat does too).
    • - -

    Three more, found the same way, are fixed: a reloaded web plugin lost its own React state (now usePageState, +tx.update. +Four more, found the same way, are fixed: a reloaded web plugin lost its own React state (now usePageState, chapter VIII), a plugin's contract.ts could not be imported outside builds (the API is now a workspace package, -chapter I), and plugins could not throw Chord's invalid (Chord exports invalid and forbidden, chapter II).

    +chapter I), plugins could not throw Chord's invalid (Chord exports invalid and forbidden, chapter II), and a +live window over a table needed a one-time read in self.open, which made the store restart at every reload of its +plugin (now self.query, a live query Chord runs, chapters II and VIII). +
    A

    The App's Files

    @@ -1372,7 +1378,7 @@ chapter I), and plugins could not throw Chord's invalid (Chord host kinds and contracts -host.serve(Type, factory); factory self: table, writable, open, derive, tx, get, contribute +host.serve(Type, factory); factory self: table, query, derive, tx, get, contribute II an object's parts diff --git a/packages/chord/src/documents/transaction.ts b/packages/chord/src/documents/transaction.ts index 8fc02a29e..b0151220f 100644 --- a/packages/chord/src/documents/transaction.ts +++ b/packages/chord/src/documents/transaction.ts @@ -9,6 +9,7 @@ import type { JobHandle, Outcome, Run, StartOptions } from "../jobs/index.ts"; import type { Job } from "../jobs/internal.ts"; import type { Columns, PutRowOf, RowOf } from "../store/columns.ts"; import type { Condition, SetOf } from "../store/expression.ts"; +import { refreshOf } from "../store/live-query.ts"; import type { ColumnRefs, Query } from "../store/query.ts"; import { refusalOf } from "../store/refusal.ts"; import type { StagedWrite } from "../store/step.ts"; @@ -332,6 +333,7 @@ export class Transaction implements Tx { this.#assertOpen(); if (this.#pending > 0) throw new Error(`${this.#documents.label}: the callback settled before its pending tx operations`); + if (this.#tables !== undefined) this.#refreshQueries(this.#tables); this.#state = "committing"; this.#context.abortSignal?.throwIfAborted(); const adoptions: Adoption[] = []; @@ -519,6 +521,18 @@ export class Transaction implements Tx { return changes; } + /** + * Stages the re-run of the live queries over the tables this transaction wrote (decision "Live queries over an + * object's own tables"): read in the commit step after every table write staged so far (the callback's), published + * in the commit's batch. None when no written table has a live query. + */ + #refreshQueries(tables: ReadonlyMap): void { + const refresh = refreshOf(tables.keys(), this.#context); + if (refresh === undefined) return; + this.#writes!.staged.push(refresh.write); + this.stage(refresh, () => refresh); + } + /** Counts `read` as pending until the operation's promise settles (rule 3). */ async #track(read: Promise): Promise { this.#pending++; diff --git a/packages/chord/src/host/README.md b/packages/chord/src/host/README.md index 377ace2ea..23d6b91b2 100644 --- a/packages/chord/src/host/README.md +++ b/packages/chord/src/host/README.md @@ -193,8 +193,9 @@ server.accept(webSocketLink(ws), { principal, limits: { part: 16 * MiB, decoded: `activation_failed` (`object`, `failure`: every failed attempt), `object_failed` (`object`, `failure`: a live object failed, slice 6a), `storage_failed` (`error`: the host's store stopped; the host closes itself), `derive_failed` (`object`, `member?`, `error`: a derive's compute threw, once per failure episode, watched or not; `member` when it is - an exposed state member), `close_failed` (`object`, `native` position, `error`: a close that threw while nobody - waited), `close_timeout` (`object`, `native`: a close still running at its `closeTimeout`, a possible leak), `drain_timeout` (`object`: calls still running at `closeTimeout`; their natives close anyway), + an exposed state member), `query_failed` (`object`, `table`, `error`: a live query's run in a commit threw, once per + failure episode; the query is a failed member until a good run), `close_failed` (`object`, `native` position, + `error`: a close that threw while nobody waited), `close_timeout` (`object`, `native`: a close still running at its `closeTimeout`, a possible leak), `drain_timeout` (`object`: calls still running at `closeTimeout`; their natives close anyway), `listener_error` (a throwing watch listener), `delivery_failed` (a frame delivery or a link's flush that threw) and `protocol_error` (`error`: what closed a link, see `link/`) and `resolve_failed` (`address`: its key, `error`: a route's `resolve` rejected; the address's holders fail `waiting` with it), `routine_failed` (`object`, `routine`: its position, @@ -364,13 +365,23 @@ server.accept(webSocketLink(ws), { principal, limits: { part: 16 * MiB, decoded: - `createHost({ store?, ... })` (slice 6a): the host's store (`openStore` in `store/`), which the app opens, gives to one open host at a time (`invalid/args` otherwise) and closes after it. A backend error stops the store: the host reports `storage_failed` once and closes itself. A `local` type with empty params may then declare - `self.table(definition)` (once per table name) and persisted documents (`defineState(..., { persist: true, - checkpointWhen })`); the types refuse both elsewhere (P3), and so does the factory at runtime. Its activation - loads the persisted singletons and pinned keys, then creates or migrates its tables (the `readyTimeout` deadline - paused meanwhile, resumed with its remaining budget), then binds its references. A table or document the code - cannot read (a newer version, a history edited in place, a failed `rows` step or `migrate`) fails the object - `broken`; a live object whose read fails so fails while live (`object_failed`), keeping its documents for the - next attempt. + `self.table(definition)` (once per table name), live queries over them (`self.query(table, q, limit)`, below) and + persisted documents (`defineState(..., { persist: true, checkpointWhen })`); the types refuse them elsewhere (P3), + and so does the factory at runtime. Its activation loads the persisted singletons and pinned keys, then creates or + migrates its tables (the `readyTimeout` deadline paused meanwhile, resumed with its remaining budget), then runs its + live queries, then binds its references. +- `self.query(table, q, limit)` (decision "Live queries over an object's own tables"): a JSON watchable of `q`'s rows + (at most `limit`, in its order) over a table this factory declared; a state member, a derive source, an `on` source. + Run at activation (a failure fails it `broken`) and again inside the commit step of every commit of the object that + writes the table, after the commit's writes, published as the diff in the commit's batch (`store/live-query.ts`); a + commit writing no table with a query runs none (`chord.query.runs` counts the runs). Fixed for the attempt; refused + at declaration (`invalid/args`) without a positive integer `limit`, for a query of another table, or a table this + factory did not declare. A run that fails (a row its code cannot read) makes it a failed member, as a throwing + derive: one failure call, `query_failed` once, `[["r", rows]]` at the next good run; the object stays live. It is + no native: an object with live queries swaps on reload. +- A table or document the code cannot read (a newer version, a history edited in place, a failed `rows` step or + `migrate`) fails the object `broken`; a live object whose read fails so fails while live (`object_failed`), keeping + its documents for the next attempt. - A factory `(params, self) => implementation` declares with `self.state`, `self.derive`, `self.get`, `self.open`, `self.on`, `self.onLive`, `self.job` (`jobs/`), `self.registry`, `self.writable` and `self.retryOn` (below), and writes with `self.tx`. It runs once per attempt. The implementation has exactly the contract's members: state diff --git a/packages/chord/src/host/attempt.ts b/packages/chord/src/host/attempt.ts index 725bbb191..4f1367245 100644 --- a/packages/chord/src/host/attempt.ts +++ b/packages/chord/src/host/attempt.ts @@ -25,7 +25,9 @@ import type { Contribution, Registry, RegistryValue } from "../registries/index. import type { RegistryContribution, RegistryWaiter } from "../registries/internal.ts"; import type { Writable } from "../state/index.ts"; import { Derive, type DeriveSource, WritableSource } from "../state/internal.ts"; -import type { Columns } from "../store/columns.ts"; +import type { Columns, RowOf } from "../store/columns.ts"; +import { checkLiveQuery, LiveQuery } from "../store/live-query.ts"; +import type { Query } from "../store/query.ts"; import { tableSqlName } from "../store/sql.ts"; import type { ChordStore } from "../store/store.ts"; import type { TableDefinition } from "../store/table.ts"; @@ -69,8 +71,8 @@ import { /** * What a factory declares with. Declarations (`state`, `derive`, `get`, `registry`, `writable`, `retryOn`, `open`, - * `on`, `onLive`, `job`) are valid only while it runs. `P` is the app's principal type, which the ctx of routine calls - * and job runs carry. + * `on`, `onLive`, `job`; `table`, `query` and `contribute` in `StoreSelf`) are valid only while it runs. `P` is the + * app's principal type, which the ctx of routine calls and job runs carry. */ export interface Self

    extends RoutineDeclarations

    { /** This object's documents of `definition`, in memory for its life (across its attempts). */ @@ -146,6 +148,18 @@ export interface StoreSelf

    extends Self

    { state>(definition: D): HandleOf; /** This object's table `definition` in the host's store, once per table name. */ table(definition: TableDefinition): Table; + /** + * A live query (decision "Live queries over an object's own tables"): the rows of `q` (at most `limit`, in its + * order) over `table`, a table this factory declared, as a JSON watchable (a state member, a derive source). Run + * before the object goes live, and again in the commit step of every commit of this object that writes `table` + * (one query per such commit, bounded by `limit`; not for tables written hundreds of times a second). A limit is + * required (`invalid/args`). + */ + query( + table: Table, + q: Query, + limit: number, + ): Watchable[]>; /** * Contributes `value` to registry `R` from the moment this object is live until it ends, fails or restarts (R21: a * host-lifetime object): a plain value, or a Chord watchable of one (followed: its change recomposes). At most one @@ -224,12 +238,12 @@ export function notLive(label: string, ms: number): ChordError { } /** - * One attempt of an object. Activation: factory, documents hydrated (the object's slots), references bound, then parked - * until every dependency is live (no timer: the deadline paused), then its own steps: natives opened one after - * another, followed sources attached, members checked, derives computed; bounded by the host's `readyTimeout`. - * Ending: sealed at once (calls aborted, its transactions refused and the one holding the line discarded, its job - * invocations cut off), then the drain, the closes in reverse, derives detached; its object releases its references. - * Documents and job runs stay on the object. + * One attempt of an object. Activation: factory, documents hydrated (the object's slots), tables prepared and live + * queries run, references bound, then parked until every dependency is live (no timer: the deadline paused), then its + * own steps: natives opened one after another, followed sources attached, members checked, derives computed; bounded + * by the host's `readyTimeout`. Ending: sealed at once (calls aborted, its transactions refused and the one holding the + * line discarded, its job invocations cut off), then the drain, the closes in reverse, derives and live queries + * detached; its object releases its references. Documents and job runs stay on the object. */ export class Attempt implements StoreSelf, KeyDeclarer, RegistryWaiter { readonly #object: ChordObject; @@ -264,6 +278,8 @@ export class Attempt implements StoreSelf, KeyDeclarer, RegistryWaiter { readonly #jobs = new Map(); /** The tables it declared, by name. */ readonly #tables = new Map(); + /** Its live queries, run at activation (after the tables are prepared), detached at its end. */ + readonly #queries: LiveQuery[] = []; /** References this attempt declared with `self.get` (params checked), bound during activation. */ readonly #references: { readonly binding: Binding; readonly contract: Contract; readonly params: CheckedParams }[] = []; @@ -347,6 +363,7 @@ export class Attempt implements StoreSelf, KeyDeclarer, RegistryWaiter { label, scope: object.documents, assertLive: () => object.documents.assertLive(), + queries: object.liveQueries, fail: (error: ChordError) => object.failLive(this, { kind: "broken", error }), }; // The handle's rows are the definition's, typed by `C` and `K`. @@ -355,6 +372,32 @@ export class Attempt implements StoreSelf, KeyDeclarer, RegistryWaiter { return handle as unknown as Table; } + query( + table: Table, + q: Query, + limit: number, + ): Watchable[]> { + this.#assertDeclaring("query"); + const { host, label } = this.#object; + // Plugin code may pass any table: only a handle this attempt declared is taken (its object's own table). + const handle: unknown = table; + if (!(handle instanceof TableHandle) || this.#tables.get(handle.definition.name) !== handle) { + throw new Error(`${label}: self.query: the table is not one this factory declared`); + } + checkLiveQuery(handle.definition, q as unknown as Query, limit, label); + const owner = { + label, + report: (error: unknown) => + host.report({ kind: "query_failed", object: label, table: handle.definition.name, error }), + }; + // The query is of `handle`'s table (checked above). + const live = new LiveQuery(handle, q as unknown as Query, limit, owner, host.batch); + handle.owner.queries.add(live); + this.#queries.push(live); + // Its rows are the table's, typed by `C`. + return live as unknown as Watchable[]>; + } + derive | NativeWatchable)[], T extends JsonValue>( sources: S, compute: ( @@ -918,14 +961,16 @@ export class Attempt implements StoreSelf, KeyDeclarer, RegistryWaiter { } /** - * Detaches its derives (once their object no longer needs them: reads throw `released`, their watches end), its - * followed sources (their observers detached, their watches ended) and its writables (their watches ended). + * Detaches its derives and live queries (once their object no longer needs them: reads throw `released`, their + * watches end; no commit re-runs the queries), its followed sources (their observers detached, their watches ended) + * and its writables (their watches ended). */ detach(): void { const ended = this.#object.ended(); for (const adapted of this.#adapted?.values() ?? []) adapted.detach(ended); this.#adapted = undefined; for (const derive of this.#derives) derive.detach(ended); + for (const query of this.#queries) query.detach(ended); for (const follow of this.#follows) follow.detach(); const declarations = this.#registryDeclarations; if (declarations !== undefined) for (const writable of declarations.writables) writable.detach(ended); @@ -952,6 +997,7 @@ export class Attempt implements StoreSelf, KeyDeclarer, RegistryWaiter { const loading = documents.load(this.#opening); if (loading !== undefined) await loading; if (this.#tables.size > 0) await this.#prepareTables(deadline); + for (const query of this.#queries) await query.load(this.#opening); // A persisted job's runs, loaded once per object: rebuilt before anything runs. for (const job of this.#jobs.values()) await job.load(this.#opening); const { host } = this.#object; diff --git a/packages/chord/src/host/host.ts b/packages/chord/src/host/host.ts index cd86dce47..ac11ada8e 100644 --- a/packages/chord/src/host/host.ts +++ b/packages/chord/src/host/host.ts @@ -141,6 +141,11 @@ export type HostError = * creation; a value ends the episode); `member` when it is an exposed state member. */ | { readonly kind: "derive_failed"; readonly object: string; readonly member?: string; readonly error: unknown } + /** + * A live query's run in a commit threw (a row its code cannot read), once per failure episode (a good run ends it): + * the query is a failed member until then (`self.query`). + */ + | { readonly kind: "query_failed"; readonly object: string; readonly table: string; readonly error: unknown } /** A routine's call threw, or its source failed for good; `routine` is its kind and position. */ | { readonly kind: "routine_failed"; readonly object: string; readonly routine: Position; readonly error: unknown } /** A job's `run` threw (not after an abort): its outcome is `failed { threw }`. */ diff --git a/packages/chord/src/host/object.ts b/packages/chord/src/host/object.ts index 783434655..d6971c482 100644 --- a/packages/chord/src/host/object.ts +++ b/packages/chord/src/host/object.ts @@ -21,6 +21,7 @@ import type { JsonValue } from "../json.ts"; import { CHAIN } from "../link/internal.ts"; import type { RegistryContribution } from "../registries/internal.ts"; import type { DeriveSource } from "../state/internal.ts"; +import { LiveQueries } from "../store/live-query.ts"; import { refusal } from "./adapted.ts"; import { Attempt, type PinnedKey, type Self } from "./attempt.ts"; import type { Caller } from "./ctx.ts"; @@ -223,6 +224,8 @@ export class ChordObject implements Target { #wakes: RetryWakes | undefined; /** Its swapped-out attempts while they retire (bounded by `closeTimeout`); made by the first swap. */ #retiring: Map> | undefined; + /** Its live queries by table, across its attempts; made by the first table it declares. */ + #liveQueries: LiveQueries | undefined; constructor( host: AnyHost, @@ -545,6 +548,12 @@ export class ChordObject implements Target { this.host.store!.claims.table(table, this); } + /** Its live queries by table (`self.query`), across its attempts: what its commits writing a table re-run. */ + get liveQueries(): LiveQueries { + this.#liveQueries ??= new LiveQueries(); + return this.#liveQueries; + } + /** Claims this object's documents in `namespace` of the host's store, as `claimTable`. */ claimDocuments(namespace: string): void { this.host.store!.claims.documents(this.served.contract.name, namespace, this); diff --git a/packages/chord/src/host/staging.ts b/packages/chord/src/host/staging.ts index c359ed8e2..f085e394d 100644 --- a/packages/chord/src/host/staging.ts +++ b/packages/chord/src/host/staging.ts @@ -15,6 +15,8 @@ import type { AnyDefinition } from "../documents/internal.ts"; import type { JobDefinition } from "../jobs/index.ts"; import type { Registry, RegistryProvider } from "../registries/index.ts"; import { checkProvider, type RegistryValueSource } from "../registries/internal.ts"; +import { checkLiveQuery } from "../store/live-query.ts"; +import type { Query } from "../store/query.ts"; import { tableSqlName } from "../store/sql.ts"; import type { ChordStore } from "../store/store.ts"; import type { TableDefinition } from "../store/table.ts"; @@ -276,6 +278,16 @@ class RecordingSelf { return INERT; } + /** A live query over a table this factory declared (its handle is inert: the query names the table), limited. */ + query(_table: unknown, q: Query, limit: unknown): never { + this.#assertDeclaring("query"); + const label = this.#object.label; + const table = this.#tables.get(q.table.name); + if (table !== q.table) throw new Error(`${label}: self.query: the table is not one this factory declared`); + checkLiveQuery(table, q, limit, label); + return INERT; + } + derive(): never { this.#assertDeclaring("derive"); return INERT; diff --git a/packages/chord/src/store/README.md b/packages/chord/src/store/README.md index 14238e36a..adaaa5b12 100644 --- a/packages/chord/src/store/README.md +++ b/packages/chord/src/store/README.md @@ -43,6 +43,9 @@ const copy = await importStore(sqliteBackend(db), decodeJsonl(bytes), ctx); optional data step (`rows(fn)` or `update({ where?, set })`), `drop`, `index`, `dropIndex` after it. - Queries: `query(Table).where((t) => cond).orderBy((t) => col, dir)` with `eq ne lt lte gt gte inArray isNull prefix and or not` (SQL's `NULL` rules); pages `{ items, next? }` with opaque cursors, `rows()` paging by 100, `count`. +- Live queries: `self.query(table, q, limit)` in the owning object's factory, a watchable of up to `limit` rows, run + at activation and again in the commit step of each commit of the object that writes the table (`live-query.ts`; + `host/` describes the rules). One query per such commit: not for tables written hundreds of times a second. - Writes in `self.tx`: `put` (an upsert), `update(key, set)`, `delete(key)`, `updateWhere(where, set)`, `deleteWhere(where)`; a `set` is values or expressions (`add sub max min concat replacePrefix coalesce when`); key columns never change. `tx.get` sees the transaction's own single-key writes, not after a bulk one (W10). @@ -72,10 +75,11 @@ call and at its turn while an owner holds a claim or once the store diverged, an `take()` for an export, the catalog), the commit `Step` and `StagedWrite`s in their `WriteOrder` (`step.ts`), the documents engine (`engine.ts`), the catalog (`catalog.ts`), the write vocabulary (`writes.ts`, `table-writes.ts`), rows and columns (`rows.ts`, `columns.ts`, `json-column.ts`), the compiled where/set vocabulary (`expression.ts`) and its JS evaluator (`evaluate.ts`), -SQL (`sql.ts`), the table handle and a transaction's table changes (`table-handle.ts`, `table-changes.ts`), table +SQL (`sql.ts`), the table handle and a transaction's table changes (`table-handle.ts`, `table-changes.ts`), live +queries and their re-run in a commit step (`live-query.ts`), table histories and migrations (`table.ts`, `table-evolve.ts`, `table-history.ts`, `table-migration.ts`), the outbox (`outbox.ts`), archives (`archive.ts`, `export.ts`, `import.ts`, `applier.ts`, `jsonl.ts`), the store's SQL schema (`schema.ts`) and backends (`backend.ts`). -Tests: `test/store/` (one file per source); the benchmarks in `test/bench/store.ts` (results in -`docs/benchmarks.md`). Examples: `20-tables.ts`, `22-migrations.ts`, `25-export-import.ts`. +Tests: `test/store/` (one file per source); the benchmarks in `test/bench/store.ts` and `test/bench/live-query.ts` +(results in `docs/benchmarks.md`). Examples: `20-tables.ts`, `22-migrations.ts`, `25-export-import.ts`. diff --git a/packages/chord/src/store/live-query.ts b/packages/chord/src/store/live-query.ts new file mode 100644 index 000000000..cfb07aa32 --- /dev/null +++ b/packages/chord/src/store/live-query.ts @@ -0,0 +1,301 @@ +// Live queries (decision "Live queries over an object's own tables"): `self.query(table, q, limit)`, a JSON watchable +// of a query's rows over a table of the object itself. Run once before the object goes live, then again inside the +// commit step of every commit of the object that wrote the table (after its writes, so it sees them) and published in +// that commit's batch as the diff from the previous rows (W1). Only the object writes its tables, so its commits are +// the exact invalidation; a commit writing no table with a live query runs none. + +import type { Context } from "../context/index.ts"; +import { ChordError } from "../data/index.ts"; +import { invalid } from "../data/internal.ts"; +import { diffRevisions, type Op } from "../delta/index.ts"; +import type { JsonValue } from "../json.ts"; +import type { SqliteValue } from "../sqlite/index.ts"; +import { type Batch, handedOut, Source, type WatchStart } from "../state/internal.ts"; +import type { Query } from "./query.ts"; +import type { Statement } from "./sql.ts"; +import type { StagedWrite, Step } from "./step.ts"; +import type { TableDefinition } from "./table.ts"; +import type { TableHandle } from "./table-handle.ts"; +import type { StoreWrite } from "./writes.ts"; + +/** Who a live query belongs to: its object, and where its failures after activation are reported (`query_failed`). */ +export type QueryOwner = { + readonly label: string; + report(error: unknown): void; +}; + +/** + * The rows of one run: in order, and by key with the stored row each was decoded from, which the next run compares + * against to keep an unchanged row the same object. + */ +type Rows = { readonly value: JsonValue[]; readonly byKey: ReadonlyMap }; + +/** A row as a run decoded it, with the stored row it came from. */ +type Decoded = { readonly stored: StoredRow; readonly row: JsonValue }; + +/** One run of a live query in a commit step: its rows, or why it could not read them. */ +type QueryRun = { readonly kind: "rows"; readonly rows: Rows } | { readonly kind: "failed"; readonly error: unknown }; + +/** + * Where a live query stands: before its first run (activation), with rows, failed (its last run threw; its last good + * rows kept for a watch opened meanwhile), or detached (its attempt ended). + */ +type QueryState = + | { readonly kind: "loading" } + | { readonly kind: "rows"; readonly value: JsonValue[] } + | { readonly kind: "failed"; readonly error: unknown; readonly last: JsonValue[] | undefined } + | { readonly kind: "detached"; readonly error: unknown }; + +/** A stored row as the backend returns it. */ +type StoredRow = { readonly [column: string]: SqliteValue }; + +/** + * Refuses what the types cannot (plugin code is bundled without type checks; declaration policy): a query of another + * table than `table`, and a `limit` that is no positive integer (a live query must be bounded: one run per commit + * that writes its table reads at most `limit` rows). + */ +export function checkLiveQuery(table: TableDefinition, q: Query, limit: unknown, label: string): void { + if (q.table !== table) { + throw invalid("args", `${label}: self.query: a query of table "${q.table.name}" over "${table.name}"`); + } + if (!Number.isSafeInteger(limit) || (limit as number) < 1) { + throw invalid("args", `${label}: self.query: a limit is required, a positive integer`); + } +} + +/** The live queries of one object by table (its SQL name), across its attempts: what its commits re-run. */ +export class LiveQueries { + readonly #byTable = new Map>(); + + /** Adds `query` (its attempt declared it): re-run by every commit writing its table from now on. */ + add(query: LiveQuery): void { + const table = query.handle.sql.table; + let queries = this.#byTable.get(table); + if (queries === undefined) { + queries = new Set(); + this.#byTable.set(table, queries); + } + queries.add(query); + } + + /** Removes `query` (its attempt ended). */ + delete(query: LiveQuery): void { + const table = query.handle.sql.table; + const queries = this.#byTable.get(table); + if (queries === undefined) return; + queries.delete(query); + if (queries.size === 0) this.#byTable.delete(table); + } + + /** The live queries over `table`, if any. */ + on(table: string): ReadonlySet | undefined { + return this.#byTable.get(table); + } +} + +/** + * The re-run of the live queries over the tables a transaction writes (`handles`), `undefined` when none has one (the + * common case: a map lookup per written table, nothing more). `context` is the transaction's: the runs count through it. + */ +export function refreshOf(handles: Iterable, context: Context): QueryRefresh | undefined { + let queries: Set | undefined; + for (const handle of handles) { + const on = handle.owner.queries.on(handle.sql.table); + if (on === undefined) continue; + queries ??= new Set(); + for (const query of on) queries.add(query); + } + return queries === undefined ? undefined : new QueryRefresh([...queries], context); +} + +const NO_WRITES: readonly StoreWrite[] = []; + +/** + * The re-run of live queries in one commit: `write` reads them in the commit step after the commit's own writes (the + * last write of the last phase), and `adopt` publishes their rows in the commit's batch; a refused step adopts nothing. + * A transaction stages it as its own change (`Staged`). + */ +export class QueryRefresh { + readonly #queries: readonly LiveQuery[]; + readonly #context: Context; + readonly #runs: QueryRun[] = []; + /** The staged write that reads the queries; it writes nothing and records nothing in the outbox. */ + readonly write: StagedWrite; + + constructor(queries: readonly LiveQuery[], context: Context) { + this.#queries = queries; + this.#context = context; + this.write = { order: "write", apply: (step) => this.#rerun(step) }; + } + + async #rerun(step: Step): Promise { + for (const query of this.#queries) this.#runs.push(await query.rerun(step, this.#context)); + return NO_WRITES; + } + + /** Publishes each query's new rows (or its failure), inside the commit's batch. */ + adopt(): void { + for (const [index, query] of this.#queries.entries()) query.adopt(this.#runs[index]!); + } + + committed(): void {} + + discarded(): void {} +} + +/** What the watches of a failed live query had: no rows, so its next rows replace the value whole. */ +const NO_VALUE: unique symbol = Symbol("no value"); + +const NO_ROWS: ReadonlyMap = new Map(); + +/** + * A live query: the rows of `query` (at most `limit`, in its order) over its object's table `handle`, as a JSON source + * (a state member, a derive source, an `on` source). Its statement is compiled once; each run decodes only the rows + * whose stored columns changed and keeps the others the same objects, so its frame, the diff from the rows its watches + * had, compares unchanged rows by identity (a window shifted by one row: one insert and one removal, not a comparison + * of every row with every other). + */ +export class LiveQuery extends Source { + readonly handle: TableHandle; + readonly #owner: QueryOwner; + readonly #statement: Statement; + #state: QueryState = { kind: "loading" }; + /** The last adopted rows by key, with the stored rows they were decoded from (the next run reuses unchanged ones). */ + #byKey: ReadonlyMap = NO_ROWS; + /** The rows this query's watches have, kept from its adoption to its frame in the batch; `undefined`: not listed. */ + #framed: JsonValue[] | typeof NO_VALUE | undefined; + /** Commits adopted so far: an activation run that a commit overtook keeps the commit's newer rows. */ + #adopted = 0; + /** Set from a failed run until the next good one: a failure episode is reported once. */ + #failing = false; + + constructor(handle: TableHandle, query: Query, limit: number, owner: QueryOwner, batch: Batch) { + super(batch); + this.handle = handle; + this.#owner = owner; + this.#statement = handle.sql.select(query.condition, query.order, undefined, limit); + } + + get value(): JsonValue[] { + const state = this.#state; + if (state.kind === "rows") return state.value; + if (state.kind === "loading") { + throw new ChordError({ code: "not_ready", message: `${this.#owner.label}: a live query read before it ran` }); + } + throw state.error; + } + + /** Its rows, or while failed its failure and last good rows (a detached query throws). */ + override start(): WatchStart { + const state = this.#state; + if (state.kind === "failed") return { kind: "failed", error: state.error, last: state.last }; + return { kind: "value", value: this.value }; + } + + /** + * The first run, before its object goes live (activation, after its tables are prepared): a failure throws and fails + * the activation. Rows a commit adopted meanwhile are newer and stay. + */ + async load(context: Context): Promise { + const adopted = this.#adopted; + this.#count(context); + const { sql, params } = this.#statement; + const stored = await this.handle.store.read((executor) => executor.all(sql, ...params), context); + // Decoded outside the store's read: a row the code cannot read is this query's failure, not the backend's. + const rows = this.#decode(stored, context); + if (this.#adopted !== adopted) return; + this.#byKey = rows.byKey; + this.#state = { kind: "rows", value: rows.value }; + } + + /** A run inside the commit step `step`, after its writes: the rows, or the failure; it never throws. */ + async rerun(step: Step, context: Context): Promise { + try { + this.#count(context); + const { sql, params } = this.#statement; + return { kind: "rows", rows: this.#decode(await step.all(sql, ...params), context) }; + } catch (error) { + return { kind: "failed", error }; + } + } + + /** + * Takes a commit's run, inside its batch: listed for a frame when watched, its dependents marked. A failed run makes + * the query a failed member (as a throwing derive): its watches get one failure call, the owner a report per + * failure episode; the next good run replaces the value whole. + */ + adopt(run: QueryRun): void { + const state = this.#state; + if (state.kind === "detached") return; + this.#adopted++; + if (this.watched && this.#framed === undefined) this.#framed = state.kind === "rows" ? state.value : NO_VALUE; + if (run.kind === "rows") { + this.#failing = false; + this.#byKey = run.rows.byKey; + this.#state = { kind: "rows", value: run.rows.value }; + } else { + if (!this.#failing) this.#owner.report(handedOut(run.error)); + this.#failing = true; + const last = state.kind === "rows" ? state.value : state.kind === "failed" ? state.last : undefined; + this.#state = { kind: "failed", error: run.error, last }; + } + this.changed(); + } + + /** Delivers the frame: the diff from the rows its watches had, or one failure call when they had rows. */ + deliver(context: Context): void { + const before = this.#framed; + this.#framed = undefined; + const state = this.#state; + if (state.kind === "rows") { + // Listed by `adopt`, so `before` was kept (a detach in the batch leaves no rows). + const after = state.value; + const ops: readonly Op[] = before === NO_VALUE ? [["r", after]] : diffRevisions(before!, after); + if (ops.length > 0) this.publish(after, ops, false, context); + return; + } + // Still failed since the last call: nothing new for the watches. + if (state.kind === "failed" && before !== NO_VALUE) this.failWatches(state.error, context); + } + + /** Its attempt ended: no commit re-runs it, reads throw `error` (the object's), its watches end `ended`. */ + detach(error: Error): void { + this.handle.owner.queries.delete(this); + this.#state = { kind: "detached", error }; + this.#framed = undefined; + this.endWatches(); + } + + /** One run, counted where it happens (the cost guard: a commit not writing the table runs none). */ + #count(context: Context): void { + context.telemetry.metrics?.counter("chord.query.runs").add(1); + } + + /** + * The rows of `stored`: a row whose stored columns are all the same as in the last adopted run is that run's object + * (one comparison per physical column), any other is decoded. + */ + #decode(stored: readonly StoredRow[], context: Context): Rows { + const { key, rows } = this.handle.definition; + const value: JsonValue[] = []; + const byKey = new Map(); + for (const each of stored) { + const id = key.length === 1 ? each[key[0]!] : JSON.stringify(key.map((column) => each[column])); + const before = this.#byKey.get(id); + const row = + before !== undefined && sameStored(before.stored, each, rows.physical) + ? before.row + : // A table's rows hold its columns' values: scalars and JSON. + (rows.decode(each, context) as JsonValue); + value.push(row); + byKey.set(id, { stored: each, row }); + } + return { value, byKey }; + } +} + +/** Whether two stored rows have the same value in every physical column (SQLite values compare by `===`). */ +function sameStored(a: StoredRow, b: StoredRow, physical: readonly string[]): boolean { + for (const column of physical) if (a[column] !== b[column]) return false; + return true; +} diff --git a/packages/chord/src/store/table-handle.ts b/packages/chord/src/store/table-handle.ts index c5027d795..2c2ae4412 100644 --- a/packages/chord/src/store/table-handle.ts +++ b/packages/chord/src/store/table-handle.ts @@ -7,6 +7,7 @@ import { ChordError } from "../data/index.ts"; import { invalid } from "../data/internal.ts"; import type { SqliteValue } from "../sqlite/index.ts"; import type { Columns, RowOf } from "./columns.ts"; +import type { LiveQueries } from "./live-query.ts"; import { decodeCursor, encodeCursor, type Query } from "./query.ts"; import { TableSql } from "./sql.ts"; import type { ChordStore } from "./store.ts"; @@ -36,11 +37,16 @@ export interface Table { /** The handle type of table definition `D`: what `self.table(D)` returns. */ export type TableOf = D extends TableDefinition ? Table : never; -/** Who owns a table handle: its object (whose transactions alone write it), liveness check and live failure. */ +/** + * Who owns a table handle: its object (whose transactions alone write it), its live queries, liveness check and live + * failure. + */ export type TableOwner = { readonly label: string; /** The object's documents: a transaction of another object refuses the handle. */ readonly scope: object; + /** The object's live queries, across its attempts: what a commit writing the table re-runs. */ + readonly queries: LiveQueries; /** Throws `released` once the object ended. */ assertLive(): void; /** A row the code cannot read (a `json(...)` cell, slice 6a §9) fails the owner `broken` (§1.4). */ diff --git a/packages/chord/test/bench/live-query.ts b/packages/chord/test/bench/live-query.ts new file mode 100644 index 000000000..1ac5014d5 --- /dev/null +++ b/packages/chord/test/bench/live-query.ts @@ -0,0 +1,92 @@ +// Benchmark of live queries (decision "Live queries over an object's own tables"): the cost of one commit that writes +// a table with a live query of its newest 100 rows (a chat's window: a short text per row), against the same commit +// without the query, with the query unwatched (no diff) and a commit writing another table. Each step appends one row +// (the window shifts by one) in one transaction; the value is watched, as a client's hold watches it, except in the +// unwatched case. Median and p95 per commit over several bursts, on a SQLite `:memory:` store. +// Run by hand from packages/chord (not in CI): +// node --conditions=source --experimental-strip-types --expose-gc test/bench/live-query.ts +import { performance } from "node:perf_hooks"; +import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts"; +import { arr, int, local, method, num, obj, str } from "../../src/contracts/index.ts"; +import { createHost, defineFacet } from "../../src/host/index.ts"; +import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts"; +import { constant } from "../../src/state/index.ts"; +import { defineTable, openStore, query, sqliteBackend } from "../../src/store/index.ts"; + +const STEPS = 2_000; +const RUNS = 7; +const WINDOW = 100; +const TEXT = "a message of about the length people type in a chat, give or take a few words"; + +const Messages = defineTable("messages", { + version: 1, + columns: { id: int(), author: str(), text: str(), at: num() }, + key: ["id"], +}); +const Other = defineTable("other", { version: 1, columns: { id: int(), text: str() }, key: ["id"] }); +const NEWEST = query(Messages).orderBy((message) => message.id, "desc"); + +const Message = obj({ id: int(), author: str(), text: str(), at: num() }); +const Room = local("bench.room", { + version: 1, + params: {}, + members: { messages: arr(Message), post: method(int()), other: method(int()) }, +}); + +/** + * What a burst measures: a commit of `Messages` with the live query watched, unwatched, or without it, or a commit of + * `Other` beside the watched query. + */ +type Kind = "no query" | "query" | "unwatched" | "other table"; + +/** Per-commit times (ms) of `RUNS` bursts of `STEPS` commits of `kind`, after one warm-up burst. */ +async function bursts(kind: Kind): Promise<{ readonly median: number; readonly p95: number }> { + const times: number[] = []; + for (let run = -1; run < RUNS; run++) { + const store = await openStore(sqliteBackend(await openNodeSqliteDatabase(":memory:")), context); + const host = createHost({ store, onError: () => {} }); + const facet = defineFacet((h) => + h.serve(Room, (_, self) => { + const messages = self.table(Messages); + const other = self.table(Other); + return { + messages: kind === "no query" ? constant([]) : self.query(messages, NEWEST, WINDOW), + post: async (id, ctx) => { + await self.tx(async (tx) => tx.put(messages, { id, author: "ada", text: TEXT, at: id }), ctx); + return null; + }, + other: async (id, ctx) => { + await self.tx(async (tx) => tx.put(other, { id, text: TEXT }), ctx); + return null; + }, + }; + }), + ); + await host.load([{ name: "@bench/live-query", facet }], context); + const room = await host.get(Room, {}, context); + // A full window first: every later post shifts it by one row. + for (let id = 1; id <= WINDOW; id++) await room.post(id, context); + const watch = await room.messages.watch(context); + if (kind === "unwatched") await watch.stop(); + else watch.start(async () => {}); + for (let step = 0; step < STEPS; step++) { + const id = WINDOW + 1 + step; + const before = performance.now(); + await (kind === "other table" ? room.other(id, context) : room.post(id, context)); + if (run >= 0) times.push(performance.now() - before); + } + await watch.stop(); + room.release(); + await host.close(context); + await store.close(context); + } + times.sort((a, b) => a - b); + return { median: times[Math.floor(times.length / 2)]!, p95: times[Math.floor(times.length * 0.95)]! }; +} + +for (const kind of ["no query", "query", "unwatched", "other table"] as const) { + const { median, p95 } = await bursts(kind); + console.log( + `${kind.padEnd(12)} one-row commit through a call: ${(median * 1000).toFixed(1)} µs median, ${(p95 * 1000).toFixed(1)} µs p95`, + ); +} diff --git a/packages/chord/test/bench/store.ts b/packages/chord/test/bench/store.ts index 3bb6f59f2..e4b1f4362 100644 --- a/packages/chord/test/bench/store.ts +++ b/packages/chord/test/bench/store.ts @@ -23,6 +23,7 @@ import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts"; import { sqliteBackend } from "../../src/store/backend.ts"; import { changeDocument, createDocument } from "../../src/store/engine.ts"; import { lt } from "../../src/store/expression.ts"; +import { LiveQueries } from "../../src/store/live-query.ts"; import { query } from "../../src/store/query.ts"; import { TableSql, tableSqlName } from "../../src/store/sql.ts"; import type { StagedWrite } from "../../src/store/step.ts"; @@ -336,6 +337,7 @@ async function filledEdits(count: number, body = VALUE.body, definition: typeof const handle = new TableHandle(definition, table, store, { label: "bench", scope: {}, + queries: new LiveQueries(), assertLive: () => {}, fail: () => {}, }); diff --git a/packages/chord/test/examples/59-chat/plugins/chat/worker.ts b/packages/chord/test/examples/59-chat/plugins/chat/worker.ts index 90d3e02a6..3561a2c60 100644 --- a/packages/chord/test/examples/59-chat/plugins/chat/worker.ts +++ b/packages/chord/test/examples/59-chat/plugins/chat/worker.ts @@ -1,7 +1,7 @@ // The chat plugin on a channel's worker: the room's messages in a table of the worker's store (every message kept), -// written and read by one local object. Tables have no watch, so the object keeps the live window itself: the newest -// 100 messages, read at activation and appended to after each post (it is the only writer); older ones are read by -// page on request. Each principal's `Room` posts as them and shares files through the kit's `Files` of this worker. +// written and read by one local object. The newest 100 are live through a live query (`self.query`: read before the +// object goes live, read again after every commit writing the table); older ones are read by page on request. Each +// principal's `Room` posts as them and shares files through the kit's `Files` of this worker. import { Attachment, @@ -33,7 +33,7 @@ const Messages = defineTable("chat.messages", { /** The messages, newest first. */ const NEWEST = query(Messages).orderBy((message) => message.id, "desc"); -/** How many messages the live window and a page of history hold. */ +/** How many messages the live query and a page of history hold. */ const PAGE = 100; const Store = local("chat.store", { @@ -49,24 +49,18 @@ const Store = local("chat.store", { export default defineFacet((host) => { host.serve(Store, (_params, self) => { const messages = self.table(Messages); - // The newest messages (oldest first) and the next message's number: read once, then kept here. - const window = self.writable([]); - let next = 1; - self.open( - async (ctx) => { - const { items } = await messages.page(NEWEST, PAGE, ctx); - window.set([...items].reverse()); - next = (items[0]?.id ?? 0) + 1; - }, - () => {}, - ); + // The newest messages, live. + const newest = self.query(messages, NEWEST, PAGE); return { - messages: self.derive([window], ([value]) => value), + // Oldest first, as the room shows them. + messages: self.derive([newest], ([rows]) => [...rows].reverse()), add: async ({ author, text, file }, ctx) => { - const message = { id: next++, author, text, at: Date.now(), ...(file === undefined ? {} : { file }) }; - await self.tx(async (tx) => tx.put(messages, message), ctx); - // In number order: two posts may commit in either order. - window.set([...window.value, message].sort((a, b) => a.id - b.id).slice(-PAGE)); + await self.tx(async (tx) => { + // The number after the newest message's: the store's transactions run one after another. + const { items: last } = await tx.page(messages, NEWEST, 1); + const id = (last[0]?.id ?? 0) + 1; + tx.put(messages, { id, author, text, at: Date.now(), ...(file === undefined ? {} : { file }) }); + }, ctx); return null; }, history: async ({ before }, ctx) => { diff --git a/packages/chord/test/examples/61-tutorial/plugins/lists/worker.ts b/packages/chord/test/examples/61-tutorial/plugins/lists/worker.ts index 57acd38e1..8def80ac4 100644 --- a/packages/chord/test/examples/61-tutorial/plugins/lists/worker.ts +++ b/packages/chord/test/examples/61-tutorial/plugins/lists/worker.ts @@ -1,7 +1,7 @@ // The lists plugin on a list's worker. The todos live in a table of the worker's store (every one kept), written and -// read by one local object, `Store`. A table has no watch, so the store keeps a live window itself: the newest 50 -// todos, read when it goes live and changed after each write (it is the only writer). Each principal's `Todos` view -// adds todos as that principal, and gives the owner `clear`. The store refuses a todo without text with Chord's +// read by one local object, `Store`, which shows the newest 50 live with a live query (`self.query`): Chord runs it +// before the store goes live and again after every commit that writes the table. Each principal's `Todos` view adds +// todos as that principal, and gives the owner `clear`. The store refuses a todo without text with Chord's // `invalid/args`. import { @@ -32,7 +32,7 @@ const Items = defineTable("todo.items", { /** The newest todos first. */ const NEWEST = query(Items).orderBy((item) => item.id, "desc"); -/** How many todos the live window holds. */ +/** How many todos the live query holds. */ const WINDOW = 50; /** The list's todos, once per worker: empty params, so it lives as long as the worker and may keep data in its store. */ @@ -50,36 +50,27 @@ const Store = local("todo.store", { export default defineFacet((host) => { host.serve(Store, (_params, self) => { const items = self.table(Items); - // The newest todos, oldest first, and the next todo's number: read once when the store goes live, then kept here. - const window = self.writable([]); - let next = 1; - self.open( - async (ctx) => { - const { items: newest } = await items.page(NEWEST, WINDOW, ctx); - window.set([...newest].reverse()); - next = (newest[0]?.id ?? 0) + 1; - }, - () => {}, - ); + // The newest todos, live: read before the store goes live, read again after every commit writing the table. + const newest = self.query(items, NEWEST, WINDOW); return { - items: self.derive([window], ([value]) => value), + // Oldest first, as the list shows them. + items: self.derive([newest], ([rows]) => [...rows].reverse()), add: async ({ text, by }, ctx) => { // Refused as Chord refuses arguments that fail their schema: the caller gets `invalid/args`. if (text === "") throw invalid("args", "a todo needs text"); - const item = { id: next++, text, done: false, by, at: Date.now() }; - await self.tx(async (tx) => tx.put(items, item), ctx); - // In number order: two adds may commit in either order. - window.set([...window.value, item].sort((a, b) => a.id - b.id).slice(-WINDOW)); + await self.tx(async (tx) => { + // The number after the newest todo's: the store's transactions run one after another. + const { items: last } = await tx.page(items, NEWEST, 1); + tx.put(items, { id: (last[0]?.id ?? 0) + 1, text, done: false, by, at: Date.now() }); + }, ctx); return null; }, tick: async ({ id, done }, ctx) => { await self.tx(async (tx) => tx.updateWhere(items, (item) => eq(item.id, id), { done }), ctx); - window.set(window.value.map((item) => (item.id === id ? { ...item, done } : item))); return null; }, clear: async (ctx) => { await self.tx(async (tx) => tx.deleteWhere(items, (item) => eq(item.done, true)), ctx); - window.set(window.value.filter((item) => !item.done)); return null; }, }; diff --git a/packages/chord/test/store/live-query.test.ts b/packages/chord/test/store/live-query.test.ts new file mode 100644 index 000000000..7ba6c2687 --- /dev/null +++ b/packages/chord/test/store/live-query.test.ts @@ -0,0 +1,335 @@ +// Tests of live queries (decision "Live queries over an object's own tables"): `self.query(table, q, limit)` runs +// before its object goes live, again in the commit step of every commit of the object that writes its table, and +// publishes the diff in that commit's batch; a commit writing another table runs none; a reload swaps the object; +// the limit is required; a failing run is a failed member; an ended attempt's query is collected. + +import { describe, expect, it } from "vitest"; +import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts"; +import { arr, int, local, method, obj, opt, str } from "../../src/contracts/index.ts"; +import type { ChordError } from "../../src/data/index.ts"; +import { createHost, defineFacet, type FacetHost, type HostError, type StoreSelf } from "../../src/host/index.ts"; +import { eq } from "../../src/store/expression.ts"; +import { json } from "../../src/store/json-column.ts"; +import { LiveQuery } from "../../src/store/live-query.ts"; +import { query } from "../../src/store/query.ts"; +import type { ChordStore } from "../../src/store/store.ts"; +import { defineTable } from "../../src/store/table.ts"; +import { + changesOf, + delivered, + expectCollected, + flushed, + framesOf, + instancesOf, + loaded, + memoryStore, + metered, + rejection, +} from "../support.ts"; + +const Items = defineTable("items", { + version: 1, + columns: { id: int(), text: str(), meta: opt(json(obj({}), { version: 1 })) }, + key: ["id"], +}); +const Other = defineTable("other", { version: 1, columns: { id: int() }, key: ["id"] }); + +/** The newest items first. */ +const NEWEST = query(Items).orderBy((item) => item.id, "desc"); + +const Item = obj({ id: int(), text: str(), meta: opt(obj({})) }); + +const Board = local("app.board", { + version: 1, + params: {}, + members: { + items: arr(Item), + put: method(obj({ id: int(), text: str(), meta: opt(obj({})) })), + rename: method(obj({ id: int(), text: str() })), + drop: method(int()), + other: method(int()), + }, +}); + +/** Live queries collected weakly as they run at activation (GC tests). */ +const queries = instancesOf(LiveQuery, "load"); + +/** Serves `Board` with a live query of the newest `limit` items; `declare` runs first in its factory. */ +function serveBoard(h: FacetHost, limit: number, declare: (self: StoreSelf) => void = () => {}): void { + h.serve(Board, (_, self) => { + declare(self); + const items = self.table(Items); + const other = self.table(Other); + return { + items: self.query(items, NEWEST, limit), + put: async (row, ctx) => { + await self.tx(async (tx) => tx.put(items, row), ctx); + return null; + }, + rename: async ({ id, text }, ctx) => { + await self.tx(async (tx) => tx.updateWhere(items, (item) => eq(item.id, id), { text }), ctx); + return null; + }, + drop: async (id, ctx) => { + await self.tx(async (tx) => tx.deleteWhere(items, (item) => eq(item.id, id)), ctx); + return null; + }, + other: async (id, ctx) => { + await self.tx(async (tx) => tx.put(other, { id }), ctx); + return null; + }, + }; + }); +} + +/** A host serving `Board` on `store`, metered, and a reference to the board. */ +async function board(store: ChordStore, limit = 3) { + const meter = await metered(); + // The board lives as long as its host: it activates during `load`, under the load's context. + const { host, errors } = await loaded((h) => serveBoard(h, limit), { + store, + telemetry: meter.telemetry, + context: meter.context, + }); + const reference = await host.get(Board, {}, meter.context); + return { ...meter, host, errors, board: reference }; +} + +/** The ids of `rows`. */ +function ids(rows: readonly { readonly id: number }[]): number[] { + return rows.map(({ id }) => id); +} + +describe("activation", () => { + it("runs the query before the object goes live, over the rows already stored", async () => { + const store = await memoryStore(); + const first = await board(store); + for (const id of [1, 2]) await first.board.put({ id, text: `t${id}` }, context); + await first.host.close(context); + const { board: again, host, counts } = await board(store); + expect(again.items.value).toEqual([ + { id: 2, text: "t2" }, + { id: 1, text: "t1" }, + ]); + expect(counts()["chord.query.runs"]).toBe(1); + await host.close(context); + await store.close(context); + }); + + it("a run that fails at activation fails the activation broken", async () => { + const store = await memoryStore(); + const first = await board(store); + await first.board.put({ id: 1, text: "t1", meta: {} }, context); + await first.host.close(context); + await store.read((executor) => executor.run(`UPDATE "app.board:items" SET "meta$v" = 9`), context); + const errors: HostError[] = []; + const host = createHost({ store, onError: (error) => errors.push(error) }); + await host.load([{ name: "@test/app", facet: defineFacet((h) => serveBoard(h, 3)) }], context); + const refused = (await rejection(host.get(Board, {}, context))) as ChordError; + expect(refused.message).toContain("stored at version 9"); + expect(errors.map((error) => [error.kind, "failure" in error ? error.failure.kind : ""])).toEqual([ + ["activation_failed", "broken"], + ]); + await host.close(context); + await store.close(context); + }); +}); + +describe("publication", () => { + it("an insert, an update and a delete reach a watch as ops, in the commit's batch", async () => { + const store = await memoryStore(); + const { board: b, host } = await board(store); + const watch = await b.items.watch(context); + const frames = framesOf(watch); + await b.put({ id: 1, text: "a" }, context); + // Published in the commit's batch: the value is current once the call returned. + expect(ids(b.items.value)).toEqual([1]); + await b.put({ id: 2, text: "b" }, context); + await b.rename({ id: 1, text: "A" }, context); + await b.drop(2, context); + await delivered(); + // Splices of the rows at the root (`["p", path, at, deleted, inserted]`) and a set inside one row. + expect(frames.map(([, ops]) => ops)).toEqual([ + [["p", [], 0, 0, [{ id: 1, text: "a" }]]], + [["p", [], 0, 0, [{ id: 2, text: "b" }]]], + [["s", [1, "text"], "A"]], + [["p", [], 0, 1, []]], + ]); + expect(b.items.value).toEqual([{ id: 1, text: "A" }]); + watch.stop(); + await host.close(context); + await store.close(context); + }); + + it("keeps the window at its limit: a newer row pushes the oldest out", async () => { + const store = await memoryStore(); + const { board: b, host } = await board(store, 2); + for (const id of [1, 2]) await b.put({ id, text: `t${id}` }, context); + const watch = await b.items.watch(context); + const frames = framesOf(watch); + await b.put({ id: 3, text: "t3" }, context); + await delivered(); + expect(ids(b.items.value)).toEqual([3, 2]); + expect(frames.map(([, ops]) => ops)).toEqual([ + [ + ["p", [], 0, 0, [{ id: 3, text: "t3" }]], + ["p", [], 2, 1, []], + ], + ]); + watch.stop(); + await host.close(context); + await store.close(context); + }); + + it("keeps an unchanged row the same object across runs, a changed one new", async () => { + const store = await memoryStore(); + const { board: b, host } = await board(store); + for (const id of [1, 2]) await b.put({ id, text: `t${id}`, meta: {} }, context); + const [two, one] = b.items.value; + await b.rename({ id: 2, text: "T2" }, context); + const [twoAfter, oneAfter] = b.items.value; + expect([oneAfter === one, twoAfter === two]).toEqual([true, false]); + expect(twoAfter).toEqual({ id: 2, text: "T2", meta: {} }); + await host.close(context); + await store.close(context); + }); + + it("a commit writing another table runs no query and publishes nothing (cost guard)", async () => { + const store = await memoryStore(); + const { board: b, host, context: counting, counts } = await board(store); + const watch = await b.items.watch(context); + const calls = changesOf(watch); + const statements = (): number => counts()["chord.store.statements"] ?? 0; + const runs = (): number => counts()["chord.query.runs"] ?? 0; + let before = [statements(), runs()]; + await b.other(1, counting); + const other = [statements() - before[0]!, runs() - before[1]!]; + before = [statements(), runs()]; + await b.put({ id: 1, text: "a" }, counting); + const touching = [statements() - before[0]!, runs() - before[1]!]; + await delivered(); + // The same one-row put: the touching commit costs one query, one statement more. + expect(other[1]).toBe(0); + expect(touching).toEqual([other[0]! + 1, 1]); + expect(calls).toHaveLength(1); + watch.stop(); + await host.close(context); + await store.close(context); + }); +}); + +describe("reload", () => { + it("swaps the object (no restart): the new attempt's query runs at its activation and takes over", async () => { + const store = await memoryStore(); + const { board: b, host, context: counting, counts } = await board(store); + for (const id of [1, 2, 3]) await b.put({ id, text: `t${id}` }, context); + const report = await host.reload([{ name: "@test/app", facet: defineFacet((h) => serveBoard(h, 2)) }], context); + expect(report.objects).toEqual([{ object: "app.board", kind: "swapped" }]); + await flushed(); + expect(ids(b.items.value)).toEqual([3, 2]); + const runs = counts()["chord.query.runs"] ?? 0; + await b.put({ id: 4, text: "t4" }, counting); + expect(ids(b.items.value)).toEqual([4, 3]); + // Only the new attempt's query runs: the old one left with its attempt. + expect((counts()["chord.query.runs"] ?? 0) - runs).toBe(1); + await host.close(context); + await store.close(context); + }); +}); + +describe("declaration", () => { + const refusals: [string, (self: StoreSelf) => void, string][] = [ + ["no limit", (self) => void self.query(self.table(Items), NEWEST, undefined as never), "a limit is required"], + ["a limit of 0", (self) => void self.query(self.table(Items), NEWEST, 0), "a limit is required"], + ["a fractional limit", (self) => void self.query(self.table(Items), NEWEST, 2.5), "a limit is required"], + [ + "a query of another table", + (self) => void self.query(self.table(Other), NEWEST as never, 3), + 'a query of table "items" over "other"', + ], + [ + "a table this factory did not declare", + (self) => void self.query({} as never, NEWEST, 3), + "the table is not one this factory declared", + ], + ]; + + for (const [what, declare, message] of refusals) { + it(`refuses ${what}`, async () => { + const store = await memoryStore(); + const { host } = await loaded( + (h) => + h.serve(Board, (_, self) => { + declare(self); + throw new Error("unreachable"); + }), + { store }, + ); + const refused = (await rejection(host.get(Board, {}, context))) as ChordError; + expect(refused.message).toContain(message); + await host.close(context); + await store.close(context); + }); + } + + it("staging refuses a query without a valid limit: the plugin keeps its old code", async () => { + const store = await memoryStore(); + const { board: b, host } = await board(store); + const report = await host.reload([{ name: "@test/app", facet: defineFacet((h) => serveBoard(h, 0)) }], context); + expect(report.failed.map(({ error }) => (error as Error).message)).toEqual([ + expect.stringContaining("a limit is required"), + ]); + expect(b.state.value).toEqual({ kind: "live" }); + await host.close(context); + await store.close(context); + }); +}); + +describe("failure", () => { + it("a failing run is a failed member, reported once; the next good run heals it", async () => { + const store = await memoryStore(); + const { board: b, host, errors } = await board(store); + await b.put({ id: 1, text: "a", meta: {} }, context); + const watch = await b.items.watch(context); + const calls = changesOf(watch); + await store.read((executor) => executor.run(`UPDATE "app.board:items" SET "meta$v" = 9`), context); + await b.put({ id: 2, text: "b" }, context); + await b.put({ id: 3, text: "c" }, context); + expect(() => b.items.value).toThrow("stored at version 9"); + await b.drop(1, context); + await delivered(); + expect(calls.map((call) => call.kind)).toEqual(["failed", "value"]); + expect(calls[1]).toMatchObject({ + ops: [ + [ + "r", + [ + { id: 3, text: "c" }, + { id: 2, text: "b" }, + ], + ], + ], + }); + expect(errors.map((error) => [error.kind, "table" in error ? error.table : ""])).toEqual([ + ["query_failed", "items"], + ]); + // The object stays live: only the member failed. + expect(b.state.value).toEqual({ kind: "live" }); + watch.stop(); + await host.close(context); + await store.close(context); + }); +}); + +describe("memory", () => { + it("collects a swapped-out attempt's query: no commit re-runs it", async () => { + const store = await memoryStore(); + const { host } = await board(store); + const from = queries.length - 1; + await host.reload([{ name: "@test/app", facet: defineFacet((h) => serveBoard(h, 2)) }], context); + await flushed(); + await expectCollected(queries[from]!); + await host.close(context); + await store.close(context); + }); +}); diff --git a/packages/chord/test/support.ts b/packages/chord/test/support.ts index 4cc0ba3cc..4958bcd63 100644 --- a/packages/chord/test/support.ts +++ b/packages/chord/test/support.ts @@ -63,6 +63,7 @@ import type { ArchivePart } from "../src/store/archive.ts"; import { sqliteBackend } from "../src/store/backend.ts"; import { changeDocument, createDocument, retireDocument } from "../src/store/engine.ts"; import { json } from "../src/store/json-column.ts"; +import { LiveQueries } from "../src/store/live-query.ts"; import type { StagedWrite } from "../src/store/step.ts"; import { type ChordStore, openStore, type Store } from "../src/store/store.ts"; import { defineTable, type TableDefinition } from "../src/store/table.ts"; @@ -328,7 +329,14 @@ export function holdNextCommit(store: ChordStore): { readonly entered: Promise {}, fail: (error) => void failures.push(error), failures }; + return { + label: "app.store", + scope: {}, + queries: new LiveQueries(), + assertLive: () => {}, + fail: (error) => void failures.push(error), + failures, + }; } /** Creates `definition` (and its indexes) in `store` as `app.store:` and returns a handle to it. */