API index and search · Build metadata
Source snapshot
packages/normalized/src/sync.ts
1// NormalizedSync — the client-side cross-query refcount + GC layer of the normalized2// local-first path (NORMALIZED-CHANGES-DESIGN.md §5), extended with the client column3// UNION for projection support (PROJECTION-SUPPORT-DESIGN.md §4). It sits between the4// per-query normalized streams (each an independent, validated `NormalizedOp` batch from5// the server, via `@rindle/remote`) and the client's one shared normalized store (the wasm6// `Db`'s base tables). The server already de-duplicates *intra-query* multi-path (§4.2);7// this layer does the *cross-query* refcount the server deliberately does NOT — so a base8// row shared by N queries lands in the local `Db` exactly once and is GC'd only when the9// last query stops referencing it.10//11// It is the half the design calls out as "NO CVR" ([[no-cvr-client-refcounts]]): the12// refcount is per-row on the *client*, keyed by `(table, pk)`, with each query's footprint13// tracked as a membership set. The query-id rides at the subscription envelope, never on a14// row (§5.1) — so this layer is fed per-query (`applyBatch(queryId, ops)`).15//16// **Projection (the client column union, PROJECTION-SUPPORT-DESIGN.md §4).** A *projected*17// query streams rows positional over its OWN (narrower) schema; the client scatters those18// cells into the shared base-table positions via the query's registered column map19// (`registerProjection`) and leaves the rest ABSENT (the `undefined` marshal token, §5.3).20// The same map also absorbs the reverse skew — a server WIDER than the client (an expanded21// table mid `expand-then-contract` migration): a wire column the client doesn't have maps to a22// DROP sentinel (`-1`) and its cell is discarded, so an old client keeps working against a23// newer server. (Mapping is BY NAME, so neither a narrower nor a wider server mis-aligns.)24// The base row a query group holds is the **union over all referencing queries'25// projections** (§2): widening (`add`/`edit` brings new columns) column-merges them in;26// narrowing (`dropQuery`, or a query's `remove` of a still-referenced row) recomputes the27// union from the *remaining* queries and resets dropped columns to ABSENT — the client-side28// equivalent of Zero's `remove-keys` (§4.2). A `'*'` query (no registered projection)29// contributes all columns, so any row it references is full and present everywhere: the30// union never introduces ABSENT and behavior is byte-identical to before projection (§7).31//32// Output: a NET `Mutation[]` to apply to the wasm `Db` in ONE `write()` transaction (the33// design's "stage into one txn, commit once"). This module is pure logic — it never34// touches the `Db`; the `NormalizedStore` (Slice 5) feeds the mutations to it.3536import type { Mutation, NormalizedOp, QueryId, WireValue } from "@rindle/client";3738// `NormalizedOp` (the table-tagged, path-free wire op, §3) lives in `@rindle/client` so both the39// protocol (`@rindle/remote`) and this sync layer share one type. Re-exported here for callers.40export type { NormalizedOp } from "@rindle/client";4142/** Per-table primary-key column indices (into a positional row) — how `(table, pk)` is keyed. */43export type PkCols = Record<string, number[]>;4445/** Per-table FULL column count — the width of a base/union row. Needed to scatter a projected46 * (narrower) wire row into the shared positional layout (PROJECTION-SUPPORT-DESIGN.md §5.3). */47export type ColCounts = Record<string, number>;4849/**50 * The Absent token on the JS side (PROJECTION-SUPPORT-DESIGN.md §5.3): an absent cell in a51 * positional UNION row. The wasm marshal maps `undefined` → `OwnedValue::Absent` (and `null`52 * → present-and-null), so forwarding a union row with `undefined` cells is how the local53 * engine learns a column is not present on a shared row.54 */55const ABSENT = undefined;5657/** A union-row cell: a present wire value, or `ABSENT` (`undefined`) when the column is not58 * present on the shared row. */59type UnionCell = WireValue | undefined;6061interface BaseEntry {62 table: string;63 /** The shared UNION row (present cells filled; absent cells carry {@link ABSENT}). */64 row: UnionCell[];65 /** The queries currently referencing this row. Its size is the cross-query refcount; its66 * members drive the per-row column union (§4.1). */67 queries: Set<QueryId>;68}6970/** Elementwise equality over positional rows (json columns compare as strings; `ABSENT`71 * compares equal only to `ABSENT`). */72function rowEq(a: UnionCell[], b: UnionCell[]): boolean {73 if (a.length !== b.length) return false;74 for (let i = 0; i < a.length; i++) if (a[i] !== b[i]) return false;75 return true;76}7778export class NormalizedSync {79 private readonly pkCols: PkCols;80 /** Per-table full width, for scattering projected rows. Empty ⇒ projection unsupported81 * (every query must be `'*'`), which is the pre-projection behavior. */82 private readonly colCounts: ColCounts;83 /** `(table, pk)` key → the one shared base entry. */84 private readonly base = new Map<string, BaseEntry>();85 /** queryId → the set of `(table, pk)` keys that query currently footprints. */86 private readonly qfoot = new Map<QueryId, Set<string>>();87 /** queryId → (table → the base ColIds it contributes for that table, in the order its wire88 * rows are positional against — its `required_cols` for the table, §4.1). A query spans89 * multiple base tables (root + related/EXISTS children), each projected independently; a90 * table absent from the inner map ⇒ that query syncs it `'*'` (all columns, full width). */91 private readonly qcols = new Map<QueryId, Map<string, number[]>>();9293 constructor(pkCols: PkCols, colCounts: ColCounts = {}) {94 this.pkCols = pkCols;95 this.colCounts = colCounts;96 }9798 /**99 * Register a query's projection for one base table (PROJECTION-SUPPORT-DESIGN.md §4.1): the100 * base ColIds it contributes for `table`, in the order that table's wire rows are positional101 * against. Omit a table (or never call) for a `'*'` table — it then contributes every column102 * and its rows are full width. Must be called before the query's first batch.103 *104 * A `cols[i] < 0` entry is a DROP sentinel: wire column `i` is an EXPANDED server column the105 * client doesn't have (the server side of an `expand-then-contract` migration), so its cell is106 * discarded rather than scattered. The query still contributes only its real (`>= 0`) ColIds.107 */108 registerProjection(queryId: QueryId, table: string, cols: number[]): void {109 let byTable = this.qcols.get(queryId);110 if (!byTable) {111 byTable = new Map<string, number[]>();112 this.qcols.set(queryId, byTable);113 }114 byTable.set(table, cols.slice());115 }116117 /**118 * Drop a query's projection for one table, reverting it to `'*'` (full presence, rows scattered119 * verbatim). The inverse of {@link registerProjection}; a no-op if none was registered. Needed120 * because `qcols` persists across re-hydrate epochs: if a live subscription's hello narrows from121 * an EXPANDED layout (a `-1`-bearing map) back to an exact full-width one, the stale map must be122 * cleared or it would mis-scatter the now-exact rows. A no-op for an unprojected (`'*'`) table.123 */124 unregisterProjection(queryId: QueryId, table: string): void {125 this.qcols.get(queryId)?.delete(table);126 }127128 /**129 * Register a table's primary-key columns after construction — for a **synthetic aggregate130 * table** (`AGGREGATE-SYNC-DESIGN.md` §3.3): a relationship `count` is synced as a131 * server-authoritative `__agg_*` base table that is not in the client's typed schema, so132 * the backend registers it here (and on the local engine) as queries that use it arrive.133 * Idempotent — re-registering the same table (a second query over the same aggregate)134 * is a no-op. Once registered, its rows refcount/GC exactly like any base table.135 */136 registerTable(table: string, primaryKey: number[]): void {137 this.pkCols[table] = primaryKey;138 }139140 /**141 * The inverse of {@link registerTable} for a synthetic aggregate table whose last142 * referencing query is gone (`AGGREGATE-SYNC-DESIGN.md` §4): drop its primary-key143 * registration so the table is unknown again. A balanced stream has already GC'd its rows144 * at the `1→0` transition (via {@link dropQuery}); defensively this also sweeps any residual145 * base rows + per-query footprint entries for the table, so a later re-registration of the146 * same name starts clean. A no-op for an unregistered table.147 */148 unregisterTable(table: string): void {149 delete this.pkCols[table];150 // A balanced stream GC's the rows at 1→0 (dropQuery); defensively sweep any residual base151 // rows + footprint references for this table (matched on the entry's stored table —152 // separator-agnostic) so a re-register of the same name starts clean.153 const stale = new Set<string>();154 for (const [k, e] of this.base) if (e.table === table) stale.add(k);155 for (const k of stale) this.base.delete(k);156 if (stale.size) for (const foot of this.qfoot.values()) for (const k of stale) foot.delete(k);157 }158159 /**160 * Apply one query's normalized batch (its hydrate snapshot or one transaction's ops) and161 * return the NET base-table mutations to commit to the wasm `Db` in a single transaction.162 * Cross-query refcount + per-query dedup + column union (§4, §5):163 * - `add`: counted into this query's footprint once; on the base `0→1` transition the row164 * enters at this query's projection; on a `1→N` transition the query's columns are merged165 * into the shared union (widen) and an `edit` is forwarded if the union changed.166 * - `remove`: on the base `N→0` transition the row leaves; on `N→M>0` the union is recomputed167 * from the remaining queries and an `edit` narrows the shared row.168 * - `edit`: column-merge this query's cells into the shared union; forward once (idempotent169 * across queries — same source row, same values).170 */171 applyBatch(queryId: QueryId, ops: NormalizedOp[]): Mutation[] {172 const muts: Mutation[] = [];173 const foot = this.footOf(queryId);174 for (const op of ops) {175 if (op.op === "add")176 this.add(queryId, foot, op.table, this.scatter(queryId, op.table, op.row), muts);177 else if (op.op === "remove")178 this.remove(queryId, foot, op.table, this.scatter(queryId, op.table, op.row), muts);179 else this.edit(op.table, this.scatter(queryId, op.table, op.new), muts);180 }181 return muts;182 }183184 /**185 * Re-hydrate one query under a new epoch (§5.3): the server re-sent `queryId`'s whole186 * footprint as seq-0 `add`s. Diff the new footprint against the query's current one —187 * rows only in the old set leave (refcount out, GC/narrow), rows only in the new set enter188 * (refcount in / widen), and an intersecting row whose value changed during the gap is an189 * edit. Only `queryId`'s references move; other queries' counts are untouched. Returns the190 * net mutations to commit.191 */192 rehydrate(queryId: QueryId, snapshot: NormalizedOp[]): Mutation[] {193 const muts: Mutation[] = [];194 const foot = this.footOf(queryId);195 // The new footprint, keyed; a re-hydrate snapshot is all `add`s.196 const next = new Map<string, { table: string; row: UnionCell[] }>();197 for (const op of snapshot) {198 if (op.op !== "add") continue;199 const row = this.scatter(queryId, op.table, op.row);200 next.set(this.key(op.table, row), { table: op.table, row });201 }202 // Rows that left: in the old footprint but not the new one.203 for (const key of [...foot]) {204 if (!next.has(key)) this.removeKey(queryId, foot, key, muts);205 }206 // Rows that entered or changed.207 for (const [key, { table, row }] of next) {208 if (foot.has(key)) {209 // Still referenced — only forward an edit if the union changed during the gap.210 const e = this.base.get(key);211 if (e) this.mergeInto(e, row, table, muts);212 } else {213 this.add(queryId, foot, table, row, muts);214 }215 }216 return muts;217 }218219 /**220 * Drop a query (§5.1): decrement its footprint's refcounts, GC each row at the last221 * reference (or narrow the union if others remain), and forget the query. `O(footprint)`,222 * no per-row scan. Returns the net mutations. (The caller also tells the server to223 * deregister the stream.)224 */225 dropQuery(queryId: QueryId): Mutation[] {226 const muts: Mutation[] = [];227 const foot = this.qfoot.get(queryId);228 if (!foot) return muts;229 for (const key of [...foot]) this.removeKey(queryId, foot, key, muts);230 this.qfoot.delete(queryId);231 this.qcols.delete(queryId);232 return muts;233 }234235 // --- introspection (for tests / the Store) ---------------------------------236237 /** The number of distinct base rows currently synced into the local store. */238 baseSize(): number {239 return this.base.size;240 }241242 /** How many queries currently reference `(table, row)` (0 if absent). The lookup keys by PK,243 * so a projected `row` need only carry its PK columns at the base positions. */244 refCount(table: string, row: UnionCell[]): number {245 return this.base.get(this.key(table, row))?.queries.size ?? 0;246 }247248 /**249 * The synced (server-authoritative) row currently held for `(table, pkCells)`, or250 * `undefined` if no query references it. `pkCells` are the primary-key cells in251 * `primaryKey` order (NOT a full row) — the same key the cross-query refcount uses.252 *253 * The optimistic aggregate overlay (`AGGREGATE-SYNC-DESIGN.md` §4) reads the server's254 * `__agg` count cell through this so it can compute `displayed = server_base ⊕ delta`255 * from the authoritative base rather than re-deriving it from the local engine's head256 * (which already carries the optimistic layer — a torn read).257 */258 baseRow(table: string, pkCells: WireValue[]): (WireValue | undefined)[] | undefined {259 const cols = this.pkCols[table];260 if (!cols) throw new Error(`NormalizedSync: no primary key registered for table ${table}`);261 // Place the pk cells at their schema positions so `key()` reduces this probe to the262 // exact same string it builds for any full row with these key cells.263 const probe: WireValue[] = [];264 cols.forEach((c, j) => (probe[c] = pkCells[j]));265 return this.base.get(this.key(table, probe))?.row;266 }267268 // --- internals -------------------------------------------------------------269270 private footOf(queryId: QueryId): Set<string> {271 let foot = this.qfoot.get(queryId);272 if (!foot) {273 foot = new Set<string>();274 this.qfoot.set(queryId, foot);275 }276 return foot;277 }278279 /** Scatter a query's (possibly narrower) wire row for `table` into the shared full-width280 * positional layout. No registered projection for `(queryId, table)` ⇒ a `'*'` table whose281 * row is already full width (returned as-is). Otherwise allocate a full-width row of282 * `ABSENT` and place `row[i]` at `cols[i]`. */283 private scatter(queryId: QueryId, table: string, row: WireValue[]): UnionCell[] {284 const cols = this.qcols.get(queryId)?.get(table);285 if (!cols) return row; // '*' — already full width, present everywhere286 const width = this.colCounts[table];287 if (width === undefined) {288 throw new Error(`NormalizedSync: no column count registered for projected table ${table}`);289 }290 const out: UnionCell[] = new Array<UnionCell>(width).fill(ABSENT);291 // `cols[i] < 0` ⇒ wire column `i` is an expanded server column the client lacks: drop its292 // cell, leaving the client's narrower row intact. Known columns map by name to their base293 // position, so a server wider than (or reordered vs.) the client still scatters correctly.294 for (let i = 0; i < cols.length; i++) if (cols[i] >= 0) out[cols[i]] = row[i];295 return out;296 }297298 /** Column-merge `incoming`'s present cells over `e.row` (a widen / pure-value change) and299 * forward a single `edit` if the union changed. Both rows are full width. */300 private mergeInto(e: BaseEntry, incoming: UnionCell[], table: string, muts: Mutation[]): void {301 const merged = e.row.map((cur, i) => (incoming[i] !== ABSENT ? incoming[i] : cur));302 if (!rowEq(merged, e.row)) {303 muts.push(mut("edit", table, e.row, merged));304 e.row = merged;305 }306 }307308 private add(queryId: QueryId, foot: Set<string>, table: string, row: UnionCell[], muts: Mutation[]): void {309 const key = this.key(table, row);310 if (foot.has(key)) return; // this query already references it311 foot.add(key);312 const e = this.base.get(key);313 if (e) {314 e.queries.add(queryId); // another query already synced the row → no Db add (would dup-key)315 // …but this query may carry NEW columns (widen) or a NEWER value for a shared column316 // (the gainer half of a server commit that edits the same row on a holder query's317 // stream, CRIT#3). Column-merge and forward an edit at the new union if it changed.318 this.mergeInto(e, row, table, muts);319 } else {320 this.base.set(key, { table, row, queries: new Set([queryId]) });321 muts.push(mut("add", table, undefined, row)); // 0→1322 }323 }324325 private remove(326 queryId: QueryId,327 foot: Set<string>,328 table: string,329 row: UnionCell[],330 muts: Mutation[],331 ): void {332 this.removeKey(queryId, foot, this.key(table, row), muts);333 }334335 private removeKey(queryId: QueryId, foot: Set<string>, key: string, muts: Mutation[]): void {336 if (!foot.has(key)) return; // not referenced by this query337 foot.delete(key);338 const e = this.base.get(key);339 if (!e) return; // defensive: inconsistent stream340 e.queries.delete(queryId);341 if (e.queries.size === 0) {342 this.base.delete(key);343 muts.push(mut("remove", e.table, undefined, e.row)); // N→0344 } else {345 // Others still reference it → recompute the union from the remaining queries and narrow346 // the shared row (reset any column no longer contributed to ABSENT). §4.2.347 const present = this.presentCols(e.queries, e.table);348 if (present === null) return; // a '*' query still holds it → full presence, no narrowing349 const narrowed = e.row.map((cell, i) => (present.has(i) ? cell : ABSENT));350 if (!rowEq(narrowed, e.row)) {351 muts.push(mut("edit", e.table, e.row, narrowed));352 e.row = narrowed;353 }354 }355 }356357 private edit(table: string, next: UnionCell[], muts: Mutation[]): void {358 // PK is stable across an edit, so `next` keys the existing entry.359 const key = this.key(table, next);360 const e = this.base.get(key);361 if (!e) return; // defensive: edit of an untracked row362 this.mergeInto(e, next, table, muts); // column-merge; dedups (no-op) if already applied363 }364365 /** The columns present on a `table` row footprinted by `queries`: the union of their366 * projections for that table. `null` ⇒ all columns present (some referencing query syncs367 * the table `'*'`), so the row is full. */368 private presentCols(queries: Set<QueryId>, table: string): Set<number> | null {369 const out = new Set<number>();370 for (const q of queries) {371 const cols = this.qcols.get(q)?.get(table);372 if (!cols) return null; // a '*' query contributes every column for this table373 for (const c of cols) if (c >= 0) out.add(c); // skip DROP sentinels (expanded server cols)374 }375 return out;376 }377378 private key(table: string, row: UnionCell[]): string {379 const cols = this.pkCols[table];380 if (!cols) throw new Error(`NormalizedSync: no primary key registered for table ${table}`);381 return `${table} ${JSON.stringify(cols.map((i) => row[i]))}`;382 }383}384385/** Build a `Mutation`. A union row may carry `ABSENT` (`undefined`) cells — the wasm `mutate`386 * ABI accepts them as the Absent token (§5.3) — so the row crosses as `WireValue[]` with that387 * understood sentinel. */388function mut(389 op: "add" | "remove" | "edit",390 table: string,391 old: UnionCell[] | undefined,392 row: UnionCell[],393): Mutation {394 if (op === "edit") return { op, table, old: (old as WireValue[]) ?? [], new: row as WireValue[] };395 return { op, table, row: row as WireValue[] } as Mutation;396}397