diff --git a/src/selfhost/backend-contracts.ts b/src/selfhost/backend-contracts.ts new file mode 100644 index 0000000000..3d65c6dd5e --- /dev/null +++ b/src/selfhost/backend-contracts.ts @@ -0,0 +1,132 @@ +// Shared structural contracts for the self-host runtime's three swappable-backend pairs (#4010): the durable +// job queue (sqlite-queue.ts / pg-queue.ts), the D1-shaped storage adapter (d1-adapter.ts / pg-adapter.ts), +// and the Vectorize-shaped RAG vector store (vectorize.ts / qdrant-vectorize.ts / pg-vectorize.ts). Before +// this file, each pair either declared its own independent interface reconciled only by a loose +// `T | Promise` union at the call site (the queue pair -- see DurableQueue below), or force-cast a bare +// object literal straight to one of Cloudflare's ambient bindings (D1Database, Vectorize -- both +// `declare abstract class` in worker-configuration.d.ts, so a plain object literal can only ever satisfy them +// via `as unknown as X`; there is no way to avoid that final cast) with nothing checking the literal's OWN +// shape first, so either side could silently drift from its sibling with nothing to catch it. +// +// Every interface here is the actual subset each pair's concrete implementations already satisfy. Each +// backend module assigns its returned object to one of these types BEFORE the unavoidable ambient-binding +// cast, so a future change that breaks parity between two (or three) implementations is a compile error here +// instead of a runtime surprise discovered on whichever backend the change didn't touch. +import type { DeadLetterJob, SelfHostQueueSnapshot } from "./queue-common"; +import type { MaintenancePressureSignals } from "./maintenance-admission"; +import type { BacklogRepoCount } from "./queue-fairness"; + +// ── Queue pair (sqlite-queue.ts createSqliteQueue / pg-queue.ts createPgQueue) ────────────────────────────── +// Previously two independently-declared interfaces (DurableQueue, PgDurableQueue): every method that was +// synchronous on the sqlite side was Promise-wrapped on the postgres side, and PgDurableQueue alone had an +// extra `init()`. Unified here as a strict superset of the (former) sqlite shape -- every method returns a +// Promise, since node:sqlite's synchronous calls await trivially, whereas making the postgres side +// synchronous is not possible for a real network client. +export interface DurableQueue { + binding: Queue; + /** One-time async setup (schema DDL, column backfills, crash recovery, startup jitter, the foreground- + * liveness self-heal) that MUST complete before `start()`/`binding.send()` are used. The postgres backend + * genuinely awaits `pool.query(...)` for its schema DDL here. The sqlite backend performs the equivalent + * setup SYNCHRONOUSLY inside `createSqliteQueue()` itself (node:sqlite has no connection to await), so by + * the time that factory returns, setup is already done -- its `init()` is a no-op that resolves + * immediately. Callers that treat both backends uniformly (`await createXQueue(...).init()`) get correct + * behavior either way. */ + init(): Promise; + start(): void; + stop(): Promise; + drain(): Promise; + size(): Promise; + deadCount(): Promise; + /** Jobs currently claimed and mid-flight (status='processing') -- distinct from size(), which also + * includes still-pending work. See #selfhost-queue-liveness's own observability additions. */ + processingCount(): Promise; + stats(): Promise>; + snapshot(): Promise; + /** Live-vs-maintenance queue pressure, for the /metrics gauges (see server.ts) -- the SAME signals the + * maintenance-admission policy itself consults at claim time. */ + pressureSignals(): Promise; + /** Requeues dead-lettered jobs still under the auto-retry attempts ceiling. Called on a timer while + * running (see start()), and exposed directly so tests and an operator-triggered repair path don't have + * to wait for the real interval. Returns the number of jobs revived. */ + reviveDeadLetterJobs(): Promise; + /** Foreground-liveness invariant (#selfhost-queue-liveness): pulls back any FOREGROUND-priority pending job + * whose deferral has gone stale (see foreground-liveness.ts) regardless of what deferred it. Called once at + * boot and on a timer while running, and exposed directly so tests and an operator-triggered repair path + * don't have to wait for the real interval. Returns the number released. */ + releaseStaleForegroundDeferrals(): Promise; + /** Top-N repos by backlog-convergence pending depth, for the observability dashboard's per-repo backlog panel + * (#selfhost-lane-observability). */ + topBacklogRepos(limit: number): Promise; + /** Paginated dead-letter rows, newest-death-first, for the DLQ dashboard table (#2214). Also mirrored onto + * `binding` (see queue-common.ts's SelfHostQueueDeadLetterAdmin) so Hono routes can reach it via env.JOBS. */ + listDeadLetterJobs(limit: number, offset: number): Promise; + /** Manual, operator-initiated replay of ONE dead job with a FRESH retry budget (#2215) -- unlike the automatic + * reviveDeadLetterJobs() sweep above, which deliberately preserves `attempts` under a ceiling. */ + replayDeadLetterJob(id: number): Promise; + /** Manual, operator-initiated permanent delete of ONE dead job (#2215). */ + deleteDeadLetterJob(id: number): Promise; + /** Manual, operator-initiated permanent delete of EVERY dead job (#2215). */ + purgeDeadLetterJobs(): Promise; +} + +// ── Storage adapter pair (d1-adapter.ts createD1Adapter / pg-adapter.ts createPgAdapter) ──────────────────── +// The actual subset of Cloudflare's ambient D1Database both adapters really satisfy -- notably NOT +// `withSession()`, which neither implements (self-host is single-primary; there is no replica to anchor a +// session against). `run()` is typed WITHOUT a `results` field (matching pg-adapter.ts's own D1Response- +// faithful shape and the real D1 contract, where a non-SELECT statement carries no meaningful results); +// d1-adapter.ts's own `run()` happens to return a wider object that also carries `results`, which remains +// assignable here since an implementation may always return MORE than an interface requires. +export interface SelfHostD1PreparedStatement { + bind(...values: unknown[]): SelfHostD1PreparedStatement; + all(): Promise<{ results: T[]; success: true; meta: Record }>; + first(colName?: string): Promise; + run(): Promise<{ success: true; meta: Record }>; + raw(): Promise; +} + +export interface SelfHostD1Database { + prepare(query: string): SelfHostD1PreparedStatement; + batch(statements: SelfHostD1PreparedStatement[]): Promise }>>; + exec(query: string): Promise<{ count: number; duration: number }>; + /** @deprecated present only because both self-host adapters still implement it for D1 surface completeness; + * real D1 no longer uses it either (see d1-adapter.ts / pg-adapter.ts). */ + dump(): Promise; +} + +// ── Vectorize pair (vectorize.ts createSqliteVectorize / qdrant-vectorize.ts createQdrantVectorize / +// pg-vectorize.ts createPgVectorize) ────────────────────────────────────────────────────────────────────── +// Each backend previously redeclared its own private copy of these three shapes, and only vectorize.ts's +// QueryOptions carried `returnMetadata` -- a real divergence, not a stylistic one: every backend is invoked +// through the SAME reviewVectorAdapter → vectorize.query(vector, opts) call path (src/review/adapters.ts), +// and src/review/rag.ts's own opts ALWAYS includes `returnMetadata: "all"` regardless of which backend is +// bound to env.VECTORIZE. So every backend genuinely receives this option today; the field belongs on all +// three, not none. None of the three currently branches on it -- each already returns whatever metadata it +// has stored for a match regardless of the requested retrieval level, which is a conservative behavior +// already compatible with "all" -- so adding it to the other two is a type-honesty fix (the type now matches +// what the function is actually called with), not a behavior change. The type is tightened from +// vectorize.ts's previous loose `string` to the real three-value union Cloudflare's own +// VectorizeQueryOptions.returnMetadata uses (worker-configuration.d.ts). +export interface SelfHostVectorRecord { + id: string; + values: number[]; + namespace?: string; + metadata?: Record; +} + +export interface SelfHostVectorizeQueryOptions { + topK?: number; + namespace?: string; + returnMetadata?: "all" | "none" | "indexed"; +} + +export interface SelfHostVectorizeMatch { + id: string; + score: number; + metadata?: Record; +} + +export interface SelfHostVectorize { + upsert(vectors: SelfHostVectorRecord[]): Promise<{ count: number; ids: string[] }>; + query(vector: number[], opts: SelfHostVectorizeQueryOptions): Promise<{ matches: SelfHostVectorizeMatch[] }>; + deleteByIds(ids: string[]): Promise<{ count: number }>; +} diff --git a/src/selfhost/d1-adapter.ts b/src/selfhost/d1-adapter.ts index 8f77ed3247..4f0d9d2caa 100644 --- a/src/selfhost/d1-adapter.ts +++ b/src/selfhost/d1-adapter.ts @@ -6,6 +6,14 @@ // D1's API is async; the SQLite drivers are sync — sync calls are wrapped in resolved Promises. The driver is // INJECTED behind the tiny SqliteDriver interface, so this module has no hard SQLite dependency and the // Cloudflare Worker bundle never imports it. Default driver: node:sqlite (built into Node, no native build). +// +// `Statement` implements the shared `SelfHostD1PreparedStatement` contract (backend-contracts.ts, #4010) -- +// the same one pg-adapter.ts's `PgStatement` implements -- and `createD1Adapter`'s own return value is typed +// `SelfHostD1Database` before the final `as unknown as D1Database` cast (D1Database is a `declare abstract +// class`, so a plain object can only ever satisfy it via that cast; there is no way to avoid it). That +// intermediate typed step is what's new: previously nothing checked this module's own shape against its +// Postgres sibling's before the cast erased everything to `unknown`. +import type { SelfHostD1Database, SelfHostD1PreparedStatement } from "./backend-contracts"; /** A uniform sync SQLite primitive both node:sqlite and better-sqlite3 can satisfy via a thin wrapper. `query` * ALWAYS returns rows (empty for a write) + the write metadata, so the adapter needs no reader-detection. */ @@ -20,7 +28,7 @@ function meta(changes = 0, lastRowId = 0): Record { /** One prepared (and optionally bound) statement. bind() returns a fresh instance (D1 statements are immutable * after bind). The SQLite statement is compiled per execution (drivers cache by SQL text). */ -class Statement { +class Statement implements SelfHostD1PreparedStatement { constructor( private readonly driver: SqliteDriver, private readonly sql: string, @@ -32,18 +40,18 @@ class Statement { } /** Sync core used by all()/run() (async wrappers) and batch() (inside a transaction). */ - execSync(): { results: unknown[]; success: boolean; meta: Record } { + execSync(): { results: unknown[]; success: true; meta: Record } { const r = this.driver.query(this.sql, this.values); return { results: r.rows, success: true, meta: meta(r.changes, r.lastInsertRowid) }; } - async all(): Promise<{ results: T[]; success: boolean; meta: Record }> { - return this.execSync() as { results: T[]; success: boolean; meta: Record }; + async all(): Promise<{ results: T[]; success: true; meta: Record }> { + return this.execSync() as { results: T[]; success: true; meta: Record }; } // D1's run() returns the same {results, meta} shape (results empty for a non-returning write). - async run(): Promise<{ results: T[]; success: boolean; meta: Record }> { - return this.execSync() as { results: T[]; success: boolean; meta: Record }; + async run(): Promise<{ results: T[]; success: true; meta: Record }> { + return this.execSync() as { results: T[]; success: true; meta: Record }; } async first(colName?: string): Promise { @@ -60,16 +68,15 @@ class Statement { /** Wrap a synchronous SQLite driver as a D1Database. */ export function createD1Adapter(driver: SqliteDriver): D1Database { - const adapter = { + const adapter: SelfHostD1Database = { prepare(sql: string) { return new Statement(driver, sql); }, - async batch(statements: unknown[]) { + async batch(statements: Statement[]) { // D1 runs a batch atomically, one result per statement, in order. - const list = statements as Statement[]; driver.exec("BEGIN"); try { - const out = list.map((s) => s.execSync()); + const out = statements.map((s) => s.execSync()); driver.exec("COMMIT"); return out; } catch (error) { diff --git a/src/selfhost/pg-adapter.ts b/src/selfhost/pg-adapter.ts index 2fec78743a..446e0a92c5 100644 --- a/src/selfhost/pg-adapter.ts +++ b/src/selfhost/pg-adapter.ts @@ -2,13 +2,19 @@ // app + drizzle-orm/d1 use (prepare/bind/all/first/run/raw + batch + exec), translating each SQLite query to // Postgres (pg-dialect.ts) and running it via node-postgres. A shared Postgres DB makes multi-instance // self-host possible (vs the single-file SQLite default). +// +// `PgStatement` implements the shared `SelfHostD1PreparedStatement` contract (backend-contracts.ts, #4010) -- +// the same one d1-adapter.ts's `Statement` implements -- and `createPgAdapter`'s own return value is typed +// `SelfHostD1Database` before the final `as unknown as D1Database` cast (unavoidable: D1Database is a +// `declare abstract class`, so only that cast can bridge a plain object to it). import type { Pool, PoolClient } from "pg"; +import type { SelfHostD1Database, SelfHostD1PreparedStatement } from "./backend-contracts"; import { translateDdl, translateSql } from "./pg-dialect"; type Row = Record; type Runner = Pool | PoolClient; -class PgStatement { +class PgStatement implements SelfHostD1PreparedStatement { constructor( private readonly pool: Pool, private readonly sql: string, @@ -54,13 +60,13 @@ class PgStatement { } export function createPgAdapter(pool: Pool): D1Database { - const adapter = { + const adapter: SelfHostD1Database = { prepare: (sql: string) => new PgStatement(pool, sql), async batch(statements: PgStatement[]) { const client = await pool.connect(); try { await client.query("BEGIN"); - const out: unknown[] = []; + const out: Array<{ results: Row[]; success: true; meta: Record }> = []; for (const st of statements) out.push(await st.runOn(client)); await client.query("COMMIT"); return out; diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 9b9158d5e3..a59585578f 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -3,6 +3,7 @@ // app instances sharing one Postgres can claim jobs concurrently without double-processing. size()/deadCount() // are async (the metrics gauges accept async samplers). import type { Pool, QueryResult } from "pg"; +import type { DurableQueue } from "./backend-contracts"; import { logAudit, extractPayloadType, extractPayloadContext } from "./audit"; import { incr } from "./metrics"; import { withReviewSpan } from "./tracing"; @@ -202,48 +203,6 @@ CREATE TABLE IF NOT EXISTS ${FAIRNESS_TABLE} ( last_backlog_repo TEXT );`; -export interface PgDurableQueue { - binding: Queue; - init(): Promise; - start(): void; - stop(): Promise; - drain(): Promise; - size(): Promise; - deadCount(): Promise; - /** Jobs currently claimed and mid-flight (status='processing') -- distinct from size(), which also - * includes still-pending work. See #selfhost-queue-liveness's own observability additions. */ - processingCount(): Promise; - stats(): Promise>; - snapshot(): Promise; - /** Live-vs-maintenance queue pressure, for the /metrics gauges (see server.ts) -- the SAME signals the - * maintenance-admission policy itself consults at claim time. */ - pressureSignals(): Promise; - /** Requeues dead-lettered jobs still under the auto-retry attempts ceiling. Called on a timer while - * running (see start()), and exposed directly so tests and an operator-triggered repair path don't have - * to wait for the real interval. Returns the number of jobs revived. */ - reviveDeadLetterJobs(): Promise; - /** Foreground-liveness invariant (#selfhost-queue-liveness): pulls back any FOREGROUND-priority pending job - * whose deferral has gone stale (see foreground-liveness.ts) regardless of what deferred it. Called once at - * boot and on a timer while running (see init()/start()), and exposed directly so tests and an - * operator-triggered repair path don't have to wait for the real interval. Returns the number released. */ - releaseStaleForegroundDeferrals(): Promise; - /** Top-N repos by backlog-convergence pending depth, for the observability dashboard's per-repo backlog panel - * (#selfhost-lane-observability). */ - topBacklogRepos(limit: number): Promise; - /** Paginated dead-letter rows, newest-death-first, for the DLQ dashboard table (#2214). Also mirrored onto - * `binding` (see queue-common.ts's SelfHostQueueDeadLetterAdmin) so Hono routes can reach it via env.JOBS. */ - listDeadLetterJobs(limit: number, offset: number): Promise; - /** Manually requeues ONE dead job by id with a fresh retry budget (#2215). Also mirrored onto `binding` (see - * queue-common.ts's SelfHostQueueDeadLetterAdmin) so Hono routes can reach it via env.JOBS. */ - replayDeadLetterJob(id: number): Promise; - /** Permanently deletes ONE dead job by id (#2215). Also mirrored onto `binding` (see queue-common.ts's - * SelfHostQueueDeadLetterAdmin) so Hono routes can reach it via env.JOBS. */ - deleteDeadLetterJob(id: number): Promise; - /** Permanently deletes EVERY dead job (#2215). Also mirrored onto `binding` (see queue-common.ts's - * SelfHostQueueDeadLetterAdmin) so Hono routes can reach it via env.JOBS. */ - purgeDeadLetterJobs(): Promise; -} - interface JobRow { id: string; payload: string; @@ -275,7 +234,7 @@ export function createPgQueue( pool: Pool, consume: (message: JobMessage) => Promise, opts: PgQueueOptions = {}, -): PgDurableQueue { +): DurableQueue { const maxRetries = opts.maxRetries ?? 5; const pollIntervalMs = opts.pollIntervalMs ?? 1000; const backoff = diff --git a/src/selfhost/pg-vectorize.ts b/src/selfhost/pg-vectorize.ts index 14caca85ad..be8251ae25 100644 --- a/src/selfhost/pg-vectorize.ts +++ b/src/selfhost/pg-vectorize.ts @@ -5,26 +5,23 @@ // // Enable: set DATABASE_URL to a postgres:// URI and use the pgvector/pgvector:pg16 Docker image. The // buildPostgresBackend path in server.ts calls init() at startup then injects this adapter as env.VECTORIZE. +// +// VectorRecord/QueryOptions/Match are the shared backend-contracts.ts types (#4010) also used by vectorize.ts +// and qdrant-vectorize.ts -- see vectorize.ts's own header comment for why QueryOptions' `returnMetadata` +// field belongs here too (this backend receives it identically to the other two through the same +// reviewVectorAdapter call path; it just doesn't branch on it, same as the other two). `adapter` is typed +// `SelfHostVectorize` before the final `as unknown as Vectorize` cast (unavoidable: Vectorize is a `declare +// abstract class`, so only that cast can bridge a plain object to it). import type { Pool } from "pg"; +import type { + SelfHostVectorRecord as VectorRecord, + SelfHostVectorizeQueryOptions as QueryOptions, + SelfHostVectorizeMatch as Match, + SelfHostVectorize, +} from "./backend-contracts"; const TABLE = "_selfhost_vectors"; -interface VectorRecord { - id: string; - values: number[]; - namespace?: string; - metadata?: Record; -} -interface QueryOptions { - topK?: number; - namespace?: string; -} -interface Match { - id: string; - score: number; - metadata?: Record; -} - export async function initPgVectorize(pool: Pool): Promise { await pool.query("CREATE EXTENSION IF NOT EXISTS vector"); await pool.query(` @@ -38,7 +35,7 @@ export async function initPgVectorize(pool: Pool): Promise { } export function createPgVectorize(pool: Pool): Vectorize { - const adapter = { + const adapter: SelfHostVectorize = { async upsert(vectors: VectorRecord[]): Promise<{ count: number; ids: string[] }> { for (const v of vectors) { const embedding = `[${v.values.join(",")}]`; diff --git a/src/selfhost/qdrant-vectorize.ts b/src/selfhost/qdrant-vectorize.ts index 42bae8e168..7c3a8722d9 100644 --- a/src/selfhost/qdrant-vectorize.ts +++ b/src/selfhost/qdrant-vectorize.ts @@ -10,27 +10,25 @@ // // Set QDRANT_API_KEY for deployments that require Bearer token authentication (cloud Qdrant, // production on-prem). Omit for unauthenticated local/dev deployments. +// +// VectorRecord/QueryOptions/Match are the shared backend-contracts.ts types (#4010) also used by +// vectorize.ts and pg-vectorize.ts -- see vectorize.ts's own header comment for why QueryOptions' +// `returnMetadata` field belongs here too (this backend receives it identically to the other two through +// the same reviewVectorAdapter call path; it just doesn't branch on it, same as the other two). `adapter` is +// typed `SelfHostVectorize` before the final `as unknown as Vectorize` cast (unavoidable: Vectorize is a +// `declare abstract class`, so only that cast can bridge a plain object to it). import { createHash } from "node:crypto"; import { incr } from "./metrics"; +import type { + SelfHostVectorRecord as VectorRecord, + SelfHostVectorizeQueryOptions as QueryOptions, + SelfHostVectorizeMatch as Match, + SelfHostVectorize, +} from "./backend-contracts"; const DEFAULT_COLLECTION = "gittensory"; const DEFAULT_DIM = 1024; // bge-m3 / mxbai-embed-large (1024-d); set QDRANT_DIM to override -interface VectorRecord { - id: string; - values: number[]; - namespace?: string; - metadata?: Record; -} -interface QueryOptions { - topK?: number; - namespace?: string; -} -interface Match { - id: string; - score: number; - metadata?: Record; -} interface QdrantSearchResult { result: Array<{ id: string; score: number; payload: Record }>; } @@ -94,7 +92,7 @@ export async function initQdrantCollection( export function createQdrantVectorize(url: string, collection = DEFAULT_COLLECTION): Vectorize { const base = url.replace(/\/+$/, ""); - const adapter = { + const adapter: SelfHostVectorize = { async upsert(vectors: VectorRecord[]): Promise<{ count: number; ids: string[] }> { const points = vectors.map((v) => ({ id: idToUuid(v.id), diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 58279f3258..672af9665b 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -4,6 +4,7 @@ // backing store differs. Single-process model: node:sqlite is synchronous + serial, so claim (SELECT→UPDATE) // is atomic with no row-lock dance. import type { SqliteDriver } from "./d1-adapter"; +import type { DurableQueue } from "./backend-contracts"; import { logAudit, extractPayloadType, extractPayloadContext } from "./audit"; import { incr } from "./metrics"; import { withReviewSpan } from "./tracing"; @@ -118,45 +119,6 @@ CREATE INDEX IF NOT EXISTS ${TABLE}_pending_job_key ON ${TABLE}(job_key, status) const LANE_INDEX_DDL = ` CREATE INDEX IF NOT EXISTS ${TABLE}_lane_claim ON ${TABLE}(status, foreground_lane, run_after);`; -export interface DurableQueue { - binding: Queue; - start(): void; - stop(): Promise; - drain(): Promise; - size(): number; - deadCount(): number; - /** Jobs currently claimed and mid-flight (status='processing') -- distinct from size(), which also - * includes still-pending work. See #selfhost-queue-liveness's own observability additions. */ - processingCount(): number; - stats(): Record; - snapshot(): SelfHostQueueSnapshot; - /** Live-vs-maintenance queue pressure, for the /metrics gauges (see server.ts) -- the SAME signals the - * maintenance-admission policy itself consults at claim time. */ - pressureSignals(): MaintenancePressureSignals; - /** Requeues dead-lettered jobs still under the auto-retry attempts ceiling. Called on a timer while - * running (see start()), and exposed directly so tests and an operator-triggered repair path don't have - * to wait for the real interval. Returns the number of jobs revived. */ - reviveDeadLetterJobs(): number; - /** Foreground-liveness invariant (#selfhost-queue-liveness): pulls back any FOREGROUND-priority pending job - * whose deferral has gone stale (see foreground-liveness.ts) regardless of what deferred it. Called once at - * boot and on a timer while running (see the module-init block/start()), and exposed directly so tests and - * an operator-triggered repair path don't have to wait for the real interval. Returns the number released. */ - releaseStaleForegroundDeferrals(): number; - /** Top-N repos by backlog-convergence pending depth, for the observability dashboard's per-repo backlog panel - * (#selfhost-lane-observability). */ - topBacklogRepos(limit: number): BacklogRepoCount[]; - /** Paginated dead-letter rows, newest-death-first, for the DLQ dashboard table (#2214). Also mirrored onto - * `binding` (see queue-common.ts's SelfHostQueueDeadLetterAdmin) so Hono routes can reach it via env.JOBS. */ - listDeadLetterJobs(limit: number, offset: number): DeadLetterJob[]; - /** Manual, operator-initiated replay of ONE dead job with a FRESH retry budget (#2215) -- unlike the automatic - * reviveDeadLetterJobs() sweep above, which deliberately preserves `attempts` under a ceiling. */ - replayDeadLetterJob(id: number): boolean; - /** Manual, operator-initiated permanent delete of ONE dead job (#2215). */ - deleteDeadLetterJob(id: number): boolean; - /** Manual, operator-initiated permanent delete of EVERY dead job (#2215). */ - purgeDeadLetterJobs(): number; -} - interface JobRow { id: number; payload: string; @@ -319,9 +281,9 @@ export function createSqliteQueue( // foreground-liveness.ts) and logs + records its own metric when it finds work. MUST run after `active`/ // `activeBackground` above are initialized -- a release calls kickAll(), which reads them, and both are // still in the temporal dead zone before this point (#selfhost-queue-liveness-tdz). - releaseStaleForegroundDeferrals(); + void releaseStaleForegroundDeferrals(); - function reviveDeadLetterJobs(): number { + async function reviveDeadLetterJobs(): Promise { const revived = reviveEligibleDeadJobs(driver, maxRetries); if (revived) { recordQueueMetric(driver, "gittensory_jobs_dead_letter_revived_total", revived); @@ -408,7 +370,7 @@ export function createSqliteQueue( * always represented in `eligible` regardless of how large the older-blocked backlog grows, at the same * total worst-case row/admission-check budget as before (still `maxReleasePerSweep * 2` candidates, just * split fairly across both ends of the age spectrum instead of packed entirely into the oldest end). */ - function releaseStaleForegroundDeferrals(): number { + async function releaseStaleForegroundDeferrals(): Promise { if (!foregroundLivenessConfig.enabled) return 0; const now = Date.now(); const candidateLimit = foregroundLivenessConfig.maxReleasePerSweep; @@ -468,9 +430,9 @@ export function createSqliteQueue( * reviveDeadLetterJobsSafely's own rationale: an uncaught exception here would surface as an unhandled * exception and can terminate the process when SENTRY_DSN is unset. A failed sweep just waits for the next * interval, same as a failed poll tick waits for the next poll. */ - function releaseStaleForegroundDeferralsSafely(): void { + async function releaseStaleForegroundDeferralsSafely(): Promise { try { - releaseStaleForegroundDeferrals(); + await releaseStaleForegroundDeferrals(); } catch (error) { console.error( JSON.stringify({ @@ -728,7 +690,7 @@ export function createSqliteQueue( * The COUNT/GROUP BY/ORDER BY/LIMIT run IN SQL (gate review, #selfhost-lane-observability) -- a self-host * install with a large real backlog must never pull every matching job_key into JS on every /metrics scrape * just to throw away all but the top 10; only the final, already-bounded rows ever leave the DB. */ - function topBacklogRepos(limit: number): BacklogRepoCount[] { + async function topBacklogRepos(limit: number): Promise { const { rows } = driver.query( `WITH backlog_rest AS ( SELECT substr(job_key, length(?) + 1) AS rest @@ -750,13 +712,13 @@ export function createSqliteQueue( return (rows as Array<{ repo: string; cnt: number }>).map((row) => ({ repo: row.repo, count: Number(row.cnt) })); } - function deadCount(): number { + async function deadCount(): Promise { return Number( (driver.query(`SELECT COUNT(*) AS c FROM ${TABLE} WHERE status='dead'`, []).rows[0] as { c: number }).c, ); } - function listDeadLetterJobs(limit: number, offset: number): DeadLetterJob[] { + async function listDeadLetterJobs(limit: number, offset: number): Promise { const { rows } = driver.query( `SELECT id, payload, attempts, last_error, created_at, dead_at FROM ${TABLE} @@ -792,7 +754,7 @@ export function createSqliteQueue( /** Manually requeues ONE dead job with a FRESH retry budget (attempts reset to 0) -- see the doc comment on * SelfHostQueueDeadLetterAdmin.replayDeadLetterJob in queue-common.ts for the full rationale. Returns false * if no row with that id is currently dead. */ - function replayDeadLetterJob(id: number): boolean { + async function replayDeadLetterJob(id: number): Promise { const { changes } = driver.query( `UPDATE ${TABLE} SET status='pending', run_after=?, last_error=NULL, dead_at=NULL, attempts=0 WHERE id=? AND status='dead'`, [Date.now(), id], @@ -801,13 +763,13 @@ export function createSqliteQueue( } /** Permanently deletes ONE dead job by id. Returns false if no row with that id is currently dead. */ - function deleteDeadLetterJob(id: number): boolean { + async function deleteDeadLetterJob(id: number): Promise { const { changes } = driver.query(`DELETE FROM ${TABLE} WHERE id=? AND status='dead'`, [id]); return changes > 0; } /** Permanently deletes EVERY dead job. Returns the number of rows deleted. */ - function purgeDeadLetterJobs(): number { + async function purgeDeadLetterJobs(): Promise { const { changes } = driver.query(`DELETE FROM ${TABLE} WHERE status='dead'`, []); return changes; } @@ -1227,7 +1189,7 @@ export function createSqliteQueue( ): Promise { for (const m of messages) enqueue(m.body, m.delaySeconds ?? 0); }, - snapshot() { + async snapshot() { return buildSelfHostQueueSnapshot( driver.query( `SELECT payload, status, run_after FROM ${TABLE} WHERE status IN ('pending','processing','dead')`, @@ -1241,16 +1203,24 @@ export function createSqliteQueue( deleteDeadLetterJob, purgeDeadLetterJobs, } as unknown as Queue & { - snapshot(): SelfHostQueueSnapshot; - deadCount(): number; - listDeadLetterJobs(limit: number, offset: number): DeadLetterJob[]; - replayDeadLetterJob(id: number): boolean; - deleteDeadLetterJob(id: number): boolean; - purgeDeadLetterJobs(): number; + snapshot(): Promise; + deadCount(): Promise; + listDeadLetterJobs(limit: number, offset: number): Promise; + replayDeadLetterJob(id: number): Promise; + deleteDeadLetterJob(id: number): Promise; + purgeDeadLetterJobs(): Promise; }; return { binding, + // Every setup step (DDL, column backfills, crash recovery, startup jitter, the foreground-liveness + // self-heal above) already ran SYNCHRONOUSLY above, inline in createSqliteQueue() itself -- node:sqlite + // has no connection to await, so by the time this object is returned there is nothing left to do. init() + // exists purely so callers that treat both queue backends uniformly (`await createXQueue(...).init()`, + // see server.ts's Postgres branch) get correct behavior regardless of which backend is active (#4010). + async init(): Promise { + /* no-op: see comment above */ + }, start() { if (running) return; running = true; @@ -1267,7 +1237,7 @@ export function createSqliteQueue( deadLetterReviveTimer = setInterval(() => void reviveDeadLetterJobsSafely(), queueDeadLetterReviveIntervalMs()); // Foreground-liveness sweep (#selfhost-queue-liveness): also a separate, slow interval -- see // foreground-liveness.ts for why a per-tick check would busy-loop under sustained rate-limit pressure. - foregroundLivenessTimer = setInterval(releaseStaleForegroundDeferralsSafely, foregroundLivenessConfig.checkIntervalMs); + foregroundLivenessTimer = setInterval(() => void releaseStaleForegroundDeferralsSafely(), foregroundLivenessConfig.checkIntervalMs); }, async stop() { running = false; @@ -1281,7 +1251,7 @@ export function createSqliteQueue( while (active > 0) await new Promise((r) => setTimeout(r, 5)); await pump(); }, - size() { + async size() { return Number( ( driver.query( @@ -1292,7 +1262,7 @@ export function createSqliteQueue( ); }, deadCount, - processingCount() { + async processingCount() { return Number( ( driver.query( @@ -1302,13 +1272,13 @@ export function createSqliteQueue( ).c, ); }, - stats() { + async stats() { return readQueueStats(driver); }, snapshot: binding.snapshot, reviveDeadLetterJobs, releaseStaleForegroundDeferrals, - pressureSignals() { + async pressureSignals() { return maintenancePressureSignals(driver, Date.now()); }, topBacklogRepos, diff --git a/src/selfhost/vectorize.ts b/src/selfhost/vectorize.ts index 8ad8a49a1c..ccc43eb5d5 100644 --- a/src/selfhost/vectorize.ts +++ b/src/selfhost/vectorize.ts @@ -3,7 +3,20 @@ // SQLite table with brute-force cosine similarity. For a repo's worth of chunks (hundreds–few-thousand // vectors per namespace) this is fast enough; namespaces (one per repo) keep each query's candidate set // small. Embeddings come from the OpenAI-compatible AI adapter's /embeddings path (e.g. Ollama bge-m3, 1024-d). +// +// VectorRecord/QueryOptions/Match are the shared backend-contracts.ts types (#4010) also used by +// qdrant-vectorize.ts and pg-vectorize.ts -- this module previously redeclared its own private copies, the +// only one of the three carrying `returnMetadata` (see backend-contracts.ts's SelfHostVectorizeQueryOptions +// doc comment for why that field belongs on all three, not just this one). `adapter` is typed +// `SelfHostVectorize` before the final `as unknown as Vectorize` cast (unavoidable: Vectorize is a `declare +// abstract class`, so only that cast can bridge a plain object to it). import type { SqliteDriver } from "./d1-adapter"; +import type { + SelfHostVectorRecord as VectorRecord, + SelfHostVectorizeQueryOptions as QueryOptions, + SelfHostVectorizeMatch as Match, + SelfHostVectorize, +} from "./backend-contracts"; const TABLE = "_selfhost_vectors"; const DDL = ` @@ -15,23 +28,6 @@ CREATE TABLE IF NOT EXISTS ${TABLE} ( ); CREATE INDEX IF NOT EXISTS ${TABLE}_ns ON ${TABLE}(namespace);`; -interface VectorRecord { - id: string; - values: number[]; - namespace?: string; - metadata?: Record; -} -interface QueryOptions { - topK?: number; - namespace?: string; - returnMetadata?: string; -} -interface Match { - id: string; - score: number; - metadata?: Record; -} - export function cosineSimilarity(a: number[], b: number[]): number { let dot = 0; let na = 0; @@ -50,7 +46,7 @@ export function cosineSimilarity(a: number[], b: number[]): number { export function createSqliteVectorize(driver: SqliteDriver): Vectorize { driver.exec(DDL); - const adapter = { + const adapter: SelfHostVectorize = { async upsert(vectors: VectorRecord[]): Promise<{ count: number; ids: string[] }> { for (const v of vectors) { driver.query( diff --git a/src/server.ts b/src/server.ts index 44198a8f90..bd40343845 100644 --- a/src/server.ts +++ b/src/server.ts @@ -58,9 +58,8 @@ import { runSelfHostMigrations } from "./selfhost/migrate"; import { createPgAdapter, tuneGithubRateLimitObservationsAutovacuum } from "./selfhost/pg-adapter"; import { createPgQueue } from "./selfhost/pg-queue"; import { createPgVectorize, initPgVectorize } from "./selfhost/pg-vectorize"; -import { resolvePostgresPoolMax, type SelfHostQueueSnapshot } from "./selfhost/queue-common"; -import type { BacklogRepoCount } from "./selfhost/queue-fairness"; -import type { MaintenancePressureSignals } from "./selfhost/maintenance-admission"; +import { resolvePostgresPoolMax } from "./selfhost/queue-common"; +import type { DurableQueue } from "./selfhost/backend-contracts"; import { createSqliteQueue } from "./selfhost/sqlite-queue"; import { createSqliteVectorize } from "./selfhost/vectorize"; import { createFsBlobStore } from "./selfhost/blob-store"; @@ -129,18 +128,11 @@ function loadFileSecrets(): void { interface Backend { db: D1Database; - queue: { - binding: Queue; - start(): void; - stop(): Promise; - size(): number | Promise; - deadCount(): number | Promise; - processingCount(): number | Promise; - stats(): Record | Promise>; - pressureSignals(): MaintenancePressureSignals | Promise; - snapshot(): SelfHostQueueSnapshot | Promise; - topBacklogRepos(limit: number): BacklogRepoCount[] | Promise; - }; + // Unified DurableQueue (backend-contracts.ts, #4010) -- previously an inline type papering over the sqlite + // and Postgres queue backends' independently-declared interfaces with a loose `T | Promise` union on + // every method. Both createSqliteQueue and createPgQueue now return the same fully-async DurableQueue, so + // this can reference it directly instead of re-declaring a looser subset by hand. + queue: DurableQueue; vectorize?: Vectorize; shutdown(): Promise; } diff --git a/test/contract/selfhost-d1-database.test.ts b/test/contract/selfhost-d1-database.test.ts new file mode 100644 index 0000000000..e87657dcc5 --- /dev/null +++ b/test/contract/selfhost-d1-database.test.ts @@ -0,0 +1,99 @@ +// Shared contract test for the self-host D1-shaped storage adapter pair (#4010): runs the IDENTICAL +// assertion suite against createD1Adapter (real node:sqlite) and createPgAdapter (real Postgres) via one +// shared spec function, so a future change that breaks behavioral parity between the two -- despite both +// still satisfying SelfHostD1Database structurally -- is caught here instead of discovered later as a silent +// production divergence. +// +// The sqlite side always runs (node:sqlite is built-in and instant, matching +// test/unit/selfhost-d1-adapter.test.ts's own pattern). The Postgres side needs a REAL Postgres -- there is +// no meaningful way to fake generic SQL execution for an arbitrary CREATE TABLE/INSERT/SELECT -- so it +// follows the exact same PG_TEST_URL gate test/integration/selfhost-pg.test.ts already established: unset in +// CI (skipped, not failed), set locally against a real Postgres to actually exercise it: +// docker run -d -e POSTGRES_PASSWORD=devpw -e POSTGRES_DB=gittensory -p 55432:5432 postgres:16 +// PG_TEST_URL=postgres://postgres:devpw@localhost:55432/gittensory npx vitest run test/contract/selfhost-d1-database.test.ts +// Every statement here uses only PG-native column types (INTEGER/TEXT) and `?`-style placeholders with +// explicit values (never relying on SQLite ROWID auto-assignment), which pg-dialect.ts's translateDdl/ +// translateSql already pass through/translate unchanged (see its own doc comments) -- so the same SQL text is +// valid, and means the same thing, against both backends. This does not replace either backend's own richer +// test file -- it is the narrow, identical-inputs slice both must agree on. +import { DatabaseSync } from "node:sqlite"; +import { describe, expect, it, beforeAll, afterAll } from "vitest"; +import type { Pool } from "pg"; +import { createD1Adapter, nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; +import { createPgAdapter } from "../../src/selfhost/pg-adapter"; + +/** The identical assertion suite, run against whichever concrete D1Database `make()` returns. Each `it` calls + * `make()` itself (not a shared hoisted instance) so the sqlite side gets a fresh in-memory database per + * test; the Postgres side reuses one real connection pool across tests (see below) and instead relies on + * each test using its own uniquely-named table to stay isolated. */ +function runD1DatabaseContractTests(make: () => D1Database): void { + it("creates a table, inserts, and reads it back via all()/first()", async () => { + const db = make(); + await db.exec("CREATE TABLE contract_all_first (id INTEGER PRIMARY KEY, name TEXT)"); + await db.prepare("INSERT INTO contract_all_first (id, name) VALUES (?, ?)").bind(1, "a").run(); + await db.prepare("INSERT INTO contract_all_first (id, name) VALUES (?, ?)").bind(2, "b").run(); + + const all = await db.prepare("SELECT id, name FROM contract_all_first ORDER BY id").all<{ id: number; name: string }>(); + expect(all.success).toBe(true); + expect(all.results).toEqual([ + { id: 1, name: "a" }, + { id: 2, name: "b" }, + ]); + + const first = await db.prepare("SELECT name FROM contract_all_first WHERE id = ?").bind(1).first<{ name: string }>(); + expect(first).toEqual({ name: "a" }); + }); + + it("first() returns null for no matching row", async () => { + const db = make(); + await db.exec("CREATE TABLE contract_empty (id INTEGER PRIMARY KEY)"); + expect(await db.prepare("SELECT id FROM contract_empty WHERE id = 99").first()).toBeNull(); + }); + + it("run() reports success:true and a meta object", async () => { + const db = make(); + await db.exec("CREATE TABLE contract_run (id INTEGER PRIMARY KEY, name TEXT)"); + const result = await db.prepare("INSERT INTO contract_run (id, name) VALUES (1, 'x')").run(); + expect(result.success).toBe(true); + expect(typeof result.meta).toBe("object"); + }); + + it("batch() runs every statement, in order", async () => { + const db = make(); + await db.exec("CREATE TABLE contract_batch (id INTEGER PRIMARY KEY, name TEXT)"); + await db.batch([ + db.prepare("INSERT INTO contract_batch (id, name) VALUES (1, 'x')"), + db.prepare("INSERT INTO contract_batch (id, name) VALUES (2, 'y')"), + ]); + const count = await db.prepare("SELECT COUNT(*) AS n FROM contract_batch").first<{ n: number | string }>(); + expect(Number(count?.n)).toBe(2); + }); + + it("dump() returns an ArrayBuffer", async () => { + expect(await make().dump()).toBeInstanceOf(ArrayBuffer); + }); +} + +describe("D1Database contract (sqlite, #4010)", () => { + runD1DatabaseContractTests(() => createD1Adapter(nodeSqliteDriver(new DatabaseSync(":memory:") as never))); +}); + +const PG_TEST_URL = process.env.PG_TEST_URL; +const pgSuite = PG_TEST_URL ? describe : describe.skip; + +pgSuite("D1Database contract (postgres, #4010) — real Postgres", () => { + let pool: Pool; + + beforeAll(async () => { + const pg = (await import("pg")).default; + pg.types.setTypeParser(20, (v: string) => Number.parseInt(v, 10)); // int8 (COUNT) → number, like D1 + pool = new pg.Pool({ connectionString: PG_TEST_URL }); + await pool.query("DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public;"); + }); + + afterAll(async () => { + await pool?.end(); + }); + + runD1DatabaseContractTests(() => createPgAdapter(pool)); +}); diff --git a/test/contract/selfhost-durable-queue.test.ts b/test/contract/selfhost-durable-queue.test.ts new file mode 100644 index 0000000000..cd0abc5c15 --- /dev/null +++ b/test/contract/selfhost-durable-queue.test.ts @@ -0,0 +1,142 @@ +// Shared contract test for the self-host queue pair (#4010): runs the IDENTICAL assertion suite against +// createSqliteQueue and createPgQueue via describe.each, so a future change that breaks structural parity +// between the two DurableQueue implementations (a renamed method, a changed return shape, a dropped field) +// is caught here -- not discovered later as a silent divergence on whichever backend the change didn't +// touch. This is DELIBERATELY narrow: it exercises the introspection/admin surface (size, deadCount, +// pressureSignals, the dead-letter admin methods, ...) against a freshly-initialized, EMPTY queue on both +// backends, where the correct answer (0 / [] / false) is identical and unambiguous for both. It does not +// replace either backend's own much richer implementation-specific test file (selfhost-sqlite-queue.test.ts, +// selfhost-pg-queue.test.ts), which already cover real enqueue/claim/coalesce/retry semantics per backend. +// +// The Postgres side uses a minimal mock Pool (no real Postgres required, matching the existing +// selfhost-pg-queue.test.ts / selfhost-pg-vectorize.test.ts convention of mocking `pg.Pool` for unit-level +// coverage) that answers every aggregate/count query with a single zero-valued row and every list-style query +// with an empty row set -- both are the objectively correct answers for a genuinely empty table, so this +// mock needs no per-test scripting, unlike the richer scenario-specific MockPool in selfhost-pg-queue.test.ts. +import { DatabaseSync } from "node:sqlite"; +import { describe, expect, it } from "vitest"; +import type { Pool } from "pg"; +import { nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; +import { createSqliteQueue } from "../../src/selfhost/sqlite-queue"; +import { createPgQueue } from "../../src/selfhost/pg-queue"; +import type { DurableQueue } from "../../src/selfhost/backend-contracts"; +import type { JobMessage } from "../../src/types"; + +const noopConsume = async (_message: JobMessage): Promise => undefined; + +/** A single zero-valued row satisfying every aggregate column name the queue's own pressure/count queries + * select (`c`, `cnt`, `oldest`, `runnable_cnt`, `oldest_runnable`) -- correct for ANY of those queries when + * the underlying table is genuinely empty. */ +function zeroAggregateRow(): Record { + return { c: 0, cnt: 0, oldest: null, runnable_cnt: 0, oldest_runnable: null }; +} + +/** Minimal mock `pg.Pool` for an always-empty queue table: distinguishes a plain aggregate (COUNT with no + * GROUP BY -- deadCount/pressureSignals) from a grouped aggregate or a plain list query (GROUP BY, or no + * COUNT at all -- topBacklogRepos, the backfill/recovery SELECTs, listDeadLetterJobs, ...) by SQL text, since + * an empty base table genuinely produces a single zero/null row for the former and zero rows for the + * latter. Every UPDATE/DELETE reports rowCount 0 (nothing to touch). */ +function makeEmptyPgPool(): Pool { + return { + async query(sql: string) { + const text = String(sql); + const isPlainAggregate = /count\(\*\)/i.test(text) && !/group by/i.test(text); + return isPlainAggregate ? { rows: [zeroAggregateRow()], rowCount: 0 } : { rows: [], rowCount: 0 }; + }, + } as unknown as Pool; +} + +const backends: Array<{ name: string; make: () => Promise }> = [ + { + name: "sqlite", + make: async () => { + const driver = nodeSqliteDriver(new DatabaseSync(":memory:") as never); + const q = createSqliteQueue(driver, noopConsume); + await q.init(); + return q; + }, + }, + { + name: "postgres", + make: async () => { + const q = createPgQueue(makeEmptyPgPool(), noopConsume); + await q.init(); + return q; + }, + }, +]; + +describe.each(backends)("DurableQueue contract ($name, #4010)", ({ make }) => { + it("init() resolves without throwing", async () => { + await expect(make()).resolves.toBeDefined(); + }); + + it("exposes a Queue-shaped binding (send/sendBatch)", async () => { + const q = await make(); + expect(typeof q.binding.send).toBe("function"); + expect(typeof q.binding.sendBatch).toBe("function"); + }); + + it("size/deadCount/processingCount are all 0 on a fresh empty queue", async () => { + const q = await make(); + expect(await q.size()).toBe(0); + expect(await q.deadCount()).toBe(0); + expect(await q.processingCount()).toBe(0); + }); + + it("stats() returns an empty Record", async () => { + const q = await make(); + expect(await q.stats()).toEqual({}); + }); + + it("snapshot() returns the SelfHostQueueSnapshot shape with zeroed totals", async () => { + const q = await make(); + const snapshot = await q.snapshot(); + expect(snapshot.totals).toEqual({ pending: 0, processing: 0, dead: 0, due: 0 }); + expect(snapshot.byType).toEqual([]); + }); + + it("pressureSignals() reports clear pressure on every documented field", async () => { + const q = await make(); + const signals = await q.pressureSignals(); + expect(signals).toMatchObject({ + livePendingCount: 0, + oldestLivePendingAgeMs: null, + liveRunnableNowCount: 0, + oldestLiveRunnableAgeMs: null, + maintenancePendingCount: 0, + oldestMaintenancePendingAgeMs: null, + backlogConvergencePendingCount: 0, + freshIntakePendingCount: 0, + }); + }); + + it("topBacklogRepos() returns an empty array", async () => { + const q = await make(); + expect(await q.topBacklogRepos(10)).toEqual([]); + }); + + it("listDeadLetterJobs() returns an empty array", async () => { + const q = await make(); + expect(await q.listDeadLetterJobs(10, 0)).toEqual([]); + }); + + it("replayDeadLetterJob/deleteDeadLetterJob return false for a nonexistent id", async () => { + const q = await make(); + expect(await q.replayDeadLetterJob(999_999)).toBe(false); + expect(await q.deleteDeadLetterJob(999_999)).toBe(false); + }); + + it("purgeDeadLetterJobs/reviveDeadLetterJobs/releaseStaleForegroundDeferrals all report 0 work done", async () => { + const q = await make(); + expect(await q.purgeDeadLetterJobs()).toBe(0); + expect(await q.reviveDeadLetterJobs()).toBe(0); + expect(await q.releaseStaleForegroundDeferrals()).toBe(0); + }); + + it("drain() and stop() resolve without throwing", async () => { + const q = await make(); + await expect(q.drain()).resolves.toBeUndefined(); + await expect(q.stop()).resolves.toBeUndefined(); + }); +}); diff --git a/test/contract/selfhost-vectorize.test.ts b/test/contract/selfhost-vectorize.test.ts new file mode 100644 index 0000000000..d69bb0d0af --- /dev/null +++ b/test/contract/selfhost-vectorize.test.ts @@ -0,0 +1,115 @@ +// Shared contract test for the self-host Vectorize-shaped RAG store pair (#4010): runs the IDENTICAL +// assertion suite against createSqliteVectorize, createQdrantVectorize, and createPgVectorize via +// describe.each, so a future change that breaks parity between the three -- despite all three still +// satisfying SelfHostVectorize structurally -- is caught here instead of discovered later as a silent +// production divergence. Also directly exercises the returnMetadata fix (#4010): all three backends now +// accept `returnMetadata` on their QueryOptions (see backend-contracts.ts's SelfHostVectorizeQueryOptions doc +// comment for why), and this suite calls every one of them with it set, matching how +// src/review/adapters.ts's reviewVectorAdapter always calls whichever backend is bound to env.VECTORIZE. +// +// Postgres and Qdrant are backed by lightweight mocks (a scripted `pg.Pool` / a stubbed global `fetch`), +// matching the existing per-backend test files' own conventions (selfhost-pg-vectorize.test.ts, +// selfhost-qdrant-vectorize.test.ts) -- this suite is not a replacement for either backend's own richer +// implementation-specific tests, only the narrow, identical-inputs slice all three must agree on. +import { DatabaseSync } from "node:sqlite"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { Pool } from "pg"; +import { nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; +import { createSqliteVectorize } from "../../src/selfhost/vectorize"; +import { createQdrantVectorize } from "../../src/selfhost/qdrant-vectorize"; +import { createPgVectorize } from "../../src/selfhost/pg-vectorize"; +import type { SelfHostVectorize } from "../../src/selfhost/backend-contracts"; + +const QDRANT_BASE = "http://qdrant:6333"; + +/** One seeded vector record + the metadata a matching backend should surface for it. */ +const SEED = { id: "seed-1", values: [1, 0], namespace: "contract-ns", metadata: { path: "seed.ts" } }; + +/** Build a fake fetch that returns the given JSON body for any call (mirrors + * selfhost-qdrant-vectorize.test.ts's own mockFetch helper). */ +function mockFetch(body: unknown): typeof fetch { + return (async () => new Response(JSON.stringify(body), { status: 200 })) as unknown as typeof fetch; +} + +/** A scripted `pg.Pool` that returns the given rows for ANY query (mirrors + * selfhost-pg-vectorize.test.ts's own makePool helper). */ +function makePgPool(rows: Record[]): Pool { + return { async query() { return { rows, rowCount: rows.length }; } } as unknown as Pool; +} + +const backends: Array<{ + name: string; + makeEmpty: () => SelfHostVectorize; + /** A backend pre-seeded so a query in SEED.namespace returns exactly one match for SEED. */ + makeWithSeedMatch: () => SelfHostVectorize; +}> = [ + { + name: "sqlite", + makeEmpty: () => createSqliteVectorize(nodeSqliteDriver(new DatabaseSync(":memory:") as never)) as unknown as SelfHostVectorize, + makeWithSeedMatch: () => { + const v = createSqliteVectorize(nodeSqliteDriver(new DatabaseSync(":memory:") as never)) as unknown as SelfHostVectorize; + void v.upsert([SEED]); + return v; + }, + }, + { + name: "postgres", + makeEmpty: () => createPgVectorize(makePgPool([])) as unknown as SelfHostVectorize, + makeWithSeedMatch: () => + createPgVectorize(makePgPool([{ id: SEED.id, score: 1, metadata: SEED.metadata }])) as unknown as SelfHostVectorize, + }, + { + name: "qdrant", + makeEmpty: () => { + vi.stubGlobal("fetch", mockFetch({ result: [] })); + return createQdrantVectorize(QDRANT_BASE) as unknown as SelfHostVectorize; + }, + makeWithSeedMatch: () => { + vi.stubGlobal( + "fetch", + mockFetch({ + result: [{ id: "qdrant-point-uuid", score: 1, payload: { _orig_id: SEED.id, namespace: SEED.namespace, ...SEED.metadata } }], + }), + ); + return createQdrantVectorize(QDRANT_BASE) as unknown as SelfHostVectorize; + }, + }, +]; + +describe.each(backends)("SelfHostVectorize contract ($name, #4010)", ({ makeEmpty, makeWithSeedMatch }) => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it("upsert() returns a count and ids matching the input length", async () => { + const v = makeEmpty(); + const result = await v.upsert([SEED, { ...SEED, id: "seed-2" }]); + expect(result.count).toBe(2); + expect(result.ids).toEqual([SEED.id, "seed-2"]); + }); + + it("query() against an empty store returns no matches", async () => { + const v = makeEmpty(); + const { matches } = await v.query([1, 0], { topK: 5, namespace: SEED.namespace }); + expect(matches).toEqual([]); + }); + + it("query() with returnMetadata set finds the seeded match and surfaces its metadata", async () => { + const v = makeWithSeedMatch(); + const { matches } = await v.query([1, 0], { topK: 5, namespace: SEED.namespace, returnMetadata: "all" }); + expect(matches).toHaveLength(1); + expect(matches[0]?.id).toBe(SEED.id); + expect(matches[0]?.metadata?.path).toBe("seed.ts"); + }); + + it("query() accepts returnMetadata: none/indexed without throwing", async () => { + const v = makeWithSeedMatch(); + await expect(v.query([1, 0], { topK: 5, namespace: SEED.namespace, returnMetadata: "none" })).resolves.toBeDefined(); + await expect(v.query([1, 0], { topK: 5, namespace: SEED.namespace, returnMetadata: "indexed" })).resolves.toBeDefined(); + }); + + it("deleteByIds([]) is a no-op that resolves to count 0", async () => { + const v = makeEmpty(); + expect(await v.deleteByIds([])).toEqual({ count: 0 }); + }); +}); diff --git a/test/unit/selfhost-pg-adapter.test.ts b/test/unit/selfhost-pg-adapter.test.ts new file mode 100644 index 0000000000..2e9d6d4f7c --- /dev/null +++ b/test/unit/selfhost-pg-adapter.test.ts @@ -0,0 +1,98 @@ +// Unit tests for the Postgres-backed D1Database adapter (#977). Mocks pg.Pool so no real DB is needed, +// mirroring selfhost-pg-queue.test.ts / selfhost-pg-vectorize.test.ts's own convention. Real-Postgres +// integration paths (migrations, translated-SQL correctness against a live server) live in +// test/integration/selfhost-pg.test.ts; the autovacuum tuning helper has its own file +// (selfhost-pg-adapter-autovacuum.test.ts). This file covers the adapter's own D1-shaped surface +// (prepare/bind/all/first/run/raw/batch/exec/dump) against a scripted mock, mirroring +// selfhost-d1-adapter.test.ts's coverage of the same surface for the sqlite side (#4010 contract parity). +import { describe, expect, it } from "vitest"; +import type { Pool, PoolClient } from "pg"; +import { createPgAdapter } from "../../src/selfhost/pg-adapter"; + +/** A minimal Pool mock: `query()` (on the pool OR a connected client) always answers with the given rows, + * and every query issued (via the pool directly, or via a connect()ed client inside batch()'s transaction) + * is recorded to the SAME shared log, so a test can assert on the exact SQL/order a call produced. */ +function makeMockPool(rows: Record[] = []): Pool & { queries: Array<{ sql: string; params: unknown[] }> } { + const queries: Array<{ sql: string; params: unknown[] }> = []; + async function query(sql: string, params: unknown[] = []): Promise<{ rows: Record[]; rowCount: number }> { + queries.push({ sql: String(sql), params }); + return { rows, rowCount: rows.length }; + } + const client = { query, release() {} }; + const pool = { queries, query, connect: async () => client as unknown as PoolClient }; + return pool as unknown as Pool & { queries: Array<{ sql: string; params: unknown[] }> }; +} + +describe("createPgAdapter (#977 self-host D1-over-Postgres)", () => { + it("prepare/bind/all reads translated rows (? -> $1 placeholder translation)", async () => { + const pool = makeMockPool([{ id: 1, name: "a" }]); + const db = createPgAdapter(pool); + const result = await db.prepare("SELECT id, name FROM t WHERE id = ?").bind(1).all<{ id: number; name: string }>(); + expect(result.success).toBe(true); + expect(result.results).toEqual([{ id: 1, name: "a" }]); + expect(pool.queries[0]?.sql).toContain("$1"); + }); + + it("first() returns the row, or null when there is no row", async () => { + const withRow = createPgAdapter(makeMockPool([{ name: "a" }])); + expect(await withRow.prepare("SELECT name FROM t").first<{ name: string }>()).toEqual({ name: "a" }); + + const empty = createPgAdapter(makeMockPool([])); + expect(await empty.prepare("SELECT name FROM t").first()).toBeNull(); + }); + + it("first(colName) returns just that column's value", async () => { + const db = createPgAdapter(makeMockPool([{ name: "a" }])); + expect(await db.prepare("SELECT name FROM t").first("name")).toBe("a"); + }); + + it("run() reports success:true and a meta object carrying the row count", async () => { + const db = createPgAdapter(makeMockPool([])); + const result = await db.prepare("INSERT INTO t (name) VALUES (?)").bind("x").run(); + expect(result.success).toBe(true); + expect(typeof result.meta).toBe("object"); + }); + + it("raw() returns each row as an array of column values, column order preserved", async () => { + const db = createPgAdapter(makeMockPool([{ id: 1, name: "a" }])); + expect(await db.prepare("SELECT id, name FROM t").raw()).toEqual([[1, "a"]]); + }); + + it("batch() runs BEGIN/COMMIT and returns one result per statement, in order", async () => { + const pool = makeMockPool([{ id: 1 }]); + const db = createPgAdapter(pool); + const out = await db.batch([ + db.prepare("INSERT INTO t (id) VALUES (1)"), + db.prepare("INSERT INTO t (id) VALUES (2)"), + ]); + expect(out).toHaveLength(2); + expect(out[0]).toMatchObject({ success: true }); + expect(out[1]).toMatchObject({ success: true }); + const sqls = pool.queries.map((q) => q.sql); + expect(sqls[0]).toBe("BEGIN"); + expect(sqls.at(-1)).toBe("COMMIT"); + }); + + it("batch() rolls back and rethrows when a statement fails", async () => { + const client = { + async query(sql: string): Promise<{ rows: Record[]; rowCount: number }> { + if (sql === "BEGIN" || sql === "ROLLBACK") return { rows: [], rowCount: 0 }; + throw new Error("boom"); + }, + release() {}, + }; + const pool = { connect: async () => client as unknown as PoolClient } as unknown as Pool; + const db = createPgAdapter(pool); + await expect(db.batch([db.prepare("INSERT INTO t (id) VALUES (1)")])).rejects.toThrow("boom"); + }); + + it("exec() translates DDL and reports a statement count from the semicolon-separated input", async () => { + const db = createPgAdapter(makeMockPool([])); + const result = await db.exec("CREATE TABLE a (id INTEGER); CREATE TABLE b (id INTEGER);"); + expect(result.count).toBe(2); + }); + + it("dump() returns an ArrayBuffer (D1 surface completeness)", async () => { + expect(await createPgAdapter(makeMockPool([])).dump()).toBeInstanceOf(ArrayBuffer); + }); +}); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 360174c848..9f2d27a2ba 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -87,7 +87,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.binding.send(msg("b")); await q.drain(); expect(seen).toEqual(["a", "b"]); - expect(q.size()).toBe(0); + expect(await q.size()).toBe(0); }); it("REGRESSION: releases the reserved background slot when a background claim query throws (#selfhost-bg-slot-leak)", async () => { @@ -232,7 +232,7 @@ describe("createSqliteQueue (durable #980)", () => { ["github rate-limit background admission"], ).rows[0] as { c: number }; expect(pendingBackground.c).toBe(2); - expect(q.stats()).toMatchObject({ gittensory_jobs_rate_limit_deferred_total: 2 }); + expect(await q.stats()).toMatchObject({ gittensory_jobs_rate_limit_deferred_total: 2 }); const metrics = await renderMetrics(); expect(metrics).toContain('gittensory_jobs_rate_limit_admission_deferred_total{job_type="agent-regate-pr",key_scope="installation",kind="background"} 1'); expect(metrics).toContain('gittensory_jobs_rate_limit_admission_deferred_total{job_type="rag-index-repo",key_scope="public",kind="background"} 1'); @@ -389,7 +389,7 @@ describe("createSqliteQueue (durable #980)", () => { run_after: Date.parse("2026-06-24T12:10:15.000Z"), last_error: "github rate-limit webhook admission", }); - expect(q.stats()).toMatchObject({ gittensory_jobs_rate_limit_deferred_total: 1 }); + expect(await q.stats()).toMatchObject({ gittensory_jobs_rate_limit_deferred_total: 1 }); expect(await renderMetrics()).toContain('gittensory_jobs_rate_limit_admission_deferred_total{job_type="github-webhook",key_scope="installation",kind="webhook"} 1'); } finally { if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; @@ -441,7 +441,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.binding.send(installedWebhook("fresh", 123)); await q.drain(); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); expect(await renderMetrics()).not.toContain("gittensory_jobs_rate_limit_admission_deferred_total"); } finally { if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; @@ -543,7 +543,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.drain(); expect(seen).toEqual(["github-webhook"]); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); } finally { if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; @@ -647,7 +647,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.drain(); expect(seen).toEqual(["github-webhook"]); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); }); it("does not keep webhook admission closed from stale legacy rows after a newer healthy exact observation", async () => { @@ -686,7 +686,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.drain(); expect(seen).toEqual(["github-webhook"]); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); }); it("does not pre-yield webhook jobs for another installation's persisted REST exhaustion", async () => { @@ -720,7 +720,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.drain(); expect(seen).toEqual(["github-webhook"]); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); }); it("skips the background-admission metric when the defer update changes no rows", async () => { @@ -770,7 +770,7 @@ describe("createSqliteQueue (durable #980)", () => { expect(seen).toEqual([]); expect(warned).not.toHaveBeenCalled(); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); } finally { if (oldJitter === undefined) delete process.env.QUEUE_RATE_LIMIT_JITTER_MS; else process.env.QUEUE_RATE_LIMIT_JITTER_MS = oldJitter; @@ -865,7 +865,7 @@ describe("createSqliteQueue (durable #980)", () => { `github-webhook:pr-refresh:jsonbored/gittensory#1629@${"a".repeat(40)}`, ]); expect(rows.map((row) => JSON.parse(row.payload).deliveryId).filter(Boolean).sort()).toEqual(["ci-2", "pr-2"]); - expect(q.stats()).toMatchObject({ + expect(await q.stats()).toMatchObject({ gittensory_jobs_enqueued_total: 3, gittensory_jobs_coalesced_total: 3, }); @@ -985,7 +985,7 @@ describe("createSqliteQueue (durable #980)", () => { requestedBy: "schedule", repoFullName: "JSONbored/gittensory", }); - expect(q.stats()).toMatchObject({ + expect(await q.stats()).toMatchObject({ gittensory_jobs_enqueued_total: 1, gittensory_jobs_coalesced_total: 1, }); @@ -1025,7 +1025,7 @@ describe("createSqliteQueue (durable #980)", () => { }); // The two incrementals now MERGE into one row before the full job supersedes it (#selfhost-maintenance-self-pin): // 1 insert (the first incremental) + 2 coalesces (the merge, then the supersede), not 2 inserts + 1 coalesce. - expect(q.stats()).toMatchObject({ + expect(await q.stats()).toMatchObject({ gittensory_jobs_enqueued_total: 1, gittensory_jobs_coalesced_total: 2, }); @@ -1062,7 +1062,7 @@ describe("createSqliteQueue (durable #980)", () => { paths: ["src/a.ts", "src/b.ts"], }); expect(rows[0]?.job_key).toBe(jobCoalesceKey(rows[0]!.payload)); - expect(q.stats()).toMatchObject({ + expect(await q.stats()).toMatchObject({ gittensory_jobs_enqueued_total: 1, gittensory_jobs_coalesced_total: 1, }); @@ -1153,7 +1153,7 @@ describe("createSqliteQueue (durable #980)", () => { [JSON.stringify(msg("rag-index-repo")), Date.now()], ); - const snapshot = q.snapshot(); + const snapshot = await q.snapshot(); const bindingSnapshot = await queueSnapshotFromBinding(q.binding); expect(snapshot.totals).toMatchObject({ pending: 2, processing: 1, dead: 1 }); @@ -1262,7 +1262,7 @@ describe("createSqliteQueue (durable #980)", () => { "api", "schedule", ]); - expect(q.stats()).toMatchObject({ + expect(await q.stats()).toMatchObject({ gittensory_jobs_enqueued_total: 5, gittensory_jobs_coalesced_total: 4, }); @@ -1298,7 +1298,7 @@ describe("createSqliteQueue (durable #980)", () => { expect(rows[0]?.id).toBe(first.id); expect(rows[0]?.created_at).toBe(first.created_at); // NOT reset to the re-enqueue time expect(rows[0]?.run_after).toBeGreaterThan(first.run_after); // still advances with the new request - expect(q.stats()).toMatchObject({ gittensory_jobs_coalesced_total: 1 }); + expect(await q.stats()).toMatchObject({ gittensory_jobs_coalesced_total: 1 }); } finally { vi.useRealTimers(); } @@ -1761,7 +1761,7 @@ describe("createSqliteQueue (durable #980)", () => { it("returns an empty array when no backlog-lane row is pending", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); - expect(q.topBacklogRepos(10)).toEqual([]); + expect(await q.topBacklogRepos(10)).toEqual([]); }); it("counts pending AND processing backlog-lane rows, grouped by repo, sorted by depth", async () => { @@ -1779,7 +1779,7 @@ describe("createSqliteQueue (durable #980)", () => { [JSON.stringify(backlogJob(repo, prNumber)), status, `agent-regate-pr:${repo}#${prNumber}`], ); } - expect(q.topBacklogRepos(10)).toEqual([ + expect(await q.topBacklogRepos(10)).toEqual([ { repo: "owner/b", count: 3 }, { repo: "owner/a", count: 1 }, ]); @@ -1798,7 +1798,7 @@ describe("createSqliteQueue (durable #980)", () => { [JSON.stringify(backlogJob(repo, 1)), `agent-regate-pr:${repo}#1`], ); } - expect(q.topBacklogRepos(2)).toHaveLength(2); + expect(await q.topBacklogRepos(2)).toHaveLength(2); }); it("excludes a terminal (dead/cancelled) backlog-lane row", async () => { @@ -1808,18 +1808,18 @@ describe("createSqliteQueue (durable #980)", () => { "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane) VALUES (?, 'dead', 0, 0, 1000, 9, ?, 'backlog')", [JSON.stringify(backlogJob("owner/repo", 1)), "agent-regate-pr:owner/repo#1"], ); - expect(q.topBacklogRepos(10)).toEqual([]); + expect(await q.topBacklogRepos(10)).toEqual([]); }); }); describe("listDeadLetterJobs (#2214)", () => { - it("returns an empty array when there are no dead-letter rows", () => { + it("returns an empty array when there are no dead-letter rows", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); - expect(q.listDeadLetterJobs(10, 0)).toEqual([]); + expect(await q.listDeadLetterJobs(10, 0)).toEqual([]); }); - it("maps dead rows newest-death-first, extracting job type/attempts/error, and excludes non-dead rows", () => { + it("maps dead rows newest-death-first, extracting job type/attempts/error, and excludes non-dead rows", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); driver.query( @@ -1834,13 +1834,13 @@ describe("createSqliteQueue (durable #980)", () => { "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at) VALUES (?, 'pending', 0, 0, 500)", [JSON.stringify(msg("agent-regate-sweep"))], ); - expect(q.listDeadLetterJobs(10, 0)).toEqual([ + expect(await q.listDeadLetterJobs(10, 0)).toEqual([ { id: 2, jobType: "github-webhook", attempts: 1, lastError: "kaboom", createdAtMs: 2000, deadAtMs: 9000 }, { id: 1, jobType: "agent-regate-pr", attempts: 3, lastError: "boom", createdAtMs: 1000, deadAtMs: 5000 }, ]); }); - it("falls back to created_at ordering and reports deadAtMs null for a legacy row with no dead_at", () => { + it("falls back to created_at ordering and reports deadAtMs null for a legacy row with no dead_at", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); // Legacy row: no dead_at, but its created_at (7000) is newer than the other row's real dead_at (3000) -- @@ -1853,25 +1853,25 @@ describe("createSqliteQueue (durable #980)", () => { "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, last_error, dead_at) VALUES (?, 'dead', 1, 0, 1000, 'recent failure', 3000)", [JSON.stringify(msg("agent-regate-pr"))], ); - expect(q.listDeadLetterJobs(10, 0)).toEqual([ + expect(await q.listDeadLetterJobs(10, 0)).toEqual([ { id: 1, jobType: "agent-regate-pr", attempts: 2, lastError: "legacy failure", createdAtMs: 7000, deadAtMs: null }, { id: 2, jobType: "agent-regate-pr", attempts: 1, lastError: "recent failure", createdAtMs: 1000, deadAtMs: 3000 }, ]); }); - it("reports jobType 'unknown' for an unparseable payload", () => { + it("reports jobType 'unknown' for an unparseable payload", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); driver.query( "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, last_error, dead_at) VALUES ('not-json', 'dead', 0, 0, 1000, 'unparseable payload', 1000)", [], ); - expect(q.listDeadLetterJobs(10, 0)).toEqual([ + expect(await q.listDeadLetterJobs(10, 0)).toEqual([ { id: 1, jobType: "unknown", attempts: 0, lastError: "unparseable payload", createdAtMs: 1000, deadAtMs: 1000 }, ]); }); - it("paginates via limit/offset", () => { + it("paginates via limit/offset", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); for (let i = 0; i < 3; i++) { @@ -1880,19 +1880,19 @@ describe("createSqliteQueue (durable #980)", () => { [JSON.stringify(msg("agent-regate-pr")), 1000 + i, 1000 + i], ); } - expect(q.listDeadLetterJobs(1, 1).map((job) => job.createdAtMs)).toEqual([1001]); + expect((await q.listDeadLetterJobs(1, 1)).map((job) => job.createdAtMs)).toEqual([1001]); }); }); describe("replay/delete/purge dead-letter jobs (#2215)", () => { - it("replayDeadLetterJob requeues an existing dead row with a fresh retry budget", () => { + it("replayDeadLetterJob requeues an existing dead row with a fresh retry budget", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); driver.query( "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, last_error, dead_at) VALUES (?, 'dead', 3, 0, 1000, 'boom', 5000)", [JSON.stringify(msg("agent-regate-pr"))], ); - expect(q.replayDeadLetterJob(1)).toBe(true); + expect(await q.replayDeadLetterJob(1)).toBe(true); const row = driver.query("SELECT status, attempts, last_error, dead_at, run_after FROM _selfhost_jobs WHERE id=1", []) .rows[0] as { status: string; attempts: number; last_error: string | null; dead_at: number | null; run_after: number }; expect(row.status).toBe("pending"); @@ -1902,20 +1902,20 @@ describe("createSqliteQueue (durable #980)", () => { expect(row.run_after).toBeGreaterThan(0); }); - it("replayDeadLetterJob returns false for a non-existent id", () => { + it("replayDeadLetterJob returns false for a non-existent id", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); - expect(q.replayDeadLetterJob(999)).toBe(false); + expect(await q.replayDeadLetterJob(999)).toBe(false); }); - it("replayDeadLetterJob returns false and leaves a non-dead row untouched", () => { + it("replayDeadLetterJob returns false and leaves a non-dead row untouched", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); driver.query( "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at) VALUES (?, 'pending', 0, 42, 1000)", [JSON.stringify(msg("agent-regate-pr"))], ); - expect(q.replayDeadLetterJob(1)).toBe(false); + expect(await q.replayDeadLetterJob(1)).toBe(false); const row = driver.query("SELECT status, run_after FROM _selfhost_jobs WHERE id=1", []).rows[0] as { status: string; run_after: number; @@ -1924,35 +1924,35 @@ describe("createSqliteQueue (durable #980)", () => { expect(row.run_after).toBe(42); }); - it("deleteDeadLetterJob removes an existing dead row", () => { + it("deleteDeadLetterJob removes an existing dead row", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); driver.query( "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, dead_at) VALUES (?, 'dead', 1, 0, 1000, 1000)", [JSON.stringify(msg("agent-regate-pr"))], ); - expect(q.deleteDeadLetterJob(1)).toBe(true); + expect(await q.deleteDeadLetterJob(1)).toBe(true); expect(driver.query("SELECT id FROM _selfhost_jobs WHERE id=1", []).rows).toEqual([]); }); - it("deleteDeadLetterJob returns false for a non-existent id", () => { + it("deleteDeadLetterJob returns false for a non-existent id", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); - expect(q.deleteDeadLetterJob(999)).toBe(false); + expect(await q.deleteDeadLetterJob(999)).toBe(false); }); - it("deleteDeadLetterJob returns false and does not delete a non-dead row", () => { + it("deleteDeadLetterJob returns false and does not delete a non-dead row", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); driver.query( "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at) VALUES (?, 'processing', 0, 0, 1000)", [JSON.stringify(msg("agent-regate-pr"))], ); - expect(q.deleteDeadLetterJob(1)).toBe(false); + expect(await q.deleteDeadLetterJob(1)).toBe(false); expect(driver.query("SELECT id FROM _selfhost_jobs WHERE id=1", []).rows).toHaveLength(1); }); - it("purgeDeadLetterJobs deletes every dead row and leaves non-dead rows untouched", () => { + it("purgeDeadLetterJobs deletes every dead row and leaves non-dead rows untouched", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); for (let i = 0; i < 3; i++) { @@ -1969,19 +1969,19 @@ describe("createSqliteQueue (durable #980)", () => { "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at) VALUES (?, 'processing', 0, 0, 3000)", [JSON.stringify(msg("agent-regate-pr"))], ); - expect(q.purgeDeadLetterJobs()).toBe(3); + expect(await q.purgeDeadLetterJobs()).toBe(3); expect((driver.query("SELECT COUNT(*) AS c FROM _selfhost_jobs WHERE status='dead'", []).rows[0] as { c: number }).c).toBe(0); expect((driver.query("SELECT COUNT(*) AS c FROM _selfhost_jobs WHERE status!='dead'", []).rows[0] as { c: number }).c).toBe(2); }); - it("purgeDeadLetterJobs returns 0 and touches nothing when there are no dead rows", () => { + it("purgeDeadLetterJobs returns 0 and touches nothing when there are no dead rows", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); driver.query( "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at) VALUES (?, 'pending', 0, 0, 1000)", [JSON.stringify(msg("agent-regate-pr"))], ); - expect(q.purgeDeadLetterJobs()).toBe(0); + expect(await q.purgeDeadLetterJobs()).toBe(0); expect((driver.query("SELECT COUNT(*) AS c FROM _selfhost_jobs", []).rows[0] as { c: number }).c).toBe(1); }); }); @@ -2000,10 +2000,10 @@ describe("createSqliteQueue (durable #980)", () => { await q.binding.send(msg("x")); await q.drain(); // backoff 0 → all 3 attempts run within one drain, then dead-lettered expect(calls).toBe(3); - expect(q.deadCount()).toBe(1); - expect(q.size()).toBe(0); + expect(await q.deadCount()).toBe(1); + expect(await q.size()).toBe(0); // #2214: a max-retries death also stamps dead_at, so the DLQ table can sort/report a real death time. - const [row] = q.listDeadLetterJobs(10, 0); + const [row] = await q.listDeadLetterJobs(10, 0); expect(row).toMatchObject({ jobType: "x", attempts: 3, lastError: "boom" }); expect(row!.deadAtMs).not.toBeNull(); }); @@ -2035,10 +2035,10 @@ describe("createSqliteQueue (durable #980)", () => { ); await q.binding.send(msg("x")); await q.drain(); // dies at attempts=1 (maxRetries=1) - expect(q.deadCount()).toBe(1); + expect(await q.deadCount()).toBe(1); calls = 0; - const revived = q.reviveDeadLetterJobs(); + const revived = await q.reviveDeadLetterJobs(); expect(revived).toBe(1); const { rows } = driver.query("SELECT status, attempts, last_error FROM _selfhost_jobs", []); @@ -2056,7 +2056,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.drain(); // the one extra attempt the revival granted expect(calls).toBe(1); - expect(q.deadCount()).toBe(1); // failed again -- back to dead, attempts now 2 + expect(await q.deadCount()).toBe(1); // failed again -- back to dead, attempts now 2 }); it("stops reviving a job once it reaches the auto-retry ceiling (maxRetries + extra attempts)", async () => { @@ -2066,17 +2066,17 @@ describe("createSqliteQueue (durable #980)", () => { await q.binding.send(msg("x")); await q.drain(); // attempts=1, dead (ceiling = maxRetries(1) + extra(1) = 2) - expect(q.reviveDeadLetterJobs()).toBe(1); // attempts(1) < ceiling(2) -- eligible + expect(await q.reviveDeadLetterJobs()).toBe(1); // attempts(1) < ceiling(2) -- eligible await q.drain(); // fails again -- attempts=2, dead again - expect(q.reviveDeadLetterJobs()).toBe(0); // attempts(2) is NOT < ceiling(2) -- exhausted, stays dead - expect(q.deadCount()).toBe(1); + expect(await q.reviveDeadLetterJobs()).toBe(0); // attempts(2) is NOT < ceiling(2) -- exhausted, stays dead + expect(await q.deadCount()).toBe(1); }); - it("is a no-op when there are no dead jobs", () => { + it("is a no-op when there are no dead jobs", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); - expect(q.reviveDeadLetterJobs()).toBe(0); + expect(await q.reviveDeadLetterJobs()).toBe(0); }); // REGRESSION (#2581 review defect, parity with the same fix in pg-queue.ts): the SELECT that finds eligible @@ -2091,7 +2091,7 @@ describe("createSqliteQueue (durable #980)", () => { const q = createSqliteQueue(driver, async () => { throw new Error("boom"); }, { maxRetries: 1, backoffMs: () => 0 }); await q.binding.send(msg("x")); await q.drain(); // dies at attempts=1 (maxRetries=1) - expect(q.deadCount()).toBe(1); + expect(await q.deadCount()).toBe(1); vi.spyOn(driver, "query").mockImplementation((sql: string, params: unknown[]) => { if (sql.includes("SET status='pending', run_after=?, last_error=NULL")) { @@ -2100,7 +2100,7 @@ describe("createSqliteQueue (durable #980)", () => { return realQuery(sql, params); }); - const revived = q.reviveDeadLetterJobs(); + const revived = await q.reviveDeadLetterJobs(); expect(revived).toBe(0); // the UPDATE's "AND status='dead'" matched zero rows -- not counted as revived const { rows } = driver.query("SELECT status FROM _selfhost_jobs", []); @@ -2122,7 +2122,7 @@ describe("createSqliteQueue (durable #980)", () => { ); await q.binding.send(msg("x")); await vi.advanceTimersByTimeAsync(200); // dies at attempts=1 - expect(q.deadCount()).toBe(1); + expect(await q.deadCount()).toBe(1); calls = 0; q.start(); @@ -2281,7 +2281,7 @@ describe("createSqliteQueue (durable #980)", () => { const now = Date.now(); seedForegroundPendingRow(driver, { createdAt: now - 5 * 60_000, runAfter: now + 60 * 60_000 }); - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(1); const row = driver.query("SELECT run_after FROM _selfhost_jobs", []).rows[0] as { run_after: number }; @@ -2312,7 +2312,7 @@ describe("createSqliteQueue (durable #980)", () => { ], ); - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(0); const row = driver.query("SELECT run_after FROM _selfhost_jobs", []).rows[0] as { run_after: number }; @@ -2346,7 +2346,7 @@ describe("createSqliteQueue (durable #980)", () => { ); } - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(0); const admissionReads = querySpy.mock.calls.filter(([sql]) => String(sql).includes("FROM github_rate_limit_observations")); @@ -2370,7 +2370,7 @@ describe("createSqliteQueue (durable #980)", () => { [JSON.stringify({ type: "github-webhook", deliveryId: "now-clear", eventName: "x", payload: {} }), futureRunAfter, now - 1_000], ); - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(1); expect(await renderMetrics()).toContain("gittensory_jobs_foreground_liveness_released_total 1"); @@ -2391,7 +2391,7 @@ describe("createSqliteQueue (durable #980)", () => { ["not valid json", futureRunAfter, now - 1_000], ); - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(0); }); @@ -2414,7 +2414,7 @@ describe("createSqliteQueue (durable #980)", () => { type: "build-contributor-evidence", }); - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(0); const row = driver.query("SELECT run_after FROM _selfhost_jobs", []).rows[0] as { run_after: number }; @@ -2428,7 +2428,7 @@ describe("createSqliteQueue (durable #980)", () => { const now = Date.now(); seedForegroundPendingRow(driver, { createdAt: now - 60 * 60_000, runAfter: now + 60 * 60_000 }); - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(0); const row = driver.query("SELECT run_after FROM _selfhost_jobs", []).rows[0] as { run_after: number }; @@ -2457,7 +2457,7 @@ describe("createSqliteQueue (durable #980)", () => { seedForegroundPendingRow(driver, { createdAt: staleAge, runAfter: farFuture }); } - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(3); expect(await renderMetrics()).toContain("gittensory_jobs_foreground_liveness_released_total 3"); @@ -2485,7 +2485,7 @@ describe("createSqliteQueue (durable #980)", () => { seedForegroundPendingRow(driver, { createdAt: now - ageMs, runAfter: farFuture }); } - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(2); expect(await renderMetrics()).toContain("gittensory_jobs_foreground_liveness_released_total 2"); @@ -2527,7 +2527,7 @@ describe("createSqliteQueue (durable #980)", () => { ); } - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); expect(released).toBe(2); const releasedIds = driver.query(`SELECT payload FROM _selfhost_jobs WHERE run_after @@ -2571,7 +2571,7 @@ describe("createSqliteQueue (durable #980)", () => { [JSON.stringify({ type: "github-webhook", deliveryId: "clear-newer", eventName: "x", payload: { installation: { id: 222 } } }), farFuture, now - 1_000], ); - const released = q.releaseStaleForegroundDeferrals(); + const released = await q.releaseStaleForegroundDeferrals(); const releasedIds = driver.query(`SELECT payload FROM _selfhost_jobs WHERE run_after JSON.parse((row as { payload: string }).payload).deliveryId, @@ -2635,13 +2635,13 @@ describe("createSqliteQueue (durable #980)", () => { "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority) VALUES (?, 'pending', 0, 0, 0, 0)", [JSON.stringify(msg("y"))], ); - expect(q.processingCount()).toBe(1); + expect(await q.processingCount()).toBe(1); }); it("returns 0 when no job is processing", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); - expect(q.processingCount()).toBe(0); + expect(await q.processingCount()).toBe(0); }); }); @@ -2671,7 +2671,7 @@ describe("createSqliteQueue (durable #980)", () => { last_error: string; }; expect(calls).toBe(1); - expect(q.deadCount()).toBe(0); + expect(await q.deadCount()).toBe(0); expect(row.status).toBe("pending"); expect(row.attempts).toBe(0); expect(row.run_after).toBeGreaterThan(Date.now()); @@ -2703,11 +2703,11 @@ describe("createSqliteQueue (durable #980)", () => { attempts: 2, last_error: "openai api rate limit exceeded", }); - expect(q.stats()).toMatchObject({ + expect(await q.stats()).toMatchObject({ gittensory_jobs_failed_total: 2, gittensory_jobs_dead_total: 1, }); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_rate_limited_total"); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_rate_limited_total"); }); it("does not defer GitHub work when a non-GitHub job throws a GitHub-looking rate limit", async () => { @@ -2745,8 +2745,8 @@ describe("createSqliteQueue (durable #980)", () => { expect(pending).toHaveLength(1); expect(JSON.parse(pending[0]!.payload)).toMatchObject({ type: "refresh-registry" }); expect(pending[0]!.last_error).toBe("API rate limit exceeded for installation ID 123"); - expect(q.stats()).toMatchObject({ gittensory_jobs_rate_limited_total: 1 }); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); + expect(await q.stats()).toMatchObject({ gittensory_jobs_rate_limited_total: 1 }); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_rate_limit_deferred_total"); expect(await renderMetrics()).toContain('gittensory_jobs_rate_limited_by_type_total{job_type="refresh-registry",key_scope="unknown",kind="unknown"} 1'); }); @@ -2823,7 +2823,7 @@ describe("createSqliteQueue (durable #980)", () => { expect(byType.get("blocked-installation")?.last_error).toBe("API rate limit exceeded for installation ID 123"); expect(byType.get("agent-regate-pr:9")?.last_error).toBe("github rate-limit budget deferred"); expect(byType.has("agent-regate-pr:10")).toBe(false); - expect(q.stats()).toMatchObject({ + expect(await q.stats()).toMatchObject({ gittensory_jobs_processed_total: 3, gittensory_jobs_rate_limited_total: 1, gittensory_jobs_rate_limit_deferred_total: 1, @@ -2868,7 +2868,7 @@ describe("createSqliteQueue (durable #980)", () => { expect(JSON.parse(rows[0]!.payload).deliveryId).toBe("ci-existing"); expect(rows[0]!.attempts).toBe(0); expect(rows[0]!.last_error).toContain("secondary rate limit"); - expect(q.stats()).toMatchObject({ gittensory_jobs_coalesced_total: 1 }); + expect(await q.stats()).toMatchObject({ gittensory_jobs_coalesced_total: 1 }); }); it("reschedules a keyed rate-limited job when no pending duplicate exists", async () => { @@ -2908,7 +2908,7 @@ describe("createSqliteQueue (durable #980)", () => { expect(row.attempts).toBe(0); expect(row.run_after).toBeGreaterThan(Date.now()); expect(row.last_error).toContain("secondary rate limit"); - expect(q.stats()).toMatchObject({ gittensory_jobs_rate_limited_total: 1 }); + expect(await q.stats()).toMatchObject({ gittensory_jobs_rate_limited_total: 1 }); }); it("consumes retryable incomplete review attempts and dead-letters after maxRetries", async () => { @@ -2940,7 +2940,7 @@ describe("createSqliteQueue (durable #980)", () => { last_error: string; }; expect(calls).toBe(1); - expect(q.deadCount()).toBe(0); + expect(await q.deadCount()).toBe(0); expect(row.status).toBe("pending"); expect(row.attempts).toBe(1); expect(row.run_after).toBeGreaterThanOrEqual(before + 5_000); @@ -2953,7 +2953,7 @@ describe("createSqliteQueue (durable #980)", () => { [], ).rows[0] as { status: string; attempts: number; last_error: string }; expect(calls).toBe(2); - expect(q.deadCount()).toBe(1); + expect(await q.deadCount()).toBe(1); expect(dead.status).toBe("dead"); expect(dead.attempts).toBe(2); expect(dead.last_error).toContain("AI review did not produce"); @@ -2995,7 +2995,7 @@ describe("createSqliteQueue (durable #980)", () => { expect(JSON.parse(rows[1]!.payload).deliveryId).toBe("ci-existing"); expect(rows[1]!.attempts).toBe(0); expect(rows[1]!.last_error).toBeNull(); - expect(q.stats().gittensory_jobs_coalesced_total ?? 0).toBe(0); + expect((await q.stats()).gittensory_jobs_coalesced_total ?? 0).toBe(0); }); it("SURVIVES A RESTART: a fresh queue over the same DB processes a persisted pending job", async () => { @@ -3022,7 +3022,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.drain(); expect(driver.query("SELECT status FROM _selfhost_jobs", []).rows[0]).toMatchObject({ status: "processing" }); - expect(q.stats().gittensory_jobs_recovered_total ?? 0).toBe(0); + expect((await q.stats()).gittensory_jobs_recovered_total ?? 0).toBe(0); } finally { if (old === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS; else process.env.QUEUE_PROCESSING_TIMEOUT_MS = old; @@ -3155,7 +3155,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.drain(); expect(seen).toEqual(["lease-expired"]); - expect(q.stats()).toMatchObject({ + expect(await q.stats()).toMatchObject({ gittensory_jobs_recovered_total: 1, gittensory_jobs_processed_total: 1, }); @@ -3201,7 +3201,7 @@ describe("createSqliteQueue (durable #980)", () => { await new Promise((r) => setTimeout(r, 10)); expect(seen.filter((type) => type === "slow")).toHaveLength(1); - expect(queue.stats().gittensory_jobs_recovered_total ?? 0).toBe(0); + expect((await queue.stats()).gittensory_jobs_recovered_total ?? 0).toBe(0); } finally { for (const release of releases) release(); if (q) await q.stop(); @@ -3263,7 +3263,7 @@ describe("createSqliteQueue (durable #980)", () => { ); await q.binding.send(msg("x")); await q.drain(); - expect(q.deadCount()).toBe(1); + expect(await q.deadCount()).toBe(1); }); it("dead-letters an unparseable payload", async () => { @@ -3271,9 +3271,9 @@ describe("createSqliteQueue (durable #980)", () => { const q = createSqliteQueue(driver, async () => undefined); driver.query("INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at) VALUES ('not-json','pending',0,0,0)", []); await q.drain(); - expect(q.deadCount()).toBe(1); + expect(await q.deadCount()).toBe(1); // #2214: an unparseable payload has no `type` field to extract -- the DLQ table falls back to "unknown". - const [row] = q.listDeadLetterJobs(10, 0); + const [row] = await q.listDeadLetterJobs(10, 0); expect(row).toMatchObject({ jobType: "unknown", lastError: "unparseable payload" }); expect(row!.deadAtMs).not.toBeNull(); // A malformed payload consumes the same bounded retry budget as a normal failure (previously left `attempts` @@ -3296,13 +3296,13 @@ describe("createSqliteQueue (durable #980)", () => { await q2.binding.send(msg("f")); await q2.drain(); expect(calls).toBe(1); - expect(q2.size()).toBe(1); + expect(await q2.size()).toBe(1); }); it("stop() is a no-op when start() was never called (timer is null)", async () => { const q = createSqliteQueue(makeDriver(), async () => undefined); await q.stop(); // timer=null → the false branch of `if (timer) clearTimeout(timer)` is taken - expect(q.size()).toBe(0); // still usable after a spurious stop() + expect(await q.size()).toBe(0); // still usable after a spurious stop() }); it("concurrency=1 saturates after one active pump (active >= concurrency → early return)", async () => { @@ -3319,7 +3319,7 @@ describe("createSqliteQueue (durable #980)", () => { await new Promise((r) => setTimeout(r, 60)); await q.stop(); expect(maxConcurrent).toBe(1); - expect(q.size()).toBe(0); + expect(await q.size()).toBe(0); }); it("concurrency=2 allows two jobs to run simultaneously", async () => { @@ -3335,7 +3335,7 @@ describe("createSqliteQueue (durable #980)", () => { await new Promise((r) => setTimeout(r, 60)); await q.stop(); expect(maxConcurrent).toBe(2); - expect(q.size()).toBe(0); + expect(await q.size()).toBe(0); }); it("start() is idempotent and stop() waits for an in-flight pump", async () => { @@ -3408,7 +3408,7 @@ describe("createSqliteQueue (durable #980)", () => { expect(row.status).toBe("pending"); expect(row.run_after).toBeGreaterThan(before); expect(row.last_error).toContain("live_pending_high"); - expect(q.stats()).toMatchObject({ gittensory_jobs_maintenance_admission_deferred_total: 1 }); + expect(await q.stats()).toMatchObject({ gittensory_jobs_maintenance_admission_deferred_total: 1 }); expect(await renderMetrics()).toContain( 'gittensory_jobs_maintenance_admission_deferred_by_reason_total{job_type="build-contributor-evidence",reason="live_pending_high"} 1', ); @@ -3441,7 +3441,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.binding.send(msg("build-contributor-evidence")); await q.drain(); expect(started).toEqual(["build-contributor-evidence"]); - expect(q.size()).toBe(0); + expect(await q.size()).toBe(0); }); it("never defers live/foreground work, even under the same pressure that defers maintenance work", async () => { @@ -3598,7 +3598,7 @@ describe("createSqliteQueue (durable #980)", () => { ); await q.drain(); expect(started).toEqual(["build-contributor-evidence"]); - expect(q.stats()).toMatchObject({ gittensory_jobs_maintenance_trickle_admitted_total: 1 }); + expect(await q.stats()).toMatchObject({ gittensory_jobs_maintenance_trickle_admitted_total: 1 }); const metrics = await renderMetrics(); expect(metrics).toContain('gittensory_jobs_maintenance_trickle_admitted_by_type_total{job_type="build-contributor-evidence"} 1'); expect(metrics).toContain('gittensory_jobs_maintenance_admission_granted_under_pressure_total{job_type="build-contributor-evidence",reason="trickle_max_defer_age"} 1'); @@ -3611,7 +3611,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.binding.send(msg("build-contributor-evidence")); await q.drain(); expect(started).toEqual(["build-contributor-evidence"]); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_maintenance_trickle_admitted_total"); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_maintenance_trickle_admitted_total"); expect(await renderMetrics()).not.toContain("gittensory_jobs_maintenance_trickle_admitted"); }); @@ -3635,7 +3635,7 @@ describe("createSqliteQueue (durable #980)", () => { VALUES (?, 'pending', 0, ?, ?, 0, NULL, 1)`, [JSON.stringify({ type: "rollup-product-usage", requestedBy: "test" }), now + 3_600_000, now - 5_000], ); - const signals = q.pressureSignals(); + const signals = await q.pressureSignals(); expect(signals.livePendingCount).toBe(2); expect(signals.oldestLivePendingAgeMs).toBeGreaterThanOrEqual(0); expect(signals.maintenancePendingCount).toBe(1); @@ -3646,7 +3646,7 @@ describe("createSqliteQueue (durable #980)", () => { it("pressureSignals() reports null oldest ages when a lane has no pending work", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); - const signals = q.pressureSignals(); + const signals = await q.pressureSignals(); expect(signals.livePendingCount).toBe(0); expect(signals.oldestLivePendingAgeMs).toBeNull(); expect(signals.maintenancePendingCount).toBe(0); @@ -3671,7 +3671,7 @@ describe("createSqliteQueue (durable #980)", () => { } as unknown as JobMessage); // A fresh-intake row (foreground_lane='fresh') must NOT count toward the backlog-convergence signal. await q.binding.send(prWebhook("fresh-unrelated")); - const signals = q.pressureSignals(); + const signals = await q.pressureSignals(); expect(signals.backlogConvergencePendingCount).toBe(1); }); @@ -3687,7 +3687,7 @@ describe("createSqliteQueue (durable #980)", () => { prNumber: 1, installationId: 1, } as unknown as JobMessage); - const signals = q.pressureSignals(); + const signals = await q.pressureSignals(); expect(signals.freshIntakePendingCount).toBe(1); }); @@ -3713,7 +3713,7 @@ describe("createSqliteQueue (durable #980)", () => { VALUES (?, 'pending', 0, ?, ?, 9, NULL, 0)`, [JSON.stringify(msg("agent-regate-pr")), now - 1_000, now - 10_000], ); - const signals = q.pressureSignals(); + const signals = await q.pressureSignals(); expect(signals.livePendingCount).toBe(2); expect(signals.oldestLivePendingAgeMs).toBeGreaterThanOrEqual(500_000); expect(signals.liveRunnableNowCount).toBe(1); @@ -3735,7 +3735,7 @@ describe("createSqliteQueue (durable #980)", () => { [JSON.stringify(msg("agent-regate-pr")), now + 3_600_000, now - 60_000], ); } - const signals = q.pressureSignals(); + const signals = await q.pressureSignals(); expect(signals.livePendingCount).toBe(3); expect(signals.liveRunnableNowCount).toBe(0); expect(signals.oldestLiveRunnableAgeMs).toBeNull(); @@ -3834,7 +3834,7 @@ describe("createSqliteQueue (durable #980)", () => { await q.binding.send(msg("build-contributor-evidence")); await q.drain(); expect(started).not.toContain("build-contributor-evidence"); - expect(q.stats()).not.toHaveProperty("gittensory_jobs_maintenance_admission_deferred_total"); + expect(await q.stats()).not.toHaveProperty("gittensory_jobs_maintenance_admission_deferred_total"); expect(await renderMetrics()).not.toContain("gittensory_jobs_maintenance_admission_deferred_by_reason_total"); }); });