Skip to content

feat(metadata): watch(_, since) replays from sys_metadata_history, and what a bare watch() owes is written down#11018

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
os-elon merged 2 commits into from
Aug 22, 2026
Merged

feat(metadata): watch(_, since) replays from sys_metadata_history, and what a bare watch() owes is written down #11018

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d67cb7b
feat(metadata): watch(_, since) replays from sys_metadata_history; wr…
claude Aug 22, 2026
75a71f7
docs(changeset): watch(_, since) replay + the no-since contract sentence
claude Aug 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions .changeset/watch-since-replays-from-history.md
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -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.
147 changes: 80 additions & 67 deletions packages/metadata-core/src/contract-suite.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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
Expand DownExpand Up@@ -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;
}
Expand DownExpand Up@@ -147,25 +157,16 @@ const SERIALISATION_SHAPES: ReadonlyArray<{
];

/** Drain at most `n` events from an async iterable with a timeout. */
async function take<T>(iter: AsyncIterable<T>, n: number, timeoutMs = 1000): Promise<T[]> {
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<void> {
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(
Expand DownExpand Up@@ -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();
Expand DownExpand Up@@ -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();
Expand DownExpand Up@@ -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']);
});
}
});

Expand Down
34 changes: 31 additions & 3 deletions packages/metadata-core/src/repository.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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.
*/
Expand DownExpand Up@@ -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.).
Expand Down
Loading
Loading