diff --git a/docs/adr/0005-parser-independent-relation-completion.md b/docs/adr/0005-parser-independent-relation-completion.md index f6266dd..6f21ab5 100644 --- a/docs/adr/0005-parser-independent-relation-completion.md +++ b/docs/adr/0005-parser-independent-relation-completion.md @@ -481,11 +481,27 @@ The service reference-counts one provider subscription per provider configuration and scope, then fans invalidation out to subscribed sessions. The installed callback closure supplies authenticated provider and scope identity; the untrusted invalidation payload therefore contains only an epoch. -Subscription membership is installed atomically before a provider can call -back synchronously. A newly joining session captures an already-observed epoch -without a synthetic revision bump. Disposal removes membership before -unsubscribing or running external cleanup. Fifty sessions sharing a scope do -not create fifty provider subscriptions. +Membership uses a two-phase prepare/commit protocol: the session stores its +prepared lease before commit can invoke provider code. Commit installs the +membership atomically before a provider can call back synchronously. Because a +subscription callback is always a change event, a valid callback fired during +`subscribe` advances that committed member after subscription installation +succeeds; it is not reclassified as an initial snapshot. A later joining +session captures an already-observed epoch without a synthetic revision bump. +Disposal removes membership before unsubscribing or running external cleanup. +Retirement also clears the membership's target and coordinator link before +that cleanup, so a consumer retaining a disposed lease retains only inert, +bounded membership metadata. The coordinator stores only the copied provider +ID and captured subscription function; it does not retain the unused search +function. Coordinator disposal revokes that subscription reference before +external cleanup, so a consumer retaining the disposed coordinator handle does +not retain the provider graph. +Fifty sessions sharing a scope do not create fifty provider subscriptions. +The coordinator reserves membership state and capacity before reading the +revision target callback. A throwing or invalid callback getter rolls that +reservation back with `invalid-target` unavailable evidence. Reentrant +coordinator disposal instead reports `disposed` and leaves no reserved +membership or captured callback behind. Each installed subscription has a service-owned incarnation identity captured by its callback. The identity is revoked before external unsubscribe or @@ -493,6 +509,80 @@ cleanup, and every callback checks it before entering the serialized epoch gate. Cleanup from a retired incarnation is keyed to that identity and cannot remove, mutate, or notify a reentrantly installed replacement. +The callback closes over a revocable cell rather than the service, scope state, +or sessions. Retirement clears the cell before external cleanup, so a provider +that retains an old callback retains only an inert bounded object. Synchronous +callbacks are decoded into frozen epochs while `subscribe` is running; raw +payloads and premature audience snapshots are not buffered. After the returned +cleanup closure is validated, the buffered epochs are staged together in FIFO +order. Each accepted baseline or advance snapshots active membership and +installs the epoch as one serialized step with no external call between those +operations. A member joined during one event's dispatch therefore +participates in a later pending event, but not the event already committed. +A thrown subscription or malformed returned cleanup value discards that buffer +and disables automatic invalidation for the live scope; explicit search +remains available. +The returned cleanup closure is itself untrusted: the service captures only a +function, calls it with `this === undefined` at most once, and isolates +malformed values, thrown cleanup, and rejected or hostile thenable results. +Detached cleanup settlement retains no coordinator state. A closure avoids a +structural TypeScript contract that would accept class instances or inherited +methods which the hostile runtime boundary could not safely validate. Dropping +the last owner removes the complete scope incarnation before cleanup. A later +join creates a new unobserved incarnation and may attempt a fresh subscription. + +Malformed, duplicate, stale, or token-conflicting invalidations do not mutate +state and do not tear down an otherwise valid subscription. Every raw callback +still consumes the callback reset-window allowance. Exceeding that allowance +revokes the incarnation and disables automatic invalidation before decoding +further payloads. The conservative window resets from a zero-delay timer +scheduled by the first callback. It intentionally makes no claim about formal +JavaScript task or turn boundaries, while still preventing a provider from +evading the allowance with a microtask chain. If hostile payload inspection +synchronously reenters the same callback, the nested callback consumes that +allowance but is ignored fail-closed; the coordinator does not retain the raw +payload or invert callback order. + +An epoch capture authenticates the exact provider configuration, scope +incarnation, membership, observed epoch, and notification sequence. An +equal-current response is publishable only if no accepted invalidation or +higher response changed that sequence after capture. Otherwise it is +superseded even when its epoch equals the newly current epoch. A capture is a +single-use work token: submission claims it before queue admission, and replay +is discarded even when the first submission was overloaded. + +Epoch commands use one bounded, non-recursive FIFO drain. Its hard admission +limit counts all work accepted during that drain, including commands already +processed, so a one-at-a-time reentrant chain cannot evade the bound. For every +accepted change, active membership is snapshotted and the epoch is immediately +installed before revision preparation or any external listener runs. All +affected session revisions then change before dispatch. Reentrant commands run +only after the current change has reached every still-active member, preserving +event order. The core does not coalesce revision events. The CodeMirror adapter +may coalesce the resulting UI refresh, as described below. Admission fails +before epoch mutation: +an overflowing provider callback revokes automatic invalidation for that +incarnation, while an overflowing response observation settles with closed +overload evidence. +Invalid, retired, replayed, disposed, and overflowing response submissions +return closed settlement evidence without invoking a callback. Only an admitted +FIFO command invokes its completion callback, so rejection cannot recursively +reenter the coordinator's drain. + +A revision target prepares its state change and returns a listener-dispatch +closure. Returning `null`, returning a non-function, or throwing retires that +membership. Retirement detaches state immediately, but hostile provider cleanup +is deferred until every remaining target has prepared. A higher response epoch +installed before preparation remains observed; if its submitting membership is +retired during preparation, its response settles as retired while other +successfully prepared members still dispatch. Coordinator disposal takes +precedence over that retired evidence. The outermost cleanup barrier admits and +invokes at most 1,024 captured cleanups. If reentrant work attempts to admit a +1,025th cleanup, the coordinator quarantines itself immediately: all cells and +memberships become inert and all not-yet-invoked cleanup references are +cleared without calling more provider code. Cleanup is never continued from a +timer. + The session exposes a disposable revision-change subscription for service-originated changes: @@ -671,6 +761,12 @@ The initial checked limits are: | Components per search path | 4 | | Total catalog context | 16,384 UTF-16 units | | Configured relation catalog providers | 1 | +| Live catalog scopes per service | 128 | +| Catalog memberships per service | 1,024 | +| Catalog memberships per scope | 256 | +| Raw subscription callbacks per reset window | 256 | +| Catalog epoch admissions per drain | 1,024 | +| Provider cleanups per outer barrier | 1,024 | | Catalog results per search | 100 | | Completion results after composition | 100 | | Provider ID | 256 UTF-16 units | @@ -815,19 +911,22 @@ Mixed cached assets fail the exact version check and retire the generation. 2. Attach embedded regions to session open/update transactions atomically. 3. Add the bounded partial-`SELECT` query-site recognizer. 4. Add the bounded CTE frame and visibility recognizer. -5. Add the service-owned catalog coordinator, subscriptions, cache, - in-flight sharing, cancellation, and provider contract suite. -6. Add the session completion method and deterministic composition. -7. Add the separate CodeMirror adapter and packed/browser fixtures. -8. Add the pinned packed/browser/runtime marimo fixture and 1/10/50-editor +5. Add the service-owned scoped epoch/subscription coordinator and its hostile + lifecycle contract suite. It does not invoke provider search or retain + result pages. +6. Add catalog search scheduling, cache, in-flight sharing, cancellation, + retry gates, and availability observers over the proven epoch coordinator. +7. Add the session completion method and deterministic composition. +8. Add the separate CodeMirror adapter and packed/browser fixtures. +9. Add the pinned packed/browser/runtime marimo fixture and 1/10/50-editor performance and leak evidence. -9. Validate both the in-memory/notebook and hierarchical remote provider +10. Validate both the in-memory/notebook and hierarchical remote provider shapes, then stabilize the declarations and public export surface. -10. Design scoped parser semantics and protocol v2 against PostgreSQL and +11. Design scoped parser semantics and protocol v2 against PostgreSQL and BigQuery corpora before any additional parser-derived scope feature consumes it. -Breaking refinement remains allowed through step 8. The provider and +Breaking refinement remains allowed through step 9. The provider and completion types become stable only after the working vertical slice, reusable provider contract suite, two materially different provider shapes, marimo packed/browser integration, hostile decoding, cancellation, epoch and diff --git a/package.json b/package.json index 8832267..401581f 100644 --- a/package.json +++ b/package.json @@ -27,6 +27,7 @@ "test:worker-placement": "node ./scripts/worker-placement.mjs", "test:integrity": "node ./scripts/check-test-integrity.mjs", "bench:catalog-boundary": "vitest bench --run src/vnext/__tests__/relation-catalog-boundary.bench.ts", + "bench:catalog-coordinator": "vitest bench --run src/vnext/__tests__/relation-catalog-epoch-coordinator.bench.ts", "bench:local-relation-site": "vitest bench --run src/vnext/__tests__/local-relation-site.bench.ts", "bench:parser-adapter": "vitest bench --run src/vnext/__tests__/node-sql-parser-adapter.bench.ts", "bench:query-site": "vitest bench --run src/vnext/__tests__/query-site.bench.ts", diff --git a/src/vnext/__tests__/relation-catalog-epoch-coordinator.bench.ts b/src/vnext/__tests__/relation-catalog-epoch-coordinator.bench.ts new file mode 100644 index 0000000..66f0650 --- /dev/null +++ b/src/vnext/__tests__/relation-catalog-epoch-coordinator.bench.ts @@ -0,0 +1,443 @@ +import { bench, describe } from "vitest"; +import { + captureSqlRelationCatalogProvider, +} from "../relation-catalog-boundary.js"; +import { + MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW, + createSqlCatalogEpochCoordinator, +} from "../relation-catalog-epoch-coordinator.js"; +import type { + SqlCatalogEpochCoordinator, + SqlCatalogResponseEpochDecision, + SqlCatalogResponseEpochSubmissionResult, + SqlCatalogRevisionTarget, + SqlCatalogScopeMembership, +} from "../relation-catalog-epoch-coordinator.js"; + +interface RevisionCounter { + dispatched: number; + prepared: number; + readonly target: SqlCatalogRevisionTarget; +} + +interface CoordinatorFixture { + readonly coordinator: SqlCatalogEpochCoordinator; + readonly counters: RevisionCounter[]; + readonly dispose: () => void; + readonly emitInvalidation: (generation: number) => void; + readonly memberships: SqlCatalogScopeMembership[]; + readonly subscriptionCounts: { + disposed: number; + installed: number; + }; +} + +let providerSequence = 0; + +function benchmarkFailure(message: string): never { + throw new Error( + `Catalog epoch coordinator benchmark preflight failed: ${message}`, + ); +} + +function revisionCounter(): RevisionCounter { + const counter: RevisionCounter = { + dispatched: 0, + prepared: 0, + target: { + prepareCatalogChange: (): (() => void) => { + counter.prepared += 1; + return (): void => { + counter.dispatched += 1; + }; + }, + }, + }; + return counter; +} + +function requireMembership( + coordinator: SqlCatalogEpochCoordinator, + scope: string, + counter: RevisionCounter, +): SqlCatalogScopeMembership { + const prepared = coordinator.prepareScopeMembership( + scope, + counter.target, + ); + if (prepared.status !== "prepared") { + return benchmarkFailure("membership preparation was unavailable"); + } + const activated = prepared.membership.activate(); + if (activated.status !== "active") { + return benchmarkFailure("membership activation was unavailable"); + } + return prepared.membership; +} + +function requireMember( + members: readonly SqlCatalogScopeMembership[], + index: number, +): SqlCatalogScopeMembership { + const member = members[index]; + if (!member) { + return benchmarkFailure("membership index was unavailable"); + } + return member; +} + +function requireDecision( + submit: ( + receive: ( + this: void, + decision: SqlCatalogResponseEpochDecision, + ) => void, + ) => SqlCatalogResponseEpochSubmissionResult, +): SqlCatalogResponseEpochDecision { + const decisions: SqlCatalogResponseEpochDecision[] = []; + const submission = submit((decision): void => { + decisions.push(decision); + }); + if (submission.status !== "submitted") { + return benchmarkFailure("epoch command was not submitted"); + } + const decision = decisions[0]; + if (!decision || decisions.length !== 1) { + return benchmarkFailure( + "epoch command did not settle exactly once", + ); + } + return decision; +} + +function createFixture(memberCount: number): CoordinatorFixture { + let invalidationListener: + | ((this: void, event: unknown) => void) + | null = null; + const subscriptionCounts = { + disposed: 0, + installed: 0, + }; + providerSequence += 1; + const captured = captureSqlRelationCatalogProvider({ + id: `coordinator-benchmark-${providerSequence}`, + search: (): Promise => + new Promise(() => { + // The epoch coordinator never invokes catalog search. + }), + subscribe: ( + _scope: string, + listener: (this: void, event: unknown) => void, + ) => { + subscriptionCounts.installed += 1; + invalidationListener = listener; + return (): void => { + subscriptionCounts.disposed += 1; + }; + }, + }); + if (captured.status !== "accepted") { + return benchmarkFailure("provider capture was rejected"); + } + const created = createSqlCatalogEpochCoordinator( + captured.value, + ); + if (created.status !== "created") { + return benchmarkFailure("coordinator creation was unavailable"); + } + const counters: RevisionCounter[] = []; + const memberships: SqlCatalogScopeMembership[] = []; + for (let index = 0; index < memberCount; index += 1) { + const counter = revisionCounter(); + counters.push(counter); + memberships.push( + requireMembership( + created.coordinator, + "benchmark-scope", + counter, + ), + ); + } + return { + coordinator: created.coordinator, + counters, + dispose: (): void => { + created.coordinator.dispose(); + }, + emitInvalidation: (generation: number): void => { + const listener = invalidationListener; + if (!listener) { + return benchmarkFailure( + "provider subscription listener was unavailable", + ); + } + listener({ + epoch: { + generation, + token: "benchmark-epoch", + }, + }); + }, + memberships, + subscriptionCounts, + }; +} + +function establishBaseline( + fixture: CoordinatorFixture, +): SqlCatalogScopeMembership { + const member = requireMember(fixture.memberships, 0); + const captured = member.captureEpoch(); + if (captured.status !== "captured") { + return benchmarkFailure("epoch capture was unavailable"); + } + const decision = requireDecision((receive) => + fixture.coordinator.submitResponseEpoch( + captured.capture, + { generation: 0, token: "benchmark-epoch" }, + receive, + ), + ); + if ( + decision.status !== "usable" || + decision.observation !== "baseline" + ) { + return benchmarkFailure("baseline response was not usable"); + } + return member; +} + +function assertFanout(memberCount: number): void { + const fixture = createFixture(memberCount); + const member = establishBaseline(fixture); + const captured = member.captureEpoch(); + if (captured.status !== "captured") { + return benchmarkFailure("fanout capture was unavailable"); + } + const decision = requireDecision((receive) => + fixture.coordinator.submitResponseEpoch( + captured.capture, + { generation: 1, token: "benchmark-epoch" }, + receive, + ), + ); + if (decision.status !== "superseded") { + return benchmarkFailure( + "higher response did not supersede its capture", + ); + } + for (const counter of fixture.counters) { + if (counter.prepared !== 1 || counter.dispatched !== 1) { + return benchmarkFailure( + `higher response did not reach all ${memberCount} members`, + ); + } + } + fixture.dispose(); +} + +function assertRejectedCallbacks(): void { + const fixture = createFixture(1); + fixture.emitInvalidation(2); + fixture.emitInvalidation(2); + fixture.emitInvalidation(1); + if ( + fixture.counters[0]?.prepared !== 1 || + fixture.counters[0]?.dispatched !== 1 + ) { + return benchmarkFailure( + "duplicate or stale callback changed the revision", + ); + } + fixture.dispose(); +} + +function assertBoundedStorm(): void { + const fixture = createFixture(1); + for ( + let generation = 1; + generation <= MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW; + generation += 1 + ) { + fixture.emitInvalidation(generation); + } + const counter = fixture.counters[0]; + if ( + !counter || + counter.prepared !== MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW || + counter.dispatched !== MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW + ) { + return benchmarkFailure("bounded callback storm lost an event"); + } + fixture.emitInvalidation( + MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW + 1, + ); + fixture.emitInvalidation( + MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW + 2, + ); + if ( + counter.prepared !== MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW || + counter.dispatched !== MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW || + fixture.subscriptionCounts.disposed !== 1 + ) { + return benchmarkFailure( + "callback overflow did not retire the subscription", + ); + } + fixture.dispose(); +} + +function assertAttachDetachReuse(): void { + const fixture = createFixture(1); + for (let index = 0; index < 256; index += 1) { + const counter = revisionCounter(); + const membership = requireMembership( + fixture.coordinator, + "benchmark-scope", + counter, + ); + membership.dispose(); + } + const finalMembership = requireMembership( + fixture.coordinator, + "benchmark-scope", + revisionCounter(), + ); + finalMembership.dispose(); + if ( + fixture.subscriptionCounts.installed !== 1 || + fixture.subscriptionCounts.disposed !== 0 + ) { + return benchmarkFailure( + "attach/detach churn changed the shared subscription", + ); + } + fixture.dispose(); + if (Number(fixture.subscriptionCounts.disposed) !== 1) { + return benchmarkFailure( + "shared subscription was not disposed exactly once", + ); + } +} + +for (const memberCount of [1, 10, 50, 256]) { +assertFanout(memberCount); +} +assertRejectedCallbacks(); +assertBoundedStorm(); +assertAttachDetachReuse(); + +const noopDecision = ( + _decision: SqlCatalogResponseEpochDecision, +): void => {}; + +describe("relation catalog epoch coordinator", () => { + for (const memberCount of [1, 10, 50, 256]) { + const lifecycleFixture = createFixture(memberCount); + let lifecycleCursor = 0; + bench( + `join, activate, and dispose within a ${memberCount}-member scope`, + () => { + const previous = requireMember( + lifecycleFixture.memberships, + lifecycleCursor, + ); + const nextCounter = revisionCounter(); + if (memberCount === 1) { + const next = requireMembership( + lifecycleFixture.coordinator, + "benchmark-scope", + nextCounter, + ); + previous.dispose(); + lifecycleFixture.memberships[lifecycleCursor] = next; + } else { + previous.dispose(); + lifecycleFixture.memberships[lifecycleCursor] = + requireMembership( + lifecycleFixture.coordinator, + "benchmark-scope", + nextCounter, + ); + } + lifecycleCursor = + (lifecycleCursor + 1) % memberCount; + }, + ); + + const fanoutFixture = createFixture(memberCount); + const fanoutMember = establishBaseline(fanoutFixture); + let generation = 0; + bench( + `accept higher response and fan out to ${memberCount} members`, + () => { + generation += 1; + const nextCapture = fanoutMember.captureEpoch(); + if (nextCapture.status !== "captured") { + benchmarkFailure( + "fanout benchmark capture was unavailable", + ); + } + const submission = + fanoutFixture.coordinator.submitResponseEpoch( + nextCapture.capture, + { generation, token: "benchmark-epoch" }, + noopDecision, + ); + if (submission.status !== "submitted") { + benchmarkFailure( + "fanout benchmark response was not submitted", + ); + } + }, + ); + } + + bench("reject duplicate and stale provider callbacks", () => { + const fixture = createFixture(1); + fixture.emitInvalidation(2); + fixture.emitInvalidation(2); + fixture.emitInvalidation(1); + fixture.dispose(); + }); + + bench( + "process a bounded 256-event provider callback storm for one member", + () => { + const fixture = createFixture(1); + for ( + let generation = 1; + generation <= MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW; + generation += 1 + ) { + fixture.emitInvalidation(generation); + } + fixture.dispose(); + }, + ); + + bench("fan out a bounded 256-event storm to 256 members", () => { + const fixture = createFixture(256); + for ( + let generation = 1; + generation <= MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW; + generation += 1 + ) { + fixture.emitInvalidation(generation); + } + fixture.dispose(); + }); + + bench("attach and detach 10,000 members on one scope", () => { + const fixture = createFixture(1); + for (let index = 0; index < 10_000; index += 1) { + const membership = requireMembership( + fixture.coordinator, + "benchmark-scope", + revisionCounter(), + ); + membership.dispose(); + } + fixture.dispose(); + }); +}); diff --git a/src/vnext/__tests__/relation-catalog-epoch-coordinator.test.ts b/src/vnext/__tests__/relation-catalog-epoch-coordinator.test.ts new file mode 100644 index 0000000..f0d0a50 --- /dev/null +++ b/src/vnext/__tests__/relation-catalog-epoch-coordinator.test.ts @@ -0,0 +1,2046 @@ +import { MessageChannel } from "node:worker_threads"; +import { describe, expect, it } from "vitest"; +import { + captureSqlRelationCatalogProvider, + MAX_CATALOG_SCOPE_LENGTH, + type CapturedSqlRelationCatalogProvider, +} from "../relation-catalog-boundary.js"; +import { + createSqlCatalogEpochCoordinator, + MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW, + MAX_CATALOG_CLEANUPS_PER_BARRIER, + MAX_CATALOG_EPOCH_COMMANDS, + MAX_CATALOG_LIVE_SCOPES, + MAX_CATALOG_MEMBERSHIPS, + MAX_CATALOG_MEMBERSHIPS_PER_SCOPE, + type SqlCatalogEpochCapture, + type SqlCatalogEpochCoordinator, + type SqlCatalogResponseEpochDecision, + type SqlCatalogResponseEpochSubmissionResult, + type SqlCatalogRevisionTarget, + type SqlCatalogScopeMembership, +} from "../relation-catalog-epoch-coordinator.js"; + +type RawInvalidationListener = (value: unknown) => void; +type RawSubscribe = ( + scope: string, + listener: RawInvalidationListener, +) => unknown; + +function epoch(generation: number, token = `epoch-${generation}`) { + return { generation, token }; +} + +function invalidation(generation: number, token = `epoch-${generation}`) { + return { epoch: epoch(generation, token) }; +} + +function capturedProvider( + subscribe?: RawSubscribe, + id = "catalog", +): CapturedSqlRelationCatalogProvider { + const candidate = + subscribe === undefined + ? { + id, + search: async () => null, + } + : { + id, + search: async () => null, + subscribe, + }; + const result = captureSqlRelationCatalogProvider(candidate); + expect(result.status).toBe("accepted"); + if (result.status !== "accepted") { + throw new Error("Expected an accepted provider fixture"); + } + return result.value; +} + +function coordinator( + subscribe?: RawSubscribe, + id = "catalog", +): SqlCatalogEpochCoordinator { + const result = createSqlCatalogEpochCoordinator( + capturedProvider(subscribe, id), + ); + expect(result.status).toBe("created"); + if (result.status !== "created") { + throw new Error("Expected a coordinator fixture"); + } + return result.coordinator; +} + +function prepared( + owner: SqlCatalogEpochCoordinator, + scope: unknown, + target: SqlCatalogRevisionTarget = { + prepareCatalogChange: () => () => {}, + }, +): SqlCatalogScopeMembership { + const result = owner.prepareScopeMembership(scope, target); + expect(result.status).toBe("prepared"); + if (result.status !== "prepared") { + throw new Error("Expected a prepared membership fixture"); + } + return result.membership; +} + +function active( + owner: SqlCatalogEpochCoordinator, + scope: string, + target?: SqlCatalogRevisionTarget, +): SqlCatalogScopeMembership { + const membership = prepared(owner, scope, target); + expect(membership.activate()).toEqual({ status: "active" }); + return membership; +} + +function capture( + membership: SqlCatalogScopeMembership, +): SqlCatalogEpochCapture { + const result = membership.captureEpoch(); + expect(result.status).toBe("captured"); + if (result.status !== "captured") { + throw new Error("Expected a capture fixture"); + } + return result.capture; +} + +function submit( + owner: SqlCatalogEpochCoordinator, + captured: unknown, + value: unknown, +): { + readonly decisions: readonly SqlCatalogResponseEpochDecision[]; + readonly result: SqlCatalogResponseEpochSubmissionResult; +} { + const decisions: SqlCatalogResponseEpochDecision[] = []; + const result = owner.submitResponseEpoch( + captured, + value, + (decision) => decisions.push(decision), + ); + return { decisions, result }; +} + +function listenerProvider(): { + readonly listeners: RawInvalidationListener[]; + readonly scopes: string[]; + readonly subscribe: RawSubscribe; + cleanupCalls: number; +} { + const listeners: RawInvalidationListener[] = []; + const scopes: string[] = []; + const harness = { + cleanupCalls: 0, + listeners, + scopes, + subscribe: ( + scope: string, + listener: RawInvalidationListener, + ) => { + harness.scopes.push(scope); + harness.listeners.push(listener); + return () => { + harness.cleanupCalls += 1; + }; + }, + }; + return harness; +} + +function counterTarget( + onDispatch?: () => void, +): { + readonly target: SqlCatalogRevisionTarget; + prepared: number; + dispatched: number; +} { + const counter = { + dispatched: 0, + prepared: 0, + target: { + prepareCatalogChange: () => { + counter.prepared += 1; + return () => { + counter.dispatched += 1; + onDispatch?.(); + }; + }, + }, + }; + return counter; +} + +describe("catalog epoch coordinator construction and membership", () => { + it("authenticates providers and freezes its closed API", () => { + expect(createSqlCatalogEpochCoordinator(null)).toEqual({ + reason: "invalid-provider", + status: "unavailable", + }); + expect( + createSqlCatalogEpochCoordinator({ + ...capturedProvider(undefined, "copied"), + }), + ).toEqual({ + reason: "invalid-provider", + status: "unavailable", + }); + + const owner = coordinator(undefined, "provider-a"); + expect(owner.providerId).toBe("provider-a"); + expect(Object.isFrozen(owner)).toBe(true); + }); + + it("validates exact bounded well-formed scopes without raw errors", () => { + const owner = coordinator(); + expect( + owner.prepareScopeMembership("", counterTarget().target), + ).toEqual({ + reason: "invalid-scope", + status: "unavailable", + }); + for (const scope of [ + null, + 1, + "bad\0scope", + "\ud800", + "\ud800a", + "\udc00", + "x".repeat(MAX_CATALOG_SCOPE_LENGTH + 1), + ]) { + expect( + owner.prepareScopeMembership(scope, counterTarget().target), + ).toEqual({ + reason: "invalid-scope", + status: "unavailable", + }); + } + expect( + owner.prepareScopeMembership( + `\ud83d\ude80${"x".repeat(MAX_CATALOG_SCOPE_LENGTH - 2)}`, + counterTarget().target, + ).status, + ).toBe("prepared"); + }); + + it("uses two-phase activation and exposes only a frozen expected epoch", () => { + const owner = coordinator(); + const membership = prepared(owner, "scope"); + expect(membership.captureEpoch()).toEqual({ + reason: "inactive", + status: "unavailable", + }); + expect(membership.activate()).toEqual({ status: "active" }); + expect(membership.activate()).toEqual({ status: "active" }); + const first = capture(membership); + expect(first.expectedEpoch).toBeNull(); + expect(Object.keys(first)).toEqual(["expectedEpoch"]); + expect(Object.isFrozen(first)).toBe(true); + membership.dispose(); + membership.dispose(); + expect(membership.captureEpoch()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(membership.activate()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + }); + + it("does not retain a membership if target capture disposes the coordinator", () => { + const owner = coordinator(); + const target = new Proxy( + { + prepareCatalogChange: () => null, + }, + { + get(value, property, receiver) { + if (property === "prepareCatalogChange") { + owner.dispose(); + return 1; + } + return Reflect.get(value, property, receiver); + }, + }, + ); + const result = owner.prepareScopeMembership("scope", target); + expect(result).toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect( + owner.prepareScopeMembership("later", counterTarget().target), + ).toEqual({ + reason: "disposed", + status: "unavailable", + }); + + const throwingOwner = coordinator(); + expect( + throwingOwner.prepareScopeMembership("scope", { + get prepareCatalogChange(): never { + throwingOwner.dispose(); + throw new Error("dispose before hostile getter failure"); + }, + }), + ).toEqual({ + reason: "disposed", + status: "unavailable", + }); + }); + + it("reserves membership capacity before a hostile target getter can reenter", () => { + const owner = coordinator(); + const memberships: SqlCatalogScopeMembership[] = []; + const rejected: string[] = []; + let getterCalls = 0; + let remaining = MAX_CATALOG_MEMBERSHIPS + 16; + const target = { + get prepareCatalogChange(): SqlCatalogRevisionTarget["prepareCatalogChange"] { + getterCalls += 1; + if (remaining > 0) { + remaining -= 1; + const nested = owner.prepareScopeMembership("scope", target); + if (nested.status === "prepared") { + memberships.push(nested.membership); + } else { + rejected.push(nested.reason); + } + } + return () => null; + }, + }; + + const outer = owner.prepareScopeMembership("scope", target); + if (outer.status === "prepared") { + memberships.push(outer.membership); + } else { + rejected.push(outer.reason); + } + expect(memberships).toHaveLength(MAX_CATALOG_MEMBERSHIPS); + expect(getterCalls).toBe(MAX_CATALOG_MEMBERSHIPS); + expect(remaining).toBe(16); + expect(rejected).toEqual(["membership-capacity"]); + expect( + owner.prepareScopeMembership("overflow", counterTarget().target), + ).toEqual({ + reason: "membership-capacity", + status: "unavailable", + }); + + for (const membership of memberships) membership.dispose(); + const reused = Array.from( + { length: MAX_CATALOG_MEMBERSHIPS }, + (_, index) => + owner.prepareScopeMembership( + `reused-${index}`, + counterTarget().target, + ), + ); + expect(reused.every((result) => result.status === "prepared")).toBe( + true, + ); + for (const result of reused) { + if (result.status === "prepared") result.membership.dispose(); + } + }); + + it.each([ + { + name: "a throwing getter", + target: { + get prepareCatalogChange(): never { + throw new Error("hostile target getter"); + }, + }, + }, + { + name: "a non-callable runtime value", + target: new Proxy( + { + prepareCatalogChange: () => null, + }, + { + get(value, property, receiver) { + return property === "prepareCatalogChange" + ? 1 + : Reflect.get(value, property, receiver); + }, + }, + ), + }, + ])( + "rejects $name without consuming the reserved membership slot", + ({ target }) => { + const owner = coordinator(); + const existing = Array.from( + { length: MAX_CATALOG_MEMBERSHIPS - 1 }, + (_, index) => prepared(owner, `existing-${index}`), + ); + expect( + owner.prepareScopeMembership("invalid-target", target), + ).toEqual({ + reason: "invalid-target", + status: "unavailable", + }); + const final = owner.prepareScopeMembership( + "final", + counterTarget().target, + ); + expect(final.status).toBe("prepared"); + expect( + owner.prepareScopeMembership( + "overflow", + counterTarget().target, + ), + ).toEqual({ + reason: "membership-capacity", + status: "unavailable", + }); + + for (const membership of existing) membership.dispose(); + if (final.status === "prepared") final.membership.dispose(); + }, + ); + + it.each([1, 10, 50])( + "shares one subscription across %i same-scope members", + (count) => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const memberships = Array.from({ length: count }, () => + active(owner, "shared"), + ); + expect(harness.scopes).toEqual(["shared"]); + const secondScope = active(owner, "other"); + expect(harness.scopes).toEqual(["shared", "other"]); + for (const membership of memberships.slice(0, -1)) { + membership.dispose(); + } + expect(harness.cleanupCalls).toBe(0); + memberships.at(-1)?.dispose(); + expect(harness.cleanupCalls).toBe(1); + secondScope.dispose(); + expect(harness.cleanupCalls).toBe(2); + }, + ); + + it("supports providers with no subscription", () => { + const owner = coordinator(); + const membership = active(owner, "static"); + const first = capture(membership); + const outcome = submit(owner, first, epoch(3)); + expect(outcome.result).toEqual({ status: "submitted" }); + expect(outcome.decisions).toEqual([ + { + epoch: epoch(3), + observation: "baseline", + status: "usable", + }, + ]); + expect(Object.isFrozen(outcome.result)).toBe(true); + expect(Object.isFrozen(outcome.decisions[0])).toBe(true); + expect(capture(membership).expectedEpoch).toEqual(epoch(3)); + }); + + it("never invokes provider search", () => { + let searchCalls = 0; + const captured = captureSqlRelationCatalogProvider({ + id: "search-is-out-of-scope", + search: async () => { + searchCalls += 1; + throw new Error("search must remain scheduler-owned"); + }, + }); + if (captured.status !== "accepted") { + throw new Error("Expected an accepted provider fixture"); + } + const created = createSqlCatalogEpochCoordinator(captured.value); + if (created.status !== "created") { + throw new Error("Expected a coordinator fixture"); + } + const membership = active(created.coordinator, "scope"); + submit(created.coordinator, capture(membership), epoch(1)); + created.coordinator.dispose(); + expect(searchCalls).toBe(0); + }); + + it("enforces exact capacity edges and reuses released capacity", () => { + const owner = coordinator(); + const inactive = Array.from( + { length: MAX_CATALOG_MEMBERSHIPS }, + (_, index) => prepared(owner, `prepared-${index}`), + ); + expect( + owner.prepareScopeMembership("overflow", counterTarget().target), + ).toEqual({ + reason: "membership-capacity", + status: "unavailable", + }); + inactive[0]?.dispose(); + expect( + owner.prepareScopeMembership("reused", counterTarget().target) + .status, + ).toBe("prepared"); + + const scopeOwner = coordinator(); + const scopes = Array.from( + { length: MAX_CATALOG_LIVE_SCOPES }, + (_, index) => active(scopeOwner, `scope-${index}`), + ); + const blocked = prepared(scopeOwner, "scope-overflow"); + expect(blocked.activate()).toEqual({ + reason: "scope-capacity", + status: "unavailable", + }); + scopes[0]?.dispose(); + expect(blocked.activate()).toEqual({ status: "active" }); + + const memberOwner = coordinator(); + const sameScope = Array.from( + { length: MAX_CATALOG_MEMBERSHIPS_PER_SCOPE }, + () => active(memberOwner, "crowded"), + ); + const extra = prepared(memberOwner, "crowded"); + expect(extra.activate()).toEqual({ + reason: "scope-membership-capacity", + status: "unavailable", + }); + sameScope[0]?.dispose(); + expect(extra.activate()).toEqual({ status: "active" }); + }); + + it("reclaims scope structure across ten thousand incarnations", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + for (let index = 0; index < 10_000; index += 1) { + active(owner, `scope-${index}`).dispose(); + } + expect(harness.scopes).toHaveLength(10_000); + expect(harness.cleanupCalls).toBe(10_000); + expect(active(owner, "after-churn").captureEpoch().status).toBe( + "captured", + ); + }); +}); + +describe("response epoch observations and capture authority", () => { + it("classifies baseline, equal, advance, stale, conflicts, and malformed epochs", () => { + const target = counterTarget(); + const owner = coordinator(); + const membership = active(owner, "scope", target.target); + + const baseline = submit(owner, capture(membership), epoch(1)); + expect(baseline.decisions).toEqual([ + { + epoch: epoch(1), + observation: "baseline", + status: "usable", + }, + ]); + expect(target.dispatched).toBe(0); + + const equal = submit(owner, capture(membership), epoch(1)); + expect(equal.decisions[0]).toMatchObject({ + observation: "equal", + status: "usable", + }); + + const advance = submit(owner, capture(membership), epoch(2)); + expect(advance.decisions).toEqual([ + { epoch: epoch(2), status: "superseded" }, + ]); + expect(target.dispatched).toBe(1); + expect(capture(membership).expectedEpoch).toEqual(epoch(2)); + + expect( + submit(owner, capture(membership), epoch(1)).decisions, + ).toEqual([{ reason: "stale", status: "discarded" }]); + expect( + submit(owner, capture(membership), epoch(2, "other")).decisions, + ).toEqual([ + { reason: "token-conflict", status: "discarded" }, + ]); + for (const malformed of [ + null, + {}, + { generation: -1, token: "bad" }, + { generation: 3, token: "\ud800" }, + ]) { + expect( + submit(owner, capture(membership), malformed).decisions, + ).toEqual([{ reason: "malformed", status: "discarded" }]); + } + }); + + it("claims captures once and keeps immediate rejection callback-free", () => { + const owner = coordinator(); + const membership = active(owner, "scope"); + const oneUse = capture(membership); + let callbackCalls = 0; + expect( + owner.submitResponseEpoch(oneUse, epoch(1), () => { + callbackCalls += 1; + const replay = owner.submitResponseEpoch( + oneUse, + epoch(1), + () => { + callbackCalls += 100; + }, + ); + expect(replay).toEqual({ + decision: { reason: "retired", status: "discarded" }, + status: "settled", + }); + }), + ).toEqual({ status: "submitted" }); + expect(callbackCalls).toBe(1); + expect( + owner.submitResponseEpoch(oneUse, epoch(1), () => { + callbackCalls += 100; + }), + ).toEqual({ + decision: { reason: "retired", status: "discarded" }, + status: "settled", + }); + expect(callbackCalls).toBe(1); + + for (const forged of [ + null, + {}, + { ...capture(membership) }, + ]) { + expect( + owner.submitResponseEpoch(forged, epoch(1), () => { + callbackCalls += 100; + }), + ).toEqual({ + decision: { reason: "malformed", status: "discarded" }, + status: "settled", + }); + } + expect(callbackCalls).toBe(1); + }); + + it("supersedes equal-current work captured before an accepted change", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const membership = active(owner, "scope"); + submit(owner, capture(membership), epoch(1)); + const beforeInvalidation = capture(membership); + harness.listeners[0]?.(invalidation(2)); + expect( + submit(owner, beforeInvalidation, epoch(2)).decisions, + ).toEqual([{ epoch: epoch(2), status: "superseded" }]); + + const beforeResponse = capture(membership); + expect( + submit(owner, capture(membership), epoch(3)).decisions, + ).toEqual([{ epoch: epoch(3), status: "superseded" }]); + expect(submit(owner, beforeResponse, epoch(3)).decisions).toEqual([ + { epoch: epoch(3), status: "superseded" }, + ]); + }); + + it("rejects cross-coordinator, retired, and cross-incarnation captures", () => { + const provider = capturedProvider(); + const createdA = createSqlCatalogEpochCoordinator(provider); + const createdB = createSqlCatalogEpochCoordinator(provider); + if (createdA.status !== "created" || createdB.status !== "created") { + throw new Error("Expected coordinator fixtures"); + } + const first = active(createdA.coordinator, "scope"); + const stale = capture(first); + expect( + createdB.coordinator.submitResponseEpoch( + stale, + epoch(1), + () => { + throw new Error("must not run"); + }, + ), + ).toEqual({ + decision: { reason: "malformed", status: "discarded" }, + status: "settled", + }); + first.dispose(); + active(createdA.coordinator, "scope"); + expect( + createdA.coordinator.submitResponseEpoch( + stale, + epoch(1), + () => { + throw new Error("must not run"); + }, + ), + ).toEqual({ + decision: { reason: "retired", status: "discarded" }, + status: "settled", + }); + }); + + it("fails closed when hostile response decoding retires membership or coordinator", () => { + const memberOwner = coordinator(); + const member = active(memberOwner, "scope"); + const memberEpoch = new Proxy(epoch(1), { + ownKeys(value) { + member.dispose(); + return Reflect.ownKeys(value); + }, + }); + const retired = submit( + memberOwner, + capture(member), + memberEpoch, + ); + expect(retired.result).toEqual({ status: "submitted" }); + expect(retired.decisions).toEqual([ + { reason: "retired", status: "discarded" }, + ]); + + const serviceOwner = coordinator(); + const serviceMember = active(serviceOwner, "scope"); + const serviceEpoch = new Proxy(epoch(1), { + ownKeys(value) { + serviceOwner.dispose(); + return Reflect.ownKeys(value); + }, + }); + const disposed = submit( + serviceOwner, + capture(serviceMember), + serviceEpoch, + ); + expect(disposed.result).toEqual({ status: "submitted" }); + expect(disposed.decisions).toEqual([ + { reason: "disposed", status: "discarded" }, + ]); + }); +}); + +describe("subscription invalidation ordering and isolation", () => { + it("classifies baseline/equal/advance while preserving state-before-listener and insertion order", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const order: string[] = []; + let first: SqlCatalogScopeMembership; + const firstTarget: SqlCatalogRevisionTarget = { + prepareCatalogChange: () => { + order.push(`prepare-a:${capture(first).expectedEpoch?.generation}`); + return () => { + order.push( + `dispatch-a:${capture(first).expectedEpoch?.generation}`, + ); + }; + }, + }; + first = active(owner, "scope", firstTarget); + const second = active(owner, "scope", { + prepareCatalogChange: () => { + order.push("prepare-b"); + return () => order.push("dispatch-b"); + }, + }); + const notify = harness.listeners[0]; + notify?.(invalidation(1)); + notify?.(invalidation(1)); + notify?.(invalidation(0)); + notify?.(invalidation(1, "conflict")); + notify?.(invalidation(2)); + expect(order).toEqual([ + "prepare-a:1", + "prepare-b", + "dispatch-a:1", + "dispatch-b", + "prepare-a:2", + "prepare-b", + "dispatch-a:2", + "dispatch-b", + ]); + expect(capture(second).expectedEpoch).toEqual(epoch(2)); + }); + + it("includes members activated after a reentrant invalidation is queued", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const joinedCounter = counterTarget(); + let joined: SqlCatalogScopeMembership | undefined; + let joinedAtActivation: + | SqlCatalogEpochCapture["expectedEpoch"] + | undefined; + let firstPrepared = 0; + let firstDispatched = 0; + const first = active(owner, "scope", { + prepareCatalogChange: () => { + firstPrepared += 1; + return () => { + firstDispatched += 1; + if (firstDispatched !== 1) return; + harness.listeners[0]?.(invalidation(2)); + joined = active(owner, "scope", { + ...joinedCounter.target, + }); + joinedAtActivation = capture(joined).expectedEpoch; + }; + }, + }); + + harness.listeners[0]?.(invalidation(1)); + expect(joinedAtActivation).toEqual(epoch(1)); + expect({ firstDispatched, firstPrepared }).toEqual({ + firstDispatched: 2, + firstPrepared: 2, + }); + expect(joinedCounter).toMatchObject({ + dispatched: 1, + prepared: 1, + }); + expect(capture(first).expectedEpoch).toEqual(epoch(2)); + expect(capture(joined ?? first).expectedEpoch).toEqual(epoch(2)); + }); + + it("isolates listener failure and skips a member disposed by an earlier listener", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const events: string[] = []; + let second: SqlCatalogScopeMembership; + active(owner, "scope", { + prepareCatalogChange: () => () => { + events.push("first"); + second.dispose(); + throw new Error("listener failure"); + }, + }); + second = active(owner, "scope", { + prepareCatalogChange: () => () => events.push("second"), + }); + active(owner, "scope", { + prepareCatalogChange: () => { + throw new Error("prepare failure"); + }, + }); + active(owner, "scope", { + prepareCatalogChange: () => () => events.push("last"), + }); + expect(() => + harness.listeners[0]?.(invalidation(1)), + ).not.toThrow(); + expect(events).toEqual(["first", "last"]); + }); + + it("skips a queued audience member disposed during preparation", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const events: string[] = []; + let second: SqlCatalogScopeMembership; + active(owner, "scope", { + prepareCatalogChange: () => { + events.push("prepare-first"); + second.dispose(); + return () => events.push("dispatch-first"); + }, + }); + second = active(owner, "scope", { + prepareCatalogChange: () => { + events.push("prepare-second"); + return () => events.push("dispatch-second"); + }, + }); + harness.listeners[0]?.(invalidation(1)); + expect(events).toEqual([ + "prepare-first", + "dispatch-first", + ]); + }); + + it("retires failed revision targets without blocking live members or early cleanup", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const nullTarget = active(owner, "scope", { + prepareCatalogChange: () => null, + }); + const throwingTarget = active(owner, "scope", { + prepareCatalogChange: () => { + throw new Error("prepare failed"); + }, + }); + const good = counterTarget(); + const goodMembership = active(owner, "scope", good.target); + + harness.listeners[0]?.(invalidation(1)); + expect(nullTarget.captureEpoch()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(throwingTarget.captureEpoch()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(good).toMatchObject({ dispatched: 1, prepared: 1 }); + expect(capture(goodMembership).expectedEpoch).toEqual(epoch(1)); + expect(harness.cleanupCalls).toBe(0); + + nullTarget.dispose(); + throwingTarget.dispose(); + expect(harness.cleanupCalls).toBe(0); + goodMembership.dispose(); + expect(harness.cleanupCalls).toBe(1); + }); + + it("fails closed against hostile decode reentrancy without reversing epochs", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const observed: number[] = []; + const membership = active(owner, "scope", { + prepareCatalogChange: () => () => { + const generation = capture(membership).expectedEpoch?.generation; + if (generation !== undefined) observed.push(generation); + }, + }); + const hostileEpoch = new Proxy(epoch(1), { + ownKeys(target) { + harness.listeners[0]?.(invalidation(2)); + return Reflect.ownKeys(target); + }, + }); + harness.listeners[0]?.({ epoch: hostileEpoch }); + expect(observed).toEqual([1]); + expect(capture(membership).expectedEpoch).toEqual(epoch(1)); + }); + + it("drops decoded work if hostile decoding retires its subscription", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const target = counterTarget(); + const membership = active(owner, "scope", target.target); + const hostileEpoch = new Proxy(epoch(1), { + ownKeys(value) { + membership.dispose(); + return Reflect.ownKeys(value); + }, + }); + expect(() => + harness.listeners[0]?.({ epoch: hostileEpoch }), + ).not.toThrow(); + expect(target.dispatched).toBe(0); + expect(membership.captureEpoch()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + }); + + it("fans higher responses out atomically and drains reentrant responses FIFO", () => { + const owner = coordinator(); + const events: string[] = []; + let first: SqlCatalogScopeMembership; + let reentered = false; + first = active(owner, "scope", { + prepareCatalogChange: () => { + events.push( + `prepare-a:${capture(first).expectedEpoch?.generation}`, + ); + return () => { + events.push("dispatch-a"); + if (!reentered) { + reentered = true; + const nested = owner.submitResponseEpoch( + capture(first), + epoch(3), + (decision) => events.push(`decision-${decision.status}`), + ); + expect(nested).toEqual({ status: "submitted" }); + } + }; + }, + }); + const second = active(owner, "scope", { + prepareCatalogChange: () => { + events.push( + `prepare-b:${capture(first).expectedEpoch?.generation}`, + ); + return () => events.push("dispatch-b"); + }, + }); + submit(owner, capture(first), epoch(1)); + const result = owner.submitResponseEpoch( + capture(first), + epoch(2), + (decision) => events.push(`decision-${decision.status}`), + ); + expect(result).toEqual({ status: "submitted" }); + expect(events).toEqual([ + "prepare-a:2", + "prepare-b:2", + "decision-superseded", + "dispatch-a", + "dispatch-b", + "prepare-a:3", + "prepare-b:3", + "decision-superseded", + "dispatch-a", + "dispatch-b", + ]); + expect(capture(first).expectedEpoch).toEqual(epoch(3)); + expect(capture(second).expectedEpoch).toEqual(epoch(3)); + }); + + it("includes members activated after a higher response is queued", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const decisions: SqlCatalogResponseEpochDecision[] = []; + const joinedCounter = counterTarget(); + let admission: SqlCatalogResponseEpochSubmissionResult | undefined; + let joined: SqlCatalogScopeMembership | undefined; + let joinedAtActivation: + | SqlCatalogEpochCapture["expectedEpoch"] + | undefined; + let firstPrepared = 0; + let firstDispatched = 0; + let responseCapture: SqlCatalogEpochCapture; + const first = active(owner, "scope", { + prepareCatalogChange: () => { + firstPrepared += 1; + return () => { + firstDispatched += 1; + if (firstDispatched !== 1) return; + admission = owner.submitResponseEpoch( + responseCapture, + epoch(2), + (decision) => decisions.push(decision), + ); + joined = active(owner, "scope", joinedCounter.target); + joinedAtActivation = capture(joined).expectedEpoch; + }; + }, + }); + responseCapture = capture(first); + + harness.listeners[0]?.(invalidation(1)); + expect(admission).toEqual({ status: "submitted" }); + expect(joinedAtActivation).toEqual(epoch(1)); + expect(decisions).toEqual([ + { epoch: epoch(2), status: "superseded" }, + ]); + expect({ firstDispatched, firstPrepared }).toEqual({ + firstDispatched: 2, + firstPrepared: 2, + }); + expect(joinedCounter).toMatchObject({ + dispatched: 1, + prepared: 1, + }); + expect(capture(first).expectedEpoch).toEqual(epoch(2)); + expect(capture(joined ?? first).expectedEpoch).toEqual(epoch(2)); + }); + + it("does not dispatch a higher response after preparation disposes the coordinator", () => { + const owner = coordinator(); + let dispatchCalls = 0; + let disposeDuringPrepare = false; + const membership = active(owner, "scope", { + prepareCatalogChange: () => { + if (disposeDuringPrepare) owner.dispose(); + return () => { + dispatchCalls += 1; + }; + }, + }); + submit(owner, capture(membership), epoch(1)); + disposeDuringPrepare = true; + const outcome = submit(owner, capture(membership), epoch(2)); + expect(outcome.result).toEqual({ status: "submitted" }); + expect(outcome.decisions).toEqual([ + { reason: "disposed", status: "discarded" }, + ]); + expect(dispatchCalls).toBe(0); + }); + + it("discards a higher response whose capture retires during preparation", () => { + const owner = coordinator(); + let retireDuringPrepare = false; + let dispatchCalls = 0; + let membership: SqlCatalogScopeMembership; + membership = active(owner, "scope", { + prepareCatalogChange: () => { + if (retireDuringPrepare) membership.dispose(); + return () => { + dispatchCalls += 1; + }; + }, + }); + submit(owner, capture(membership), epoch(1)); + retireDuringPrepare = true; + + const outcome = submit(owner, capture(membership), epoch(2)); + expect(outcome.result).toEqual({ status: "submitted" }); + expect(outcome.decisions).toEqual([ + { reason: "retired", status: "discarded" }, + ]); + expect(dispatchCalls).toBe(0); + expect(membership.captureEpoch()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + }); + + it("settles a retired response before cleaning a last membership disposed during preparation", () => { + const events: string[] = []; + let disposeDuringPrepare = false; + let membership: SqlCatalogScopeMembership; + const owner = coordinator( + () => () => { + events.push("cleanup"); + }, + ); + membership = active(owner, "scope", { + prepareCatalogChange: () => { + if (disposeDuringPrepare) { + events.push("prepare-start"); + membership.dispose(); + events.push("prepare-end"); + } + return () => events.push("dispatch"); + }, + }); + submit(owner, capture(membership), epoch(1)); + disposeDuringPrepare = true; + + const result = owner.submitResponseEpoch( + capture(membership), + epoch(2), + (decision) => { + events.push(`decision-${decision.status}`); + }, + ); + expect(result).toEqual({ status: "submitted" }); + expect(events).toEqual([ + "prepare-start", + "prepare-end", + "decision-discarded", + "cleanup", + ]); + }); + + it("discards an admitted response when its membership retires before processing", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const decisions: SqlCatalogResponseEpochDecision[] = []; + let admission: SqlCatalogResponseEpochSubmissionResult | undefined; + let membership: SqlCatalogScopeMembership; + membership = active(owner, "scope", { + prepareCatalogChange: () => () => { + admission = owner.submitResponseEpoch( + capture(membership), + epoch(2), + (decision) => decisions.push(decision), + ); + membership.dispose(); + }, + }); + harness.listeners[0]?.(invalidation(1)); + expect(admission).toEqual({ status: "submitted" }); + expect(decisions).toEqual([ + { reason: "retired", status: "discarded" }, + ]); + }); +}); + +describe("hostile subscription lifecycle", () => { + it("buffers a valid synchronous callback until subscribe succeeds", () => { + let listenerCalls = 0; + const owner = coordinator((_scope, notify) => { + notify(invalidation(1)); + return () => {}; + }); + const membership = prepared(owner, "scope", { + prepareCatalogChange: () => () => { + listenerCalls += 1; + }, + }); + expect(membership.activate()).toEqual({ status: "active" }); + expect(listenerCalls).toBe(1); + expect(capture(membership).expectedEpoch).toEqual(epoch(1)); + }); + + it("flushes a synchronous callback to every member active when subscribe commits", () => { + let second: SqlCatalogScopeMembership; + const firstCounter = counterTarget(); + const secondCounter = counterTarget(); + const owner = coordinator((_scope, notify) => { + notify(invalidation(1)); + expect(second.activate()).toEqual({ status: "active" }); + return () => {}; + }); + const first = prepared(owner, "scope", firstCounter.target); + second = prepared(owner, "scope", secondCounter.target); + + expect(first.activate()).toEqual({ status: "active" }); + expect(firstCounter).toMatchObject({ dispatched: 1, prepared: 1 }); + expect(secondCounter).toMatchObject({ dispatched: 1, prepared: 1 }); + expect(capture(first).expectedEpoch).toEqual(epoch(1)); + expect(capture(second).expectedEpoch).toEqual(epoch(1)); + }); + + it("includes a member activated between two buffered synchronous invalidations", () => { + const firstCounter = { + dispatched: 0, + prepared: 0, + }; + const secondCounter = counterTarget(); + let second: SqlCatalogScopeMembership; + let secondAtActivation: + | SqlCatalogEpochCapture["expectedEpoch"] + | undefined; + const owner = coordinator((_scope, notify) => { + notify(invalidation(1)); + notify(invalidation(2)); + return () => {}; + }); + const first = prepared(owner, "scope", { + prepareCatalogChange: () => { + firstCounter.prepared += 1; + return () => { + firstCounter.dispatched += 1; + if (firstCounter.dispatched !== 1) return; + expect(second.activate()).toEqual({ status: "active" }); + secondAtActivation = capture(second).expectedEpoch; + }; + }, + }); + second = prepared(owner, "scope", secondCounter.target); + + expect(first.activate()).toEqual({ status: "active" }); + expect(secondAtActivation).toEqual(epoch(1)); + expect(firstCounter).toEqual({ dispatched: 2, prepared: 2 }); + expect(secondCounter).toMatchObject({ + dispatched: 1, + prepared: 1, + }); + expect(capture(first).expectedEpoch).toEqual(epoch(2)); + expect(capture(second).expectedEpoch).toEqual(epoch(2)); + }); + + it("discards malformed buffered callbacks and all buffered work when subscribe fails", () => { + let listenerCalls = 0; + for (const subscribe of [ + (_scope: string, notify: RawInvalidationListener) => { + notify({ epoch: { generation: -1, token: "bad" } }); + return () => {}; + }, + (_scope: string, notify: RawInvalidationListener) => { + notify(invalidation(1)); + throw new Error("subscribe failed"); + }, + (_scope: string, notify: RawInvalidationListener) => { + notify(invalidation(1)); + return null; + }, + ]) { + const owner = coordinator(subscribe); + const membership = active(owner, "scope", { + prepareCatalogChange: () => () => { + listenerCalls += 1; + }, + }); + expect(capture(membership).expectedEpoch).toBeNull(); + } + expect(listenerCalls).toBe(0); + }); + + it("captures one cleanup closure, invokes it this-free once, and rejects non-functions without getters", () => { + const receivers: unknown[] = []; + let getterCalls = 0; + const validOwner = coordinator( + () => + function (this: unknown) { + receivers.push(this); + }, + ); + const membership = active(validOwner, "scope"); + membership.dispose(); + membership.dispose(); + expect(membership.activate()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(membership.captureEpoch()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + validOwner.dispose(); + expect(receivers).toEqual([undefined]); + + const nonEnumerable = {}; + Object.defineProperty(nonEnumerable, "dispose", { + enumerable: false, + value: () => {}, + }); + const invalidCleanupValues: unknown[] = [ + {}, + Object.create({ dispose() {} }), + nonEnumerable, + new Proxy( + { dispose() {} }, + { + getOwnPropertyDescriptor() { + throw new Error("hostile"); + }, + }, + ), + { + get dispose() { + getterCalls += 1; + return () => {}; + }, + }, + ]; + for (const cleanupValue of invalidCleanupValues) { + const owner = coordinator(() => cleanupValue); + expect(active(owner, "scope").captureEpoch().status).toBe( + "captured", + ); + } + expect(getterCalls).toBe(0); + }); + + it("isolates thrown cleanup and retains no live callback after failed subscribe", () => { + let retained: RawInvalidationListener | undefined; + const owner = coordinator((_scope, notify) => { + retained = notify; + return () => { + throw new Error("cleanup failed"); + }; + }); + const target = counterTarget(); + const membership = active(owner, "scope", target.target); + expect(() => membership.dispose()).not.toThrow(); + retained?.(invalidation(1)); + expect(target.dispatched).toBe(0); + + let failedRetained: RawInvalidationListener | undefined; + const failedOwner = coordinator((_scope, notify) => { + failedRetained = notify; + return null; + }); + const failedTarget = counterTarget(); + active(failedOwner, "scope", failedTarget.target); + failedRetained?.(invalidation(1)); + expect(failedTarget.dispatched).toBe(0); + }); + + it("drains rejected and hostile cleanup results after making state inert", async () => { + let asyncCleanupCalls = 0; + const rejectingOwner = coordinator( + () => async () => { + asyncCleanupCalls += 1; + throw new Error("async cleanup failed"); + }, + ); + const rejectingMembership = active(rejectingOwner, "scope"); + rejectingMembership.dispose(); + rejectingMembership.dispose(); + + let applyCalls = 0; + let thenReads = 0; + const thenProperty = ["th", "en"].join(""); + const hostileResult = Object.defineProperty({}, thenProperty, { + get() { + thenReads += 1; + throw new Error("hostile then getter"); + }, + }); + const hostileCleanup = new Proxy(() => hostileResult, { + apply(target, receiver, argumentsList) { + applyCalls += 1; + expect(receiver).toBeUndefined(); + return Reflect.apply(target, receiver, argumentsList); + }, + }); + const hostileOwner = coordinator(() => hostileCleanup); + const hostileMembership = active(hostileOwner, "scope"); + hostileMembership.dispose(); + hostileMembership.dispose(); + + let catchReads = 0; + const throwingCatchResult = Promise.reject( + new Error("shadowed catch rejection"), + ); + Object.defineProperty(throwingCatchResult, "catch", { + get() { + catchReads += 1; + throw new Error("hostile catch getter"); + }, + }); + const throwingCatchOwner = coordinator( + () => () => throwingCatchResult, + ); + active(throwingCatchOwner, "scope").dispose(); + + const nonCallableCatchResult = Promise.reject( + new Error("non-callable catch rejection"), + ); + Object.defineProperty(nonCallableCatchResult, "catch", { + value: null, + }); + const nonCallableCatchOwner = coordinator( + () => () => nonCallableCatchResult, + ); + active(nonCallableCatchOwner, "scope").dispose(); + + let thenCalls = 0; + let reentrantOwner: SqlCatalogEpochCoordinator; + const reentrantResult = Object.defineProperty( + {}, + thenProperty, + { + value: (resolve: () => void): void => { + thenCalls += 1; + const replacement = active( + reentrantOwner, + "replacement", + ); + replacement.dispose(); + resolve(); + }, + }, + ); + const reentrantCleanup = () => reentrantResult; + let reentrantSubscriptions = 0; + reentrantOwner = coordinator(() => { + reentrantSubscriptions += 1; + return reentrantSubscriptions === 1 + ? reentrantCleanup + : () => {}; + }); + active(reentrantOwner, "scope").dispose(); + + await Promise.resolve(); + await Promise.resolve(); + expect(asyncCleanupCalls).toBe(1); + expect(applyCalls).toBe(1); + expect(thenReads).toBe(1); + expect(catchReads).toBe(0); + expect(thenCalls).toBe(1); + }); + + it("retries a failed subscription only in a new last-owner incarnation", () => { + const listeners: RawInvalidationListener[] = []; + let attempts = 0; + const owner = coordinator((_scope, notify) => { + attempts += 1; + listeners.push(notify); + if (attempts === 1) return null; + return () => {}; + }); + const first = active(owner, "scope"); + const second = active(owner, "scope"); + expect(attempts).toBe(1); + first.dispose(); + expect(attempts).toBe(1); + second.dispose(); + const replacement = active(owner, "scope"); + expect(attempts).toBe(2); + listeners[0]?.(invalidation(10)); + expect(capture(replacement).expectedEpoch).toBeNull(); + listeners[1]?.(invalidation(1)); + expect(capture(replacement).expectedEpoch).toEqual(epoch(1)); + }); + + it("retires old callbacks before reentrant cleanup installs a replacement", () => { + const listeners: RawInvalidationListener[] = []; + const memberships: SqlCatalogScopeMembership[] = []; + let owner: SqlCatalogEpochCoordinator; + const provider = (_scope: string, notify: RawInvalidationListener) => { + listeners.push(notify); + return () => { + if (memberships.length === 1) { + memberships.push(active(owner, "scope")); + } + }; + }; + owner = coordinator(provider); + memberships.push(active(owner, "scope")); + memberships[0]?.dispose(); + expect(listeners).toHaveLength(2); + listeners[0]?.(invalidation(10)); + expect( + capture(memberships[1] ?? memberships[0]!).expectedEpoch, + ).toBeNull(); + listeners[1]?.(invalidation(1)); + expect( + capture(memberships[1] ?? memberships[0]!).expectedEpoch, + ).toEqual(epoch(1)); + }); + + it("returns unavailable when synchronous activation work disposes membership or coordinator", () => { + let membership: SqlCatalogScopeMembership; + const memberOwner = coordinator((_scope, notify) => { + notify(invalidation(1)); + return () => {}; + }); + membership = prepared(memberOwner, "scope", { + prepareCatalogChange: () => () => membership.dispose(), + }); + expect(membership.activate()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + + let service: SqlCatalogEpochCoordinator; + service = coordinator((_scope, notify) => { + notify(invalidation(1)); + return () => {}; + }); + const serviceMembership = prepared(service, "scope", { + prepareCatalogChange: () => () => service.dispose(), + }); + expect(serviceMembership.activate()).toEqual({ + reason: "coordinator-disposed", + status: "unavailable", + }); + }); + + it("quarantines 257 callbacks across microtasks and cleans an install-time overload once", async () => { + let notify: RawInvalidationListener | undefined; + let cleanupCalls = 0; + const owner = coordinator((_scope, listener) => { + notify = listener; + return () => { + cleanupCalls += 1; + }; + }); + const membership = active(owner, "scope"); + for ( + let index = 0; + index < MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW; + index += 1 + ) { + await Promise.resolve(); + notify?.({ malformed: index }); + } + await Promise.resolve(); + notify?.(invalidation(1)); + expect(cleanupCalls).toBe(1); + expect(capture(membership).expectedEpoch).toBeNull(); + + let installCleanup = 0; + const installOwner = coordinator((_scope, listener) => { + for ( + let index = 0; + index <= MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW; + index += 1 + ) { + listener({ malformed: index }); + } + return () => { + installCleanup += 1; + }; + }); + active(installOwner, "scope"); + expect(installCleanup).toBe(1); + }); + + it("resets callback allowance only after its reset timer fires", async () => { + let notify: RawInvalidationListener | undefined; + const owner = coordinator((_scope, listener) => { + notify = listener; + return () => {}; + }); + const membership = active(owner, "scope"); + for ( + let index = 0; + index < MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW; + index += 1 + ) { + notify?.({ malformed: index }); + } + await new Promise((resolve) => { + setTimeout(resolve, 0); + }); + notify?.(invalidation(1)); + expect(capture(membership).expectedEpoch).toEqual(epoch(1)); + }); + + it("conservatively counts an earlier queued MessageChannel task in the reset window", async () => { + let notify: RawInvalidationListener | undefined; + let cleanupCalls = 0; + const owner = coordinator((_scope, listener) => { + notify = listener; + return () => { + cleanupCalls += 1; + }; + }); + const membership = active(owner, "scope"); + const channel = new MessageChannel(); + const messageTask = new Promise((resolve) => { + channel.port2.once("message", () => { + notify?.(invalidation(1)); + resolve(); + }); + }); + channel.port1.postMessage("queued-before-reset-timer"); + for ( + let index = 0; + index < MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW; + index += 1 + ) { + notify?.({ malformed: index }); + } + await messageTask; + channel.port1.close(); + channel.port2.close(); + + expect(cleanupCalls).toBe(1); + expect(capture(membership).expectedEpoch).toBeNull(); + }); +}); + +describe("service disposal and bounded command draining", () => { + it("deduplicates install-time overload cleanup within one outer barrier", () => { + let driverNotify: RawInvalidationListener | undefined; + let overloadedCleanupCalls = 0; + let driverCleanupCalls = 0; + let preparedOverloads = false; + let owner: SqlCatalogEpochCoordinator; + owner = coordinator((scope, notify) => { + if (scope === "driver") { + driverNotify = notify; + } else { + for ( + let index = 0; + index <= MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW; + index += 1 + ) { + notify({ malformed: index }); + } + } + return () => { + if (scope === "driver") { + driverCleanupCalls += 1; + } else { + overloadedCleanupCalls += 1; + } + }; + }); + const driver = active(owner, "driver", { + prepareCatalogChange: () => { + if (!preparedOverloads) { + preparedOverloads = true; + for ( + let index = 0; + index < MAX_CATALOG_CLEANUPS_PER_BARRIER; + index += 1 + ) { + active(owner, `overloaded-${index}`).dispose(); + } + } + return () => {}; + }, + }); + + driverNotify?.(invalidation(1)); + expect(overloadedCleanupCalls).toBe( + MAX_CATALOG_CLEANUPS_PER_BARRIER, + ); + expect(driverCleanupCalls).toBe(0); + expect(capture(driver).expectedEpoch).toEqual(epoch(1)); + expect( + owner.prepareScopeMembership( + "after-deduplicated-cleanup", + counterTarget().target, + ).status, + ).toBe("prepared"); + + driver.dispose(); + expect(driverCleanupCalls).toBe(1); + }); + + it("quarantines cleanup reentrancy at the documented barrier limit", () => { + const listeners = new Map(); + let cleanupCalls = 0; + let cleanupDepth = 0; + let lateCleanupCalls = 0; + let maximumCleanupDepth = 0; + let sentinelCleanupCalls = 0; + let subscriptionCalls = 0; + let owner: SqlCatalogEpochCoordinator; + owner = coordinator((scope, notify) => { + subscriptionCalls += 1; + listeners.set(scope, notify); + const installsAfterCleanupLimit = + subscriptionCalls === + MAX_CATALOG_CLEANUPS_PER_BARRIER + 2; + if (installsAfterCleanupLimit) { + for ( + let index = 0; + index <= MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW; + index += 1 + ) { + notify({ malformed: index }); + } + } + return () => { + if (installsAfterCleanupLimit) { + lateCleanupCalls += 1; + return; + } + cleanupDepth += 1; + maximumCleanupDepth = Math.max( + maximumCleanupDepth, + cleanupDepth, + ); + cleanupCalls += 1; + if (scope === "sentinel") { + sentinelCleanupCalls += 1; + } else { + const next = active( + owner, + `cleanup-chain-${subscriptionCalls}`, + ); + next.dispose(); + } + cleanupDepth -= 1; + }; + }); + const sentinelTarget = counterTarget(); + const sentinel = active( + owner, + "sentinel", + sentinelTarget.target, + ); + active(owner, "cleanup-chain-0", { + prepareCatalogChange: () => null, + }); + + listeners.get("cleanup-chain-0")?.(invalidation(1)); + expect(subscriptionCalls).toBe( + MAX_CATALOG_CLEANUPS_PER_BARRIER + 2, + ); + expect(cleanupCalls).toBe( + MAX_CATALOG_CLEANUPS_PER_BARRIER, + ); + expect(maximumCleanupDepth).toBe(1); + expect(lateCleanupCalls).toBe(0); + expect(sentinelCleanupCalls).toBe(0); + expect(sentinel.captureEpoch()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect( + owner.prepareScopeMembership( + "after-overflow", + counterTarget().target, + ), + ).toEqual({ + reason: "disposed", + status: "unavailable", + }); + + listeners.get("sentinel")?.(invalidation(2)); + owner.dispose(); + sentinel.dispose(); + expect(sentinelTarget.dispatched).toBe(0); + expect(cleanupCalls).toBe( + MAX_CATALOG_CLEANUPS_PER_BARRIER, + ); + expect(lateCleanupCalls).toBe(0); + expect(sentinelCleanupCalls).toBe(0); + }); + + it("retires callbacks, members, and cleanup exactly once on disposal", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const target = counterTarget(); + const membership = active(owner, "scope", target.target); + const retained = harness.listeners[0]; + owner.dispose(); + owner.dispose(); + expect(harness.cleanupCalls).toBe(1); + retained?.(invalidation(1)); + expect(target.dispatched).toBe(0); + expect(membership.activate()).toEqual({ + reason: "coordinator-disposed", + status: "unavailable", + }); + expect(membership.captureEpoch()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + membership.dispose(); + membership.dispose(); + expect(harness.cleanupCalls).toBe(1); + expect( + owner.prepareScopeMembership("new", target.target), + ).toEqual({ + reason: "disposed", + status: "unavailable", + }); + let callbackCalls = 0; + expect( + owner.submitResponseEpoch({}, epoch(1), () => { + callbackCalls += 1; + }), + ).toEqual({ + decision: { reason: "disposed", status: "discarded" }, + status: "settled", + }); + expect(callbackCalls).toBe(0); + }); + + it("retires the provider callback before reentrant cleanup", () => { + let retained: RawInvalidationListener | undefined; + let cleanupCalls = 0; + let payloadReads = 0; + const hostilePayload = new Proxy( + {}, + { + get(target, property, receiver) { + payloadReads += 1; + return Reflect.get(target, property, receiver); + }, + getOwnPropertyDescriptor(target, property) { + payloadReads += 1; + return Reflect.getOwnPropertyDescriptor(target, property); + }, + ownKeys(target) { + payloadReads += 1; + return Reflect.ownKeys(target); + }, + }, + ); + const owner = coordinator((_scope, listener) => { + retained = listener; + return () => { + cleanupCalls += 1; + retained?.(hostilePayload); + }; + }); + const target = counterTarget(); + const membership = active(owner, "scope", target.target); + + expect(() => owner.dispose()).not.toThrow(); + owner.dispose(); + expect(cleanupCalls).toBe(1); + expect(payloadReads).toBe(0); + expect(target.dispatched).toBe(0); + + expect(() => retained?.(hostilePayload)).not.toThrow(); + expect(payloadReads).toBe(0); + expect(target.dispatched).toBe(0); + expect(membership.activate()).toEqual({ + reason: "coordinator-disposed", + status: "unavailable", + }); + expect(membership.captureEpoch()).toEqual({ + reason: "disposed", + status: "unavailable", + }); + membership.dispose(); + membership.dispose(); + expect(cleanupCalls).toBe(1); + expect( + owner.prepareScopeMembership("new", target.target), + ).toEqual({ + reason: "disposed", + status: "unavailable", + }); + let callbackCalls = 0; + expect( + owner.submitResponseEpoch({}, epoch(2), () => { + callbackCalls += 1; + }), + ).toEqual({ + decision: { reason: "disposed", status: "discarded" }, + status: "settled", + }); + expect(callbackCalls).toBe(0); + }); + + it("settles queued response sinks as disposed without recursive callbacks", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const decisions: SqlCatalogResponseEpochDecision[] = []; + let queuedResult: SqlCatalogResponseEpochSubmissionResult | undefined; + const membership = active(owner, "scope", { + prepareCatalogChange: () => () => { + queuedResult = owner.submitResponseEpoch( + capture(membership), + epoch(2), + (decision) => decisions.push(decision), + ); + owner.dispose(); + }, + }); + harness.listeners[0]?.(invalidation(1)); + expect(queuedResult).toEqual({ status: "submitted" }); + expect(decisions).toEqual([ + { reason: "disposed", status: "discarded" }, + ]); + }); + + it("isolates thrown decision callbacks and continues draining", () => { + const owner = coordinator(); + const membership = active(owner, "scope"); + expect(() => + owner.submitResponseEpoch( + capture(membership), + epoch(1), + () => { + throw new Error("consumer failure"); + }, + ), + ).not.toThrow(); + const next = submit(owner, capture(membership), epoch(1)); + expect(next.decisions).toEqual([ + { + epoch: epoch(1), + observation: "equal", + status: "usable", + }, + ]); + }); + + it("bounds response commands and keeps overload rejection nonrecursive", () => { + const harness = listenerProvider(); + const owner = coordinator(harness.subscribe); + const membership = active(owner, "scope", { + prepareCatalogChange: () => () => { + const admitted: SqlCatalogEpochCapture[] = []; + for ( + let index = 0; + index < MAX_CATALOG_EPOCH_COMMANDS; + index += 1 + ) { + admitted.push(capture(membership)); + } + for (const item of admitted) { + expect( + owner.submitResponseEpoch(item, epoch(1), () => {}), + ).toEqual({ status: "submitted" }); + } + let recursiveCalls = 0; + const overflow = capture(membership); + const result = owner.submitResponseEpoch( + overflow, + epoch(1), + () => { + recursiveCalls += 1; + owner.submitResponseEpoch( + capture(membership), + epoch(1), + () => { + recursiveCalls += 1; + }, + ); + }, + ); + expect(result).toEqual({ + decision: { reason: "overloaded", status: "discarded" }, + status: "settled", + }); + expect(recursiveCalls).toBe(0); + expect( + owner.submitResponseEpoch(overflow, epoch(1), () => { + recursiveCalls += 1; + }), + ).toEqual({ + decision: { reason: "retired", status: "discarded" }, + status: "settled", + }); + }, + }); + harness.listeners[0]?.(invalidation(1)); + expect(capture(membership).expectedEpoch).toEqual(epoch(1)); + }); + + it("caps one-at-a-time response chains by total admissions per drain", () => { + const owner = coordinator(); + const membership = active(owner, "scope"); + const captures = Array.from( + { length: MAX_CATALOG_EPOCH_COMMANDS + 1 }, + () => capture(membership), + ); + let callbacks = 0; + let callbackDepth = 0; + let maximumCallbackDepth = 0; + let overflow: SqlCatalogResponseEpochSubmissionResult | undefined; + + const onDecision = (): void => { + callbackDepth += 1; + maximumCallbackDepth = Math.max( + maximumCallbackDepth, + callbackDepth, + ); + callbacks += 1; + const next = captures[callbacks]; + if (next) { + const result = owner.submitResponseEpoch( + next, + epoch(1), + onDecision, + ); + if (result.status === "settled") overflow = result; + } + callbackDepth -= 1; + }; + + const first = captures[0]; + if (!first) throw new Error("Expected an epoch capture"); + expect( + owner.submitResponseEpoch(first, epoch(1), onDecision), + ).toEqual({ status: "submitted" }); + expect(callbacks).toBe(MAX_CATALOG_EPOCH_COMMANDS); + expect(maximumCallbackDepth).toBe(1); + expect(overflow).toEqual({ + decision: { reason: "overloaded", status: "discarded" }, + status: "settled", + }); + + let afterDrain = 0; + expect( + owner.submitResponseEpoch( + capture(membership), + epoch(1), + () => { + afterDrain += 1; + }, + ), + ).toEqual({ status: "submitted" }); + expect(afterDrain).toBe(1); + }); + + it("quarantines the overflowing invalidation subscription and retires its queued commands", () => { + const listeners = new Map(); + const cleanupScopes: string[] = []; + const owner = coordinator((scope, notify) => { + listeners.set(scope, notify); + return () => cleanupScopes.push(scope); + }); + const stormMemberships = Array.from({ length: 5 }, (_, index) => + active(owner, `storm-${index}`), + ); + let fired = false; + active(owner, "trigger", { + prepareCatalogChange: () => () => { + if (fired) return; + fired = true; + for (let scopeIndex = 0; scopeIndex < 5; scopeIndex += 1) { + const notify = listeners.get(`storm-${scopeIndex}`); + for (let generation = 1; generation <= 205; generation += 1) { + notify?.(invalidation(generation)); + } + } + }, + }); + listeners.get("trigger")?.(invalidation(1)); + expect(cleanupScopes).toEqual(["storm-4"]); + for (const membership of stormMemberships.slice(0, 4)) { + expect(capture(membership).expectedEpoch).toEqual(epoch(205)); + } + expect(capture(stormMemberships[4]!).expectedEpoch).toBeNull(); + }); +}); diff --git a/src/vnext/relation-catalog-epoch-coordinator.ts b/src/vnext/relation-catalog-epoch-coordinator.ts new file mode 100644 index 0000000..ef96b25 --- /dev/null +++ b/src/vnext/relation-catalog-epoch-coordinator.ts @@ -0,0 +1,1223 @@ +import { + MAX_CATALOG_SCOPE_LENGTH, + compareSqlCatalogEpoch, + decodeSqlCatalogInvalidation, + resolveSqlRelationCatalogProvider, +} from "./relation-catalog-boundary.js"; +import type { SqlCapturedRelationCatalogProviderContext } from "./relation-catalog-boundary.js"; +import type { SqlCatalogEpoch } from "./relation-completion-types.js"; + +export const MAX_CATALOG_LIVE_SCOPES = 128; +export const MAX_CATALOG_MEMBERSHIPS = 1_024; +export const MAX_CATALOG_MEMBERSHIPS_PER_SCOPE = 256; +export const MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW = 256; +export const MAX_CATALOG_EPOCH_COMMANDS = 1_024; +export const MAX_CATALOG_CLEANUPS_PER_BARRIER = 1_024; + +export interface SqlCatalogRevisionTarget { + readonly prepareCatalogChange: ( + this: void, + ) => ((this: void) => void) | null; +} + +const epochCaptureBrand: unique symbol = Symbol( + "SqlCatalogEpochCapture", +); +const EPOCH_CAPTURE_KIND = "SqlCatalogEpochCapture"; + +export interface SqlCatalogEpochCapture { + readonly [epochCaptureBrand]: "SqlCatalogEpochCapture"; + readonly expectedEpoch: SqlCatalogEpoch | null; +} + +export type SqlCatalogMembershipActivationResult = + | { readonly status: "active" } + | { + readonly status: "unavailable"; + readonly reason: + | "disposed" + | "scope-capacity" + | "scope-membership-capacity" + | "coordinator-disposed"; + }; + +export type SqlCatalogEpochCaptureResult = + | { + readonly status: "captured"; + readonly capture: SqlCatalogEpochCapture; + } + | { + readonly status: "unavailable"; + readonly reason: "inactive" | "disposed"; + }; + +export interface SqlCatalogScopeMembership { + readonly activate: ( + this: void, + ) => SqlCatalogMembershipActivationResult; + readonly captureEpoch: ( + this: void, + ) => SqlCatalogEpochCaptureResult; + readonly dispose: (this: void) => void; +} + +export type SqlCatalogScopeMembershipResult = + | { + readonly status: "prepared"; + readonly membership: SqlCatalogScopeMembership; + } + | { + readonly status: "unavailable"; + readonly reason: + | "disposed" + | "invalid-scope" + | "invalid-target" + | "membership-capacity"; + }; + +export type SqlCatalogResponseEpochDecision = + | { + readonly status: "usable"; + readonly observation: "baseline" | "equal"; + readonly epoch: SqlCatalogEpoch; + } + | { + readonly status: "superseded"; + readonly epoch: SqlCatalogEpoch; + } + | { + readonly status: "discarded"; + readonly reason: + | "stale" + | "token-conflict" + | "retired" + | "disposed" + | "malformed" + | "overloaded"; + }; + +export type SqlCatalogResponseEpochSubmissionResult = + | { + readonly status: "submitted"; + } + | { + readonly status: "settled"; + readonly decision: Extract< + SqlCatalogResponseEpochDecision, + { readonly status: "discarded" } + >; + }; + +export interface SqlCatalogEpochCoordinator { + readonly providerId: string; + readonly prepareScopeMembership: ( + this: void, + scope: unknown, + target: SqlCatalogRevisionTarget, + ) => SqlCatalogScopeMembershipResult; + readonly submitResponseEpoch: ( + this: void, + capture: unknown, + epoch: unknown, + onDecision: ( + this: void, + decision: SqlCatalogResponseEpochDecision, + ) => void, + ) => SqlCatalogResponseEpochSubmissionResult; + readonly dispose: (this: void) => void; +} + +export type SqlCatalogEpochCoordinatorResult = + | { + readonly status: "created"; + readonly coordinator: SqlCatalogEpochCoordinator; + } + | { + readonly status: "unavailable"; + readonly reason: "invalid-provider"; + }; + +interface CoordinatorState { + cleanupAdmissions: number; + cleanupDeferralDepth: number; + cleanupOverloaded: boolean; + disposed: boolean; + draining: boolean; + readonly captures: WeakMap; + readonly commands: EpochCommand[]; + readonly deferredCleanup: Set; + readonly memberships: Set; + readonly providerId: string; + readonly scopes: Map; + subscribe: SqlCapturedRelationCatalogProviderContext["subscribe"]; +} + +interface ScopeEntry { + readonly members: Set; + notificationSequence: number; + observedEpoch: SqlCatalogEpoch | null; + readonly scope: string; + subscription: SubscriptionState | null; + subscriptionAttempted: boolean; +} + +interface MembershipState { + active: boolean; + disposed: boolean; + disposedByCoordinator: boolean; + entry: ScopeEntry | null; + owner: CoordinatorState | null; + prepareCatalogChange: (this: void) => unknown; + readonly scope: string; +} + +interface CaptureState { + claimed: boolean; + readonly entry: ScopeEntry; + readonly membership: MembershipState; + readonly notificationSequence: number; + readonly owner: CoordinatorState; +} + +interface CallbackCell { + active: boolean; + callbackCount: number; + decode: + | ((value: unknown) => SqlCatalogEpoch | null) + | null; + decoding: boolean; + deliver: + | ((value: SqlCatalogEpoch) => void) + | null; + overload: (() => void) | null; + resetTimer: ReturnType | null; +} + +interface SubscriptionState { + readonly cell: CallbackCell; + cleanupCalled: boolean; + cleanup: Function | null; + failed: boolean; + installing: boolean; + readonly pending: SqlCatalogEpoch[]; +} + +interface InvalidationCommand { + readonly entry: ScopeEntry; + readonly kind: "invalidation"; + readonly subscription: SubscriptionState; + readonly epoch: SqlCatalogEpoch; +} + +interface ResponseCommand { + readonly capture: CaptureState; + readonly epoch: unknown; + readonly kind: "response"; + readonly onDecision: ( + this: void, + decision: SqlCatalogResponseEpochDecision, + ) => void; +} + +type EpochCommand = InvalidationCommand | ResponseCommand; + +const ACTIVE_RESULT: SqlCatalogMembershipActivationResult = + Object.freeze({ status: "active" }); +const SUBMITTED_RESULT: SqlCatalogResponseEpochSubmissionResult = + Object.freeze({ status: "submitted" }); +const NO_PREPARE_CATALOG_CHANGE = (): null => null; +const IGNORE_CLEANUP_REJECTION = (): void => {}; + +function unavailableActivation( + reason: Exclude< + SqlCatalogMembershipActivationResult, + { readonly status: "active" } + >["reason"], +): SqlCatalogMembershipActivationResult { + return Object.freeze({ reason, status: "unavailable" }); +} + +function discarded( + reason: Extract< + SqlCatalogResponseEpochDecision, + { readonly status: "discarded" } + >["reason"], +): Extract< + SqlCatalogResponseEpochDecision, + { readonly status: "discarded" } +> { + return Object.freeze({ reason, status: "discarded" }); +} + +function settled( + decision: Extract< + SqlCatalogResponseEpochDecision, + { readonly status: "discarded" } + >, +): SqlCatalogResponseEpochSubmissionResult { + return Object.freeze({ decision, status: "settled" }); +} + +function settleDecision( + onDecision: ( + this: void, + decision: SqlCatalogResponseEpochDecision, + ) => void, + decision: SqlCatalogResponseEpochDecision, +): void { + try { + onDecision(decision); + } catch { + // Consumers cannot break the serialized epoch gate. + } +} + +function isValidScope(candidate: unknown): candidate is string { + if ( + typeof candidate !== "string" || + candidate.length < 1 || + candidate.length > MAX_CATALOG_SCOPE_LENGTH + ) { + return false; + } + for (let index = 0; index < candidate.length; index += 1) { + const code = candidate.charCodeAt(index); + if (code === 0) return false; + if (code >= 0xd800 && code <= 0xdbff) { + if (index + 1 >= candidate.length) return false; + const next = candidate.charCodeAt(index + 1); + if (!(next >= 0xdc00 && next <= 0xdfff)) return false; + index += 1; + } else if (code >= 0xdc00 && code <= 0xdfff) { + return false; + } + } + return true; +} + +function snapshotAudience( + entry: ScopeEntry, +): readonly MembershipState[] { + const audience: MembershipState[] = []; + for (const member of entry.members) { + audience.push(member); + } + return audience; +} + +function resetCallbackAllowance(cell: CallbackCell): void { + cell.resetTimer = null; + cell.callbackCount = 0; +} + +function receiveSubscriptionCallback( + cell: CallbackCell, + value: unknown, +): void { + if (!cell.active) return; + cell.callbackCount += 1; + if (cell.resetTimer === null) { + cell.resetTimer = setTimeout( + resetCallbackAllowance, + 0, + cell, + ); + } + if ( + cell.callbackCount > + MAX_CATALOG_CALLBACKS_PER_RESET_WINDOW + ) { + const overload = cell.overload; + cell.active = false; + cell.deliver = null; + cell.overload = null; + clearTimeout(cell.resetTimer); + cell.resetTimer = null; + overload?.(); + return; + } + if (cell.decoding) return; + const decode = cell.decode; + if (!decode) return; + cell.decoding = true; + let decoded: SqlCatalogEpoch | null; + try { + decoded = decode(value); + } finally { + cell.decoding = false; + } + if (decoded) cell.deliver?.(decoded); +} + +function retireCallbackCell(cell: CallbackCell): void { + cell.active = false; + cell.callbackCount = 0; + cell.decode = null; + cell.decoding = false; + cell.deliver = null; + cell.overload = null; + if (cell.resetTimer !== null) { + clearTimeout(cell.resetTimer); + cell.resetTimer = null; + } +} + +function captureCleanup(candidate: unknown): Function | null { + return typeof candidate === "function" ? candidate : null; +} + +function cleanupSubscription(subscription: SubscriptionState): void { + if (subscription.cleanupCalled) { + subscription.cleanup = null; + return; + } + const cleanup = subscription.cleanup; + if (!cleanup) return; + subscription.cleanupCalled = true; + subscription.cleanup = null; + try { + const result = Reflect.apply(cleanup, undefined, []); + const settlement = new Promise((resolve) => { + resolve(result); + }); + void settlement.then(undefined, IGNORE_CLEANUP_REJECTION); + } catch { + // Provider cleanup is isolated after state is inert. + } +} + +function abandonSubscriptionCleanup( + subscription: SubscriptionState, +): void { + subscription.cleanupCalled = true; + subscription.cleanup = null; +} + +function quarantineCleanupOverflow( + state: CoordinatorState, + rejected: SubscriptionState, +): void { + abandonSubscriptionCleanup(rejected); + state.cleanupOverloaded = true; + disposeCoordinator(state); + for (const subscription of state.deferredCleanup) { + abandonSubscriptionCleanup(subscription); + } + state.deferredCleanup.clear(); +} + +function scheduleSubscriptionCleanup( + state: CoordinatorState, + subscription: SubscriptionState, +): void { + if (subscription.cleanupCalled) { + subscription.cleanup = null; + return; + } + if (state.cleanupOverloaded) { + abandonSubscriptionCleanup(subscription); + return; + } + if (state.deferredCleanup.has(subscription)) return; + if ( + state.cleanupAdmissions >= + MAX_CATALOG_CLEANUPS_PER_BARRIER + ) { + quarantineCleanupOverflow(state, subscription); + return; + } + state.cleanupAdmissions += 1; + state.deferredCleanup.add(subscription); + if (state.cleanupDeferralDepth === 0) { + flushDeferredCleanup(state); + } +} + +function enterCleanupBarrier(state: CoordinatorState): void { + if (state.cleanupDeferralDepth === 0) { + state.cleanupAdmissions = 0; + } + state.cleanupDeferralDepth += 1; +} + +function flushDeferredCleanup(state: CoordinatorState): void { + if (state.deferredCleanup.size === 0) return; + state.cleanupDeferralDepth = 1; + try { + while (state.deferredCleanup.size > 0) { + for (const subscription of state.deferredCleanup) { + state.deferredCleanup.delete(subscription); + cleanupSubscription(subscription); + break; + } + } + } finally { + state.cleanupDeferralDepth = 0; + if (!state.cleanupOverloaded) { + state.cleanupAdmissions = 0; + } + } +} + +function leaveCleanupBarrier(state: CoordinatorState): void { + state.cleanupDeferralDepth -= 1; + if (state.cleanupDeferralDepth === 0) { + flushDeferredCleanup(state); + if (!state.cleanupOverloaded) { + state.cleanupAdmissions = 0; + } + } +} + +function disableSubscription( + state: CoordinatorState, + entry: ScopeEntry, + subscription: SubscriptionState, +): void { + if (entry.subscription === subscription) { + entry.subscription = null; + } + subscription.failed = true; + subscription.pending.length = 0; + retireCallbackCell(subscription.cell); + scheduleSubscriptionCleanup(state, subscription); +} + +function enqueueCommand( + state: CoordinatorState, + command: EpochCommand, +): boolean { + if (state.commands.length >= MAX_CATALOG_EPOCH_COMMANDS) { + if (command.kind === "response") { + return false; + } else { + disableSubscription( + state, + command.entry, + command.subscription, + ); + } + return false; + } + state.commands.push(command); + if (!state.draining) drainCommands(state); + return true; +} + +function enqueueInvalidation( + state: CoordinatorState, + entry: ScopeEntry, + subscription: SubscriptionState, + epoch: SqlCatalogEpoch, +): void { + if ( + state.disposed || + entry.subscription !== subscription || + subscription.failed + ) { + return; + } + enqueueCommand(state, { + entry, + epoch, + kind: "invalidation", + subscription, + }); +} + +function decodeSubscriptionValue( + state: CoordinatorState, + entry: ScopeEntry, + subscription: SubscriptionState, + value: unknown, +): SqlCatalogEpoch | null { + const decoded = decodeSqlCatalogInvalidation(value); + if (decoded.status !== "accepted") return null; + if ( + state.disposed || + entry.subscription !== subscription || + subscription.failed + ) { + return null; + } + return decoded.value.epoch; +} + +function deliverSubscriptionValue( + state: CoordinatorState, + entry: ScopeEntry, + subscription: SubscriptionState, + epoch: SqlCatalogEpoch, +): void { + if (subscription.installing) { + subscription.pending.push(epoch); + } else { + enqueueInvalidation( + state, + entry, + subscription, + epoch, + ); + } +} + +function installSubscription( + state: CoordinatorState, + entry: ScopeEntry, +): void { + const subscribe = state.subscribe; + if ( + state.disposed || + entry.subscriptionAttempted || + !subscribe + ) { + entry.subscriptionAttempted = true; + return; + } + entry.subscriptionAttempted = true; + const cell: CallbackCell = { + active: true, + callbackCount: 0, + decode: null, + decoding: false, + deliver: null, + overload: null, + resetTimer: null, + }; + const subscription: SubscriptionState = { + cell, + cleanupCalled: false, + cleanup: null, + failed: false, + installing: true, + pending: [], + }; + entry.subscription = subscription; + cell.decode = ( + value: unknown, + ): SqlCatalogEpoch | null => + decodeSubscriptionValue( + state, + entry, + subscription, + value, + ); + cell.deliver = (value: SqlCatalogEpoch): void => { + deliverSubscriptionValue( + state, + entry, + subscription, + value, + ); + }; + cell.overload = (): void => { + disableSubscription(state, entry, subscription); + }; + let rawCleanup: unknown; + try { + rawCleanup = subscribe( + entry.scope, + (value: unknown): void => { + receiveSubscriptionCallback(cell, value); + }, + ); + } catch { + disableSubscription(state, entry, subscription); + return; + } + const cleanup = captureCleanup(rawCleanup); + if (!cleanup) { + disableSubscription(state, entry, subscription); + return; + } + subscription.cleanup = cleanup; + subscription.installing = false; + if ( + subscription.failed || + state.disposed || + entry.subscription !== subscription + ) { + disableSubscription(state, entry, subscription); + return; + } + const pending = subscription.pending.splice( + 0, + subscription.pending.length, + ); + const stageWholeFlush = !state.draining; + if (stageWholeFlush) state.draining = true; + try { + for (const epoch of pending) { + enqueueInvalidation( + state, + entry, + subscription, + epoch, + ); + } + } finally { + if (stageWholeFlush) { + state.draining = false; + drainCommands(state); + } + } +} + +interface PreparedRevision { + readonly dispatch: (this: void) => unknown; + readonly member: MembershipState; +} + +function prepareRevisions( + audience: readonly MembershipState[], + entry: ScopeEntry, +): readonly PreparedRevision[] { + const prepared: PreparedRevision[] = []; + for (const member of audience) { + if ( + !member.active || + member.disposed || + member.entry !== entry + ) { + continue; + } + let dispatch: unknown = null; + try { + const prepareCatalogChange = + member.prepareCatalogChange; + dispatch = prepareCatalogChange(); + } catch { + dispatch = null; + } + if ( + typeof dispatch === "function" && + member.active && + !member.disposed && + member.entry === entry + ) { + const dispatchFunction = dispatch; + prepared.push({ + dispatch: (): unknown => + Reflect.apply(dispatchFunction, undefined, []), + member, + }); + continue; + } + detachMembership(member); + } + return prepared; +} + +function dispatchRevisions( + prepared: readonly PreparedRevision[], + entry: ScopeEntry, +): void { + for (const item of prepared) { + if ( + !item.member.active || + item.member.disposed || + item.member.entry !== entry + ) { + continue; + } + try { + const dispatch = item.dispatch; + dispatch(); + } catch { + // Listener failures are isolated from the epoch gate. + } + } +} + +function processInvalidation( + state: CoordinatorState, + command: InvalidationCommand, +): void { + enterCleanupBarrier(state); + try { + if ( + state.disposed || + state.scopes.get(command.entry.scope) !== command.entry || + command.entry.subscription !== command.subscription || + command.subscription.failed + ) { + return; + } + const comparison = compareSqlCatalogEpoch( + command.entry.observedEpoch, + command.epoch, + ); + if ( + comparison.kind !== "baseline" && + comparison.kind !== "advance" + ) { + return; + } + const audience = snapshotAudience(command.entry); + command.entry.observedEpoch = comparison.epoch; + command.entry.notificationSequence += 1; + const prepared = prepareRevisions( + audience, + command.entry, + ); + dispatchRevisions(prepared, command.entry); + } finally { + leaveCleanupBarrier(state); + } +} + +function isLiveCapture( + state: CoordinatorState, + capture: CaptureState, +): boolean { + return ( + capture.owner === state && + capture.membership.active && + !capture.membership.disposed && + capture.membership.entry === capture.entry && + state.scopes.get(capture.entry.scope) === capture.entry + ); +} + +function processResponse( + state: CoordinatorState, + command: ResponseCommand, +): void { + enterCleanupBarrier(state); + try { + if (state.disposed) { + settleDecision(command.onDecision, discarded("disposed")); + return; + } + if (!isLiveCapture(state, command.capture)) { + settleDecision(command.onDecision, discarded("retired")); + return; + } + const entry = command.capture.entry; + const comparison = compareSqlCatalogEpoch( + entry.observedEpoch, + command.epoch, + ); + if (state.disposed) { + settleDecision(command.onDecision, discarded("disposed")); + return; + } + if (!isLiveCapture(state, command.capture)) { + settleDecision(command.onDecision, discarded("retired")); + return; + } + if (comparison.kind === "malformed") { + settleDecision(command.onDecision, discarded("malformed")); + return; + } + if (comparison.kind === "stale") { + settleDecision(command.onDecision, discarded("stale")); + return; + } + if (comparison.kind === "token-conflict") { + settleDecision( + command.onDecision, + discarded("token-conflict"), + ); + return; + } + if (comparison.kind === "equal") { + if ( + command.capture.notificationSequence !== + entry.notificationSequence + ) { + settleDecision( + command.onDecision, + Object.freeze({ + epoch: comparison.epoch, + status: "superseded", + }), + ); + } else { + settleDecision( + command.onDecision, + Object.freeze({ + epoch: comparison.epoch, + observation: "equal", + status: "usable", + }), + ); + } + return; + } + if (comparison.kind === "baseline") { + entry.observedEpoch = comparison.epoch; + settleDecision( + command.onDecision, + Object.freeze({ + epoch: comparison.epoch, + observation: "baseline", + status: "usable", + }), + ); + return; + } + const audience = snapshotAudience(entry); + entry.observedEpoch = comparison.epoch; + entry.notificationSequence += 1; + const prepared = prepareRevisions(audience, entry); + if (state.disposed) { + settleDecision(command.onDecision, discarded("disposed")); + return; + } + if (isLiveCapture(state, command.capture)) { + settleDecision( + command.onDecision, + Object.freeze({ + epoch: comparison.epoch, + status: "superseded", + }), + ); + } else { + settleDecision(command.onDecision, discarded("retired")); + } + dispatchRevisions(prepared, entry); + } finally { + leaveCleanupBarrier(state); + } +} + +function drainCommands(state: CoordinatorState): void { + state.draining = true; + try { + for (const command of state.commands) { + if (command.kind === "invalidation") { + processInvalidation(state, command); + } else { + processResponse(state, command); + } + } + } finally { + state.commands.length = 0; + state.draining = false; + } +} + +function detachMembership( + member: MembershipState, +): void { + if (member.disposed) return; + const owner = member.owner; + const entry = member.entry; + member.disposed = true; + member.active = false; + member.entry = null; + member.owner = null; + member.prepareCatalogChange = NO_PREPARE_CATALOG_CHANGE; + if (!owner) return; + owner.memberships.delete(member); + if (!entry) return; + entry.members.delete(member); + if (entry.members.size !== 0) return; + owner.scopes.delete(entry.scope); + const subscription = entry.subscription; + entry.subscription = null; + if (subscription) { + retireCallbackCell(subscription.cell); + scheduleSubscriptionCleanup(owner, subscription); + } +} + +function disposeMembership(member: MembershipState): void { + detachMembership(member); +} + +function createMembershipHandle( + member: MembershipState, +): SqlCatalogScopeMembership { + return Object.freeze({ + activate: (): SqlCatalogMembershipActivationResult => + activateMembership(member), + captureEpoch: (): SqlCatalogEpochCaptureResult => + captureMembershipEpoch(member), + dispose: (): void => { + disposeMembership(member); + }, + }); +} + +function activateMembership( + member: MembershipState, +): SqlCatalogMembershipActivationResult { + if (member.disposed) { + return unavailableActivation( + member.disposedByCoordinator + ? "coordinator-disposed" + : "disposed", + ); + } + const owner = member.owner; + if (!owner || owner.disposed) { + return unavailableActivation("coordinator-disposed"); + } + if (member.active) return ACTIVE_RESULT; + let entry = owner.scopes.get(member.scope); + if (!entry) { + if ( + owner.scopes.size >= MAX_CATALOG_LIVE_SCOPES + ) { + return unavailableActivation("scope-capacity"); + } + entry = { + members: new Set(), + notificationSequence: 0, + observedEpoch: null, + scope: member.scope, + subscription: null, + subscriptionAttempted: false, + }; + owner.scopes.set(member.scope, entry); + } + if ( + entry.members.size >= MAX_CATALOG_MEMBERSHIPS_PER_SCOPE + ) { + return unavailableActivation( + "scope-membership-capacity", + ); + } + member.entry = entry; + member.active = true; + entry.members.add(member); + installSubscription(owner, entry); + if (member.disposed) { + return unavailableActivation( + member.disposedByCoordinator + ? "coordinator-disposed" + : "disposed", + ); + } + return ACTIVE_RESULT; +} + +function captureMembershipEpoch( + member: MembershipState, +): SqlCatalogEpochCaptureResult { + const owner = member.owner; + if (member.disposed || !owner || owner.disposed) { + return Object.freeze({ + reason: "disposed", + status: "unavailable", + }); + } + const entry = member.entry; + if (!member.active || !entry) { + return Object.freeze({ + reason: "inactive", + status: "unavailable", + }); + } + const captureValue: SqlCatalogEpochCapture = { + [epochCaptureBrand]: EPOCH_CAPTURE_KIND, + expectedEpoch: entry.observedEpoch, + }; + const capture = Object.freeze(captureValue); + owner.captures.set(capture, { + claimed: false, + entry, + membership: member, + notificationSequence: entry.notificationSequence, + owner, + }); + return Object.freeze({ capture, status: "captured" }); +} + +function prepareMembership( + state: CoordinatorState, + scope: unknown, + target: SqlCatalogRevisionTarget, +): SqlCatalogScopeMembershipResult { + if (state.disposed) { + return Object.freeze({ + reason: "disposed", + status: "unavailable", + }); + } + if (!isValidScope(scope)) { + return Object.freeze({ + reason: "invalid-scope", + status: "unavailable", + }); + } + if (state.memberships.size >= MAX_CATALOG_MEMBERSHIPS) { + return Object.freeze({ + reason: "membership-capacity", + status: "unavailable", + }); + } + const member: MembershipState = { + active: false, + disposed: false, + disposedByCoordinator: false, + entry: null, + owner: state, + prepareCatalogChange: NO_PREPARE_CATALOG_CHANGE, + scope, + }; + state.memberships.add(member); + let prepareCandidate: unknown; + try { + prepareCandidate = target.prepareCatalogChange; + } catch { + const reason = + state.disposed || member.disposed + ? "disposed" + : "invalid-target"; + detachMembership(member); + return Object.freeze({ + reason, + status: "unavailable", + }); + } + if (state.disposed || member.disposed) { + detachMembership(member); + return Object.freeze({ + reason: "disposed", + status: "unavailable", + }); + } + if (typeof prepareCandidate !== "function") { + detachMembership(member); + return Object.freeze({ + reason: "invalid-target", + status: "unavailable", + }); + } + const prepareFunction = prepareCandidate; + member.prepareCatalogChange = (): unknown => + Reflect.apply(prepareFunction, undefined, []); + const membership = createMembershipHandle(member); + return Object.freeze({ membership, status: "prepared" }); +} + +function submitResponse( + state: CoordinatorState, + captureCandidate: unknown, + epoch: unknown, + onDecision: ( + this: void, + decision: SqlCatalogResponseEpochDecision, + ) => void, +): SqlCatalogResponseEpochSubmissionResult { + if (state.disposed) { + return settled(discarded("disposed")); + } + if ( + captureCandidate === null || + typeof captureCandidate !== "object" + ) { + return settled(discarded("malformed")); + } + const capture = state.captures.get(captureCandidate); + if (!capture) { + return settled(discarded("malformed")); + } + if (capture.claimed) { + return settled(discarded("retired")); + } + if (!isLiveCapture(state, capture)) { + return settled(discarded("retired")); + } + capture.claimed = true; + const admitted = enqueueCommand(state, { + capture, + epoch, + kind: "response", + onDecision, + }); + return admitted + ? SUBMITTED_RESULT + : settled(discarded("overloaded")); +} + +function disposeCoordinator(state: CoordinatorState): void { + if (state.disposed) return; + state.disposed = true; + state.subscribe = null; + const subscriptions: SubscriptionState[] = []; + for (const entry of state.scopes.values()) { + if (entry.subscription) { + subscriptions.push(entry.subscription); + entry.subscription = null; + } + entry.members.clear(); + } + state.scopes.clear(); + for (const member of state.memberships) { + member.active = false; + member.disposed = true; + member.disposedByCoordinator = true; + member.entry = null; + member.owner = null; + member.prepareCatalogChange = NO_PREPARE_CATALOG_CHANGE; + } + state.memberships.clear(); + for (const subscription of subscriptions) { + retireCallbackCell(subscription.cell); + } + for (const subscription of subscriptions) { + scheduleSubscriptionCleanup(state, subscription); + } + if (!state.draining) drainCommands(state); +} + +function createCoordinatorHandle( + state: CoordinatorState, +): SqlCatalogEpochCoordinator { + return Object.freeze({ + dispose: (): void => { + disposeCoordinator(state); + }, + prepareScopeMembership: ( + scope: unknown, + target: SqlCatalogRevisionTarget, + ): SqlCatalogScopeMembershipResult => + prepareMembership(state, scope, target), + providerId: state.providerId, + submitResponseEpoch: ( + capture: unknown, + epoch: unknown, + onDecision: ( + this: void, + decision: SqlCatalogResponseEpochDecision, + ) => void, + ): SqlCatalogResponseEpochSubmissionResult => + submitResponse(state, capture, epoch, onDecision), + }); +} + +export function createSqlCatalogEpochCoordinator( + capturedProvider: unknown, +): SqlCatalogEpochCoordinatorResult { + const provider = resolveSqlRelationCatalogProvider( + capturedProvider, + ); + if (!provider) { + return Object.freeze({ + reason: "invalid-provider", + status: "unavailable", + }); + } + const providerId = provider.id; + const subscribe = provider.subscribe; + const state: CoordinatorState = { + cleanupAdmissions: 0, + cleanupDeferralDepth: 0, + cleanupOverloaded: false, + captures: new WeakMap(), + commands: [], + deferredCleanup: new Set(), + disposed: false, + draining: false, + memberships: new Set(), + providerId, + scopes: new Map(), + subscribe, + }; + const coordinator = createCoordinatorHandle(state); + return Object.freeze({ coordinator, status: "created" }); +} diff --git a/src/vnext/relation-completion-types.ts b/src/vnext/relation-completion-types.ts index 6a7094a..f5a6e8c 100644 --- a/src/vnext/relation-completion-types.ts +++ b/src/vnext/relation-completion-types.ts @@ -9,9 +9,13 @@ import type { // Provisional package-private declarations until the vertical slice is proven. export interface SqlDisposable { - readonly dispose: () => void; + readonly dispose: (this: void) => void; } +export type SqlCatalogSubscriptionCleanup = ( + this: void, +) => void | PromiseLike; + export type SqlCatalogContainerRole = | "catalog" | "schema" @@ -125,7 +129,7 @@ export interface SqlRelationCatalogProvider { this: void, scope: string, onInvalidation: (event: SqlCatalogInvalidation) => void, - ) => SqlDisposable; + ) => SqlCatalogSubscriptionCleanup; } export type SqlIdentifierDecodeResult = diff --git a/test/vnext-types/marimo-relation-completion.test-d.ts b/test/vnext-types/marimo-relation-completion.test-d.ts index 8656ba1..1a6973f 100644 --- a/test/vnext-types/marimo-relation-completion.test-d.ts +++ b/test/vnext-types/marimo-relation-completion.test-d.ts @@ -14,8 +14,10 @@ import type { SqlCatalogRelation, SqlCatalogSearchRequest, SqlCatalogSearchResponse, + SqlCatalogSubscriptionCleanup, SqlCompletionCancellationReason, SqlCompletionIssue, + SqlDisposable, SqlRelationCompletionItem, SqlRelationCompletionList, SqlRelationCatalogProvider, @@ -125,7 +127,7 @@ const provider: SqlRelationCatalogProvider = { }, subscribe: (_scope, onInvalidation) => { onInvalidation({ epoch: { generation: 1, token: "tables-updated" } }); - return { dispose: () => undefined }; + return () => undefined; }, }; void provider; @@ -148,6 +150,21 @@ const receiverDependentProvider: SqlRelationCatalogProvider = { search: receiverDependentSearch, }; +const receiverDependentDispose = function ( + this: { readonly id: string }, +): void { + void this.id; +}; +const receiverDependentDisposable: SqlDisposable = { + // @ts-expect-error disposable callbacks are this-free closures + dispose: receiverDependentDispose, +}; + +// @ts-expect-error catalog cleanup callbacks are this-free closures +const receiverDependentCatalogCleanup: SqlCatalogSubscriptionCleanup = + receiverDependentDispose; +void receiverDependentCatalogCleanup; + const relationPath = [ { quoted: false, role: "catalog", value: "memory" }, { quoted: false, role: "schema", value: "main" }, @@ -328,5 +345,6 @@ void openWithRegions; void openWithoutRegions; void providerRenderedSql; void receiverDependentProvider; +void receiverDependentDisposable; void synchronousProvider; void undefinedContext;