diff --git a/docs/adr/0005-parser-independent-relation-completion.md b/docs/adr/0005-parser-independent-relation-completion.md index 13d6ae7..967ec05 100644 --- a/docs/adr/0005-parser-independent-relation-completion.md +++ b/docs/adr/0005-parser-independent-relation-completion.md @@ -347,7 +347,11 @@ search request contains only copied, recursively frozen plain data: The `AbortSignal` is passed separately. Providers never receive a session, revision, `EditorState`, DOM object, credential object, or arbitrary live host context. Connection identity belongs in the catalog scope; live resources stay -inside the provider closure. The provider object is caller-owned. The service +inside the provider closure. The scope is an opaque host-generated identity +that includes the live connection incarnation, not merely a reusable display +name. Rebinding a marimo engine variable to a different connection therefore +uses a new scope even when the variable name is unchanged. The provider object +is caller-owned. The service captures its own-data callbacks once and owns searches, subscriptions, and their cleanup, but does not imply provider-wide disposal. Provider callbacks are closure functions with a declared `this: void` contract and are invoked @@ -389,15 +393,23 @@ Search responses distinguish: - loading; and - failed with a closed code and retry policy. -A terminal `loading` response is not cached. A provider that later becomes -ready must publish a strictly higher epoch through its subscription; same-epoch -duplicate invalidation is not a readiness signal. A still-pending search -promise can instead resolve ready under the service-owned response/refresh -rules below. Every completion result carrying `catalog-loading`, whether caused -by terminal loading or soft expiry, also carries a checked remaining -completion-intent lease. For terminal loading that lease is independent of -in-flight work ownership and is keyed to the exact session, query, and captured -observed or unobserved epoch. +A terminal `loading` response is not cached. It installs a bounded barrier for +the exact request key and captured epoch. When the provider has a live +subscription, matching requests observe loading without invoking it again +until a strictly higher epoch retires the barrier. Same-epoch duplicate +invalidation is not a readiness signal. Without a subscription, a later +explicit request may become the single shared probe behind that barrier. The +probe uses the normal queue and execution budgets and may publish readiness +only with a strictly higher epoch; a same-epoch ready response cannot cross the +loading barrier. A still-pending original search promise can instead resolve +ready before returning terminal loading under the service-owned +response/refresh rules below. + +Every completion result carrying `catalog-loading`, whether caused by terminal +loading or soft expiry, also carries a checked remaining completion-intent +lease. For terminal loading that lease is independent of in-flight work +ownership and is keyed to the exact session, query, scope incarnation, and +captured observed or unobserved epoch. Partial positive entities may be offered. A partial or absent result never proves that a relation does not exist. A complete empty search means only that @@ -501,9 +513,12 @@ searches remain usable; only the epoch observation is a duplicate. A terminal `loading` response requires a higher epoch before that exact search can become ready. Providers without subscriptions may return `loading`, but they cannot cause automatic refresh; a later explicit request must observe the higher -epoch. Failed responses are attempt evidence: `next-request` may recover at -the same epoch, while `after-invalidation` and `never` create bounded retry -gates rather than cached search results. +epoch through the single shared probe. Failed responses are attempt evidence, +not cache entries. `next-request` installs no gate and may recover at the same +epoch. `after-invalidation` blocks the exact request until a strictly higher +epoch. `never` blocks it for the remaining scope incarnation and is retired +only with that incarnation or the service. Retry gates are bounded internal +metadata; they never acquire result authority or extend work deadlines. The service reference-counts one provider subscription per provider configuration and scope, then fans invalidation out to subscribed sessions. @@ -733,8 +748,10 @@ and CTE evidence is composed first and never waits past the remaining response budget. On soft expiry the request settles ready, possibly empty, with `isIncomplete: true`, `catalog-loading`, and bounded remaining completion-intent lease metadata; its session may atomically retag the request -consumer as a service-owned refresh observer. The intent lease is no longer -than the remaining work lease. +consumer as a service-owned refresh observer. The transfer removes consumer +authority and installs observer authority as one serialized mutation, with no +zero-owner interval in which shared work can be aborted or a settlement can be +lost. The intent lease is no longer than the remaining work lease. That bounded refresh lease can retain the shared operation only until its existing hard deadline and active/queued service limits. With no consumers or live observers, the service removes and aborts the work. An observer is bound @@ -754,24 +771,44 @@ 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. +Terminal loading has no work consumer to transfer. The subsequent session +composition increment owns its independent intent, retaining only the bounded +exact key, scope incarnation, captured epoch, and lease. A higher accepted +response or invalidation retires it and emits only the normal higher-epoch +`catalog` revision. By contrast, a compatible same-epoch pending-work +settlement emits exactly one `catalog-availability` notification. A higher +epoch retires both kinds of observer and never emits a second availability +notification. + +The package-private coordinator is delivered in two increments. The first +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. The second is +one combined cache/loading/retry/observer increment over that scheduler. It +adds the ready-page cache, terminal-loading barriers and probes, retry gates, +and atomic consumer-to-observer transfer. The session increment then adds +independent terminal-loading intents together with the revision events they +own; splitting the store states would permit stale cache hits or missed +readiness transitions, while placing a session event lease in the scheduler +would invert ownership. + +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. + +The combined increment exposes only package-private lifecycle outcomes and +observer primitives. Selection of the 40 ms completion-response budget, +session revision ownership, conversion of readiness into session events, +pagination composition, ranking, and CodeMirror menu refresh remain subsequent +increments. A marimo adapter will map connection replacement, deferred +schema/table resolution, local-table changes, and DuckDB catalog DDL to higher +epochs. Its unresolved `schemas_resolved`, `tables_resolved`, and +`child_schemas_resolved` states map to terminal loading, not complete-empty +catalog evidence. The exact structural cache and shared-work key contains: @@ -792,15 +829,34 @@ to its epoch. Only decoded ready responses are cached under their response epoch. Loading, failure, malformed, timeout, cancellation, supersession, queue overload, and disposal outcomes are not cached. Complete-empty results are reused only for the exact key and never prove an unknown-object diagnostic. -Partial and paginated entries retain incomplete coverage; pages combine only -for the identical base key and epoch, and each continuation remains a distinct -request key. Higher-epoch results may be cached only after older scope entries -are cleared and never publish to the revision that observed the older epoch. - -Decoded entity IDs are unique for `(provider configuration, scope, epoch)` -across every page composed into one result. A repeated ID, even with otherwise -identical data, makes the page chain malformed before ranking or caching; array -and page order therefore cannot resolve conflicting identity records. +Partial and paginated entries retain incomplete coverage. This increment +caches individual decoded pages only; each continuation is a distinct request +key, and automatic page following or composition remains deferred. +Higher-epoch results may be cached only after older scope entries are cleared +and never publish to the revision that observed the older epoch. + +Cache retention is bounded simultaneously to 256 entries and 2 MiB of +deterministically estimated retained bytes, including the copied structural key +and frozen decoded page. A successful exact-key lookup and an insertion each +receive the next service-owned access sequence. Before an insertion becomes +visible, least-recently used entries are evicted in ascending access sequence +until both bounds hold. A page whose own estimate exceeds the byte bound +remains usable for its attached consumers but is not cached. Epoch retirement +removes incompatible entries before admission, and eviction invokes no +provider or host callback. + +Loading and retry gates share the bounds but are correctness state, not cache +entries. Their admission evicts ready pages first; an admitted gate is never +evicted before its epoch or scope-incarnation retirement rule. If the bounds +contain only live gates and cannot admit another, the producing request fails +closed as overloaded and the policy store remains overloaded until its provider +configuration is replaced. This configuration-level quarantine prevents a +later request from publishing across the barrier that could not be retained. + +Decoded entity IDs are unique within each cached page for `(provider +configuration, scope, epoch)`. Cross-page uniqueness is checked later by the +pagination-composition increment; array or page order can never resolve +conflicting identity records. ### Initial resource budgets @@ -973,8 +1029,9 @@ Mixed cached assets fail the exact version check and retire the generation. 5. Add the service-owned scoped epoch/subscription coordinator and its hostile lifecycle contract suite. It does not invoke provider search or retain result pages. -6. Add catalog search scheduling, cache, in-flight sharing, cancellation, - retry gates, and availability observers over the proven epoch coordinator. +6. Add catalog search scheduling and in-flight sharing, then one combined + package-private cache/loading/retry/availability-observer increment over the + proven epoch coordinator. 7. Add the session completion method and deterministic composition. 8. Add the separate CodeMirror adapter and packed/browser fixtures. 9. Add the pinned packed/browser/runtime marimo fixture and 1/10/50-editor diff --git a/implementation.md b/implementation.md index 47493ab..906ccb1 100644 --- a/implementation.md +++ b/implementation.md @@ -570,6 +570,29 @@ Prefer reviewers with different prompts or models; two identical agents are correlated evidence. Independent downstream work may target an accepted interface contract, but cannot merge until its dependency is accepted. +### Delivery cadence + +Keep the rigor at candidate boundaries without paying the full cost after every +small edit: + +- Combine naturally coupled package-private work into one coherent vertical + slice when separate PRs would only create temporary APIs and duplicate + review cycles. +- Run focused unit, type, lint, and benchmark checks while developing. Run the + complete local gate once at a clean candidate commit and again only after a + material fix. +- Start the two independent reviews in parallel against that coherent + candidate. A reviewer may recheck a narrow non-architectural amendment + without repeating unrelated probes; public-contract, concurrency, or + lifecycle changes still require two full exact-head approvals. +- Treat hosted CI as the authoritative cross-platform and package-matrix rerun. + Do not duplicate the same unchanged full matrix between every local commit. +- Parallelize independent design, implementation, consumer research, and + review work. Time-box exploratory research and record nonblocking ideas as + follow-ups instead of expanding the active slice. +- Request Copilot once when the PR first has a coherent reviewable head. Never + delay a later exact-head fix solely to obtain another Copilot pass. + ### Review automation Generate one review packet containing: diff --git a/src/vnext/__tests__/relation-catalog-search-policy-store.test.ts b/src/vnext/__tests__/relation-catalog-search-policy-store.test.ts new file mode 100644 index 0000000..9f5d414 --- /dev/null +++ b/src/vnext/__tests__/relation-catalog-search-policy-store.test.ts @@ -0,0 +1,694 @@ +import { describe, expect, it } from "vitest"; +import type { + SqlValidatedCatalogRelation, + SqlValidatedCatalogSearchResponse, +} from "../relation-catalog-boundary.js"; +import { + createSqlCatalogSearchPolicyStore, + MAX_CATALOG_POLICY_STORE_ENTRIES, + MAX_CATALOG_POLICY_STORE_RETAINED_BYTES, +} from "../relation-catalog-search-policy-store.js"; +import { + DUCKDB_SQL_RELATION_DIALECT, + POSTGRESQL_SQL_RELATION_DIALECT, +} from "../relation-dialect.js"; +import type { + SqlCanonicalRelationPath, + SqlCatalogEpoch, + SqlCatalogReadyCoverage, + SqlCatalogSearchRequest, +} from "../relation-completion-types.js"; + +function epoch(generation: number): SqlCatalogEpoch { + return Object.freeze({ + generation, + token: `epoch-${generation}`, + }); +} + +function request( + prefix: string, + expectedEpoch: SqlCatalogEpoch | null, + options: { + readonly continuationToken?: string | null; + readonly dialectId?: string; + readonly limit?: number; + readonly qualifier?: string; + readonly quoted?: boolean; + readonly searchPath?: string; + readonly scope?: string; + } = {}, +): SqlCatalogSearchRequest { + return Object.freeze({ + continuationToken: options.continuationToken ?? null, + dialectId: options.dialectId ?? "postgresql", + expectedEpoch, + limit: options.limit ?? 20, + prefix: Object.freeze({ + quoted: options.quoted ?? false, + value: prefix, + }), + qualifier: Object.freeze([ + Object.freeze({ + quoted: false, + value: options.qualifier ?? "public", + }), + ]), + scope: options.scope ?? "scope-a", + searchPaths: Object.freeze([ + Object.freeze([ + Object.freeze({ + quoted: false, + value: options.searchPath ?? "public", + }), + ]), + ]), + }); +} + +function relation( + entityId: string, + detail?: string, +): SqlValidatedCatalogRelation { + const path: SqlCanonicalRelationPath = Object.freeze([ + Object.freeze({ + quoted: false, + role: "relation" as const, + value: entityId, + }), + ]); + const base = { + canonicalPath: path, + completionPath: path, + completionPathStart: 0, + completionText: entityId, + entityId, + matchQuality: "exact" as const, + relationKind: "table" as const, + }; + return Object.freeze( + detail === undefined ? base : { ...base, detail }, + ); +} + +function ready( + value: SqlCatalogEpoch, + coverage: SqlCatalogReadyCoverage = { kind: "complete" }, + relations: readonly SqlValidatedCatalogRelation[] = [], +): Extract< + SqlValidatedCatalogSearchResponse, + { readonly status: "ready" } +> { + return Object.freeze({ + coverage: Object.freeze(coverage), + epoch: value, + relations: Object.freeze(relations), + status: "ready", + }); +} + +function loading( + value: SqlCatalogEpoch, +): Extract< + SqlValidatedCatalogSearchResponse, + { readonly status: "loading" } +> { + return Object.freeze({ epoch: value, status: "loading" }); +} + +function failed( + value: SqlCatalogEpoch, + retry: "after-invalidation" | "never" | "next-request", +): Extract< + SqlValidatedCatalogSearchResponse, + { readonly status: "failed" } +> { + return Object.freeze({ + code: "unavailable", + epoch: value, + retry, + status: "failed", + }); +} + +describe("relation catalog search policy store", () => { + it("caches ready baseline pages only under their concrete exact epoch key", () => { + const store = createSqlCatalogSearchPolicyStore(); + const firstEpoch = epoch(1); + const baseline = request("u", null); + const response = ready( + firstEpoch, + { + continuationToken: "next", + kind: "paginated", + }, + [relation("users")], + ); + + expect( + store.record( + baseline, + POSTGRESQL_SQL_RELATION_DIALECT, + response, + ), + ).toMatchObject({ retained: "ready", status: "accepted" }); + expect( + store.probe( + baseline, + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toEqual({ loadingEpoch: null, status: "miss" }); + expect( + store.probe( + request("u", firstEpoch), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toEqual({ response, status: "ready" }); + expect( + store.probe( + request("u", firstEpoch, { + continuationToken: "next", + }), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "miss" }); + expect( + store.probe( + request("u", firstEpoch), + DUCKDB_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "miss" }); + }); + + it("retains partial and complete-empty results without upgrading coverage", () => { + const store = createSqlCatalogSearchPolicyStore(); + const current = epoch(2); + const partial = ready(current, { kind: "partial" }, [ + relation("partial"), + ]); + const empty = ready(current); + + store.record( + request("p", current), + POSTGRESQL_SQL_RELATION_DIALECT, + partial, + ); + store.record( + request("e", current), + POSTGRESQL_SQL_RELATION_DIALECT, + empty, + ); + + expect( + store.probe( + request("p", current), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toEqual({ response: partial, status: "ready" }); + expect( + store.probe( + request("e", current), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toEqual({ response: empty, status: "ready" }); + }); + + it("uses every structural key field and replaces an exact cached value", () => { + const store = createSqlCatalogSearchPolicyStore(); + const current = epoch(2); + const key = request("key", current); + const replacement = ready(current, { kind: "partial" }); + store.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + ready(current), + ); + store.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + replacement, + ); + + const variants = [ + request("key", current, { continuationToken: "page" }), + request("key", current, { dialectId: "other" }), + request("key", current, { limit: 19 }), + request("key", current, { qualifier: "other" }), + request("key", current, { quoted: true }), + request("key", current, { searchPath: "other" }), + request("key", current, { scope: "other" }), + request("other", current), + Object.freeze({ + ...key, + qualifier: Object.freeze([]), + }), + ]; + for (const variant of variants) { + expect( + store.probe( + variant, + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "miss" }); + } + expect( + store.probe( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toEqual({ response: replacement, status: "ready" }); + const continued = request("continued", current, { + continuationToken: "page-2", + }); + store.record( + continued, + POSTGRESQL_SQL_RELATION_DIALECT, + ready(current), + ); + expect( + store.probe( + continued, + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "ready" }); + expect(store.metrics().entries).toBe(2); + }); + + it("permits loading probes but rejects same-epoch ready publication", () => { + const store = createSqlCatalogSearchPolicyStore(); + const current = epoch(3); + const key = request("load", current); + + expect( + store.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + loading(current), + ), + ).toMatchObject({ + retained: "loading-barrier", + status: "accepted", + }); + expect( + store.probe(key, POSTGRESQL_SQL_RELATION_DIALECT), + ).toEqual({ + loadingEpoch: current, + status: "miss", + }); + expect( + store.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + loading(current), + ), + ).toMatchObject({ retained: "loading-barrier" }); + expect( + store.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + ready(current), + ), + ).toEqual({ + reason: "loading-transition", + status: "conflict", + }); + expect( + store.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + ready(epoch(4)), + ), + ).toEqual({ + reason: "epoch-mismatch", + status: "conflict", + }); + expect(store.metrics()).toMatchObject({ + loadingBarriers: 1, + }); + expect( + store.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + failed(current, "next-request"), + ), + ).toMatchObject({ retained: "none" }); + expect( + store.probe(key, POSTGRESQL_SQL_RELATION_DIALECT), + ).toEqual({ loadingEpoch: current, status: "miss" }); + }); + + it("implements all failure retry policies", () => { + const current = epoch(4); + const key = request("retry", current); + + const nextStore = createSqlCatalogSearchPolicyStore(); + expect( + nextStore.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + failed(current, "next-request"), + ), + ).toMatchObject({ retained: "none" }); + expect( + nextStore.probe( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "miss" }); + + const invalidationStore = + createSqlCatalogSearchPolicyStore(); + invalidationStore.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + failed(current, "after-invalidation"), + ); + expect( + invalidationStore.probe( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ + retry: "after-invalidation", + status: "failed", + }); + expect( + invalidationStore.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + ready(current), + ), + ).toEqual({ + reason: "retry-gated", + status: "conflict", + }); + expect( + invalidationStore.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + loading(current), + ), + ).toEqual({ + reason: "retry-gated", + status: "conflict", + }); + invalidationStore.advanceScope("scope-a", epoch(5)); + expect( + invalidationStore.probe( + request("retry", epoch(5)), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "miss" }); + + const neverStore = createSqlCatalogSearchPolicyStore(); + neverStore.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + failed(current, "never"), + ); + neverStore.advanceScope("scope-a", epoch(5)); + expect( + neverStore.probe( + request("retry", epoch(5)), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toEqual({ + code: "unavailable", + epoch: epoch(5), + retry: "never", + status: "failed", + }); + expect( + neverStore.probe( + request("retry", null), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toEqual({ loadingEpoch: null, status: "miss" }); + expect( + neverStore.record( + request("retry", epoch(5)), + POSTGRESQL_SQL_RELATION_DIALECT, + loading(epoch(5)), + ), + ).toEqual({ + reason: "retry-gated", + status: "conflict", + }); + }); + + it("invalidates only older epoch-scoped state in the selected scope", () => { + const store = createSqlCatalogSearchPolicyStore(); + const oldEpoch = epoch(7); + const newEpoch = epoch(8); + const keepResponse = ready(newEpoch); + + store.record( + request("old", oldEpoch), + POSTGRESQL_SQL_RELATION_DIALECT, + ready(oldEpoch), + ); + store.record( + request("keep", newEpoch), + POSTGRESQL_SQL_RELATION_DIALECT, + keepResponse, + ); + store.record( + request("other", oldEpoch, { scope: "scope-b" }), + POSTGRESQL_SQL_RELATION_DIALECT, + ready(oldEpoch), + ); + store.advanceScope("scope-a", newEpoch); + store.advanceScope("scope-missing", newEpoch); + + expect( + store.probe( + request("old", newEpoch), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "miss" }); + expect( + store.probe( + request("keep", newEpoch), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toEqual({ response: keepResponse, status: "ready" }); + expect(store.metrics()).toMatchObject({ + entries: 2, + readyEntries: 2, + }); + }); + + it("evicts deterministically by recency at the shared entry ceiling", () => { + const store = createSqlCatalogSearchPolicyStore(); + const current = epoch(9); + for ( + let index = 0; + index < MAX_CATALOG_POLICY_STORE_ENTRIES; + index += 1 + ) { + store.record( + request(`key-${index}`, current), + POSTGRESQL_SQL_RELATION_DIALECT, + ready(current), + ); + } + store.probe( + request("key-0", current), + POSTGRESQL_SQL_RELATION_DIALECT, + ); + store.record( + request("newest", current), + POSTGRESQL_SQL_RELATION_DIALECT, + failed(current, "after-invalidation"), + ); + + expect(store.metrics()).toMatchObject({ + entries: MAX_CATALOG_POLICY_STORE_ENTRIES, + failureGates: 1, + }); + expect( + store.probe( + request("key-0", current), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "ready" }); + expect( + store.probe( + request("key-1", current), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "miss" }); + }); + + it("never evicts correctness gates when bounded capacity is exhausted", () => { + const store = createSqlCatalogSearchPolicyStore(); + const current = epoch(9); + for ( + let index = 0; + index < MAX_CATALOG_POLICY_STORE_ENTRIES; + index += 1 + ) { + expect( + store.record( + request(`gate-${index}`, current), + POSTGRESQL_SQL_RELATION_DIALECT, + loading(current), + ), + ).toMatchObject({ + retained: "loading-barrier", + status: "accepted", + }); + } + + expect( + store.record( + request("overflow", current), + POSTGRESQL_SQL_RELATION_DIALECT, + failed(current, "after-invalidation"), + ), + ).toEqual({ + reason: "capacity", + status: "conflict", + }); + expect( + store.record( + request("loading-overflow", current), + POSTGRESQL_SQL_RELATION_DIALECT, + loading(current), + ), + ).toEqual({ + reason: "capacity", + status: "conflict", + }); + expect( + store.probe( + request("gate-0", current), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toEqual({ status: "overloaded" }); + expect(store.metrics()).toMatchObject({ + entries: MAX_CATALOG_POLICY_STORE_ENTRIES, + loadingBarriers: MAX_CATALOG_POLICY_STORE_ENTRIES, + }); + }); + + it("enforces the retained-byte ceiling without truncating responses", () => { + const store = createSqlCatalogSearchPolicyStore(); + const current = epoch(10); + const relations = Array.from( + { length: 50 }, + (_, index) => + relation(`large-${index}`, "x".repeat(800)), + ); + for (let index = 0; index < 80; index += 1) { + store.record( + request(`large-${index}`, current), + POSTGRESQL_SQL_RELATION_DIALECT, + ready(current, { kind: "partial" }, relations), + ); + } + + expect(store.metrics().retainedBytes).toBeLessThanOrEqual( + MAX_CATALOG_POLICY_STORE_RETAINED_BYTES, + ); + expect(store.metrics().entries).toBeLessThan(80); + expect( + store.probe( + request("large-0", current), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "miss" }); + expect( + store.probe( + request("large-79", current), + POSTGRESQL_SQL_RELATION_DIALECT, + ), + ).toMatchObject({ status: "ready" }); + }); + + it("does not retain one response larger than the byte ceiling", () => { + const store = createSqlCatalogSearchPolicyStore(); + const current = epoch(10); + const oversized = ready(current, { kind: "partial" }, [ + relation( + "oversized", + "x".repeat( + MAX_CATALOG_POLICY_STORE_RETAINED_BYTES, + ), + ), + ]); + + expect( + store.record( + request("oversized", current), + POSTGRESQL_SQL_RELATION_DIALECT, + oversized, + ), + ).toMatchObject({ + retained: "none", + response: oversized, + status: "accepted", + }); + expect(store.metrics().entries).toBe(0); + }); + + it("rejects an individually oversized correctness gate", () => { + const store = createSqlCatalogSearchPolicyStore(); + const current = epoch(10); + const oversizedKey = request( + "x".repeat( + MAX_CATALOG_POLICY_STORE_RETAINED_BYTES, + ), + current, + ); + + expect( + store.record( + oversizedKey, + POSTGRESQL_SQL_RELATION_DIALECT, + failed(current, "after-invalidation"), + ), + ).toEqual({ + reason: "capacity", + status: "conflict", + }); + expect(store.metrics()).toMatchObject({ + entries: 0, + failureGates: 0, + }); + }); + + it("disposes retained state and leaves the frozen handle inert", () => { + const store = createSqlCatalogSearchPolicyStore(); + const current = epoch(11); + const key = request("disposed", current); + store.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + ready(current), + ); + + expect(Object.isFrozen(store)).toBe(true); + store.dispose(); + store.dispose(); + store.advanceScope("scope-a", epoch(12)); + + expect(store.metrics()).toEqual({ + entries: 0, + failureGates: 0, + loadingBarriers: 0, + readyEntries: 0, + retainedBytes: 0, + }); + expect( + store.probe(key, POSTGRESQL_SQL_RELATION_DIALECT), + ).toEqual({ status: "disposed" }); + expect( + store.record( + key, + POSTGRESQL_SQL_RELATION_DIALECT, + ready(current), + ), + ).toEqual({ status: "disposed" }); + }); +}); diff --git a/src/vnext/__tests__/relation-catalog-search-work.test.ts b/src/vnext/__tests__/relation-catalog-search-work.test.ts index a16aa0d..a6914c7 100644 --- a/src/vnext/__tests__/relation-catalog-search-work.test.ts +++ b/src/vnext/__tests__/relation-catalog-search-work.test.ts @@ -16,6 +16,7 @@ import { type SqlCatalogSearchWorkOutcome, type SqlCatalogSearchWorkTicket, } from "../relation-catalog-search-work.js"; +import { MAX_CATALOG_POLICY_STORE_ENTRIES } from "../relation-catalog-search-policy-store.js"; import { DUCKDB_SQL_RELATION_DIALECT, POSTGRESQL_SQL_RELATION_DIALECT, @@ -281,6 +282,8 @@ describe("catalog search coordinator construction", () => { { executionDeadlineMs: 5_001 }, { queueDeadlineMs: 9 }, { queueDeadlineMs: 2_001 }, + { refreshLeaseMs: 0 }, + { refreshLeaseMs: 5_001 }, { deadlineScheduler: { clearTimeout() {}, @@ -710,6 +713,567 @@ describe("catalog search ownership and active-slot lifecycle", () => { expect(provider.calls).toHaveLength(2); expect(scheduler.pendingCount).toBe(0); }); + + it("transfers pending ownership to a bounded refresh observer and dispatches ready availability once", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + refreshLeaseMs: 50, + }); + const ticket = owner(service).request(input("observed")); + let preparations = 0; + let dispatches = 0; + + expect( + ticket.retainForRefresh(() => { + preparations += 1; + expect(provider.calls[0]?.signal.aborted).toBe(false); + return (): undefined => { + dispatches += 1; + return undefined; + }; + }), + ).toEqual({ + remainingLeaseMs: 50, + status: "retained", + }); + await expect(ticket.result).resolves.toEqual({ + status: "cancelled", + }); + expect( + ticket.retainForRefresh(() => null), + ).toEqual({ + reason: "not-retainable", + status: "unavailable", + }); + + provider.calls[0]?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + expect(preparations).toBe(1); + expect(dispatches).toBe(1); + scheduler.advanceBy(50); + expect(dispatches).toBe(1); + }); + + it("rejects invalid and already-expired refresh retention without transferring ownership", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 20, + }); + const ticket = owner(service).request(input("invalid-retention")); + + expect( + Reflect.apply(ticket.retainForRefresh, undefined, [null]), + ).toEqual({ + reason: "invalid-target", + status: "unavailable", + }); + scheduler.nowValue = 20; + expect(ticket.retainForRefresh(() => null)).toEqual({ + reason: "expired", + status: "unavailable", + }); + + ticket.cancel(); + await expect(ticket.result).resolves.toEqual({ + status: "cancelled", + }); + }); + + it("fails refresh retention closed when its clock disposes or supersedes the owner", async () => { + for (const reentry of ["dispose", "replace"] as const) { + let trigger = false; + let service: SqlCatalogSearchWorkCoordinator | undefined; + let session: SqlCatalogSearchWorkOwner | undefined; + const replacements: SqlCatalogSearchWorkTicket[] = []; + const provider = providerHarness(); + service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout() {}, + now() { + if (trigger) { + trigger = false; + if (reentry === "dispose") { + service?.dispose(); + } else { + const replacement = session?.request( + input("replacement"), + ); + if (replacement) replacements.push(replacement); + } + } + return 0; + }, + setTimeout() { + return 1; + }, + }, + }); + session = owner(service); + const ticket = session.request(input("observed")); + trigger = true; + + expect(ticket.retainForRefresh(() => null)).toEqual({ + reason: + reentry === "dispose" ? "disposed" : "superseded", + status: "unavailable", + }); + await expect(ticket.result).resolves.toEqual( + reentry === "dispose" + ? { reason: "disposed", status: "unavailable" } + : { status: "superseded" }, + ); + + service.dispose(); + for (const replacement of replacements) { + await expect(replacement.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + } + } + }); + + it("returns expired when a hostile scheduler fires the refresh lease synchronously", async () => { + let nextHandle = 0; + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: { + clearTimeout() {}, + now: () => 0, + setTimeout(callback, delayMs) { + nextHandle += 1; + if (delayMs === 50) callback(); + return nextHandle; + }, + }, + refreshLeaseMs: 50, + }); + const ticket = owner(service).request(input("sync-expiry")); + + expect(ticket.retainForRefresh(() => null)).toEqual({ + reason: "expired", + status: "unavailable", + }); + await expect(ticket.result).resolves.toEqual({ + status: "cancelled", + }); + expect(provider.calls[0]?.signal.aborted).toBe(true); + }); + + it("contains hostile availability preparation and dispatch results", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const prepared: string[] = []; + const dispatched: string[] = []; + const targets = [ + () => { + prepared.push("throw"); + throw new Error("preparation failed"); + }, + () => { + prepared.push("non-function"); + return Promise.reject(new Error("detached preparation")); + }, + () => { + prepared.push("null"); + return null; + }, + () => { + prepared.push("throwing-dispatch"); + return () => { + dispatched.push("throw"); + throw new Error("dispatch failed"); + }; + }, + () => { + prepared.push("async-dispatch"); + return () => { + dispatched.push("async"); + return Promise.reject(new Error("detached dispatch")); + }; + }, + ] as const; + const tickets = targets.map(() => + owner(service).request(input("hostile-availability")), + ); + for (const [index, ticket] of tickets.entries()) { + expect( + Reflect.apply(ticket.retainForRefresh, undefined, [ + targets[index] ?? (() => null), + ]), + ).toMatchObject({ status: "retained" }); + } + + provider.calls[0]?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + expect(prepared).toEqual([ + "throw", + "non-function", + "null", + "throwing-dispatch", + "async-dispatch", + ]); + expect(dispatched).toEqual(["throw", "async"]); + await Promise.all( + tickets.map((ticket) => + expect(ticket?.result).resolves.toEqual({ + status: "cancelled", + }), + ), + ); + }); + + it("stops availability preparation and dispatch when preparation disposes the coordinator", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + let secondPreparation = 0; + let dispatches = 0; + const disposing = owner(service).request( + input("disposing-availability"), + ); + const skipped = owner(service).request( + input("disposing-availability"), + ); + + expect( + disposing.retainForRefresh(() => { + service.dispose(); + return () => { + dispatches += 1; + }; + }).status, + ).toBe("retained"); + expect( + skipped.retainForRefresh(() => { + secondPreparation += 1; + return null; + }).status, + ).toBe("retained"); + + provider.calls[0]?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + expect(secondPreparation).toBe(0); + expect(dispatches).toBe(0); + }); + + it.each(["owner-dispose", "new-request", "higher-epoch"] as const)( + "suppresses prepared availability after reentrant %s", + async (reentry) => { + 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); + const session = owner(service); + const ticket = session.request(input("availability-race")); + let dispatches = 0; + expect( + ticket.retainForRefresh(() => { + if (reentry === "owner-dispose") { + session.dispose(); + } else if (reentry === "new-request") { + session.request(input("newer")); + } else { + listener.current?.({ epoch: epoch(2) }); + } + return () => { + dispatches += 1; + }; + }).status, + ).toBe("retained"); + + calls[0]?.settlement.resolve(readyResponse(1)); + await flushMicrotasks(); + expect(dispatches).toBe(0); + service.dispose(); + }, + ); + + it("revalidates each prepared dispatch when a later preparation retires one owner", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const firstOwner = owner(service); + const secondOwner = owner(service); + const first = firstOwner.request(input("shared-availability")); + const second = secondOwner.request(input("shared-availability")); + let firstDispatches = 0; + let secondDispatches = 0; + expect( + first.retainForRefresh(() => () => { + firstDispatches += 1; + }).status, + ).toBe("retained"); + expect( + second.retainForRefresh(() => { + firstOwner.dispose(); + return () => { + secondDispatches += 1; + }; + }).status, + ).toBe("retained"); + + provider.calls[0]?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + expect(firstDispatches).toBe(0); + expect(secondDispatches).toBe(1); + service.dispose(); + }); + + it("starts observer-only queued work when an active slot becomes free", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const active = Array.from( + { length: MAX_CATALOG_ACTIVE_SEARCH_WORK }, + (_, index) => + owner(service).request(input(`active-${index}`)), + ); + const queued = owner(service).request(input("queued-observer")); + let availability = 0; + + expect( + queued.retainForRefresh(() => { + availability += 1; + return null; + }), + ).toMatchObject({ + status: "retained", + }); + await expect(queued.result).resolves.toEqual({ + status: "cancelled", + }); + expect(provider.calls).toHaveLength( + MAX_CATALOG_ACTIVE_SEARCH_WORK, + ); + + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(active[0]?.result).resolves.toMatchObject({ + status: "usable", + }); + await flushMicrotasks(); + expect(provider.calls).toHaveLength( + MAX_CATALOG_ACTIVE_SEARCH_WORK + 1, + ); + expect(provider.calls.at(-1)?.request.prefix.value).toBe( + "queued-observer", + ); + provider.calls.at(-1)?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + expect(availability).toBe(1); + + service.dispose(); + await Promise.all( + active.slice(1).map((ticket) => + expect(ticket.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }), + ), + ); + }); + + it("retires a refresh observer when its owner starts different work or is disposed", async () => { + for (const action of ["request", "dispose"] as const) { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const observed = session.request(input(`observed-${action}`)); + expect( + observed.retainForRefresh(() => null).status, + ).toBe("retained"); + const observedSignal = provider.calls[0]?.signal; + + if (action === "request") { + const replacement = session.request( + input("different-work"), + ); + expect(observedSignal?.aborted).toBe(true); + expect(provider.calls).toHaveLength(2); + service.dispose(); + await expect(replacement.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + } else { + session.dispose(); + expect(observedSignal?.aborted).toBe(true); + } + await expect(observed.result).resolves.toEqual({ + status: "cancelled", + }); + service.dispose(); + } + }); + + it("hands an observed same-key operation back to a consumer without aborting or restarting it", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const observed = session.request(input("same-observed")); + let availability = 0; + expect( + observed.retainForRefresh(() => { + availability += 1; + return null; + }).status, + ).toBe("retained"); + const signal = provider.calls[0]?.signal; + + const replacement = session.request(input("same-observed")); + expect(provider.calls).toHaveLength(1); + expect(signal?.aborted).toBe(false); + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(replacement.result).resolves.toMatchObject({ + status: "usable", + }); + expect(availability).toBe(0); + }); + + it("counts consumers and observers together before issuing the last-owner abort", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const first = owner(service).request(input("shared-observed")); + const second = owner(service).request(input("shared-observed")); + const signal = provider.calls[0]?.signal; + expect( + first.retainForRefresh(() => null).status, + ).toBe("retained"); + + second.cancel(); + await expect(second.result).resolves.toEqual({ + status: "cancelled", + }); + expect(signal?.aborted).toBe(false); + + first.cancel(); + expect(signal?.aborted).toBe(true); + provider.calls[0]?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + }); + + it("clips observer retention to the hard work deadline and never notifies on expiry", async () => { + const scheduler = new ManualDeadlineScheduler(); + const provider = providerHarness(); + const service = coordinator(provider.captured, { + deadlineScheduler: scheduler, + executionDeadlineMs: 25, + refreshLeaseMs: 50, + }); + const ticket = owner(service).request(input("leased")); + let availability = 0; + expect( + ticket.retainForRefresh(() => { + availability += 1; + return null; + }), + ).toEqual({ + remainingLeaseMs: 25, + status: "retained", + }); + + scheduler.advanceBy(25); + expect(provider.calls[0]?.signal.aborted).toBe(true); + expect(availability).toBe(0); + provider.calls[0]?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + expect(availability).toBe(0); + }); + + it.each(["loading", "failed"] as const)( + "removes observers without availability for a terminal %s response", + async (status) => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const ticket = owner(service).request(input(status)); + let availability = 0; + expect( + ticket.retainForRefresh(() => { + availability += 1; + return null; + }).status, + ).toBe("retained"); + provider.calls[0]?.settlement.resolve( + status === "loading" + ? { + epoch: epoch(), + status, + } + : { + code: "unavailable", + epoch: epoch(), + retry: "next-request", + status, + }, + ); + await flushMicrotasks(); + expect(availability).toBe(0); + expect(provider.calls[0]?.signal.aborted).toBe(false); + }, + ); + + it("clears observers without availability on higher epochs and disposal", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const baselineOwner = owner(service); + const baseline = baselineOwner.request(input("baseline")); + provider.calls[0]?.settlement.resolve(readyResponse(1)); + await expect(baseline.result).resolves.toMatchObject({ + status: "usable", + }); + + const higher = baselineOwner.request(input("higher")); + let higherAvailability = 0; + expect( + higher.retainForRefresh(() => { + higherAvailability += 1; + return null; + }).status, + ).toBe("retained"); + provider.calls[1]?.settlement.resolve(readyResponse(2)); + await flushMicrotasks(); + expect(higherAvailability).toBe(0); + + const disposable = owner(service, "other-scope").request( + input("disposed-observer"), + ); + let disposedAvailability = 0; + expect( + disposable.retainForRefresh(() => { + disposedAvailability += 1; + return null; + }).status, + ).toBe("retained"); + const disposableSignal = provider.calls[2]?.signal; + service.dispose(); + expect(disposableSignal?.aborted).toBe(true); + expect(disposedAvailability).toBe(0); + provider.calls[2]?.settlement.resolve(readyResponse()); + await flushMicrotasks(); + expect(disposedAvailability).toBe(0); + }); }); describe("catalog search absolute deadlines", () => { @@ -1485,6 +2049,373 @@ describe("catalog search epoch authority and isolation", () => { }); }); +describe("catalog search policy integration", () => { + it("fails closed when loading barriers exhaust policy capacity", async () => { + const captured = accepted( + captureSqlRelationCatalogProvider({ + id: "capacity-catalog", + search(request: SqlCatalogSearchRequest) { + return { + epoch: request.expectedEpoch ?? epoch(1), + status: "loading", + }; + }, + }), + ); + const service = coordinator(captured); + const session = owner(service); + + for ( + let index = 0; + index < MAX_CATALOG_POLICY_STORE_ENTRIES; + index += 1 + ) { + await expect( + session.request(input(`gate-${index}`)).result, + ).resolves.toMatchObject({ + response: { status: "loading" }, + status: "usable", + }); + } + await expect( + session.request(input("gate-overflow")).result, + ).resolves.toEqual({ + reason: "overloaded", + status: "unavailable", + }); + service.dispose(); + }); + + it("serves a repeated ready search from the exact-epoch cache", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const first = session.request(input("cached")); + + provider.calls[0]?.settlement.resolve(readyResponse(3)); + const firstOutcome = await first.result; + expect(firstOutcome).toMatchObject({ + observation: "baseline", + status: "usable", + }); + + const cached = session.request(input("cached")); + expect(provider.calls).toHaveLength(1); + await expect(cached.result).resolves.toEqual({ + observation: "equal", + response: + firstOutcome.status === "usable" + ? firstOutcome.response + : undefined, + status: "usable", + }); + + const distinct = session.request(input("cached", { limit: 19 })); + expect(provider.calls).toHaveLength(2); + service.dispose(); + await expect(distinct.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + }); + + it("supersedes a cache hit when retiring old work reenters with a newer request", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const baseline = session.request(input("cached-reentrant")); + provider.calls[0]?.settlement.resolve(readyResponse()); + await expect(baseline.result).resolves.toMatchObject({ + status: "usable", + }); + + const old = session.request(input("old-work")); + const newest: { + current: SqlCatalogSearchWorkTicket | null; + } = { current: null }; + provider.calls[1]?.signal.addEventListener("abort", () => { + newest.current = session.request(input("newest-work")); + }); + const displacedCacheHit = session.request( + input("cached-reentrant"), + ); + + await expect(old.result).resolves.toEqual({ + status: "superseded", + }); + await expect(displacedCacheHit.result).resolves.toEqual({ + status: "superseded", + }); + expect(provider.calls).toHaveLength(3); + provider.calls[2]?.settlement.resolve(readyResponse()); + if (!newest.current) { + throw new Error("Expected reentrant request"); + } + await expect(newest.current.result).resolves.toMatchObject({ + status: "usable", + }); + }); + + it("keeps immediate tickets non-retainable and fails a reentrant cache hit closed on disposal", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const session = owner(service); + const baseline = session.request(input("cached-disposal")); + provider.calls[0]?.settlement.resolve(readyResponse()); + await baseline.result; + + const observed = session.request(input("observed-before-cache")); + expect( + observed.retainForRefresh(() => null).status, + ).toBe("retained"); + const cached = session.request(input("cached-disposal")); + expect(cached.retainForRefresh(() => null)).toEqual({ + reason: "not-retainable", + status: "unavailable", + }); + await expect(cached.result).resolves.toMatchObject({ + status: "usable", + }); + + const active = session.request(input("dispose-on-abort")); + provider.calls.at(-1)?.signal.addEventListener("abort", () => { + service.dispose(); + }); + const disposedHit = session.request(input("cached-disposal")); + await expect(active.result).resolves.toEqual({ + status: "superseded", + }); + await expect(disposedHit.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + expect(disposedHit.retainForRefresh(() => null)).toEqual({ + reason: "disposed", + status: "unavailable", + }); + }); + + it("gates loading until a higher epoch is observed and then caches readiness", 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(1)); + await expect(baseline.result).resolves.toMatchObject({ + observation: "baseline", + status: "usable", + }); + + const loading = session.request(input("loading")); + provider.calls[1]?.settlement.resolve({ + epoch: epoch(1), + status: "loading", + }); + await expect(loading.result).resolves.toMatchObject({ + response: { status: "loading" }, + status: "usable", + }); + + const sameEpochReady = session.request(input("loading")); + provider.calls[2]?.settlement.resolve(readyResponse(1)); + await expect(sameEpochReady.result).resolves.toEqual({ + status: "superseded", + }); + + const advancing = session.request(input("loading")); + provider.calls[3]?.settlement.resolve(readyResponse(2)); + await expect(advancing.result).resolves.toEqual({ + status: "superseded", + }); + + const current = session.request(input("loading")); + provider.calls[4]?.settlement.resolve(readyResponse(2)); + const currentOutcome = await current.result; + expect(currentOutcome).toMatchObject({ + observation: "equal", + status: "usable", + }); + + const cached = session.request(input("loading")); + expect(provider.calls).toHaveLength(5); + await expect(cached.result).resolves.toEqual({ + observation: "equal", + response: + currentOutcome.status === "usable" + ? currentOutcome.response + : undefined, + status: "usable", + }); + }); + + it("reuses loading without a provider call while a scope subscription is live", async () => { + const listeners = new Map< + string, + (event: unknown) => void + >(); + 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, + ) { + listeners.set(scope, onInvalidation); + return () => undefined; + }, + }), + ); + const service = coordinator(captured); + const session = owner(service); + const first = session.request(input("subscribed-loading")); + calls[0]?.settlement.resolve({ + epoch: epoch(1), + status: "loading", + }); + const firstOutcome = await first.result; + expect(firstOutcome).toMatchObject({ + observation: "baseline", + response: { status: "loading" }, + status: "usable", + }); + + const retained = session.request( + input("subscribed-loading"), + ); + expect(calls).toHaveLength(1); + await expect(retained.result).resolves.toEqual({ + observation: "equal", + response: + firstOutcome.status === "usable" + ? firstOutcome.response + : undefined, + status: "usable", + }); + + listeners + .get("connection:primary") + ?.({ epoch: epoch(2) }); + const retried = session.request( + input("subscribed-loading"), + ); + expect(calls).toHaveLength(2); + service.dispose(); + await expect(retried.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + }); + + it("reuses after-invalidation failures only until the scope advances", 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); + const session = owner(service); + const failing = session.request(input("failure")); + calls[0]?.settlement.resolve({ + code: "unavailable", + epoch: epoch(1), + retry: "after-invalidation", + status: "failed", + }); + const failedOutcome = await failing.result; + expect(failedOutcome).toMatchObject({ + observation: "baseline", + response: { status: "failed" }, + status: "usable", + }); + + const gated = session.request(input("failure")); + expect(calls).toHaveLength(1); + await expect(gated.result).resolves.toEqual({ + observation: "equal", + response: + failedOutcome.status === "usable" + ? failedOutcome.response + : undefined, + status: "usable", + }); + + listener.current?.({ epoch: epoch(2) }); + const retried = session.request(input("failure")); + expect(calls).toHaveLength(2); + service.dispose(); + await expect(retried.result).resolves.toEqual({ + reason: "disposed", + status: "unavailable", + }); + }); + + it("re-establishes epoch authority before reusing a never gate after remount", async () => { + const provider = providerHarness(); + const service = coordinator(provider.captured); + const firstOwner = owner(service); + const failing = firstOwner.request(input("never-remount")); + provider.calls[0]?.settlement.resolve({ + code: "invalid-configuration", + epoch: epoch(1), + retry: "never", + status: "failed", + }); + await expect(failing.result).resolves.toMatchObject({ + observation: "baseline", + status: "usable", + }); + firstOwner.dispose(); + + const remounted = owner(service); + const baseline = remounted.request(input("never-remount")); + expect(provider.calls).toHaveLength(2); + expect(provider.calls[1]?.request.expectedEpoch).toBeNull(); + provider.calls[1]?.settlement.resolve(readyResponse(1)); + await expect(baseline.result).resolves.toEqual({ + status: "superseded", + }); + + const gated = remounted.request(input("never-remount")); + expect(provider.calls).toHaveLength(2); + await expect(gated.result).resolves.toMatchObject({ + observation: "equal", + response: { + code: "invalid-configuration", + status: "failed", + }, + status: "usable", + }); + }); +}); + describe("catalog search disposal and late settlement", () => { it("settles active and queued owners before abort and drains late rejection", async () => { const scheduler = new ManualDeadlineScheduler(); diff --git a/src/vnext/relation-catalog-epoch-coordinator.ts b/src/vnext/relation-catalog-epoch-coordinator.ts index a82018e..b567d3c 100644 --- a/src/vnext/relation-catalog-epoch-coordinator.ts +++ b/src/vnext/relation-catalog-epoch-coordinator.ts @@ -117,6 +117,10 @@ export type SqlCatalogResponseEpochSubmissionResult = }; export interface SqlCatalogEpochCoordinator { + readonly hasLiveSubscription: ( + this: void, + scope: string, + ) => boolean; readonly providerId: string; readonly prepareScopeMembership: ( this: void, @@ -1274,6 +1278,17 @@ function createCoordinatorHandle( dispose: (): void => { disposeCoordinator(state); }, + hasLiveSubscription: (scope: string): boolean => { + const subscription = + state.scopes.get(scope)?.subscription; + return Boolean( + !state.disposed && + subscription && + !subscription.failed && + !subscription.installing && + subscription.cell.active, + ); + }, prepareScopeMembership: ( scope: unknown, target: SqlCatalogRevisionTarget, diff --git a/src/vnext/relation-catalog-search-policy-store.ts b/src/vnext/relation-catalog-search-policy-store.ts new file mode 100644 index 0000000..2f155d7 --- /dev/null +++ b/src/vnext/relation-catalog-search-policy-store.ts @@ -0,0 +1,762 @@ +import type { + SqlValidatedCatalogSearchResponse, +} from "./relation-catalog-boundary.js"; +import type { SqlRelationDialectRuntime } from "./relation-dialect.js"; +import type { + SqlCatalogEpoch, + SqlCatalogFailureCode, + SqlCatalogRetryPolicy, + SqlCatalogSearchRequest, +} from "./relation-completion-types.js"; +import type { + SqlIdentifierComponent, + SqlIdentifierPath, +} from "./types.js"; + +export const MAX_CATALOG_POLICY_STORE_ENTRIES = 256; +export const MAX_CATALOG_POLICY_STORE_RETAINED_BYTES = + 2 * 1_024 * 1_024; + +export type SqlCatalogSearchPolicyProbe = + | { + readonly status: "ready"; + readonly response: Extract< + SqlValidatedCatalogSearchResponse, + { readonly status: "ready" } + >; + } + | { + readonly status: "failed"; + readonly code: SqlCatalogFailureCode; + readonly epoch: SqlCatalogEpoch; + readonly retry: Exclude< + SqlCatalogRetryPolicy, + "next-request" + >; + } + | { + readonly status: "miss"; + readonly loadingEpoch: SqlCatalogEpoch | null; + } + | { + readonly status: "disposed"; + } + | { + readonly status: "overloaded"; + }; + +export type SqlCatalogSearchPolicyRecordResult = + | { + readonly status: "accepted"; + readonly response: SqlValidatedCatalogSearchResponse; + readonly retained: + | "failure-gate" + | "loading-barrier" + | "none" + | "ready"; + } + | { + readonly status: "conflict"; + readonly reason: + | "capacity" + | "epoch-mismatch" + | "loading-transition" + | "retry-gated"; + } + | { + readonly status: "disposed"; + }; + +export interface SqlCatalogSearchPolicyStoreMetrics { + readonly entries: number; + readonly failureGates: number; + readonly loadingBarriers: number; + readonly readyEntries: number; + readonly retainedBytes: number; +} + +export interface SqlCatalogSearchPolicyStore { + readonly advanceScope: ( + this: void, + scope: string, + epoch: SqlCatalogEpoch, + ) => void; + readonly dispose: (this: void) => void; + readonly metrics: ( + this: void, + ) => SqlCatalogSearchPolicyStoreMetrics; + readonly probe: ( + this: void, + request: SqlCatalogSearchRequest, + dialect: SqlRelationDialectRuntime, + ) => SqlCatalogSearchPolicyProbe; + readonly record: ( + this: void, + request: SqlCatalogSearchRequest, + dialect: SqlRelationDialectRuntime, + response: SqlValidatedCatalogSearchResponse, + ) => SqlCatalogSearchPolicyRecordResult; +} + +interface StructuralKey { + readonly continuationToken: string | null; + readonly dialect: SqlRelationDialectRuntime; + readonly dialectId: string; + readonly limit: number; + readonly prefix: SqlIdentifierComponent; + readonly qualifier: SqlIdentifierPath; + readonly scope: string; + readonly searchPaths: readonly SqlIdentifierPath[]; +} + +interface RetainedRecordBase { + readonly bytes: number; + readonly key: StructuralKey; + lastUse: bigint; +} + +interface ReadyRecord extends RetainedRecordBase { + readonly kind: "ready"; + readonly response: Extract< + SqlValidatedCatalogSearchResponse, + { readonly status: "ready" } + >; +} + +interface LoadingRecord extends RetainedRecordBase { + readonly epoch: SqlCatalogEpoch; + readonly kind: "loading"; +} + +interface FailureRecord extends RetainedRecordBase { + readonly code: SqlCatalogFailureCode; + readonly epoch: SqlCatalogEpoch; + readonly kind: "failure"; + readonly retry: Exclude< + SqlCatalogRetryPolicy, + "next-request" + >; +} + +type RetainedRecord = + | FailureRecord + | LoadingRecord + | ReadyRecord; + +interface StoreState { + disposed: boolean; + readonly records: RetainedRecord[]; + retainedBytes: number; + saturated: boolean; + sequence: bigint; +} + +const DISPOSED_PROBE: SqlCatalogSearchPolicyProbe = + Object.freeze({ status: "disposed" }); +const DISPOSED_RECORD: SqlCatalogSearchPolicyRecordResult = + Object.freeze({ status: "disposed" }); +const KEY_FIXED_BYTES = 160; +const PATH_FIXED_BYTES = 24; +const COMPONENT_FIXED_BYTES = 24; +const RESPONSE_FIXED_BYTES = 96; +const RELATION_FIXED_BYTES = 128; + +function sameEpoch( + left: SqlCatalogEpoch, + right: SqlCatalogEpoch, +): boolean { + return ( + left.generation === right.generation && + left.token === right.token + ); +} + +function sameComponent( + left: SqlIdentifierComponent, + right: SqlIdentifierComponent, +): boolean { + return ( + left.quoted === right.quoted && + left.value === right.value + ); +} + +function samePath( + left: SqlIdentifierPath, + right: SqlIdentifierPath, +): boolean { + if (left.length !== right.length) return false; + for (let index = 0; index < left.length; index += 1) { + const leftComponent = left[index]; + const rightComponent = right[index]; + if ( + !leftComponent || + !rightComponent || + !sameComponent(leftComponent, rightComponent) + ) { + return false; + } + } + return true; +} + +function sameBaseKey( + left: StructuralKey, + right: StructuralKey, +): boolean { + if ( + left.dialect !== right.dialect || + left.dialectId !== right.dialectId || + left.scope !== right.scope || + left.limit !== right.limit || + left.continuationToken !== right.continuationToken || + !sameComponent(left.prefix, right.prefix) || + !samePath(left.qualifier, right.qualifier) || + left.searchPaths.length !== right.searchPaths.length + ) { + return false; + } + for ( + let index = 0; + index < left.searchPaths.length; + index += 1 + ) { + const leftPath = left.searchPaths[index]; + const rightPath = right.searchPaths[index]; + if ( + !leftPath || + !rightPath || + !samePath(leftPath, rightPath) + ) { + return false; + } + } + return true; +} + +function keyFrom( + request: SqlCatalogSearchRequest, + dialect: SqlRelationDialectRuntime, +): StructuralKey { + return { + continuationToken: request.continuationToken, + dialect, + dialectId: request.dialectId, + limit: request.limit, + prefix: request.prefix, + qualifier: request.qualifier, + scope: request.scope, + searchPaths: request.searchPaths, + }; +} + +function textBytes(value: string): number { + return 16 + value.length * 2; +} + +function componentBytes( + component: SqlIdentifierComponent, +): number { + return COMPONENT_FIXED_BYTES + textBytes(component.value); +} + +function pathBytes(path: SqlIdentifierPath): number { + let bytes = PATH_FIXED_BYTES; + for (const component of path) { + bytes += componentBytes(component); + } + return bytes; +} + +function keyBytes(key: StructuralKey): number { + let bytes = + KEY_FIXED_BYTES + + textBytes(key.scope) + + textBytes(key.dialectId) + + componentBytes(key.prefix) + + pathBytes(key.qualifier); + if (key.continuationToken !== null) { + bytes += textBytes(key.continuationToken); + } + for (const path of key.searchPaths) { + bytes += pathBytes(path); + } + return bytes; +} + +function epochBytes(epoch: SqlCatalogEpoch): number { + return 32 + textBytes(epoch.token); +} + +function readyResponseBytes( + response: Extract< + SqlValidatedCatalogSearchResponse, + { readonly status: "ready" } + >, +): number { + let bytes = + RESPONSE_FIXED_BYTES + epochBytes(response.epoch); + if (response.coverage.kind === "paginated") { + bytes += textBytes(response.coverage.continuationToken); + } + for (const relation of response.relations) { + bytes += + RELATION_FIXED_BYTES + + textBytes(relation.entityId) + + textBytes(relation.completionText); + if (relation.detail !== undefined) { + bytes += textBytes(relation.detail); + } + for (const component of relation.canonicalPath) { + bytes += componentBytes(component); + } + for (const component of relation.completionPath) { + bytes += componentBytes(component); + } + } + return bytes; +} + +function touch( + state: StoreState, + record: RetainedRecord, +): void { + state.sequence += 1n; + record.lastUse = state.sequence; +} + +function removeAt(state: StoreState, index: number): void { + const removed = state.records.splice(index, 1); + for (const record of removed) { + state.retainedBytes -= record.bytes; + } +} + +function removeWhere( + state: StoreState, + predicate: (record: RetainedRecord) => boolean, +): void { + for ( + let index = state.records.length - 1; + index >= 0; + index -= 1 + ) { + const record = state.records[index]; + if (record && predicate(record)) removeAt(state, index); + } +} + +function retainReady( + state: StoreState, + record: ReadyRecord, +): boolean { + if ( + record.bytes > + MAX_CATALOG_POLICY_STORE_RETAINED_BYTES + ) { + return false; + } + touch(state, record); + state.records.push(record); + state.retainedBytes += record.bytes; + while ( + state.records.length > + MAX_CATALOG_POLICY_STORE_ENTRIES || + state.retainedBytes > + MAX_CATALOG_POLICY_STORE_RETAINED_BYTES + ) { + let oldestIndex = state.records.indexOf(record); + for (let index = 0; index < state.records.length; index += 1) { + const candidate = state.records[index]; + const oldest = state.records[oldestIndex]; + if ( + candidate?.kind === "ready" && + oldest && + candidate.lastUse < oldest.lastUse + ) { + oldestIndex = index; + } + } + removeAt(state, oldestIndex); + } + return state.records.includes(record); +} + +function retainGate( + state: StoreState, + record: FailureRecord | LoadingRecord, +): boolean { + if ( + record.bytes > + MAX_CATALOG_POLICY_STORE_RETAINED_BYTES + ) { + return false; + } + while ( + state.records.length >= + MAX_CATALOG_POLICY_STORE_ENTRIES || + state.retainedBytes + record.bytes > + MAX_CATALOG_POLICY_STORE_RETAINED_BYTES + ) { + let oldestReadyIndex = -1; + for (let index = 0; index < state.records.length; index += 1) { + const candidate = state.records[index]; + const oldest = + oldestReadyIndex < 0 + ? null + : state.records[oldestReadyIndex]; + if ( + candidate?.kind === "ready" && + (!oldest || candidate.lastUse < oldest.lastUse) + ) { + oldestReadyIndex = index; + } + } + if (oldestReadyIndex < 0) return false; + removeAt(state, oldestReadyIndex); + } + touch(state, record); + state.records.push(record); + state.retainedBytes += record.bytes; + return true; +} + +function findReady( + state: StoreState, + key: StructuralKey, + epoch: SqlCatalogEpoch, +): ReadyRecord | null { + for (const record of state.records) { + if ( + record.kind === "ready" && + sameEpoch(record.response.epoch, epoch) && + sameBaseKey(record.key, key) + ) { + return record; + } + } + return null; +} + +function findNeverGate( + state: StoreState, + key: StructuralKey, +): FailureRecord | null { + for (const record of state.records) { + if ( + record.kind === "failure" && + record.retry === "never" && + sameBaseKey(record.key, key) + ) { + return record; + } + } + return null; +} + +function findEpochGate( + state: StoreState, + key: StructuralKey, + epoch: SqlCatalogEpoch, +): FailureRecord | LoadingRecord | null { + for (const record of state.records) { + if ( + record.kind !== "ready" && + (record.kind === "loading" || + record.retry !== "never") && + sameEpoch(record.epoch, epoch) && + sameBaseKey(record.key, key) + ) { + return record; + } + } + return null; +} + +function accepted( + response: SqlValidatedCatalogSearchResponse, + retained: + | "failure-gate" + | "loading-barrier" + | "none" + | "ready", +): SqlCatalogSearchPolicyRecordResult { + return Object.freeze({ + response, + retained, + status: "accepted", + }); +} + +function conflict( + reason: + | "capacity" + | "epoch-mismatch" + | "loading-transition" + | "retry-gated", +): SqlCatalogSearchPolicyRecordResult { + return Object.freeze({ reason, status: "conflict" }); +} + +function probe( + state: StoreState, + request: SqlCatalogSearchRequest, + dialect: SqlRelationDialectRuntime, +): SqlCatalogSearchPolicyProbe { + if (state.disposed) return DISPOSED_PROBE; + if (state.saturated) { + return Object.freeze({ status: "overloaded" }); + } + const key = keyFrom(request, dialect); + const epoch = request.expectedEpoch; + if (epoch === null) { + return Object.freeze({ + loadingEpoch: null, + status: "miss", + }); + } + const neverGate = findNeverGate(state, key); + if (neverGate) { + touch(state, neverGate); + return Object.freeze({ + code: neverGate.code, + epoch: request.expectedEpoch ?? neverGate.epoch, + retry: neverGate.retry, + status: "failed", + }); + } + const ready = findReady(state, key, epoch); + if (ready) { + touch(state, ready); + return Object.freeze({ + response: ready.response, + status: "ready", + }); + } + const gate = findEpochGate(state, key, epoch); + if (!gate) { + return Object.freeze({ + loadingEpoch: null, + status: "miss", + }); + } + touch(state, gate); + if (gate.kind === "loading") { + return Object.freeze({ + loadingEpoch: gate.epoch, + status: "miss", + }); + } + return Object.freeze({ + code: gate.code, + epoch: gate.epoch, + retry: gate.retry, + status: "failed", + }); +} + +function recordResponse( + state: StoreState, + request: SqlCatalogSearchRequest, + dialect: SqlRelationDialectRuntime, + response: SqlValidatedCatalogSearchResponse, +): SqlCatalogSearchPolicyRecordResult { + if (state.disposed) return DISPOSED_RECORD; + if (state.saturated) return conflict("capacity"); + if ( + request.expectedEpoch !== null && + !sameEpoch(request.expectedEpoch, response.epoch) + ) { + return conflict("epoch-mismatch"); + } + const key = keyFrom(request, dialect); + const neverGate = findNeverGate(state, key); + if (neverGate) return conflict("retry-gated"); + const epochGate = findEpochGate( + state, + key, + response.epoch, + ); + if (response.status === "ready") { + if (epochGate?.kind === "loading") { + return conflict("loading-transition"); + } + if (epochGate?.kind === "failure") { + return conflict("retry-gated"); + } + removeWhere( + state, + (candidate) => + candidate.kind === "ready" && + sameEpoch(candidate.response.epoch, response.epoch) && + sameBaseKey(candidate.key, key), + ); + const retained = retainReady(state, { + bytes: keyBytes(key) + readyResponseBytes(response), + key, + kind: "ready", + lastUse: 0n, + response, + }); + return accepted(response, retained ? "ready" : "none"); + } + if (response.status === "loading") { + if (epochGate?.kind === "failure") { + return conflict("retry-gated"); + } + if (epochGate?.kind === "loading") { + touch(state, epochGate); + return accepted(response, "loading-barrier"); + } + removeWhere( + state, + (candidate) => + candidate.kind === "loading" && + sameEpoch(candidate.epoch, response.epoch) && + sameBaseKey(candidate.key, key), + ); + const retained = retainGate(state, { + bytes: keyBytes(key) + epochBytes(response.epoch) + 64, + epoch: response.epoch, + key, + kind: "loading", + lastUse: 0n, + }); + if (retained) { + return accepted(response, "loading-barrier"); + } + state.saturated = true; + return conflict("capacity"); + } + if (response.retry === "next-request") { + return accepted(response, "none"); + } + const replacedGates = state.records.filter( + (candidate): candidate is FailureRecord | LoadingRecord => + candidate.kind !== "ready" && + ((candidate.kind === "loading" && + sameEpoch(candidate.epoch, response.epoch)) || + (candidate.kind === "failure" && + (response.retry === "never" || + candidate.retry === response.retry))) && + sameBaseKey(candidate.key, key), + ); + removeWhere( + state, + (candidate) => + candidate.kind === "loading" && + sameEpoch(candidate.epoch, response.epoch) && + sameBaseKey(candidate.key, key), + ); + removeWhere( + state, + (candidate) => + candidate.kind === "failure" && + (response.retry === "never" || + candidate.retry === response.retry) && + sameBaseKey(candidate.key, key), + ); + const retained = retainGate(state, { + bytes: + keyBytes(key) + + epochBytes(response.epoch) + + textBytes(response.code) + + 80, + code: response.code, + epoch: response.epoch, + key, + kind: "failure", + lastUse: 0n, + retry: response.retry, + }); + if (!retained) { + for (const replaced of replacedGates) { + retainGate(state, replaced); + } + state.saturated = true; + } + return retained + ? accepted(response, "failure-gate") + : conflict("capacity"); +} + +function advanceScope( + state: StoreState, + scope: string, + epoch: SqlCatalogEpoch, +): void { + if (state.disposed) return; + removeWhere(state, (record) => { + if (record.key.scope !== scope) return false; + if ( + record.kind === "failure" && + record.retry === "never" + ) { + return false; + } + const recordEpoch = + record.kind === "ready" + ? record.response.epoch + : record.epoch; + return !sameEpoch(recordEpoch, epoch); + }); +} + +function metrics( + state: StoreState, +): SqlCatalogSearchPolicyStoreMetrics { + let failureGates = 0; + let loadingBarriers = 0; + let readyEntries = 0; + for (const record of state.records) { + if (record.kind === "failure") failureGates += 1; + else if (record.kind === "loading") { + loadingBarriers += 1; + } else readyEntries += 1; + } + return Object.freeze({ + entries: state.records.length, + failureGates, + loadingBarriers, + readyEntries, + retainedBytes: state.retainedBytes, + }); +} + +export function createSqlCatalogSearchPolicyStore(): SqlCatalogSearchPolicyStore { + const state: StoreState = { + disposed: false, + records: [], + retainedBytes: 0, + saturated: false, + sequence: 0n, + }; + return Object.freeze({ + advanceScope: ( + scope: string, + epoch: SqlCatalogEpoch, + ): void => { + advanceScope(state, scope, epoch); + }, + dispose: (): void => { + if (state.disposed) return; + state.disposed = true; + state.records.length = 0; + state.retainedBytes = 0; + state.saturated = false; + }, + metrics: (): SqlCatalogSearchPolicyStoreMetrics => + metrics(state), + probe: ( + request: SqlCatalogSearchRequest, + dialect: SqlRelationDialectRuntime, + ): SqlCatalogSearchPolicyProbe => + probe(state, request, dialect), + record: ( + request: SqlCatalogSearchRequest, + dialect: SqlRelationDialectRuntime, + response: SqlValidatedCatalogSearchResponse, + ): SqlCatalogSearchPolicyRecordResult => + recordResponse(state, request, dialect, response), + }); +} diff --git a/src/vnext/relation-catalog-search-work.ts b/src/vnext/relation-catalog-search-work.ts index 6fc35ee..ca1034e 100644 --- a/src/vnext/relation-catalog-search-work.ts +++ b/src/vnext/relation-catalog-search-work.ts @@ -18,7 +18,14 @@ import type { SqlCatalogRevisionTarget, SqlCatalogScopeMembership, } from "./relation-catalog-epoch-coordinator.js"; +import { + createSqlCatalogSearchPolicyStore, +} from "./relation-catalog-search-policy-store.js"; +import type { + SqlCatalogSearchPolicyStore, +} from "./relation-catalog-search-policy-store.js"; import type { + SqlCatalogEpoch, SqlCatalogSearchRequest, } from "./relation-completion-types.js"; import type { SqlRelationDialectRuntime } from "./relation-dialect.js"; @@ -35,12 +42,15 @@ 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 DEFAULT_CATALOG_REFRESH_LEASE_MS = 1_000; 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 const MIN_CATALOG_REFRESH_LEASE_MS = 1; +export const MAX_CATALOG_REFRESH_LEASE_MS = 5_000; export interface SqlCatalogSearchDeadlineScheduler { readonly clearTimeout: ( @@ -59,6 +69,7 @@ export interface SqlCatalogSearchWorkOptions { readonly deadlineScheduler?: SqlCatalogSearchDeadlineScheduler; readonly executionDeadlineMs?: number; readonly queueDeadlineMs?: number; + readonly refreshLeaseMs?: number; readonly synchronousBudgetMs?: number; } @@ -97,8 +108,33 @@ export type SqlCatalogSearchWorkOutcome = readonly reason: SqlCatalogSearchWorkUnavailableReason; }; +export type SqlCatalogSearchAvailabilityTarget = ( + this: void, +) => + | ((this: void) => undefined) + | null; + +export type SqlCatalogSearchRefreshRetentionResult = + | { + readonly status: "retained"; + readonly remainingLeaseMs: number; + } + | { + readonly status: "unavailable"; + readonly reason: + | "disposed" + | "expired" + | "invalid-target" + | "not-retainable" + | "superseded"; + }; + export interface SqlCatalogSearchWorkTicket { readonly cancel: (this: void) => void; + readonly retainForRefresh: ( + this: void, + prepareAvailability: SqlCatalogSearchAvailabilityTarget, + ) => SqlCatalogSearchRefreshRetentionResult; readonly result: Promise; } @@ -151,6 +187,7 @@ interface NormalizedOptions { readonly deadlineScheduler: SqlCatalogSearchDeadlineScheduler; readonly executionDeadlineMs: number; readonly queueDeadlineMs: number; + readonly refreshLeaseMs: number; readonly synchronousBudgetMs: number; } @@ -160,6 +197,7 @@ interface OwnerState { disposed: boolean; readonly membership: SqlCatalogScopeMembership; owner: CoordinatorState | null; + refreshObserver: RefreshObserverState | null; requestToken: object | null; readonly scope: string; } @@ -167,12 +205,29 @@ interface OwnerState { interface ConsumerState { readonly capture: SqlCatalogEpochCapture; cancelled: boolean; + readonly kind: "consumer"; owner: OwnerState | null; resolve: (outcome: SqlCatalogSearchWorkOutcome) => void; + readonly retention: TicketRetentionState; settled: boolean; work: WorkState | null; } +interface TicketRetentionState { + consumer: ConsumerState | null; + observer: RefreshObserverState | null; +} + +interface RefreshObserverState { + readonly capture: SqlCatalogEpochCapture; + readonly kind: "observer"; + readonly prepareAvailability: SqlCatalogSearchAvailabilityTarget; + readonly owner: OwnerState; + readonly retention: TicketRetentionState; + timer: DeadlineCell | null; + work: WorkState | null; +} + interface WorkState { abortController: AbortController | null; abortIssued: boolean; @@ -181,6 +236,7 @@ interface WorkState { executionDeadline: number | null; executionTimer: DeadlineCell | null; joinable: boolean; + readonly observers: Set; readonly owners: Set; phase: | "active" @@ -218,8 +274,10 @@ interface CoordinatorState { readonly epochs: SqlCatalogEpochCoordinator; readonly joinable: Set; lastNow: number; + readonly notifications: Set; readonly options: NormalizedOptions; readonly owners: Set; + readonly policyStore: SqlCatalogSearchPolicyStore; pumpRequested: boolean; pumping: boolean; readonly queue: WorkState[]; @@ -236,14 +294,18 @@ interface CoordinatorState { interface ResponseDecisionCell { active: boolean; advancing: boolean; - candidates: readonly ConsumerState[]; + candidates: readonly ResponseCandidate[]; index: number; pending: SqlCatalogResponseEpochDecision | null; - response: SqlValidatedCatalogSearchResponse | null; - state: CoordinatorState | null; - work: WorkState | null; + readonly dialect: SqlRelationDialectRuntime; + readonly request: SqlCatalogSearchRequest; + readonly response: SqlValidatedCatalogSearchResponse; + readonly state: CoordinatorState; + readonly work: WorkState; } +type ResponseCandidate = ConsumerState | RefreshObserverState; + interface DetachedSettlement { readonly outcome: SqlCatalogSearchWorkOutcome; readonly resolve: ( @@ -251,8 +313,26 @@ interface DetachedSettlement { ) => void; } +interface AvailabilityNotificationState { + active: boolean; + readonly owner: OwnerState; + readonly requestToken: object | null; + readonly scope: string; +} + +interface AvailabilityPreparation { + readonly notification: AvailabilityNotificationState; + readonly prepare: SqlCatalogSearchAvailabilityTarget; +} + +interface AvailabilityDispatch { + readonly dispatch: Function; + readonly notification: AvailabilityNotificationState; +} + interface Effects { readonly aborts: AbortController[]; + readonly availabilityPreparations: AvailabilityPreparation[]; pump: boolean; readonly settlements: DetachedSettlement[]; readonly timers: DeadlineCell[]; @@ -303,6 +383,7 @@ function unavailableOutcome( function effects(): Effects { return { aborts: [], + availabilityPreparations: [], pump: false, settlements: [], timers: [], @@ -335,6 +416,9 @@ function normalizeOptions( const executionDeadlineMs = candidate?.executionDeadlineMs ?? DEFAULT_CATALOG_EXECUTION_DEADLINE_MS; + const refreshLeaseMs = + candidate?.refreshLeaseMs ?? + DEFAULT_CATALOG_REFRESH_LEASE_MS; const synchronousBudgetMs = candidate?.synchronousBudgetMs ?? DEFAULT_CATALOG_SYNCHRONOUS_BUDGET_MS; @@ -360,6 +444,11 @@ function normalizeOptions( MIN_CATALOG_SYNCHRONOUS_BUDGET_MS, MAX_CATALOG_SYNCHRONOUS_BUDGET_MS, ) || + !isDuration( + refreshLeaseMs, + MIN_CATALOG_REFRESH_LEASE_MS, + MAX_CATALOG_REFRESH_LEASE_MS, + ) || typeof nowMethod !== "function" || typeof setTimeoutMethod !== "function" || typeof clearTimeoutMethod !== "function" @@ -402,6 +491,7 @@ function normalizeOptions( deadlineScheduler: capturedScheduler, executionDeadlineMs, queueDeadlineMs, + refreshLeaseMs, synchronousBudgetMs, }), }; @@ -559,6 +649,7 @@ function settleDetached( outcome: SqlCatalogSearchWorkOutcome, ): void { consumer.settled = true; + consumer.retention.consumer = null; const resolve = consumer.resolve; consumer.resolve = IGNORE_DETACHED_REJECTION; consumer.work = null; @@ -572,9 +663,38 @@ function runEffects( state: CoordinatorState, pending: Effects, ): void { + const availabilityDispatches: AvailabilityDispatch[] = []; for (const timer of pending.timers) { clearDeadline(state, timer); } + for (const preparation of pending.availabilityPreparations) { + const { notification, prepare } = preparation; + if (!isCurrentNotification(state, notification)) { + retireNotification(state, notification); + continue; + } + let candidate: unknown; + try { + candidate = Reflect.apply(prepare, undefined, []); + } catch { + retireNotification(state, notification); + continue; + } + if ( + typeof candidate === "function" && + isCurrentNotification(state, notification) + ) { + availabilityDispatches.push({ + dispatch: candidate, + notification, + }); + } else if (candidate !== null) { + drainDetachedSettlement(candidate); + retireNotification(state, notification); + } else { + retireNotification(state, notification); + } + } for (const settlement of pending.settlements) { settlement.resolve(settlement.outcome); } @@ -585,9 +705,65 @@ function runEffects( // Work was made inert before provider abort. } } + for (const item of availabilityDispatches) { + const { dispatch, notification } = item; + if (!isCurrentNotification(state, notification)) { + retireNotification(state, notification); + continue; + } + retireNotification(state, notification); + try { + const result = Reflect.apply(dispatch, undefined, []); + if (result !== undefined) { + drainDetachedSettlement(result); + } + } catch { + // Availability state was already detached and cannot reopen. + } + } if (pending.pump) pump(state); } +function isCurrentNotification( + state: CoordinatorState, + notification: AvailabilityNotificationState, +): boolean { + const owner = notification.owner; + return ( + notification.active && + !state.disposed && + !owner.disposed && + owner.owner === state && + owner.requestToken === notification.requestToken + ); +} + +function retireNotification( + state: CoordinatorState, + notification: AvailabilityNotificationState, +): void { + if (!notification.active) return; + notification.active = false; + state.notifications.delete(notification); +} + +function retireNotifications( + state: CoordinatorState, + predicate: ( + notification: AvailabilityNotificationState, + ) => boolean, +): void { + for (const notification of state.notifications) { + if (predicate(notification)) { + retireNotification(state, notification); + } + } +} + +function hasWorkOwners(work: WorkState): boolean { + return work.owners.size > 0 || work.observers.size > 0; +} + function sameComponent( left: SqlIdentifierComponent, right: SqlIdentifierComponent, @@ -698,9 +874,6 @@ function revokeDecisionCell(work: WorkState): void { cell.active = false; cell.candidates = []; cell.pending = null; - cell.response = null; - cell.state = null; - cell.work = null; } function detachAbort( @@ -727,6 +900,32 @@ function detachOwners( } } +function detachObserverState( + observer: RefreshObserverState, + pending: Effects, +): void { + const work = observer.work; + if (!work) return; + observer.work = null; + work.observers.delete(observer); + const owner = observer.owner; + owner.refreshObserver = null; + const timer = observer.timer; + observer.timer = null; + if (timer) pending.timers.push(timer); + observer.retention.observer = null; +} + +function detachObservers( + work: WorkState, + pending: Effects, +): void { + const observers = [...work.observers]; + for (const observer of observers) { + detachObserverState(observer, pending); + } +} + function finishWork( state: CoordinatorState, work: WorkState, @@ -756,6 +955,7 @@ function finishWork( revokeProviderCell(work); revokeDecisionCell(work); detachOwners(work, outcome, pending); + detachObservers(work, pending); if (abort) detachAbort(work, pending); else work.abortController = null; work.dialect = null; @@ -772,12 +972,44 @@ function retireActiveOwners( removeJoinable(state, work); work.phase = "retired"; detachOwners(work, outcome, pending); + detachObservers(work, pending); detachAbort(work, pending); work.dialect = null; work.request = null; work.scope = ""; } +function retireUnownedWork( + state: CoordinatorState, + work: WorkState, + pending: Effects, +): void { + 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 detachConsumerInto( state: CoordinatorState, consumer: ConsumerState, @@ -789,31 +1021,8 @@ function detachConsumerInto( 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, - ); - } + if (work && !hasWorkOwners(work)) { + retireUnownedWork(state, work, pending); } } @@ -832,6 +1041,26 @@ function detachConsumer( runEffects(state, pending); } +function detachRefreshObserverInto( + state: CoordinatorState, + observer: RefreshObserverState, + pending: Effects, +): void { + const work = observer.work; + detachObserverState(observer, pending); + if (!work || hasWorkOwners(work)) return; + retireUnownedWork(state, work, pending); +} + +function detachRefreshObserver( + state: CoordinatorState, + observer: RefreshObserverState, +): void { + const pending = effects(); + detachRefreshObserverInto(state, observer, pending); + runEffects(state, pending); +} + function handleQueueTimeout( state: CoordinatorState, work: WorkState, @@ -958,6 +1187,64 @@ function finishDecision( runEffects(state, pending); } +function prepareAvailabilityObservers( + state: CoordinatorState, + work: WorkState, + pending: Effects, +): void { + const observers = [...work.observers]; + for (const observer of observers) { + const prepare = observer.prepareAvailability; + const notification: AvailabilityNotificationState = { + active: true, + owner: observer.owner, + requestToken: observer.owner.requestToken, + scope: observer.owner.scope, + }; + state.notifications.add(notification); + detachObserverState(observer, pending); + pending.availabilityPreparations.push({ + notification, + prepare, + }); + } +} + +function finishUsableDecision( + cell: ResponseDecisionCell, + state: CoordinatorState, + work: WorkState, + outcome: Extract< + SqlCatalogSearchWorkOutcome, + { readonly status: "usable" } + >, +): void { + const recorded = state.policyStore.record( + cell.request, + cell.dialect, + outcome.response, + ); + if (recorded.status !== "accepted") { + finishDecision( + cell, + state, + work, + recorded.status === "conflict" && + recorded.reason === "capacity" + ? unavailableOutcome("overloaded") + : SUPERSEDED_OUTCOME, + ); + return; + } + cell.active = false; + const pending = effects(); + if (outcome.response.status === "ready") { + prepareAvailabilityObservers(state, work, pending); + } + finishWork(state, work, outcome, pending, false); + runEffects(state, pending); +} + function decisionOutcome( decision: Exclude< SqlCatalogResponseEpochDecision, @@ -1011,7 +1298,6 @@ function advanceDecision(cell: ResponseDecisionCell): void { 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) { @@ -1024,7 +1310,7 @@ function advanceDecision(cell: ResponseDecisionCell): void { if (pendingDecision.observation === "baseline") { rekeyUnobservedWork(state, work, response); } - finishDecision( + finishUsableDecision( cell, state, work, @@ -1051,7 +1337,9 @@ function advanceDecision(cell: ResponseDecisionCell): void { cell.index += 1; if ( candidate && - !candidate.settled && + (candidate.kind === "observer" + ? candidate.work !== null + : !candidate.settled) && candidate.work === work ) { capture = candidate.capture; @@ -1178,7 +1466,7 @@ function handleProviderResult( if ( state.disposed || work.phase !== "deciding" || - work.owners.size === 0 + !hasWorkOwners(work) ) { return; } @@ -1186,7 +1474,7 @@ function handleProviderResult( if ( state.disposed || work.phase !== "deciding" || - work.owners.size === 0 || + !hasWorkOwners(work) || afterDecode === null || work.executionDeadline === null || afterDecode >= work.executionDeadline @@ -1209,9 +1497,14 @@ function handleProviderResult( const cell: ResponseDecisionCell = { active: true, advancing: false, - candidates: [...work.owners], + candidates: [ + ...work.owners, + ...work.observers, + ], + dialect, index: 0, pending: null, + request, response: decoded.value, state, work, @@ -1228,7 +1521,7 @@ function startWork( if ( state.disposed || work.phase !== "queued" || - work.owners.size === 0 + !hasWorkOwners(work) ) { return; } @@ -1358,6 +1651,16 @@ function makeImmediateTicket( ): SqlCatalogSearchWorkTicket { return Object.freeze({ cancel: (): void => {}, + retainForRefresh: + (): SqlCatalogSearchRefreshRetentionResult => + Object.freeze({ + reason: + outcome.status === "unavailable" && + outcome.reason === "disposed" + ? "disposed" + : "not-retainable", + status: "unavailable", + }), result: new INTRINSIC_PROMISE( (resolve) => { resolve(outcome); @@ -1366,6 +1669,110 @@ function makeImmediateTicket( }); } +function unavailableRetention( + reason: Extract< + SqlCatalogSearchRefreshRetentionResult, + { readonly status: "unavailable" } + >["reason"], +): SqlCatalogSearchRefreshRetentionResult { + return Object.freeze({ reason, status: "unavailable" }); +} + +function retainConsumerForRefresh( + state: CoordinatorState, + consumer: ConsumerState, + owner: OwnerState, + work: WorkState, + prepareAvailability: SqlCatalogSearchAvailabilityTarget, +): SqlCatalogSearchRefreshRetentionResult { + if (typeof prepareAvailability !== "function") { + return unavailableRetention("invalid-target"); + } + const retention = consumer.retention; + const hardDeadline = + work.phase === "queued" + ? work.queueDeadline + : work.executionDeadline; + const requestToken = owner.requestToken; + const now = readNow(state); + if ( + state.disposed || + owner.disposed || + owner.owner !== state + ) { + return unavailableRetention("disposed"); + } + if ( + owner.requestToken !== requestToken || + consumer.settled || + consumer.work !== work || + owner.current !== consumer + ) { + return unavailableRetention("superseded"); + } + if ( + now === null || + hardDeadline === null || + now >= hardDeadline + ) { + return unavailableRetention("expired"); + } + const leaseDeadline = deadlineFrom( + now, + state.options.refreshLeaseMs, + ); + if (leaseDeadline === null) { + return unavailableRetention("expired"); + } + const deadline = Math.min( + hardDeadline, + leaseDeadline, + ); + const observer: RefreshObserverState = { + capture: consumer.capture, + kind: "observer", + owner, + prepareAvailability, + retention, + timer: null, + work, + }; + work.observers.add(observer); + owner.refreshObserver = observer; + retention.observer = observer; + const pending = effects(); + work.owners.delete(consumer); + settleDetached( + pending.settlements, + consumer, + CANCELLED_OUTCOME, + ); + const timer = scheduleDeadline( + state, + deadline, + () => detachRefreshObserver(state, observer), + ); + if (observer.work) { + observer.timer = timer; + } else { + clearDeadline(state, timer); + } + runEffects(state, pending); + if (!observer.work) { + return unavailableRetention( + state.disposed || owner.disposed + ? "disposed" + : owner.requestToken !== requestToken + ? "superseded" + : "expired", + ); + } + return Object.freeze({ + remainingLeaseMs: deadline - now, + status: "retained", + }); +} + function makeConsumer( capture: SqlCatalogEpochCapture, owner: OwnerState, @@ -1381,20 +1788,37 @@ function makeConsumer( resolve = settle; }, ); + const retention: TicketRetentionState = { + consumer: null, + observer: null, + }; const consumer: ConsumerState = { cancelled: false, capture, + kind: "consumer", owner, resolve, + retention, settled: false, work: null, }; + retention.consumer = consumer; return { consumer, ticket: Object.freeze({ cancel: (): void => { - if (consumer.cancelled || consumer.settled) return; + if (consumer.cancelled) return; consumer.cancelled = true; + const observer = retention.observer; + const observerState = observer?.owner?.owner; + if (observer && observerState) { + detachRefreshObserver( + observerState, + observer, + ); + return; + } + if (consumer.settled) return; const state = consumer.owner?.owner; if (state) { detachConsumer( @@ -1404,6 +1828,30 @@ function makeConsumer( ); } }, + retainForRefresh: ( + prepareAvailability: + SqlCatalogSearchAvailabilityTarget, + ): SqlCatalogSearchRefreshRetentionResult => { + const retainedConsumer = retention.consumer; + const retainedOwner = retainedConsumer?.owner; + const retainedWork = retainedConsumer?.work; + const state = retainedOwner?.owner; + if ( + !retainedConsumer || + !retainedOwner || + !retainedWork || + !state + ) { + return unavailableRetention("not-retainable"); + } + return retainConsumerForRefresh( + state, + retainedConsumer, + retainedOwner, + retainedWork, + prepareAvailability, + ); + }, result, }), }; @@ -1428,6 +1876,7 @@ function replaceOwnerConsumer( ): void { const pending = effects(); const previous = owner.current; + const previousObserver = owner.refreshObserver; work.owners.add(consumer); consumer.work = work; owner.current = consumer; @@ -1439,9 +1888,56 @@ function replaceOwnerConsumer( pending, ); } + if (previousObserver) { + detachRefreshObserverInto( + state, + previousObserver, + pending, + ); + } runEffects(state, pending); } +function replaceOwnerWithImmediate( + state: CoordinatorState, + owner: OwnerState, + requestToken: object, + outcome: SqlCatalogSearchWorkOutcome, +): SqlCatalogSearchWorkTicket { + const pending = effects(); + const previous = owner.current; + const previousObserver = owner.refreshObserver; + if (previous) { + detachConsumerInto( + state, + previous, + SUPERSEDED_OUTCOME, + pending, + ); + } + if (previousObserver) { + detachRefreshObserverInto( + state, + previousObserver, + pending, + ); + } + runEffects(state, pending); + if ( + state.disposed || + owner.disposed || + owner.owner !== state + ) { + return makeImmediateTicket( + unavailableOutcome("disposed"), + ); + } + if (owner.requestToken !== requestToken) { + return makeImmediateTicket(SUPERSEDED_OUTCOME); + } + return makeImmediateTicket(outcome); +} + function requestWork( state: CoordinatorState, owner: OwnerState, @@ -1473,6 +1969,13 @@ function requestWork( pending, ); } + if (owner.refreshObserver) { + detachRefreshObserverInto( + state, + owner.refreshObserver, + pending, + ); + } runEffects(state, pending); return makeImmediateTicket( unavailableOutcome( @@ -1513,21 +2016,73 @@ function requestWork( 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( + return replaceOwnerWithImmediate( + state, + owner, + requestToken, unavailableOutcome("invalid-request"), ); } + const policy = state.policyStore.probe( + request, + owner.dialect, + ); + if (policy.status === "ready") { + return replaceOwnerWithImmediate( + state, + owner, + requestToken, + Object.freeze({ + observation: "equal", + response: policy.response, + status: "usable", + }), + ); + } + if (policy.status === "failed") { + return replaceOwnerWithImmediate( + state, + owner, + requestToken, + Object.freeze({ + observation: "equal", + response: Object.freeze({ + code: policy.code, + epoch: policy.epoch, + retry: policy.retry, + status: "failed", + }), + status: "usable", + }), + ); + } + if (policy.status === "overloaded") { + return replaceOwnerWithImmediate( + state, + owner, + requestToken, + unavailableOutcome("overloaded"), + ); + } + if ( + policy.status === "miss" && + policy.loadingEpoch !== null && + state.epochs.hasLiveSubscription(owner.scope) + ) { + return replaceOwnerWithImmediate( + state, + owner, + requestToken, + Object.freeze({ + observation: "equal", + response: Object.freeze({ + epoch: policy.loadingEpoch, + status: "loading", + }), + status: "usable", + }), + ); + } const created = makeConsumer(captured.capture, owner); const existing = findWork( state, @@ -1545,6 +2100,7 @@ function requestWork( } const pending = effects(); const previous = owner.current; + const previousObserver = owner.refreshObserver; if (previous) { detachConsumerInto( state, @@ -1553,6 +2109,13 @@ function requestWork( pending, ); } + if (previousObserver) { + detachRefreshObserverInto( + state, + previousObserver, + pending, + ); + } if ( state.queue.length >= MAX_CATALOG_QUEUED_SEARCH_WORK @@ -1602,6 +2165,7 @@ function requestWork( executionDeadline: null, executionTimer: null, joinable: true, + observers: new Set(), owners: new Set([created.consumer]), phase: "queued", providerCell: null, @@ -1636,8 +2200,13 @@ function requestWork( function prepareTransition( state: CoordinatorState, scope: string, + epoch: SqlCatalogEpoch, ): (() => undefined) | null { - if (state.disposed) return null; + retireNotifications( + state, + (notification) => notification.scope === scope, + ); + state.policyStore.advanceScope(scope, epoch); const pending = effects(); for (const work of state.works) { if ( @@ -1693,11 +2262,35 @@ function disposeOwner(owner: OwnerState): void { owner.disposed = true; owner.owner = null; owner.requestToken = null; - state?.owners.delete(owner); + if (state) { + retireNotifications( + state, + (notification) => notification.owner === owner, + ); + state.owners.delete(owner); + } const current = owner.current; + const observer = owner.refreshObserver; owner.current = null; - if (state && current) { - detachConsumer(state, current, CANCELLED_OUTCOME); + owner.refreshObserver = null; + if (state && (current || observer)) { + const pending = effects(); + if (current) { + detachConsumerInto( + state, + current, + CANCELLED_OUTCOME, + pending, + ); + } + if (observer) { + detachRefreshObserverInto( + state, + observer, + pending, + ); + } + runEffects(state, pending); } owner.membership.dispose(); } @@ -1756,6 +2349,7 @@ function prepareOwner( disposed: false, membership: prepared.membership, owner: state, + refreshObserver: null, requestToken: null, scope, }; @@ -1773,6 +2367,7 @@ function disposeCoordinatorState( state.disposed = true; state.search = null; state.pumpRequested = false; + retireNotifications(state, () => true); const pending = effects(); for (const work of state.works) { finishWork( @@ -1787,9 +2382,11 @@ function disposeCoordinatorState( owner.disposed = true; owner.current = null; owner.owner = null; + owner.refreshObserver = null; owner.requestToken = null; } state.owners.clear(); + state.policyStore.dispose(); state.epochs.dispose(); runEffects(state, pending); } @@ -1815,8 +2412,8 @@ export function createSqlCatalogSearchWorkCoordinator( let state: CoordinatorState | null = null; const epochResult = createSqlCatalogEpochCoordinator( provider, - (scope): (() => undefined) | null => - state ? prepareTransition(state, scope) : null, + (scope, epoch): (() => undefined) | null => + state ? prepareTransition(state, scope, epoch) : null, (): undefined => { if (state) disposeCoordinatorState(state); return undefined; @@ -1834,8 +2431,10 @@ export function createSqlCatalogSearchWorkCoordinator( epochs: epochResult.coordinator, joinable: new Set(), lastNow: normalized.initialNow, + notifications: new Set(), options: normalized.options, owners: new Set(), + policyStore: createSqlCatalogSearchPolicyStore(), pumpRequested: false, pumping: false, queue: [], diff --git a/test/vnext-types/relation-catalog-search-work.test-d.ts b/test/vnext-types/relation-catalog-search-work.test-d.ts index 8176b88..d34f12f 100644 --- a/test/vnext-types/relation-catalog-search-work.test-d.ts +++ b/test/vnext-types/relation-catalog-search-work.test-d.ts @@ -9,6 +9,7 @@ import { createSqlCatalogEpochCoordinator, } from "../../src/vnext/relation-catalog-epoch-coordinator.js"; import type { + SqlCatalogSearchAvailabilityTarget, SqlCatalogSearchWorkCoordinator, SqlCatalogSearchWorkInput, SqlCatalogSearchWorkOutcome, @@ -82,6 +83,31 @@ const thisFreeRequest: ( const thisFreeCancel: ( this: void, ) => void = ticket.cancel; +const thisFreeRetainForRefresh: SqlCatalogSearchWorkTicket["retainForRefresh"] = + ticket.retainForRefresh; +const retention = ticket.retainForRefresh( + (): ((this: void) => undefined) => + (): undefined => undefined, +); +if (retention.status === "retained") { + void retention.remainingLeaseMs; +} + +const asyncAvailabilityDispatch = (): Promise => + Promise.resolve(undefined); +const invalidAsyncAvailabilityTarget: SqlCatalogSearchAvailabilityTarget = + // @ts-expect-error availability dispatch is synchronously exact-undefined + () => asyncAvailabilityDispatch; + +const receiverDependentAvailability = function ( + this: { readonly active: boolean }, +): null { + void this.active; + return null; +}; +// @ts-expect-error availability preparation cannot depend on a receiver +const invalidAvailabilityReceiver: SqlCatalogSearchAvailabilityTarget = + receiverDependentAvailability; const receiverDependentRequest = function ( this: { readonly active: boolean }, @@ -241,10 +267,13 @@ void thisFreeActivate; void thisFreeOwnerDispose; void thisFreeRequest; void thisFreeCancel; +void thisFreeRetainForRefresh; void invalidPrepareOwnerReceiver; void invalidActivateReceiver; void invalidRequestReceiver; void invalidCancelReceiver; +void invalidAsyncAvailabilityTarget; +void invalidAvailabilityReceiver; void scopeLeakingInput; void providerLeakingInput; void providerHandleLeakingInput;