diff --git a/docs/adr/0005-parser-independent-relation-completion.md b/docs/adr/0005-parser-independent-relation-completion.md index 9c4cb42..13d6ae7 100644 --- a/docs/adr/0005-parser-independent-relation-completion.md +++ b/docs/adr/0005-parser-independent-relation-completion.md @@ -484,6 +484,12 @@ state preparation and suppresses later revision listeners; an accidental Promise result receives best-effort detached rejection draining. Coordinator disposal revokes the preparation closure before external cleanup. The hook remains package-private and is not a provider or session extension point. +The combined search coordinator also installs one package-owned disposal +target. Epoch self-quarantine makes the outer coordinator inert before +subscription cleanup continues, so search work cannot outlive its epoch +authority. The target is receiver-free, invoked at most once, and its failure +cannot reopen disposal. It is synchronously exact-return and must return +`undefined`; any other runtime result is detached and rejection-drained. A search that discovers a higher epoch supersedes itself instead of publishing against its older captured revision. Pages and cache entries from different @@ -555,7 +561,9 @@ callback can always create a poisoned or independent unhandled rejection that no JavaScript library can retroactively contain. Legitimate asynchronous teardown must consume or report its own failure before the cleanup closure returns `undefined`. The valid `undefined` path returns without Promise -allocation or a microtask. A closure avoids a structural TypeScript +allocation or a microtask. Detached assimilation uses module-captured Promise +intrinsics, so replacing the global Promise constructor cannot disable the +drain. 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 @@ -654,6 +662,8 @@ outcomes settle through a discriminated request result. Completion is latest-wins per session: - a new completion request supersedes the previous request; +- a new request that cannot capture an active epoch still supersedes and + detaches the previous request before reporting its unavailable outcome; - same-key supersession atomically attaches the new request consumer, or retags the existing observer as that consumer, before removing old request ownership; different-key supersession revokes the old observer first; @@ -744,6 +754,25 @@ refresh notification and leaves the already-returned incomplete result valid. No optional catalog promise can keep `complete()` pending indefinitely or block the local baseline past its product response budget. +The first implementation increment of this section is intentionally +package-private. It combines an authenticated provider with the epoch +coordinator and owns the fixed 8-active/64-queued scheduler, exact-key +in-flight sharing, one latest-wins consumer per owner, independent +cancellation, absolute queue and execution deadlines, response decoding, and +epoch publication. An owner captures its scope and dialect when prepared, so +individual requests cannot substitute provider, scope, dialect, or epoch +authority. The authenticated dialect runtime owns its canonical provider ID; +callers cannot pair an unrelated ID and runtime. Establishing the first +baseline re-keys other joinable unobserved work in that scope, allowing +newly-observed consumers to join it without duplicating a provider call. + +Cache entries, loading/retry policy, refresh observers and their leases, the +40 ms completion-response budget, pagination composition, ranking, session +composition, and CodeMirror integration do not belong to that increment. They +remain explicit follow-up increments; the coordinator must not expose a +premature public surface that makes those deferred semantics difficult to add +or test. + The exact structural cache and shared-work key contains: - service-owned provider configuration identity and unique provider ID; diff --git a/src/vnext/__tests__/relation-catalog-boundary.test.ts b/src/vnext/__tests__/relation-catalog-boundary.test.ts index 93ef31a..0e610bc 100644 --- a/src/vnext/__tests__/relation-catalog-boundary.test.ts +++ b/src/vnext/__tests__/relation-catalog-boundary.test.ts @@ -5,6 +5,7 @@ import { createSqlCatalogSearchRequest, decodeSqlCatalogInvalidation, decodeSqlCatalogSearchResponse, + isValidSqlCatalogScope, MAX_CATALOG_CONTINUATION_TOKEN_LENGTH, MAX_CATALOG_DETAIL_LENGTH, MAX_CATALOG_ENTITY_ID_LENGTH, @@ -345,6 +346,29 @@ describe("relation catalog provider capture", () => { }); }); +describe("catalog scope validation", () => { + it("accepts only bounded, non-NUL, well-formed text", () => { + expect(isValidSqlCatalogScope("connection:primary")).toBe(true); + expect( + isValidSqlCatalogScope( + `\ud83d\ude80${"x".repeat(MAX_CATALOG_SCOPE_LENGTH - 2)}`, + ), + ).toBe(true); + for (const candidate of [ + null, + 1, + "", + "bad\0scope", + "\ud800", + "\ud800x", + "\udc00", + "x".repeat(MAX_CATALOG_SCOPE_LENGTH + 1), + ]) { + expect(isValidSqlCatalogScope(candidate)).toBe(false); + } + }); +}); + describe("catalog search request snapshots", () => { it("copies and recursively freezes the exact provider request", () => { const raw = request(); diff --git a/src/vnext/__tests__/relation-catalog-epoch-coordinator.test.ts b/src/vnext/__tests__/relation-catalog-epoch-coordinator.test.ts index 24413b3..fddbbab 100644 --- a/src/vnext/__tests__/relation-catalog-epoch-coordinator.test.ts +++ b/src/vnext/__tests__/relation-catalog-epoch-coordinator.test.ts @@ -236,6 +236,108 @@ describe("catalog epoch coordinator construction and membership", () => { expect(reads).toBe(0); }); + it("validates, drains, and invokes the package disposal target exactly once", async () => { + expect( + Reflect.apply(createSqlCatalogEpochCoordinator, undefined, [ + capturedProvider(), + undefined, + 1, + ]), + ).toEqual({ + reason: "invalid-disposal-target", + status: "unavailable", + }); + + let disposalCalls = 0; + const created = createSqlCatalogEpochCoordinator( + capturedProvider(), + undefined, + () => { + disposalCalls += 1; + throw new Error("package disposal target failed"); + }, + ); + expect(created.status).toBe("created"); + if (created.status !== "created") { + throw new Error("Expected a coordinator fixture"); + } + expect(() => { + created.coordinator.dispose(); + created.coordinator.dispose(); + }).not.toThrow(); + expect(disposalCalls).toBe(1); + + const invalidReturn = Promise.reject( + new Error("invalid async package disposal"), + ); + const withInvalidReturn = Reflect.apply( + createSqlCatalogEpochCoordinator, + undefined, + [ + capturedProvider(), + undefined, + () => invalidReturn, + ], + ); + expect(withInvalidReturn.status).toBe("created"); + if (withInvalidReturn.status !== "created") { + throw new Error("Expected a coordinator fixture"); + } + withInvalidReturn.coordinator.dispose(); + await Promise.resolve(); + await Promise.resolve(); + }); + + it("drains a plain thenable with captured Promise intrinsics after the global constructor is replaced", async () => { + const intrinsicPromise = Promise; + let thenCalls = 0; + const thenable = new Proxy( + {}, + { + get(target, property, receiver): unknown { + if (property === "then") { + return ( + _resolve: (value: unknown) => void, + reject: (reason?: unknown) => void, + ): void => { + thenCalls += 1; + reject(new Error("detached plain thenable")); + }; + } + return Reflect.get(target, property, receiver); + }, + }, + ); + function HostilePromise(): never { + throw new Error("mutable global Promise was used"); + } + expect( + Reflect.set(globalThis, "Promise", HostilePromise), + ).toBe(true); + try { + const created = Reflect.apply( + createSqlCatalogEpochCoordinator, + undefined, + [ + capturedProvider(), + undefined, + () => thenable, + ], + ); + expect(created.status).toBe("created"); + if (created.status !== "created") { + throw new Error("Expected a coordinator fixture"); + } + const createdCoordinator = created.coordinator; + expect(() => createdCoordinator.dispose()).not.toThrow(); + } finally { + Reflect.set(globalThis, "Promise", intrinsicPromise); + } + await intrinsicPromise.resolve(); + await intrinsicPromise.resolve(); + expect(thenCalls).toBe(1); + }); + it("validates exact bounded well-formed scopes without raw errors", () => { const owner = coordinator(); expect( diff --git a/src/vnext/__tests__/relation-catalog-search-work.bench.ts b/src/vnext/__tests__/relation-catalog-search-work.bench.ts new file mode 100644 index 0000000..1591d84 --- /dev/null +++ b/src/vnext/__tests__/relation-catalog-search-work.bench.ts @@ -0,0 +1,417 @@ +import { bench, describe } from "vitest"; +import { + captureSqlRelationCatalogProvider, +} from "../relation-catalog-boundary.js"; +import type { + CapturedSqlRelationCatalogProvider, + SqlCatalogBoundaryResult, +} from "../relation-catalog-boundary.js"; +import { + createSqlCatalogSearchWorkCoordinator, + MAX_CATALOG_ACTIVE_SEARCH_WORK, + MAX_CATALOG_EXECUTION_DEADLINE_MS, + MAX_CATALOG_QUEUED_SEARCH_WORK, + MAX_CATALOG_QUEUE_DEADLINE_MS, +} from "../relation-catalog-search-work.js"; +import type { + SqlCatalogSearchWorkCoordinator, + SqlCatalogSearchWorkInput, + SqlCatalogSearchWorkOutcome, + SqlCatalogSearchWorkOwner, + SqlCatalogSearchWorkTicket, +} from "../relation-catalog-search-work.js"; +import { + POSTGRESQL_SQL_RELATION_DIALECT, +} from "../relation-dialect.js"; +import type { + SqlCatalogSearchRequest, +} from "../relation-completion-types.js"; + +interface Deferred { + readonly promise: Promise; + readonly resolve: (value: unknown) => void; +} + +interface ProviderCall { + readonly request: SqlCatalogSearchRequest; + readonly settlement: Deferred; + readonly signal: AbortSignal; +} + +interface ProviderFixture { + readonly calls: ProviderCall[]; + readonly captured: CapturedSqlRelationCatalogProvider; + readonly emitInvalidation: ( + this: void, + generation: number, + ) => void; +} + +let providerSequence = 0; + +function benchmarkFailure(message: string): never { + throw new Error( + `Catalog search-work benchmark preflight failed: ${message}`, + ); +} + +function accepted( + result: SqlCatalogBoundaryResult, +): Value { + if (result.status !== "accepted") { + return benchmarkFailure( + `provider capture was rejected: ${result.reason}`, + ); + } + return result.value; +} + +function deferred(): Deferred { + let resolve: ((value: unknown) => void) | null = null; + const promise = new Promise((onResolve) => { + resolve = onResolve; + }); + return { + promise, + resolve: (value: unknown): void => { + if (!resolve) { + return benchmarkFailure("deferred resolver was unavailable"); + } + resolve(value); + }, + }; +} + +function readyResponse(generation = 1): unknown { + return { + coverage: { kind: "complete" }, + epoch: { + generation, + token: `benchmark-epoch-${generation}`, + }, + relations: [], + status: "ready", + }; +} + +function input(prefix: string): SqlCatalogSearchWorkInput { + return { + continuationToken: null, + limit: 20, + prefix: { quoted: false, value: prefix }, + qualifier: [{ quoted: false, value: "public" }], + searchPaths: [[{ quoted: false, value: "public" }]], + }; +} + +function providerFixture(subscribe: boolean): ProviderFixture { + const calls: ProviderCall[] = []; + let invalidation: + | ((this: void, event: unknown) => void) + | null = null; + providerSequence += 1; + const provider = subscribe + ? { + id: `search-work-benchmark-${providerSequence}`, + search( + request: SqlCatalogSearchRequest, + signal: AbortSignal, + ): Promise { + const settlement = deferred(); + calls.push({ request, settlement, signal }); + return settlement.promise; + }, + subscribe( + _scope: string, + listener: (this: void, event: unknown) => void, + ): (this: void) => undefined { + invalidation = listener; + return (): undefined => undefined; + }, + } + : { + id: `search-work-benchmark-${providerSequence}`, + search( + request: SqlCatalogSearchRequest, + signal: AbortSignal, + ): Promise { + const settlement = deferred(); + calls.push({ request, settlement, signal }); + return settlement.promise; + }, + }; + return { + calls, + captured: accepted( + captureSqlRelationCatalogProvider(provider), + ), + emitInvalidation: (generation: number): void => { + const listener = invalidation; + if (!listener) { + return benchmarkFailure( + "invalidation listener was unavailable", + ); + } + listener({ + epoch: { + generation, + token: `benchmark-epoch-${generation}`, + }, + }); + }, + }; +} + +function coordinator( + fixture: ProviderFixture, +): SqlCatalogSearchWorkCoordinator { + const created = createSqlCatalogSearchWorkCoordinator( + fixture.captured, + { + executionDeadlineMs: MAX_CATALOG_EXECUTION_DEADLINE_MS, + queueDeadlineMs: MAX_CATALOG_QUEUE_DEADLINE_MS, + }, + ); + if (created.status !== "created") { + return benchmarkFailure( + `coordinator creation was unavailable: ${created.reason}`, + ); + } + return created.coordinator; +} + +function owner( + service: SqlCatalogSearchWorkCoordinator, +): SqlCatalogSearchWorkOwner { + const prepared = service.prepareOwner( + "benchmark-scope", + POSTGRESQL_SQL_RELATION_DIALECT, + { + prepareCatalogChange: () => () => undefined, + }, + ); + if (prepared.status !== "prepared") { + return benchmarkFailure( + `owner preparation was unavailable: ${prepared.reason}`, + ); + } + const activated = prepared.owner.activate(); + if (activated.status !== "active") { + return benchmarkFailure( + `owner activation was unavailable: ${activated.reason}`, + ); + } + return prepared.owner; +} + +function requireCall( + fixture: ProviderFixture, + index: number, +): ProviderCall { + const call = fixture.calls[index]; + if (!call) { + return benchmarkFailure( + `provider call ${index} was unavailable`, + ); + } + return call; +} + +async function requireUsable( + ticket: SqlCatalogSearchWorkTicket, +): Promise { + const outcome = await ticket.result; + if (outcome.status !== "usable") { + benchmarkFailure( + `expected usable work, received ${outcome.status}`, + ); + } +} + +async function requireStatus( + ticket: SqlCatalogSearchWorkTicket, + status: SqlCatalogSearchWorkOutcome["status"], +): Promise { + const outcome = await ticket.result; + if (outcome.status !== status) { + benchmarkFailure( + `expected ${status} work, received ${outcome.status}`, + ); + } +} + +async function flushProviderCalls( + fixture: ProviderFixture, + expected: number, +): Promise { + for (let index = 0; index < expected; index += 1) { + for ( + let attempts = 0; + !fixture.calls[index] && attempts < 4; + attempts += 1 + ) { + await Promise.resolve(); + } + requireCall(fixture, index).settlement.resolve( + readyResponse(), + ); + await Promise.resolve(); + } +} + +async function benchmarkSameKeyJoins( + ownerCount: number, +): Promise { + const fixture = providerFixture(false); + const service = coordinator(fixture); + const tickets = Array.from( + { length: ownerCount }, + () => owner(service).request(input("shared")), + ); + if (fixture.calls.length !== 1) { + benchmarkFailure( + `${ownerCount} same-key owners created ${fixture.calls.length} calls`, + ); + } + requireCall(fixture, 0).settlement.resolve(readyResponse()); + await Promise.all(tickets.map(requireUsable)); + service.dispose(); +} + +async function benchmarkQueuePump(): Promise { + const fixture = providerFixture(false); + const service = coordinator(fixture); + const workCount = + MAX_CATALOG_ACTIVE_SEARCH_WORK + + MAX_CATALOG_QUEUED_SEARCH_WORK; + const tickets = Array.from( + { length: workCount }, + (_, index) => + owner(service).request(input(`queue-${index}`)), + ); + if ( + fixture.calls.length !== MAX_CATALOG_ACTIVE_SEARCH_WORK + ) { + benchmarkFailure("active search capacity was not filled"); + } + await flushProviderCalls(fixture, workCount); + await Promise.all(tickets.map(requireUsable)); + if (fixture.calls.length !== workCount) { + benchmarkFailure("queued search work was not fully pumped"); + } + service.dispose(); +} + +async function benchmarkWorstCaseExactKeyScan(): Promise { + const fixture = providerFixture(false); + const service = coordinator(fixture); + const workCount = + MAX_CATALOG_ACTIVE_SEARCH_WORK + + MAX_CATALOG_QUEUED_SEARCH_WORK; + const tickets = Array.from( + { length: workCount }, + (_, index) => + owner(service).request(input(`scan-${index}`)), + ); + const joined = owner(service).request( + input(`scan-${workCount - 1}`), + ); + if ( + fixture.calls.length !== MAX_CATALOG_ACTIVE_SEARCH_WORK + ) { + benchmarkFailure("worst-case exact key did not join queued work"); + } + for (const ticket of tickets) ticket.cancel(); + joined.cancel(); + await Promise.all([ + ...tickets.map((ticket) => + requireStatus(ticket, "cancelled"), + ), + requireStatus(joined, "cancelled"), + ]); + for (const call of fixture.calls) { + call.settlement.resolve(readyResponse()); + } + await Promise.resolve(); + await Promise.resolve(); + service.dispose(); +} + +async function benchmarkScopeRetirement(): Promise { + const fixture = providerFixture(true); + const service = coordinator(fixture); + const workCount = + MAX_CATALOG_ACTIVE_SEARCH_WORK + + MAX_CATALOG_QUEUED_SEARCH_WORK; + const tickets = Array.from( + { length: workCount }, + (_, index) => + owner(service).request(input(`epoch-${index}`)), + ); + fixture.emitInvalidation(1); + await Promise.all( + tickets.map((ticket) => + requireStatus(ticket, "superseded"), + ), + ); + for (const call of fixture.calls) { + if (!call.signal.aborted) { + benchmarkFailure( + "scope retirement left active provider work un-aborted", + ); + } + call.settlement.resolve(readyResponse()); + } + await Promise.resolve(); + await Promise.resolve(); + service.dispose(); +} + +async function benchmarkAcquireCancel10k(): Promise { + const fixture = providerFixture(false); + const service = coordinator(fixture); + const session = owner(service); + for (let index = 0; index < 10_000; index += 1) { + session.request(input(`cancel-${index}`)).cancel(); + } + if ( + fixture.calls.length !== MAX_CATALOG_ACTIVE_SEARCH_WORK + ) { + benchmarkFailure( + "cancelled active work did not retain its slot until settlement", + ); + } + for (const call of fixture.calls) { + if (!call.signal.aborted) { + benchmarkFailure("cancelled active work was not aborted"); + } + call.settlement.resolve(readyResponse()); + } + await Promise.resolve(); + await Promise.resolve(); + service.dispose(); +} + +describe("relation catalog search work", () => { + for (const ownerCount of [1, 10, 50]) { + bench(`${ownerCount} same-key owner joins`, async () => { + await benchmarkSameKeyJoins(ownerCount); + }); + } + + bench("pump 8 active and 64 queued searches", async () => { + await benchmarkQueuePump(); + }); + + bench("scan the worst-case exact key at capacity", async () => { + await benchmarkWorstCaseExactKeyScan(); + }); + + bench("retire a full scope on an epoch change", async () => { + await benchmarkScopeRetirement(); + }); + + bench("acquire and cancel 10,000 requests", async () => { + await benchmarkAcquireCancel10k(); + }); +}); diff --git a/src/vnext/__tests__/relation-catalog-search-work.test.ts b/src/vnext/__tests__/relation-catalog-search-work.test.ts new file mode 100644 index 0000000..a16aa0d --- /dev/null +++ b/src/vnext/__tests__/relation-catalog-search-work.test.ts @@ -0,0 +1,3105 @@ +import { describe, expect, it, vi } from "vitest"; +import { + captureSqlRelationCatalogProvider, + type CapturedSqlRelationCatalogProvider, + type SqlCatalogBoundaryResult, +} from "../relation-catalog-boundary.js"; +import { + createSqlCatalogSearchWorkCoordinator, + MAX_CATALOG_ACTIVE_SEARCH_WORK, + MAX_CATALOG_QUEUED_SEARCH_WORK, + type SqlCatalogSearchDeadlineScheduler, + type SqlCatalogSearchWorkCoordinator, + type SqlCatalogSearchWorkInput, + type SqlCatalogSearchWorkOptions, + type SqlCatalogSearchWorkOwner, + type SqlCatalogSearchWorkOutcome, + type SqlCatalogSearchWorkTicket, +} from "../relation-catalog-search-work.js"; +import { + DUCKDB_SQL_RELATION_DIALECT, + POSTGRESQL_SQL_RELATION_DIALECT, + type SqlRelationDialectRuntime, +} from "../relation-dialect.js"; +import type { SqlCatalogSearchRequest } from "../relation-completion-types.js"; +import type { SqlIdentifierComponent } from "../types.js"; + +interface Deferred { + readonly promise: Promise; + readonly reject: (reason?: unknown) => void; + readonly resolve: (value: Value) => void; +} + +interface ProviderCall { + readonly request: SqlCatalogSearchRequest; + readonly settlement: Deferred; + readonly signal: AbortSignal; +} + +interface ProviderHarness { + readonly calls: ProviderCall[]; + readonly captured: CapturedSqlRelationCatalogProvider; +} + +interface ScheduledDeadline { + active: boolean; + readonly callback: (this: void) => void; + readonly deadline: number; +} + +class ManualDeadlineScheduler + implements SqlCatalogSearchDeadlineScheduler +{ + nowValue = 0; + readonly tasks = new Map(); + private nextHandle = 1; + + readonly clearTimeout = (handle: unknown): void => { + if (typeof handle !== "number") return; + const task = this.tasks.get(handle); + if (task) task.active = false; + }; + + readonly now = (): number => this.nowValue; + + readonly setTimeout = ( + callback: (this: void) => void, + delayMs: number, + ): unknown => { + const handle = this.nextHandle; + this.nextHandle += 1; + this.tasks.set(handle, { + active: true, + callback, + deadline: this.nowValue + delayMs, + }); + return handle; + }; + + advanceBy(deltaMs: number): void { + this.nowValue += deltaMs; + this.flushDue(); + } + + flushDue(): void { + for (;;) { + const due = [...this.tasks.entries()] + .filter( + ([, task]) => + task.active && task.deadline <= this.nowValue, + ) + .sort( + ([leftHandle, left], [rightHandle, right]) => + left.deadline - right.deadline || + leftHandle - rightHandle, + )[0]; + if (!due) return; + const [handle, task] = due; + task.active = false; + this.tasks.delete(handle); + Reflect.apply(task.callback, undefined, []); + } + } + + get pendingCount(): number { + let count = 0; + for (const task of this.tasks.values()) { + if (task.active) count += 1; + } + return count; + } +} + +function accepted( + result: SqlCatalogBoundaryResult, +): Value { + expect(result.status).toBe("accepted"); + if (result.status !== "accepted") { + throw new Error(`Expected accepted, received ${result.reason}`); + } + return result.value; +} + +function deferred(): Deferred { + let resolve!: (value: Value) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((onResolve, onReject) => { + resolve = onResolve; + reject = onReject; + }); + return { promise, reject, resolve }; +} + +function epoch( + generation = 1, + token = `snapshot-${generation}`, +): { + readonly generation: number; + readonly token: string; +} { + return { generation, token }; +} + +function component( + value: string, + quoted = false, +): SqlIdentifierComponent { + return { quoted, value }; +} + +function pathComponent( + role: "relation" | "schema", + value: string, +) { + return { quoted: false, role, value }; +} + +function readyResponse(generation = 1) { + return { + coverage: { kind: "complete" }, + epoch: epoch(generation), + relations: [ + { + canonicalPath: [ + pathComponent("schema", "public"), + pathComponent("relation", `users-${generation}`), + ], + completionPathStart: 0, + entityId: `relation-${generation}`, + matchQuality: "exact", + relationKind: "table", + }, + ], + status: "ready", + }; +} + +function input( + prefix = "us", + overrides: Partial = {}, +): SqlCatalogSearchWorkInput { + return { + continuationToken: null, + limit: 20, + prefix: component(prefix), + qualifier: [component("public")], + searchPaths: [[component("public")]], + ...overrides, + }; +} + +function providerHarness(): ProviderHarness { + const calls: ProviderCall[] = []; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search( + request: SqlCatalogSearchRequest, + signal: AbortSignal, + ) { + const settlement = deferred(); + calls.push({ request, settlement, signal }); + return settlement.promise; + }, + }), + ); + return { calls, captured }; +} + +function coordinator( + captured: CapturedSqlRelationCatalogProvider, + options: SqlCatalogSearchWorkOptions = {}, +): SqlCatalogSearchWorkCoordinator { + const created = createSqlCatalogSearchWorkCoordinator( + captured, + options, + ); + expect(created.status).toBe("created"); + if (created.status !== "created") { + throw new Error( + `Expected coordinator, received ${created.reason}`, + ); + } + return created.coordinator; +} + +function owner( + service: SqlCatalogSearchWorkCoordinator, + scope = "connection:primary", + dialect: SqlRelationDialectRuntime = + POSTGRESQL_SQL_RELATION_DIALECT, +): SqlCatalogSearchWorkOwner { + const prepared = service.prepareOwner( + scope, + dialect, + { + prepareCatalogChange: () => () => {}, + }, + ); + expect(prepared.status).toBe("prepared"); + if (prepared.status !== "prepared") { + throw new Error(`Expected owner, received ${prepared.reason}`); + } + expect(prepared.owner.activate()).toEqual({ status: "active" }); + return prepared.owner; +} + +async function settled( + ticket: Promise, +): Promise { + await Promise.resolve(); + return ticket; +} + +async function flushMicrotasks(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +} + +describe("catalog search coordinator construction", () => { + it("rejects unauthenticated providers and malformed scheduler options", () => { + expect( + Reflect.apply( + createSqlCatalogSearchWorkCoordinator, + undefined, + [ + { + id: "catalog", + search: () => readyResponse(), + }, + ], + ), + ).toEqual({ + reason: "invalid-provider", + status: "unavailable", + }); + + const provider = providerHarness(); + for (const options of [ + { executionDeadlineMs: 9 }, + { executionDeadlineMs: 5_001 }, + { queueDeadlineMs: 9 }, + { queueDeadlineMs: 2_001 }, + { + deadlineScheduler: { + clearTimeout() {}, + now: 1, + setTimeout() { + return 1; + }, + }, + }, + ]) { + expect( + Reflect.apply( + createSqlCatalogSearchWorkCoordinator, + undefined, + [provider.captured, options], + ), + ).toEqual({ + reason: "invalid-options", + status: "unavailable", + }); + } + }); + + it("freezes public handles, tickets, outcomes, and decoded responses", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const ticket = session.request(input()); + expect(Object.isFrozen(service)).toBe(true); + expect(Object.isFrozen(session)).toBe(true); + expect(Object.isFrozen(ticket)).toBe(true); + + provider.calls[0]?.settlement.resolve(readyResponse()); + const outcome = await ticket.result; + expect(Object.isFrozen(outcome)).toBe(true); + expect(outcome.status).toBe("usable"); + if (outcome.status === "usable") { + expect(Object.isFrozen(outcome.response)).toBe(true); + } + }); + + it("does not expire default-scheduled work synchronously", async () => { + let calls = 0; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + calls += 1; + return readyResponse(); + }, + }), + ); + const service = coordinator(captured); + const ticket = owner(service).request(input()); + + expect(calls).toBe(1); + await expect(ticket.result).resolves.toMatchObject({ + status: "usable", + }); + }); + + it("rejects structurally copied dialect runtimes without invoking them", () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const result = service.prepareOwner( + "scope", + { ...POSTGRESQL_SQL_RELATION_DIALECT }, + { prepareCatalogChange: () => () => {} }, + ); + expect(result).toEqual({ + reason: "invalid-dialect", + status: "unavailable", + }); + }); + + it("returns closed tickets for inactive, malformed, and disposed owners", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const prepared = service.prepareOwner( + "scope", + POSTGRESQL_SQL_RELATION_DIALECT, + { prepareCatalogChange: () => () => {} }, + ); + expect(prepared.status).toBe("prepared"); + if (prepared.status !== "prepared") { + throw new Error("Expected prepared owner"); + } + + await expect( + prepared.owner.request(input()).result, + ).resolves.toEqual({ + reason: "inactive", + status: "unavailable", + }); + expect(prepared.owner.activate()).toEqual({ + status: "active", + }); + await expect( + prepared.owner.request(input("bad", { limit: 0 })).result, + ).resolves.toEqual({ + reason: "invalid-request", + status: "unavailable", + }); + prepared.owner.dispose(); + await expect( + prepared.owner.request(input()).result, + ).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(provider.calls).toHaveLength(0); + }); +}); + +describe("catalog search structural sharing and admission", () => { + it("shares exactly equal copied requests while preserving structural distinctions", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const first = owner(service); + const second = owner(service); + const sharedInput = input(); + const firstTicket = first.request(sharedInput); + const secondTicket = second.request({ + ...sharedInput, + prefix: { ...sharedInput.prefix }, + qualifier: sharedInput.qualifier.map((part) => ({ + ...part, + })), + searchPaths: sharedInput.searchPaths.map((path) => + path.map((part) => ({ ...part })), + ), + }); + + expect(provider.calls).toHaveLength(1); + expect(provider.calls[0]?.request).not.toBe(sharedInput); + expect(Object.isFrozen(provider.calls[0]?.request)).toBe(true); + + const distinct = [ + input("us", { continuationToken: "page-2" }), + input("us", { limit: 19 }), + input("US"), + input("us", { prefix: component("us", true) }), + input("us", { qualifier: [component("other")] }), + input("us", { searchPaths: [[component("other")]] }), + ]; + for (const candidate of distinct) { + owner(service).request(candidate); + } + owner( + service, + "connection:primary", + DUCKDB_SQL_RELATION_DIALECT, + ).request(input()); + expect(provider.calls).toHaveLength(2 + distinct.length); + expect(provider.calls.at(-1)?.request.dialectId).toBe( + "duckdb", + ); + + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(firstTicket.result).resolves.toMatchObject({ + status: "usable", + }); + await expect(secondTicket.result).resolves.toMatchObject({ + status: "usable", + }); + service.dispose(); + }); + + it("admits a same-key join at full queue capacity but rejects a new key", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + }); + const tickets = Array.from( + { + length: + MAX_CATALOG_ACTIVE_SEARCH_WORK + + MAX_CATALOG_QUEUED_SEARCH_WORK, + }, + (_, index) => + owner(service).request(input(`key-${index}`)), + ); + expect(provider.calls).toHaveLength( + MAX_CATALOG_ACTIVE_SEARCH_WORK, + ); + + const queuedIndex = MAX_CATALOG_ACTIVE_SEARCH_WORK; + const joined = owner(service).request( + input(`key-${queuedIndex}`), + ); + const rejected = owner(service).request(input("overflow")); + await expect(rejected.result).resolves.toEqual({ + reason: "overloaded", + status: "unavailable", + }); + + provider.calls[0]?.settlement.resolve(readyResponse()); + await Promise.resolve(); + await Promise.resolve(); + expect(provider.calls.at(-1)?.request.prefix.value).toBe( + `key-${queuedIndex}`, + ); + provider.calls.at(-1)?.settlement.resolve(readyResponse()); + await expect(tickets[queuedIndex]?.result).resolves.toMatchObject( + { status: "usable" }, + ); + await expect(joined.result).resolves.toMatchObject({ + status: "usable", + }); + service.dispose(); + }); + + it("promotes queued work FIFO without recursively invoking providers", async () => { + let depth = 0; + let maximumDepth = 0; + const order: string[] = []; + let service!: SqlCatalogSearchWorkCoordinator; + const reentrantTickets: SqlCatalogSearchWorkTicket[] = []; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search(request: SqlCatalogSearchRequest) { + depth += 1; + maximumDepth = Math.max(maximumDepth, depth); + order.push(request.prefix.value); + if ( + request.prefix.value === + `key-${MAX_CATALOG_ACTIVE_SEARCH_WORK}` && + reentrantTickets.length === 0 + ) { + reentrantTickets.push( + owner(service).request( + input("key-reentrant"), + ), + ); + } + depth -= 1; + return Promise.resolve(readyResponse()); + }, + }), + ); + service = coordinator(captured); + const tickets = Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK + 3 }, + (_, index) => owner(service).request(input(`key-${index}`)), + ); + await Promise.all(tickets.map((ticket) => ticket.result)); + const reentrantTicket = reentrantTickets[0]; + if (!reentrantTicket) { + throw new Error("Expected reentrant queued ticket"); + } + await reentrantTicket.result; + + expect(order).toEqual( + [ + ...Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK + 3 }, + (_, index) => `key-${index}`, + ), + "key-reentrant", + ], + ); + expect(maximumDepth).toBe(1); + }); +}); + +describe("catalog search ownership and active-slot lifecycle", () => { + it("keeps cancellation idempotent before and after settlement", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const cancelled = owner(service).request(input("cancelled")); + cancelled.cancel(); + cancelled.cancel(); + await expect(cancelled.result).resolves.toEqual({ + status: "cancelled", + }); + + const completed = owner(service).request(input("completed")); + provider.calls.at(-1)?.settlement.resolve(readyResponse()); + await expect(completed.result).resolves.toMatchObject({ + status: "usable", + }); + completed.cancel(); + completed.cancel(); + service.dispose(); + }); + + it("transfers same-key latest-wins ownership before detaching the old consumer", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const first = session.request(input("same")); + const signal = provider.calls[0]?.signal; + const replacement = session.request(input("same")); + + await expect(first.result).resolves.toEqual({ + status: "superseded", + }); + expect(provider.calls).toHaveLength(1); + expect(signal?.aborted).toBe(false); + + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(replacement.result).resolves.toMatchObject({ + status: "usable", + }); + }); + + it("aborts only after the last shared owner leaves and retains the active slot until settlement", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const firstOwner = owner(service); + const secondOwner = owner(service); + const first = firstOwner.request(input("shared")); + const second = secondOwner.request(input("shared")); + const filling = Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK - 1 }, + (_, index) => owner(service).request(input(`active-${index}`)), + ); + const queued = owner(service).request(input("queued")); + const sharedCall = provider.calls[0]; + + first.cancel(); + await expect(first.result).resolves.toEqual({ + status: "cancelled", + }); + expect(sharedCall?.signal.aborted).toBe(false); + + second.cancel(); + await expect(second.result).resolves.toEqual({ + status: "cancelled", + }); + expect(sharedCall?.signal.aborted).toBe(true); + expect(provider.calls).toHaveLength( + MAX_CATALOG_ACTIVE_SEARCH_WORK, + ); + + sharedCall?.settlement.resolve(readyResponse()); + await Promise.resolve(); + await Promise.resolve(); + expect(provider.calls).toHaveLength( + MAX_CATALOG_ACTIVE_SEARCH_WORK + 1, + ); + expect(provider.calls.at(-1)?.request.prefix.value).toBe( + "queued", + ); + + for (const call of provider.calls.slice(1)) { + call.settlement.resolve(readyResponse()); + } + await Promise.all([ + ...filling.map((ticket) => settled(ticket.result)), + settled(queued.result), + ]); + }); + + it("supersedes a different key and aborts the abandoned work exactly once", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const first = session.request(input("first")); + let aborts = 0; + provider.calls[0]?.signal.addEventListener("abort", () => { + aborts += 1; + }); + const replacement = session.request(input("replacement")); + + await expect(first.result).resolves.toEqual({ + status: "superseded", + }); + expect(provider.calls).toHaveLength(2); + expect(provider.calls[0]?.signal.aborted).toBe(true); + expect(aborts).toBe(1); + + provider.calls[0]?.settlement.reject( + new Error("late abandoned rejection"), + ); + provider.calls[1]?.settlement.resolve(readyResponse()); + await expect(replacement.result).resolves.toMatchObject({ + status: "usable", + }); + await flushMicrotasks(); + expect(aborts).toBe(1); + }); + + it("keeps one deterministic current owner when an abandoned abort reenters with a third key", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + }); + const session = owner(service); + const abandoned = session.request(input("abandoned")); + const reentrantTickets: SqlCatalogSearchWorkTicket[] = []; + provider.calls[0]?.signal.addEventListener("abort", () => { + reentrantTickets.push( + session.request(input("reentrant")), + ); + }); + + const displaced = session.request(input("displaced")); + await expect(abandoned.result).resolves.toEqual({ + status: "superseded", + }); + await expect(displaced.result).resolves.toEqual({ + status: "superseded", + }); + expect( + provider.calls.map((call) => call.request.prefix.value), + ).toEqual(["abandoned", "reentrant"]); + expect(provider.calls[0]?.signal.aborted).toBe(true); + + const reentrantTicket = reentrantTickets[0]; + if (!reentrantTicket) { + throw new Error("Expected abort-listener reentrant request"); + } + provider.calls[0]?.settlement.reject( + new Error("late abandoned rejection"), + ); + provider.calls[1]?.settlement.resolve(readyResponse()); + await expect(reentrantTicket.result).resolves.toMatchObject({ + status: "usable", + }); + displaced.cancel(); + abandoned.cancel(); + await flushMicrotasks(); + expect(provider.calls).toHaveLength(2); + expect(scheduler.pendingCount).toBe(0); + }); +}); + +describe("catalog search absolute deadlines", () => { + it("does not retain or invoke work when a deadline expires while it is armed", async () => { + const cases = [ + { + expectedReason: "execution-timeout", + nowValues: [0, 0, 0, 0, 20], + }, + ] as const; + for (const testCase of cases) { + let index = 0; + let providerCalls = 0; + const activeHandles = new Set(); + let nextHandle = 0; + const scheduler: SqlCatalogSearchDeadlineScheduler = { + clearTimeout(handle) { + if (typeof handle === "number") { + activeHandles.delete(handle); + } + }, + now() { + const value = testCase.nowValues[index]; + index += 1; + return ( + value ?? + testCase.nowValues.at(-1) ?? + Number.NaN + ); + }, + setTimeout() { + nextHandle += 1; + activeHandles.add(nextHandle); + return nextHandle; + }, + }; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + providerCalls += 1; + return readyResponse(); + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 20, + queueDeadlineMs: 10, + synchronousBudgetMs: 5, + }); + const ticket = owner(service).request(input()); + + await expect(ticket.result).resolves.toEqual({ + reason: testCase.expectedReason, + status: "unavailable", + }); + expect(providerCalls).toBe(1); + expect(activeHandles.size).toBe(0); + service.dispose(); + } + }); + + it("fails closed when adding a duration would overflow the clock", async () => { + let providerCalls = 0; + let timerCalls = 0; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + providerCalls += 1; + return readyResponse(); + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: { + clearTimeout() {}, + now: () => Number.MAX_VALUE, + setTimeout() { + timerCalls += 1; + return 1; + }, + }, + }); + const ticket = owner(service).request(input()); + + await expect(ticket.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + expect(providerCalls).toBe(0); + expect(timerCalls).toBe(0); + service.dispose(); + }); + + it("does not let a late joiner extend the first enqueue deadline", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 20, + queueDeadlineMs: 10, + synchronousBudgetMs: 5, + }); + Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK }, + (_, index) => owner(service).request(input(`active-${index}`)), + ); + const first = owner(service).request(input("queued")); + scheduler.advanceBy(9); + const joined = owner(service).request(input("queued")); + scheduler.advanceBy(1); + + await expect(first.result).resolves.toEqual({ + reason: "queue-timeout", + status: "unavailable", + }); + await expect(joined.result).resolves.toEqual({ + reason: "queue-timeout", + status: "unavailable", + }); + expect(provider.calls).toHaveLength( + MAX_CATALOG_ACTIVE_SEARCH_WORK, + ); + service.dispose(); + }); + + it("checks the absolute queue deadline during promotion when the timer is delayed", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 20, + queueDeadlineMs: 10, + synchronousBudgetMs: 5, + }); + Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK }, + (_, index) => owner(service).request(input(`active-${index}`)), + ); + const queued = owner(service).request(input("queued")); + + scheduler.nowValue = 10; + provider.calls[0]?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + + await expect(queued.result).resolves.toEqual({ + reason: "queue-timeout", + status: "unavailable", + }); + expect(provider.calls).toHaveLength( + MAX_CATALOG_ACTIVE_SEARCH_WORK, + ); + service.dispose(); + }); + + it("does not let a late joiner or delayed timer publish at the execution deadline", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 20, + queueDeadlineMs: 10, + synchronousBudgetMs: 5, + }); + const first = owner(service).request(input("active")); + scheduler.nowValue = 19; + const joined = owner(service).request(input("active")); + scheduler.nowValue = 20; + provider.calls[0]?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + + await expect(first.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + await expect(joined.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + expect(provider.calls[0]?.signal.aborted).toBe(true); + expect(scheduler.pendingCount).toBe(0); + }); + + it("discards a synchronous provider return that exceeds the observation budget", async () => { + const scheduler = new ManualDeadlineScheduler(); + let signal: AbortSignal | undefined; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search( + _request: SqlCatalogSearchRequest, + nextSignal: AbortSignal, + ) { + signal = nextSignal; + scheduler.nowValue += 6; + return readyResponse(); + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 20, + queueDeadlineMs: 10, + synchronousBudgetMs: 5, + }); + const ticket = owner(service).request(input()); + + await expect(ticket.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + expect(signal?.aborted).toBe(true); + expect(scheduler.pendingCount).toBe(0); + }); + + it("does not let synchronously repeated early timer callbacks violate the absolute deadline", async () => { + let timerCallback: (() => void) | undefined; + let providerCalls = 0; + let synchronousFirings = 0; + const delays: number[] = []; + const scheduler: SqlCatalogSearchDeadlineScheduler = { + clearTimeout() {}, + now: () => 0, + setTimeout(callback, delayMs) { + delays.push(delayMs); + timerCallback = callback; + if ( + delayMs === 20 && + synchronousFirings < 1 + ) { + synchronousFirings += 1; + callback(); + callback(); + } + return 1; + }, + }; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + providerCalls += 1; + return readyResponse(); + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 20, + queueDeadlineMs: 10, + synchronousBudgetMs: 5, + }); + const ticket = owner(service).request(input()); + + expect({ + delays, + providerCalls, + synchronousFirings, + }).toEqual({ + delays: [20, 20], + providerCalls: 1, + synchronousFirings: 1, + }); + await expect(ticket.result).resolves.toMatchObject({ + status: "usable", + }); + expect(() => timerCallback?.()).not.toThrow(); + }); +}); + +describe("catalog provider failures and hostile settlement", () => { + it("maps synchronous throws, rejections, and malformed responses to closed outcomes", async () => { + const values: readonly { + readonly expected: SqlCatalogSearchWorkOutcome; + readonly search: () => unknown; + }[] = [ + { + expected: { + reason: "provider-failed", + status: "unavailable", + }, + search: () => { + throw new Error("synchronous provider failure"); + }, + }, + { + expected: { + reason: "provider-failed", + status: "unavailable", + }, + search: () => + Promise.reject(new Error("asynchronous provider failure")), + }, + { + expected: { + reason: "malformed-response", + status: "unavailable", + }, + search: () => ({ status: "ready" }), + }, + ]; + + for (const { expected, search } of values) { + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search, + }), + ); + const service = coordinator(captured); + await expect( + owner(service).request(input()).result, + ).resolves.toEqual(expected); + service.dispose(); + } + }); + + it("settles once when a hostile thenable resolves and rejects repeatedly", async () => { + let calls = 0; + const thenable: Record = {}; + Object.defineProperty(thenable, ["th", "en"].join(""), { + value( + resolve: (value: unknown) => void, + reject: (reason: unknown) => void, + ) { + calls += 1; + resolve(readyResponse()); + reject(new Error("late rejection")); + resolve({ status: "invalid" }); + }, + }); + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + return thenable; + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: new ManualDeadlineScheduler(), + }); + const ticket = owner(service).request(input()); + + await expect(ticket.result).resolves.toMatchObject({ + status: "usable", + }); + expect(calls).toBe(1); + }); + + it("drains a rejected native promise without reading poisoned own then or catch properties", async () => { + let thenReads = 0; + let catchReads = 0; + const thenProperty = ["th", "en"].join(""); + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + const poisoned = Promise.reject( + new Error("poisoned provider rejection"), + ); + Object.defineProperty(poisoned, thenProperty, { + get() { + thenReads += 1; + throw new Error("poisoned own then"); + }, + }); + Object.defineProperty(poisoned, "catch", { + get() { + catchReads += 1; + throw new Error("poisoned own catch"); + }, + }); + return poisoned; + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: new ManualDeadlineScheduler(), + }); + + await expect( + owner(service).request(input()).result, + ).resolves.toEqual({ + reason: "provider-failed", + status: "unavailable", + }); + expect(thenReads).toBe(0); + expect(catchReads).toBe(0); + }); + + it("handles a throwing then getter that reenters with a replacement", async () => { + let service!: SqlCatalogSearchWorkCoordinator; + let session!: SqlCatalogSearchWorkOwner; + const reentrantTickets: SqlCatalogSearchWorkTicket[] = []; + let thenReads = 0; + let calls = 0; + const thenable: Record = {}; + Object.defineProperty(thenable, ["th", "en"].join(""), { + get() { + thenReads += 1; + reentrantTickets.push( + session.request(input("replacement")), + ); + throw new Error("hostile provider then getter"); + }, + }); + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + calls += 1; + return calls === 1 ? thenable : readyResponse(); + }, + }), + ); + service = coordinator(captured, { + deadlineScheduler: new ManualDeadlineScheduler(), + }); + session = owner(service); + const first = session.request(input("hostile")); + + await expect(first.result).resolves.toEqual({ + status: "superseded", + }); + const replacement = reentrantTickets[0]; + if (!replacement) { + throw new Error("Expected then-getter replacement"); + } + await expect(replacement.result).resolves.toMatchObject({ + status: "usable", + }); + expect(thenReads).toBe(1); + expect(calls).toBe(2); + }); + + it("invokes the captured provider receiver-free with immutable input and a work-owned signal", async () => { + let observedThisIsUndefined = false; + let requestFrozen = false; + let observedSignal: AbortSignal | undefined; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search: function ( + this: void, + request: SqlCatalogSearchRequest, + signal: AbortSignal, + ) { + observedThisIsUndefined = this === undefined; + requestFrozen = Object.isFrozen(request); + observedSignal = signal; + return readyResponse(); + }, + }), + ); + const service = coordinator(captured); + const ticket = owner(service).request(input()); + + await expect(ticket.result).resolves.toMatchObject({ + status: "usable", + }); + expect(observedThisIsUndefined).toBe(true); + expect(requestFrozen).toBe(true); + expect(observedSignal).toBeInstanceOf(AbortSignal); + }); +}); + +describe("catalog search epoch authority and isolation", () => { + it("keeps shared response authority when the first owner is disposed", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const firstOwner = owner(service); + const remainingOwner = owner(service); + const first = firstOwner.request(input("shared")); + const remaining = remainingOwner.request(input("shared")); + + firstOwner.dispose(); + await expect(first.result).resolves.toEqual({ + status: "cancelled", + }); + expect(provider.calls[0]?.signal.aborted).toBe(false); + + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(remaining.result).resolves.toMatchObject({ + status: "usable", + }); + }); + + it("rekeys same-scope unobserved work after a baseline without crossing scope boundaries", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const baseline = owner(service, "scope-a").request( + input("baseline"), + ); + const pendingInScope = owner(service, "scope-a").request( + input("pending"), + ); + const pendingInOtherScope = owner(service, "scope-b").request( + input("pending"), + ); + + expect(provider.calls).toHaveLength(3); + expect( + provider.calls.map((call) => call.request.expectedEpoch), + ).toEqual([null, null, null]); + + provider.calls[0]?.settlement.resolve(readyResponse(7)); + await expect(baseline.result).resolves.toMatchObject({ + observation: "baseline", + status: "usable", + }); + + const joinedInScope = owner(service, "scope-a").request( + input("pending"), + ); + const joinedInOtherScope = owner(service, "scope-b").request( + input("pending"), + ); + expect(provider.calls).toHaveLength(3); + + provider.calls[1]?.settlement.resolve(readyResponse(7)); + const [originalScopeResult, joinedScopeResult] = + await Promise.all([ + pendingInScope.result, + joinedInScope.result, + ]); + expect(originalScopeResult).toMatchObject({ + observation: "equal", + status: "usable", + }); + expect(joinedScopeResult).toBe(originalScopeResult); + + provider.calls[2]?.settlement.resolve(readyResponse(11)); + const [originalOtherResult, joinedOtherResult] = + await Promise.all([ + pendingInOtherScope.result, + joinedInOtherScope.result, + ]); + expect(originalOtherResult).toMatchObject({ + observation: "baseline", + status: "usable", + }); + expect(joinedOtherResult).toBe(originalOtherResult); + }); + + it("retires same-scope work before abort and revision dispatch while isolating another scope", async () => { + const listeners = new Map< + string, + (event: unknown) => void + >(); + const calls: ProviderCall[] = []; + const events: string[] = []; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search( + request: SqlCatalogSearchRequest, + signal: AbortSignal, + ) { + const settlement = deferred(); + signal.addEventListener("abort", () => { + events.push(`abort-${request.scope}`); + }); + calls.push({ request, settlement, signal }); + return settlement.promise; + }, + subscribe( + scope: string, + listener: (event: unknown) => void, + ) { + listeners.set(scope, listener); + return () => undefined; + }, + }), + ); + const service = coordinator(captured); + const prepareOwner = (scope: string) => { + const prepared = service.prepareOwner( + scope, + POSTGRESQL_SQL_RELATION_DIALECT, + { + prepareCatalogChange: () => { + events.push(`prepare-${scope}`); + return () => { + events.push(`dispatch-${scope}`); + }; + }, + }, + ); + if (prepared.status !== "prepared") { + throw new Error("Expected prepared owner"); + } + expect(prepared.owner.activate()).toEqual({ + status: "active", + }); + return prepared.owner; + }; + const firstScope = prepareOwner("scope-a").request(input("a")); + const otherScope = prepareOwner("scope-b").request(input("b")); + + listeners.get("scope-a")?.({ epoch: epoch(1) }); + await expect(firstScope.result).resolves.toEqual({ + status: "superseded", + }); + expect(events).toEqual([ + "prepare-scope-a", + "abort-scope-a", + "dispatch-scope-a", + ]); + expect(calls[0]?.signal.aborted).toBe(true); + expect(calls[1]?.signal.aborted).toBe(false); + + calls[1]?.settlement.resolve(readyResponse()); + await expect(otherScope.result).resolves.toMatchObject({ + status: "usable", + }); + }); + + it("supersedes reentrant work when its retiring membership makes the next capture fail", async () => { + const listener: { + current: ((event: unknown) => void) | null; + } = { current: null }; + const calls: ProviderCall[] = []; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search( + request: SqlCatalogSearchRequest, + signal: AbortSignal, + ) { + const settlement = deferred(); + calls.push({ request, settlement, signal }); + return settlement.promise; + }, + subscribe( + _scope: string, + onInvalidation: (event: unknown) => void, + ) { + listener.current = onInvalidation; + return () => undefined; + }, + }), + ); + const service = coordinator(captured); + let session: SqlCatalogSearchWorkOwner | null = null; + const reentrant: { + current: SqlCatalogSearchWorkTicket | null; + } = { current: null }; + const prepared = service.prepareOwner( + "scope", + POSTGRESQL_SQL_RELATION_DIALECT, + { + prepareCatalogChange: () => { + reentrant.current = + session?.request(input("reentrant")) ?? null; + return null; + }, + }, + ); + expect(prepared.status).toBe("prepared"); + if (prepared.status !== "prepared") { + throw new Error("Expected prepared owner"); + } + session = prepared.owner; + expect(session.activate()).toEqual({ status: "active" }); + const initial = session.request(input("initial")); + + listener.current?.({ epoch: epoch(1) }); + await expect(initial.result).resolves.toEqual({ + status: "superseded", + }); + expect(reentrant.current).not.toBeNull(); + expect(calls).toHaveLength(2); + expect(calls[0]?.signal.aborted).toBe(true); + expect(calls[1]?.signal.aborted).toBe(false); + + const afterRetirement = session.request(input("after")); + await expect(afterRetirement.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + if (!reentrant.current) { + throw new Error("Expected reentrant request"); + } + await expect(reentrant.current.result).resolves.toEqual({ + status: "superseded", + }); + expect(calls[1]?.signal.aborted).toBe(true); + }); + + it("makes a higher response self-supersede and retire other same-scope work only", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const baselineOwner = owner(service, "scope-a"); + const baseline = baselineOwner.request(input("baseline")); + provider.calls[0]?.settlement.resolve(readyResponse(1)); + await expect(baseline.result).resolves.toMatchObject({ + status: "usable", + }); + + const producing = owner(service, "scope-a").request( + input("producing"), + ); + const sameScope = owner(service, "scope-a").request( + input("same-scope"), + ); + const otherScope = owner(service, "scope-b").request( + input("other-scope"), + ); + const producingCall = provider.calls.find( + (call) => call.request.prefix.value === "producing", + ); + const sameScopeCall = provider.calls.find( + (call) => call.request.prefix.value === "same-scope", + ); + const otherScopeCall = provider.calls.find( + (call) => call.request.prefix.value === "other-scope", + ); + + producingCall?.settlement.resolve(readyResponse(2)); + await expect(producing.result).resolves.toEqual({ + status: "superseded", + }); + await expect(sameScope.result).resolves.toEqual({ + status: "superseded", + }); + expect(producingCall?.signal.aborted).toBe(false); + expect(sameScopeCall?.signal.aborted).toBe(true); + expect(otherScopeCall?.signal.aborted).toBe(false); + + otherScopeCall?.settlement.resolve(readyResponse()); + await expect(otherScope.result).resolves.toMatchObject({ + status: "usable", + }); + }); + + it("fails closed for stale, token-conflicting, and malformed response epochs", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const baseline = session.request(input("baseline")); + provider.calls[0]?.settlement.resolve(readyResponse(2)); + await expect(baseline.result).resolves.toMatchObject({ + status: "usable", + }); + + const stale = session.request(input("stale")); + provider.calls[1]?.settlement.resolve(readyResponse(1)); + await expect(stale.result).resolves.toEqual({ + status: "superseded", + }); + + const conflicting = session.request(input("conflicting")); + provider.calls[2]?.settlement.resolve({ + ...readyResponse(2), + epoch: epoch(2, "conflicting-token"), + }); + await expect(conflicting.result).resolves.toEqual({ + status: "superseded", + }); + + const malformed = session.request(input("malformed")); + provider.calls[3]?.settlement.resolve({ + ...readyResponse(3), + epoch: { generation: -1, token: "invalid" }, + }); + await expect(malformed.result).resolves.toEqual({ + reason: "malformed-response", + status: "unavailable", + }); + }); +}); + +describe("catalog search disposal and late settlement", () => { + it("settles active and queued owners before abort and drains late rejection", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 20, + queueDeadlineMs: 10, + synchronousBudgetMs: 5, + }); + const tickets = Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK + 1 }, + (_, index) => owner(service).request(input(`key-${index}`)), + ); + const events: string[] = []; + let reentrantStatus: string | undefined; + provider.calls[0]?.signal.addEventListener("abort", () => { + events.push("abort"); + reentrantStatus = service.prepareOwner( + "reentrant", + POSTGRESQL_SQL_RELATION_DIALECT, + { + prepareCatalogChange: () => () => {}, + }, + ).status; + void tickets[0]?.result.then(() => { + events.push("settled-before-abort"); + }); + }); + + service.dispose(); + await expect(tickets.at(-1)?.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + await Promise.all( + tickets.map((ticket) => + expect(ticket.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }), + ), + ); + expect(provider.calls).toHaveLength( + MAX_CATALOG_ACTIVE_SEARCH_WORK, + ); + expect( + provider.calls.every((call) => call.signal.aborted), + ).toBe(true); + expect(reentrantStatus).toBe("unavailable"); + expect(scheduler.pendingCount).toBe(0); + + for (const call of provider.calls) { + call.settlement.reject(new Error("late rejection")); + } + await flushMicrotasks(); + expect(events).toContain("abort"); + }); + + it("propagates epoch cleanup quarantine across scopes into search disposal", async () => { + for (const cleanupFailure of [ + "non-undefined", + "throw", + ] as const) { + const calls: ProviderCall[] = []; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: `catalog-${cleanupFailure}`, + search( + request: SqlCatalogSearchRequest, + signal: AbortSignal, + ) { + const settlement = deferred(); + calls.push({ request, settlement, signal }); + return settlement.promise; + }, + subscribe(scope: string) { + if (scope !== "scope-a") { + return () => undefined; + } + if (cleanupFailure === "throw") { + return () => { + throw new Error("cleanup failed"); + }; + } + return () => 1; + }, + }), + ); + const service = coordinator(captured); + const scopeA = owner(service, "scope-a"); + const scopeB = owner(service, "scope-b"); + const active = scopeB.request(input("scope-b-active")); + expect(calls).toHaveLength(1); + + scopeA.dispose(); + await expect(active.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(calls[0]?.signal.aborted).toBe(true); + await expect( + scopeB.request(input("after-quarantine")).result, + ).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(calls).toHaveLength(1); + calls[0]?.settlement.reject( + new Error("late quarantined rejection"), + ); + await flushMicrotasks(); + } + }); +}); + +describe("catalog search defensive lifecycle coverage", () => { + it("returns disposed when the request clock disposes its coordinator", async () => { + let disposeOnNow = false; + let service: SqlCatalogSearchWorkCoordinator | undefined; + const provider = providerHarness(); + service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout() {}, + now() { + if (disposeOnNow) service?.dispose(); + return 0; + }, + setTimeout() { + return 1; + }, + }, + }); + const session = owner(service); + disposeOnNow = true; + + await expect(session.request(input()).result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(provider.calls).toHaveLength(0); + }); + + it("contains disposal while installing a queued deadline", async () => { + let service: SqlCatalogSearchWorkCoordinator | undefined; + const provider = providerHarness(); + service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout() {}, + now: () => 0, + setTimeout(_callback, delayMs) { + if (delayMs === 100) service?.dispose(); + return 1; + }, + }, + executionDeadlineMs: 20, + queueDeadlineMs: 100, + }); + const tickets = Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK + 1 }, + (_, index) => + owner(service).request(input(`work-${index}`)), + ); + + await Promise.all( + tickets.map((ticket) => + expect(ticket.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }), + ), + ); + expect(provider.calls).toHaveLength( + MAX_CATALOG_ACTIVE_SEARCH_WORK, + ); + }); + + it("rejects throwing, non-finite, negative, and malformed scheduler configuration", () => { + const provider = providerHarness(); + const validMethods = { + clearTimeout() {}, + now: () => 0, + setTimeout() { + return 1; + }, + }; + const candidates: unknown[] = [ + { + get queueDeadlineMs() { + throw new Error("hostile options"); + }, + }, + { + deadlineScheduler: { + ...validMethods, + now() { + throw new Error("hostile clock"); + }, + }, + }, + { + deadlineScheduler: { + ...validMethods, + now: () => Number.NaN, + }, + }, + { + deadlineScheduler: { + ...validMethods, + now: () => -1, + }, + }, + { + deadlineScheduler: { + ...validMethods, + now: () => "now", + }, + }, + { + synchronousBudgetMs: 0, + }, + { + synchronousBudgetMs: 51, + }, + { + deadlineScheduler: { + clearTimeout: 1, + now: () => 0, + setTimeout() { + return 1; + }, + }, + }, + { + deadlineScheduler: { + clearTimeout() {}, + now: () => 0, + setTimeout: 1, + }, + }, + ]; + + for (const candidate of candidates) { + expect( + Reflect.apply( + createSqlCatalogSearchWorkCoordinator, + undefined, + [provider.captured, candidate], + ), + ).toEqual({ + reason: "invalid-options", + status: "unavailable", + }); + } + }); + + it("fails closed when a live monotonic clock throws, becomes non-finite, or moves backward", async () => { + const provider = providerHarness(); + const laterValues: Array<() => number> = [ + () => { + throw new Error("clock failed"); + }, + () => Number.NaN, + () => -1, + ]; + + for (const later of laterValues) { + let reads = 0; + const service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout() {}, + now() { + reads += 1; + return reads === 1 ? 0 : later(); + }, + setTimeout() { + return 1; + }, + }, + }); + await expect( + owner(service).request(input(`clock-${reads}`)).result, + ).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + service.dispose(); + } + expect(provider.calls).toHaveLength(0); + }); + + it("contains throwing timer installation and cleanup", async () => { + const provider = providerHarness(); + const installationFailure = coordinator( + provider.captured, + { + deadlineScheduler: { + clearTimeout() {}, + now: () => 0, + setTimeout() { + throw new Error("timer install failed"); + }, + }, + }, + ); + await expect( + owner(installationFailure).request( + input("install-failure"), + ).result, + ).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + + const cleanupProvider = accepted( + captureSqlRelationCatalogProvider({ + id: "cleanup", + search() { + return readyResponse(); + }, + }), + ); + const cleanupFailure = coordinator(cleanupProvider, { + deadlineScheduler: { + clearTimeout() { + throw new Error("timer cleanup failed"); + }, + now: () => 0, + setTimeout() { + return 1; + }, + }, + }); + await expect( + owner(cleanupFailure).request(input("cleanup-failure")) + .result, + ).resolves.toMatchObject({ status: "usable" }); + expect(() => cleanupFailure.dispose()).not.toThrow(); + }); + + it("bounds a scheduler that fires every deadline synchronously without advancing time", async () => { + let timerCalls = 0; + let providerCalls = 0; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + providerCalls += 1; + return readyResponse(); + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: { + clearTimeout() {}, + now: () => 0, + setTimeout(callback) { + timerCalls += 1; + callback(); + return timerCalls; + }, + }, + }); + const ticket = owner(service).request(input()); + + await expect(ticket.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + expect(timerCalls).toBe(257); + expect(providerCalls).toBe(0); + }); + + it("rearms an early asynchronous deadline and ignores its obsolete generation", async () => { + const callbacks: Array<() => void> = []; + const scheduler: SqlCatalogSearchDeadlineScheduler = { + clearTimeout() {}, + now: () => 0, + setTimeout(callback) { + callbacks.push(callback); + if (callbacks.length === 1) callback(); + return callbacks.length; + }, + }; + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + }); + const ticket = owner(service).request(input()); + + expect(callbacks.length).toBeGreaterThanOrEqual(2); + expect(() => callbacks[0]?.()).not.toThrow(); + expect(() => callbacks[1]?.()).not.toThrow(); + ticket.cancel(); + await expect(ticket.result).resolves.toEqual({ + status: "cancelled", + }); + provider.calls[0]?.settlement.reject( + new Error("late rejection"), + ); + await flushMicrotasks(); + }); + + it("expires safely when execution scheduling fires synchronously before provider invocation", async () => { + let providerCalls = 0; + let timerCalls = 0; + let nowValue = 0; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + providerCalls += 1; + return readyResponse(); + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: { + clearTimeout() {}, + now: () => nowValue, + setTimeout(callback, delayMs) { + timerCalls += 1; + if (delayMs === 250) { + nowValue = 250; + callback(); + } + return timerCalls; + }, + }, + }); + const ticket = owner(service).request(input()); + + await expect(ticket.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + expect(providerCalls).toBe(0); + expect(timerCalls).toBe(1); + }); + + it("stays inert when clearing the execution timer reentrantly disposes the coordinator", async () => { + let service: SqlCatalogSearchWorkCoordinator | undefined; + let providerCalls = 0; + let clearCalls = 0; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + providerCalls += 1; + return readyResponse(); + }, + }), + ); + service = coordinator(captured, { + deadlineScheduler: { + clearTimeout() { + clearCalls += 1; + if (clearCalls === 1) service?.dispose(); + }, + now: () => 0, + setTimeout() { + return clearCalls + 1; + }, + }, + }); + const ticket = owner(service).request(input()); + + await expect(ticket.result).resolves.toMatchObject({ + status: "usable", + }); + expect(providerCalls).toBe(1); + expect(clearCalls).toBeGreaterThan(0); + }); + + it("validates scope and dialect runtime without leaking malformed UTF-16 into memberships", () => { + const service = coordinator(providerHarness().captured); + const invalidTexts: unknown[] = [ + null, + "", + "a".repeat(513), + "nul\u0000value", + "\ud800", + "\ud800x", + "\udc00", + ]; + const target = { prepareCatalogChange: () => () => {} }; + + for (const scope of invalidTexts) { + expect( + Reflect.apply(service.prepareOwner, undefined, [ + scope, + POSTGRESQL_SQL_RELATION_DIALECT, + target, + ]), + ).toEqual({ + reason: "invalid-scope", + status: "unavailable", + }); + } + for (const dialect of invalidTexts) { + expect( + Reflect.apply(service.prepareOwner, undefined, [ + "scope", + dialect, + target, + ]), + ).toEqual({ + reason: "invalid-dialect", + status: "unavailable", + }); + } + const validAstral = service.prepareOwner( + "scope-\ud83d\ude80", + POSTGRESQL_SQL_RELATION_DIALECT, + target, + ); + expect(validAstral.status).toBe("prepared"); + if (validAstral.status === "prepared") { + validAstral.owner.dispose(); + validAstral.owner.dispose(); + } + service.dispose(); + service.dispose(); + }); + + it("maps hostile and capacity-rejected revision targets to closed owner failures", () => { + const service = coordinator(providerHarness().captured); + const throwingTarget = Object.defineProperty( + {}, + "prepareCatalogChange", + { + get() { + throw new Error("hostile target"); + }, + }, + ); + expect( + Reflect.apply(service.prepareOwner, undefined, [ + "scope", + POSTGRESQL_SQL_RELATION_DIALECT, + throwingTarget, + ]), + ).toEqual({ + reason: "invalid-target", + status: "unavailable", + }); + + const owners = Array.from({ length: 1_024 }, (_, index) => + service.prepareOwner( + `bounded-scope-${index}`, + POSTGRESQL_SQL_RELATION_DIALECT, + { prepareCatalogChange: () => () => {} }, + ), + ); + expect(owners.every((result) => result.status === "prepared")) + .toBe(true); + expect( + service.prepareOwner( + "bounded-scope-overflow", + POSTGRESQL_SQL_RELATION_DIALECT, + { prepareCatalogChange: () => () => {} }, + ), + ).toEqual({ + reason: "membership-capacity", + status: "unavailable", + }); + service.dispose(); + }); + + it("maps a throwing request getter to invalid-request after superseding current work", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const previous = session.request(input("previous")); + const hostile = Object.defineProperty( + {}, + "continuationToken", + { + enumerable: true, + get() { + throw new Error("hostile request"); + }, + }, + ); + + const rejected = Reflect.apply( + session.request, + undefined, + [hostile], + ); + await expect(rejected.result).resolves.toEqual({ + reason: "invalid-request", + status: "unavailable", + }); + await expect(previous.result).resolves.toEqual({ + status: "superseded", + }); + expect(provider.calls[0]?.signal.aborted).toBe(true); + }); + + it("lets a request getter reenter with a newer request without the older frame overwriting it", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + let nested: SqlCatalogSearchWorkTicket | undefined; + const outerInput = { + get continuationToken() { + nested ??= session.request(input("nested")); + return null; + }, + limit: 20, + prefix: component("outer"), + qualifier: [component("public")], + searchPaths: [[component("public")]], + }; + + const outer = session.request(outerInput); + await expect(outer.result).resolves.toEqual({ + status: "superseded", + }); + expect(provider.calls).toHaveLength(1); + expect(provider.calls[0]?.request.prefix.value).toBe("nested"); + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(nested?.result).resolves.toMatchObject({ + status: "usable", + }); + }); + + it("fails closed when request decoding reentrantly disposes the coordinator", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const hostile = Object.defineProperty( + input("disposing-request"), + "continuationToken", + { + enumerable: true, + get() { + service.dispose(); + return null; + }, + }, + ); + + await expect( + session.request(hostile).result, + ).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(provider.calls).toHaveLength(0); + }); + + it("revalidates active response work after the clock reenters with a replacement", async () => { + const provider = providerHarness(); + let onNow: (() => void) | undefined; + let reading = false; + const service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout() {}, + now() { + if (!reading && onNow) { + reading = true; + const callback = onNow; + onNow = undefined; + callback(); + reading = false; + } + return 0; + }, + setTimeout() { + return 1; + }, + }, + }); + const session = owner(service); + const stale = session.request(input("clock-stale")); + let replacement: SqlCatalogSearchWorkTicket | undefined; + onNow = () => { + replacement = session.request(input("clock-current")); + }; + + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(stale.result).resolves.toEqual({ + status: "superseded", + }); + expect(provider.calls).toHaveLength(2); + provider.calls[1]?.settlement.resolve(readyResponse()); + await expect(replacement?.result).resolves.toMatchObject({ + status: "usable", + }); + }); + + it("stops provider-result handling when the clock reentrantly disposes the service", async () => { + const provider = providerHarness(); + let onNow: (() => void) | undefined; + let reading = false; + const service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout() {}, + now() { + if (!reading && onNow) { + reading = true; + const callback = onNow; + onNow = undefined; + callback(); + reading = false; + } + return 0; + }, + setTimeout() { + return 1; + }, + }, + }); + const ticket = owner(service).request(input("clock-dispose")); + onNow = () => service.dispose(); + + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(ticket.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + await flushMicrotasks(); + }); + + it("revalidates queued promotion after the clock reenters with newer work", async () => { + const provider = providerHarness(); + let onNow: (() => void) | undefined; + let reading = false; + let skippedNowCallbacks = 0; + const service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout() {}, + now() { + if (!reading && onNow) { + if (skippedNowCallbacks > 0) { + skippedNowCallbacks -= 1; + return 0; + } + reading = true; + const callback = onNow; + onNow = undefined; + callback(); + reading = false; + } + return 0; + }, + setTimeout() { + return 1; + }, + }, + }); + const active = Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK }, + (_, index) => + owner(service).request(input(`active-${index}`)), + ); + const queuedOwner = owner(service); + const staleQueued = queuedOwner.request( + input("queued-stale"), + ); + let replacement: SqlCatalogSearchWorkTicket | undefined; + onNow = () => { + replacement = queuedOwner.request( + input("queued-current"), + ); + }; + skippedNowCallbacks = 1; + + provider.calls[0]?.settlement.reject( + new Error("release active slot"), + ); + await expect(staleQueued.result).resolves.toEqual({ + status: "superseded", + }); + await flushMicrotasks(); + expect( + provider.calls.some( + (call) => + call.request.prefix.value === "queued-current", + ), + ).toBe(true); + const replacementCall = provider.calls.find( + (call) => + call.request.prefix.value === "queued-current", + ); + replacementCall?.settlement.resolve(readyResponse()); + await expect(replacement?.result).resolves.toMatchObject({ + status: "usable", + }); + for (const ticket of active.slice(1)) ticket.cancel(); + service.dispose(); + }); + + it("does not let a clock-reentrant request overwrite the newer request", async () => { + const provider = providerHarness(); + let onNow: (() => void) | undefined; + let reading = false; + const service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout() {}, + now() { + if (!reading && onNow) { + reading = true; + const callback = onNow; + onNow = undefined; + callback(); + reading = false; + } + return 0; + }, + setTimeout() { + return 1; + }, + }, + }); + const session = owner(service); + let current: SqlCatalogSearchWorkTicket | undefined; + onNow = () => { + current = session.request(input("clock-newer")); + }; + + const obsolete = session.request(input("clock-older")); + await expect(obsolete.result).resolves.toEqual({ + status: "superseded", + }); + expect(provider.calls).toHaveLength(1); + expect(provider.calls[0]?.request.prefix.value).toBe( + "clock-newer", + ); + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(current?.result).resolves.toMatchObject({ + status: "usable", + }); + }); + + it("ignores stale queue and execution callbacks after successful settlement", async () => { + const callbacks: Array<() => void> = []; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + return readyResponse(); + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: { + clearTimeout() {}, + now: () => 0, + setTimeout(callback) { + callbacks.push(callback); + return callbacks.length; + }, + }, + }); + const ticket = owner(service).request(input("stale-timers")); + await expect(ticket.result).resolves.toMatchObject({ + status: "usable", + }); + + expect(callbacks).toHaveLength(1); + expect(() => { + callbacks[0]?.(); + callbacks[0]?.(); + }).not.toThrow(); + }); + + it("fails closed when the execution deadline cannot advance a large monotonic clock", async () => { + const base = 2 ** 57; + let reads = 0; + let providerCalls = 0; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + providerCalls += 1; + return readyResponse(); + }, + }), + ); + const service = coordinator(captured, { + deadlineScheduler: { + clearTimeout() {}, + now() { + reads += 1; + return reads < 4 ? base : base + 1_952; + }, + setTimeout() { + return 1; + }, + }, + executionDeadlineMs: 10, + queueDeadlineMs: 2_000, + }); + const ticket = owner(service).request(input("large-clock")); + + await expect(ticket.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + expect(providerCalls).toBe(0); + }); + + it("fails closed when a promoted queued search cannot allocate an execution deadline", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 10, + queueDeadlineMs: 2_000, + }); + Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK }, + (_, index) => owner(service).request(input(`active-${index}`)), + ); + scheduler.nowValue = 2 ** 57 - 1_024; + const queued = owner(service).request(input("large-promoted")); + scheduler.nowValue = 2 ** 57; + provider.calls[0]?.settlement.reject( + new Error("release active slot"), + ); + + await expect(queued.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + expect( + provider.calls.some( + (call) => + call.request.prefix.value === "large-promoted", + ), + ).toBe(false); + service.dispose(); + }); + + it("clears a queued deadline when execution expiry fires synchronously during promotion", async () => { + let nowValue = 0; + let fireExecution = false; + const callbacks = new Map void>(); + let nextHandle = 0; + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout(handle) { + if (typeof handle === "number") { + callbacks.delete(handle); + } + }, + now: () => nowValue, + setTimeout(callback, delayMs) { + nextHandle += 1; + callbacks.set(nextHandle, callback); + if (fireExecution && delayMs === 20) { + nowValue += 20; + callback(); + } + return nextHandle; + }, + }, + executionDeadlineMs: 20, + queueDeadlineMs: 100, + }); + Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK }, + (_, index) => owner(service).request(input(`active-${index}`)), + ); + const queued = owner(service).request( + input("synchronous-promotion-expiry"), + ); + fireExecution = true; + provider.calls[0]?.settlement.reject( + new Error("release active slot"), + ); + + await expect(queued.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + expect( + provider.calls.some( + (call) => + call.request.prefix.value === + "synchronous-promotion-expiry", + ), + ).toBe(false); + service.dispose(); + }); + + it("contains disposal reentrancy while a promoted search clears its queue deadline", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout(handle) { + scheduler.clearTimeout(handle); + if (skipClears > 0) { + skipClears -= 1; + } else { + service.dispose(); + } + }, + now: scheduler.now, + setTimeout: scheduler.setTimeout, + }, + executionDeadlineMs: 20, + queueDeadlineMs: 100, + }); + let skipClears = Number.MAX_SAFE_INTEGER; + Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK }, + (_, index) => owner(service).request(input(`active-${index}`)), + ); + const queued = owner(service).request( + input("dispose-during-promotion"), + ); + skipClears = 1; + provider.calls[0]?.settlement.reject( + new Error("release active slot"), + ); + + await expect(queued.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect( + provider.calls.some( + (call) => + call.request.prefix.value === + "dispose-during-promotion", + ), + ).toBe(false); + }); + + it("cancels reentrantly during response decoding and ignores the now-obsolete result", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const ticket = session.request(input("decode-cancel")); + const response = new Proxy(readyResponse(), { + ownKeys(target) { + ticket.cancel(); + return Reflect.ownKeys(target); + }, + }); + + provider.calls[0]?.settlement.resolve(response); + await expect(ticket.result).resolves.toEqual({ + status: "cancelled", + }); + await flushMicrotasks(); + expect(provider.calls[0]?.signal.aborted).toBe(false); + }); + + it("keeps a response that reenters with a replacement from publishing stale epoch evidence", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const stale = session.request(input("stale")); + let replacement: SqlCatalogSearchWorkTicket | undefined; + const response = new Proxy(readyResponse(), { + ownKeys(target) { + replacement ??= session.request(input("replacement")); + return Reflect.ownKeys(target); + }, + }); + + provider.calls[0]?.settlement.resolve(response); + await expect(stale.result).resolves.toEqual({ + status: "superseded", + }); + expect(replacement).toBeDefined(); + expect(provider.calls).toHaveLength(2); + provider.calls[1]?.settlement.resolve(readyResponse()); + await expect(replacement?.result).resolves.toMatchObject({ + observation: "baseline", + status: "usable", + }); + }); + + it("does not conflate paths with different component counts", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const tickets = [ + owner(service).request( + input("path-length", { + qualifier: [component("public")], + }), + ), + owner(service).request( + input("path-length", { + qualifier: [ + component("catalog"), + component("public"), + ], + }), + ), + owner(service).request( + input("path-length", { + searchPaths: [[component("public")]], + }), + ), + owner(service).request( + input("path-length", { + searchPaths: [ + [ + component("catalog"), + component("public"), + ], + ], + }), + ), + ]; + + expect(provider.calls).toHaveLength(3); + service.dispose(); + await Promise.all( + tickets.map((ticket) => + expect(ticket.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }), + ), + ); + }); + + it("applies the absolute execution deadline when response decoding advances the clock", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 20, + queueDeadlineMs: 10, + synchronousBudgetMs: 5, + }); + const ticket = owner(service).request(input("decode-timeout")); + const response = new Proxy(readyResponse(), { + ownKeys(target) { + scheduler.nowValue = 20; + return Reflect.ownKeys(target); + }, + }); + + provider.calls[0]?.settlement.resolve(response); + await expect(ticket.result).resolves.toEqual({ + reason: "execution-timeout", + status: "unavailable", + }); + expect(provider.calls[0]?.signal.aborted).toBe(true); + }); + + it("does not publish a decoded response that disposes the coordinator through a getter", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const ticket = owner(service).request(input("decode-dispose")); + const response = new Proxy(readyResponse(), { + ownKeys(target) { + service.dispose(); + return Reflect.ownKeys(target); + }, + }); + + provider.calls[0]?.settlement.resolve(response); + await expect(ticket.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + await flushMicrotasks(); + }); + + it("drains provider results returned after synchronous service disposal", async () => { + for (const kind of ["pending", "rejected"] as const) { + let service: + | SqlCatalogSearchWorkCoordinator + | undefined; + const pending = deferred(); + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: `dispose-${kind}`, + search() { + service?.dispose(); + if (kind === "rejected") { + return Promise.reject( + new Error("rejected after disposal"), + ); + } + return pending.promise; + }, + }), + ); + service = coordinator(captured); + const ticket = owner(service).request(input(kind)); + + await expect(ticket.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + if (kind === "pending") { + pending.reject(new Error("late pending rejection")); + } + await flushMicrotasks(); + } + }); + + it("accepts an epoch transition with no search work and keeps disposal idempotent", () => { + let invalidation: ((event: unknown) => void) | undefined; + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + return readyResponse(); + }, + subscribe( + _scope: string, + listener: (event: unknown) => void, + ) { + invalidation = listener; + return () => undefined; + }, + }), + ); + const service = coordinator(captured); + const session = owner(service, "idle-scope"); + + expect(() => invalidation?.({ epoch: epoch() })).not.toThrow(); + const immediate = service.prepareOwner( + "inactive", + POSTGRESQL_SQL_RELATION_DIALECT, + { prepareCatalogChange: () => () => {} }, + ); + expect(immediate.status).toBe("prepared"); + if (immediate.status === "prepared") { + const ticket = immediate.owner.request(input()); + expect(() => { + ticket.cancel(); + ticket.cancel(); + }).not.toThrow(); + } + session.dispose(); + session.dispose(); + service.dispose(); + service.dispose(); + expect( + service.prepareOwner( + "late", + POSTGRESQL_SQL_RELATION_DIALECT, + { prepareCatalogChange: () => () => {} }, + ), + ).toEqual({ + reason: "disposed", + status: "unavailable", + }); + }); +}); + +describe("catalog search epoch dependency failures", () => { + it("maps every closed epoch decision and rotates retired captures", async () => { + type Decision = + | "disposed" + | "malformed" + | "overloaded" + | "retired-exhausted" + | "retired-then-usable" + | "superseded"; + let decision: Decision = "disposed"; + let captureFailure = false; + let captureHook: (() => void) | undefined; + let capturedExpectedEpoch: ReturnType | null = + null; + let disposeCandidate: (() => void) | undefined; + let factoryFailure = false; + let membershipFailure = false; + let searchCalls = 0; + let submissions = 0; + vi.resetModules(); + vi.doMock( + "../relation-catalog-epoch-coordinator.js", + async (importOriginal) => { + const actual = + await importOriginal< + typeof import("../relation-catalog-epoch-coordinator.js") + >(); + return { + ...actual, + createSqlCatalogEpochCoordinator() { + if (factoryFailure) { + return { + reason: "invalid-provider", + status: "unavailable", + }; + } + return { + coordinator: { + dispose() {}, + prepareScopeMembership() { + if (membershipFailure) { + return { + reason: "disposed", + status: "unavailable", + }; + } + return { + membership: { + activate() { + return { status: "active" }; + }, + captureEpoch() { + const hook = captureHook; + captureHook = undefined; + hook?.(); + if (captureFailure) { + return { + reason: "disposed", + status: "unavailable", + }; + } + return { + capture: { + expectedEpoch: capturedExpectedEpoch, + }, + status: "captured", + }; + }, + dispose() {}, + }, + status: "prepared", + }; + }, + providerId: "catalog", + submitResponseEpoch( + _capture: unknown, + responseEpoch: ReturnType, + onDecision: (value: unknown) => void, + ) { + submissions += 1; + if ( + (decision === "retired-then-usable" || + decision === "retired-exhausted") && + submissions === 1 + ) { + disposeCandidate?.(); + return { + decision: { + reason: "retired", + status: "discarded", + }, + status: "settled", + }; + } + if (decision === "superseded") { + onDecision({ + epoch: responseEpoch, + status: "superseded", + }); + } else if ( + decision === "retired-then-usable" + ) { + onDecision({ + epoch: responseEpoch, + observation: "baseline", + status: "usable", + }); + } else { + onDecision({ + reason: decision, + status: "discarded", + }); + } + return { status: "submitted" }; + }, + }, + status: "created", + }; + }, + }; + }, + ); + const isolated = await import( + "../relation-catalog-search-work.js" + ); + const isolatedBoundary = await import( + "../relation-catalog-boundary.js" + ); + const isolatedDialect = await import( + "../relation-dialect.js" + ); + const captured = accepted( + isolatedBoundary.captureSqlRelationCatalogProvider({ + id: "catalog", + search() { + searchCalls += 1; + return readyResponse(); + }, + }), + ); + + const expected = new Map< + Exclude, + SqlCatalogSearchWorkOutcome + >([ + [ + "disposed", + { reason: "disposed", status: "unavailable" }, + ], + [ + "malformed", + { + reason: "malformed-response", + status: "unavailable", + }, + ], + [ + "overloaded", + { reason: "overloaded", status: "unavailable" }, + ], + ["superseded", { status: "superseded" }], + ]); + for (const [nextDecision, outcome] of expected) { + decision = nextDecision; + submissions = 0; + const created = + isolated.createSqlCatalogSearchWorkCoordinator(captured); + expect(created.status).toBe("created"); + if (created.status !== "created") continue; + const session = owner( + created.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ); + await expect( + session.request(input(nextDecision)).result, + ).resolves.toEqual(outcome); + created.coordinator.dispose(); + } + + decision = "retired-then-usable"; + submissions = 0; + const created = + isolated.createSqlCatalogSearchWorkCoordinator(captured); + expect(created.status).toBe("created"); + if (created.status === "created") { + const first = owner( + created.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ); + const second = owner( + created.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ); + const firstTicket = first.request(input("rotation")); + const secondTicket = second.request(input("rotation")); + await expect(firstTicket.result).resolves.toMatchObject({ + status: "usable", + }); + await expect(secondTicket.result).resolves.toMatchObject({ + status: "usable", + }); + expect(submissions).toBe(2); + created.coordinator.dispose(); + } + + decision = "retired-exhausted"; + submissions = 0; + const exhausted = + isolated.createSqlCatalogSearchWorkCoordinator(captured); + expect(exhausted.status).toBe("created"); + if (exhausted.status === "created") { + const first = owner( + exhausted.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ); + const second = owner( + exhausted.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ); + const firstTicket = first.request(input("exhausted")); + const secondTicket = second.request(input("exhausted")); + disposeCandidate = () => second.dispose(); + await expect(secondTicket.result).resolves.toEqual({ + status: "cancelled", + }); + await expect(firstTicket.result).resolves.toEqual({ + status: "superseded", + }); + expect(submissions).toBe(1); + exhausted.coordinator.dispose(); + } + + decision = "disposed"; + capturedExpectedEpoch = null; + searchCalls = 0; + const epochMismatch = + isolated.createSqlCatalogSearchWorkCoordinator(captured); + expect(epochMismatch.status).toBe("created"); + if (epochMismatch.status === "created") { + const first = owner( + epochMismatch.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ).request(input("epoch-key")); + capturedExpectedEpoch = epoch(7); + const second = owner( + epochMismatch.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ).request(input("epoch-key")); + expect(searchCalls).toBe(2); + epochMismatch.coordinator.dispose(); + await expect(first.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + await expect(second.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + } + capturedExpectedEpoch = null; + + decision = "retired-then-usable"; + captureFailure = true; + const captureRejected = + isolated.createSqlCatalogSearchWorkCoordinator(captured); + expect(captureRejected.status).toBe("created"); + if (captureRejected.status === "created") { + const session = owner( + captureRejected.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ); + await expect( + session.request(input("capture-disposed")).result, + ).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + captureRejected.coordinator.dispose(); + } + captureFailure = false; + + const captureReentrant = + isolated.createSqlCatalogSearchWorkCoordinator(captured); + expect(captureReentrant.status).toBe("created"); + if (captureReentrant.status === "created") { + const session = owner( + captureReentrant.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ); + let current: SqlCatalogSearchWorkTicket | undefined; + captureHook = () => { + current = session.request(input("capture-current")); + }; + const obsolete = session.request(input("capture-obsolete")); + await expect(obsolete.result).resolves.toEqual({ + status: "superseded", + }); + await expect(current?.result).resolves.toMatchObject({ + status: "usable", + }); + captureReentrant.coordinator.dispose(); + } + + const captureDisposal = + isolated.createSqlCatalogSearchWorkCoordinator(captured); + expect(captureDisposal.status).toBe("created"); + if (captureDisposal.status === "created") { + const session = owner( + captureDisposal.coordinator, + "connection:primary", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + ); + captureHook = () => { + captureDisposal.coordinator.dispose(); + }; + await expect( + session.request(input("capture-disposal")).result, + ).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + } + + membershipFailure = true; + const membershipRejected = + isolated.createSqlCatalogSearchWorkCoordinator(captured); + expect(membershipRejected.status).toBe("created"); + if (membershipRejected.status === "created") { + expect( + membershipRejected.coordinator.prepareOwner( + "scope", + isolatedDialect.POSTGRESQL_SQL_RELATION_DIALECT, + { prepareCatalogChange: () => () => {} }, + ), + ).toEqual({ + reason: "disposed", + status: "unavailable", + }); + membershipRejected.coordinator.dispose(); + } + membershipFailure = false; + + factoryFailure = true; + expect( + isolated.createSqlCatalogSearchWorkCoordinator(captured), + ).toEqual({ + reason: "invalid-provider", + status: "unavailable", + }); + vi.doUnmock("../relation-catalog-epoch-coordinator.js"); + vi.resetModules(); + }); +}); diff --git a/src/vnext/__tests__/relation-dialect.test.ts b/src/vnext/__tests__/relation-dialect.test.ts index 38a4c9f..667b416 100644 --- a/src/vnext/__tests__/relation-dialect.test.ts +++ b/src/vnext/__tests__/relation-dialect.test.ts @@ -110,8 +110,9 @@ describe("built-in relation dialect runtime", () => { }); it("owns stable, deeply frozen, coherent views", () => { - for (const runtime of Object.values(RUNTIMES)) { + for (const [id, runtime] of Object.entries(RUNTIMES)) { expectDeepFrozenRuntime(runtime); + expect(runtime.id).toBe(id); expect(runtime.cteLayout.lexicalProfile).toBe( runtime.querySite.lexicalProfile, ); diff --git a/src/vnext/relation-catalog-boundary.ts b/src/vnext/relation-catalog-boundary.ts index 274f5a8..8149513 100644 --- a/src/vnext/relation-catalog-boundary.ts +++ b/src/vnext/relation-catalog-boundary.ts @@ -341,6 +341,18 @@ function isWellFormed(value: string): boolean { return true; } +export function isValidSqlCatalogScope( + candidate: unknown, +): candidate is string { + return ( + typeof candidate === "string" && + candidate.length > 0 && + candidate.length <= MAX_CATALOG_SCOPE_LENGTH && + !candidate.includes("\0") && + isWellFormed(candidate) + ); +} + function readRecord( state: DecodeState, value: unknown, diff --git a/src/vnext/relation-catalog-epoch-coordinator.ts b/src/vnext/relation-catalog-epoch-coordinator.ts index 3a3a1fe..a82018e 100644 --- a/src/vnext/relation-catalog-epoch-coordinator.ts +++ b/src/vnext/relation-catalog-epoch-coordinator.ts @@ -1,7 +1,7 @@ import { - MAX_CATALOG_SCOPE_LENGTH, compareSqlCatalogEpoch, decodeSqlCatalogInvalidation, + isValidSqlCatalogScope, resolveSqlRelationCatalogProvider, } from "./relation-catalog-boundary.js"; import type { SqlCapturedRelationCatalogProviderContext } from "./relation-catalog-boundary.js"; @@ -143,6 +143,7 @@ export type SqlCatalogEpochCoordinatorResult = | { readonly status: "unavailable"; readonly reason: + | "invalid-disposal-target" | "invalid-provider" | "invalid-transition-target"; }; @@ -157,6 +158,7 @@ interface CoordinatorState { readonly commands: EpochCommand[]; readonly deferredCleanup: Set; readonly memberships: Set; + onDispose: ((this: void) => undefined) | null; prepareEpochTransition: Function | null; readonly providerId: string; readonly scopes: Map; @@ -238,6 +240,8 @@ const SUBMITTED_RESULT: SqlCatalogResponseEpochSubmissionResult = Object.freeze({ status: "submitted" }); const NO_PREPARE_CATALOG_CHANGE = (): null => null; const IGNORE_DETACHED_REJECTION = (): void => {}; +const INTRINSIC_PROMISE = Promise; +const INTRINSIC_PROMISE_RESOLVE = Promise.resolve; const INTRINSIC_PROMISE_THEN = Promise.prototype.then; const FAILED_EPOCH_TRANSITION: unique symbol = Symbol( "FailedSqlCatalogEpochTransition", @@ -287,29 +291,6 @@ function settleDecision( } } -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[] { @@ -390,21 +371,18 @@ function drainDetachedSettlement(result: unknown): void { return; } try { - Reflect.apply(INTRINSIC_PROMISE_THEN, result, [ + const settlement = Reflect.apply( + INTRINSIC_PROMISE_RESOLVE, + INTRINSIC_PROMISE, + [result], + ); + Reflect.apply(INTRINSIC_PROMISE_THEN, settlement, [ undefined, IGNORE_DETACHED_REJECTION, ]); - return; } catch { - // Non-native thenables are assimilated through a fresh wrapper. + // The detached value is hostile and cannot be observed safely. } - const settlement = new Promise((resolve) => { - resolve(result); - }); - Reflect.apply(INTRINSIC_PROMISE_THEN, settlement, [ - undefined, - IGNORE_DETACHED_REJECTION, - ]); } function cleanupSubscription( @@ -1148,7 +1126,7 @@ function prepareMembership( status: "unavailable", }); } - if (!isValidScope(scope)) { + if (!isValidSqlCatalogScope(scope)) { return Object.freeze({ reason: "invalid-scope", status: "unavailable", @@ -1248,6 +1226,8 @@ function submitResponse( function disposeCoordinator(state: CoordinatorState): void { if (state.disposed) return; state.disposed = true; + const onDispose = state.onDispose; + state.onDispose = null; state.prepareEpochTransition = null; state.subscribe = null; const subscriptions: SubscriptionState[] = []; @@ -1271,6 +1251,16 @@ function disposeCoordinator(state: CoordinatorState): void { for (const subscription of subscriptions) { retireCallbackCell(subscription.cell); } + if (onDispose) { + try { + const result = Reflect.apply(onDispose, undefined, []); + if (result !== undefined) { + drainDetachedSettlement(result); + } + } catch { + // Disposal remains authoritative if its package owner fails. + } + } for (const subscription of subscriptions) { scheduleSubscriptionCleanup(state, subscription); } @@ -1305,6 +1295,7 @@ function createCoordinatorHandle( export function createSqlCatalogEpochCoordinator( capturedProvider: unknown, prepareEpochTransition?: SqlCatalogEpochTransitionTarget, + onDispose?: (this: void) => undefined, ): SqlCatalogEpochCoordinatorResult { const provider = resolveSqlRelationCatalogProvider( capturedProvider, @@ -1324,6 +1315,15 @@ export function createSqlCatalogEpochCoordinator( status: "unavailable", }); } + if ( + onDispose !== undefined && + typeof onDispose !== "function" + ) { + return Object.freeze({ + reason: "invalid-disposal-target", + status: "unavailable", + }); + } const providerId = provider.id; const subscribe = provider.subscribe; const state: CoordinatorState = { @@ -1336,6 +1336,7 @@ export function createSqlCatalogEpochCoordinator( disposed: false, draining: false, memberships: new Set(), + onDispose: onDispose ?? null, prepareEpochTransition: prepareEpochTransition ?? null, providerId, diff --git a/src/vnext/relation-catalog-search-work.ts b/src/vnext/relation-catalog-search-work.ts new file mode 100644 index 0000000..6fc35ee --- /dev/null +++ b/src/vnext/relation-catalog-search-work.ts @@ -0,0 +1,1866 @@ +import { + createSqlCatalogSearchRequest, + decodeSqlCatalogSearchResponse, + isValidSqlCatalogScope, + resolveSqlRelationCatalogProvider, +} from "./relation-catalog-boundary.js"; +import type { + CapturedSqlRelationCatalogProvider, + SqlValidatedCatalogSearchResponse, +} from "./relation-catalog-boundary.js"; +import { + createSqlCatalogEpochCoordinator, +} from "./relation-catalog-epoch-coordinator.js"; +import type { + SqlCatalogEpochCapture, + SqlCatalogEpochCoordinator, + SqlCatalogResponseEpochDecision, + SqlCatalogRevisionTarget, + SqlCatalogScopeMembership, +} from "./relation-catalog-epoch-coordinator.js"; +import type { + SqlCatalogSearchRequest, +} from "./relation-completion-types.js"; +import type { SqlRelationDialectRuntime } from "./relation-dialect.js"; +import { + isSqlRelationDialectRuntime, +} from "./relation-runtime-auth.js"; +import type { + SqlIdentifierComponent, + SqlIdentifierPath, +} from "./types.js"; + +export const MAX_CATALOG_ACTIVE_SEARCH_WORK = 8; +export const MAX_CATALOG_QUEUED_SEARCH_WORK = 64; +export const DEFAULT_CATALOG_QUEUE_DEADLINE_MS = 100; +export const DEFAULT_CATALOG_EXECUTION_DEADLINE_MS = 250; +export const DEFAULT_CATALOG_SYNCHRONOUS_BUDGET_MS = 8; +export const MIN_CATALOG_QUEUE_DEADLINE_MS = 10; +export const MAX_CATALOG_QUEUE_DEADLINE_MS = 2_000; +export const MIN_CATALOG_EXECUTION_DEADLINE_MS = 10; +export const MAX_CATALOG_EXECUTION_DEADLINE_MS = 5_000; +export const MIN_CATALOG_SYNCHRONOUS_BUDGET_MS = 1; +export const MAX_CATALOG_SYNCHRONOUS_BUDGET_MS = 50; + +export interface SqlCatalogSearchDeadlineScheduler { + readonly clearTimeout: ( + this: void, + handle: unknown, + ) => void; + readonly now: (this: void) => number; + readonly setTimeout: ( + this: void, + callback: (this: void) => void, + delayMs: number, + ) => unknown; +} + +export interface SqlCatalogSearchWorkOptions { + readonly deadlineScheduler?: SqlCatalogSearchDeadlineScheduler; + readonly executionDeadlineMs?: number; + readonly queueDeadlineMs?: number; + readonly synchronousBudgetMs?: number; +} + +export interface SqlCatalogSearchWorkInput { + readonly continuationToken: string | null; + readonly limit: number; + readonly prefix: SqlIdentifierComponent; + readonly qualifier: SqlIdentifierPath; + readonly searchPaths: readonly SqlIdentifierPath[]; +} + +export type SqlCatalogSearchWorkUnavailableReason = + | "disposed" + | "execution-timeout" + | "inactive" + | "invalid-request" + | "malformed-response" + | "overloaded" + | "provider-failed" + | "queue-timeout"; + +export type SqlCatalogSearchWorkOutcome = + | { + readonly status: "usable"; + readonly observation: "baseline" | "equal"; + readonly response: SqlValidatedCatalogSearchResponse; + } + | { + readonly status: "superseded"; + } + | { + readonly status: "cancelled"; + } + | { + readonly status: "unavailable"; + readonly reason: SqlCatalogSearchWorkUnavailableReason; + }; + +export interface SqlCatalogSearchWorkTicket { + readonly cancel: (this: void) => void; + readonly result: Promise; +} + +export interface SqlCatalogSearchWorkOwner { + readonly activate: SqlCatalogScopeMembership["activate"]; + readonly request: ( + this: void, + input: SqlCatalogSearchWorkInput, + ) => SqlCatalogSearchWorkTicket; + readonly dispose: (this: void) => void; +} + +export type SqlCatalogSearchWorkOwnerResult = + | { + readonly status: "prepared"; + readonly owner: SqlCatalogSearchWorkOwner; + } + | { + readonly status: "unavailable"; + readonly reason: + | "disposed" + | "invalid-dialect" + | "invalid-scope" + | "invalid-target" + | "membership-capacity"; + }; + +export interface SqlCatalogSearchWorkCoordinator { + readonly prepareOwner: ( + this: void, + scope: unknown, + dialect: SqlRelationDialectRuntime, + target: SqlCatalogRevisionTarget, + ) => SqlCatalogSearchWorkOwnerResult; + readonly providerId: string; + readonly dispose: (this: void) => void; +} + +export type SqlCatalogSearchWorkCoordinatorResult = + | { + readonly status: "created"; + readonly coordinator: SqlCatalogSearchWorkCoordinator; + } + | { + readonly status: "unavailable"; + readonly reason: "invalid-options" | "invalid-provider"; + }; + +interface NormalizedOptions { + readonly deadlineScheduler: SqlCatalogSearchDeadlineScheduler; + readonly executionDeadlineMs: number; + readonly queueDeadlineMs: number; + readonly synchronousBudgetMs: number; +} + +interface OwnerState { + current: ConsumerState | null; + readonly dialect: SqlRelationDialectRuntime; + disposed: boolean; + readonly membership: SqlCatalogScopeMembership; + owner: CoordinatorState | null; + requestToken: object | null; + readonly scope: string; +} + +interface ConsumerState { + readonly capture: SqlCatalogEpochCapture; + cancelled: boolean; + owner: OwnerState | null; + resolve: (outcome: SqlCatalogSearchWorkOutcome) => void; + settled: boolean; + work: WorkState | null; +} + +interface WorkState { + abortController: AbortController | null; + abortIssued: boolean; + dialect: SqlRelationDialectRuntime | null; + decisionCell: ResponseDecisionCell | null; + executionDeadline: number | null; + executionTimer: DeadlineCell | null; + joinable: boolean; + readonly owners: Set; + phase: + | "active" + | "deciding" + | "queued" + | "retired" + | "terminal"; + providerCell: ProviderResultCell | null; + queueDeadline: number; + queueTimer: DeadlineCell | null; + request: SqlCatalogSearchRequest | null; + scope: string; +} + +interface DeadlineCell { + active: boolean; + generation: number; + handle: unknown; + tick: ((generation: number) => void) | null; +} + +interface ProviderResultCell { + active: boolean; + deliver: + | (( + fulfilled: boolean, + value: unknown, + ) => void) + | null; +} + +interface CoordinatorState { + activeCount: number; + disposed: boolean; + readonly epochs: SqlCatalogEpochCoordinator; + readonly joinable: Set; + lastNow: number; + readonly options: NormalizedOptions; + readonly owners: Set; + pumpRequested: boolean; + pumping: boolean; + readonly queue: WorkState[]; + search: + | (( + this: void, + request: SqlCatalogSearchRequest, + signal: AbortSignal, + ) => unknown) + | null; + readonly works: Set; +} + +interface ResponseDecisionCell { + active: boolean; + advancing: boolean; + candidates: readonly ConsumerState[]; + index: number; + pending: SqlCatalogResponseEpochDecision | null; + response: SqlValidatedCatalogSearchResponse | null; + state: CoordinatorState | null; + work: WorkState | null; +} + +interface DetachedSettlement { + readonly outcome: SqlCatalogSearchWorkOutcome; + readonly resolve: ( + outcome: SqlCatalogSearchWorkOutcome, + ) => void; +} + +interface Effects { + readonly aborts: AbortController[]; + pump: boolean; + readonly settlements: DetachedSettlement[]; + readonly timers: DeadlineCell[]; +} + +const HOST_CLEAR_TIMEOUT = globalThis.clearTimeout; +const HOST_SET_TIMEOUT = globalThis.setTimeout; +const HOST_PERFORMANCE = globalThis.performance; +const HOST_NOW = HOST_PERFORMANCE.now; +const INTRINSIC_PROMISE = Promise; +const INTRINSIC_PROMISE_RESOLVE = Promise.resolve; +const INTRINSIC_PROMISE_THEN = Promise.prototype.then; +const IGNORE_DETACHED_REJECTION = (): void => {}; +const MAX_SYNCHRONOUS_DEADLINE_REARMS = 256; + +const DEFAULT_DEADLINE_SCHEDULER: SqlCatalogSearchDeadlineScheduler = + Object.freeze({ + clearTimeout(handle: unknown): void { + Reflect.apply(HOST_CLEAR_TIMEOUT, globalThis, [ + handle, + ]); + }, + now(): number { + return Reflect.apply(HOST_NOW, HOST_PERFORMANCE, []); + }, + setTimeout( + callback: (this: void) => void, + delayMs: number, + ): unknown { + return Reflect.apply(HOST_SET_TIMEOUT, globalThis, [ + callback, + delayMs, + ]); + }, + }); + +const CANCELLED_OUTCOME: SqlCatalogSearchWorkOutcome = + Object.freeze({ status: "cancelled" }); +const SUPERSEDED_OUTCOME: SqlCatalogSearchWorkOutcome = + Object.freeze({ status: "superseded" }); + +function unavailableOutcome( + reason: SqlCatalogSearchWorkUnavailableReason, +): SqlCatalogSearchWorkOutcome { + return Object.freeze({ reason, status: "unavailable" }); +} + +function effects(): Effects { + return { + aborts: [], + pump: false, + settlements: [], + timers: [], + }; +} + +function isDuration( + value: unknown, + minimum: number, + maximum: number, +): value is number { + return ( + typeof value === "number" && + Number.isFinite(value) && + value >= minimum && + value <= maximum + ); +} + +function normalizeOptions( + candidate: SqlCatalogSearchWorkOptions | undefined, +): { + readonly initialNow: number; + readonly options: NormalizedOptions; +} | null { + try { + const queueDeadlineMs = + candidate?.queueDeadlineMs ?? + DEFAULT_CATALOG_QUEUE_DEADLINE_MS; + const executionDeadlineMs = + candidate?.executionDeadlineMs ?? + DEFAULT_CATALOG_EXECUTION_DEADLINE_MS; + const synchronousBudgetMs = + candidate?.synchronousBudgetMs ?? + DEFAULT_CATALOG_SYNCHRONOUS_BUDGET_MS; + const deadlineScheduler = + candidate?.deadlineScheduler ?? + DEFAULT_DEADLINE_SCHEDULER; + const clearTimeoutMethod = deadlineScheduler.clearTimeout; + const nowMethod = deadlineScheduler.now; + const setTimeoutMethod = deadlineScheduler.setTimeout; + if ( + !isDuration( + queueDeadlineMs, + MIN_CATALOG_QUEUE_DEADLINE_MS, + MAX_CATALOG_QUEUE_DEADLINE_MS, + ) || + !isDuration( + executionDeadlineMs, + MIN_CATALOG_EXECUTION_DEADLINE_MS, + MAX_CATALOG_EXECUTION_DEADLINE_MS, + ) || + !isDuration( + synchronousBudgetMs, + MIN_CATALOG_SYNCHRONOUS_BUDGET_MS, + MAX_CATALOG_SYNCHRONOUS_BUDGET_MS, + ) || + typeof nowMethod !== "function" || + typeof setTimeoutMethod !== "function" || + typeof clearTimeoutMethod !== "function" + ) { + return null; + } + const initialNow = Reflect.apply( + nowMethod, + undefined, + [], + ); + if ( + typeof initialNow !== "number" || + !Number.isFinite(initialNow) || + initialNow < 0 + ) { + return null; + } + const capturedScheduler: SqlCatalogSearchDeadlineScheduler = + Object.freeze({ + clearTimeout(handle: unknown): void { + Reflect.apply(clearTimeoutMethod, undefined, [handle]); + }, + now(): number { + return Reflect.apply(nowMethod, undefined, []); + }, + setTimeout( + callback: (this: void) => void, + delayMs: number, + ): unknown { + return Reflect.apply(setTimeoutMethod, undefined, [ + callback, + delayMs, + ]); + }, + }); + return { + initialNow, + options: Object.freeze({ + deadlineScheduler: capturedScheduler, + executionDeadlineMs, + queueDeadlineMs, + synchronousBudgetMs, + }), + }; + } catch { + return null; + } +} + +function readNow(state: CoordinatorState): number | null { + let value: unknown; + try { + value = Reflect.apply( + state.options.deadlineScheduler.now, + undefined, + [], + ); + } catch { + return null; + } + if ( + typeof value !== "number" || + !Number.isFinite(value) || + value < state.lastNow + ) { + return null; + } + state.lastNow = value; + return value; +} + +function deadlineFrom( + now: number, + duration: number, +): number | null { + const deadline = now + duration; + return Number.isFinite(deadline) && deadline > now + ? deadline + : null; +} + +function clearDeadline( + state: CoordinatorState, + cell: DeadlineCell, +): void { + if (!cell.active) return; + cell.active = false; + cell.generation += 1; + cell.tick = null; + try { + Reflect.apply( + state.options.deadlineScheduler.clearTimeout, + undefined, + [cell.handle], + ); + } catch { + // Deadline cleanup cannot reopen retired work. + } +} + +function scheduleDeadline( + state: CoordinatorState, + deadline: number, + expire: () => void, +): DeadlineCell { + const cell: DeadlineCell = { + active: true, + generation: 0, + handle: undefined, + tick: null, + }; + let arming = false; + let synchronousRearms = 0; + const expireNow = (): void => { + if (!cell.active) return; + cell.active = false; + cell.generation += 1; + cell.tick = null; + expire(); + }; + const arm = (): void => { + arming = true; + try { + for (;;) { + const now = readNow(state); + if (now === null || now >= deadline) { + expireNow(); + return; + } + const generation = cell.generation + 1; + cell.generation = generation; + const token = { fired: false }; + const callback = (): void => { + token.fired = true; + cell.tick?.(generation); + }; + let handle: unknown; + try { + handle = Reflect.apply( + state.options.deadlineScheduler.setTimeout, + undefined, + [callback, deadline - now], + ); + } catch { + expireNow(); + return; + } + if (!token.fired) { + cell.handle = handle; + return; + } + try { + Reflect.apply( + state.options.deadlineScheduler.clearTimeout, + undefined, + [handle], + ); + } catch { + // The synchronously fired generation is already obsolete. + } + if (!cell.active) return; + synchronousRearms += 1; + if ( + synchronousRearms > + MAX_SYNCHRONOUS_DEADLINE_REARMS + ) { + expireNow(); + return; + } + } + } finally { + arming = false; + } + }; + cell.tick = (generation): void => { + if ( + !cell.active || + generation !== cell.generation + ) { + return; + } + const now = readNow(state); + if (now === null || now >= deadline) { + expireNow(); + return; + } + if (!arming) arm(); + }; + arm(); + return cell; +} + +function settleDetached( + list: DetachedSettlement[], + consumer: ConsumerState, + outcome: SqlCatalogSearchWorkOutcome, +): void { + consumer.settled = true; + const resolve = consumer.resolve; + consumer.resolve = IGNORE_DETACHED_REJECTION; + consumer.work = null; + const owner = consumer.owner; + consumer.owner = null; + if (owner?.current === consumer) owner.current = null; + list.push({ outcome, resolve }); +} + +function runEffects( + state: CoordinatorState, + pending: Effects, +): void { + for (const timer of pending.timers) { + clearDeadline(state, timer); + } + for (const settlement of pending.settlements) { + settlement.resolve(settlement.outcome); + } + for (const controller of pending.aborts) { + try { + controller.abort(); + } catch { + // Work was made inert before provider abort. + } + } + if (pending.pump) pump(state); +} + +function sameComponent( + left: SqlIdentifierComponent, + right: SqlIdentifierComponent, +): boolean { + return ( + left.value === right.value && + left.quoted === right.quoted + ); +} + +function samePath( + left: SqlIdentifierPath, + right: SqlIdentifierPath, +): boolean { + if (left.length !== right.length) return false; + for (let index = 0; index < left.length; index += 1) { + const leftPart = left[index]; + const rightPart = right[index]; + if ( + leftPart === undefined || + rightPart === undefined || + !sameComponent(leftPart, rightPart) + ) { + return false; + } + } + return true; +} + +function sameRequest( + work: WorkState, + request: SqlCatalogSearchRequest, + dialect: SqlRelationDialectRuntime, +): boolean { + const existing = work.request; + if ( + !work.joinable || + existing === null || + work.dialect !== dialect || + existing.scope !== request.scope || + existing.dialectId !== request.dialectId || + existing.limit !== request.limit || + existing.continuationToken !== request.continuationToken || + !sameComponent(existing.prefix, request.prefix) || + !samePath(existing.qualifier, request.qualifier) || + existing.searchPaths.length !== request.searchPaths.length + ) { + return false; + } + const leftEpoch = existing.expectedEpoch; + const rightEpoch = request.expectedEpoch; + if ( + (leftEpoch === null) !== (rightEpoch === null) || + (leftEpoch !== null && + rightEpoch !== null && + (leftEpoch.generation !== rightEpoch.generation || + leftEpoch.token !== rightEpoch.token)) + ) { + return false; + } + for ( + let index = 0; + index < existing.searchPaths.length; + index += 1 + ) { + const left = existing.searchPaths[index]; + const right = request.searchPaths[index]; + if ( + left === undefined || + right === undefined || + !samePath(left, right) + ) { + return false; + } + } + return true; +} + +function removeQueued( + state: CoordinatorState, + work: WorkState, +): void { + const index = state.queue.indexOf(work); + if (index >= 0) state.queue.splice(index, 1); +} + +function removeJoinable( + state: CoordinatorState, + work: WorkState, +): void { + if (!work.joinable) return; + work.joinable = false; + state.joinable.delete(work); +} + +function revokeProviderCell(work: WorkState): void { + const cell = work.providerCell; + work.providerCell = null; + if (!cell) return; + cell.active = false; + cell.deliver = null; +} + +function revokeDecisionCell(work: WorkState): void { + const cell = work.decisionCell; + work.decisionCell = null; + if (!cell) return; + cell.active = false; + cell.candidates = []; + cell.pending = null; + cell.response = null; + cell.state = null; + cell.work = null; +} + +function detachAbort( + work: WorkState, + pending: Effects, +): void { + const controller = work.abortController; + work.abortController = null; + if (controller && !work.abortIssued) { + work.abortIssued = true; + pending.aborts.push(controller); + } +} + +function detachOwners( + work: WorkState, + outcome: SqlCatalogSearchWorkOutcome, + pending: Effects, +): void { + const consumers = [...work.owners]; + work.owners.clear(); + for (const consumer of consumers) { + settleDetached(pending.settlements, consumer, outcome); + } +} + +function finishWork( + state: CoordinatorState, + work: WorkState, + outcome: SqlCatalogSearchWorkOutcome, + pending: Effects, + abort: boolean, +): void { + if (work.phase === "terminal") return; + const occupied = + work.phase === "active" || + work.phase === "deciding" || + work.phase === "retired"; + removeJoinable(state, work); + removeQueued(state, work); + state.works.delete(work); + work.phase = "terminal"; + if (occupied && state.activeCount > 0) { + state.activeCount -= 1; + pending.pump = true; + } + const queueTimer = work.queueTimer; + const executionTimer = work.executionTimer; + work.queueTimer = null; + work.executionTimer = null; + if (queueTimer) pending.timers.push(queueTimer); + if (executionTimer) pending.timers.push(executionTimer); + revokeProviderCell(work); + revokeDecisionCell(work); + detachOwners(work, outcome, pending); + if (abort) detachAbort(work, pending); + else work.abortController = null; + work.dialect = null; + work.request = null; + work.scope = ""; +} + +function retireActiveOwners( + state: CoordinatorState, + work: WorkState, + outcome: SqlCatalogSearchWorkOutcome, + pending: Effects, +): void { + removeJoinable(state, work); + work.phase = "retired"; + detachOwners(work, outcome, pending); + detachAbort(work, pending); + work.dialect = null; + work.request = null; + work.scope = ""; +} + +function detachConsumerInto( + state: CoordinatorState, + consumer: ConsumerState, + outcome: SqlCatalogSearchWorkOutcome, + pending: Effects, +): void { + const work = consumer.work; + if (work) { + work.owners.delete(consumer); + } + settleDetached(pending.settlements, consumer, outcome); + if (work && work.owners.size === 0) { + if (work.phase === "queued") { + finishWork( + state, + work, + CANCELLED_OUTCOME, + pending, + false, + ); + } else if (work.phase === "active") { + retireActiveOwners( + state, + work, + CANCELLED_OUTCOME, + pending, + ); + } else if (work.phase === "deciding") { + finishWork( + state, + work, + CANCELLED_OUTCOME, + pending, + false, + ); + } + } +} + +function detachConsumer( + state: CoordinatorState, + consumer: ConsumerState, + outcome: SqlCatalogSearchWorkOutcome, +): void { + const pending = effects(); + detachConsumerInto( + state, + consumer, + outcome, + pending, + ); + runEffects(state, pending); +} + +function handleQueueTimeout( + state: CoordinatorState, + work: WorkState, +): void { + if (work.phase !== "queued") return; + const pending = effects(); + finishWork( + state, + work, + unavailableOutcome("queue-timeout"), + pending, + false, + ); + runEffects(state, pending); +} + +function handleExecutionTimeout( + state: CoordinatorState, + work: WorkState, +): void { + if ( + work.phase !== "active" && + work.phase !== "retired" && + work.phase !== "deciding" + ) { + return; + } + const pending = effects(); + finishWork( + state, + work, + unavailableOutcome("execution-timeout"), + pending, + true, + ); + runEffects(state, pending); +} + +function workPhase(work: WorkState): WorkState["phase"] { + return work.phase; +} + +function deliverProviderResult( + cell: ProviderResultCell, + fulfilled: boolean, + value: unknown, +): void { + if (!cell.active) return; + cell.active = false; + const deliver = cell.deliver; + cell.deliver = null; + deliver?.(fulfilled, value); +} + +function attachProviderResult( + state: CoordinatorState, + work: WorkState, + value: unknown, +): boolean { + const cell: ProviderResultCell = { + active: true, + deliver: (fulfilled, result): void => { + handleProviderResult(state, work, fulfilled, result); + }, + }; + work.providerCell = cell; + const onFulfilled = (result: unknown): void => { + deliverProviderResult(cell, true, result); + }; + const onRejected = (reason: unknown): void => { + deliverProviderResult(cell, false, reason); + }; + try { + const promise = Reflect.apply( + INTRINSIC_PROMISE_RESOLVE, + INTRINSIC_PROMISE, + [value], + ); + Reflect.apply(INTRINSIC_PROMISE_THEN, promise, [ + onFulfilled, + onRejected, + ]); + return true; + } catch { + revokeProviderCell(work); + return false; + } +} + +function drainDetachedSettlement(value: unknown): void { + try { + const promise = Reflect.apply( + INTRINSIC_PROMISE_RESOLVE, + INTRINSIC_PROMISE, + [value], + ); + Reflect.apply(INTRINSIC_PROMISE_THEN, promise, [ + undefined, + IGNORE_DETACHED_REJECTION, + ]); + } catch { + // The detached result has no authority over coordinator state. + } +} + +function settleDecision( + cell: ResponseDecisionCell, + decision: SqlCatalogResponseEpochDecision, +): void { + if (!cell.active) return; + cell.pending = decision; + advanceDecision(cell); +} + +function finishDecision( + cell: ResponseDecisionCell, + state: CoordinatorState, + work: WorkState, + outcome: SqlCatalogSearchWorkOutcome, +): void { + cell.active = false; + const pending = effects(); + finishWork(state, work, outcome, pending, false); + runEffects(state, pending); +} + +function decisionOutcome( + decision: Exclude< + SqlCatalogResponseEpochDecision, + { readonly status: "usable" } + >, +): SqlCatalogSearchWorkOutcome { + if (decision.status === "superseded") { + return SUPERSEDED_OUTCOME; + } + switch (decision.reason) { + case "disposed": + return unavailableOutcome("disposed"); + case "malformed": + return unavailableOutcome("malformed-response"); + case "overloaded": + return unavailableOutcome("overloaded"); + case "retired": + case "stale": + case "token-conflict": + return SUPERSEDED_OUTCOME; + } +} + +function rekeyUnobservedWork( + state: CoordinatorState, + producingWork: WorkState, + response: SqlValidatedCatalogSearchResponse, +): void { + const scope = producingWork.scope; + for (const work of state.joinable) { + if (work === producingWork || work.scope !== scope) { + continue; + } + const request = work.request; + if (!request || request.expectedEpoch !== null) { + continue; + } + work.request = Object.freeze({ + ...request, + expectedEpoch: response.epoch, + }); + } +} + +function advanceDecision(cell: ResponseDecisionCell): void { + if (!cell.active || cell.advancing) return; + cell.advancing = true; + try { + for (;;) { + if (!cell.active) return; + const state = cell.state; + const work = cell.work; + const response = cell.response; + if (!state || !work || !response) return; + const pendingDecision = cell.pending; + cell.pending = null; + if (pendingDecision) { + if ( + pendingDecision.status === "discarded" && + pendingDecision.reason === "retired" + ) { + // Try a capture belonging to another live shared owner. + } else if (pendingDecision.status === "usable") { + if (pendingDecision.observation === "baseline") { + rekeyUnobservedWork(state, work, response); + } + finishDecision( + cell, + state, + work, + Object.freeze({ + observation: pendingDecision.observation, + response, + status: "usable", + }), + ); + return; + } else { + finishDecision( + cell, + state, + work, + decisionOutcome(pendingDecision), + ); + return; + } + } + let capture: SqlCatalogEpochCapture | null = null; + while (cell.index < cell.candidates.length) { + const candidate = cell.candidates[cell.index]; + cell.index += 1; + if ( + candidate && + !candidate.settled && + candidate.work === work + ) { + capture = candidate.capture; + break; + } + } + if (!capture) { + finishDecision( + cell, + state, + work, + SUPERSEDED_OUTCOME, + ); + return; + } + const submitted = state.epochs.submitResponseEpoch( + capture, + response.epoch, + (decision): void => { + settleDecision(cell, decision); + }, + ); + if (submitted.status === "submitted") { + if (cell.pending === null) return; + continue; + } + cell.pending = submitted.decision; + } + } finally { + cell.advancing = false; + if (cell.active && cell.pending !== null) { + advanceDecision(cell); + } + } +} + +function handleProviderResult( + state: CoordinatorState, + work: WorkState, + fulfilled: boolean, + raw: unknown, +): void { + if ( + work.phase !== "active" && + work.phase !== "retired" + ) { + return; + } + revokeProviderCell(work); + if (work.phase === "retired") { + const pending = effects(); + finishWork( + state, + work, + CANCELLED_OUTCOME, + pending, + false, + ); + runEffects(state, pending); + return; + } + const now = readNow(state); + const phaseAfterClock = workPhase(work); + if ( + state.disposed || + (phaseAfterClock !== "active" && + phaseAfterClock !== "retired") + ) { + return; + } + if (phaseAfterClock === "retired") { + const pending = effects(); + finishWork( + state, + work, + CANCELLED_OUTCOME, + pending, + false, + ); + runEffects(state, pending); + return; + } + if ( + now === null || + work.executionDeadline === null || + now >= work.executionDeadline + ) { + handleExecutionTimeout(state, work); + return; + } + removeJoinable(state, work); + work.phase = "deciding"; + if (!fulfilled) { + const pending = effects(); + finishWork( + state, + work, + unavailableOutcome("provider-failed"), + pending, + false, + ); + runEffects(state, pending); + return; + } + const request = work.request; + const dialect = work.dialect; + if (!request || !dialect) { + const pending = effects(); + finishWork( + state, + work, + unavailableOutcome("disposed"), + pending, + false, + ); + runEffects(state, pending); + return; + } + const decoded = decodeSqlCatalogSearchResponse( + raw, + request.limit, + dialect, + ); + if ( + state.disposed || + work.phase !== "deciding" || + work.owners.size === 0 + ) { + return; + } + const afterDecode = readNow(state); + if ( + state.disposed || + work.phase !== "deciding" || + work.owners.size === 0 || + afterDecode === null || + work.executionDeadline === null || + afterDecode >= work.executionDeadline + ) { + handleExecutionTimeout(state, work); + return; + } + if (decoded.status !== "accepted") { + const pending = effects(); + finishWork( + state, + work, + unavailableOutcome("malformed-response"), + pending, + false, + ); + runEffects(state, pending); + return; + } + const cell: ResponseDecisionCell = { + active: true, + advancing: false, + candidates: [...work.owners], + index: 0, + pending: null, + response: decoded.value, + state, + work, + }; + work.decisionCell = cell; + advanceDecision(cell); +} + +function startWork( + state: CoordinatorState, + work: WorkState, +): void { + const startedAt = readNow(state); + if ( + state.disposed || + work.phase !== "queued" || + work.owners.size === 0 + ) { + return; + } + if ( + startedAt === null || + startedAt >= work.queueDeadline + ) { + handleQueueTimeout(state, work); + return; + } + const queueTimer = work.queueTimer; + work.queueTimer = null; + work.phase = "active"; + state.activeCount += 1; + const controller = new AbortController(); + work.abortController = controller; + const executionDeadline = deadlineFrom( + startedAt, + state.options.executionDeadlineMs, + ); + if (executionDeadline === null) { + handleExecutionTimeout(state, work); + if (queueTimer) clearDeadline(state, queueTimer); + return; + } + work.executionDeadline = executionDeadline; + const executionTimer = scheduleDeadline( + state, + executionDeadline, + () => handleExecutionTimeout(state, work), + ); + if (work.phase !== "active") { + clearDeadline(state, executionTimer); + if (queueTimer) clearDeadline(state, queueTimer); + return; + } + work.executionTimer = executionTimer; + if (queueTimer) clearDeadline(state, queueTimer); + const search = state.search; + const request = work.request; + if (!search || !request || state.disposed) { + const pending = effects(); + finishWork( + state, + work, + unavailableOutcome("disposed"), + pending, + true, + ); + runEffects(state, pending); + return; + } + let returned: unknown; + try { + returned = Reflect.apply(search, undefined, [ + request, + controller.signal, + ]); + } catch { + const pending = effects(); + finishWork( + state, + work, + unavailableOutcome("provider-failed"), + pending, + true, + ); + runEffects(state, pending); + return; + } + const phaseAfterSearch = workPhase(work); + if ( + state.disposed || + phaseAfterSearch === "terminal" + ) { + drainDetachedSettlement(returned); + return; + } + const attached = attachProviderResult(state, work, returned); + const observedAt = readNow(state); + if ( + !attached || + observedAt === null || + observedAt - startedAt > + state.options.synchronousBudgetMs + ) { + const pending = effects(); + finishWork( + state, + work, + unavailableOutcome( + attached + ? "execution-timeout" + : "provider-failed", + ), + pending, + true, + ); + runEffects(state, pending); + } +} + +function pump(state: CoordinatorState): void { + state.pumpRequested = true; + if (state.pumping || state.disposed) return; + state.pumping = true; + try { + while (state.pumpRequested && !state.disposed) { + state.pumpRequested = false; + while ( + state.activeCount < + MAX_CATALOG_ACTIVE_SEARCH_WORK && + state.queue.length > 0 + ) { + const work = state.queue.shift(); + if (!work) break; + startWork(state, work); + } + } + } finally { + state.pumping = false; + } +} + +function makeImmediateTicket( + outcome: SqlCatalogSearchWorkOutcome, +): SqlCatalogSearchWorkTicket { + return Object.freeze({ + cancel: (): void => {}, + result: new INTRINSIC_PROMISE( + (resolve) => { + resolve(outcome); + }, + ), + }); +} + +function makeConsumer( + capture: SqlCatalogEpochCapture, + owner: OwnerState, +): { + readonly consumer: ConsumerState; + readonly ticket: SqlCatalogSearchWorkTicket; +} { + let resolve: (outcome: SqlCatalogSearchWorkOutcome) => void = + IGNORE_DETACHED_REJECTION; + const result = + new INTRINSIC_PROMISE( + (settle) => { + resolve = settle; + }, + ); + const consumer: ConsumerState = { + cancelled: false, + capture, + owner, + resolve, + settled: false, + work: null, + }; + return { + consumer, + ticket: Object.freeze({ + cancel: (): void => { + if (consumer.cancelled || consumer.settled) return; + consumer.cancelled = true; + const state = consumer.owner?.owner; + if (state) { + detachConsumer( + state, + consumer, + CANCELLED_OUTCOME, + ); + } + }, + result, + }), + }; +} + +function findWork( + state: CoordinatorState, + request: SqlCatalogSearchRequest, + dialect: SqlRelationDialectRuntime, +): WorkState | null { + for (const work of state.joinable) { + if (sameRequest(work, request, dialect)) return work; + } + return null; +} + +function replaceOwnerConsumer( + state: CoordinatorState, + owner: OwnerState, + consumer: ConsumerState, + work: WorkState, +): void { + const pending = effects(); + const previous = owner.current; + work.owners.add(consumer); + consumer.work = work; + owner.current = consumer; + if (previous && previous !== consumer) { + detachConsumerInto( + state, + previous, + SUPERSEDED_OUTCOME, + pending, + ); + } + runEffects(state, pending); +} + +function requestWork( + state: CoordinatorState, + owner: OwnerState, + input: SqlCatalogSearchWorkInput, +): SqlCatalogSearchWorkTicket { + const requestToken = {}; + owner.requestToken = requestToken; + const captured = owner.membership.captureEpoch(); + if ( + state.disposed || + owner.disposed || + owner.owner !== state + ) { + return makeImmediateTicket( + unavailableOutcome("disposed"), + ); + } + if (owner.requestToken !== requestToken) { + return makeImmediateTicket(SUPERSEDED_OUTCOME); + } + if (captured.status !== "captured") { + const pending = effects(); + const previous = owner.current; + if (previous) { + detachConsumerInto( + state, + previous, + SUPERSEDED_OUTCOME, + pending, + ); + } + runEffects(state, pending); + return makeImmediateTicket( + unavailableOutcome( + captured.reason === "inactive" + ? "inactive" + : "disposed", + ), + ); + } + let request: SqlCatalogSearchRequest | null = null; + try { + const requestResult = createSqlCatalogSearchRequest({ + continuationToken: input.continuationToken, + dialectId: owner.dialect.id, + expectedEpoch: captured.capture.expectedEpoch, + limit: input.limit, + prefix: input.prefix, + qualifier: input.qualifier, + scope: owner.scope, + searchPaths: input.searchPaths, + }); + if (requestResult.status === "accepted") { + request = requestResult.value; + } + } catch { + // Hostile runtime input is an invalid request. + } + if ( + state.disposed || + owner.disposed || + owner.owner !== state + ) { + return makeImmediateTicket( + unavailableOutcome("disposed"), + ); + } + if (owner.requestToken !== requestToken) { + return makeImmediateTicket(SUPERSEDED_OUTCOME); + } + if (!request) { + const pending = effects(); + const previous = owner.current; + if (previous) { + detachConsumerInto( + state, + previous, + SUPERSEDED_OUTCOME, + pending, + ); + } + runEffects(state, pending); + return makeImmediateTicket( + unavailableOutcome("invalid-request"), + ); + } + const created = makeConsumer(captured.capture, owner); + const existing = findWork( + state, + request, + owner.dialect, + ); + if (existing) { + replaceOwnerConsumer( + state, + owner, + created.consumer, + existing, + ); + return created.ticket; + } + const pending = effects(); + const previous = owner.current; + if (previous) { + detachConsumerInto( + state, + previous, + SUPERSEDED_OUTCOME, + pending, + ); + } + if ( + state.queue.length >= + MAX_CATALOG_QUEUED_SEARCH_WORK + ) { + runEffects(state, pending); + return makeImmediateTicket( + unavailableOutcome("overloaded"), + ); + } + const now = readNow(state); + const disposedDuringClock = + state.disposed || + owner.disposed || + owner.owner !== state; + if ( + owner.requestToken !== requestToken || + disposedDuringClock + ) { + runEffects(state, pending); + return makeImmediateTicket( + disposedDuringClock + ? unavailableOutcome("disposed") + : SUPERSEDED_OUTCOME, + ); + } + if (now === null) { + runEffects(state, pending); + return makeImmediateTicket( + unavailableOutcome("execution-timeout"), + ); + } + const queueDeadline = deadlineFrom( + now, + state.options.queueDeadlineMs, + ); + if (queueDeadline === null) { + runEffects(state, pending); + return makeImmediateTicket( + unavailableOutcome("execution-timeout"), + ); + } + const work: WorkState = { + abortController: null, + abortIssued: false, + decisionCell: null, + dialect: owner.dialect, + executionDeadline: null, + executionTimer: null, + joinable: true, + owners: new Set([created.consumer]), + phase: "queued", + providerCell: null, + queueDeadline, + queueTimer: null, + request, + scope: owner.scope, + }; + created.consumer.work = work; + owner.current = created.consumer; + state.joinable.add(work); + state.works.add(work); + state.queue.push(work); + runEffects(state, pending); + pump(state); + if (work.phase !== "queued") { + return created.ticket; + } + const queueTimer = scheduleDeadline( + state, + queueDeadline, + () => handleQueueTimeout(state, work), + ); + if (work.phase === "queued") { + work.queueTimer = queueTimer; + } else { + clearDeadline(state, queueTimer); + } + return created.ticket; +} + +function prepareTransition( + state: CoordinatorState, + scope: string, +): (() => undefined) | null { + if (state.disposed) return null; + const pending = effects(); + for (const work of state.works) { + if ( + work.scope !== scope || + work.phase === "retired" || + work.phase === "terminal" + ) { + continue; + } + if (work.phase === "active") { + retireActiveOwners( + state, + work, + SUPERSEDED_OUTCOME, + pending, + ); + } else { + finishWork( + state, + work, + SUPERSEDED_OUTCOME, + pending, + false, + ); + } + } + if ( + pending.aborts.length === 0 && + pending.settlements.length === 0 && + pending.timers.length === 0 && + !pending.pump + ) { + return null; + } + return (): undefined => { + runEffects(state, pending); + return undefined; + }; +} + +function unavailableOwner( + reason: Exclude< + SqlCatalogSearchWorkOwnerResult, + { readonly status: "prepared" } + >["reason"], +): SqlCatalogSearchWorkOwnerResult { + return Object.freeze({ reason, status: "unavailable" }); +} + +function disposeOwner(owner: OwnerState): void { + if (owner.disposed) return; + const state = owner.owner; + owner.disposed = true; + owner.owner = null; + owner.requestToken = null; + state?.owners.delete(owner); + const current = owner.current; + owner.current = null; + if (state && current) { + detachConsumer(state, current, CANCELLED_OUTCOME); + } + owner.membership.dispose(); +} + +function createOwnerHandle( + owner: OwnerState, +): SqlCatalogSearchWorkOwner { + return Object.freeze({ + activate: () => { + return owner.membership.activate(); + }, + dispose: (): void => { + disposeOwner(owner); + }, + request: ( + input: SqlCatalogSearchWorkInput, + ): SqlCatalogSearchWorkTicket => { + const state = owner.owner; + if (!state || owner.disposed) { + return makeImmediateTicket( + unavailableOutcome("disposed"), + ); + } + return requestWork(state, owner, input); + }, + }); +} + +function prepareOwner( + state: CoordinatorState, + scope: unknown, + dialect: SqlRelationDialectRuntime, + target: SqlCatalogRevisionTarget, +): SqlCatalogSearchWorkOwnerResult { + if (state.disposed) return unavailableOwner("disposed"); + if (!isValidSqlCatalogScope(scope)) { + return unavailableOwner("invalid-scope"); + } + if (!isSqlRelationDialectRuntime(dialect)) { + return unavailableOwner("invalid-dialect"); + } + const prepared = state.epochs.prepareScopeMembership( + scope, + target, + ); + if (prepared.status !== "prepared") { + return unavailableOwner( + prepared.reason === "disposed" + ? "disposed" + : prepared.reason, + ); + } + const owner: OwnerState = { + current: null, + dialect, + disposed: false, + membership: prepared.membership, + owner: state, + requestToken: null, + scope, + }; + state.owners.add(owner); + return Object.freeze({ + owner: createOwnerHandle(owner), + status: "prepared", + }); +} + +function disposeCoordinatorState( + state: CoordinatorState, +): void { + if (state.disposed) return; + state.disposed = true; + state.search = null; + state.pumpRequested = false; + const pending = effects(); + for (const work of state.works) { + finishWork( + state, + work, + unavailableOutcome("disposed"), + pending, + true, + ); + } + for (const owner of state.owners) { + owner.disposed = true; + owner.current = null; + owner.owner = null; + owner.requestToken = null; + } + state.owners.clear(); + state.epochs.dispose(); + runEffects(state, pending); +} + +export function createSqlCatalogSearchWorkCoordinator( + provider: CapturedSqlRelationCatalogProvider, + options?: SqlCatalogSearchWorkOptions, +): SqlCatalogSearchWorkCoordinatorResult { + const context = resolveSqlRelationCatalogProvider(provider); + if (!context) { + return Object.freeze({ + reason: "invalid-provider", + status: "unavailable", + }); + } + const normalized = normalizeOptions(options); + if (!normalized) { + return Object.freeze({ + reason: "invalid-options", + status: "unavailable", + }); + } + let state: CoordinatorState | null = null; + const epochResult = createSqlCatalogEpochCoordinator( + provider, + (scope): (() => undefined) | null => + state ? prepareTransition(state, scope) : null, + (): undefined => { + if (state) disposeCoordinatorState(state); + return undefined; + }, + ); + if (epochResult.status !== "created") { + return Object.freeze({ + reason: "invalid-provider", + status: "unavailable", + }); + } + state = { + activeCount: 0, + disposed: false, + epochs: epochResult.coordinator, + joinable: new Set(), + lastNow: normalized.initialNow, + options: normalized.options, + owners: new Set(), + pumpRequested: false, + pumping: false, + queue: [], + search: context.search, + works: new Set(), + }; + const capturedState = state; + return Object.freeze({ + coordinator: Object.freeze({ + dispose: (): void => { + disposeCoordinatorState(capturedState); + }, + prepareOwner: ( + scope: unknown, + dialect: SqlRelationDialectRuntime, + target: SqlCatalogRevisionTarget, + ): SqlCatalogSearchWorkOwnerResult => + prepareOwner( + capturedState, + scope, + dialect, + target, + ), + providerId: context.id, + }), + status: "created", + }); +} diff --git a/src/vnext/relation-dialect.ts b/src/vnext/relation-dialect.ts index 76bbf50..39c2fe5 100644 --- a/src/vnext/relation-dialect.ts +++ b/src/vnext/relation-dialect.ts @@ -35,10 +35,11 @@ import type { SqlIdentifierComponent } from "./types.js"; export interface SqlRelationDialectRuntime { readonly completion: SqlRelationCompletionDialectRuntime; readonly cteLayout: SqlCteLayoutDialect; + readonly id: SqlRelationDialectId; readonly querySite: SqlQuerySiteDialect; } -type DialectKind = +export type SqlRelationDialectId = | "bigquery" | "dremio" | "duckdb" @@ -46,7 +47,7 @@ type DialectKind = interface RelationDialectSpec { readonly cteGrammar: SqlCteLayoutDialect["grammar"]; - readonly kind: DialectKind; + readonly kind: SqlRelationDialectId; readonly lexicalProfile: SqlLexicalProfile; readonly maximumPathDepth: number; readonly supportsNaturalJoin: boolean; @@ -245,7 +246,7 @@ function decodedIdentifier( function decodeDoubleQuoted( token: string, mode: "complete" | "completion-prefix", - kind: Exclude, + kind: Exclude, ): SqlIdentifierDecodeResult { if (token.length > MAX_STANDARD_QUOTED_IDENTIFIER_RAW_LENGTH) { return INVALID_IDENTIFIER; @@ -598,7 +599,7 @@ function decodeSegmentedPath( maximumPathDepth: number, quote: "\"" | "`", decodeIdentifier: SqlRelationCompletionDialectRuntime["decodeIdentifier"], - kind: DialectKind, + kind: SqlRelationDialectId, ): SqlDecodedQueryPath { if ( typeof rawPath !== "string" || @@ -883,7 +884,7 @@ function createPathDecoder( } function createQueryClassifier( - kind: DialectKind, + kind: SqlRelationDialectId, decodeIdentifier: SqlRelationCompletionDialectRuntime["decodeIdentifier"], ): SqlQuerySiteDialect["classifyIdentifierToken"] { return (rawIdentifier, quoted, role) => { @@ -1167,7 +1168,7 @@ function quoteBigQuery(value: string): string { } function legalRoleSequence( - kind: DialectKind, + kind: SqlRelationDialectId, roles: readonly string[], ): boolean { if (roles.length === 0 || roles.at(-1) !== "relation") { @@ -1420,6 +1421,7 @@ function createRuntime(spec: RelationDialectSpec): SqlRelationDialectRuntime { Object.freeze({ completion, cteLayout, + id: spec.kind, querySite, }), ); diff --git a/test/vnext-types/relation-catalog-search-work.test-d.ts b/test/vnext-types/relation-catalog-search-work.test-d.ts new file mode 100644 index 0000000..8176b88 --- /dev/null +++ b/test/vnext-types/relation-catalog-search-work.test-d.ts @@ -0,0 +1,257 @@ +import type { + CapturedSqlRelationCatalogProvider, + SqlValidatedCatalogSearchResponse, +} from "../../src/vnext/relation-catalog-boundary.js"; +import type { + SqlCatalogRevisionTarget, +} from "../../src/vnext/relation-catalog-epoch-coordinator.js"; +import { + createSqlCatalogEpochCoordinator, +} from "../../src/vnext/relation-catalog-epoch-coordinator.js"; +import type { + SqlCatalogSearchWorkCoordinator, + SqlCatalogSearchWorkInput, + SqlCatalogSearchWorkOutcome, + SqlCatalogSearchWorkOwner, + SqlCatalogSearchWorkOwnerResult, + SqlCatalogSearchWorkTicket, +} from "../../src/vnext/relation-catalog-search-work.js"; +import { + createSqlCatalogSearchWorkCoordinator, +} from "../../src/vnext/relation-catalog-search-work.js"; +import type { + SqlRelationDialectRuntime, +} from "../../src/vnext/relation-dialect.js"; + +declare const capturedProvider: CapturedSqlRelationCatalogProvider; +declare const coordinator: SqlCatalogSearchWorkCoordinator; +declare const dialect: SqlRelationDialectRuntime; +declare const owner: SqlCatalogSearchWorkOwner; +declare const target: SqlCatalogRevisionTarget; +declare const response: SqlValidatedCatalogSearchResponse; +declare const ticket: SqlCatalogSearchWorkTicket; + +const input: SqlCatalogSearchWorkInput = { + continuationToken: null, + limit: 25, + prefix: { quoted: false, value: "ord" }, + qualifier: [{ quoted: false, value: "analytics" }], + searchPaths: [ + [ + { quoted: false, value: "warehouse" }, + { quoted: false, value: "public" }, + ], + ], +}; + +const created = createSqlCatalogSearchWorkCoordinator( + capturedProvider, +); +if (created.status === "created") { + const prepared = created.coordinator.prepareOwner( + "notebook:demo", + dialect, + target, + ); + if (prepared.status === "prepared") { + const activation = prepared.owner.activate(); + if (activation.status === "active") { + const ticket = prepared.owner.request(input); + ticket.cancel(); + void ticket.result; + } + prepared.owner.dispose(); + } + created.coordinator.dispose(); +} + +const thisFreeCoordinatorDispose: ( + this: void, +) => void = coordinator.dispose; +const thisFreePrepareOwner: SqlCatalogSearchWorkCoordinator["prepareOwner"] = + coordinator.prepareOwner; +const thisFreeActivate: SqlCatalogSearchWorkOwner["activate"] = + owner.activate; +const thisFreeOwnerDispose: ( + this: void, +) => void = owner.dispose; +const thisFreeRequest: ( + this: void, + input: SqlCatalogSearchWorkInput, +) => SqlCatalogSearchWorkTicket = owner.request; +const thisFreeCancel: ( + this: void, +) => void = ticket.cancel; + +const receiverDependentRequest = function ( + this: { readonly active: boolean }, + _input: SqlCatalogSearchWorkInput, +): SqlCatalogSearchWorkTicket { + void this.active; + throw new Error("type fixture only"); +}; +// @ts-expect-error request callbacks cannot depend on a receiver +const invalidRequestReceiver: SqlCatalogSearchWorkOwner["request"] = + receiverDependentRequest; + +const receiverDependentPrepareOwner = function ( + this: { readonly active: boolean }, + _scope: unknown, + _dialect: unknown, + _target: SqlCatalogRevisionTarget, +): SqlCatalogSearchWorkOwnerResult { + void this.active; + return { reason: "disposed", status: "unavailable" }; +}; +// @ts-expect-error owner preparation cannot depend on a receiver +const invalidPrepareOwnerReceiver: SqlCatalogSearchWorkCoordinator["prepareOwner"] = + receiverDependentPrepareOwner; + +const asyncDisposalTarget = async (): Promise => {}; +type EpochDisposalTarget = NonNullable< + Parameters[2] +>; +// @ts-expect-error package disposal targets are synchronously exact-undefined +const invalidAsyncDisposalTarget: EpochDisposalTarget = + asyncDisposalTarget; +void invalidAsyncDisposalTarget; + +const receiverDependentActivate = function ( + this: { readonly active: boolean }, +): ReturnType { + void this.active; + return { status: "active" }; +}; +// @ts-expect-error owner activation cannot depend on a receiver +const invalidActivateReceiver: SqlCatalogSearchWorkOwner["activate"] = + receiverDependentActivate; + +const receiverDependentCancel = function ( + this: { readonly active: boolean }, +): void { + void this.active; +}; +// @ts-expect-error cancellation callbacks cannot depend on a receiver +const invalidCancelReceiver: SqlCatalogSearchWorkTicket["cancel"] = + receiverDependentCancel; + +// @ts-expect-error request input is readonly +input.limit = 50; +// @ts-expect-error ticket result ownership is readonly +ticket.result = Promise.resolve({ status: "cancelled" }); + +const scopeLeakingInput: SqlCatalogSearchWorkInput = { + ...input, + // @ts-expect-error scope belongs to the prepared owner, not a request + scope: "notebook:other", +}; +const providerLeakingInput: SqlCatalogSearchWorkInput = { + ...input, + // @ts-expect-error provider identity belongs to the coordinator + providerId: "foreign", +}; +const providerHandleLeakingInput: SqlCatalogSearchWorkInput = { + ...input, + // @ts-expect-error provider handles belong to the coordinator + provider: capturedProvider, +}; +const epochLeakingInput: SqlCatalogSearchWorkInput = { + ...input, + // @ts-expect-error epochs are captured by the prepared owner + expectedEpoch: { generation: 1, token: "foreign" }, +}; +const epochAliasLeakingInput: SqlCatalogSearchWorkInput = { + ...input, + // @ts-expect-error epoch aliases are captured by the prepared owner + epoch: { generation: 1, token: "foreign" }, +}; +const dialectLeakingInput: SqlCatalogSearchWorkInput = { + ...input, + // @ts-expect-error dialect identity belongs to the prepared owner + dialectId: "postgresql", +}; +const runtimeLeakingInput: SqlCatalogSearchWorkInput = { + ...input, + // @ts-expect-error dialect runtime belongs to the prepared owner + dialect, +}; + +const usableOutcome: SqlCatalogSearchWorkOutcome = { + observation: "baseline", + response, + status: "usable", +}; +// @ts-expect-error outcomes are readonly +usableOutcome.status = "cancelled"; +const extraOutcomeField: SqlCatalogSearchWorkOutcome = { + // @ts-expect-error outcome variants reject foreign fields + providerId: "foreign", + observation: "equal", + response, + status: "usable", +}; +const unknownOutcomeStatus: SqlCatalogSearchWorkOutcome = { + // @ts-expect-error outcome status is a closed discriminant + status: "loading", +}; + +function consumeOutcome(outcome: SqlCatalogSearchWorkOutcome): void { + switch (outcome.status) { + case "usable": + void outcome.response; + break; + case "superseded": + case "cancelled": + break; + case "unavailable": + switch (outcome.reason) { + case "disposed": + case "execution-timeout": + case "inactive": + case "invalid-request": + case "malformed-response": + case "overloaded": + case "provider-failed": + case "queue-timeout": + break; + default: { + const exhaustiveReason: never = outcome.reason; + void exhaustiveReason; + } + } + break; + default: { + const exhaustiveOutcome: never = outcome; + void exhaustiveOutcome; + } + } +} + +// @ts-expect-error captured providers are authentic, not structural objects +const foreignCapturedProvider: CapturedSqlRelationCatalogProvider = {}; + +declare const ownerResult: SqlCatalogSearchWorkOwnerResult; +if (ownerResult.status === "prepared") { + void ownerResult.owner.request(input).result.then(consumeOutcome); +} + +void thisFreeCoordinatorDispose; +void thisFreePrepareOwner; +void thisFreeActivate; +void thisFreeOwnerDispose; +void thisFreeRequest; +void thisFreeCancel; +void invalidPrepareOwnerReceiver; +void invalidActivateReceiver; +void invalidRequestReceiver; +void invalidCancelReceiver; +void scopeLeakingInput; +void providerLeakingInput; +void providerHandleLeakingInput; +void epochLeakingInput; +void epochAliasLeakingInput; +void dialectLeakingInput; +void runtimeLeakingInput; +void extraOutcomeField; +void unknownOutcomeStatus; +void foreignCapturedProvider;