diff --git a/.agents/pm/history/pm-github-503u.jsonl b/.agents/pm/history/pm-github-503u.jsonl index b24090c8..48823544 100644 --- a/.agents/pm/history/pm-github-503u.jsonl +++ b/.agents/pm/history/pm-github-503u.jsonl @@ -1 +1,6 @@ {"ts":"2026-07-17T00:17:34.148Z","author":"claude-fleet","op":"create","patch":[{"op":"add","path":"/metadata/id","value":"pm-github-503u"},{"op":"add","path":"/metadata/title","value":"Serialize marker-dedupe read-and-append across concurrent imports"},{"op":"add","path":"/metadata/description","value":"CodeRabbit (PR #33, comment 3599527301) correctly notes the native comment sync's read-markers-then-append sequence is not atomic across two concurrent 'pm github import' processes targeting the same workspace: both can observe a GitHub comment id as absent and append it twice. Individual writes go through the pm CLI (which locks per mutation), but our check-then-act spans multiple mutations. Current disposition: documented limitation (see syncGithubCommentsToAnnotations docstring) — concurrent same-repo imports are an operator edge case and worst case yields a duplicate comment, not data loss. Proper fix: cross-process lock (e.g. lockfile in workspace runtime dir) or an SDK-level conditional append (append-if-marker-absent) — the latter is worth proposing upstream."},{"op":"add","path":"/metadata/type","value":"Task"},{"op":"add","path":"/metadata/status","value":"open"},{"op":"add","path":"/metadata/priority","value":2},{"op":"add","path":"/metadata/tags","value":[]},{"op":"add","path":"/metadata/created_at","value":"2026-07-17T00:17:34.148Z"},{"op":"add","path":"/metadata/updated_at","value":"2026-07-17T00:17:34.148Z"},{"op":"add","path":"/metadata/author","value":"claude-fleet"},{"op":"add","path":"/metadata/acceptance_criteria","value":"Concurrent 'pm github import --comments-mode annotations' runs against the same workspace never duplicate a comment (verified by a parallel-import test), or the limitation is lifted via an upstream SDK conditional-append primitive"}],"before_hash":"3cc22dff72be7b14824654a7a64ea62b04799939b2fee54c1b5f52ca60bf6df0","after_hash":"23571755a1196598f1bd247d25fd76b4fc2f8c0a1867418368ba7a36fcc42891","message":""} +{"ts":"2026-07-18T09:54:19.859Z","author":"pi-agent","op":"claim","patch":[{"op":"replace","path":"/metadata/updated_at","value":"2026-07-18T09:54:19.859Z"},{"op":"add","path":"/metadata/assignee","value":"pi-agent"}],"before_hash":"23571755a1196598f1bd247d25fd76b4fc2f8c0a1867418368ba7a36fcc42891","after_hash":"399675b8b100dec1a3fca002f3de0968ea3444b2d92cc2b9ce1ff39209a18ba1"} +{"ts":"2026-07-18T09:54:35.851Z","author":"pi-agent","op":"update","patch":[{"op":"replace","path":"/metadata/updated_at","value":"2026-07-18T09:54:35.851Z"},{"op":"replace","path":"/metadata/status","value":"in_progress"}],"before_hash":"399675b8b100dec1a3fca002f3de0968ea3444b2d92cc2b9ce1ff39209a18ba1","after_hash":"9e170f0cba9f28d818380738d5794ca1af1a747f9fe0c829cea8928f16373a04"} +{"ts":"2026-07-18T09:54:36.132Z","author":"pi-agent","op":"note_add","patch":[{"op":"replace","path":"/metadata/updated_at","value":"2026-07-18T09:54:36.132Z"},{"op":"add","path":"/metadata/notes","value":[{"created_at":"2026-07-18T09:54:36.132Z","author":"pi-agent","text":"Started work in worktree .wt-pm-github-503u on branch feat/serialize-comment-import-lock. Plan: cross-process lockfile around comment marker read->append critical section, stale-lock TTL break, parallel-import test + lock unit tests."}]}],"before_hash":"9e170f0cba9f28d818380738d5794ca1af1a747f9fe0c829cea8928f16373a04","after_hash":"85e9f5570fa8e6972c3702c9f0fbad70eb13d49c72c5a415006e22fca60c7149"} +{"ts":"2026-07-18T10:20:49.442Z","author":"pi-agent","op":"note_add","patch":[{"op":"add","path":"/metadata/notes/1","value":{"created_at":"2026-07-18T10:20:49.441Z","author":"pi-agent","text":"Implemented cross-process serialization of the comment marker read->append critical section: acquireImportLock() creates /locks/pm-github.comment-sync..lock via fs.openSync(wx) (atomic O_EXCL), payload mirrors pm CLI lock convention ({id,pid,owner,created_at,ttl_seconds}) so pm gc can sweep it. Stale locks (older than 5min TTL, or dead owner PID) broken with stderr warning; unparseable payloads fall back to mtime so a live writer mid-write is never broken. Live contention waits with jittered backoff up to 30s; on timeout the item's comment sync is skipped with a warning (never run unlocked), on lock-mechanism failure it degrades to unlocked with a warning. Lock scope is per item, not per import run: unrelated issues in concurrent imports still sync in parallel. Tests: 10 new in test/import-lock.test.ts incl. cross-process acceptance test spawning two real concurrent child syncs against one workspace -> 20 unique comments, zero duplicates, loser skips all. Full suite 122/122 green x3 runs, typecheck clean."}},{"op":"replace","path":"/metadata/updated_at","value":"2026-07-18T10:20:49.442Z"}],"before_hash":"85e9f5570fa8e6972c3702c9f0fbad70eb13d49c72c5a415006e22fca60c7149","after_hash":"e8707e73f81fe4d12e12f2c5ba4b47d9074e43c86d278856b2e6c7936a2ba6bc"} +{"ts":"2026-07-18T10:21:18.914Z","author":"pi-agent","op":"close","patch":[{"op":"remove","path":"/metadata/assignee"},{"op":"replace","path":"/metadata/updated_at","value":"2026-07-18T10:21:18.914Z"},{"op":"replace","path":"/metadata/status","value":"closed"},{"op":"add","path":"/metadata/closed_at","value":"2026-07-18T10:21:18.913Z"},{"op":"add","path":"/metadata/close_reason","value":"Fixed: cross-process per-item lockfile (acquireImportLock in index.ts) serializes the comment marker read->append section across concurrent pm github import processes. Stale locks broken by 5min TTL or dead-PID check with stderr warning; live contention waits up to 30s then skips the item's sync with a warning (never appends unlocked). Acceptance verified by parallel-import test: two concurrent OS processes syncing 20 identical comments into one workspace item produce exactly 20 stored comments, zero duplicate markers; lock unit tests cover acquire/contend/stale-break. Full suite 122/122 green."}],"before_hash":"e8707e73f81fe4d12e12f2c5ba4b47d9074e43c86d278856b2e6c7936a2ba6bc","after_hash":"9cc3905782b940ea4d58223ba582530eea2fb57495475eb94fac36791693b09d"} diff --git a/.agents/pm/tasks/pm-github-503u.toon b/.agents/pm/tasks/pm-github-503u.toon index 8e28face..df5ab826 100644 --- a/.agents/pm/tasks/pm-github-503u.toon +++ b/.agents/pm/tasks/pm-github-503u.toon @@ -2,11 +2,16 @@ id: pm-github-503u title: Serialize marker-dedupe read-and-append across concurrent imports description: "CodeRabbit (PR #33, comment 3599527301) correctly notes the native comment sync's read-markers-then-append sequence is not atomic across two concurrent 'pm github import' processes targeting the same workspace: both can observe a GitHub comment id as absent and append it twice. Individual writes go through the pm CLI (which locks per mutation), but our check-then-act spans multiple mutations. Current disposition: documented limitation (see syncGithubCommentsToAnnotations docstring) — concurrent same-repo imports are an operator edge case and worst case yields a duplicate comment, not data loss. Proper fix: cross-process lock (e.g. lockfile in workspace runtime dir) or an SDK-level conditional append (append-if-marker-absent) — the latter is worth proposing upstream." type: Task -status: open +status: closed priority: 2 tags: [] created_at: "2026-07-17T00:17:34.148Z" -updated_at: "2026-07-17T00:17:34.148Z" +updated_at: "2026-07-18T10:21:18.914Z" +closed_at: "2026-07-18T10:21:18.913Z" author: claude-fleet acceptance_criteria: "Concurrent 'pm github import --comments-mode annotations' runs against the same workspace never duplicate a comment (verified by a parallel-import test), or the limitation is lifted via an upstream SDK conditional-append primitive" +notes[2]{created_at,author,text}: + "2026-07-18T09:54:36.132Z",pi-agent,"Started work in worktree .wt-pm-github-503u on branch feat/serialize-comment-import-lock. Plan: cross-process lockfile around comment marker read->append critical section, stale-lock TTL break, parallel-import test + lock unit tests." + "2026-07-18T10:20:49.441Z",pi-agent,"Implemented cross-process serialization of the comment marker read->append critical section: acquireImportLock() creates /locks/pm-github.comment-sync..lock via fs.openSync(wx) (atomic O_EXCL), payload mirrors pm CLI lock convention ({id,pid,owner,created_at,ttl_seconds}) so pm gc can sweep it. Stale locks (older than 5min TTL, or dead owner PID) broken with stderr warning; unparseable payloads fall back to mtime so a live writer mid-write is never broken. Live contention waits with jittered backoff up to 30s; on timeout the item's comment sync is skipped with a warning (never run unlocked), on lock-mechanism failure it degrades to unlocked with a warning. Lock scope is per item, not per import run: unrelated issues in concurrent imports still sync in parallel. Tests: 10 new in test/import-lock.test.ts incl. cross-process acceptance test spawning two real concurrent child syncs against one workspace -> 20 unique comments, zero duplicates, loser skips all. Full suite 122/122 green x3 runs, typecheck clean." +close_reason: "Fixed: cross-process per-item lockfile (acquireImportLock in index.ts) serializes the comment marker read->append section across concurrent pm github import processes. Stale locks broken by 5min TTL or dead-PID check with stderr warning; live contention waits up to 30s then skips the item's sync with a warning (never appends unlocked). Acceptance verified by parallel-import test: two concurrent OS processes syncing 20 identical comments into one workspace item produce exactly 20 stored comments, zero duplicate markers; lock unit tests cover acquire/contend/stale-break. Full suite 122/122 green." body: "" diff --git a/CHANGELOG.md b/CHANGELOG.md index b2cf2a68..a0735a15 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### Other + +- Serialize marker-dedupe read-and-append across concurrent imports ([pm-github-503u](https://github.com/unbraind/pm-github/blob/main/.agents/pm/tasks/pm-github-503u.toon)) + ## 2026.7.17 - 2026-07-17 ### Added diff --git a/index.ts b/index.ts index f1adb44c..4f9cc325 100644 --- a/index.ts +++ b/index.ts @@ -14,6 +14,9 @@ // section below and the pure plan/mapping logic in ./projects.ts). import https from "node:https"; +import crypto from "node:crypto"; +import fs from "node:fs"; +import path from "node:path"; import { spawnSync } from "node:child_process"; import type { defineExtension as defineExtensionType } from "@unbrained/pm-cli/sdk"; @@ -708,6 +711,347 @@ export function composeBody(issue: GhIssue, comments: GhComment[]): string { return body; } +// --------------------------------------------------------------------------- +// Cross-process comment-sync lock (serializes marker dedupe, pm-github-503u) +// --------------------------------------------------------------------------- +// +// The native comment sync below does a read-markers-then-append sequence that +// spans several pm CLI mutations. Each individual mutation is locked by the pm +// CLI, but the check-then-act as a whole is not atomic across two concurrent +// `pm github import` processes on the same workspace: both can observe a +// GitHub comment id as absent and append it twice. The helper here closes that +// race with a lockfile in the workspace's own locks/ dir, following the pm +// CLI's locking convention (see core/lock/lock.js in pm-cli): +// +// - location /locks/pm-github.comment-sync..lock — the +// same locks/ dir the pm CLI itself uses (gitignored runtime state), with +// a payload shape identical to the CLI's so `pm gc` can sweep our stale +// locks by their embedded ttl_seconds. The "pm-github.comment-sync." +// prefix keeps our namespace disjoint from the CLI's per-item mutation +// locks (.lock): we hold our lock ACROSS several CLI mutations, so +// sharing a name would make the very mutations we wrap conflict with it. +// - acquire fs.openSync(path, "wx") — atomic O_EXCL create on POSIX. +// - stale a lock is broken (with a stderr warning) only when its owner +// is provably gone: the recorded owner PID is dead, or it equals our own +// PID (we know we don't hold it, so the PID was recycled). A lock with a +// LIVE owner is never age-broken — a slow-but-alive holder keeps its lock +// no matter how long it runs (a >TTL holder would otherwise lose the lock +// mid-append and the race would reopen). Only when the payload is missing +// or unparseable (caught mid-write, then abandoned) does the TTL (default +// 5 min, age from file mtime) apply as the break criterion. +// - release token-checked: each acquisition embeds a unique token in the +// payload and release() unlinks the file only while it still carries that +// token, so a holder whose lock was stale-broken (or swept by `pm gc`) +// can never unlink a successor's lock. +// - breaking stale locks are removed under a breaker election (an O_EXCL +// `.break` sidecar): only the single election winner may unlink, +// and it re-verifies staleness under that mutex first — two concurrent +// breakers can therefore never double-break, and a breaker can never +// unlink a fresh lock that replaced the stale one it inspected. +// - contend a live, fresh lock is waited out with jittered backoff up to a +// budget (default 30s). On timeout the caller reports "contended" and the +// comment sync for that item is SKIPPED (never run unlocked) — a wedged +// concurrent import can then cost a comment sync, recoverable by +// re-running import, but can never produce a duplicate comment. When the +// lock mechanism itself is unavailable (read-only fs etc.) the caller +// reports "degraded" and proceeds unlocked with a warning, matching the +// pre-lock best-effort behavior. + +export const IMPORT_LOCK_TTL_MS_DEFAULT = 5 * 60_000; +export const IMPORT_LOCK_WAIT_MS_DEFAULT = 30_000; + +// Payload written into the lock file. Mirrors the pm CLI's own lock payload so +// the CLI's `pm gc` lock sweep (which reads ttl_seconds) treats our locks +// exactly like its own. +export interface ImportLockPayload { + id: string; + pid: number; + owner: string; + /** Unique per acquisition; release() only unlinks a file carrying it. */ + token: string; + created_at: string; + ttl_seconds: number; +} + +export interface ImportLock { + /** Absolute path of the held lock file (diagnostics/tests). */ + path: string; + /** Release the lock. Best-effort, idempotent, never throws. */ + release(): void; +} + +export type ImportLockAcquisition = + | { status: "acquired"; lock: ImportLock } + // Another live process held the lock past the wait budget. + | { status: "contended" } + // The lock mechanism itself failed (fs error) — caller may proceed unlocked. + | { status: "degraded" }; + +// Resolve the pm data dir (the dir holding settings.json and locks/) from the +// pmRoot the extension host hands the command, which may be either the +// workspace root (the dir containing .agents/pm) or the data dir itself — the +// pm CLI accepts both for --path, and tests pass the workspace root. +export function resolvePmDataDir(pmRoot: string): string { + const nested = path.join(pmRoot, ".agents", "pm"); + try { + if (fs.statSync(nested).isDirectory()) return nested; + } catch { + // Not the workspace-root form — assume pmRoot already is the data dir. + } + return pmRoot; +} + +// Absolute lock file path for one item's comment-sync critical section. The +// item id is sanitized defensively (pm ids are already filename-safe, but the +// lock path must never become a traversal vector if that ever changes). +export function importCommentSyncLockPath(pmRoot: string, itemId: string): string { + const safe = itemId.replace(/[^A-Za-z0-9._-]/g, "_"); + return path.join(resolvePmDataDir(pmRoot), "locks", `pm-github.comment-sync.${safe}.lock`); +} + +function readImportLockPayload(lockPath: string): ImportLockPayload | undefined { + let raw: string; + try { + raw = fs.readFileSync(lockPath, "utf8"); + } catch { + return undefined; + } + try { + const parsed = JSON.parse(raw); + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) return undefined; + return parsed as ImportLockPayload; + } catch { + return undefined; + } +} + +function isLockOwnerAlive(pid: unknown): boolean { + if (typeof pid !== "number" || !Number.isInteger(pid) || pid <= 0) return false; + try { + process.kill(pid, 0); + return true; + } catch (err: unknown) { + // EPERM means the process exists but belongs to another user — still alive. + // ESRCH means no such process — the lock owner is gone. + return (err as NodeJS.ErrnoException)?.code === "EPERM"; + } +} + +// A held lock is stale only when its owner is provably gone: the recorded +// owner PID is dead (crash without release), or it equals our own PID (we +// know we do not hold this lock, so the PID must have been recycled). A lock +// with a LIVE recorded owner is never age-broken — breaking a slow-but-alive +// holder mid-critical-section would reopen the duplicate-marker race this +// lock exists to close. Only when there is no usable owner PID (payload +// missing or unparseable — e.g. caught mid-write, then abandoned) does the +// TTL apply, with age taken from the payload's created_at falling back to +// the file mtime. (PID reuse can make a dead lock look alive and thus block +// until the wait budget — conservative; it can never break a live one.) +// Lock paths currently held by THIS process (any async task). Distinguishes a +// same-pid payload that is legitimately ours (in-process contention → wait) +// from a leftover written by a dead process whose PID we inherited (→ break). +const heldImportLocks = new Set(); + +function importLockStaleReason( + lockPath: string, + payload: ImportLockPayload | undefined, + mtimeMs: number, + ttlMs: number, +): string | undefined { + const pid = payload?.pid; + if (typeof pid === "number" && Number.isInteger(pid) && pid > 0) { + if (pid === process.pid) { + // Our own PID: either another async task in THIS process holds it (the + // in-process race the caller serializes too — contend, don't break), or + // we provably don't hold it and the PID was recycled by a dead owner. + return heldImportLocks.has(lockPath) + ? undefined + : `owner pid ${pid} is this process but the lock is not held in-process (pid recycled, lock abandoned)`; + } + return isLockOwnerAlive(pid) ? undefined : `owner pid ${pid} is dead`; + } + const createdMs = payload ? Date.parse(payload.created_at) : Number.NaN; + const ageBase = Number.isFinite(createdMs) ? createdMs : mtimeMs; + if (!Number.isFinite(ageBase) || Date.now() - ageBase > ttlMs) { + return `no usable owner pid and older than TTL ${Math.round(ttlMs / 1000)}s`; + } + return undefined; +} + +const IMPORT_LOCK_INITIAL_BACKOFF_MS = 25; +const IMPORT_LOCK_MAX_BACKOFF_MS = 200; +const IMPORT_LOCK_MAX_STALE_BREAKS = 3; +// A breaker's critical section is a handful of syscalls (re-stat, re-read, +// unlink) — a crashed breaker's sidecar goes stale in seconds, not minutes. +export const IMPORT_LOCK_BREAKER_TTL_MS = 10_000; + +// Break a stale lock under a breaker election so the unlink can never race +// another breaker: two contenders may both judge the same file stale, but +// without mutual exclusion the slower one would execute its unlink AFTER the +// winner already re-created a fresh, live lock — silently unlinking that +// replacement and letting both proceed (the exact double-acquire this module +// exists to prevent). The election is an O_EXCL sidecar (`.break`): +// only its single winner may unlink, and it re-verifies staleness under the +// mutex first, so a lock that became live since the caller's check survives. +// Returns true when the stale lock was removed (caller should retry acquire +// immediately), false when the election was lost or the lock is live again +// (caller should fall through to the normal wait/backoff). +function breakStaleImportLock(lockPath: string, ttlMs: number, reason: string): boolean { + const breakerPath = `${lockPath}.break`; + let bfd: number | undefined; + try { + bfd = fs.openSync(breakerPath, "wx"); + } catch (err: unknown) { + if ((err as NodeJS.ErrnoException)?.code === "EEXIST") { + // Another breaker is active. If IT crashed mid-break, its sidecar is + // old — clear it and let the next iteration re-run the election. + try { + if (Date.now() - fs.statSync(breakerPath).mtimeMs > IMPORT_LOCK_BREAKER_TTL_MS) { + fs.unlinkSync(breakerPath); + } + } catch { + // Sidecar vanished (election finished) — retry normally. + } + } + return false; + } + try { + fs.closeSync(bfd); + // Re-verify under the mutex: the lock we judged stale may have been + // broken and re-acquired by a live owner since we looked at it. + let payload: ImportLockPayload | undefined; + let mtimeMs = Number.NaN; + try { + mtimeMs = fs.statSync(lockPath).mtimeMs; + payload = readImportLockPayload(lockPath); + } catch { + return true; // already gone — acquire can proceed immediately + } + if (!importLockStaleReason(lockPath, payload, mtimeMs, ttlMs)) return false; + console.error(`pm-github: breaking stale comment-sync lock ${lockPath} (${reason})`); + try { + fs.unlinkSync(lockPath); + } catch { + // Already gone. + } + return true; + } finally { + try { + fs.unlinkSync(breakerPath); + } catch { + // Best-effort: an aged-out sidecar may have been cleared by a peer. + } + } +} + +// Acquire the cross-process comment-sync lock for one item. Never throws: +// every failure mode maps to one of the three acquisition statuses so the +// caller can decide how to degrade. `ttlMs`/`waitMs` are injectable for tests. +export async function acquireImportLock( + pmRoot: string, + itemId: string, + opts: { ttlMs?: number; waitMs?: number } = {}, +): Promise { + const ttlMs = opts.ttlMs ?? IMPORT_LOCK_TTL_MS_DEFAULT; + const waitMs = opts.waitMs ?? IMPORT_LOCK_WAIT_MS_DEFAULT; + let lockPath: string; + try { + lockPath = importCommentSyncLockPath(pmRoot, itemId); + fs.mkdirSync(path.dirname(lockPath), { recursive: true }); + } catch (err: unknown) { + const msg = err instanceof Error ? err.message : String(err); + console.error(`pm-github: comment-sync lock unavailable (${msg}) — proceeding without cross-process serialization`); + return { status: "degraded" }; + } + const payload: ImportLockPayload = { + id: path.basename(lockPath, ".lock"), + pid: process.pid, + owner: "pm-github", + token: crypto.randomUUID(), + created_at: new Date().toISOString(), + ttl_seconds: Math.ceil(ttlMs / 1000), + }; + const startedAt = Date.now(); + let backoffMs = IMPORT_LOCK_INITIAL_BACKOFF_MS; + let staleBreaks = 0; + for (;;) { + let fd: number | undefined; + try { + fd = fs.openSync(lockPath, "wx"); + fs.writeFileSync(fd, `${JSON.stringify(payload, null, 2)}\n`); + fs.closeSync(fd); + fd = undefined; + heldImportLocks.add(lockPath); + let released = false; + return { + status: "acquired", + lock: { + path: lockPath, + release() { + if (released) return; + released = true; + heldImportLocks.delete(lockPath); + try { + // Token check: only unlink the file while it is still OUR lock. + // If it was stale-broken or gc-swept and re-acquired, the path + // now holds the successor's lock — leave it alone. + const current = readImportLockPayload(lockPath); + if (current?.token !== payload.token) return; + fs.unlinkSync(lockPath); + } catch { + // Best-effort: already gone (stale-broken by someone else, gc). + } + }, + }, + }; + } catch (err: unknown) { + if (fd !== undefined) { + try { + fs.closeSync(fd); + } catch { + // Ignore close errors on the failure path. + } + // We created the lockfile but failed to write/close it. Remove the + // empty/partial file so other processes are not blocked until the + // TTL expires on a lock that was never validly held. + try { + fs.unlinkSync(lockPath); + } catch { + // Best-effort: already gone. + } + } + const code = (err as NodeJS.ErrnoException)?.code; + if (code !== "EEXIST") { + const msg = err instanceof Error ? err.message : String(err); + console.error(`pm-github: comment-sync lock failed (${msg}) — proceeding without cross-process serialization`); + return { status: "degraded" }; + } + // The lock exists. If it vanished before we could stat it, retry at once. + let existing: ImportLockPayload | undefined; + let mtimeMs = Number.NaN; + try { + mtimeMs = fs.statSync(lockPath).mtimeMs; + existing = readImportLockPayload(lockPath); + } catch { + continue; + } + const staleReason = importLockStaleReason(lockPath, existing, mtimeMs, ttlMs); + if (staleReason && staleBreaks < IMPORT_LOCK_MAX_STALE_BREAKS) { + staleBreaks++; + if (breakStaleImportLock(lockPath, ttlMs, staleReason)) continue; + // Lost the breaker election or the lock turned out live on re-check — + // fall through to the normal wait/backoff below. + } + const elapsedMs = Date.now() - startedAt; + if (elapsedMs >= waitMs) return { status: "contended" }; + // Jittered backoff, mirroring the pm CLI's lock wait (0.5x–1.5x jitter). + const jittered = Math.max(1, Math.round(backoffMs * (0.5 + Math.random()))); + await sleep(Math.min(jittered, waitMs - elapsedMs)); + backoffMs = Math.min(backoffMs * 2, IMPORT_LOCK_MAX_BACKOFF_MS); + } + } +} + // --------------------------------------------------------------------------- // Native comment sync (GitHub issue comments → pm comments collection) // --------------------------------------------------------------------------- @@ -765,11 +1109,19 @@ export function parseCreatedItemId(stdout: string): string | undefined { // duplicates. Each GitHub comment becomes one pm comment authored by the // GitHub login. Failures are logged and never abort the import. // -// Known limitation (tracked in pm-github-503u): the read-markers-then-append -// sequence is not atomic across processes. Two concurrent imports of the same -// repository into the same workspace can both observe a comment id as absent -// and append it twice. Individual appends are still atomic (pm CLI locks per -// mutation), so the worst case is a duplicate comment, never corruption. +// Concurrency (pm-github-503u, limitation lifted): the read-markers-then-append +// critical section is serialized across processes by a per-item lockfile (see +// acquireImportLock above), so two concurrent `pm github import` runs against +// the same workspace can no longer both observe a comment id as absent and +// append it twice. The lock is per ITEM, not per import run: unrelated issues +// in concurrent imports still sync in parallel, and only the one item whose +// comments are actively being synced is serialized — the smallest scope that +// closes the race. Stale locks (older than the TTL or owned by a dead PID) are +// broken with a stderr warning. If a live concurrent import holds the lock +// past the wait budget, this item's comment sync is skipped with a warning +// (never run unlocked, so a wedged peer can cost a sync — recoverable on the +// next import — but never cause a duplicate); if the lock mechanism itself is +// unavailable the sync proceeds unlocked as before, also with a warning. export async function syncGithubCommentsToAnnotations( itemId: string, comments: GhComment[], @@ -777,35 +1129,48 @@ export async function syncGithubCommentsToAnnotations( issueNumber: number, ): Promise<{ added: number; skipped: number }> { if (comments.length === 0) return { added: 0, skipped: 0 }; - let pmComments: PmCommentsFn; - let existing: { text?: string }[] = []; - try { - pmComments = await loadPmComments(); - const list = await pmComments(itemId, {}, { pmRoot }); - existing = (list?.comments ?? []) as { text?: string }[]; - } catch (err: unknown) { - const msg = err instanceof Error ? err.message : String(err); - console.error(`#${issueNumber}: could not read existing comments for ${itemId} — ${msg}`); + const acquisition = await acquireImportLock(pmRoot, itemId); + if (acquisition.status === "contended") { + console.error( + `#${issueNumber}: comment sync for ${itemId} skipped — another import holds the ` + + `comment-sync lock; re-run import to pick up the comments`, + ); return { added: 0, skipped: 0 }; } - const synced = extractSyncedCommentIds(existing); - let added = 0; - let skipped = 0; - for (const c of comments) { - if (synced.has(c.id)) { - skipped++; - continue; - } - const author = c.user?.login ?? "github"; + const release = acquisition.status === "acquired" ? () => acquisition.lock.release() : () => {}; + try { + let pmComments: PmCommentsFn; + let existing: { text?: string }[] = []; try { - await pmComments(itemId, { add: buildCommentText(c), author }, { pmRoot }); - added++; + pmComments = await loadPmComments(); + const list = await pmComments(itemId, {}, { pmRoot }); + existing = (list?.comments ?? []) as { text?: string }[]; } catch (err: unknown) { const msg = err instanceof Error ? err.message : String(err); - console.error(`#${issueNumber}: comment ${c.id} sync failed — ${msg}`); + console.error(`#${issueNumber}: could not read existing comments for ${itemId} — ${msg}`); + return { added: 0, skipped: 0 }; + } + const synced = extractSyncedCommentIds(existing); + let added = 0; + let skipped = 0; + for (const c of comments) { + if (synced.has(c.id)) { + skipped++; + continue; + } + const author = c.user?.login ?? "github"; + try { + await pmComments(itemId, { add: buildCommentText(c), author }, { pmRoot }); + added++; + } catch (err: unknown) { + const msg = err instanceof Error ? err.message : String(err); + console.error(`#${issueNumber}: comment ${c.id} sync failed — ${msg}`); + } } + return { added, skipped }; + } finally { + release(); } - return { added, skipped }; } // A draft PR is an issue that is both a pull request and flagged `draft: true`. diff --git a/test/helpers/comment-sync-child.mjs b/test/helpers/comment-sync-child.mjs new file mode 100644 index 00000000..a48578ab --- /dev/null +++ b/test/helpers/comment-sync-child.mjs @@ -0,0 +1,39 @@ +// Child-process entry for the cross-process concurrent comment-sync test. +// +// Spawning real OS processes (rather than two in-process calls) exercises the +// lockfile serialization the way two concurrent `pm github import` runs hit +// it: separate processes, same workspace, same item, same GitHub comments. +// +// Usage: node test/helpers/comment-sync-child.mjs +// Input: FAKE_COMMENTS env var — JSON array of GhComment-shaped objects. +// BARRIER_FILE env var (optional) — path both children poll for before +// starting, so the parent can line them up and the test exercises real +// concurrent entry into the sync instead of an accidental sequence. +// Output: one JSON line on stdout with the { added, skipped } result. + +import { existsSync } from "node:fs"; +import { setTimeout as delay } from "node:timers/promises"; + +import { syncGithubCommentsToAnnotations } from "../../dist/index.js"; + +const [itemId, pmRoot] = process.argv.slice(2); +if (!itemId || !pmRoot) { + console.error("usage: comment-sync-child.mjs "); + process.exit(2); +} + +const barrier = process.env.BARRIER_FILE; +if (barrier) { + const deadline = Date.now() + 10_000; + while (!existsSync(barrier)) { + if (Date.now() > deadline) { + console.error("barrier file never appeared"); + process.exit(2); + } + await delay(5); + } +} + +const comments = JSON.parse(process.env.FAKE_COMMENTS || "[]"); +const result = await syncGithubCommentsToAnnotations(itemId, comments, pmRoot, 1); +process.stdout.write(`${JSON.stringify(result)}\n`); diff --git a/test/import-lock.test.ts b/test/import-lock.test.ts new file mode 100644 index 00000000..cf47c570 --- /dev/null +++ b/test/import-lock.test.ts @@ -0,0 +1,496 @@ +// Tests for the cross-process comment-sync lock (pm-github-503u). +// +// Unit tests cover the lock helper in isolation: acquire/release roundtrip +// (incl. token-checked release), contention (wait → contended → acquire after +// release), and stale-lock breaking (dead owner PID, recycled own PID, the +// mtime-TTL fallback for unparseable payloads — and, critically, that a LIVE +// owner is never age-broken). The integration test spawns two REAL concurrent child processes +// syncing the same GitHub comments into the same item of a throwaway +// workspace — the exact `pm github import` race from pm-github-503u — and +// asserts no duplicate comment markers result. + +import assert from "node:assert/strict"; +import test from "node:test"; +import { execFile, execFileSync, spawn, spawnSync } from "node:child_process"; +import { + existsSync, + mkdirSync, + mkdtempSync, + readFileSync, + rmSync, + utimesSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { + IMPORT_LOCK_TTL_MS_DEFAULT, + acquireImportLock, + importCommentSyncLockPath, + resolvePmDataDir, + syncGithubCommentsToAnnotations, +} from "../dist/index.js"; + +// Minimal factories + workspace helpers -------------------------------------- + +function ghComment(overrides: Record = {}): any { + return { + id: 1001, + user: { login: "alice" }, + created_at: "2026-01-01T00:00:00Z", + body: "looks good to me", + ...overrides, + }; +} + +// Resolve the platform-appropriate pm shim (win32 needs pm.cmd via shell). +const PM_BIN = fileURLToPath( + new URL(`../node_modules/.bin/pm${process.platform === "win32" ? ".cmd" : ""}`, import.meta.url), +); + +const CHILD_SCRIPT = fileURLToPath( + new URL("../test/helpers/comment-sync-child.mjs", import.meta.url), +); + +function makeWorkspace(): string { + const root = mkdtempSync(join(tmpdir(), "pm-github-lock-test-")); + const env = { ...process.env, PM_AUTHOR: "tester" }; + const opts = { stdio: "ignore" as const, env, shell: process.platform === "win32" }; + try { + try { + execFileSync(PM_BIN, ["init", "-y", "--force", "--workspace", root, "--author", "tester"], opts); + } catch { + execFileSync(PM_BIN, ["init", "-y", "--force", root, "--author", "tester"], opts); + } + execFileSync(PM_BIN, ["--pm-path", root, "create", "task", "Synced item", "--description", "d"], opts); + return root; + } catch (err) { + rmSync(root, { recursive: true, force: true }); + throw err; + } +} + +function firstItemId(root: string): string { + const out = execFileSync(PM_BIN, ["--pm-path", root, "--json", "list", "--full"], { encoding: "utf-8" }); + const parsed = JSON.parse(out); + const arr = Array.isArray(parsed) ? parsed : (parsed.items ?? parsed.results ?? []); + if (!arr.length) throw new Error("no items in test workspace"); + return arr[0].id as string; +} + +// Run fn with console.error captured; returns the messages it logged. +async function captureStderr(fn: () => Promise): Promise { + const original = console.error; + const messages: string[] = []; + console.error = (...args: unknown[]) => { + messages.push(args.map(String).join(" ")); + }; + try { + await fn(); + } finally { + console.error = original; + } + return messages; +} + +// Plant a lock file as if another process created it. +function plantLock(root: string, itemId: string, payload: unknown): string { + const lockPath = importCommentSyncLockPath(root, itemId); + mkdirSync(dirname(lockPath), { recursive: true }); + writeFileSync(lockPath, typeof payload === "string" ? payload : `${JSON.stringify(payload)}\n`); + return lockPath; +} + +// resolvePmDataDir / importCommentSyncLockPath -------------------------------- + +test("resolvePmDataDir prefers a nested .agents/pm and falls back to pmRoot itself", () => { + const root = mkdtempSync(join(tmpdir(), "pm-github-datadir-test-")); + try { + assert.strictEqual(resolvePmDataDir(root), root, "no .agents/pm → pmRoot is the data dir"); + mkdirSync(join(root, ".agents", "pm"), { recursive: true }); + assert.strictEqual(resolvePmDataDir(root), join(root, ".agents", "pm")); + // A pmRoot that already IS the data dir keeps working. + assert.strictEqual(resolvePmDataDir(join(root, ".agents", "pm")), join(root, ".agents", "pm")); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("importCommentSyncLockPath lands in locks/ with a pm-github prefix and sanitizes the item id", () => { + const root = mkdtempSync(join(tmpdir(), "pm-github-lockpath-test-")); + try { + mkdirSync(join(root, ".agents", "pm"), { recursive: true }); + const p = importCommentSyncLockPath(root, "pm-ab12"); + assert.strictEqual(p, join(root, ".agents", "pm", "locks", "pm-github.comment-sync.pm-ab12.lock")); + // Filename-unsafe characters are replaced, so the lock always stays inside + // the locks dir even for adversarial item ids. + const nasty = importCommentSyncLockPath(root, "../evil/x"); + assert.strictEqual( + nasty, + join(root, ".agents", "pm", "locks", "pm-github.comment-sync..._evil_x.lock"), + "path separators cannot escape the locks dir", + ); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +// acquire / release ------------------------------------------------------------ + +test("acquireImportLock creates a CLI-convention payload and release removes it", async () => { + const root = makeWorkspace(); + try { + const acq = await acquireImportLock(root, "pm-lock1"); + assert.strictEqual(acq.status, "acquired"); + if (acq.status !== "acquired") return; + assert.ok(existsSync(acq.lock.path), "lock file exists while held"); + assert.ok(acq.lock.path.startsWith(join(root, ".agents", "pm", "locks")), "lock is in the workspace locks dir"); + const payload = JSON.parse(readFileSync(acq.lock.path, "utf-8")); + assert.strictEqual(payload.pid, process.pid); + assert.strictEqual(payload.owner, "pm-github"); + assert.strictEqual(payload.ttl_seconds, Math.ceil(IMPORT_LOCK_TTL_MS_DEFAULT / 1000)); + assert.ok(Number.isFinite(Date.parse(payload.created_at)), "created_at is an ISO timestamp"); + assert.ok(typeof payload.token === "string" && payload.token.length > 0, "payload carries an ownership token"); + acq.lock.release(); + assert.ok(!existsSync(acq.lock.path), "release removes the lock file"); + acq.lock.release(); // idempotent — must not throw + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("a live held lock contends: second acquire times out, then succeeds after release", async () => { + const root = makeWorkspace(); + try { + const first = await acquireImportLock(root, "pm-lock2"); + assert.strictEqual(first.status, "acquired"); + + // While held (fresh, own PID alive): not stale → waits out the budget. + const started = Date.now(); + const second = await acquireImportLock(root, "pm-lock2", { waitMs: 250 }); + assert.strictEqual(second.status, "contended"); + assert.ok(Date.now() - started >= 200, "actually waited for the budget"); + + if (first.status === "acquired") first.lock.release(); + const third = await acquireImportLock(root, "pm-lock2", { waitMs: 1000 }); + assert.strictEqual(third.status, "acquired"); + if (third.status === "acquired") third.lock.release(); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("a waiter acquires the lock once the holder releases mid-wait", async () => { + const root = makeWorkspace(); + try { + const first = await acquireImportLock(root, "pm-lock3"); + assert.strictEqual(first.status, "acquired"); + const timer = setTimeout(() => { + if (first.status === "acquired") first.lock.release(); + }, 200); + const second = await acquireImportLock(root, "pm-lock3", { waitMs: 5000 }); + clearTimeout(timer); + assert.strictEqual(second.status, "acquired", "waiter gets the lock after release, not a contention timeout"); + if (second.status === "acquired") second.lock.release(); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +// Stale-lock breaking ---------------------------------------------------------- + +test("a lock recorded under our own PID is broken (recycled pid), even when fresh", async () => { + const root = makeWorkspace(); + try { + const lockPath = plantLock(root, "pm-stale1", { + id: "pm-github.comment-sync.pm-stale1", + pid: process.pid, // we demonstrably don't hold it → the pid was recycled + owner: "pm-github", + created_at: new Date().toISOString(), // fresh — age is irrelevant here + ttl_seconds: 300, + }); + const messages = await captureStderr(async () => { + const acq = await acquireImportLock(root, "pm-stale1", { waitMs: 1000 }); + assert.strictEqual(acq.status, "acquired", "recycled-pid lock is broken and re-acquired"); + if (acq.status === "acquired") { + const payload = JSON.parse(readFileSync(acq.lock.path, "utf-8")); + assert.strictEqual(payload.pid, process.pid, "the broken lock is replaced with our payload"); + acq.lock.release(); + } + }); + assert.ok( + messages.some((m) => m.includes("breaking stale comment-sync lock") && m.includes("recycled")), + `expected a recycled-pid stale-break warning, got: ${messages.join(" | ")}`, + ); + assert.ok(!existsSync(lockPath), "lock released at the end"); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("an over-TTL lock with a LIVE foreign owner is NOT broken — the waiter contends", async () => { + const root = makeWorkspace(); + // A real live foreign process: a child that sleeps well past the test. + const holder = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60_000)"], { stdio: "ignore" }); + try { + assert.ok(holder.pid && holder.pid > 0, "need a live child pid for this test"); + plantLock(root, "pm-live1", { + id: "pm-github.comment-sync.pm-live1", + pid: holder.pid, + owner: "pm-github", + created_at: new Date(Date.now() - 10 * 60_000).toISOString(), // way past TTL + ttl_seconds: 300, + }); + const acq = await acquireImportLock(root, "pm-live1", { waitMs: 300 }); + assert.strictEqual( + acq.status, + "contended", + "a slow-but-alive holder keeps its lock: age alone must never break a live owner's lock", + ); + } finally { + holder.kill("SIGKILL"); + rmSync(root, { recursive: true, force: true }); + } +}); + +test("release() is token-checked: a broken-then-replaced lock is never unlinked by the old holder", async () => { + const root = makeWorkspace(); + try { + const first = await acquireImportLock(root, "pm-token1"); + assert.strictEqual(first.status, "acquired"); + if (first.status !== "acquired") return; + // Simulate a stale-break + successor acquisition: replace the file content + // with a different owner/token while `first` still believes it holds it. + const successor = { + id: "pm-github.comment-sync.pm-token1", + pid: process.pid, + owner: "pm-github", + token: "successor-token", + created_at: new Date().toISOString(), + ttl_seconds: 300, + }; + writeFileSync(first.lock.path, `${JSON.stringify(successor)}\n`); + first.lock.release(); + assert.ok(existsSync(first.lock.path), "release must not unlink a successor's lock"); + const after = JSON.parse(readFileSync(first.lock.path, "utf-8")); + assert.strictEqual(after.token, "successor-token", "successor payload is untouched"); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("a fresh lock whose owner PID is dead is broken with a stderr warning", async () => { + const root = makeWorkspace(); + try { + // A process that has already exited gives us a (practically) dead PID. + const dead = spawnSync(process.execPath, ["-e", ""]); + assert.ok(dead.pid && dead.pid > 0, "need a real child pid for this test"); + plantLock(root, "pm-stale2", { + id: "pm-github.comment-sync.pm-stale2", + pid: dead.pid, + owner: "pm-github", + created_at: new Date().toISOString(), // fresh — only the dead PID makes it stale + ttl_seconds: 300, + }); + const messages = await captureStderr(async () => { + const acq = await acquireImportLock(root, "pm-stale2", { waitMs: 500 }); + assert.strictEqual(acq.status, "acquired", "dead-owner lock is broken without waiting out the TTL"); + if (acq.status === "acquired") acq.lock.release(); + }); + assert.ok( + messages.some((m) => m.includes("breaking stale comment-sync lock") && m.includes("dead")), + `expected a dead-owner warning, got: ${messages.join(" | ")}`, + ); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("a stale lock is NOT broken while another breaker's fresh election sidecar exists", async () => { + const root = makeWorkspace(); + try { + // Dead-owner lock (breakable) + a FRESH `.break` sidecar: some other + // process is mid-election, so we must lose the election and wait instead + // of double-breaking. + const dead = spawnSync(process.execPath, ["-e", ""]); + assert.ok(dead.pid && dead.pid > 0); + const lockPath = plantLock(root, "pm-break1", { + id: "pm-github.comment-sync.pm-break1", + pid: dead.pid, + owner: "pm-github", + token: "stale-owner", + created_at: new Date().toISOString(), + ttl_seconds: 300, + }); + writeFileSync(`${lockPath}.break`, ""); + const acq = await acquireImportLock(root, "pm-break1", { waitMs: 300 }); + assert.strictEqual(acq.status, "contended", "an election in progress blocks a second breaker"); + assert.ok(existsSync(lockPath), "the stale lock is left for the election winner"); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("a crashed breaker's aged-out sidecar is cleared and the stale lock is then broken", async () => { + const root = makeWorkspace(); + try { + const dead = spawnSync(process.execPath, ["-e", ""]); + assert.ok(dead.pid && dead.pid > 0); + const lockPath = plantLock(root, "pm-break2", { + id: "pm-github.comment-sync.pm-break2", + pid: dead.pid, + owner: "pm-github", + token: "stale-owner", + created_at: new Date().toISOString(), + ttl_seconds: 300, + }); + // A sidecar left behind by a breaker that crashed mid-election, well past + // the breaker TTL. + const sidecar = `${lockPath}.break`; + writeFileSync(sidecar, ""); + const old = new Date(Date.now() - 30_000); + utimesSync(sidecar, old, old); + const messages = await captureStderr(async () => { + const acq = await acquireImportLock(root, "pm-break2", { waitMs: 2000 }); + assert.strictEqual(acq.status, "acquired", "aged-out sidecar is cleared, then the stale lock is broken"); + if (acq.status === "acquired") acq.lock.release(); + }); + assert.ok( + messages.some((m) => m.includes("breaking stale comment-sync lock")), + `expected a stale-break warning, got: ${messages.join(" | ")}`, + ); + assert.ok(!existsSync(sidecar), "the election sidecar is cleaned up"); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("an unparseable lock payload falls back to file mtime: fresh contends, old breaks", async () => { + const root = makeWorkspace(); + try { + // Fresh garbage payload — could be a live writer mid-write, so it must NOT + // be broken; the waiter contends instead. + const lockPath = plantLock(root, "pm-stale3", "not json at all"); + const fresh = await acquireImportLock(root, "pm-stale3", { waitMs: 200 }); + assert.strictEqual(fresh.status, "contended", "young unparseable lock is treated as held, not broken"); + + // Backdate the mtime beyond the TTL — now it is stale and gets broken. + const old = new Date(Date.now() - 10 * 60_000); + utimesSync(lockPath, old, old); + const messages = await captureStderr(async () => { + const acq = await acquireImportLock(root, "pm-stale3", { waitMs: 1000 }); + assert.strictEqual(acq.status, "acquired"); + if (acq.status === "acquired") acq.lock.release(); + }); + assert.ok(messages.some((m) => m.includes("breaking stale comment-sync lock")), "stale garbage lock is broken"); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +// Concurrency: in-process and cross-process ------------------------------------ + +test("two in-process concurrent syncs of the same comments never duplicate", async () => { + const root = makeWorkspace(); + try { + const id = firstItemId(root); + const comments = Array.from({ length: 8 }, (_, i) => + ghComment({ id: 7000 + i, user: { login: "racer" }, body: `in-process race ${i}` }), + ); + const [r1, r2] = await Promise.all([ + syncGithubCommentsToAnnotations(id, comments, root, 1), + syncGithubCommentsToAnnotations(id, comments, root, 1), + ]); + assert.strictEqual(r1.added + r2.added, comments.length, "each comment added exactly once across both syncs"); + const { comments: stored } = await import("@unbrained/pm-cli/sdk").then((m) => + m.comments(id, {}, { pmRoot: root }), + ); + assert.strictEqual(stored.length, comments.length, "no duplicate comments stored"); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +// The pm-github-503u acceptance test: two concurrent OS processes (the real +// `pm github import` race shape) sync identical fake GitHub comments into the +// same item of the same workspace. The lockfile must serialize them so every +// GitHub comment id lands exactly once. +test( + "two concurrent processes importing the same comments never duplicate (pm-github-503u)", + { timeout: 180_000 }, + async () => { + const root = makeWorkspace(); + try { + const id = firstItemId(root); + const comments = Array.from({ length: 20 }, (_, i) => + ghComment({ id: 9000 + i, user: { login: "racer" }, body: `cross-process race ${i}` }), + ); + // Start barrier: both children are spawned first and only begin syncing + // once the barrier file appears, so the test deterministically exercises + // concurrent entry instead of an accidental sequential pass. + const barrierFile = join(root, "start-barrier"); + const env = { + ...process.env, + FAKE_COMMENTS: JSON.stringify(comments), + BARRIER_FILE: barrierFile, + }; + const runChild = () => + new Promise<{ added: number; skipped: number }>((resolve, reject) => { + execFile( + process.execPath, + [CHILD_SCRIPT, id, root], + { env }, + (err, stdout, stderr) => { + if (err) return reject(new Error(`child failed: ${err.message}\n${stderr}`)); + assert.ok( + !stderr.includes("another import holds the comment-sync lock"), + `neither child may be starved off the lock: ${stderr}`, + ); + try { + resolve(JSON.parse(stdout.trim())); + } catch (parseErr) { + reject(new Error(`unparseable child output: ${stdout} (${parseErr})`)); + } + }, + ); + }); + + const childrenDone = Promise.all([runChild(), runChild()]); + writeFileSync(barrierFile, "go\n"); + const [r1, r2] = await childrenDone; + assert.strictEqual( + r1.added + r2.added, + comments.length, + `each comment added exactly once across both processes (got ${r1.added}+${r2.added})`, + ); + assert.strictEqual(r1.skipped + r2.skipped, comments.length, "the loser of the race skips every duplicate"); + + const { comments: stored } = await import("@unbrained/pm-cli/sdk").then((m) => + m.comments(id, {}, { pmRoot: root }), + ); + assert.strictEqual(stored.length, comments.length, "exactly one stored comment per GitHub comment"); + const markerIds = stored.map((c: { text?: string }) => { + const m = //.exec(c.text ?? ""); + assert.ok(m, `stored comment carries a marker: ${c.text}`); + return Number(m![1]); + }); + assert.strictEqual(new Set(markerIds).size, comments.length, "no marker id appears twice"); + assert.deepEqual( + [...markerIds].sort((a, b) => a - b), + comments.map((c) => c.id).sort((a, b) => a - b), + ); + + // The loser's re-run must be a clean no-op (idempotency preserved). + const r3 = await new Promise<{ added: number; skipped: number }>((resolve, reject) => { + execFile(process.execPath, [CHILD_SCRIPT, id, root], { env }, (err, stdout) => + err ? reject(err) : resolve(JSON.parse(stdout.trim())), + ); + }); + assert.deepEqual(r3, { added: 0, skipped: comments.length }); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }, +);