fix(selfhost): stabilize REES validation and RAG coalescing

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<void> {
   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<number> {
@@ -191,7 +216,7 @@ async function main(): Promise<number> {
       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),

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();
+  }
+});

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(

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);