fix(e2e): serialize cleanup lock ownership

Bill Leoutsakos · cursoragent · Bill Leoutsakos · commit 4633f06c4dbc · 2026-07-21T16:27:27.000-07:00
Co-authored-by: Cursor &lt;cursoragent@cursor.com&gt;
diff --git a/apps/sim/e2e/fixtures/scenario-billing.ts b/apps/sim/e2e/fixtures/scenario-billing.ts
@@ -8,7 +8,7 @@ export function sameBillingReference(
   left: ScenarioSubscription,
   right: ScenarioSubscription
 ): boolean {
-  return billingReferenceIdentity(left) === billingReferenceIdentity(right)
+  return billingReferenceKey(left) === billingReferenceKey(right)
 }
 
 export function initialSubscriptionStatus(
@@ -51,7 +51,7 @@ export function expectedUsageLimit(scenario: ResolvedScenario, userKey: string):
   return '5'
 }
 
-function billingReferenceIdentity(subscription: ScenarioSubscription): string {
+export function billingReferenceKey(subscription: ScenarioSubscription): string {
   return subscription.billingReference.kind === 'user'
     ? `user/${subscription.billingReference.userKey}`
     : `organization/${subscription.billingReference.organizationKey}`
diff --git a/apps/sim/e2e/fixtures/validate-scenario.ts b/apps/sim/e2e/fixtures/validate-scenario.ts
@@ -13,6 +13,7 @@ import type {
   ScenarioWorkspaceGrant,
 } from './scenario'
 import { SCENARIO_VERSION } from './scenario'
+import { billingReferenceKey, isEntitledSubscription } from './scenario-billing'
 
 export class ScenarioValidationError extends Error {
   readonly issues: readonly string[]
@@ -412,7 +413,7 @@ function validateSubscriptions(
       }
     }
 
-    if (subscription.status === 'active' || subscription.status === 'past_due') {
+    if (isEntitledSubscription(subscription)) {
       if (entitledBillingReferences.has(reference)) {
         issues.push(`billing reference "${reference}" has duplicate entitled subscriptions`)
       }
@@ -461,7 +462,7 @@ function validateWorkspaces(
     }
     const entitledSubscription = definition.subscriptions.find(
       (candidate) =>
-        (candidate.status === 'active' || candidate.status === 'past_due') &&
+        isEntitledSubscription(candidate) &&
         billingReferenceKey(candidate) === payerReferenceKey(workspace)
     )
     if (entitledSubscription && subscription?.key !== entitledSubscription.key) {
@@ -685,7 +686,7 @@ function validatePersonaWorkspaceExpectation(
     : undefined
   const entitledSubscription = [...subscriptionsByKey.values()].find(
     (candidate) =>
-      (candidate.status === 'active' || candidate.status === 'past_due') &&
+      isEntitledSubscription(candidate) &&
       billingReferenceKey(candidate) === payerReferenceKey(workspace)
   )
   const actualPlan = entitledSubscription?.plan ?? 'free'
@@ -754,12 +755,6 @@ function membershipFor(
   )
 }
 
-function billingReferenceKey(subscription: ScenarioSubscription): string {
-  return subscription.billingReference.kind === 'user'
-    ? `user/${subscription.billingReference.userKey}`
-    : `organization/${subscription.billingReference.organizationKey}`
-}
-
 function payerReferenceKey(workspace: ScenarioWorkspace): string {
   return workspace.payer.kind === 'user'
     ? `user/${workspace.payer.userKey}`
diff --git a/apps/sim/e2e/foundation/build-manifest.spec.ts b/apps/sim/e2e/foundation/build-manifest.spec.ts
@@ -166,6 +166,43 @@ test('orchestrator lock rejects live ownership and recovers stale descriptors',
   }
 })
 
+test('transferred run lock rejects mutations from its former owner', async () => {
+  const directory = mkdtempSync(path.join(os.tmpdir(), 'sim-e2e-transferred-run-lock-'))
+  const lockPath = path.join(directory, 'orchestrator.lock')
+  const supervisor = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1_000)'], {
+    stdio: 'ignore',
+  })
+  try {
+    await once(supervisor, 'spawn')
+    if (!supervisor.pid) throw new Error('Expected cleanup supervisor PID')
+    const lock = acquireE2eRunLock(lockPath)
+    expect(lock.transfer(supervisor.pid)).toBe(true)
+    expect(lock.transfer(process.pid)).toBe(false)
+
+    lock.setProcessGroupIds([supervisor.pid])
+    lock.retain('former owner must not retain')
+    lock.release()
+
+    const descriptor = JSON.parse(
+      readFileSync(path.join(lockPath, 'owner.json'), 'utf8')
+    ) as Record<string, unknown>
+    expect(descriptor.pid).toBe(supervisor.pid)
+    expect(descriptor.processGroupIds).toEqual([])
+    expect(descriptor.retainedFailure).toBeUndefined()
+    expect(existsSync(lockPath)).toBe(true)
+
+    supervisor.kill('SIGKILL')
+    await once(supervisor, 'exit')
+    const recovered = acquireE2eRunLock(lockPath)
+    recovered.release()
+  } finally {
+    if (supervisor.exitCode === null && supervisor.signalCode === null) {
+      supervisor.kill('SIGKILL')
+    }
+    rmSync(directory, { recursive: true, force: true })
+  }
+})
+
 test('stale orchestrator recovery terminates its persisted process groups', async () => {
   test.skip(process.platform === 'win32', 'POSIX process-group cleanup is tested on Unix')
   const directory = mkdtempSync(path.join(os.tmpdir(), 'sim-e2e-stale-process-group-'))
diff --git a/apps/sim/e2e/foundation/safety.spec.ts b/apps/sim/e2e/foundation/safety.spec.ts
@@ -257,12 +257,26 @@ test.describe('foundation safety guards', () => {
     expect(cleanup.start('SIGINT')).toBe(first)
     expect(cleanup.start('SIGTERM')).toBe(first)
     expect(cleanup.isStarted()).toBe(true)
+    expect(cleanup.claimNormalFinalization()).toBe(false)
     await Promise.resolve()
     expect(handledSignals).toEqual(['SIGINT'])
     finishCleanup()
     await first
   })
 
+  test('normal finalization atomically prevents later signal cleanup', async () => {
+    let signalCleanupCalls = 0
+    const cleanup = createSingleFlightSignalCleanup(async () => {
+      signalCleanupCalls += 1
+    })
+
+    expect(cleanup.claimNormalFinalization()).toBe(true)
+    expect(cleanup.claimNormalFinalization()).toBe(false)
+    await cleanup.start('SIGTERM')
+    expect(cleanup.isStarted()).toBe(false)
+    expect(signalCleanupCalls).toBe(0)
+  })
+
   test('port preflight rejects an existing listener', async () => {
     const server = createServer()
     await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve))
diff --git a/apps/sim/e2e/scripts/run.ts b/apps/sim/e2e/scripts/run.ts
@@ -164,6 +164,8 @@ async function main(): Promise<void> {
   }
 
   const signalCleanup = createSingleFlightSignalCleanup(async (signal) => {
+    const cleanupProcessGroupIds = getActiveManagedProcessGroupIds()
+    setManagedProcessGroupObserver(null)
     failed = true
     const exitCode = signal === 'SIGINT' ? 130 : 143
     process.exitCode = exitCode
@@ -197,7 +199,7 @@ async function main(): Promise<void> {
                 E2E_PG_ADMIN_URL: adminDatabaseUrl,
                 E2E_DATABASE_NAME: runDatabase.name,
                 E2E_DATABASE_CREATION_COMPLETE: String(databaseCreationComplete),
-                E2E_CLEANUP_PROCESS_GROUPS: getActiveManagedProcessGroupIds().join(','),
+                E2E_CLEANUP_PROCESS_GROUPS: cleanupProcessGroupIds.join(','),
                 E2E_CLEANUP_DIRECTORIES: JSON.stringify([
                   storageStateDirectory,
                   privateDirectory,
@@ -368,79 +370,80 @@ async function main(): Promise<void> {
     console.error(error)
     process.exitCode = 1
   } finally {
-    if (runDatabase && existsSync(manifestPath)) {
+    if (signalCleanup.claimNormalFinalization()) {
+      process.off('SIGINT', handleSigint)
+      process.off('SIGTERM', handleSigterm)
+
+      if (runDatabase && existsSync(manifestPath)) {
+        try {
+          await assertManifestWorkspaceIdentities(runDatabase.url, manifestPath)
+        } catch (error) {
+          failed = true
+          process.exitCode = 1
+          console.error(error)
+        }
+      }
       try {
-        await assertManifestWorkspaceIdentities(runDatabase.url, manifestPath)
+        leakCanarySecrets = loadSyntheticSecretCanaryForScan(leakCanarySecrets, canarySecretsPath)
       } catch (error) {
+        canaryCoverageComplete = false
         failed = true
         process.exitCode = 1
         console.error(error)
+      } finally {
+        rmSync(credentialsPath, { force: true })
+        rmSync(canarySecretsPath, { force: true })
       }
-    }
-    try {
-      leakCanarySecrets = loadSyntheticSecretCanaryForScan(leakCanarySecrets, canarySecretsPath)
-    } catch (error) {
-      canaryCoverageComplete = false
-      failed = true
-      process.exitCode = 1
-      console.error(error)
-    } finally {
-      rmSync(credentialsPath, { force: true })
-      rmSync(canarySecretsPath, { force: true })
-    }
-    let cleanupSucceeded = false
-    try {
-      await cleanup()
-      cleanupSucceeded = true
-    } catch (error) {
-      failed = true
-      process.exitCode = 1
-      console.error(error)
-    }
-    try {
-      assertNoForbiddenProviderTraffic(
-        [app?.logPath, realtime?.logPath].filter((value): value is string => Boolean(value))
-      )
-    } catch (error) {
-      failed = true
-      process.exitCode = 1
-      console.error(error)
-    }
-    if (canaryCoverageComplete && leakCanarySecrets.length > 0) {
+      let cleanupSucceeded = false
       try {
-        await assertNoSyntheticSecretLeaks({
-          secrets: leakCanarySecrets,
-          roots: diagnosticRoots,
-          excludedPaths: [privateDirectory, storageStateDirectory],
-        })
-        writeFileSync(
-          path.join(markerDirectory, 'leak-scan-complete.json'),
-          `${JSON.stringify({ runId, completedAt: new Date().toISOString() })}\n`,
-          { mode: 0o600 }
+        await cleanup()
+        cleanupSucceeded = true
+      } catch (error) {
+        failed = true
+        process.exitCode = 1
+        console.error(error)
+      }
+      try {
+        assertNoForbiddenProviderTraffic(
+          [app?.logPath, realtime?.logPath].filter((value): value is string => Boolean(value))
         )
       } catch (error) {
         failed = true
         process.exitCode = 1
         console.error(error)
+      }
+      if (canaryCoverageComplete && leakCanarySecrets.length > 0) {
+        try {
+          await assertNoSyntheticSecretLeaks({
+            secrets: leakCanarySecrets,
+            roots: diagnosticRoots,
+            excludedPaths: [privateDirectory, storageStateDirectory],
+          })
+          writeFileSync(
+            path.join(markerDirectory, 'leak-scan-complete.json'),
+            `${JSON.stringify({ runId, completedAt: new Date().toISOString() })}\n`,
+            { mode: 0o600 }
+          )
+        } catch (error) {
+          failed = true
+          process.exitCode = 1
+          console.error(error)
+          diagnosticsRetained = scrubDiagnostics(diagnosticRoots)
+          if (diagnosticsRetained) cleanupSucceeded = false
+        }
+      } else if (!canaryCoverageComplete) {
         diagnosticsRetained = scrubDiagnostics(diagnosticRoots)
         if (diagnosticsRetained) cleanupSucceeded = false
       }
-    } else if (!canaryCoverageComplete) {
-      diagnosticsRetained = scrubDiagnostics(diagnosticRoots)
-      if (diagnosticsRetained) cleanupSucceeded = false
-    }
-    if (!signalCleanup.isStarted()) {
-      process.off('SIGINT', handleSigint)
-      process.off('SIGTERM', handleSigterm)
       setManagedProcessGroupObserver(null)
       if (cleanupSucceeded) runLock.release()
       else runLock.retain('normal cleanup failed; inspect diagnostics and clean resources manually')
+      console.info(
+        diagnosticsRetained
+          ? `E2E ${failed ? 'failed' : 'completed'}; diagnostics: ${runDirectory}`
+          : 'E2E failed; diagnostics were scrubbed because complete secret scanning was impossible'
+      )
     }
-    console.info(
-      diagnosticsRetained
-        ? `E2E ${failed ? 'failed' : 'completed'}; diagnostics: ${runDirectory}`
-        : 'E2E failed; diagnostics were scrubbed because complete secret scanning was impossible'
-    )
   }
 }
 
diff --git a/apps/sim/e2e/scripts/signal-cleanup.ts b/apps/sim/e2e/scripts/signal-cleanup.ts
@@ -55,13 +55,13 @@ process.exit(failures.length > 0 ? 1 : 0)
 
 function releaseRunLock(): void {
   if (runLockPath && runLockToken) {
-    releaseE2eRunLock(runLockPath, runLockToken)
+    releaseE2eRunLock(runLockPath, runLockToken, process.pid)
   }
 }
 
 function retainRunLock(reason: string): void {
   if (runLockPath && runLockToken) {
-    retainE2eRunLock(runLockPath, runLockToken, reason)
+    retainE2eRunLock(runLockPath, runLockToken, process.pid, reason)
   }
 }
 
diff --git a/apps/sim/e2e/support/run-lock.ts b/apps/sim/e2e/support/run-lock.ts
@@ -59,7 +59,7 @@ export function acquireE2eRunLock(
         token: descriptor.token,
         setProcessGroupIds(processGroupIds: number[]): void {
           const current = readDescriptor(lockPath)
-          if (current?.token !== descriptor.token) return
+          if (current?.token !== descriptor.token || current.pid !== descriptor.pid) return
           const normalized = [...new Set(processGroupIds)].filter(
             (processGroupId) => Number.isInteger(processGroupId) && processGroupId > 0
           )
@@ -76,7 +76,7 @@ export function acquireE2eRunLock(
         },
         transfer(pid: number): boolean {
           const current = readDescriptor(lockPath)
-          if (current?.token !== descriptor.token) return false
+          if (current?.token !== descriptor.token || current.pid !== descriptor.pid) return false
           writeDescriptor(lockPath, {
             ...current,
             pid,
@@ -86,10 +86,10 @@ export function acquireE2eRunLock(
           return transferred?.token === descriptor.token && transferred.pid === pid
         },
         retain(reason: string): void {
-          retainE2eRunLock(lockPath, descriptor.token, reason)
+          retainE2eRunLock(lockPath, descriptor.token, descriptor.pid, reason)
         },
         release(): void {
-          releaseE2eRunLock(lockPath, descriptor.token)
+          releaseE2eRunLock(lockPath, descriptor.token, descriptor.pid)
         },
       }
     } catch (error) {
@@ -119,17 +119,22 @@ export function acquireE2eRunLock(
   throw new Error(`Unable to acquire E2E orchestrator lock: ${lockPath}`)
 }
 
-export function releaseE2eRunLock(lockPath: string, token: string): void {
+export function releaseE2eRunLock(lockPath: string, token: string, expectedOwnerPid: number): void {
   if (!existsSync(lockPath)) return
   const current = readDescriptor(lockPath)
-  if (current?.token === token) {
+  if (current?.token === token && current.pid === expectedOwnerPid) {
     rmSync(lockPath, { recursive: true, force: true })
   }
 }
 
-export function retainE2eRunLock(lockPath: string, token: string, reason: string): void {
+export function retainE2eRunLock(
+  lockPath: string,
+  token: string,
+  expectedOwnerPid: number,
+  reason: string
+): void {
   const current = readDescriptor(lockPath)
-  if (current?.token !== token) return
+  if (current?.token !== token || current.pid !== expectedOwnerPid) return
   writeDescriptor(lockPath, { ...current, retainedFailure: reason })
 }
 
diff --git a/apps/sim/e2e/support/signal-cleanup.ts b/apps/sim/e2e/support/signal-cleanup.ts
@@ -9,16 +9,25 @@ export function parseProcessGroupIds(rawValue: string | undefined): number[] {
 
 export interface SingleFlightSignalCleanup {
   isStarted(): boolean
+  claimNormalFinalization(): boolean
   start(signal: NodeJS.Signals): Promise<void>
 }
 
 export function createSingleFlightSignalCleanup(
   cleanup: (signal: NodeJS.Signals) => Promise<void>
 ): SingleFlightSignalCleanup {
+  let finalizationOwner: 'none' | 'normal' | 'signal' = 'none'
   let cleanupPromise: Promise<void> | null = null
   return {
-    isStarted: () => cleanupPromise !== null,
+    isStarted: () => finalizationOwner === 'signal',
+    claimNormalFinalization() {
+      if (finalizationOwner !== 'none') return false
+      finalizationOwner = 'normal'
+      return true
+    },
     start(signal) {
+      if (finalizationOwner === 'normal') return Promise.resolve()
+      finalizationOwner = 'signal'
       cleanupPromise ??= Promise.resolve().then(() => cleanup(signal))
       return cleanupPromise
     },

Original file line number	Diff line number	Diff line change
`@@ -13,6 +13,7 @@ import type {`
`13`	`13`	`ScenarioWorkspaceGrant,`
`14`	`14`	`} from './scenario'`
`15`	`15`	`import { SCENARIO_VERSION } from './scenario'`
	`16`	`+import { billingReferenceKey, isEntitledSubscription } from './scenario-billing'`
`16`	`17`
`17`	`18`	`export class ScenarioValidationError extends Error {`
`18`	`19`	`readonly issues: readonly string[]`
`@@ -412,7 +413,7 @@ function validateSubscriptions(`
`412`	`413`	`}`
`413`	`414`	`}`
`414`	`415`
`415`		`- if (subscription.status === 'active' \|\| subscription.status === 'past_due') {`
	`416`	`+ if (isEntitledSubscription(subscription)) {`
`416`	`417`	`if (entitledBillingReferences.has(reference)) {`
`417`	`418`	issues.push(`billing reference "${reference}" has duplicate entitled subscriptions`)
`418`	`419`	`}`
`@@ -461,7 +462,7 @@ function validateWorkspaces(`
`461`	`462`	`}`
`462`	`463`	`const entitledSubscription = definition.subscriptions.find(`
`463`	`464`	`(candidate) =>`
`464`		`- (candidate.status === 'active' \|\| candidate.status === 'past_due') &&`
	`465`	`+ isEntitledSubscription(candidate) &&`
`465`	`466`	`billingReferenceKey(candidate) === payerReferenceKey(workspace)`
`466`	`467`	`)`
`467`	`468`	`if (entitledSubscription && subscription?.key !== entitledSubscription.key) {`
`@@ -685,7 +686,7 @@ function validatePersonaWorkspaceExpectation(`
`685`	`686`	`: undefined`
`686`	`687`	`const entitledSubscription = [...subscriptionsByKey.values()].find(`
`687`	`688`	`(candidate) =>`
`688`		`- (candidate.status === 'active' \|\| candidate.status === 'past_due') &&`
	`689`	`+ isEntitledSubscription(candidate) &&`
`689`	`690`	`billingReferenceKey(candidate) === payerReferenceKey(workspace)`
`690`	`691`	`)`
`691`	`692`	`const actualPlan = entitledSubscription?.plan ?? 'free'`
`@@ -754,12 +755,6 @@ function membershipFor(`
`754`	`755`	`)`
`755`	`756`	`}`
`756`	`757`
`757`		`-function billingReferenceKey(subscription: ScenarioSubscription): string {`
`758`		`- return subscription.billingReference.kind === 'user'`
`759`		- ? `user/${subscription.billingReference.userKey}`
`760`		- : `organization/${subscription.billingReference.organizationKey}`
`761`		`-}`
`762`		`-`
`763`	`758`	`function payerReferenceKey(workspace: ScenarioWorkspace): string {`
`764`	`759`	`return workspace.payer.kind === 'user'`
`765`	`760`	? `user/${workspace.payer.userKey}`
Original file line number	Diff line number	Diff line change
`@@ -55,13 +55,13 @@ process.exit(failures.length > 0 ? 1 : 0)`
`55`	`55`
`56`	`56`	`function releaseRunLock(): void {`
`57`	`57`	`if (runLockPath && runLockToken) {`
`58`		`- releaseE2eRunLock(runLockPath, runLockToken)`
	`58`	`+ releaseE2eRunLock(runLockPath, runLockToken, process.pid)`
`59`	`59`	`}`
`60`	`60`	`}`
`61`	`61`
`62`	`62`	`function retainRunLock(reason: string): void {`
`63`	`63`	`if (runLockPath && runLockToken) {`
`64`		`- retainE2eRunLock(runLockPath, runLockToken, reason)`
	`64`	`+ retainE2eRunLock(runLockPath, runLockToken, process.pid, reason)`
`65`	`65`	`}`
`66`	`66`	`}`
`67`	`67`