fix(queue): guard ISSUE_CLOSURE solvedByPr write with state = CLOSED

packages/das/src/queue/fetch.processor.spec.ts

@@ -66,3 +66,118 @@ describe("FetchProcessor rate-limit deferral", () => {
     expect(job.moveToDelayed).not.toHaveBeenCalled();
   });
 });
+
+// Regression coverage for the ISSUE_CLOSURE reopen race (issue #199): the
+// async closure job must not write solvedByPr back onto an issue that was
+// reopened while its (slow) GraphQL fetch was in flight.
+interface IssueRow {
+  repoFullName: string;
+  issueNumber: number;
+  state: string;
+  solvedByPr: number | null;
+}
+
+// Minimal in-memory stand-in for the TypeORM Issue repository.
+// - findOneBy returns a *copy* of the row, modelling the stale in-memory read
+//   the handler holds across the async fetch.
+// - update honours the criteria: the partial is only applied when every key in
+//   the criteria object matches the current row (mirroring SQL `WHERE`), so a
+//   `state: "CLOSED"` predicate makes the write a no-op once the row is OPEN.
+function makeIssueRepo(row: IssueRow): {
+  findOneBy: jest.Mock;
+  update: jest.Mock;
+} {
+  return {
+    findOneBy: jest.fn().mockImplementation(() => Promise.resolve({ ...row })),
+    update: jest
+      .fn()
+      .mockImplementation(
+        (criteria: Partial<IssueRow>, patch: Partial<IssueRow>) => {
+          const matches = Object.entries(criteria).every(
+            ([k, v]) => row[k as keyof IssueRow] === v,
+          );
+          if (matches) Object.assign(row, patch);
+          return Promise.resolve({ affected: matches ? 1 : 0 });
+        },
+      ),
+  };
+}
+
+function closureJob(): Job {
+  return {
+    name: FETCH_JOBS.ISSUE_CLOSURE,
+    id: "closure-acme/repo-7",
+    data: { repoFullName: "acme/repo", issueNumber: 7 },
+  } as unknown as Job;
+}
+
+describe("FetchProcessor ISSUE_CLOSURE reopen race", () => {
+  beforeEach(() => {
+    jest.spyOn(Logger.prototype, "log").mockImplementation(() => undefined);
+    jest.spyOn(Logger.prototype, "warn").mockImplementation(() => undefined);
+  });
+
+  afterEach(() => jest.restoreAllMocks());
+
+  it("does not re-attach solvedByPr when the issue was reopened during the fetch", async () => {
+    const row: IssueRow = {
+      repoFullName: "acme/repo",
+      issueNumber: 7,
+      state: "CLOSED",
+      solvedByPr: null,
+    };
+    const issueRepo = makeIssueRepo(row);
+    // The GraphQL fetch is slow: model a `issues.reopened` webhook landing
+    // mid-flight (state -> OPEN, solver cleared) before it resolves a PR.
+    const fetcher = {
+      fetchIssueClosingPr: jest.fn().mockImplementation(() => {
+        row.state = "OPEN";
+        row.solvedByPr = null;
+        return Promise.resolve(123);
+      }),
+    } as any;
+
+    const processor = new FetchProcessor(
+      fetcher,
+      {} as any,
+      issueRepo as any,
+      {} as any,
+    );
+
+    await processor.process(closureJob());
+
+    // The reopen must survive: OPEN, and no solver attribution.
+    expect(row.state).toBe("OPEN");
+    expect(row.solvedByPr).toBeNull();
+    // The final write is guarded by state = 'CLOSED'.
+    expect(issueRepo.update).toHaveBeenLastCalledWith(
+      expect.objectContaining({ state: "CLOSED" }),
+      { solvedByPr: 123 },
+    );
+  });
+
+  it("writes solvedByPr when the issue stays closed", async () => {
+    const row: IssueRow = {
+      repoFullName: "acme/repo",
+      issueNumber: 7,
+      state: "CLOSED",
+      solvedByPr: null,
+    };
+    const issueRepo = makeIssueRepo(row);
+    const fetcher = {
+      fetchIssueClosingPr: jest.fn().mockResolvedValue(55),
+    } as any;
+
+    const processor = new FetchProcessor(
+      fetcher,
+      {} as any,
+      issueRepo as any,
+      {} as any,
+    );
+
+    await processor.process(closureJob());
+
+    expect(row.state).toBe("CLOSED");
+    expect(row.solvedByPr).toBe(55);
+  });
+});

packages/das/src/queue/fetch.processor.ts

@@ -136,7 +136,15 @@ export class FetchProcessor extends WorkerHost {
       issueNumber,
     );
 
-    await this.issueRepo.update({ repoFullName, issueNumber }, { solvedByPr });
+    // Guard the write with state = 'CLOSED'. The GraphQL fetch above is slow,
+    // and a `issues.reopened` webhook can land while it is in flight — that
+    // handler upserts state = OPEN and clears solvedByPr. Without this guard we
+    // would clobber the reopen and leave an OPEN issue with a solver link. The
+    // state predicate makes the update a no-op if the issue is no longer closed.
+    await this.issueRepo.update(
+      { repoFullName, issueNumber, state: "CLOSED" },
+      { solvedByPr },
+    );
   }
 
   private async handlePrMetadata(