fix(matrix): serialize credential backfill writes

extensions/matrix/src/matrix/credentials.test.ts

@@ -1,4 +1,5 @@
 import fs from "node:fs";
+import fsPromises from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 import { afterEach, describe, expect, it, vi } from "vitest";
@@ -148,6 +149,72 @@ describe("matrix credentials storage", () => {
     });
   });
 
+  it("serializes stale backfill writes behind newer credential saves", async () => {
+    setupStateDir();
+    await saveMatrixCredentials(
+      {
+        homeserver: "https://matrix.example.org",
+        userId: "@bot:example.org",
+        accessToken: "tok-old",
+      },
+      {},
+      "default",
+    );
+
+    let releaseFirstWrite: (() => void) | undefined;
+    let firstWriteStarted = false;
+    const originalRename = fsPromises.rename.bind(fsPromises);
+    const renameSpy = vi
+      .spyOn(fsPromises, "rename")
+      .mockImplementation(async (...args: Parameters<typeof fsPromises.rename>) => {
+        if (!firstWriteStarted) {
+          firstWriteStarted = true;
+          await new Promise<void>((resolve) => {
+            releaseFirstWrite = resolve;
+          });
+        }
+        await originalRename(...args);
+      });
+
+    try {
+      const staleBackfillPromise = saveBackfilledMatrixDeviceId(
+        {
+          homeserver: "https://matrix.example.org",
+          userId: "@bot:example.org",
+          accessToken: "tok-old",
+          deviceId: "DEVICE123",
+        },
+        {},
+        "default",
+      );
+
+      await vi.waitFor(() => {
+        expect(firstWriteStarted).toBe(true);
+      });
+
+      const newerSavePromise = saveMatrixCredentials(
+        {
+          homeserver: "https://matrix.example.org",
+          userId: "@bot:example.org",
+          accessToken: "tok-new",
+          deviceId: "DEVICE999",
+        },
+        {},
+        "default",
+      );
+
+      releaseFirstWrite?.();
+      await Promise.all([staleBackfillPromise, newerSavePromise]);
+
+      expect(loadMatrixCredentials({}, "default")).toMatchObject({
+        accessToken: "tok-new",
+        deviceId: "DEVICE999",
+      });
+    } finally {
+      renameSpy.mockRestore();
+    }
+  });
+
   it("migrates legacy matrix credential files on read", async () => {
     const stateDir = setupStateDir({
       channels: {

extensions/matrix/src/matrix/credentials.ts

@@ -1,4 +1,5 @@
 import { writeJsonFileAtomically } from "../runtime-api.js";
+import { createAsyncLock, type AsyncLock } from "./async-lock.js";
 import { loadMatrixCredentials, resolveMatrixCredentialsPath } from "./credentials-read.js";
 import type { MatrixStoredCredentials } from "./credentials-read.js";
 
@@ -11,23 +12,44 @@ export {
 } from "./credentials-read.js";
 export type { MatrixStoredCredentials } from "./credentials-read.js";
 
+const credentialWriteLocks = new Map<string, AsyncLock>();
+
+function withCredentialWriteLock<T>(credPath: string, fn: () => Promise<T>): Promise<T> {
+  let withLock = credentialWriteLocks.get(credPath);
+  if (!withLock) {
+    withLock = createAsyncLock();
+    credentialWriteLocks.set(credPath, withLock);
+  }
+  return withLock(fn);
+}
+
+async function writeMatrixCredentialsUnlocked(params: {
+  credPath: string;
+  credentials: Omit<MatrixStoredCredentials, "createdAt" | "lastUsedAt">;
+  existing: MatrixStoredCredentials | null;
+}): Promise<void> {
+  const now = new Date().toISOString();
+  const toSave: MatrixStoredCredentials = {
+    ...params.credentials,
+    createdAt: params.existing?.createdAt ?? now,
+    lastUsedAt: now,
+  };
+  await writeJsonFileAtomically(params.credPath, toSave);
+}
+
 export async function saveMatrixCredentials(
   credentials: Omit<MatrixStoredCredentials, "createdAt" | "lastUsedAt">,
   env: NodeJS.ProcessEnv = process.env,
   accountId?: string | null,
 ): Promise<void> {
   const credPath = resolveMatrixCredentialsPath(env, accountId);
-
-  const existing = loadMatrixCredentials(env, accountId);
-  const now = new Date().toISOString();
-
-  const toSave: MatrixStoredCredentials = {
-    ...credentials,
-    createdAt: existing?.createdAt ?? now,
-    lastUsedAt: now,
-  };
-
-  await writeJsonFileAtomically(credPath, toSave);
+  await withCredentialWriteLock(credPath, async () => {
+    await writeMatrixCredentialsUnlocked({
+      credPath,
+      credentials,
+      existing: loadMatrixCredentials(env, accountId),
+    });
+  });
 }
 
 export async function saveBackfilledMatrixDeviceId(
@@ -35,30 +57,39 @@ export async function saveBackfilledMatrixDeviceId(
   env: NodeJS.ProcessEnv = process.env,
   accountId?: string | null,
 ): Promise<"saved" | "skipped"> {
-  const existing = loadMatrixCredentials(env, accountId);
-  if (
-    existing &&
-    (existing.homeserver !== credentials.homeserver ||
-      existing.userId !== credentials.userId ||
-      existing.accessToken !== credentials.accessToken)
-  ) {
-    return "skipped";
-  }
+  const credPath = resolveMatrixCredentialsPath(env, accountId);
+  return await withCredentialWriteLock(credPath, async () => {
+    const existing = loadMatrixCredentials(env, accountId);
+    if (
+      existing &&
+      (existing.homeserver !== credentials.homeserver ||
+        existing.userId !== credentials.userId ||
+        existing.accessToken !== credentials.accessToken)
+    ) {
+      return "skipped";
+    }
 
-  await saveMatrixCredentials(credentials, env, accountId);
-  return "saved";
+    await writeMatrixCredentialsUnlocked({
+      credPath,
+      credentials,
+      existing,
+    });
+    return "saved";
+  });
 }
 
 export async function touchMatrixCredentials(
   env: NodeJS.ProcessEnv = process.env,
   accountId?: string | null,
 ): Promise<void> {
-  const existing = loadMatrixCredentials(env, accountId);
-  if (!existing) {
-    return;
-  }
-
-  existing.lastUsedAt = new Date().toISOString();
   const credPath = resolveMatrixCredentialsPath(env, accountId);
-  await writeJsonFileAtomically(credPath, existing);
+  await withCredentialWriteLock(credPath, async () => {
+    const existing = loadMatrixCredentials(env, accountId);
+    if (!existing) {
+      return;
+    }
+
+    existing.lastUsedAt = new Date().toISOString();
+    await writeJsonFileAtomically(credPath, existing);
+  });
 }