diff --git a/review-enrichment/src/upload-sourcemaps.ts b/review-enrichment/src/upload-sourcemaps.ts index 1fd46a2449..d6065001a9 100644 --- a/review-enrichment/src/upload-sourcemaps.ts +++ b/review-enrichment/src/upload-sourcemaps.ts @@ -1,6 +1,7 @@ import { spawnSync } from "node:child_process"; import { existsSync, readFileSync, readdirSync, statSync } from "node:fs"; import { dirname, relative, resolve } from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; import { fileURLToPath } from "node:url"; import { @@ -108,28 +109,52 @@ function shouldValidateRelease(): boolean { return !/^(0|false|no|off)$/i.test(process.env.REES_SENTRY_VALIDATE_RELEASE ?? ""); } -function runReleaseValidation(release: string, fields: { sha?: string; deployName: string; environment: string }): void { +function numericEnv(name: string, fallback: number, max: number): number { + const raw = Number(nonBlank(process.env[name])); + return Number.isFinite(raw) && raw >= 0 ? Math.min(Math.floor(raw), max) : fallback; +} + +async function runReleaseValidation( + release: string, + fields: { sha?: string; deployName: string; environment: string }, +): Promise { if (!shouldValidateRelease()) return; - const result = spawnSync(process.execPath, ["scripts/validate-sentry-release.mjs"], { - cwd: appDir, - env: { - ...process.env, - SENTRY_RELEASE: release, - SENTRY_COMMIT_SHA: fields.sha ?? "", - SENTRY_DEPLOY_NAME: fields.deployName, - SENTRY_ENVIRONMENT: fields.environment, - SENTRY_REQUIRE_COMMITS: "true", - SENTRY_REQUIRE_DEPLOY: "true", - SENTRY_REQUIRE_FINALIZED: "true", - }, - encoding: "utf8", - }); - const output = `${result.stdout ?? ""}${result.stderr ?? ""}`.trim(); - if (result.status === 0) { - if (output) log("rees_sentry_release_validation", { output: output.slice(0, 500) }); - return; + const attempts = Math.max(1, numericEnv("REES_SENTRY_VALIDATE_ATTEMPTS", 5, 20)); + const retryDelayMs = numericEnv("REES_SENTRY_VALIDATE_RETRY_DELAY_MS", 1_000, 30_000); + let output = ""; + let status: number | null = null; + for (let attempt = 1; attempt <= attempts; attempt += 1) { + const result = spawnSync(process.execPath, ["scripts/validate-sentry-release.mjs"], { + cwd: appDir, + env: { + ...process.env, + SENTRY_RELEASE: release, + SENTRY_COMMIT_SHA: fields.sha ?? "", + SENTRY_DEPLOY_NAME: fields.deployName, + SENTRY_ENVIRONMENT: fields.environment, + SENTRY_REQUIRE_COMMITS: "true", + SENTRY_REQUIRE_DEPLOY: "true", + SENTRY_REQUIRE_FINALIZED: "true", + }, + encoding: "utf8", + }); + status = result.status; + output = `${result.stdout ?? ""}${result.stderr ?? ""}`.trim(); + if (result.status === 0) { + if (output) log("rees_sentry_release_validation", { output: output.slice(0, 500), attempt }); + return; + } + if (attempt < attempts) { + warn("rees_sentry_release_validation_retry", { + attempt, + attempts, + retryDelayMs, + message: output.slice(0, 500), + }); + if (retryDelayMs > 0) await sleep(retryDelayMs); + } } - throw new Error(`Sentry release validation failed (${result.status}): ${output.slice(0, 500)}`); + throw new Error(`Sentry release validation failed (${status}): ${output.slice(0, 500)}`); } async function main(): Promise { @@ -191,7 +216,7 @@ async function main(): Promise { nonBlank(process.env.RAILWAY_DEPLOYMENT_ID) ?? "railway", ]); runSentry(["releases", ...projectArgs, "finalize", release!]); - runReleaseValidation(release!, { + await runReleaseValidation(release!, { sha, deployName: nonBlank(process.env.RAILWAY_DEPLOYMENT_ID) ?? "railway", environment: resolveSentryEnvironment(process.env), diff --git a/review-enrichment/test/sentry-upload.test.ts b/review-enrichment/test/sentry-upload.test.ts index 1e9ff06e2a..c7ce97dac0 100644 --- a/review-enrichment/test/sentry-upload.test.ts +++ b/review-enrichment/test/sentry-upload.test.ts @@ -28,17 +28,20 @@ test("resolveTracesSampleRate clamps malformed or out-of-range config", () => { assert.equal(resolveTracesSampleRate({ SENTRY_TRACES_SAMPLE_RATE: "2" }), 1); }); -async function sentryApiServer() { +async function sentryApiServer(options: { missingCommitOnFirstValidation?: boolean } = {}) { const seen: string[] = []; + let releaseReads = 0; const server = createServer((req, res) => { seen.push(req.url ?? ""); res.setHeader("content-type", "application/json"); if (req.url === "/api/0/organizations/jsonbored/releases/gittensory-rees%40abc123/") { + releaseReads += 1; + const missingCommit = options.missingCommitOnFirstValidation && releaseReads === 1; res.end( JSON.stringify({ version: "gittensory-rees@abc123", dateReleased: "2026-06-29T00:00:00Z", - commitCount: 1, + commitCount: missingCommit ? 0 : 1, deployCount: 1, projects: [{ slug: "rees" }], lastDeploy: { name: "deploy-1", environment: "production" }, @@ -47,7 +50,11 @@ async function sentryApiServer() { return; } if (req.url === "/api/0/organizations/jsonbored/releases/gittensory-rees%40abc123/commits/") { - res.end(JSON.stringify([{ id: "abc123" }])); + res.end( + JSON.stringify( + options.missingCommitOnFirstValidation && releaseReads === 1 ? [] : [{ id: "abc123" }], + ), + ); return; } if (req.url === "/api/0/organizations/jsonbored/releases/gittensory-rees%40abc123/deploys/") { @@ -69,6 +76,18 @@ async function sentryApiServer() { }; } +function sentryCliStub() { + const dir = mkdtempSync(resolve(tmpdir(), "rees-sentry-cli-")); + const logPath = resolve(dir, "calls.jsonl"); + const cliPath = resolve(dir, "sentry-cli"); + writeFileSync( + cliPath, + `#!/bin/sh\nnode -e 'require("fs").appendFileSync(process.argv[1], JSON.stringify(process.argv.slice(2)) + "\\n")' '${logPath}' "$@"\n`, + ); + chmodSync(cliPath, 0o755); + return { cliPath, logPath }; +} + async function runUploadSourcemaps(env: NodeJS.ProcessEnv) { return new Promise<{ status: number | null; stdout: string; stderr: string }>((resolveRun, reject) => { const child = spawn(process.execPath, ["dist/upload-sourcemaps.js"], { @@ -93,14 +112,7 @@ async function runUploadSourcemaps(env: NodeJS.ProcessEnv) { test("upload-sourcemaps calls Sentry CLI with release association on upload", async () => { const api = await sentryApiServer(); - const dir = mkdtempSync(resolve(tmpdir(), "rees-sentry-cli-")); - const logPath = resolve(dir, "calls.jsonl"); - const cliPath = resolve(dir, "sentry-cli"); - writeFileSync( - cliPath, - `#!/bin/sh\nnode -e 'require("fs").appendFileSync(process.argv[1], JSON.stringify(process.argv.slice(2)) + "\\n")' '${logPath}' "$@"\n`, - ); - chmodSync(cliPath, 0o755); + const { cliPath, logPath } = sentryCliStub(); try { const result = await runUploadSourcemaps({ @@ -131,6 +143,18 @@ test("upload-sourcemaps calls Sentry CLI with release association on upload", as "new", "gittensory-rees@abc123", ]); + assert.deepEqual(calls[1], [ + "releases", + "--org", + "jsonbored", + "--project", + "rees", + "set-commits", + "gittensory-rees@abc123", + "--commit", + "JSONbored/gittensory@abc123", + "--ignore-missing", + ]); assert.deepEqual(calls[2], ["sourcemaps", "--org", "jsonbored", "--project", "rees", "inject", "dist"]); assert.deepEqual(calls[3], [ "sourcemaps", @@ -153,3 +177,35 @@ test("upload-sourcemaps calls Sentry CLI with release association on upload", as await api.close(); } }); + +test("upload-sourcemaps retries release validation until Sentry exposes associated commits", async () => { + const api = await sentryApiServer({ missingCommitOnFirstValidation: true }); + const { cliPath } = sentryCliStub(); + + try { + const result = await runUploadSourcemaps({ + ...process.env, + SENTRY_AUTH_TOKEN: "test-token", + SENTRY_ORG: "jsonbored", + SENTRY_PROJECT: "rees", + SENTRY_CLI_PATH: cliPath, + SENTRY_URL: api.url, + RAILWAY_GIT_COMMIT_SHA: "abc123", + RAILWAY_DEPLOYMENT_ID: "deploy-1", + RAILWAY_ENVIRONMENT_NAME: "production", + REES_SENTRY_UPLOAD_STRICT: "true", + REES_SENTRY_VALIDATE_ATTEMPTS: "2", + REES_SENTRY_VALIDATE_RETRY_DELAY_MS: "0", + }); + + assert.equal(result.status, 0, `${result.stdout}\n${result.stderr}`); + assert.match(result.stderr, /rees_sentry_release_validation_retry/); + const commitPath = "/api/0/organizations/jsonbored/releases/gittensory-rees%40abc123/commits/"; + assert.equal( + api.seen.filter((path) => path === commitPath).length, + 2, + ); + } finally { + await api.close(); + } +}); diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 180c6aa6a5..191e1c9a16 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -17,7 +17,9 @@ import { githubRateLimitMetricContext, githubRateLimitRetryDelayMs, buildSelfHostQueueSnapshot, + jobCoalesceAbsorbedByKey, jobCoalesceKey, + jobCoalesceSupersededKeyPrefix, jobPriority, queueBackgroundConcurrency, queueProcessingTimeoutMs, @@ -229,6 +231,47 @@ export function createPgQueue( const priority = jobPriority(payload); const key = jobCoalesceKey(payload); const runAfter = now + delaySeconds * 1000; + const absorbedByKey = jobCoalesceAbsorbedByKey(payload); + if (absorbedByKey) { + const existingFull = ( + await pool.query( + `SELECT id FROM ${TABLE} WHERE status='pending' AND job_key=$1 ORDER BY priority DESC, run_after DESC, id LIMIT 1`, + [absorbedByKey], + ) + ).rows[0] as { id: string } | undefined; + if (existingFull) { + await recordQueueMetric("gittensory_jobs_coalesced_total"); + kickOne(); + return; + } + } + const supersededKeyPrefix = jobCoalesceSupersededKeyPrefix(payload); + if (key && supersededKeyPrefix) { + const existing = ( + await pool.query( + `SELECT id FROM ${TABLE} + WHERE status='pending' AND job_key IS NOT NULL AND left(job_key, $1)=$2 + ORDER BY priority DESC, run_after DESC, id LIMIT 1`, + [supersededKeyPrefix.length, supersededKeyPrefix], + ) + ).rows[0] as { id: string } | undefined; + if (existing) { + await pool.query( + `UPDATE ${TABLE} + SET payload=$1, run_after=GREATEST(run_after, $2), created_at=$3, priority=GREATEST(priority, $4), job_key=$5, last_error=NULL + WHERE id=$6`, + [payload, runAfter, now, priority, key, existing.id], + ); + await pool.query( + `DELETE FROM ${TABLE} + WHERE status='pending' AND id<>$1 AND job_key IS NOT NULL AND left(job_key, $2)=$3`, + [existing.id, supersededKeyPrefix.length, supersededKeyPrefix], + ); + await recordQueueMetric("gittensory_jobs_coalesced_total"); + kickOne(); + return; + } + } if (key) { const existing = ( await pool.query( diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index 840e293553..7b202f7002 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -579,32 +579,67 @@ export function scheduledEnqueueDelaySeconds(jobType: string): number { ); } +type CoalesceMessage = { + type?: unknown; + eventName?: unknown; + requestedBy?: unknown; + repoFullName?: unknown; + prNumber?: unknown; + attempt?: unknown; + force?: unknown; + mode?: unknown; + segment?: unknown; + cursor?: unknown; + login?: unknown; + day?: unknown; + days?: unknown; + dryRun?: unknown; + variant?: unknown; + paths?: unknown; + runId?: unknown; + deliveryId?: unknown; + draftId?: unknown; + event?: { dedupKey?: unknown } | null; + logins?: unknown; + payload?: GitHubWebhookPayload | null; +}; + +function parseCoalesceMessage(payload: string): CoalesceMessage | null { + try { + return JSON.parse(payload) as CoalesceMessage; + } catch { + return null; + } +} + +function ragIndexFullKey(repo: string): string { + return keyOf("rag-index-repo", repo, "full"); +} + +function ragIndexRepoKeyPrefix(repo: string): string { + return keyOf("rag-index-repo", repo, ""); +} + +export function jobCoalesceSupersededKeyPrefix(payload: string): string | null { + const message = parseCoalesceMessage(payload); + if (message?.type !== "rag-index-repo") return null; + const repo = normalizedRepo(message.repoFullName); + if (!repo || normalizedPathScope(message.paths)) return null; + return ragIndexRepoKeyPrefix(repo); +} + +export function jobCoalesceAbsorbedByKey(payload: string): string | null { + const message = parseCoalesceMessage(payload); + if (message?.type !== "rag-index-repo") return null; + const repo = normalizedRepo(message.repoFullName); + if (!repo || !normalizedPathScope(message.paths)) return null; + return ragIndexFullKey(repo); +} + export function jobCoalesceKey(payload: string): string | null { try { - const message = JSON.parse(payload) as { - type?: unknown; - eventName?: unknown; - requestedBy?: unknown; - repoFullName?: unknown; - prNumber?: unknown; - attempt?: unknown; - force?: unknown; - mode?: unknown; - segment?: unknown; - cursor?: unknown; - login?: unknown; - day?: unknown; - days?: unknown; - dryRun?: unknown; - variant?: unknown; - paths?: unknown; - runId?: unknown; - deliveryId?: unknown; - draftId?: unknown; - event?: { dedupKey?: unknown } | null; - logins?: unknown; - payload?: GitHubWebhookPayload | null; - }; + const message = parseCoalesceMessage(payload); + if (!message) return null; const type = typeof message.type === "string" ? message.type : ""; if (type === "agent-regate-pr") { const repo = normalizedRepo(message.repoFullName); diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index fc4a92cb07..a9d9ccf39f 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -18,7 +18,9 @@ import { githubRateLimitMetricContext, githubRateLimitRetryDelayMs, buildSelfHostQueueSnapshot, + jobCoalesceAbsorbedByKey, jobCoalesceKey, + jobCoalesceSupersededKeyPrefix, jobPriority, queueBackgroundConcurrency, queueProcessingTimeoutMs, @@ -172,6 +174,44 @@ export function createSqliteQueue( const priority = jobPriority(payload); const key = jobCoalesceKey(payload); const runAfter = now + delaySeconds * 1000; + const absorbedByKey = jobCoalesceAbsorbedByKey(payload); + if (absorbedByKey) { + const existingFull = driver.query( + `SELECT id FROM ${TABLE} WHERE status='pending' AND job_key=? ORDER BY priority DESC, run_after DESC, id LIMIT 1`, + [absorbedByKey], + ).rows[0] as { id: number } | undefined; + if (existingFull) { + recordQueueMetric(driver, "gittensory_jobs_coalesced_total"); + kickOne(); + return; + } + } + const supersededKeyPrefix = jobCoalesceSupersededKeyPrefix(payload); + if (key && supersededKeyPrefix) { + const prefixLength = supersededKeyPrefix.length; + const existing = driver.query( + `SELECT id FROM ${TABLE} + WHERE status='pending' AND job_key IS NOT NULL AND substr(job_key, 1, ?)=? + ORDER BY priority DESC, run_after DESC, id LIMIT 1`, + [prefixLength, supersededKeyPrefix], + ).rows[0] as { id: number } | undefined; + if (existing) { + driver.query( + `UPDATE ${TABLE} + SET payload=?, run_after=max(run_after, ?), created_at=?, priority=max(priority, ?), job_key=?, last_error=NULL + WHERE id=?`, + [payload, runAfter, now, priority, key, existing.id], + ); + driver.query( + `DELETE FROM ${TABLE} + WHERE status='pending' AND id<>? AND job_key IS NOT NULL AND substr(job_key, 1, ?)=?`, + [existing.id, prefixLength, supersededKeyPrefix], + ); + recordQueueMetric(driver, "gittensory_jobs_coalesced_total"); + kickOne(); + return; + } + } if (key) { const existing = driver.query( `SELECT id FROM ${TABLE} WHERE status='pending' AND job_key=? ORDER BY priority DESC, run_after DESC, id LIMIT 1`, diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index a805aa04c1..fe842d4589 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -335,6 +335,62 @@ describe("createPgQueue (durable #977)", () => { ); }); + it("lets a pending full RAG index absorb a later repo incremental", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [{ id: "existing-full" }], rowCount: 1 }); + + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts"], + }); + + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("WHERE status='pending' AND job_key=$1"), + ["rag-index-repo:jsonbored/gittensory:full"], + ); + expect(m.pool.query).not.toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO _selfhost_jobs (payload"), + expect.arrayContaining([expect.stringContaining('"paths":["src/a.ts"]')]), + ); + }); + + it("lets a full RAG index supersede pending repo incrementals", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [{ id: "existing-incremental" }], rowCount: 1 }); + + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/gittensory", + }); + + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("left(job_key, $1)=$2"), + ["rag-index-repo:jsonbored/gittensory:".length, "rag-index-repo:jsonbored/gittensory:"], + ); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET payload=$1, run_after=GREATEST"), + expect.arrayContaining([ + expect.stringContaining('"requestedBy":"schedule"'), + expect.any(Number), + expect.any(Number), + 0, + "rag-index-repo:jsonbored/gittensory:full", + "existing-incremental", + ]), + ); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("DELETE FROM _selfhost_jobs"), + ["existing-incremental", "rag-index-repo:jsonbored/gittensory:".length, "rag-index-repo:jsonbored/gittensory:"], + ); + }); + it("coalesces recurring maintenance jobs by semantic scope and preserves distinct scopes", async () => { const m = makePool(); const q = createPgQueue(m.pool, async () => undefined); diff --git a/test/unit/selfhost-queue-common.test.ts b/test/unit/selfhost-queue-common.test.ts index 80c3d63a35..468b8a54d0 100644 --- a/test/unit/selfhost-queue-common.test.ts +++ b/test/unit/selfhost-queue-common.test.ts @@ -14,7 +14,9 @@ import { githubWebhookRateLimitDelayMs, isGitHubBudgetBackgroundJob, isForegroundJobPriority, + jobCoalesceAbsorbedByKey, jobCoalesceKey, + jobCoalesceSupersededKeyPrefix, jobPriority, nonConsumingRetryDelayMs, queueBackgroundConcurrency, @@ -618,16 +620,17 @@ describe("self-host queue common helpers", () => { }), ), ).toBe("generate-weekly-value-report:public:31"); - expect( - jobCoalesceKey( - payload({ - type: "rag-index-repo", - requestedBy: "webhook", - repoFullName: "JSONbored/Gittensory", - paths: ["README.md", "src/a.ts", "README.md"], - }), - ), - ).toBe("rag-index-repo:jsonbored/gittensory:sha256:8812e979fc698c98d98665ad4ccd8630e396dabdce08ebf87b41600c94bb1df5"); + const incrementalRagJob = payload({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/Gittensory", + paths: ["README.md", "src/a.ts", "README.md"], + }); + expect(jobCoalesceKey(incrementalRagJob)).toBe( + "rag-index-repo:jsonbored/gittensory:sha256:8812e979fc698c98d98665ad4ccd8630e396dabdce08ebf87b41600c94bb1df5", + ); + expect(jobCoalesceAbsorbedByKey(incrementalRagJob)).toBe("rag-index-repo:jsonbored/gittensory:full"); + expect(jobCoalesceSupersededKeyPrefix(incrementalRagJob)).toBeNull(); expect( jobCoalesceKey( payload({ @@ -668,6 +671,14 @@ describe("self-host queue common helpers", () => { }), ), ).toBe("rag-index-repo:jsonbored/gittensory:full"); + const fullRagJob = payload({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/Gittensory", + }); + expect(jobCoalesceKey(fullRagJob)).toBe("rag-index-repo:jsonbored/gittensory:full"); + expect(jobCoalesceSupersededKeyPrefix(fullRagJob)).toBe("rag-index-repo:jsonbored/gittensory:"); + expect(jobCoalesceAbsorbedByKey(fullRagJob)).toBeNull(); expect(jobCoalesceKey(payload({ type: "prune-retention", requestedBy: "schedule", dryRun: true }))).toBe( "prune-retention:1", ); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 642547b2f2..fe7a5533b5 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -820,6 +820,76 @@ describe("createSqliteQueue (durable #980)", () => { }); }); + it("lets a pending full RAG index absorb later repo incrementals", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/gittensory", + }, { delaySeconds: 60 }); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts"], + }, { delaySeconds: 1 }); + + const rows = driver.query( + "SELECT payload, job_key FROM _selfhost_jobs ORDER BY id", + [], + ).rows as Array<{ payload: string; job_key: string }>; + expect(rows).toHaveLength(1); + expect(rows[0]?.job_key).toBe("rag-index-repo:jsonbored/gittensory:full"); + expect(JSON.parse(rows[0]!.payload)).toEqual({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/gittensory", + }); + expect(q.stats()).toMatchObject({ + gittensory_jobs_enqueued_total: 1, + gittensory_jobs_coalesced_total: 1, + }); + }); + + it("lets a full RAG index supersede pending repo incrementals without dropping to one path set", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts"], + }, { delaySeconds: 60 }); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/b.ts"], + }, { delaySeconds: 60 }); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/gittensory", + }, { delaySeconds: 1 }); + + const rows = driver.query( + "SELECT payload, job_key FROM _selfhost_jobs ORDER BY id", + [], + ).rows as Array<{ payload: string; job_key: string }>; + expect(rows).toHaveLength(1); + expect(rows[0]?.job_key).toBe("rag-index-repo:jsonbored/gittensory:full"); + expect(JSON.parse(rows[0]!.payload)).toEqual({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/gittensory", + }); + expect(q.stats()).toMatchObject({ + gittensory_jobs_enqueued_total: 2, + gittensory_jobs_coalesced_total: 1, + }); + }); + it("snapshot() reports pending/processing/dead queue depth by job type", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined);