From 408d052dbfc1f9eab6698b1ed80b727cbc818c47 Mon Sep 17 00:00:00 2001 From: Programmable <309941960+0xprogrammable@users.noreply.github.com> Date: Mon, 3 Aug 2026 07:14:23 +0200 Subject: [PATCH] fix(read-model): attest indexed market health --- app/api/ops/health/route.ts | 29 + config/read-model-operations.v1.json | 6 +- ...didate-projector-runtime-binding.server.ts | 31 +- lib/data-pipeline/read-model-health.server.ts | 527 ++++++++++++++++++ .../perf/read-model-ops-source-contracts.mjs | 11 +- ...803000100_market_projector_health_view.sql | 267 +++++++++ .../022_market_projector_health_view.test.sql | 357 ++++++++++++ ...andidate-projector-runtime-binding.test.ts | 65 ++- .../operations-health-route.test.ts | 198 +++++++ tests/data-pipeline/read-model-health.test.ts | 354 ++++++++++++ 10 files changed, 1839 insertions(+), 6 deletions(-) create mode 100644 lib/data-pipeline/read-model-health.server.ts create mode 100644 supabase/migrations/20260803000100_market_projector_health_view.sql create mode 100644 supabase/tests/database/022_market_projector_health_view.test.sql create mode 100644 tests/data-pipeline/operations-health-route.test.ts create mode 100644 tests/data-pipeline/read-model-health.test.ts diff --git a/app/api/ops/health/route.ts b/app/api/ops/health/route.ts index c7d6d9bc..fe44a2f6 100644 --- a/app/api/ops/health/route.ts +++ b/app/api/ops/health/route.ts @@ -5,6 +5,7 @@ import { readDurableExploreModel, readIndependentRpcHealth, } from "../../../../lib/onchain"; +import { readIndexedReadModelHealth } from "../../../../lib/data-pipeline/read-model-health.server"; export const dynamic = "force-dynamic"; export const runtime = "nodejs"; @@ -14,12 +15,40 @@ const MAX_INDEX_AGE_MS = 15 * 60 * 1_000; export async function GET() { const startedAt = Date.now(); try { + const indexed = await readIndexedReadModelHealth(); const deployment = getOperationalOnchainDeployment("production"); if (deployment.status !== "ready") { throw new Error( "The verified production release is not operationally eligible", ); } + if (indexed) { + if (indexed.chainId !== deployment.chainId) { + throw new Error("Indexed read-model chain binding is unavailable"); + } + const rpc = await readIndependentRpcHealth(deployment); + if ( + rpc.chainId !== deployment.chainId || + rpc.chainId !== indexed.chainId + ) { + throw new Error("Indexed read-model chain binding is unavailable"); + } + return NextResponse.json( + { + status: "healthy", + chainId: indexed.chainId, + index: indexed.index, + rpc, + checkedAt: new Date().toISOString(), + }, + { + headers: { + "Cache-Control": "public, max-age=0, s-maxage=30", + }, + }, + ); + } + const [index, rpc] = await Promise.all([ readDurableExploreModel(deployment, MAX_INDEX_AGE_MS), readIndependentRpcHealth(deployment), diff --git a/config/read-model-operations.v1.json b/config/read-model-operations.v1.json index 3da4f84f..4057a6b0 100644 --- a/config/read-model-operations.v1.json +++ b/config/read-model-operations.v1.json @@ -30,7 +30,7 @@ "dependencies": [ { "path": "lib/data-pipeline/candidate-projector-runtime-binding.server.ts", - "sha256": "32efa13d740614f7e66fd20a0158edf3383f4f6643a7fe34268fabda6261931c" + "sha256": "9a97556d3fca540586b56eb1bccf8c9c779e56c890cf03e585cd172206944615" } ], "migrations": [ @@ -97,6 +97,10 @@ { "path": "supabase/migrations/20260802092800_market_projector_fast_lane.sql", "sha256": "70c2719af30e0d3438e3de306376c7fa62d0196be98f81d7bd6b327559c14dc7" + }, + { + "path": "supabase/migrations/20260803000100_market_projector_health_view.sql", + "sha256": "946000d60600f8b144fb535579f6808b0acfd6da3331f17511f712e7bb24b2fd" } ] } diff --git a/lib/data-pipeline/candidate-projector-runtime-binding.server.ts b/lib/data-pipeline/candidate-projector-runtime-binding.server.ts index 14edc4a8..8b350e07 100644 --- a/lib/data-pipeline/candidate-projector-runtime-binding.server.ts +++ b/lib/data-pipeline/candidate-projector-runtime-binding.server.ts @@ -116,6 +116,11 @@ export type CandidateDatabasePromotionBinding = Readonly<{ stagedDeploymentId: string; }>; +export type DatabasePhysicalIdentity = Readonly<{ + databaseName: string; + systemIdentifier: string; +}>; + export type ProjectorRuntimeBindingSelection = | Readonly<{ mode: typeof CANDIDATE_PROJECTOR_RUNTIME_MODE; @@ -404,8 +409,8 @@ export async function assertCandidateDatabaseBootstrapState(input: Readonly<{ export async function assertCandidateDatabasePromotedState(input: Readonly<{ executor: PostgresExecutor; binding: CandidateDatabasePromotionBinding; -}>): Promise { - await input.executor.transaction(async (transaction) => { +}>): Promise { + return input.executor.transaction(async (transaction) => { const login = await transaction.query<{ session_user: unknown }>( "select session_user::text as session_user", ); @@ -446,5 +451,27 @@ export async function assertCandidateDatabasePromotedState(input: Readonly<{ if (rows.length !== 1 || rows[0]?.verified !== true) { return invalidCandidateBinding(); } + const identityRows = await transaction.query<{ + database_name: unknown; + system_identifier: unknown; + }>( + `select + pg_catalog.current_database()::text as database_name, + ((pg_catalog.pg_control_system()).system_identifier)::text + as system_identifier`, + ); + const databaseName = identityRows[0]?.database_name; + const systemIdentifier = identityRows[0]?.system_identifier; + if ( + identityRows.length !== 1 || + typeof databaseName !== "string" || + !/^[a-z][a-z0-9_]{0,62}$/u.test(databaseName) || + typeof systemIdentifier !== "string" || + !/^[1-9]\d{0,19}$/u.test(systemIdentifier) || + BigInt(systemIdentifier) > 18_446_744_073_709_551_615n + ) { + return invalidCandidateBinding(); + } + return Object.freeze({ databaseName, systemIdentifier }); }); } diff --git a/lib/data-pipeline/read-model-health.server.ts b/lib/data-pipeline/read-model-health.server.ts new file mode 100644 index 00000000..40aa5abe --- /dev/null +++ b/lib/data-pipeline/read-model-health.server.ts @@ -0,0 +1,527 @@ +import "server-only"; + +import { + assertCandidateDatabasePromotedState, + type CandidateDatabasePromotionBinding, + type DatabasePhysicalIdentity, +} from "./candidate-projector-runtime-binding.server"; +import { INDEXED_ROUTE_FLAG_NAMES } from "./config"; +import { + createPostgresExecutor, + type PostgresExecutor, +} from "./postgres"; +import { validatedPostgresConnectionTarget } from "./postgres-connection.server"; +import { loadProjectorRuntimeConfig } from "./projector-runtime-config.server"; +import { + readExactRouteSnapshotReadiness, +} from "./public-route-readiness.server"; +import { reconcilerRouteKeysForScope } from "./reconciler-preparity"; +import { getServerReadModel } from "./read-model.server"; +import { + ALL_REVIEWED_ROUTE_SCOPES, + INDEXED_ROUTE_KEYS, + type IndexedRouteKey, + type ReviewedRouteScope, +} from "./route-coordinator.server"; + +type Environment = Readonly>; +type ServerReadModel = NonNullable< + Awaited> +>; + +type IndexedReadModelHealthDependencies = Readonly<{ + getServerReadModel: typeof getServerReadModel; + loadProjectorRuntimeConfig: typeof loadProjectorRuntimeConfig; + createPostgresExecutor: typeof createPostgresExecutor; + assertCandidateDatabasePromotedState: typeof assertCandidateDatabasePromotedState; + readExactRouteSnapshotReadiness: typeof readExactRouteSnapshotReadiness; + nowMs: () => number; +}>; + +type CheckpointHealthRow = Readonly<{ + chain_id: unknown; + release_id: unknown; + model_id: unknown; + source_group: unknown; + block_number: unknown; + created_at: unknown; +}>; + +type IndexedHealthResult = Readonly<{ + chainId: 1; + index: Readonly<{ + ageSeconds: number; + blockNumber: string; + tokenCount: number; + }>; +}>; + +const MAXIMUM_CHECKPOINT_AGE_MS = 15 * 60 * 1_000; +const MAXIMUM_CLOCK_SKEW_MS = 60 * 1_000; +const DATABASE_IDENTITY_SINGLETON = Symbol.for( + "programmable.read-model-health.promoted-database-identity.v1", +); + +type IdentityRegistry = { + [DATABASE_IDENTITY_SINGLETON]?: Readonly<{ + key: string; + promise: Promise; + }>; +}; + +function identityRegistry(): IdentityRegistry { + return globalThis as typeof globalThis & IdentityRegistry; +} + +const DEFAULT_DEPENDENCIES: IndexedReadModelHealthDependencies = Object.freeze({ + getServerReadModel, + loadProjectorRuntimeConfig, + createPostgresExecutor, + assertCandidateDatabasePromotedState, + readExactRouteSnapshotReadiness, + nowMs: Date.now, +}); + +function fail(): never { + throw new Error("Indexed read-model health is unavailable"); +} + +function canonicalInteger(value: unknown): string { + if (typeof value === "bigint") { + if (value < 0n) return fail(); + return value.toString(); + } + if (typeof value !== "string" || !/^(?:0|[1-9]\d*)$/u.test(value)) { + return fail(); + } + return value; +} + +function bytes32(value: unknown): string { + if (value instanceof Uint8Array && value.byteLength === 32) { + return Array.from(value, (byte) => byte.toString(16).padStart(2, "0")).join(""); + } + if (typeof value === "string" && /^(?:0x|\\x)[0-9a-fA-F]{64}$/u.test(value)) { + return value.slice(2).toLowerCase(); + } + return fail(); +} + +function timestampMs(value: unknown): number { + const parsed = value instanceof Date ? value.valueOf() : + typeof value === "string" ? Date.parse(value) : Number.NaN; + if (!Number.isFinite(parsed)) return fail(); + return parsed; +} + +function nullableTimestampMs(value: unknown): number | null { + return value === null || value === undefined ? null : timestampMs(value); +} + +function validatePhysicalIdentity(value: unknown): DatabasePhysicalIdentity { + if (typeof value !== "object" || value === null) return fail(); + const candidate = value as Record; + if ( + typeof candidate.databaseName !== "string" || + !/^[a-z][a-z0-9_]{0,62}$/u.test(candidate.databaseName) || + typeof candidate.systemIdentifier !== "string" || + !/^[1-9]\d{0,19}$/u.test(candidate.systemIdentifier) || + BigInt(candidate.systemIdentifier) > 18_446_744_073_709_551_615n + ) { + return fail(); + } + return Object.freeze({ + databaseName: candidate.databaseName, + systemIdentifier: candidate.systemIdentifier, + }); +} + +function physicalIdentityFromRows( + rows: readonly Record[], +): DatabasePhysicalIdentity { + if (rows.length !== 1) return fail(); + return validatePhysicalIdentity({ + databaseName: rows[0]?.database_name, + systemIdentifier: rows[0]?.system_identifier, + }); +} + +function indexedActivationState(env: Environment): "indexed" | "legacy" { + const values = INDEXED_ROUTE_FLAG_NAMES.map((name) => env[name]); + const shadow = env.INDEXED_READ_SHADOW_COMPARE_ENABLED; + const shadowOff = shadow === undefined || shadow === "" || shadow === "false"; + const shadowOn = shadow === "true"; + if (!shadowOff && !shadowOn) return fail(); + const routeOff = (value: string | undefined) => + value === undefined || value === "" || value === "false"; + if (values.some((value) => !routeOff(value) && value !== "true")) return fail(); + if (shadowOff && values.every(routeOff)) return "legacy"; + return "indexed"; +} + +function promotionCacheKey( + binding: CandidateDatabasePromotionBinding, + connectionString: string, +): string { + const target = validatedPostgresConnectionTarget(connectionString); + const url = new URL(target.connectionString); + return JSON.stringify([ + target.hostname, + target.port, + url.pathname, + decodeURIComponent(url.username), + binding.providerDeploymentId, + binding.deploymentCommitment, + binding.schemaCommitment, + binding.initializationInputCommitment, + binding.initializedAt, + binding.productCommit, + binding.stagedDeploymentId, + ]); +} + +async function readPromotedDatabaseIdentity(input: Readonly<{ + binding: CandidateDatabasePromotionBinding; + connectionString: string; + sslCaPem: string; + dependencies: IndexedReadModelHealthDependencies; +}>): Promise { + const key = promotionCacheKey(input.binding, input.connectionString); + const registry = identityRegistry(); + const existing = registry[DATABASE_IDENTITY_SINGLETON]; + if (existing?.key === key) return existing.promise; + + const target = validatedPostgresConnectionTarget(input.connectionString); + const executor: PostgresExecutor = input.dependencies.createPostgresExecutor({ + connectionString: input.connectionString, + sslCaPem: input.sslCaPem, + allowInsecureLoopback: target.isLoopback, + maxConnections: 1, + connectTimeoutMs: 2_000, + idleTimeoutMs: 15_000, + }); + const promise = input.dependencies.assertCandidateDatabasePromotedState({ + executor, + binding: input.binding, + }).then(validatePhysicalIdentity).finally(() => executor.close()); + registry[DATABASE_IDENTITY_SINGLETON] = Object.freeze({ key, promise }); + try { + return await promise; + } finally { + if (registry[DATABASE_IDENTITY_SINGLETON]?.promise === promise) { + delete registry[DATABASE_IDENTITY_SINGLETON]; + } + } +} + +function exactRouteScopes( + releaseScopes: ReturnType["releaseScopes"], +): readonly Readonly<{ + route: IndexedRouteKey; + scopes: readonly ReviewedRouteScope[]; +}>[] { + const reviewed = ALL_REVIEWED_ROUTE_SCOPES.filter((scope) => + releaseScopes.some( + (release) => + release.releaseId === scope.releaseVersion && + release.modelId === scope.model, + ), + ); + if (reviewed.length !== releaseScopes.length) return fail(); + return INDEXED_ROUTE_KEYS.map((route) => { + const scopes = reviewed.filter((scope) => + reconcilerRouteKeysForScope( + scope.releaseVersion, + scope.model, + ).includes(route), + ); + if (scopes.length === 0) return fail(); + return Object.freeze({ route, scopes: Object.freeze(scopes) }); + }); +} + +function checkpointHealth(input: Readonly<{ + rows: readonly CheckpointHealthRow[]; + releaseScopes: ReturnType["releaseScopes"]; + nowMs: number; +}>): Readonly<{ ageSeconds: number; blockNumber: string }> { + const matched = input.releaseScopes.map((scope) => { + const rows = input.rows.filter( + (row) => + canonicalInteger(row.chain_id) === "1" && + row.release_id === scope.releaseId && + row.model_id === scope.modelId && + row.source_group === scope.sourceGroup, + ); + if (rows.length !== 1) return fail(); + const blockNumber = canonicalInteger(rows[0]!.block_number); + const createdAtMs = timestampMs(rows[0]!.created_at); + if ( + BigInt(blockNumber) < 1n || + createdAtMs > input.nowMs + MAXIMUM_CLOCK_SKEW_MS || + input.nowMs - createdAtMs > MAXIMUM_CHECKPOINT_AGE_MS + ) { + return fail(); + } + return Object.freeze({ blockNumber, createdAtMs }); + }); + const minimumBlock = matched.reduce( + (minimum, row) => + BigInt(row.blockNumber) < BigInt(minimum) ? row.blockNumber : minimum, + matched[0]!.blockNumber, + ); + const oldestCreatedAt = Math.min(...matched.map((row) => row.createdAtMs)); + return Object.freeze({ + ageSeconds: Math.max(0, Math.floor((input.nowMs - oldestCreatedAt) / 1_000)), + blockNumber: minimumBlock, + }); +} + +async function readApiDatabaseHealth(input: Readonly<{ + readModel: ServerReadModel; + releaseScopes: ReturnType["releaseScopes"]; + nowMs: number; + dependencies: IndexedReadModelHealthDependencies; +}>): Promise> { + const routeScopes = exactRouteScopes(input.releaseScopes); + return input.readModel.repeatableReadSnapshot(async (transaction) => { + const [ + identityRows, + checkpointRows, + circuitRows, + corpusRows, + marketRows, + ...readiness + ] = await Promise.all([ + transaction.query>( + `select + pg_catalog.current_database()::text as database_name, + ((pg_catalog.pg_control_system()).system_identifier)::text + as system_identifier`, + ), + transaction.query( + `select chain_id, release_id, model_id, source_group, + block_number, created_at + from programmable_private.checkpoint_summary_v1 + order by chain_id, release_id, model_id, source_group`, + ), + transaction.query>( + `select dependency, circuit_status + from programmable_private.health_summary_v1 + order by dependency`, + ), + transaction.query>( + `select release_id, model_id, source_group, + '0x' || pg_catalog.encode(pool_id, 'hex') as pool_id + from programmable_private.launch_by_token_v2 + where chain_id = $1 + order by release_id, model_id, source_group, pool_id`, + ["1"], + ), + transaction.query>( + `select * + from programmable_private.market_projector_health_v1 + where chain_id = $1 + order by release_id, model_id, source_group, pool_id`, + ["1"], + ), + ...routeScopes.map(({ route, scopes }) => + input.dependencies.readExactRouteSnapshotReadiness({ + transaction, + route, + chainId: 1, + scope: scopes, + }), + ), + ]); + + if ( + circuitRows.some((row) => row.circuit_status !== "closed") || + readiness.length !== routeScopes.length || + readiness.some((snapshot, index) => + snapshot.readiness.length !== routeScopes[index]!.scopes.length || + snapshot.readiness.some( + (member) => + member.eligibility !== "eligible" || + member.parity !== "current" || + member.version === undefined, + ), + ) + ) { + return fail(); + } + if (corpusRows.length < 1 || corpusRows.length > 1_000_000) return fail(); + const expectedMarkets = new Set( + corpusRows.map(marketScopeKey), + ); + if (expectedMarkets.size !== corpusRows.length) return fail(); + validateMarketHealth({ + rows: marketRows, + expectedMarkets, + releaseScopes: input.releaseScopes, + nowMs: input.nowMs, + }); + const checkpoint = checkpointHealth({ + rows: checkpointRows, + releaseScopes: input.releaseScopes, + nowMs: input.nowMs, + }); + return Object.freeze({ + identity: physicalIdentityFromRows(identityRows), + index: Object.freeze({ + ...checkpoint, + tokenCount: corpusRows.length, + }), + }); + }); +} + +function marketScopeKey(row: Record): string { + if ( + typeof row.release_id !== "string" || + typeof row.model_id !== "string" || + typeof row.source_group !== "string" + ) return fail(); + return JSON.stringify([ + row.release_id, + row.model_id, + row.source_group, + bytes32(row.pool_id), + ]); +} + +function validateMarketHealth(input: Readonly<{ + rows: readonly Record[]; + expectedMarkets: ReadonlySet; + releaseScopes: ReturnType["releaseScopes"]; + nowMs: number; +}>): void { + if (input.rows.length !== input.expectedMarkets.size) return fail(); + const markets = new Set(); + for (const row of input.rows) { + if (canonicalInteger(row.chain_id) !== "1") return fail(); + const release = input.releaseScopes.find( + (scope) => + scope.releaseId === row.release_id && + scope.modelId === row.model_id && + scope.sourceGroup === row.source_group, + ); + if (!release) return fail(); + const market = marketScopeKey(row); + if (!input.expectedMarkets.has(market) || markets.has(market)) return fail(); + markets.add(market); + if ( + typeof row.market_projector_version !== "string" || + !/^[a-z0-9]+(?:[._-][a-z0-9]+)*$/u.test(row.market_projector_version) || + typeof row.source_projector_version !== "string" || + !/^[a-z0-9]+(?:[._-][a-z0-9]+)*$/u.test(row.source_projector_version) || + row.cursor_epoch_id !== row.source_checkpoint_epoch_id || + canonicalInteger(row.cursor_pointer_generation) !== + canonicalInteger(row.source_checkpoint_pointer_generation) || + BigInt(canonicalInteger(row.cursor_block_number)) > + BigInt(canonicalInteger(row.source_checkpoint_block_number)) || + BigInt(canonicalInteger(row.cursor_source_reorg_generation)) < + BigInt(canonicalInteger(row.source_checkpoint_reorg_generation)) || + canonicalInteger(row.source_checkpoint_cursor_block_global_log_index) !== + "4294967295" || + row.source_checkpoint_cursor_candidate_id !== "empty-page" || + BigInt(canonicalInteger(row.cursor_generation)) < 1n || + BigInt(canonicalInteger(row.source_checkpoint_generation)) < 1n + ) { + return fail(); + } + canonicalInteger(row.cursor_reorg_generation); + canonicalInteger(row.source_checkpoint_reorg_generation); + + const cursorAdvancedAt = timestampMs(row.cursor_advanced_at); + const sourceCreatedAt = timestampMs(row.source_checkpoint_created_at); + if ( + cursorAdvancedAt > input.nowMs + MAXIMUM_CLOCK_SKEW_MS || + sourceCreatedAt > input.nowMs + MAXIMUM_CLOCK_SKEW_MS || + input.nowMs - sourceCreatedAt > MAXIMUM_CHECKPOINT_AGE_MS + ) { + return fail(); + } + if ( + BigInt(canonicalInteger(row.latest_snapshot_block_number)) < 1n || + BigInt(canonicalInteger(row.latest_snapshot_block_number)) > + BigInt(canonicalInteger(row.cursor_block_number)) + ) { + return fail(); + } + const hourCoverageEnd = nullableTimestampMs(row.hour_coverage_end); + const dayCoverageEnd = nullableTimestampMs(row.day_coverage_end); + const latestSnapshotObservedAt = timestampMs(row.latest_snapshot_observed_at); + const latestSnapshotAttachedAt = timestampMs(row.latest_snapshot_attached_at); + const latestSnapshotReconciledAt = timestampMs( + row.latest_snapshot_reconciled_at, + ); + if ( + (hourCoverageEnd !== null && + hourCoverageEnd > input.nowMs + MAXIMUM_CLOCK_SKEW_MS) || + (dayCoverageEnd !== null && + dayCoverageEnd > input.nowMs + 24 * 60 * 60 * 1_000) || + latestSnapshotObservedAt > input.nowMs + MAXIMUM_CLOCK_SKEW_MS || + latestSnapshotAttachedAt > input.nowMs + MAXIMUM_CLOCK_SKEW_MS || + latestSnapshotReconciledAt > input.nowMs + MAXIMUM_CLOCK_SKEW_MS + ) { + return fail(); + } + } + if (markets.size !== input.expectedMarkets.size) return fail(); +} + +/** + * Returns null without parsing unrelated data-pipeline configuration only for + * the exact legacy-off activation state. Indexed mode proves one promoted + * physical database, current route parity, source checkpoints and market data. + */ +export async function readIndexedReadModelHealth( + env: Environment = process.env, + dependencies: IndexedReadModelHealthDependencies = DEFAULT_DEPENDENCIES, +): Promise { + if (indexedActivationState(env) === "legacy") return null; + const nowMs = dependencies.nowMs(); + if (!Number.isSafeInteger(nowMs) || nowMs < 0) return fail(); + + const runtimeConfig = dependencies.loadProjectorRuntimeConfig(env); + if ( + runtimeConfig.binding.mode !== "release" || + runtimeConfig.binding.promotedDatabase === null + ) { + return fail(); + } + const readModel = await dependencies.getServerReadModel({ required: true }); + if (!readModel) return fail(); + + const [projectorIdentity, apiHealth] = await Promise.all([ + readPromotedDatabaseIdentity({ + binding: runtimeConfig.binding.promotedDatabase, + connectionString: runtimeConfig.database.projectorConnectionString, + sslCaPem: runtimeConfig.database.sslCaPem, + dependencies, + }), + readApiDatabaseHealth({ + readModel, + releaseScopes: runtimeConfig.releaseScopes, + nowMs, + dependencies, + }), + ]); + if ( + projectorIdentity.databaseName !== apiHealth.identity.databaseName || + projectorIdentity.systemIdentifier !== apiHealth.identity.systemIdentifier + ) { + return fail(); + } + return Object.freeze({ chainId: 1 as const, index: apiHealth.index }); +} + +/** Test isolation only; production promotion verification is deduped in-flight. */ +export function resetReadModelHealthForTests(): void { + if (process.env.NODE_ENV !== "test") return fail(); + delete identityRegistry()[DATABASE_IDENTITY_SINGLETON]; +} diff --git a/scripts/perf/read-model-ops-source-contracts.mjs b/scripts/perf/read-model-ops-source-contracts.mjs index 7c288f56..bb726e89 100644 --- a/scripts/perf/read-model-ops-source-contracts.mjs +++ b/scripts/perf/read-model-ops-source-contracts.mjs @@ -35,7 +35,7 @@ const APPROVED_OPERATIONS = Object.freeze({ dependencies: Object.freeze([ Object.freeze({ path: "lib/data-pipeline/candidate-projector-runtime-binding.server.ts", - sha256: "32efa13d740614f7e66fd20a0158edf3383f4f6643a7fe34268fabda6261931c", + sha256: "9a97556d3fca540586b56eb1bccf8c9c779e56c890cf03e585cd172206944615", }), ]), migrations: Object.freeze([ @@ -103,6 +103,10 @@ const APPROVED_OPERATIONS = Object.freeze({ path: "supabase/migrations/20260802092800_market_projector_fast_lane.sql", sha256: "70c2719af30e0d3438e3de306376c7fa62d0196be98f81d7bd6b327559c14dc7", }), + Object.freeze({ + path: "supabase/migrations/20260803000100_market_projector_health_view.sql", + sha256: "946000d60600f8b144fb535579f6808b0acfd6da3331f17511f712e7bb24b2fd", + }), ]), }), ]), @@ -838,7 +842,7 @@ export function evaluateReadModelOperationsSourceContracts( ); check( "ops-market-projector-migration", - marketWorker?.migrations?.length === 2 && + marketWorker?.migrations?.length === 3 && migrationContract( "market-projector", source(marketWorker.migrations[0]?.path), @@ -846,6 +850,9 @@ export function evaluateReadModelOperationsSourceContracts( migrationContract( "market-projector-fast-lane", source(marketWorker.migrations[1]?.path), + ) && + source(marketWorker.migrations[2]?.path)?.includes( + "market_projector_health_v1", ), "the market worker is bound to exact lineage, terminal checkpoint and lease SQL", ); diff --git a/supabase/migrations/20260803000100_market_projector_health_view.sql b/supabase/migrations/20260803000100_market_projector_health_view.sql new file mode 100644 index 00000000..ea6f735b --- /dev/null +++ b/supabase/migrations/20260803000100_market_projector_health_view.sql @@ -0,0 +1,267 @@ +-- Narrow API-reader health surface for the event-driven market projector. +-- A row exists for every current launch only when its cursor is on the current +-- release lineage, no canonical market event remains unprojected, and at least +-- one successfully reconciled snapshot belongs to the active market lineage. + +set role programmable_migrator; + +create index market_snapshot_lineage_health_latest_idx + on programmable_private.market_snapshot_lineage_memberships ( + chain_id, release_id, model_id, source_group, projector_version, + pool_id, reorg_generation, attached_at desc, market_snapshot_id + ); + +create view programmable_private.market_projector_health_v1 +with (security_invoker = false, security_barrier = true) +as +select + current_cursor.chain_id, + current_cursor.release_id, + current_cursor.model_id, + current_cursor.source_group, + current_cursor.projector_version as market_projector_version, + current_cursor.pool_id, + cursor_history.market_cursor_id, + cursor_history.epoch_id as cursor_epoch_id, + cursor_history.pointer_generation as cursor_pointer_generation, + cursor_history.cursor_generation, + cursor_history.reorg_generation as cursor_reorg_generation, + cursor_history.source_checkpoint_id as cursor_source_checkpoint_id, + cursor_history.source_checkpoint_generation + as cursor_source_checkpoint_generation, + cursor_history.source_reorg_generation + as cursor_source_reorg_generation, + cursor_history.block_number::bigint as cursor_block_number, + cursor_history.block_hash::bytea as cursor_block_hash, + cursor_history.advanced_at as cursor_advanced_at, + cursor_history.hour_coverage_end, + cursor_history.day_coverage_end, + source_checkpoint.projector_version as source_projector_version, + source_checkpoint.checkpoint_id as source_checkpoint_id, + source_checkpoint.epoch_id as source_checkpoint_epoch_id, + source_checkpoint.pointer_generation + as source_checkpoint_pointer_generation, + source_checkpoint.checkpoint_generation + as source_checkpoint_generation, + source_checkpoint.reorg_generation + as source_checkpoint_reorg_generation, + source_checkpoint.block_number::bigint + as source_checkpoint_block_number, + source_checkpoint.block_hash::bytea as source_checkpoint_block_hash, + source_checkpoint.cursor_block_global_log_index::bigint + as source_checkpoint_cursor_block_global_log_index, + source_checkpoint.cursor_candidate_id::text + as source_checkpoint_cursor_candidate_id, + source_checkpoint.created_at as source_checkpoint_created_at, + latest_snapshot.block_number::bigint as latest_snapshot_block_number, + latest_snapshot.observed_at as latest_snapshot_observed_at, + latest_snapshot.attached_at as latest_snapshot_attached_at, + latest_snapshot.reconciled_at as latest_snapshot_reconciled_at +from programmable_private.projector_checkpoint_current + as current_source_checkpoint +join programmable_private.projector_checkpoints as source_checkpoint + on source_checkpoint.checkpoint_id = + current_source_checkpoint.checkpoint_id + and source_checkpoint.chain_id = current_source_checkpoint.chain_id + and source_checkpoint.release_id = current_source_checkpoint.release_id + and source_checkpoint.model_id = current_source_checkpoint.model_id + and source_checkpoint.source_group = current_source_checkpoint.source_group + and source_checkpoint.projector_version = + current_source_checkpoint.projector_version + and source_checkpoint.checkpoint_generation = + current_source_checkpoint.checkpoint_generation + and source_checkpoint.reorg_generation = + current_source_checkpoint.reorg_generation +join programmable_private.release_epoch_current as current_epoch + on current_epoch.chain_id = source_checkpoint.chain_id + and current_epoch.release_id = source_checkpoint.release_id + and current_epoch.model_id = source_checkpoint.model_id + and current_epoch.source_group = source_checkpoint.source_group + and current_epoch.epoch_id = source_checkpoint.epoch_id + and current_epoch.generation = source_checkpoint.pointer_generation +join programmable_private.market_projector_cursor_current as current_cursor + on current_cursor.chain_id = source_checkpoint.chain_id + and current_cursor.release_id = source_checkpoint.release_id + and current_cursor.model_id = source_checkpoint.model_id + and current_cursor.source_group = source_checkpoint.source_group +join programmable_private.market_projector_cursor_history as cursor_history + on cursor_history.market_cursor_id = current_cursor.market_cursor_id + and cursor_history.chain_id = current_cursor.chain_id + and cursor_history.release_id = current_cursor.release_id + and cursor_history.model_id = current_cursor.model_id + and cursor_history.source_group = current_cursor.source_group + and cursor_history.projector_version = current_cursor.projector_version + and cursor_history.pool_id = current_cursor.pool_id + and cursor_history.cursor_generation = current_cursor.cursor_generation + and cursor_history.reorg_generation = current_cursor.reorg_generation +join programmable_private.projector_checkpoints as bound_source_checkpoint + on bound_source_checkpoint.checkpoint_id = + cursor_history.source_checkpoint_id + and bound_source_checkpoint.chain_id = cursor_history.chain_id + and bound_source_checkpoint.release_id = cursor_history.release_id + and bound_source_checkpoint.model_id = cursor_history.model_id + and bound_source_checkpoint.source_group = cursor_history.source_group + and bound_source_checkpoint.projector_version = + source_checkpoint.projector_version + and bound_source_checkpoint.epoch_id = cursor_history.epoch_id + and bound_source_checkpoint.pointer_generation = + cursor_history.pointer_generation + and bound_source_checkpoint.checkpoint_generation = + cursor_history.source_checkpoint_generation + and bound_source_checkpoint.reorg_generation = + cursor_history.source_reorg_generation +join lateral ( + select + snapshot.block_number, + snapshot.observed_at, + membership.attached_at, + fact_outcome.finished_at as reconciled_at + from programmable_private.market_snapshot_lineage_memberships as membership + join programmable_private.market_snapshots as snapshot + on snapshot.market_snapshot_id = membership.market_snapshot_id + and snapshot.chain_id = membership.chain_id + and snapshot.pool_id = membership.pool_id + join programmable_private.reconciliation_records as fact_reconciliation + on fact_reconciliation.reconciliation_id = snapshot.reconciliation_id + and fact_reconciliation.chain_id = snapshot.chain_id + and fact_reconciliation.mismatch_count = 0 + join programmable_private.run_headers as fact_run + on fact_run.run_id = fact_reconciliation.run_id + and fact_run.run_kind = 'reconciliation' + and fact_run.chain_id = membership.chain_id + and fact_run.release_id = membership.release_id + and fact_run.model_id = membership.model_id + and fact_run.source_group = membership.source_group + and fact_run.epoch_id = cursor_history.epoch_id + and fact_run.captured_pointer_generation = + cursor_history.pointer_generation + join programmable_private.run_lifecycle_outcomes as fact_outcome + on fact_outcome.run_id = fact_run.run_id + and fact_outcome.status = 'succeeded' + join programmable_private.reconciliation_records as attached_reconciliation + on attached_reconciliation.reconciliation_id = + membership.attached_reconciliation_id + and attached_reconciliation.chain_id = membership.chain_id + and attached_reconciliation.mismatch_count = 0 + join programmable_private.run_headers as attached_run + on attached_run.run_id = attached_reconciliation.run_id + and attached_run.run_kind = 'reconciliation' + and attached_run.chain_id = membership.chain_id + and attached_run.release_id = membership.release_id + and attached_run.model_id = membership.model_id + and attached_run.source_group = membership.source_group + and attached_run.epoch_id = cursor_history.epoch_id + and attached_run.captured_pointer_generation = + cursor_history.pointer_generation + join programmable_private.run_lifecycle_outcomes as attached_outcome + on attached_outcome.run_id = attached_run.run_id + and attached_outcome.status = 'succeeded' + where membership.chain_id = current_cursor.chain_id + and membership.release_id = current_cursor.release_id + and membership.model_id = current_cursor.model_id + and membership.source_group = current_cursor.source_group + and membership.projector_version = current_cursor.projector_version + and membership.pool_id = current_cursor.pool_id + and membership.reorg_generation = current_cursor.reorg_generation + and snapshot.block_number <= cursor_history.block_number + order by membership.attached_at desc, membership.market_snapshot_id desc + limit 1 +) as latest_snapshot on true +where source_checkpoint.cursor_block_global_log_index = 4294967295 + and source_checkpoint.cursor_candidate_id = 'empty-page' + and cursor_history.epoch_id = source_checkpoint.epoch_id + and cursor_history.pointer_generation = source_checkpoint.pointer_generation + and cursor_history.source_reorg_generation >= + source_checkpoint.reorg_generation + and cursor_history.block_number <= source_checkpoint.block_number + and not exists ( + select 1 + from ( + select distinct on ( + occurrence.block_number, occurrence.block_hash + ) + occurrence.occurrence_id, + occurrence.block_number, + occurrence.block_hash + from programmable_private.chain_event_occurrences as occurrence + join programmable_private.chain_event_occurrence_materializations + as materialization + on materialization.occurrence_id = occurrence.occurrence_id + and materialization.chain_id = source_checkpoint.chain_id + and materialization.release_id = source_checkpoint.release_id + and materialization.model_id = source_checkpoint.model_id + and materialization.source_group = source_checkpoint.source_group + and materialization.epoch_id = source_checkpoint.epoch_id + and materialization.pointer_generation = + source_checkpoint.pointer_generation + and case source_checkpoint.model_id + when 'classic' + then materialization.event_type = 'NativeSwapFeesAccrued' + when 'stock-paired' + then materialization.event_type = 'QuoteSwapFeesAccrued' + else false + end + join programmable_private.chain_event_current_canonical as canonical + on canonical.occurrence_id = occurrence.occurrence_id + and canonical.logical_event_id = occurrence.logical_event_id + and canonical.block_hash = occurrence.block_hash + where occurrence.chain_id = source_checkpoint.chain_id + and occurrence.block_number <= source_checkpoint.block_number + and pg_catalog.lower(materialization.decoded_payload ->> 'poolId') = + '0x' || pg_catalog.encode(current_cursor.pool_id, 'hex') + order by occurrence.block_number, occurrence.block_hash, + occurrence.transaction_index desc, + occurrence.block_global_log_index desc, + occurrence.occurrence_id desc + ) as required_close + where required_close.block_number > cursor_history.block_number + or ( + required_close.block_number = cursor_history.block_number + and not exists ( + select 1 + from programmable_private.market_block_closes as projected_close + join programmable_private.reconciliation_records + as close_reconciliation + on close_reconciliation.reconciliation_id = + projected_close.reconciliation_id + and close_reconciliation.mismatch_count = 0 + join programmable_private.run_headers as close_run + on close_run.run_id = close_reconciliation.run_id + and close_run.run_kind = 'reconciliation' + join programmable_private.run_lifecycle_outcomes as close_outcome + on close_outcome.run_id = close_run.run_id + and close_outcome.status = 'succeeded' + where projected_close.chain_id = source_checkpoint.chain_id + and projected_close.release_id = source_checkpoint.release_id + and projected_close.model_id = source_checkpoint.model_id + and projected_close.source_group = source_checkpoint.source_group + and projected_close.epoch_id = source_checkpoint.epoch_id + and projected_close.pointer_generation = + source_checkpoint.pointer_generation + and projected_close.pool_id = current_cursor.pool_id + and projected_close.last_source_occurrence_id = + required_close.occurrence_id + and projected_close.block_number = required_close.block_number + and projected_close.block_hash = required_close.block_hash + ) + ) + ); + +comment on view programmable_private.market_projector_health_v1 is + 'Fail-closed event-driven market cursor, terminal source checkpoint, and reconciled latest-snapshot health for exact API launch-corpus evaluation.'; + +revoke all on programmable_private.market_projector_health_v1 +from public, anon, authenticated, service_role, + programmable_projector, programmable_reconciler, + programmable_api_reader, programmable_profile_binder, + programmable_profile_recovery, programmable_profile_writer, + programmable_maintenance, programmable_operator, + programmable_projector_runtime, programmable_release_probe_nonce, + programmable_api_reader_login, programmable_projector_login, + programmable_reconciler_login, programmable_projector_runtime_login, + programmable_release_probe_nonce_login; + +grant select on programmable_private.market_projector_health_v1 +to programmable_api_reader; + +reset role; diff --git a/supabase/tests/database/022_market_projector_health_view.test.sql b/supabase/tests/database/022_market_projector_health_view.test.sql new file mode 100644 index 00000000..ce8564d7 --- /dev/null +++ b/supabase/tests/database/022_market_projector_health_view.test.sql @@ -0,0 +1,357 @@ +begin; + +set local timezone = 'UTC'; + +select plan(9); + +select ok( + to_regclass('programmable_private.market_projector_health_v1') is not null + and ( + select relation.relkind = 'v' + and relation.reloptions @> array[ + 'security_barrier=true', 'security_invoker=false' + ]::text[] + from pg_catalog.pg_class as relation + where relation.oid = + 'programmable_private.market_projector_health_v1'::regclass + ), + 'market health is a security-definer security-barrier view' +); + +select ok( + pg_catalog.strpos( + pg_catalog.pg_get_viewdef( + 'programmable_private.market_projector_health_v1'::regclass, true + ), + 'NativeSwapFeesAccrued' + ) > 0 + and pg_catalog.strpos( + pg_catalog.pg_get_viewdef( + 'programmable_private.market_projector_health_v1'::regclass, true + ), + 'market_block_closes' + ) > 0 + and pg_catalog.strpos( + pg_catalog.pg_get_viewdef( + 'programmable_private.market_projector_health_v1'::regclass, true + ), + 'cursor_history.source_reorg_generation' + ) > 0 + and pg_catalog.strpos( + pg_catalog.pg_get_viewdef( + 'programmable_private.market_projector_health_v1'::regclass, true + ), + 'DISTINCT ON (occurrence.block_number, occurrence.block_hash)' + ) > 0 + and pg_catalog.strpos( + pg_catalog.pg_get_viewdef( + 'programmable_private.market_projector_health_v1'::regclass, true + ), + 'occurrence.block_global_log_index DESC' + ) > 0, + 'health checks only the canonical last fee event in each covered block' +); + +select is( + ( + select pg_catalog.array_agg(attribute.attname::text order by attribute.attnum) + from pg_catalog.pg_attribute as attribute + where attribute.attrelid = + 'programmable_private.market_projector_health_v1'::regclass + and attribute.attnum > 0 + and not attribute.attisdropped + ), + array[ + 'chain_id', 'release_id', 'model_id', 'source_group', + 'market_projector_version', 'pool_id', + 'market_cursor_id', 'cursor_epoch_id', 'cursor_pointer_generation', + 'cursor_generation', 'cursor_reorg_generation', + 'cursor_source_checkpoint_id', 'cursor_source_checkpoint_generation', + 'cursor_source_reorg_generation', 'cursor_block_number', + 'cursor_block_hash', 'cursor_advanced_at', 'hour_coverage_end', + 'day_coverage_end', 'source_projector_version', 'source_checkpoint_id', + 'source_checkpoint_epoch_id', 'source_checkpoint_pointer_generation', + 'source_checkpoint_generation', 'source_checkpoint_reorg_generation', + 'source_checkpoint_block_number', 'source_checkpoint_block_hash', + 'source_checkpoint_cursor_block_global_log_index', + 'source_checkpoint_cursor_candidate_id', 'source_checkpoint_created_at', + 'latest_snapshot_block_number', 'latest_snapshot_observed_at', + 'latest_snapshot_attached_at', 'latest_snapshot_reconciled_at' + ]::text[], + 'the view exposes only scoped lineage and latest reconciled snapshot health' +); + +select ok( + exists ( + select 1 + from pg_catalog.pg_indexes + where schemaname = 'programmable_private' + and indexname = 'market_snapshot_lineage_health_latest_idx' + and indexdef like '%attached_at DESC%' + ), + 'latest snapshot health has a bounded index-backed lookup' +); + +select ok( + pg_catalog.has_table_privilege( + 'programmable_api_reader', + 'programmable_private.market_projector_health_v1', 'SELECT' + ) + and not pg_catalog.has_table_privilege( + 'anon', 'programmable_private.market_projector_health_v1', 'SELECT' + ) + and not pg_catalog.has_table_privilege( + 'authenticated', + 'programmable_private.market_projector_health_v1', 'SELECT' + ) + and not pg_catalog.has_table_privilege( + 'service_role', + 'programmable_private.market_projector_health_v1', 'SELECT' + ) + and not pg_catalog.has_table_privilege( + 'programmable_projector', + 'programmable_private.market_projector_health_v1', 'SELECT' + ) + and not pg_catalog.has_table_privilege( + 'programmable_reconciler', + 'programmable_private.market_projector_health_v1', 'SELECT' + ), + 'only the API reader receives the health capability' +); + +select ok( + not pg_catalog.has_table_privilege( + 'programmable_api_reader', + 'programmable_private.market_projector_cursor_current', 'SELECT' + ) + and not pg_catalog.has_table_privilege( + 'programmable_api_reader', + 'programmable_private.market_projector_cursor_history', 'SELECT' + ) + and not pg_catalog.has_table_privilege( + 'programmable_api_reader', + 'programmable_private.market_snapshot_lineage_memberships', 'SELECT' + ), + 'the health grant does not broaden access to base relations' +); + +set local session_replication_role = replica; + +insert into programmable_private.release_epoch_current ( + chain_id, release_id, model_id, source_group, epoch_id, generation, + changed_at, changed_by_audit_id +) values ( + 1, 'classic-v3', 'classic', 'core', + '22000000-0000-4000-8000-000000000001', 1, + '2026-08-03T04:00:00Z', + '22000000-0000-4000-8000-000000000002' +); + +insert into programmable_private.projector_checkpoints ( + checkpoint_id, chain_id, release_id, model_id, source_group, + projector_version, epoch_id, pointer_generation, lease_generation, + checkpoint_generation, reorg_generation, block_number, block_hash, + cursor_block_global_log_index, cursor_candidate_id, + safe_head_observation_id, target_block_evidence_id, run_id, + terminal_outcome_id, created_at +) values + ( + '22000000-0000-4000-8000-000000000010', + 1, 'classic-v3', 'classic', 'core', 'source-projector-v1', + '22000000-0000-4000-8000-000000000001', 1, 1, 1, 2, 1200, + pg_catalog.decode(pg_catalog.repeat('10', 32), 'hex'), + 4294967295, 'empty-page', + '22000000-0000-4000-8000-000000000011', + '22000000-0000-4000-8000-000000000012', + '22000000-0000-4000-8000-000000000013', + '22000000-0000-4000-8000-000000000014', + '2026-08-03T03:59:00Z' + ), + ( + '22000000-0000-4000-8000-000000000020', + 1, 'classic-v3', 'classic', 'core', 'source-projector-v1', + '22000000-0000-4000-8000-000000000001', 1, 1, 2, 2, 1250, + pg_catalog.decode(pg_catalog.repeat('20', 32), 'hex'), + 4294967295, 'empty-page', + '22000000-0000-4000-8000-000000000021', + '22000000-0000-4000-8000-000000000022', + '22000000-0000-4000-8000-000000000023', + '22000000-0000-4000-8000-000000000024', + '2026-08-03T04:01:00Z' + ); + +insert into programmable_private.projector_checkpoint_current ( + chain_id, release_id, model_id, source_group, projector_version, + checkpoint_id, checkpoint_generation, reorg_generation, changed_at +) values ( + 1, 'classic-v3', 'classic', 'core', 'source-projector-v1', + '22000000-0000-4000-8000-000000000010', 1, 2, + '2026-08-03T03:59:00Z' +); + +insert into programmable_private.run_headers ( + run_id, run_kind, chain_id, release_id, model_id, source_group, + epoch_id, captured_pointer_generation, worker_version, + request_commitment, caller_role, started_at, opened_by_audit_id +) values ( + '22000000-0000-4000-8000-000000000050', 'reconciliation', + 1, 'classic-v3', 'classic', 'core', + '22000000-0000-4000-8000-000000000001', 1, + 'market-reconciler-v1', + pg_catalog.decode(pg_catalog.repeat('50', 32), 'hex'), + 'programmable_reconciler', '2026-08-03T03:59:30Z', + '22000000-0000-4000-8000-000000000051' +); + +insert into programmable_private.reconciliation_records ( + reconciliation_id, run_id, chain_id, release_id, model_id, + epoch_id, pointer_generation, comparison_kind, severity, + source_from_block, source_to_block, compared_count, mismatch_count, + evidence_commitment, mismatch_identity_commitments, + resolved_at, recorded_at, audit_id +) values ( + '22000000-0000-4000-8000-000000000032', + '22000000-0000-4000-8000-000000000050', + 1, 'classic-v3', 'classic', + '22000000-0000-4000-8000-000000000001', 1, + 'market-health', 'info', 1100, 1200, 1, 0, + pg_catalog.decode(pg_catalog.repeat('52', 32), 'hex'), + array[]::bytea[], null, '2026-08-03T04:00:00Z', + '22000000-0000-4000-8000-000000000052' +); + +insert into programmable_private.run_lifecycle_outcomes ( + outcome_id, run_id, status, result_commitment, + caller_role, finished_at, audit_id +) values ( + '22000000-0000-4000-8000-000000000053', + '22000000-0000-4000-8000-000000000050', 'succeeded', + pg_catalog.decode(pg_catalog.repeat('53', 32), 'hex'), + 'programmable_reconciler', '2026-08-03T04:00:05Z', + '22000000-0000-4000-8000-000000000054' +); + +insert into programmable_private.market_snapshots ( + market_snapshot_id, chain_id, pool_id, source_deployment_id, + block_evidence_id, block_number, block_hash, sqrt_price_x96, liquidity, + market_volume_token0, market_volume_token1, market_volume_usd, + hook_gross_volume, observed_at, reconciliation_id, audit_id +) values ( + '22000000-0000-4000-8000-000000000101', 1, + pg_catalog.decode(pg_catalog.repeat('30', 32), 'hex'), + '22000000-0000-4000-8000-000000000110', + '22000000-0000-4000-8000-000000000111', 1200, + pg_catalog.decode(pg_catalog.repeat('10', 32), 'hex'), + 100, 1000, 1, 2, 3, 4, '2026-08-03T04:00:00Z', + '22000000-0000-4000-8000-000000000032', + '22000000-0000-4000-8000-000000000112' +); + +insert into programmable_private.market_projector_cursor_history ( + market_cursor_id, chain_id, release_id, model_id, source_group, + projector_version, pool_id, epoch_id, pointer_generation, + cursor_generation, reorg_generation, source_checkpoint_id, + source_checkpoint_generation, source_reorg_generation, + block_evidence_id, block_number, block_hash, provider_cursor, + hour_coverage_end, day_coverage_end, page_commitment, + reconciliation_id, advanced_at, audit_id +) values ( + '22000000-0000-4000-8000-000000000030', + 1, 'classic-v3', 'classic', 'core', 'market-projector-v1', + pg_catalog.decode(pg_catalog.repeat('30', 32), 'hex'), + '22000000-0000-4000-8000-000000000001', 1, 3, 2, + '22000000-0000-4000-8000-000000000010', 1, 2, + '22000000-0000-4000-8000-000000000031', 1200, + pg_catalog.decode(pg_catalog.repeat('10', 32), 'hex'), + 'block:1200:1010101010101010', null, null, + pg_catalog.decode(pg_catalog.repeat('31', 32), 'hex'), + '22000000-0000-4000-8000-000000000032', + '2026-08-03T04:00:00Z', + '22000000-0000-4000-8000-000000000033' +); + +insert into programmable_private.market_projector_cursor_current ( + chain_id, release_id, model_id, source_group, projector_version, pool_id, + market_cursor_id, cursor_generation, reorg_generation, changed_at, + changed_by_audit_id +) values ( + 1, 'classic-v3', 'classic', 'core', 'market-projector-v1', + pg_catalog.decode(pg_catalog.repeat('30', 32), 'hex'), + '22000000-0000-4000-8000-000000000030', 3, 2, + '2026-08-03T04:00:00Z', + '22000000-0000-4000-8000-000000000034' +); + +insert into programmable_private.market_snapshot_lineage_memberships ( + chain_id, release_id, model_id, source_group, projector_version, pool_id, + reorg_generation, market_snapshot_id, attached_reconciliation_id, + attached_at, audit_id +) values ( + 1, 'classic-v3', 'classic', 'core', 'market-projector-v1', + pg_catalog.decode(pg_catalog.repeat('30', 32), 'hex'), 2, + '22000000-0000-4000-8000-000000000101', + '22000000-0000-4000-8000-000000000032', + '2026-08-03T04:00:04Z', + '22000000-0000-4000-8000-000000000141' +); + +set local session_replication_role = origin; +set local role programmable_api_reader; + +select is( + (select pg_catalog.count(*) + from programmable_private.market_projector_health_v1), + 1::bigint, + 'a current launch with a caught-up reconciled cursor is healthy' +); + +reset role; +set local session_replication_role = replica; + +update programmable_private.projector_checkpoint_current +set checkpoint_id = '22000000-0000-4000-8000-000000000020', + checkpoint_generation = 2, + changed_at = '2026-08-03T04:01:00Z' +where chain_id = 1 + and release_id = 'classic-v3' + and model_id = 'classic' + and source_group = 'core' + and projector_version = 'source-projector-v1'; + +set local session_replication_role = origin; +set local role programmable_api_reader; + +select is( + (select pg_catalog.concat_ws('|', + cursor_block_number::text, source_checkpoint_block_number::text, + source_checkpoint_generation::text) + from programmable_private.market_projector_health_v1), + '1200|1250|2'::text, + 'an inactive cursor remains healthy across a newer empty source checkpoint' +); + +reset role; +set local session_replication_role = replica; + +update programmable_private.projector_checkpoints +set reorg_generation = 3 +where checkpoint_id = '22000000-0000-4000-8000-000000000020'; +update programmable_private.projector_checkpoint_current +set reorg_generation = 3 +where checkpoint_id = '22000000-0000-4000-8000-000000000020'; + +set local session_replication_role = origin; +set local role programmable_api_reader; + +select is( + (select pg_catalog.count(*) + from programmable_private.market_projector_health_v1), + 0::bigint, + 'a source reorg beyond the cursor generation disappears fail-closed' +); + +reset role; + +select * from finish(); + +rollback; diff --git a/tests/data-pipeline/candidate-projector-runtime-binding.test.ts b/tests/data-pipeline/candidate-projector-runtime-binding.test.ts index dcb45ab9..a964ab71 100644 --- a/tests/data-pipeline/candidate-projector-runtime-binding.test.ts +++ b/tests/data-pipeline/candidate-projector-runtime-binding.test.ts @@ -280,6 +280,12 @@ describe("candidate projector runtime binding", () => { if (text.includes("verify_candidate_database_promoted_v2")) { return [{ verified: true }]; } + if (text.includes("pg_control_system")) { + return [{ + database_name: "postgres", + system_identifier: "72623859790382856", + }]; + } return []; }); const executor = { @@ -290,7 +296,10 @@ describe("candidate projector runtime binding", () => { await expect(assertCandidateDatabasePromotedState({ executor, binding: selection.promotedDatabase, - })).resolves.toBeUndefined(); + })).resolves.toEqual({ + databaseName: "postgres", + systemIdentifier: "72623859790382856", + }); expect(query).toHaveBeenCalledWith( expect.stringContaining("verify_candidate_database_promoted_v2"), expect.arrayContaining([ @@ -300,6 +309,60 @@ describe("candidate projector runtime binding", () => { "dpl_12345678901234567890", ]), ); + expect(query).toHaveBeenCalledWith( + expect.stringContaining("pg_catalog.pg_control_system()"), + ); + }); + + it.each([ + ["missing", []], + ["multiple", [ + { database_name: "postgres", system_identifier: "72623859790382856" }, + { database_name: "postgres", system_identifier: "72623859790382856" }, + ]], + ["malformed database", [ + { database_name: "postgres;select", system_identifier: "72623859790382856" }, + ]], + ["zero system identifier", [ + { database_name: "postgres", system_identifier: "0" }, + ]], + ["oversized system identifier", [ + { database_name: "postgres", system_identifier: "18446744073709551616" }, + ]], + ["non-text system identifier", [ + { database_name: "postgres", system_identifier: 72623859790382856n }, + ]], + ])("rejects a %s physical database identity", async (_label, identityRows) => { + const selection = selectProjectorRuntimeBinding({ + env: promotedReleaseEnvironment(), + canonicalBinding: canonicalCandidateBinding(), + }); + if (!selection.promotedDatabase) throw new Error("missing promotion proof"); + const query = vi.fn(async (text: string) => { + if (text.includes("current_role::text")) { + return [{ + session_user: "programmable_projector_login", + current_role: "programmable_projector", + }]; + } + if (text === "select session_user::text as session_user") { + return [{ session_user: "programmable_projector_login" }]; + } + if (text.includes("verify_candidate_database_promoted_v2")) { + return [{ verified: true }]; + } + if (text.includes("pg_control_system")) return identityRows; + return []; + }); + const executor = { + transaction: vi.fn(async (work) => work({ query })), + close: vi.fn(), + } as never; + + await expect(assertCandidateDatabasePromotedState({ + executor, + binding: selection.promotedDatabase, + })).rejects.toThrow(); }); it("validates real Envio progress against the selected candidate identity", async () => { diff --git a/tests/data-pipeline/operations-health-route.test.ts b/tests/data-pipeline/operations-health-route.test.ts new file mode 100644 index 00000000..1227a9f3 --- /dev/null +++ b/tests/data-pipeline/operations-health-route.test.ts @@ -0,0 +1,198 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + readIndexedReadModelHealth: vi.fn(), + getOperationalOnchainDeployment: vi.fn(), + readDurableExploreModel: vi.fn(), + readIndependentRpcHealth: vi.fn(), +})); + +vi.mock("../../lib/data-pipeline/read-model-health.server", () => ({ + readIndexedReadModelHealth: mocks.readIndexedReadModelHealth, +})); + +vi.mock("../../lib/onchain", () => ({ + getOperationalOnchainDeployment: mocks.getOperationalOnchainDeployment, + readDurableExploreModel: mocks.readDurableExploreModel, + readIndependentRpcHealth: mocks.readIndependentRpcHealth, +})); + +import { GET } from "../../app/api/ops/health/route"; + +describe("operations health route", () => { + beforeEach(() => { + Object.values(mocks).forEach((mock) => mock.mockReset()); + }); + + it("preserves the complete legacy health behavior in legacy-only mode", async () => { + mocks.readIndexedReadModelHealth.mockResolvedValue(null); + mocks.getOperationalOnchainDeployment.mockReturnValue({ status: "ready" }); + mocks.readDurableExploreModel.mockResolvedValue({ + status: "ready", + ageMs: 5_500, + envelope: { + payload: { + model: { + snapshot: { blockNumber: "25600000" }, + tokens: [{}, {}], + }, + }, + }, + }); + mocks.readIndependentRpcHealth.mockResolvedValue({ + chainId: 1, + heads: ["25600012", "25600012"], + confirmedBlock: { + number: "25600000", + hash: `0x${"11".repeat(32)}`, + }, + }); + + const response = await GET(); + const body = await response.json(); + + expect(response.status).toBe(200); + expect(response.headers.get("cache-control")).toBe( + "public, max-age=0, s-maxage=30", + ); + expect(body).toMatchObject({ + status: "healthy", + chainId: 1, + index: { + ageSeconds: 5, + blockNumber: "25600000", + tokenCount: 2, + }, + }); + expect(mocks.readDurableExploreModel).toHaveBeenCalledOnce(); + expect(mocks.readIndependentRpcHealth).toHaveBeenCalledOnce(); + }); + + it("binds indexed health to the production deployment and independent RPC chain", async () => { + mocks.readIndexedReadModelHealth.mockResolvedValue({ + chainId: 1, + index: { + ageSeconds: 4, + blockNumber: "25600010", + tokenCount: 281, + }, + }); + mocks.getOperationalOnchainDeployment.mockReturnValue({ + status: "ready", + chainId: 1, + }); + mocks.readIndependentRpcHealth.mockResolvedValue({ + chainId: 1, + heads: ["25600012", "25600012"], + confirmedBlock: { + number: "25600010", + hash: `0x${"22".repeat(32)}`, + }, + }); + + const response = await GET(); + const body = await response.json(); + + expect(response.status).toBe(200); + expect(response.headers.get("cache-control")).toBe( + "public, max-age=0, s-maxage=30", + ); + expect(body).toMatchObject({ + status: "healthy", + chainId: 1, + index: { + ageSeconds: 4, + blockNumber: "25600010", + tokenCount: 281, + }, + rpc: { + heads: ["25600012", "25600012"], + }, + }); + expect(mocks.getOperationalOnchainDeployment).toHaveBeenCalledWith( + "production", + ); + expect(mocks.readDurableExploreModel).not.toHaveBeenCalled(); + expect(mocks.readIndependentRpcHealth).toHaveBeenCalledOnce(); + }); + + it("fails closed when indexed health and the deployment chain differ", async () => { + vi.spyOn(console, "error").mockImplementation(() => undefined); + mocks.readIndexedReadModelHealth.mockResolvedValue({ + chainId: 1, + index: { + ageSeconds: 4, + blockNumber: "25600010", + tokenCount: 281, + }, + }); + mocks.getOperationalOnchainDeployment.mockReturnValue({ + status: "ready", + chainId: 11_155_111, + }); + + const response = await GET(); + const body = await response.json(); + + expect(response.status).toBe(503); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(body).toMatchObject({ status: "unhealthy" }); + expect(JSON.stringify(body)).not.toContain("chain binding"); + expect(mocks.readIndependentRpcHealth).not.toHaveBeenCalled(); + expect(mocks.readDurableExploreModel).not.toHaveBeenCalled(); + }); + + it("fails closed when the independent RPC chain differs", async () => { + vi.spyOn(console, "error").mockImplementation(() => undefined); + mocks.readIndexedReadModelHealth.mockResolvedValue({ + chainId: 1, + index: { + ageSeconds: 4, + blockNumber: "25600010", + tokenCount: 281, + }, + }); + mocks.getOperationalOnchainDeployment.mockReturnValue({ + status: "ready", + chainId: 1, + }); + mocks.readIndependentRpcHealth.mockResolvedValue({ + chainId: 11_155_111, + heads: ["25600012", "25600012"], + confirmedBlock: { + number: "25600010", + hash: `0x${"22".repeat(32)}`, + }, + }); + + const response = await GET(); + const body = await response.json(); + + expect(response.status).toBe(503); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(body).toMatchObject({ status: "unhealthy" }); + expect(JSON.stringify(body)).not.toContain("chain binding"); + expect(mocks.readIndependentRpcHealth).toHaveBeenCalledOnce(); + expect(mocks.readDurableExploreModel).not.toHaveBeenCalled(); + }); + + it("fails closed without details when indexed health validation fails", async () => { + vi.spyOn(console, "error").mockImplementation(() => undefined); + mocks.readIndexedReadModelHealth.mockRejectedValue( + new Error("private database detail"), + ); + + const response = await GET(); + const body = await response.json(); + + expect(response.status).toBe(503); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(body).toMatchObject({ + status: "unhealthy", + }); + expect(JSON.stringify(body)).not.toContain( + "private database detail", + ); + expect(mocks.getOperationalOnchainDeployment).not.toHaveBeenCalled(); + }); +}); diff --git a/tests/data-pipeline/read-model-health.test.ts b/tests/data-pipeline/read-model-health.test.ts new file mode 100644 index 00000000..38b3efec --- /dev/null +++ b/tests/data-pipeline/read-model-health.test.ts @@ -0,0 +1,354 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("server-only", () => ({})); + +import { INDEXED_ROUTE_FLAG_NAMES } from "../../lib/data-pipeline/config"; +import { + readIndexedReadModelHealth, + resetReadModelHealthForTests, +} from "../../lib/data-pipeline/read-model-health.server"; + +const NOW = Date.parse("2026-08-03T04:00:00.000Z"); +const PHYSICAL_IDENTITY: Readonly<{ + databaseName: string; + systemIdentifier: string; +}> = Object.freeze({ + databaseName: "postgres", + systemIdentifier: "7666007964130682852", +}); +const RELEASE_SCOPES = Object.freeze([ + Object.freeze({ releaseId: "classic-v2", modelId: "classic", sourceGroup: "core" }), + Object.freeze({ releaseId: "classic-v3", modelId: "classic", sourceGroup: "core" }), + Object.freeze({ releaseId: "stock-paired-v1", modelId: "stock-paired", sourceGroup: "core" }), + Object.freeze({ releaseId: "stock-paired-v2", modelId: "stock-paired", sourceGroup: "core" }), + Object.freeze({ releaseId: "stock-paired-v3", modelId: "stock-paired", sourceGroup: "core" }), +]); + +function indexedEnvironment() { + return Object.fromEntries( + INDEXED_ROUTE_FLAG_NAMES.map((name) => [name, "true"]), + ); +} + +function checkpointRows(createdAt = "2026-08-03T03:59:00.000Z") { + return RELEASE_SCOPES.map((scope, index) => ({ + chain_id: "1", + release_id: scope.releaseId, + model_id: scope.modelId, + source_group: scope.sourceGroup, + block_number: String(25_600_000 + index), + created_at: createdAt, + })); +} + +function marketRows(options: Readonly<{ + cursorAdvancedAt?: string; + sourceBlockOffset?: number; + sourceCreatedAt?: string; +}> = {}) { + return RELEASE_SCOPES.map((scope, index) => { + const blockNumber = String(25_600_000 + index); + const sourceBlockNumber = String( + 25_600_000 + index + (options.sourceBlockOffset ?? 0), + ); + const blockHash = `0x${String(index + 1).padStart(64, "0")}`; + const epochId = `00000000-0000-4000-8000-${String(index + 1).padStart(12, "0")}`; + return { + chain_id: "1", + release_id: scope.releaseId, + model_id: scope.modelId, + source_group: scope.sourceGroup, + market_projector_version: "market-projector-v1", + pool_id: `0x${String(index + 11).padStart(64, "0")}`, + cursor_epoch_id: epochId, + cursor_pointer_generation: "1", + cursor_generation: "2", + cursor_reorg_generation: "0", + cursor_source_reorg_generation: "0", + cursor_block_number: blockNumber, + cursor_block_hash: blockHash, + cursor_advanced_at: + options.cursorAdvancedAt ?? "2026-08-03T03:59:00.000Z", + hour_coverage_end: "2026-08-03T04:00:00.000Z", + day_coverage_end: "2026-08-03T00:00:00.000Z", + source_projector_version: "projector-v1", + source_checkpoint_epoch_id: epochId, + source_checkpoint_pointer_generation: "1", + source_checkpoint_generation: "3", + source_checkpoint_reorg_generation: "0", + source_checkpoint_block_number: sourceBlockNumber, + source_checkpoint_block_hash: blockHash, + source_checkpoint_cursor_block_global_log_index: "4294967295", + source_checkpoint_cursor_candidate_id: "empty-page", + source_checkpoint_created_at: + options.sourceCreatedAt ?? "2026-08-03T03:59:00.000Z", + latest_snapshot_block_number: blockNumber, + latest_snapshot_observed_at: "2026-08-03T03:59:00.000Z", + latest_snapshot_attached_at: "2026-08-03T03:59:00.000Z", + latest_snapshot_reconciled_at: "2026-08-03T03:59:00.000Z", + }; + }); +} + +function fixture(options: Readonly<{ + projectorIdentity?: typeof PHYSICAL_IDENTITY; + apiIdentity?: typeof PHYSICAL_IDENTITY; + bindingError?: Error; + checkpointRows?: readonly Record[]; + marketRows?: readonly Record[]; + circuitState?: "closed" | "open"; + readinessParity?: "current" | "pending" | "stale" | "mismatch" | "missing"; +}> = {}) { + const close = vi.fn(async () => undefined); + const query = vi.fn(async (text: string) => { + if (text.includes("pg_control_system")) { + const identity = options.apiIdentity ?? PHYSICAL_IDENTITY; + return [{ + database_name: identity.databaseName, + system_identifier: identity.systemIdentifier, + }]; + } + if (text.includes("checkpoint_summary_v1")) { + return options.checkpointRows ?? checkpointRows(); + } + if (text.includes("health_summary_v1")) { + return options.circuitState + ? [{ dependency: "postgres", circuit_status: options.circuitState }] + : []; + } + if (text.includes("launch_by_token_v2")) { + return marketRows().map((row) => ({ + release_id: row.release_id, + model_id: row.model_id, + source_group: row.source_group, + pool_id: row.pool_id, + })); + } + if (text.includes("market_projector_health_v1")) { + return options.marketRows ?? marketRows(); + } + throw new Error(`unexpected query: ${text}`); + }); + const assertCandidateDatabasePromotedState = options.bindingError + ? vi.fn(async () => { + throw options.bindingError; + }) + : vi.fn(async () => options.projectorIdentity ?? PHYSICAL_IDENTITY); + const readExactRouteSnapshotReadiness = vi.fn(async (input: { + scope: readonly Readonly<{ model: string; releaseVersion: string }>[]; + }) => ({ + readiness: input.scope.map((scope, index) => ({ + ...scope, + eligibility: "eligible" as const, + parity: options.readinessParity ?? ("current" as const), + version: { + checkpointId: `00000000-0000-4000-8000-${String(index + 20).padStart(12, "0")}`, + sourceGroup: "core", + projectorVersion: "projector-v1", + epochId: `00000000-0000-4000-8000-${String(index + 30).padStart(12, "0")}`, + pointerGeneration: "1", + checkpointGeneration: "1", + reorgGeneration: "0", + blockNumber: "25600000", + blockHash: `0x${"11".repeat(32)}`, + }, + })), + })); + const dependencies = { + getServerReadModel: vi.fn(async () => ({ + repeatableReadSnapshot: async (work: (transaction: { query: typeof query }) => Promise) => + work({ query }), + })), + loadProjectorRuntimeConfig: vi.fn(() => ({ + binding: { + mode: "release", + candidate: null, + promotedDatabase: { + providerDeploymentId: "d08b62a6-74fb-5e0a-a698-dc6877150db4", + deploymentCommitment: `0x${"11".repeat(32)}`, + schemaCommitment: `0x${"22".repeat(32)}`, + initializationInputCommitment: `0x${"33".repeat(32)}`, + initializedAt: "2026-08-01T09:00:00.000Z", + productCommit: "a".repeat(40), + stagedDeploymentId: "dpl_aaaaaaaaaaaaaaaaaaaaaaaa", + }, + }, + database: { + projectorConnectionString: + "postgresql://programmable_projector_login:password@127.0.0.1:5432/postgres?sslmode=disable", + runtimeConnectionString: + "postgresql://programmable_projector_runtime_login:password@127.0.0.1:5432/postgres?sslmode=disable", + sslCaPem: "unused-in-loopback-test", + }, + releaseScopes: RELEASE_SCOPES, + })), + createPostgresExecutor: vi.fn(() => ({ close })), + assertCandidateDatabasePromotedState, + readExactRouteSnapshotReadiness, + nowMs: vi.fn(() => NOW), + }; + return { + dependencies, + close, + query, + assertCandidateDatabasePromotedState, + readExactRouteSnapshotReadiness, + }; +} + +describe("indexed read-model operations health", () => { + beforeEach(() => resetReadModelHealthForTests()); + + it("preserves legacy health without parsing unrelated pipeline configuration", async () => { + const input = fixture(); + await expect(readIndexedReadModelHealth( + { PROGRAMMABLE_ENVIO_GRAPHQL_URL: "not-a-url" }, + input.dependencies as never, + )).resolves.toBeNull(); + expect(input.dependencies.loadProjectorRuntimeConfig).not.toHaveBeenCalled(); + expect(input.dependencies.getServerReadModel).not.toHaveBeenCalled(); + expect(input.dependencies.createPostgresExecutor).not.toHaveBeenCalled(); + }); + + it("treats partial indexed activation as a Postgres-backed rollout", async () => { + const input = fixture(); + await expect(readIndexedReadModelHealth( + { [INDEXED_ROUTE_FLAG_NAMES[0]]: "true" }, + input.dependencies as never, + )).resolves.toMatchObject({ chainId: 1 }); + expect(input.dependencies.loadProjectorRuntimeConfig).toHaveBeenCalledOnce(); + }); + + it("rejects non-canonical activation values before opening a database", async () => { + const input = fixture(); + await expect(readIndexedReadModelHealth( + { [INDEXED_ROUTE_FLAG_NAMES[0]]: "1" }, + input.dependencies as never, + )).rejects.toThrow("Indexed read-model health is unavailable"); + expect(input.dependencies.loadProjectorRuntimeConfig).not.toHaveBeenCalled(); + }); + + it("revalidates the promoted physical database and current routes on each health request", async () => { + const input = fixture({ circuitState: "closed" }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + input.dependencies as never, + )).resolves.toEqual({ + chainId: 1, + index: { ageSeconds: 60, blockNumber: "25600000", tokenCount: 5 }, + }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + input.dependencies as never, + )).resolves.toMatchObject({ chainId: 1 }); + expect(input.assertCandidateDatabasePromotedState).toHaveBeenCalledTimes(2); + expect(input.dependencies.createPostgresExecutor).toHaveBeenCalledTimes(2); + expect(input.close).toHaveBeenCalledTimes(2); + expect(input.readExactRouteSnapshotReadiness).toHaveBeenCalledTimes(12); + }); + + it("closes and retries after a failed immutable promotion binding", async () => { + const input = fixture({ bindingError: new Error("not promoted") }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + input.dependencies as never, + )).rejects.toThrow("not promoted"); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + input.dependencies as never, + )).rejects.toThrow("not promoted"); + expect(input.assertCandidateDatabasePromotedState).toHaveBeenCalledTimes(2); + expect(input.close).toHaveBeenCalledTimes(2); + }); + + it("rejects a promoted projector and API reader on different clusters", async () => { + const input = fixture({ + apiIdentity: { + databaseName: "postgres", + systemIdentifier: "7666007964130682853", + }, + }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + input.dependencies as never, + )).rejects.toThrow("Indexed read-model health is unavailable"); + }); + + it("rejects non-current route parity and an open circuit", async () => { + const parity = fixture({ readinessParity: "stale" }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + parity.dependencies as never, + )).rejects.toThrow("Indexed read-model health is unavailable"); + + resetReadModelHealthForTests(); + const circuit = fixture({ circuitState: "open" }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + circuit.dependencies as never, + )).rejects.toThrow("Indexed read-model health is unavailable"); + }); + + it("rejects missing market-projector coverage", async () => { + const missing = fixture({ marketRows: [] }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + missing.dependencies as never, + )).rejects.toThrow("Indexed read-model health is unavailable"); + + }); + + it("accepts an inactive caught-up market cursor after empty source checkpoints", async () => { + const input = fixture({ + marketRows: marketRows({ + cursorAdvancedAt: "2026-08-01T04:00:00.000Z", + sourceBlockOffset: 500, + }), + }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + input.dependencies as never, + )).resolves.toMatchObject({ chainId: 1 }); + }); + + it("accepts null candle coverage for an inactive pool with no completed trades", async () => { + const rows = marketRows().map((row, index) => index === 0 ? { + ...row, + hour_coverage_end: null, + day_coverage_end: null, + } : row); + const input = fixture({ marketRows: rows }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + input.dependencies as never, + )).resolves.toMatchObject({ chainId: 1 }); + }); + + it("treats shadow comparison as a Postgres-backed health mode", async () => { + const input = fixture(); + await expect(readIndexedReadModelHealth( + { INDEXED_READ_SHADOW_COMPARE_ENABLED: "true" }, + input.dependencies as never, + )).resolves.toMatchObject({ chainId: 1 }); + expect(input.dependencies.loadProjectorRuntimeConfig).toHaveBeenCalledOnce(); + }); + + it("rejects stale or implausibly future-dated source checkpoints", async () => { + const stale = fixture({ + checkpointRows: checkpointRows("2026-08-03T03:44:59.999Z"), + }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + stale.dependencies as never, + )).rejects.toThrow("Indexed read-model health is unavailable"); + + resetReadModelHealthForTests(); + const future = fixture({ + checkpointRows: checkpointRows("2026-08-03T04:01:00.001Z"), + }); + await expect(readIndexedReadModelHealth( + indexedEnvironment(), + future.dependencies as never, + )).rejects.toThrow("Indexed read-model health is unavailable"); + }); +});