diff --git a/.changeset/serialize-secrets-state-writes.md b/.changeset/serialize-secrets-state-writes.md new file mode 100644 index 000000000..892e72423 --- /dev/null +++ b/.changeset/serialize-secrets-state-writes.md @@ -0,0 +1,5 @@ +--- +"@tailor-platform/sdk": patch +--- + +Fix concurrent deploys to the same workspace and application corrupting the local secrets hash state. Secret and auth-connection updates now hold a target-scoped lock while writing remote values and saving their hashes, so a later deploy no longer skips a secret update based on a hash that no longer matches the deployed value. diff --git a/packages/sdk/src/cli/commands/deploy/auth-connection.test.ts b/packages/sdk/src/cli/commands/deploy/auth-connection.test.ts index 1afb92ece..655bb6bd6 100644 --- a/packages/sdk/src/cli/commands/deploy/auth-connection.test.ts +++ b/packages/sdk/src/cli/commands/deploy/auth-connection.test.ts @@ -27,6 +27,7 @@ vi.mock("./secrets-state", async (importOriginal) => { ...actual, loadSecretsState: (...args: unknown[]) => mockLoadSecretsState(...args), saveSecretsState: (...args: unknown[]) => mockSaveSecretsState(...args), + withSecretsStateLock: (_scope: unknown, fn: () => Promise) => fn(), // Deterministic hash so tests can pin the "unchanged" secret state. hashValue: () => "fixed-hash", }; diff --git a/packages/sdk/src/cli/commands/deploy/auth-connection.ts b/packages/sdk/src/cli/commands/deploy/auth-connection.ts index 4ce6bf2ee..aa6e0b7af 100644 --- a/packages/sdk/src/cli/commands/deploy/auth-connection.ts +++ b/packages/sdk/src/cli/commands/deploy/auth-connection.ts @@ -9,7 +9,12 @@ import { logger } from "#/cli/shared/logger"; import { createChangeSet } from "./change-set"; import { buildMetaRequest, resourceTrn, sdkNameLabelKey, type WithLabel } from "./label"; import { trackDesiredResourceOwnership, trackRemainingResourceOwner } from "./owned-resource"; -import { hashValue, loadSecretsState, saveSecretsState } from "./secrets-state"; +import { + hashValue, + loadSecretsState, + saveSecretsState, + withSecretsStateLock, +} from "./secrets-state"; import type { AuthConnectionConfig } from "#/types/auth-connection.generated"; import type { OwnerConflict, UnmanagedResource } from "./confirm"; import type { ApplyPhase } from "./phase"; @@ -255,28 +260,55 @@ export async function applyAuthConnections( const { changeSet, stateScope } = result; if (phase === "create-update") { - await Promise.all( - changeSet.creates.map(async (create) => { - await client.createAuthConnection(create.request); - await client.setMetadata(create.metaRequest); - logger.info( - `Connection "${create.name}" was created. Authorize it with:\n` + - ` tailor-sdk authconnection authorize --name ${create.name}\n` + - `Or via the Console: tailor-sdk authconnection open`, + if (changeSet.creates.length > 0 || changeSet.replaces.length > 0) { + await withSecretsStateLock(stateScope, async () => { + await Promise.all( + changeSet.creates.map(async (create) => { + await client.createAuthConnection(create.request); + await client.setMetadata(create.metaRequest); + logger.info( + `Connection "${create.name}" was created. Authorize it with:\n` + + ` tailor-sdk authconnection authorize --name ${create.name}\n` + + `Or via the Console: tailor-sdk authconnection open`, + ); + }), ); - }), - ); - for (const replace of changeSet.replaces) { - const resp = await client.updateAuthConnection(replace.updateRequest); - if (resp.connection?.status === AuthConnection_Status.UNAUTHORIZED) { - logger.warn( - `Connection "${replace.name}" requires re-authorization. Authorize with:\n` + - ` tailor-sdk authconnection authorize --name ${replace.name}\n` + - `Or via the Console: tailor-sdk authconnection open`, + for (const replace of changeSet.replaces) { + const resp = await client.updateAuthConnection(replace.updateRequest); + if (resp.connection?.status === AuthConnection_Status.UNAUTHORIZED) { + logger.warn( + `Connection "${replace.name}" requires re-authorization. Authorize with:\n` + + ` tailor-sdk authconnection authorize --name ${replace.name}\n` + + `Or via the Console: tailor-sdk authconnection open`, + ); + } + await client.setMetadata(replace.metaRequest); + } + + const secretReplaces = changeSet.replaces.filter((replace) => + replace.updateRequest.updateMask?.paths?.includes("oauth2.client_secret"), ); - } - await client.setMetadata(replace.metaRequest); + if (changeSet.creates.length > 0 || secretReplaces.length > 0) { + const state = loadSecretsState(stateScope); + if (!state.connections) { + state.connections = {}; + } + for (const create of changeSet.creates) { + const conn = create.request.connection; + if (conn?.config?.case === "oauth2") { + state.connections[create.name] = hashValue(conn.config.value.clientSecret ?? ""); + } + } + for (const replace of secretReplaces) { + const conn = replace.updateRequest.connection; + if (conn?.config?.case === "oauth2") { + state.connections[replace.name] = hashValue(conn.config.value.clientSecret ?? ""); + } + } + saveSecretsState(stateScope, state); + } + }); } // Metadata-only updates: backfill the SDK ownership label on connections @@ -286,37 +318,14 @@ export async function applyAuthConnections( await client.setMetadata(update.metaRequest); }), ); + } else if (changeSet.deletes.length > 0) { + await withSecretsStateLock(stateScope, async () => { + await Promise.all( + changeSet.deletes.map(async (del) => { + await client.deleteAuthConnection(del.request); + }), + ); - const secretReplaces = changeSet.replaces.filter((replace) => - replace.updateRequest.updateMask?.paths?.includes("oauth2.client_secret"), - ); - if (changeSet.creates.length > 0 || secretReplaces.length > 0) { - const state = loadSecretsState(stateScope); - if (!state.connections) { - state.connections = {}; - } - for (const create of changeSet.creates) { - const conn = create.request.connection; - if (conn?.config?.case === "oauth2") { - state.connections[create.name] = hashValue(conn.config.value.clientSecret ?? ""); - } - } - for (const replace of secretReplaces) { - const conn = replace.updateRequest.connection; - if (conn?.config?.case === "oauth2") { - state.connections[replace.name] = hashValue(conn.config.value.clientSecret ?? ""); - } - } - saveSecretsState(stateScope, state); - } - } else { - await Promise.all( - changeSet.deletes.map(async (del) => { - await client.deleteAuthConnection(del.request); - }), - ); - - if (changeSet.deletes.length > 0) { const state = loadSecretsState(stateScope); if (state.connections) { for (const del of changeSet.deletes) { @@ -324,6 +333,6 @@ export async function applyAuthConnections( } saveSecretsState(stateScope, state); } - } + }); } } diff --git a/packages/sdk/src/cli/commands/deploy/secret-manager.test.ts b/packages/sdk/src/cli/commands/deploy/secret-manager.test.ts index 26a290e72..498dd1642 100644 --- a/packages/sdk/src/cli/commands/deploy/secret-manager.test.ts +++ b/packages/sdk/src/cli/commands/deploy/secret-manager.test.ts @@ -21,6 +21,7 @@ vi.mock("./secrets-state", async (importOriginal) => { ...actual, loadSecretsState: (...args: unknown[]) => mockLoadSecretsState(...args), saveSecretsState: (...args: unknown[]) => mockSaveSecretsState(...args), + withSecretsStateLock: (_scope: unknown, fn: () => Promise) => fn(), }; }); diff --git a/packages/sdk/src/cli/commands/deploy/secret-manager.ts b/packages/sdk/src/cli/commands/deploy/secret-manager.ts index 63654047e..9ffe9603e 100644 --- a/packages/sdk/src/cli/commands/deploy/secret-manager.ts +++ b/packages/sdk/src/cli/commands/deploy/secret-manager.ts @@ -7,7 +7,12 @@ import { trackDesiredResourceOwnership, trackRemainingResourceOwner, } from "./owned-resource"; -import { hashValue, loadSecretsState, saveSecretsState } from "./secrets-state"; +import { + hashValue, + loadSecretsState, + saveSecretsState, + withSecretsStateLock, +} from "./secrets-state"; import type { ApplyPhase, PlanContext } from "#/cli/commands/deploy/types"; import type { Application } from "#/cli/services/application"; import type { OwnerConflict, UnmanagedResource } from "./confirm"; @@ -337,57 +342,61 @@ export async function applySecretManager( ); } - // Create new secrets - await Promise.all( - secretChangeSet.creates.map((create) => - client.createSecretManagerSecret(secretCreateRequest(create)), - ), - ); + const secretHashUpdates = [...secretChangeSet.creates, ...secretChangeSet.updates]; + if (secretHashUpdates.length > 0) { + await withSecretsStateLock(stateScope, async () => { + // Create new secrets + await Promise.all( + secretChangeSet.creates.map((create) => + client.createSecretManagerSecret(secretCreateRequest(create)), + ), + ); - // Update existing secrets - await Promise.all( - secretChangeSet.updates.map((update) => - client.updateSecretManagerSecret(secretUpdateRequest(update)), - ), - ); + // Update existing secrets + await Promise.all( + secretChangeSet.updates.map((update) => + client.updateSecretManagerSecret(secretUpdateRequest(update)), + ), + ); - const secretHashUpdates = [...secretChangeSet.creates, ...secretChangeSet.updates]; - if (application && secretHashUpdates.length > 0) { - const state = loadSecretsState(stateScope); - for (const secret of secretHashUpdates) { - if (!Object.hasOwn(state.vaults, secret.vaultName)) { - state.vaults[secret.vaultName] = {}; + if (application) { + const state = loadSecretsState(stateScope); + for (const secret of secretHashUpdates) { + if (!Object.hasOwn(state.vaults, secret.vaultName)) { + state.vaults[secret.vaultName] = {}; + } + assertDefined(state.vaults[secret.vaultName], "vault state entry missing")[ + secret.secretName + ] = hashValue(secret.value); + } + saveSecretsState(stateScope, state); } - assertDefined(state.vaults[secret.vaultName], "vault state entry missing")[ - secret.secretName - ] = hashValue(secret.value); - } - saveSecretsState(stateScope, state); + }); } - } else { - // Delete orphan secrets - await Promise.all( - secretChangeSet.deletes.map((del) => - client.deleteSecretManagerSecret({ - workspaceId: del.workspaceId, - secretmanagerVaultName: del.vaultName, - secretmanagerSecretName: del.secretName, - }), - ), - ); + } else if (secretChangeSet.deletes.length > 0 || vaultChangeSet.deletes.length > 0) { + await withSecretsStateLock(stateScope, async () => { + // Delete orphan secrets + await Promise.all( + secretChangeSet.deletes.map((del) => + client.deleteSecretManagerSecret({ + workspaceId: del.workspaceId, + secretmanagerVaultName: del.vaultName, + secretmanagerSecretName: del.secretName, + }), + ), + ); - // Delete orphan vaults - await Promise.all( - vaultChangeSet.deletes.map((del) => - client.deleteSecretManagerVault({ - workspaceId: del.workspaceId, - secretmanagerVaultName: del.name, - }), - ), - ); + // Delete orphan vaults + await Promise.all( + vaultChangeSet.deletes.map((del) => + client.deleteSecretManagerVault({ + workspaceId: del.workspaceId, + secretmanagerVaultName: del.name, + }), + ), + ); - // Remove deleted secrets and vaults from hash state - if (secretChangeSet.deletes.length > 0 || vaultChangeSet.deletes.length > 0) { + // Remove deleted secrets and vaults from hash state const state = loadSecretsState(stateScope); for (const del of secretChangeSet.deletes) { if (Object.hasOwn(state.vaults, del.vaultName)) { @@ -406,6 +415,6 @@ export async function applySecretManager( delete state.vaults[del.name]; } saveSecretsState(stateScope, state); - } + }); } } diff --git a/packages/sdk/src/cli/commands/deploy/secrets-state-concurrency.test.ts b/packages/sdk/src/cli/commands/deploy/secrets-state-concurrency.test.ts new file mode 100644 index 000000000..ddb8156d9 --- /dev/null +++ b/packages/sdk/src/cli/commands/deploy/secrets-state-concurrency.test.ts @@ -0,0 +1,180 @@ +import { existsSync, rmSync } from "node:fs"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { applyAuthConnections, type planAuthConnections } from "./auth-connection"; +import { applySecretManager, type planSecretManager } from "./secret-manager"; +import { hashValue, loadSecretsState } from "./secrets-state"; +import type { Application } from "#/cli/services/application"; +import type { OperatorClient } from "#/cli/shared/client"; + +const distDir = "/tmp/tailor-sdk-test-secrets-state-concurrency"; + +vi.mock("#/cli/shared/dist-dir", () => ({ + getDistDir: () => "/tmp/tailor-sdk-test-secrets-state-concurrency", +})); + +const stateScope = { + workspaceId: "ws-1", + applicationId: "app-1", + applicationName: "test-app", +}; + +const application = { + name: "test-app", + id: "app-1", + secrets: [], +} as unknown as Application; + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((r) => (resolve = r)); + return { promise, resolve }; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function removeStateDir(): void { + if (existsSync(distDir)) { + rmSync(distDir, { recursive: true }); + } +} + +function secretUpdate(secretName: string, value: string) { + return { + name: `my-vault/${secretName}`, + secretName, + workspaceId: "ws-1", + vaultName: "my-vault", + value, + }; +} + +function secretPlanResult(updates: Array>) { + return { + stateScope, + vaultChangeSet: { creates: [], updates: [], deletes: [], replaces: [] }, + secretChangeSet: { creates: [], updates, deletes: [], replaces: [] }, + } as unknown as Awaited>; +} + +function connectionReplace(name: string, clientSecret: string) { + return { + name, + updateRequest: { + workspaceId: "ws-1", + connection: { + name, + config: { case: "oauth2" as const, value: { clientSecret } }, + }, + updateMask: { paths: ["oauth2.client_secret"] }, + }, + metaRequest: { trn: `trn:${name}`, labels: {} }, + }; +} + +function connectionPlanResult(replaces: Array>) { + return { + stateScope, + changeSet: { creates: [], updates: [], deletes: [], replaces }, + } as unknown as Awaited>; +} + +describe("concurrent deploys to the same target", () => { + beforeEach(removeStateDir); + afterEach(removeStateDir); + + test("secret hash state matches the last remote write", async () => { + const remote = new Map(); + const bSettled = deferred(); + + // Deploy A: the trailing secret's remote update settles only after deploy B + // has fully finished (or after a grace period when B is made to wait). + const clientA = { + updateSecretManagerSecret: vi.fn().mockImplementation(async (req) => { + remote.set(req.secretmanagerSecretName, req.secretmanagerSecretValue); + if (req.secretmanagerSecretName === "trailing-secret") { + await Promise.race([bSettled.promise, sleep(50)]); + } + return {}; + }), + } as unknown as OperatorClient; + + const clientB = { + updateSecretManagerSecret: vi.fn().mockImplementation(async (req) => { + remote.set(req.secretmanagerSecretName, req.secretmanagerSecretValue); + return {}; + }), + } as unknown as OperatorClient; + + const applyA = applySecretManager( + clientA, + secretPlanResult([ + secretUpdate("shared-secret", "value-from-a"), + secretUpdate("trailing-secret", "trailing-value"), + ]), + "create-update", + application, + ); + const applyB = applySecretManager( + clientB, + secretPlanResult([secretUpdate("shared-secret", "value-from-b")]), + "create-update", + application, + ); + void applyB.catch(() => {}).finally(() => bSettled.resolve()); + + await Promise.all([applyA, applyB]); + + const state = loadSecretsState(stateScope); + expect(state.vaults["my-vault"]?.["shared-secret"]).toBe( + hashValue(remote.get("shared-secret")!), + ); + expect(state.vaults["my-vault"]?.["trailing-secret"]).toBe(hashValue("trailing-value")); + }); + + test("auth connection hash state matches the last remote write", async () => { + const remote = new Map(); + const bSettled = deferred(); + + const clientA = { + updateAuthConnection: vi.fn().mockImplementation(async (req) => { + remote.set(req.connection.name, req.connection.config.value.clientSecret); + if (req.connection.name === "trailing-conn") { + await Promise.race([bSettled.promise, sleep(50)]); + } + return {}; + }), + setMetadata: vi.fn().mockResolvedValue({}), + } as unknown as OperatorClient; + + const clientB = { + updateAuthConnection: vi.fn().mockImplementation(async (req) => { + remote.set(req.connection.name, req.connection.config.value.clientSecret); + return {}; + }), + setMetadata: vi.fn().mockResolvedValue({}), + } as unknown as OperatorClient; + + const applyA = applyAuthConnections( + clientA, + connectionPlanResult([ + connectionReplace("shared-conn", "secret-from-a"), + connectionReplace("trailing-conn", "trailing-secret"), + ]), + "create-update", + ); + const applyB = applyAuthConnections( + clientB, + connectionPlanResult([connectionReplace("shared-conn", "secret-from-b")]), + "create-update", + ); + void applyB.catch(() => {}).finally(() => bSettled.resolve()); + + await Promise.all([applyA, applyB]); + + const state = loadSecretsState(stateScope); + expect(state.connections?.["shared-conn"]).toBe(hashValue(remote.get("shared-conn")!)); + expect(state.connections?.["trailing-conn"]).toBe(hashValue("trailing-secret")); + }); +}); diff --git a/packages/sdk/src/cli/commands/deploy/secrets-state.test.ts b/packages/sdk/src/cli/commands/deploy/secrets-state.test.ts index 8584106fd..b1bce5626 100644 --- a/packages/sdk/src/cli/commands/deploy/secrets-state.test.ts +++ b/packages/sdk/src/cli/commands/deploy/secrets-state.test.ts @@ -1,4 +1,4 @@ -import { existsSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, rmSync, utimesSync, writeFileSync } from "node:fs"; import * as path from "pathe"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { @@ -6,6 +6,7 @@ import { hashValue, loadSecretsState, saveSecretsState, + withSecretsStateLock, } from "./secrets-state"; import type { SecretsStateScope } from "./secrets-state"; @@ -194,3 +195,102 @@ describe("secrets-state", () => { expect(hash1).not.toBe(hash2); }); }); + +describe("withSecretsStateLock", () => { + beforeEach(removeStateFiles); + afterEach(removeStateFiles); + + function lockPathFor(scope: SecretsStateScope): string { + return `${getSecretsStatePath(scope)}.lock`; + } + + test("returns the critical section result and releases the lock", async () => { + const result = await withSecretsStateLock(scopeA, async () => 42); + expect(result).toBe(42); + expect(existsSync(lockPathFor(scopeA))).toBe(false); + }); + + test("serializes concurrent critical sections for the same scope", async () => { + const events: string[] = []; + let releaseFirst!: () => void; + const firstGate = new Promise((resolve) => (releaseFirst = resolve)); + + const first = withSecretsStateLock(scopeA, async () => { + events.push("first-start"); + await firstGate; + events.push("first-end"); + }); + const second = withSecretsStateLock(scopeA, async () => { + events.push("second-start"); + }); + + await vi.waitFor(() => expect(events).toContain("first-start")); + expect(events).toEqual(["first-start"]); + + releaseFirst(); + await Promise.all([first, second]); + expect(events).toEqual(["first-start", "first-end", "second-start"]); + }); + + test("releases the lock when the critical section throws", async () => { + await expect( + withSecretsStateLock(scopeA, async () => { + throw new Error("boom"); + }), + ).rejects.toThrow("boom"); + + expect(existsSync(lockPathFor(scopeA))).toBe(false); + await expect(withSecretsStateLock(scopeA, async () => "recovered")).resolves.toBe("recovered"); + }); + + test("waits while another process holds a fresh lock", async () => { + const lockPath = lockPathFor(scopeA); + mkdirSync(lockPath, { recursive: true }); + writeFileSync( + path.join(lockPath, "owner.json"), + JSON.stringify({ pid: 12345, token: "other" }), + ); + + const fn = vi.fn().mockResolvedValue("done"); + const pending = withSecretsStateLock(scopeA, fn); + + await new Promise((resolve) => setTimeout(resolve, 150)); + expect(fn).not.toHaveBeenCalled(); + + rmSync(lockPath, { recursive: true }); + await expect(pending).resolves.toBe("done"); + }); + + test("release leaves a lock taken over by another process", async () => { + const lockPath = lockPathFor(scopeA); + await withSecretsStateLock(scopeA, async () => { + writeFileSync( + path.join(lockPath, "owner.json"), + JSON.stringify({ pid: 12345, token: "other" }), + ); + }); + + expect(existsSync(lockPath)).toBe(true); + }); + + test("steals a lock whose lease has expired", async () => { + const lockPath = lockPathFor(scopeA); + mkdirSync(lockPath, { recursive: true }); + writeFileSync( + path.join(lockPath, "owner.json"), + JSON.stringify({ pid: 12345, token: "other" }), + ); + const expired = new Date(Date.now() - 2 * 60 * 1000); + utimesSync(lockPath, expired, expired); + + await expect(withSecretsStateLock(scopeA, async () => "stolen")).resolves.toBe("stolen"); + expect(existsSync(lockPath)).toBe(false); + }); + + test("runs without locking when the scope has no stable application id", async () => { + const scopeWithoutId = { ...scopeA, applicationId: undefined }; + const result = await withSecretsStateLock(scopeWithoutId, async () => "no-lock"); + expect(result).toBe("no-lock"); + expect(existsSync(path.dirname(getSecretsStatePath(scopeA)))).toBe(false); + }); +}); diff --git a/packages/sdk/src/cli/commands/deploy/secrets-state.ts b/packages/sdk/src/cli/commands/deploy/secrets-state.ts index fe147dba4..3f7826017 100644 --- a/packages/sdk/src/cli/commands/deploy/secrets-state.ts +++ b/packages/sdk/src/cli/commands/deploy/secrets-state.ts @@ -1,5 +1,14 @@ -import { createHash } from "node:crypto"; -import { mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { createHash, randomUUID } from "node:crypto"; +import { + mkdirSync, + readFileSync, + renameSync, + rmSync, + statSync, + utimesSync, + writeFileSync, +} from "node:fs"; +import pLimit, { type LimitFunction } from "p-limit"; import * as path from "pathe"; import { z } from "zod"; import { getDistDir } from "#/cli/shared/dist-dir"; @@ -82,8 +91,10 @@ export function saveSecretsState(scope: SecretsStateScope, state: SecretsState): const filePath = getSecretsStatePath(scope); const dir = path.dirname(filePath); mkdirSync(dir, { recursive: true }); + // Write via a temp file and rename so concurrent readers never see torn JSON. + const tempPath = `${filePath}.tmp-${process.pid}`; writeFileSync( - filePath, + tempPath, JSON.stringify( { version: 1, @@ -96,6 +107,7 @@ export function saveSecretsState(scope: SecretsStateScope, state: SecretsState): ), "utf-8", ); + renameSync(tempPath, filePath); } /** @@ -106,3 +118,151 @@ export function saveSecretsState(scope: SecretsStateScope, state: SecretsState): export function hashValue(value: string): string { return createHash("sha256").update(value).digest("hex"); } + +const LOCK_POLL_INTERVAL_MS = 100; +const LOCK_ACQUIRE_TIMEOUT_MS = 5 * 60 * 1000; +// A holder refreshes the lock directory mtime while working, so a lock whose +// mtime is older than the lease can only belong to a crashed or stopped +// process and is safe to steal. +const LOCK_HEARTBEAT_INTERVAL_MS = 10 * 1000; +const LOCK_LEASE_MS = 60 * 1000; + +const lockQueues = new Map(); + +/** + * Run a remote-update/state-save sequence exclusively for one target's secrets state. + * + * Serializes concurrent deploys to the same workspace and application (across + * processes sharing the same output directory) so the persisted hash state + * always reflects the last remote write. + * @param scope - Workspace and application identity for the deployment + * @param fn - Critical section performing remote updates and the state save + * @returns The value returned by fn + */ +export async function withSecretsStateLock( + scope: SecretsStateScope, + fn: () => Promise, +): Promise { + if (!scope.applicationId) { + return fn(); + } + const lockPath = `${getSecretsStatePath(scope)}.lock`; + let queue = lockQueues.get(lockPath); + if (!queue) { + queue = pLimit(1); + lockQueues.set(lockPath, queue); + } + return queue(async () => { + const token = await acquireFileLock(lockPath); + const heartbeat = setInterval(() => refreshLock(lockPath, token), LOCK_HEARTBEAT_INTERVAL_MS); + heartbeat.unref(); + try { + return await fn(); + } finally { + clearInterval(heartbeat); + releaseFileLock(lockPath, token); + } + }); +} + +async function acquireFileLock(lockPath: string): Promise { + mkdirSync(path.dirname(lockPath), { recursive: true }); + const token = randomUUID(); + const deadline = Date.now() + LOCK_ACQUIRE_TIMEOUT_MS; + for (;;) { + let created = true; + try { + mkdirSync(lockPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") { + throw error; + } + created = false; + } + if (created) { + try { + writeFileSync( + path.join(lockPath, "owner.json"), + JSON.stringify({ pid: process.pid, token }), + ); + } catch (error) { + rmSync(lockPath, { recursive: true, force: true }); + throw error; + } + return token; + } + if (isLockExpired(lockPath) && stealLock(lockPath)) { + continue; + } + if (Date.now() >= deadline) { + throw new Error( + `Timed out waiting for the secrets state lock at "${lockPath}". ` + + "Another deploy to the same workspace and application may still be running; " + + "remove the lock directory if no such deploy exists.", + ); + } + await new Promise((resolve) => setTimeout(resolve, LOCK_POLL_INTERVAL_MS)); + } +} + +function isLockExpired(lockPath: string): boolean { + try { + return Date.now() - statSync(lockPath).mtimeMs > LOCK_LEASE_MS; + } catch { + return false; + } +} + +function readLockToken(lockPath: string): unknown { + try { + return ( + JSON.parse(readFileSync(path.join(lockPath, "owner.json"), "utf-8")) as { token?: unknown } + ).token; + } catch { + return undefined; + } +} + +function refreshLock(lockPath: string, token: string): void { + if (readLockToken(lockPath) !== token) { + return; + } + const now = new Date(); + try { + utimesSync(lockPath, now, now); + } catch { + // The lock disappeared or was stolen; the release path handles ownership. + } +} + +function releaseFileLock(lockPath: string, token: string): void { + // Only remove a lock this process still owns, so a holder whose lease + // expired cannot delete the current holder's lock. + if (readLockToken(lockPath) !== token) { + return; + } + rmSync(lockPath, { recursive: true, force: true }); +} + +function stealLock(lockPath: string): boolean { + // Rename first so concurrent stealers cannot remove a lock that another + // contender has just re-acquired. + const trash = `${lockPath}.stale-${process.pid}-${Date.now()}`; + try { + renameSync(lockPath, trash); + } catch { + return false; + } + if (isLockExpired(trash)) { + rmSync(trash, { recursive: true, force: true }); + return true; + } + // A live holder re-acquired or refreshed the lock between the expiry check + // and the rename; hand it back. + try { + renameSync(trash, lockPath); + } catch { + rmSync(trash, { recursive: true, force: true }); + } + return false; +}