From d67cb7b984dfbab7204c599052c1446f7d526edf Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 22 Aug 2026 07:51:32 +0000 Subject: [PATCH 1/2] feat(metadata): watch(_, since) replays from sys_metadata_history; write down what a bare watch() owes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Invariant 6 (resumability) had one written half and one inherited half. Face 1 — the written MUST. `SysMetadataRepository.watch()` read `since` only as a drop filter on live events, so an event that had already committed was unreachable however low `since` was set, even though the repository holds a durable per-org `event_seq` log and already reads it org-wide in `nextEventSeq()`. A numeric `since` now replays every logged event with `seq > since` before any live event, through that same query and the row→event mapping extracted out of `history()`. The live listener registers before the durable read starts and a `delivered` set of `seq` closes the replay→live seam, so an event committing during the read is delivered exactly once. A read failure is raised to the consumer rather than downgraded to a silent live-only tail. Face 2 — the inherited half, now written. `repository.ts` spoke only of `seq > since`; with no `since` there is no such set, so "no `since` replays everything" lived only as `InMemoryRepository`'s implementation and the shared contract suite silently leaned on it. Invariant 6 now states the floor: a bare `watch(filter)` is owed live events only, an implementation MAY additionally deliver what already committed, and a caller MUST NOT rely on it. The suite's filter clause is rewritten to that floor — subscription first, writes after — because its old shape asserted a replay the contract does not owe. `declaredDivergences: { resumableWatch: '#10842' }` is deleted; the pin it swapped in went red when replay landed, which is the mechanism working. Fixes #10842 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_019yDEhPBC3tcGkW9bkce1HM --- packages/metadata-core/src/contract-suite.ts | 147 ++++++------ packages/metadata-core/src/repository.ts | 34 ++- .../sys-metadata-repository.contract.test.ts | 218 +++++++++++++++++- .../src/sys-metadata-repository.ts | 176 ++++++++++++-- 4 files changed, 474 insertions(+), 101 deletions(-) diff --git a/packages/metadata-core/src/contract-suite.ts b/packages/metadata-core/src/contract-suite.ts index a5f87fd2ec..7e648496f3 100644 --- a/packages/metadata-core/src/contract-suite.ts +++ b/packages/metadata-core/src/contract-suite.ts @@ -14,7 +14,8 @@ * 3. Optimistic locking (ConflictError) * 4. Canonical hashing (hash === hashSpec(body)) * 5. Event ordering (monotonic seq, no gaps) - * 6. Resumability (watch with `since` replays) + * 6. Resumability (a numeric `since` replays; a bare `watch(filter)` is + * owed live events only) * 7. Tombstones (delete event emitted, get returns null) * * Two knobs, both narrow on purpose. `primaryType` / `secondaryType` move the @@ -75,13 +76,22 @@ export interface ContractSuiteOptions { /** @see ContractSuiteOptions.declaredDivergences */ export interface DeclaredDivergences { /** - * **Invariant 6 (resumability).** The implementation's `watch()` delivers - * live events only; it never replays from its durable log, so neither - * `watch(filter, since)` nor a `watch(filter)` opened after a write can - * surface an event that already committed. + * **Invariant 6, first half (a numeric `since` replays).** The + * implementation's `watch(filter, since)` never replays from its durable + * log, so an event that committed before the subscription cannot be + * surfaced however high `since` is set. * - * Value is the tracking issue, e.g. `'#10842'` — `SysMetadataRepository`, - * the only declaration today. + * ⚠️ Scoped to the NUMERIC-`since` half on purpose. A `watch(filter)` with + * no `since` that surfaces nothing already committed is not a divergence at + * all — invariant 6 owes such a subscriber live events only, and replaying + * for it is a MAY. That sentence used to be unwritten, and this member's + * own doc used to name the no-`since` case as part of the divergence. + * + * Value is the tracking issue, e.g. `'#10842'`. **No declaration today:** + * `SysMetadataRepository`, the only one there has ever been, was fixed and + * deleted its line — the pin below is what told it to. An empty ledger is + * the mechanism at rest, not dead code; the shrink-only direction is the + * only one it travels without a new tracking issue. */ resumableWatch?: string; } @@ -147,25 +157,16 @@ const SERIALISATION_SHAPES: ReadonlyArray<{ ]; /** Drain at most `n` events from an async iterable with a timeout. */ -async function take(iter: AsyncIterable, n: number, timeoutMs = 1000): Promise { - const out: T[] = []; - const it = iter[Symbol.asyncIterator](); +/** + * Poll `cond` until it holds or `timeoutMs` elapses. Returns either way — the + * caller's `expect` is what fails, so a timeout produces the real assertion + * message instead of "timed out". + */ +async function until(cond: () => boolean, timeoutMs: number): Promise { const deadline = Date.now() + timeoutMs; - while (out.length < n) { - const remaining = deadline - Date.now(); - if (remaining <= 0) break; - const result = await Promise.race([ - it.next(), - new Promise<{ value: undefined; done: true }>((resolve) => - setTimeout(() => resolve({ value: undefined, done: true }), remaining), - ), - ]); - if (result.done) break; - out.push(result.value as T); + while (!cond() && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 10)); } - // Close the iterator so the repo subscriber is freed. - await it.return?.(undefined); - return out; } export function runRepositoryContractTests( @@ -441,6 +442,55 @@ export function runRepositoryContractTests( expect(evts.every((e, i) => i === 0 || e.seq > evts[i - 1]!.seq)).toBe(true); }); + // ── Invariant 6, second half — the FILTER clause, live-stream shaped ── + // + // Unconditional, and deliberately NOT written as "put twice, then open a + // watch and expect the match back". That older shape asserted a replay + // the contract does not owe: invariant 6 states that a `watch()` with no + // `since` is owed LIVE events only, and that events which had already + // committed MAY be delivered but MUST NOT be relied on. The clause passed + // for two implementations because they happen to replay the whole + // matching log, and failed for the third for conforming — a suite defect, + // not a repository defect, and the reason the sentence it leaned on is now + // written down in `repository.ts` instead of inherited from whichever + // implementation was read first. + // + // So the subscription is established FIRST and the writes follow it. What + // is asserted is the floor every implementation owes: the post-subscribe + // event that matches the filter arrives, and the one that does not is + // never delivered. An implementation that additionally replays is not + // failed here — that is the MAY, and pinning its absence is an + // implementation-local question, kept out of this table on purpose (the + // table's whole value is having no per-implementation columns). + it('watch filters by type and name — over the live stream', async () => { + const repo = await factory(); + const iter = repo + .watch({ org: 'system', type: primaryType, name: 'a' }) + [Symbol.asyncIterator](); + const collected: MetadataEvent[] = []; + const pump = (async () => { + for (;;) { + const r = await iter.next(); + if (r.done) return; + collected.push(r.value as MetadataEvent); + } + })(); + + await repo.put(refOf({ name: 'a' }), spec('a'), { parentVersion: null, actor: 't' }); + await repo.put(refOf({ name: 'b' }), spec('b'), { parentVersion: null, actor: 't' }); + // Wait for the match to actually land rather than for a fixed sleep to + // elapse — a filesystem-backed implementation routes it through a real + // watcher and a bare `setTimeout` turns that into a flake. + await until(() => collected.length >= 1, 2000); + // …then let a wrongly-delivered non-match have its chance to show up. + await new Promise((resolve) => setTimeout(resolve, 100)); + await iter.return?.(undefined); + await pump; + + expect(collected.map((e) => e.ref.name)).toEqual(['a']); + expect(collected.every((e, i) => i === 0 || e.seq > collected[i - 1]!.seq)).toBe(true); + }); + if (resumableWatchDivergence === undefined) { it('watch(sinceSeq) replays subsequent events then goes live', async () => { const repo = await factory(); @@ -469,29 +519,14 @@ export function runRepositoryContractTests( await it.return?.(undefined); }); - - it('watch filters by type and name', async () => { - const repo = await factory(); - await repo.put(refOf({ name: 'a' }), spec('a'), { parentVersion: null, actor: 't' }); - await repo.put(refOf({ name: 'b' }), spec('b'), { parentVersion: null, actor: 't' }); - const events = await take( - repo.watch({ org: 'system', type: primaryType, name: 'a' }), - 5, - 200, - ); - expect(events.length).toBe(1); - expect(events[0]!.ref.name).toBe('a'); - }); } else { - // ── DECLARED DIVERGENCE — invariant 6 is not satisfied here ────── + // ── DECLARED DIVERGENCE — invariant 6's first half is unmet here ── // - // Both clauses above lean on the implementation replaying from its - // durable log. This implementation does not, and the two replacements - // below are NOT relaxations of the pair: the first PINS the absence of - // replay (so it reds the day replay lands and this whole branch has to - // go), and the second re-asks the filter question the second clause is - // named for, sourced from the live stream instead of the replay buffer, - // so filter coverage is not silently traded away for the exception. + // Only the numeric-`since` clause above is swapped out; the filter + // clause is shared, because it now asserts the live-stream floor every + // implementation owes rather than a replay only some perform. The + // replacement below is NOT a relaxation: it PINS the absence of replay, + // so it reds the day replay lands and this whole branch has to go. it(`watch(sinceSeq) does NOT replay, then goes live — DECLARED DIVERGENCE ${resumableWatchDivergence}`, async () => { const repo = await factory(); @@ -522,28 +557,6 @@ export function runRepositoryContractTests( await it.return?.(undefined); }); - it(`watch filters by type and name — over the live stream — DECLARED DIVERGENCE ${resumableWatchDivergence}`, async () => { - const repo = await factory(); - const it = repo - .watch({ org: 'system', type: primaryType, name: 'a' }) - [Symbol.asyncIterator](); - const collected: MetadataEvent[] = []; - const pump = (async () => { - for (;;) { - const r = await it.next(); - if (r.done) return; - collected.push(r.value as MetadataEvent); - } - })(); - - await repo.put(refOf({ name: 'a' }), spec('a'), { parentVersion: null, actor: 't' }); - await repo.put(refOf({ name: 'b' }), spec('b'), { parentVersion: null, actor: 't' }); - await new Promise((resolve) => setTimeout(resolve, 100)); - await it.return?.(undefined); - await pump; - - expect(collected.map((e) => e.ref.name)).toEqual(['a']); - }); } }); diff --git a/packages/metadata-core/src/repository.ts b/packages/metadata-core/src/repository.ts index 8c1077d4d1..2df69c50ea 100644 --- a/packages/metadata-core/src/repository.ts +++ b/packages/metadata-core/src/repository.ts @@ -24,8 +24,30 @@ * 4. **Canonical hashing.** `item.hash === hashSpec(item.body)` — always. * 5. **Event ordering.** Subscribers to `watch()` receive events in * monotonically-increasing `seq` order with no gaps. - * 6. **Resumability.** `watch(_, since)` MUST replay all events with - * `seq > since` before delivering live events. + * 6. **Resumability, and where it stops.** `watch(_, since)` called with a + * NUMBER MUST replay all events with `seq > since` before delivering live + * events. Called with NO `since`, `watch()` owes **live events only** — + * the events that commit after the subscription is established; an + * implementation MAY additionally deliver events that had already + * committed, but a caller MUST NOT rely on it, and a caller that needs the + * already-committed prefix MUST pass a numeric `since` (or read + * `history()`). Neither form may deliver the same `seq` twice. + * + * The second sentence is written down because it was load-bearing while + * unwritten. Invariant 6 spoke only of `seq > since`, and with no `since` + * there is no such set — so "no `since` replays everything" existed only + * as `InMemoryRepository`'s implementation, and the shared contract suite + * silently leaned on it. Two of the three implementations shipped today do + * replay the whole matching log on a bare `watch(filter)` + * (`InMemoryRepository`, `FileSystemRepository`); `SysMetadataRepository` + * delivers live events only. That spread is exactly why this is a MAY and + * not a MUST in either direction: forbidding the replay would break two + * implementations and the consumers that lean on them, requiring it would + * flood every `MetadataManager.setRepository()` / `MetadataCache.start()` + * — both of which subscribe with no `since` — with the org's entire + * history at attach time. What a consumer may *rely* on is the floor, and + * the floor is now stated instead of inherited from whichever + * implementation was read first. * 7. **Tombstones, not holes.** `delete` produces a `delete` event; * `get` returns null but `history` still shows the lineage. */ @@ -83,7 +105,13 @@ export interface MetadataRepository { /** * Live event stream. The iterator MUST: * - * - Replay all events with `seq > since` before yielding any new event. + * - When `since` is a number: replay all events with `seq > since` + * before yielding any new event. + * - When `since` is omitted: deliver live events only — the events that + * commit after this subscription is established. Events that had + * already committed MAY also be delivered, but callers MUST NOT rely + * on it; a caller that needs them passes a numeric `since` or reads + * `history()`. See invariant 6. * - Stay open until the consumer breaks the loop. * - Survive transient backend disconnects (implementation's choice * how to resume — Postgres LISTEN reconnect, JSONL tail, etc.). diff --git a/packages/metadata-protocol/src/sys-metadata-repository.contract.test.ts b/packages/metadata-protocol/src/sys-metadata-repository.contract.test.ts index d77bde3842..269d474be0 100644 --- a/packages/metadata-protocol/src/sys-metadata-repository.contract.test.ts +++ b/packages/metadata-protocol/src/sys-metadata-repository.contract.test.ts @@ -58,6 +58,7 @@ import { hashSpec, } from '@objectstack/metadata-core'; import { runRepositoryContractTests } from '@objectstack/metadata-core/testing'; +import type { MetadataEvent } from '@objectstack/metadata-core'; import { SysMetadataRepository } from './sys-metadata-repository.js'; interface Row { @@ -182,17 +183,13 @@ runRepositoryContractTests('SysMetadataRepository', makeRepo, { // type — is not, on purpose (packaged objects are locked); `'dashboard'` is. primaryType: 'view', secondaryType: 'dashboard', - // #10842 — the one invariant this implementation does NOT satisfy, found by - // this very file: `watch()` registers an in-memory listener and reads `since` - // only as a drop-filter on live events, so it never replays from - // `sys_metadata_history`. Declaring it does not skip the clauses: the suite - // swaps in two that pin the divergence and re-ask the filter question over - // the live stream, so this line has to be deleted the day replay lands. - // Not fixed here — both production `watch()` consumers subscribe with no - // `since`, so a full-log replay would flood HMR and cache invalidation at - // every `setRepository()`, and what `watch()` with no `since` owes is not - // written in `repository.ts` at all. #10842 carries the fork. - declaredDivergences: { resumableWatch: '#10842' }, + // #10842 — the `declaredDivergences: { resumableWatch: '#10842' }` line that + // stood here is GONE, deleted by the PR that made invariant 6 true for this + // implementation: `watch(filter, since)` now replays from + // `sys_metadata_history` before going live. The pin clause the declaration + // swapped in went red the moment replay landed, which is the mechanism + // working — it is what told this call site to delete the line rather than + // let a fixed divergence keep being declared. }); /** @@ -252,3 +249,202 @@ describe('SysMetadataRepository — what the contract suite does and does not re expect(got!.hash).toBe(hashSpec(got!.body)); }); }); + +/** + * #10842 — invariant 6's TWO halves, pinned where they are implementation + * facts rather than table entries. + * + * The shared suite asserts the floor every `MetadataRepository` owes and + * deliberately carries no per-implementation column — that is the property + * that keeps it one table. The half below is precisely per-implementation: + * `repository.ts` says a `watch()` with no `since` MAY additionally replay, + * and `SysMetadataRepository` is the implementation that does NOT. That is a + * load-bearing choice, not an omission, so it is pinned here. + * + * ⛔ Why it is load-bearing: both production subscribers attach with no + * `since` — `MetadataManager.startRepositoryWatch()` issues `repo.watch({})` + * from `setRepository()`, and `MetadataCache.start()` issues + * `repo.watch(this.watchFilter)`. Every event they receive invalidates a + * registry entry, drops the `list()` cache and re-emits to every watcher + * (ObjectQLPlugin, Studio HMR). A replay here means the org's whole + * `sys_metadata_history` arrives as "this just changed" at every attach — + * thousands of rows on a mature environment. That flood is the option the + * maintainer explicitly declined, so its absence gets a test, not a comment. + */ +describe('SysMetadataRepository — invariant 6, both halves (#10842)', () => { + const ref = { org: 'system', type: 'view' as const, name: 'sample_view' }; + + /** Drain up to `n` events, or until `timeoutMs` elapses. Never throws. */ + async function drain( + iter: AsyncIterator, + n: number, + timeoutMs: number, + ): Promise { + const out: MetadataEvent[] = []; + const deadline = Date.now() + timeoutMs; + while (out.length < n) { + const remaining = deadline - Date.now(); + if (remaining <= 0) break; + const r = await Promise.race([ + iter.next(), + new Promise>((resolve) => + setTimeout(() => resolve({ value: undefined as any, done: true }), remaining), + ), + ]); + if (r.done) break; + out.push(r.value); + } + return out; + } + + // ── THE ACCIDENTAL-OPTION-A GUARD ────────────────────────────────────── + // + // ⛔ This case FAILS if `watch()` ever starts replaying without `since`. + // That is its entire job: it is the falsifiable form of the sentence + // `repository.ts` now carries, and the tripwire on the attach-time flood. + it.each([ + ['MetadataManager.startRepositoryWatch()', {}], + ['MetadataCache.start() with the default filter', {}], + ['MetadataCache.start() with a type filter', { type: 'view' as const }], + ])('replays NOTHING at attach time for %s', async (_label, filter) => { + const repo = makeRepo(); + const a = await repo.put(ref, { label: '1' }, { parentVersion: null, actor: 't' }); + await repo.put(ref, { label: '2' }, { parentVersion: a.version, actor: 't' }); + + // Two events are committed and durably logged. A no-`since` subscriber is + // owed none of them. + const iter = repo.watch(filter)[Symbol.asyncIterator](); + expect(await drain(iter, 5, 250)).toEqual([]); + await iter.return?.(undefined); + }); + + it('…and the same no-`since` subscriber still receives what commits AFTER it attached', async () => { + const repo = makeRepo(); + const a = await repo.put(ref, { label: '1' }, { parentVersion: null, actor: 't' }); + + const iter = repo.watch({})[Symbol.asyncIterator](); + const pending = drain(iter, 1, 2000); + const b = await repo.put(ref, { label: '2' }, { parentVersion: a.version, actor: 't' }); + + expect((await pending).map((e) => e.seq)).toEqual([b.seq]); + await iter.return?.(undefined); + }); + + // ── FACE 1 — the durable replay ──────────────────────────────────────── + + it('a numeric `since` replays `seq > since` from `sys_metadata_history`, in seq order, then goes live', async () => { + const repo = makeRepo(); + const a = await repo.put(ref, { label: '1' }, { parentVersion: null, actor: 't' }); + const b = await repo.put(ref, { label: '2' }, { parentVersion: a.version, actor: 't' }); + const c = await repo.put(ref, { label: '3' }, { parentVersion: b.version, actor: 't' }); + + const iter = repo.watch({ org: 'system' }, a.seq)[Symbol.asyncIterator](); + // b and c committed BEFORE the subscription and are replayed out of the + // durable log — the read this repository used to never issue. + expect((await drain(iter, 2, 2000)).map((e) => e.seq)).toEqual([b.seq, c.seq]); + + const pending = drain(iter, 1, 2000); + const d = await repo.put(ref, { label: '4' }, { parentVersion: c.version, actor: 't' }); + expect((await pending).map((e) => e.seq)).toEqual([d.seq]); + await iter.return?.(undefined); + }); + + it('the replayed event carries the ROW’s own (type, name), not the filter’s', async () => { + const repo = makeRepo(); + const view = await repo.put(ref, { label: 'v' }, { parentVersion: null, actor: 't' }); + await repo.put( + { org: 'system', type: 'dashboard', name: 'board' }, + { label: 'd' }, + { parentVersion: null, actor: 't' }, + ); + + // Org-wide watch: the replay is org-scoped, so the two rows must come back + // wearing their own refs. A mapping that stamped one ref on every row (the + // shape `history()` legitimately uses, because it resolved a single ref + // first) would pass a same-type fixture and silently mislabel this one. + const iter = repo.watch({ org: 'system' }, view.seq - 1)[Symbol.asyncIterator](); + const got = await drain(iter, 2, 2000); + expect(got.map((e) => `${e.ref.type}/${e.ref.name}`)).toEqual([ + 'view/sample_view', + 'dashboard/board', + ]); + await iter.return?.(undefined); + }); + + it('an event committing DURING the durable read is delivered exactly once', async () => { + const base = makeFakeEngine(); + let onHistoryRead: (() => void) | null = null; + const gated = { + ...base, + async find(table: string, opts: { where: Record; limit?: number }) { + if (table === 'sys_metadata_history' && onHistoryRead) { + const fire = onHistoryRead; + onHistoryRead = null; + fire(); + // Hold the read open long enough for the interleaved write to commit + // and broadcast, so the same event is in BOTH the replay batch and + // the live queue — which is the seam `delivered` exists to close. + await new Promise((resolve) => setTimeout(resolve, 30)); + } + return base.find(table, opts); + }, + }; + const repo = new SysMetadataRepository({ + engine: gated, + organizationId: null, + orgLabel: 'system', + }); + created.push(repo); + + const a = await repo.put(ref, { label: '1' }, { parentVersion: null, actor: 't' }); + const b = await repo.put(ref, { label: '2' }, { parentVersion: a.version, actor: 't' }); + + let cSeq = -1; + onHistoryRead = () => { + void repo + .put(ref, { label: '3' }, { parentVersion: b.version, actor: 't' }) + .then((r) => { + cSeq = r.seq; + }); + }; + const iter = repo.watch({ org: 'system' }, a.seq)[Symbol.asyncIterator](); + + // Ask for THREE; only two exist. The third slot is what would catch a + // duplicate, so the assertion is on the whole drained list, not a prefix. + const got = await drain(iter, 3, 2000); + expect(cSeq).toBeGreaterThan(0); + expect(got.map((e) => e.seq)).toEqual([b.seq, cSeq]); + await iter.return?.(undefined); + }); + + it('a durable-read failure reaches the consumer instead of silently degrading to live-only', async () => { + const base = makeFakeEngine(); + let failHistoryReads = false; + const flaky = { + ...base, + async find(table: string, opts: { where: Record; limit?: number }) { + if (table === 'sys_metadata_history' && failHistoryReads) { + throw new Error('connection reset'); + } + return base.find(table, opts); + }, + }; + const repo = new SysMetadataRepository({ + engine: flaky, + organizationId: null, + orgLabel: 'system', + }); + created.push(repo); + + const a = await repo.put(ref, { label: '1' }, { parentVersion: null, actor: 't' }); + await repo.put(ref, { label: '2' }, { parentVersion: a.version, actor: 't' }); + + failHistoryReads = true; + const iter = repo.watch({ org: 'system' }, a.seq)[Symbol.asyncIterator](); + // A consumer that asked to resume from `a.seq` and got a quiet live tail + // would believe it holds events it does not hold. #4867's rule, one seam + // over: a cursor we could not read is not a cursor we may invent. + await expect(iter.next()).rejects.toThrow('connection reset'); + await iter.return?.(undefined); + }); +}); diff --git a/packages/metadata-protocol/src/sys-metadata-repository.ts b/packages/metadata-protocol/src/sys-metadata-repository.ts index 58aeef4cd7..f32891caca 100644 --- a/packages/metadata-protocol/src/sys-metadata-repository.ts +++ b/packages/metadata-protocol/src/sys-metadata-repository.ts @@ -14,7 +14,10 @@ * project/branch concepts removed — see ADR-0008 §0 amendment) * - hash stamping with `hashSpec` (PR-10a guarantees stability) * - watch() implemented via an in-memory event broadcaster fed by - * every successful put/delete on THIS instance + * every successful put/delete on THIS instance; a NUMERIC `since` + * additionally replays `sys_metadata_history` before going live + * (invariant 6). With no `since` the subscriber is owed live events + * only — see `watch()` for why that is not an oversight. * - whitelist enforcement: refuses to persist types whose registry * entry has `allowOrgOverride: false` (Prime Directive #8) * - **M1**: every successful put/delete appends a durable row to @@ -1009,29 +1012,112 @@ export class SysMetadataRepository implements MetadataRepository { if (opts?.sinceSeq !== undefined && (row.event_seq ?? 0) <= opts.sinceSeq) continue; if (opts?.limit !== undefined && yielded >= opts.limit) break; yielded++; - yield { - seq: (row.event_seq as number) ?? 0, - op: (row.operation_type as MetadataEvent['op']) ?? 'update', - ref: full, - hash: (row.checksum as string | null) ?? null, - parentHash: (row.previous_checksum as string | null) ?? null, - version: typeof row.version === 'number' ? row.version : undefined, - // #4556 — surface the absence, do not paper it over with a label. - // An audit timeline that must show "who changed this" needs to know - // the answer is "the platform", not a user literally named 'unknown'. - actor: (row.recorded_by as string | null | undefined) ?? null, - message: (row.change_note as string | undefined) ?? undefined, - ts: (row.recorded_at as string) ?? new Date(0).toISOString(), - source: (row.source as string | undefined) ?? 'sys-metadata-repo', - }; + yield this.rowToEvent(row, full); } } + /** + * One `sys_metadata_history` row → one `MetadataEvent`. + * + * Extracted from {@link history} so {@link watch}'s replay reads the log + * through the SAME mapping rather than a second hand-written copy. Two + * copies of this would be two answers to "what is `actor` when + * `recorded_by` is null" — and the #4556 answer below (absent, not a user + * called 'unknown') is the kind that drifts silently when duplicated. + * + * `ref` is passed in because the two callers derive it differently: + * `history()` already resolved one `(type, name)` and stamps it on every + * row, while the org-wide replay carries a different `(type, name)` per row. + */ + private rowToEvent(row: any, ref: MetaRef): MetadataEvent { + return { + seq: (row.event_seq as number) ?? 0, + op: (row.operation_type as MetadataEvent['op']) ?? 'update', + ref, + hash: (row.checksum as string | null) ?? null, + parentHash: (row.previous_checksum as string | null) ?? null, + version: typeof row.version === 'number' ? row.version : undefined, + // #4556 — surface the absence, do not paper it over with a label. + // An audit timeline that must show "who changed this" needs to know + // the answer is "the platform", not a user literally named 'unknown'. + actor: (row.recorded_by as string | null | undefined) ?? null, + message: (row.change_note as string | undefined) ?? undefined, + ts: (row.recorded_at as string) ?? new Date(0).toISOString(), + source: (row.source as string | undefined) ?? 'sys-metadata-repo', + }; + } + + /** + * The durable half of invariant 6 — every logged event with + * `seq > since` that matches `filter`, in `seq` order. + * + * Reads `sys_metadata_history` org-wide, which is the same query + * {@link nextEventSeq} already issues; the per-row `(type, name)` is what + * makes the resulting refs differ from `history()`'s single-ref stream. + * + * ⚠️ Called ONLY when `since` is a number. A `watch(filter)` with no `since` + * is owed live events only (invariant 6's second half), and replaying for it + * would flood every `MetadataManager.setRepository()` and + * `MetadataCache.start()` — both subscribe with no `since` — with the org's + * whole history at attach time, as "this just changed". + */ + private async replayFromHistory( + filter: WatchFilter, + since: number, + ): Promise { + const rows = await this.engine.find(this.historyTable, { + where: { organization_id: this.organizationId }, + }); + const out: MetadataEvent[] = []; + for (const row of rows as Array>) { + const type = row.type as MetadataEvent['ref']['type'] | undefined; + const name = row.name as string | undefined; + if (!type || !name) continue; + const evt = this.rowToEvent(row, this.fullRef({ type, name })); + if (evt.seq <= since) continue; + if (!this.matchesFilter(evt, filter)) continue; + out.push(evt); + } + out.sort((a, b) => a.seq - b.seq); + return out; + } + /** * Live event stream. Fires for every successful put/delete on THIS * instance — cross-replica fan-out is M1. Manual AsyncIterator (not * an async generator) so we can deterministically tear down via * `iter.return()`, matching the pattern used by InMemoryRepository. + * + * ## `since` — invariant 6, both halves + * + * **Numeric `since`** (#10842): every logged event with `seq > since` is + * replayed out of `sys_metadata_history` before any live event is yielded. + * `since` used to be nothing but a DROP filter on live events, so an event + * that had already committed was unreachable through `watch()` however low + * `since` was set — the repository held a durable per-org `event_seq` log + * and never read it here. It does now, through the same + * `find(sys_metadata_history, { organization_id })` query `nextEventSeq()` + * issues and the same row→event mapping `history()` uses. + * + * **No `since`**: live events only. Deliberate, and now written into the + * contract (`repository.ts` invariant 6) rather than left to be inherited + * from `InMemoryRepository`, which replays its whole matching log. Both + * production subscribers here — `MetadataManager.startRepositoryWatch()` + * (`repo.watch({})`) and `MetadataCache.start()` + * (`repo.watch(this.watchFilter)`) — attach with no `since`, so replaying + * for them would push the org's entire history through HMR and cache + * invalidation as "this just changed" on every `setRepository()`. A + * consumer that wants the already-committed prefix asks for it by number. + * + * ## Ordering and the replay→live handoff + * + * The live listener is registered SYNCHRONOUSLY, before the durable read is + * issued, so an event committing during the read is buffered rather than + * dropped. It then appears in both the replay batch and the live queue, and + * `delivered` (a set of `seq`) collapses the pair — invariant 5's "no gaps, + * no duplicates" across the seam. Replay is always drained first, so seq + * order holds across it. `seqCounter` mirrors the durable `event_seq` + * (see `put()`), which is what lets one set of numbers dedup both sides. */ watch(filter: WatchFilter, since?: number): AsyncIterable { const self = this; @@ -1040,26 +1126,76 @@ export class SysMetadataRepository implements MetadataRepository { const queue: MetadataEvent[] = []; let pendingResolve: ((r: IteratorResult) => void) | null = null; let stopped = false; + /** `seq`s already handed to the consumer — the replay/live dedup. */ + const delivered = new Set(); + let replay: MetadataEvent[] = []; + let replayIdx = 0; + /** + * A read failure is REPORTED, never swallowed into "live only": a + * consumer that asked to resume from `since` and silently got a live + * tail would believe it had the events it does not have. Held rather + * than left as a rejected promise so an iterator nobody pulls on + * cannot raise an unhandled rejection. + */ + let replayError: unknown = null; const dispatch = (evt: MetadataEvent) => { if (stopped) return; if (!self.matchesFilter(evt, filter)) return; if (since !== undefined && evt.seq <= since) return; + if (delivered.has(evt.seq)) return; if (pendingResolve) { const r = pendingResolve; pendingResolve = null; + delivered.add(evt.seq); r({ value: evt, done: false }); } else { queue.push(evt); } }; + // Registered BEFORE the durable read starts — see the handoff note. self.watchers.add(dispatch); + const replayReady: Promise = + since === undefined + ? Promise.resolve() + : self.replayFromHistory(filter, since).then( + (evts) => { + replay = evts; + }, + (err) => { + replayError = err; + }, + ); + + const drain = (): IteratorResult | null => { + while (replayIdx < replay.length) { + const evt = replay[replayIdx++]!; + if (delivered.has(evt.seq)) continue; + delivered.add(evt.seq); + return { value: evt, done: false }; + } + while (queue.length > 0) { + const evt = queue.shift()!; + if (delivered.has(evt.seq)) continue; + delivered.add(evt.seq); + return { value: evt, done: false }; + } + return null; + }; + return { - next(): Promise> { - if (stopped) return Promise.resolve({ value: undefined as any, done: true }); - const buffered = queue.shift(); - if (buffered) return Promise.resolve({ value: buffered, done: false }); + async next(): Promise> { + if (stopped) return { value: undefined as any, done: true }; + await replayReady; + if (replayError !== null) { + const err = replayError; + replayError = null; + throw err; + } + if (stopped) return { value: undefined as any, done: true }; + const ready = drain(); + if (ready) return ready; return new Promise((resolve) => { pendingResolve = resolve; }); From 75a71f7a22dbe297bca64645c85a515dae580e7f Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 22 Aug 2026 07:54:34 +0000 Subject: [PATCH 2/2] docs(changeset): watch(_, since) replay + the no-since contract sentence Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_019yDEhPBC3tcGkW9bkce1HM --- .../watch-since-replays-from-history.md | 48 +++++++++++++++++++ 1 file changed, 48 insertions(+) create mode 100644 .changeset/watch-since-replays-from-history.md diff --git a/.changeset/watch-since-replays-from-history.md b/.changeset/watch-since-replays-from-history.md new file mode 100644 index 0000000000..7338b76d67 --- /dev/null +++ b/.changeset/watch-since-replays-from-history.md @@ -0,0 +1,48 @@ +--- +"@objectstack/metadata-protocol": minor +"@objectstack/metadata-core": minor +--- + +`MetadataRepository.watch()` — a numeric `since` now replays from the durable +log, and what a bare `watch(filter)` owes is written into the contract. + +**`SysMetadataRepository.watch(filter, since)`** read `since` only as a drop +filter on live events, so an event that had already committed was unreachable +through `watch()` however low `since` was set — even though the repository holds +a durable per-org `event_seq` log in `sys_metadata_history` and already reads it +org-wide in `nextEventSeq()`. Invariant 6 of the repository contract +("`watch(_, since)` MUST replay all events with `seq > since` before delivering +live events") was therefore unimplemented in the repository backing every +production metadata write. It now replays through that same query, using the +row-to-event mapping extracted out of `history()`. The live listener is +registered before the durable read is issued and a set of delivered `seq` +numbers closes the replay-to-live seam, so an event committing mid-read arrives +exactly once; a failed durable read is raised to the consumer rather than +degraded into a silent live-only tail. + +**No behaviour change for a `watch()` with no `since`** — deliberately. Both +in-repo production subscribers (`MetadataManager.startRepositoryWatch()` and +`MetadataCache.start()`) attach that way, and replaying for them would push an +org's entire history through cache invalidation and HMR as "this just changed" +at every attach. + +**Contract text (`@objectstack/metadata-core`, `repository.ts`).** Invariant 6 +now states its own boundary: a `watch()` with no `since` is owed **live events +only**; an implementation MAY additionally deliver events that had already +committed, but a caller MUST NOT rely on it, and a caller that needs the +already-committed prefix passes a numeric `since` or reads `history()`. That +half was previously unwritten and load-bearing — "no `since` replays +everything" existed only as `InMemoryRepository`'s implementation, and the +shared contract suite silently depended on it. + +**If you run `runRepositoryContractTests` from +`@objectstack/metadata-core/testing` against your own implementation**, one +clause changed shape. `watch filters by type and name` (which wrote twice, then +opened a watch and expected the match back) is replaced by `watch filters by +type and name — over the live stream`, which opens the subscription first and +writes after. FROM: an implementation passed by replaying its whole matching log +on a bare `watch(filter)`. TO: it passes by delivering, and filtering, the +events that commit after the subscription is established. An implementation that +replays as well still passes — the new clause asserts the floor, not the +maximum. If yours only replayed and never delivered live events, it was relying +on unspecified behaviour and now needs a live path.