From edbf70fd27ee1a41d7b8e9d07892054efb6f2603 Mon Sep 17 00:00:00 2001 From: galuis116 Date: Sun, 12 Jul 2026 17:01:00 -0400 Subject: [PATCH] test(miner): add real concurrent-process races test for claim-ledger and portfolio-queue (#4867) Only worktree-allocator.js had a dedicated multi-process collision test before; claim-ledger and portfolio-queue atomicity was only exercised per-function, single process. Spawns real Node child processes racing the same on-disk SQLite file (mirroring acquire-child.mjs's fixture pattern) and asserts no double-claim, double-dequeue, or corrupted state under real concurrent load. --- .../miner-concurrent-stores/claim-child.mjs | 34 +++ .../miner-concurrent-stores/dequeue-child.mjs | 34 +++ .../unit/miner-concurrent-store-races.test.ts | 214 ++++++++++++++++++ 3 files changed, 282 insertions(+) create mode 100644 test/fixtures/miner-concurrent-stores/claim-child.mjs create mode 100644 test/fixtures/miner-concurrent-stores/dequeue-child.mjs create mode 100644 test/unit/miner-concurrent-store-races.test.ts diff --git a/test/fixtures/miner-concurrent-stores/claim-child.mjs b/test/fixtures/miner-concurrent-stores/claim-child.mjs new file mode 100644 index 0000000000..b5093b2fa3 --- /dev/null +++ b/test/fixtures/miner-concurrent-stores/claim-child.mjs @@ -0,0 +1,34 @@ +#!/usr/bin/env node +// Cross-process helper for claim-ledger concurrent-race tests (#4867). +// Opens the shared ledger, waits for a stdin "go" signal, then calls claimIssue() so multiple Node +// processes contend on the same UNIQUE(repo_full_name, issue_number) row via the same dbPath. +import { openClaimLedger } from "../../../packages/gittensory-miner/lib/claim-ledger.js"; + +const [dbPath, repoFullName, issueNumberStr, note] = process.argv.slice(2); +if (!dbPath || !repoFullName || !issueNumberStr) { + process.stderr.write("usage: claim-child.mjs [note]\n"); + process.exit(2); +} + +const ledger = openClaimLedger(dbPath); +let started = false; + +function runClaim() { + if (started) return; + started = true; + try { + const claim = ledger.claimIssue(repoFullName, Number(issueNumberStr), note || null); + process.stdout.write(`${JSON.stringify({ ok: true, claim })}\n`); + process.exit(0); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + process.stdout.write(`${JSON.stringify({ ok: false, message })}\n`); + process.exit(1); + } finally { + ledger.close(); + } +} + +process.stdin.setEncoding("utf8"); +process.stdin.on("data", () => runClaim()); +process.stdout.write("READY\n"); diff --git a/test/fixtures/miner-concurrent-stores/dequeue-child.mjs b/test/fixtures/miner-concurrent-stores/dequeue-child.mjs new file mode 100644 index 0000000000..96e75cd5a0 --- /dev/null +++ b/test/fixtures/miner-concurrent-stores/dequeue-child.mjs @@ -0,0 +1,34 @@ +#!/usr/bin/env node +// Cross-process helper for portfolio-queue concurrent-race tests (#4867). +// Opens the shared queue, waits for a stdin "go" signal, then calls dequeueNext() so multiple Node +// processes contend on the same atomic UPDATE...RETURNING claim via the same dbPath. +import { initPortfolioQueueStore } from "../../../packages/gittensory-miner/lib/portfolio-queue.js"; + +const [dbPath] = process.argv.slice(2); +if (!dbPath) { + process.stderr.write("usage: dequeue-child.mjs \n"); + process.exit(2); +} + +const store = initPortfolioQueueStore(dbPath); +let started = false; + +function runDequeue() { + if (started) return; + started = true; + try { + const entry = store.dequeueNext(); + process.stdout.write(`${JSON.stringify({ ok: true, entry })}\n`); + process.exit(0); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + process.stdout.write(`${JSON.stringify({ ok: false, message })}\n`); + process.exit(1); + } finally { + store.close(); + } +} + +process.stdin.setEncoding("utf8"); +process.stdin.on("data", () => runDequeue()); +process.stdout.write("READY\n"); diff --git a/test/unit/miner-concurrent-store-races.test.ts b/test/unit/miner-concurrent-store-races.test.ts new file mode 100644 index 0000000000..c3b61182e4 --- /dev/null +++ b/test/unit/miner-concurrent-store-races.test.ts @@ -0,0 +1,214 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it } from "vitest"; +import { openClaimLedger } from "../../packages/gittensory-miner/lib/claim-ledger.js"; +import { initPortfolioQueueStore } from "../../packages/gittensory-miner/lib/portfolio-queue.js"; + +// Real cross-process concurrency coverage for the claim-ledger and portfolio-queue stores (#4867). Only the +// worktree-allocator had a dedicated multi-process collision test before this; claim-ledger/portfolio-queue +// atomicity was previously only exercised per-function (single process). This spawns two real Node child +// processes racing the same on-disk SQLite file and asserts no double-claim/double-dequeue or corrupted state +// results — the store's own atomic UPSERT/UPDATE...RETURNING statements are what's under test, not the +// conflict-resolution logic (who "should" win a race), which is explicitly out of scope per the issue. + +const claimChildScript = join( + dirname(fileURLToPath(import.meta.url)), + "../fixtures/miner-concurrent-stores/claim-child.mjs", +); +const dequeueChildScript = join( + dirname(fileURLToPath(import.meta.url)), + "../fixtures/miner-concurrent-stores/dequeue-child.mjs", +); + +const roots: string[] = []; + +function tempRoot(): { root: string; dbPath: string } { + const root = mkdtempSync(join(tmpdir(), "gittensory-miner-concurrent-races-")); + roots.push(root); + return { root, dbPath: join(root, "store.sqlite3") }; +} + +function spawnChild(script: string, args: string[]): ChildProcessWithoutNullStreams { + return spawn(process.execPath, [script, ...args], { stdio: ["pipe", "pipe", "pipe"] }); +} + +async function waitForReady(child: ChildProcessWithoutNullStreams): Promise { + await new Promise((resolve, reject) => { + let buffer = ""; + const onData = (chunk: Buffer | string) => { + buffer += chunk.toString(); + if (buffer.includes("READY\n")) { + child.stdout.off("data", onData); + resolve(); + } + }; + child.stdout.on("data", onData); + child.once("error", reject); + child.once("exit", (code) => { + if (code !== 0 && code !== null) reject(new Error(`child exited before READY (${code})`)); + }); + }); +} + +async function runBarriered(children: ChildProcessWithoutNullStreams[]): Promise { + await Promise.all(children.map((child) => waitForReady(child))); + for (const child of children) child.stdin.write("go\n"); + return Promise.all( + children.map( + (child) => + new Promise((resolve, reject) => { + let stdout = ""; + child.stdout.on("data", (chunk) => { + stdout += chunk.toString(); + }); + child.once("error", reject); + child.once("exit", () => { + const line = stdout + .split("\n") + .map((entry) => entry.trim()) + .find((entry) => entry.startsWith("{")); + if (!line) { + reject(new Error(`child produced no JSON result: ${stdout}`)); + return; + } + resolve(JSON.parse(line) as T); + }); + }), + ), + ); +} + +afterEach(() => { + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +type ClaimChildResult = { + ok: boolean; + claim?: { repoFullName: string; issueNumber: number; status: string }; + message?: string; +}; + +describe("claim-ledger cross-process races (#4867)", () => { + it("two processes claiming the SAME issue simultaneously produce exactly one active row, no duplication", async () => { + const { dbPath } = tempRoot(); + const children = [ + spawnChild(claimChildScript, [dbPath, "acme/widgets", "42"]), + spawnChild(claimChildScript, [dbPath, "acme/widgets", "42"]), + ]; + const results = await runBarriered(children); + + expect(results.every((result) => result.ok)).toBe(true); + expect(results.every((result) => result.claim?.status === "active")).toBe(true); + + const ledger = openClaimLedger(dbPath); + try { + const rows = ledger.listClaims({ repoFullName: "acme/widgets" }); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ repoFullName: "acme/widgets", issueNumber: 42, status: "active" }); + } finally { + ledger.close(); + } + }); + + it("two processes claiming DIFFERENT issues concurrently both succeed with distinct rows", async () => { + const { dbPath } = tempRoot(); + const children = [ + spawnChild(claimChildScript, [dbPath, "acme/widgets", "1"]), + spawnChild(claimChildScript, [dbPath, "acme/widgets", "2"]), + ]; + const results = await runBarriered(children); + + expect(results.every((result) => result.ok)).toBe(true); + const issueNumbers = results.map((result) => result.claim?.issueNumber).sort(); + expect(issueNumbers).toEqual([1, 2]); + + const ledger = openClaimLedger(dbPath); + try { + expect(ledger.listClaims({ repoFullName: "acme/widgets" })).toHaveLength(2); + } finally { + ledger.close(); + } + }); +}); + +type DequeueChildResult = { + ok: boolean; + entry?: { repoFullName: string; identifier: string } | null; + message?: string; +}; + +describe("portfolio-queue cross-process races (#4867)", () => { + it("two processes racing dequeueNext() on a single queued item: exactly one wins, the other gets null", async () => { + const { dbPath } = tempRoot(); + const bootstrap = initPortfolioQueueStore(dbPath); + bootstrap.enqueue({ repoFullName: "acme/widgets", identifier: "pr:1" }); + bootstrap.close(); + + const children = [spawnChild(dequeueChildScript, [dbPath]), spawnChild(dequeueChildScript, [dbPath])]; + const results = await runBarriered(children); + + expect(results.every((result) => result.ok)).toBe(true); + const winners = results.filter((result) => result.entry != null); + const empties = results.filter((result) => result.entry == null); + expect(winners).toHaveLength(1); + expect(empties).toHaveLength(1); + expect(winners[0]?.entry).toMatchObject({ repoFullName: "acme/widgets", identifier: "pr:1" }); + + const store = initPortfolioQueueStore(dbPath); + try { + const rows = store.listQueue("acme/widgets"); + expect(rows).toHaveLength(1); + expect(rows[0]?.status).toBe("in_progress"); + } finally { + store.close(); + } + }); + + it("N processes racing dequeueNext() over N queued items claim exactly N distinct items, none duplicated", async () => { + const { dbPath } = tempRoot(); + const bootstrap = initPortfolioQueueStore(dbPath); + const identifiers = ["pr:1", "pr:2", "pr:3", "pr:4"]; + for (const identifier of identifiers) { + bootstrap.enqueue({ repoFullName: "acme/widgets", identifier }); + } + bootstrap.close(); + + const children = identifiers.map(() => spawnChild(dequeueChildScript, [dbPath])); + const results = await runBarriered(children); + + expect(results.every((result) => result.ok)).toBe(true); + const claimedIdentifiers = results.map((result) => result.entry?.identifier).filter(Boolean); + expect(claimedIdentifiers).toHaveLength(identifiers.length); + expect(new Set(claimedIdentifiers).size).toBe(identifiers.length); + + const store = initPortfolioQueueStore(dbPath); + try { + const rows = store.listQueue("acme/widgets"); + expect(rows.every((row) => row.status === "in_progress")).toBe(true); + expect(rows).toHaveLength(identifiers.length); + } finally { + store.close(); + } + }); + + it("rejects the dequeue-child helper when required args are missing", async () => { + const child = spawn(process.execPath, [dequeueChildScript], { stdio: ["ignore", "pipe", "pipe"] }); + const exitCode = await new Promise((resolve, reject) => { + child.once("error", reject); + child.once("exit", resolve); + }); + expect(exitCode).toBe(2); + }); + + it("rejects the claim-child helper when required args are missing", async () => { + const child = spawn(process.execPath, [claimChildScript], { stdio: ["ignore", "pipe", "pipe"] }); + const exitCode = await new Promise((resolve, reject) => { + child.once("error", reject); + child.once("exit", resolve); + }); + expect(exitCode).toBe(2); + }); +});