From dbb21a9f0693d7309713ab4fe62c89791fbfa819 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 25 Aug 2026 17:38:42 +0800 Subject: [PATCH 1/2] fix(storage): bind session cleanup to process lifetime Use an OS-held native file lock as the process-incarnation identity for temporary Session-copy cleanup. Preserve PID-only recovery for leases written by older clients and hold one recovery claim across Host-side removal. Generated-by: Codex --- .../cli/src/runtime-host-session-driver.ts | 4 + packages/cli/src/runtime-host-tui-context.ts | 26 ++- packages/storage/package.json | 1 + .../fixtures/process-lifetime-owner-holder.ts | 27 +++ .../__tests__/process-lifetime-owner.test.ts | 108 +++++++++++ .../__tests__/session-copy-cleanup.test.ts | 171 ++++++++++++++++++ packages/storage/src/native-file-lock.ts | 80 ++++++++ .../src/process-lifetime-file-update-lock.ts | 65 ++----- .../storage/src/process-lifetime-owner.ts | 131 ++++++++++++++ packages/storage/src/session-copy-cleanup.ts | 88 +++++++-- 10 files changed, 635 insertions(+), 66 deletions(-) create mode 100644 packages/storage/src/__tests__/fixtures/process-lifetime-owner-holder.ts create mode 100644 packages/storage/src/__tests__/process-lifetime-owner.test.ts create mode 100644 packages/storage/src/native-file-lock.ts create mode 100644 packages/storage/src/process-lifetime-owner.ts diff --git a/packages/cli/src/runtime-host-session-driver.ts b/packages/cli/src/runtime-host-session-driver.ts index b65ba1a172..65751106d4 100644 --- a/packages/cli/src/runtime-host-session-driver.ts +++ b/packages/cli/src/runtime-host-session-driver.ts @@ -40,6 +40,7 @@ import { createSessionCopyCleanupAuthority, type SessionCopyCleanupAuthority, } from '@maka/storage/session-copy-cleanup'; +import type { ProcessLifetimeOwner } from '@maka/storage/process-lifetime-owner'; import type { OrchestrationMode } from '@maka/core/orchestration'; import type { PermissionMode } from '@maka/core/permission'; @@ -128,6 +129,8 @@ export interface RuntimeHostMakaSessionDriverInput { executionLocation?: { readonly kind: 'client_path' } | { readonly kind: 'host' }; /** Client-local durable lease parent for temporary TUI conversation copies. */ sessionCopyCleanupRoot?: string; + /** Process-incarnation owner for temporary TUI conversation copies. */ + sessionCopyCleanupOwner?: ProcessLifetimeOwner; } type RuntimeHostSessionDriverConnection = Pick< @@ -229,6 +232,7 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { resumeSessionCopy: (creation) => this.#resumeSessionCopy(creation), processId: `tui:${process.pid}`, isOwnerProcessActive: isTuiProcessActive, + processLifetimeOwner: input.sessionCopyCleanupOwner, }) : undefined; this.moveSession = diff --git a/packages/cli/src/runtime-host-tui-context.ts b/packages/cli/src/runtime-host-tui-context.ts index 6a4e998857..815476bf9a 100644 --- a/packages/cli/src/runtime-host-tui-context.ts +++ b/packages/cli/src/runtime-host-tui-context.ts @@ -28,6 +28,10 @@ import { findProjectByIdentity } from '@maka/core/project'; import type { ConnectionCatalogEntry, ConnectionCatalogSnapshot } from '@maka/core/runtime-policy'; import { SessionActivityRegistry } from '@maka/runtime/goal-turn-lifecycle'; import { type InvocableSkillEntry } from '@maka/runtime/skill-invocation'; +import { + acquireProcessLifetimeOwner, + type ProcessLifetimeOwner, +} from '@maka/storage/process-lifetime-owner'; import { readRuntimeHostAgentGraphEpochs, readRuntimeHostInvocableSkills, @@ -112,6 +116,7 @@ export async function createRuntimeHostTuiContext( }); const connection = connected.connection; let mcp: TuiMcpController | undefined; + let sessionCopyCleanupOwner: ProcessLifetimeOwner | undefined; try { const catalog = connected.catalog; const workspace = await resolveRuntimeHostTuiWorkspace(connection, connected.profile, input); @@ -126,13 +131,19 @@ export async function createRuntimeHostTuiContext( executionBoundaryDisplayMode( createGenesisExecutionBoundary(await readHostChatDefaultPermissionMode(connection)), ) ?? 'ask'; + const sessionCopyCleanupRoot = join(input.clientDataRoot, 'tui-session-copies'); + const owner = await acquireProcessLifetimeOwner( + join(sessionCopyCleanupRoot, connection.rootId), + ); + sessionCopyCleanupOwner = owner; const driverInput: RuntimeHostMakaSessionDriverInput = { connection, cwd: input.cwd, llmConnectionSlug: target.connection.slug, model: target.model, prospectivePermissionMode, - sessionCopyCleanupRoot: join(input.clientDataRoot, 'tui-session-copies'), + sessionCopyCleanupRoot, + sessionCopyCleanupOwner: owner, executionLocation: connected.profile.kind === 'local' ? { kind: 'client_path' } : { kind: 'host' }, ...(workspace ? { workspace } : {}), @@ -174,17 +185,19 @@ export async function createRuntimeHostTuiContext( onboarding: createRuntimeHostOnboardingSurface(connection), ...(mcp ? { mcp } : {}), profile: connected.profile, - close: () => closeRuntimeHostTuiContext(mcp, connected.close), + close: () => closeRuntimeHostTuiContext(mcp, owner, connected.close), }; } catch (error) { - await mcp?.close().catch(() => undefined); - await connected.close().catch(() => undefined); + await closeRuntimeHostTuiContext(mcp, sessionCopyCleanupOwner, connected.close).catch( + () => undefined, + ); throw error; } } async function closeRuntimeHostTuiContext( mcp: TuiMcpController | undefined, + sessionCopyCleanupOwner: ProcessLifetimeOwner | undefined, closeConnection: () => Promise, ): Promise { const errors: unknown[] = []; @@ -193,6 +206,11 @@ async function closeRuntimeHostTuiContext( } catch (error) { errors.push(error); } + try { + await sessionCopyCleanupOwner?.close(); + } catch (error) { + errors.push(error); + } try { await closeConnection(); } catch (error) { diff --git a/packages/storage/package.json b/packages/storage/package.json index fc466ae84f..69f0eddf97 100644 --- a/packages/storage/package.json +++ b/packages/storage/package.json @@ -32,6 +32,7 @@ "./pet-pack-store": "./dist/pet-pack-store.js", "./plan-authority": "./dist/plan-authority.js", "./process-lifetime-file-update-lock": "./dist/process-lifetime-file-update-lock.js", + "./process-lifetime-owner": "./dist/process-lifetime-owner.js", "./project-catalog": "./dist/project-catalog.js", "./project-catalog-authority": "./dist/project-catalog-authority.js", "./root-authority": "./dist/root-authority.js", diff --git a/packages/storage/src/__tests__/fixtures/process-lifetime-owner-holder.ts b/packages/storage/src/__tests__/fixtures/process-lifetime-owner-holder.ts new file mode 100644 index 0000000000..b3b3091bf6 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/process-lifetime-owner-holder.ts @@ -0,0 +1,27 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { acquireProcessLifetimeOwner } from '../../process-lifetime-owner.js'; + +const root = process.argv[2]; +if (!root) throw new Error('Missing process lifetime owner root'); + +const owner = await acquireProcessLifetimeOwner(root); +process.send?.({ reference: owner.reference }); +await new Promise(() => setInterval(() => undefined, 1_000)); diff --git a/packages/storage/src/__tests__/process-lifetime-owner.test.ts b/packages/storage/src/__tests__/process-lifetime-owner.test.ts new file mode 100644 index 0000000000..7ce56dee8e --- /dev/null +++ b/packages/storage/src/__tests__/process-lifetime-owner.test.ts @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { strict as assert } from 'node:assert'; +import { fork } from 'node:child_process'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test, type TestContext } from 'node:test'; +import { acquireProcessLifetimeOwner } from '../process-lifetime-owner.js'; + +test('claims an owner only after its process exits', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-process-lifetime-owner-')); + const child = fork( + new URL('./fixtures/process-lifetime-owner-holder.js', import.meta.url), + [root], + { stdio: ['ignore', 'ignore', 'inherit', 'ipc'] }, + ); + const observer = await acquireProcessLifetimeOwner(root); + const competingObserver = await acquireProcessLifetimeOwner(root); + t.after(async () => { + if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL'); + await observer.close().catch(() => undefined); + await competingObserver.close().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + }); + + const reference = await waitForReference(t, child); + assert.equal(await observer.tryClaimReleased(reference), undefined); + + child.kill('SIGKILL'); + await new Promise((resolve) => child.once('exit', () => resolve())); + + const claims = await Promise.all([ + observer.tryClaimReleased(reference), + competingObserver.tryClaimReleased(reference), + ]); + const acquired = claims.filter((claim) => claim !== undefined); + assert.equal(acquired.length, 1); + await acquired[0]?.retire(); +}); + +test('releases its owner reference on graceful close', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-process-lifetime-owner-')); + const owner = await acquireProcessLifetimeOwner(root); + const observer = await acquireProcessLifetimeOwner(root); + t.after(async () => { + await owner.close().catch(() => undefined); + await observer.close().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + }); + + assert.equal(await observer.tryClaimReleased(owner.reference), undefined); + await owner.close(); + const claim = await observer.tryClaimReleased(owner.reference); + assert.ok(claim); + await claim.retire(); +}); + +test('rejects owner references outside its versioned namespace', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-process-lifetime-owner-')); + const owner = await acquireProcessLifetimeOwner(root); + t.after(async () => { + await owner.close().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + }); + + await assert.rejects(owner.tryClaimReleased('../other'), /Invalid process lifetime owner/); +}); + +async function waitForReference(t: TestContext, child: ReturnType): Promise { + return new Promise((resolve, reject) => { + const onMessage = (message: unknown) => { + if ( + typeof message === 'object' && + message !== null && + 'reference' in message && + typeof message.reference === 'string' + ) { + resolve(message.reference); + } else { + reject(new Error(`Unexpected child message: ${String(message)}`)); + } + }; + child.once('message', onMessage); + child.once('error', reject); + child.once('exit', (code, signal) => { + reject(new Error(`Owner exited before acquisition (${String(code)}, ${signal})`)); + }); + t.after(() => child.off('message', onMessage)); + }); +} diff --git a/packages/storage/src/__tests__/session-copy-cleanup.test.ts b/packages/storage/src/__tests__/session-copy-cleanup.test.ts index b83ae4a65d..fc75ed38d9 100644 --- a/packages/storage/src/__tests__/session-copy-cleanup.test.ts +++ b/packages/storage/src/__tests__/session-copy-cleanup.test.ts @@ -24,6 +24,10 @@ import { join } from 'node:path'; import { afterEach, describe, it } from 'node:test'; import { DatabaseSync } from 'node:sqlite'; import { createSessionCopyCleanupAuthority } from '../session-copy-cleanup.js'; +import type { + ProcessLifetimeOwner, + ProcessLifetimeRecoveryClaim, +} from '../process-lifetime-owner.js'; const roots: string[] = []; @@ -227,6 +231,92 @@ describe('session copy cleanup authority', () => { assert.deepEqual(await readPendingIds(workspaceRoot), ['fork-live-owner']); }); + it('claims one released process incarnation across all of its copies', async () => { + const workspaceRoot = await createWorkspace(); + const original = fakeLifetimeOwner('lock-v1:11111111-1111-4111-8111-111111111111'); + const owner = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:101', + processLifetimeOwner: original.owner, + removeSession: async () => {}, + }); + for (const sessionId of ['fork-lifetime-a', 'fork-lifetime-b']) { + await owner.ownCreation( + { + sessionId, + kind: 'branch', + sourceSessionId: 'source-session', + sourceTurnId: 'source-turn', + ownerId: 'tui-side', + }, + async () => 'created', + ); + } + + const successor = fakeLifetimeOwner('lock-v1:22222222-2222-4222-8222-222222222222'); + const removed: string[] = []; + const authority = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:202', + processLifetimeOwner: successor.owner, + removeSession: async (sessionId) => { + assert.equal(successor.claimClosed, false); + removed.push(sessionId); + }, + }); + + assert.deepEqual(await authority.recover(), { + removed: ['fork-lifetime-a', 'fork-lifetime-b'], + failed: [], + }); + assert.deepEqual(removed, ['fork-lifetime-a', 'fork-lifetime-b']); + assert.equal(successor.claimAttempts, 1); + assert.equal(successor.claimRetired, true); + assert.deepEqual(await readPendingIds(workspaceRoot), []); + }); + + it('fails closed when process-incarnation ownership cannot be inspected', async () => { + const workspaceRoot = await createWorkspace(); + const original = fakeLifetimeOwner('lock-v1:33333333-3333-4333-8333-333333333333'); + const owner = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:303', + processLifetimeOwner: original.owner, + removeSession: async () => {}, + }); + await owner.ownCreation( + { + sessionId: 'fork-unknown-owner', + kind: 'branch', + sourceSessionId: 'source-session', + sourceTurnId: 'source-turn', + ownerId: 'tui-side', + }, + async () => 'created', + ); + + const failure = new Error('native lock unavailable'); + const authority = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:404', + processLifetimeOwner: { + reference: 'lock-v1:44444444-4444-4444-8444-444444444444', + tryClaimReleased: async () => { + throw failure; + }, + close: async () => {}, + }, + removeSession: async () => { + throw new Error('must not remove an indeterminate owner'); + }, + }); + + const recovery = await authority.recover(); + assert.deepEqual(recovery.removed, []); + assert.deepEqual(recovery.failed, [{ sessionId: 'fork-unknown-owner', error: failure }]); + assert.deepEqual(await readPendingIds(workspaceRoot), ['fork-unknown-owner']); + }); + it('abandons every live copy owned by a renderer that exits', async () => { const workspaceRoot = await createWorkspace(); const removed: string[] = []; @@ -255,6 +345,48 @@ describe('session copy cleanup authority', () => { assert.deepEqual(await readPendingIds(workspaceRoot), []); }); + it('abandons only copies owned by the current process incarnation', async () => { + const workspaceRoot = await createWorkspace(); + const firstLifetime = fakeLifetimeOwner('lock-v1:55555555-5555-4555-8555-555555555555'); + const secondLifetime = fakeLifetimeOwner('lock-v1:66666666-6666-4666-8666-666666666666'); + const first = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:505', + processLifetimeOwner: firstLifetime.owner, + removeSession: async () => {}, + }); + const removed: string[] = []; + const second = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:505', + processLifetimeOwner: secondLifetime.owner, + removeSession: async (sessionId) => { + removed.push(sessionId); + }, + }); + for (const [authority, sessionId] of [ + [first, 'fork-first-incarnation'], + [second, 'fork-second-incarnation'], + ] as const) { + await authority.ownCreation( + { + sessionId, + kind: 'branch', + sourceSessionId: 'source-session', + sourceTurnId: 'source-turn', + ownerId: 'tui-side', + }, + async () => 'created', + ); + } + + await second.abandonOwner('tui-side'); + await second.cleanup('fork-second-incarnation'); + + assert.deepEqual(removed, ['fork-second-incarnation']); + assert.deepEqual(await readPendingIds(workspaceRoot), ['fork-first-incarnation']); + }); + it('acknowledges abandon after the intent is durable without waiting for removal', async () => { const workspaceRoot = await createWorkspace(); let releaseRemoval: (() => void) | undefined; @@ -377,3 +509,42 @@ async function readPendingIds(workspaceRoot: string): Promise { database.close(); } } + +function fakeLifetimeOwner(reference: string): { + owner: ProcessLifetimeOwner; + readonly claimAttempts: number; + readonly claimClosed: boolean; + readonly claimRetired: boolean; +} { + let claimAttempts = 0; + let claimClosed = false; + let claimRetired = false; + const claim: ProcessLifetimeRecoveryClaim = { + retire: async () => { + claimRetired = true; + claimClosed = true; + }, + close: async () => { + claimClosed = true; + }, + }; + return { + owner: { + reference, + tryClaimReleased: async () => { + claimAttempts += 1; + return claim; + }, + close: async () => {}, + }, + get claimAttempts() { + return claimAttempts; + }, + get claimClosed() { + return claimClosed; + }, + get claimRetired() { + return claimRetired; + }, + }; +} diff --git a/packages/storage/src/native-file-lock.ts b/packages/storage/src/native-file-lock.ts new file mode 100644 index 0000000000..980750025a --- /dev/null +++ b/packages/storage/src/native-file-lock.ts @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +/// + +import { constants as fsConstants } from 'node:fs'; +import { lstat, open, unlink, type FileHandle } from 'node:fs/promises'; +import { tryLock, unlock } from 'fs-native-extensions'; + +export async function openStableNativeLockFile(path: string): Promise { + const handle = await open( + path, + fsConstants.O_CREAT | fsConstants.O_RDWR | fsConstants.O_NOFOLLOW, + 0o600, + ); + try { + await assertStableRegularFile(handle, path); + if (process.platform !== 'win32') await handle.chmod(0o600); + return handle; + } catch (error) { + await handle.close(); + throw error; + } +} + +export function tryAcquireNativeFileLock(handle: FileHandle): boolean { + return tryLock(handle.fd); +} + +export function releaseNativeFileLock(handle: FileHandle): void { + try { + unlock(handle.fd); + } catch { + // Closing the OS handle is the authoritative release path. + } +} + +export async function unlinkStableNativeLockFile(handle: FileHandle, path: string): Promise { + try { + await assertStableRegularFile(handle, path); + await unlink(path); + } catch (error) { + if (!isNodeError(error, 'ENOENT')) throw error; + } +} + +async function assertStableRegularFile(handle: FileHandle, path: string): Promise { + const [handleStat, pathStat] = await Promise.all([ + handle.stat({ bigint: true }), + lstat(path, { bigint: true }), + ]); + if ( + !handleStat.isFile() || + !pathStat.isFile() || + handleStat.dev !== pathStat.dev || + handleStat.ino !== pathStat.ino + ) { + throw new Error(`Native lock is not one stable regular file: ${path}`); + } +} + +function isNodeError(error: unknown, code: string): error is NodeJS.ErrnoException { + return error instanceof Error && 'code' in error && error.code === code; +} diff --git a/packages/storage/src/process-lifetime-file-update-lock.ts b/packages/storage/src/process-lifetime-file-update-lock.ts index 3a928820fd..e8ce28ddcb 100644 --- a/packages/storage/src/process-lifetime-file-update-lock.ts +++ b/packages/storage/src/process-lifetime-file-update-lock.ts @@ -17,11 +17,13 @@ * under the License. */ -/// - import { constants as fsConstants } from 'node:fs'; -import { lstat, open, rmdir, unlink, type FileHandle } from 'node:fs/promises'; -import { tryLock, unlock } from 'fs-native-extensions'; +import { lstat, open, rmdir, unlink } from 'node:fs/promises'; +import { + openStableNativeLockFile, + releaseNativeFileLock, + tryAcquireNativeFileLock, +} from './native-file-lock.js'; const LOCK_POLL_MS = 25; const LOCK_TIMEOUT_MS = 10_000; @@ -37,12 +39,14 @@ export async function withLegacyFileUpdateLockLease( const supervisionPath = `${targetPath}.supervised`; const deadline = Date.now() + timeoutMs; return runWithLockGate(leasePath, deadline, async () => { - const lease = await openStableLockFile(leasePath); + const lease = await openStableNativeLockFile(leasePath); let leased = false; let supervised = false; let completed = false; try { - while (!(leased = tryLock(lease.fd))) await waitForLockTurn(lockPath, deadline); + while (!(leased = tryAcquireNativeFileLock(lease))) { + await waitForLockTurn(lockPath, deadline); + } // The inherited advisory lease follows the legacy child process. A surviving // supervision marker therefore proves that its directory lock is ownerless // once a later process can acquire this lease. @@ -56,7 +60,7 @@ export async function withLegacyFileUpdateLockLease( try { if (supervised && completed) await unlink(supervisionPath).catch(ignoreMissing); } finally { - if (leased) releaseLock(lease); + if (leased) releaseNativeFileLock(lease); await lease.close(); } } @@ -72,11 +76,13 @@ export async function withProcessLifetimeFileUpdateLock( const deadline = Date.now() + timeoutMs; const leasePath = `${targetPath}.lease`; return runWithLockGate(leasePath, deadline, async () => { - const lease = await openStableLockFile(leasePath); + const lease = await openStableNativeLockFile(leasePath); let leased = false; let markerCreated = false; try { - while (!(leased = tryLock(lease.fd))) await waitForLockTurn(lockPath, deadline); + while (!(leased = tryAcquireNativeFileLock(lease))) { + await waitForLockTurn(lockPath, deadline); + } await recoverSupervisedLegacyLock(lockPath, `${targetPath}.supervised`); await acquireLegacyMarker(lockPath, deadline); markerCreated = true; @@ -85,7 +91,7 @@ export async function withProcessLifetimeFileUpdateLock( try { if (markerCreated) await unlink(lockPath).catch(ignoreMissing); } finally { - if (leased) releaseLock(lease); + if (leased) releaseNativeFileLock(lease); await lease.close(); } } @@ -161,22 +167,6 @@ async function waitForGate( } } -async function openStableLockFile(lockPath: string): Promise { - const handle = await open( - lockPath, - fsConstants.O_CREAT | fsConstants.O_RDWR | fsConstants.O_NOFOLLOW, - 0o600, - ); - try { - await assertStableRegularFile(handle, lockPath); - if (process.platform !== 'win32') await handle.chmod(0o600); - return handle; - } catch (error) { - await handle.close(); - throw error; - } -} - async function acquireLegacyMarker(lockPath: string, deadline: number): Promise { for (;;) { try { @@ -210,21 +200,6 @@ async function acquireLegacyMarker(lockPath: string, deadline: number): Promise< } } -async function assertStableRegularFile(handle: FileHandle, lockPath: string): Promise { - const [handleStat, pathStat] = await Promise.all([ - handle.stat({ bigint: true }), - lstat(lockPath, { bigint: true }), - ]); - if ( - !handleStat.isFile() || - !pathStat.isFile() || - handleStat.dev !== pathStat.dev || - handleStat.ino !== pathStat.ino - ) { - throw new Error(`File update lease is not one stable regular file: ${lockPath}`); - } -} - async function waitForLockTurn(lockPath: string, deadline: number): Promise { if (Date.now() >= deadline) throw lockTimeout(lockPath); await new Promise((resolve) => setTimeout(resolve, LOCK_POLL_MS)); @@ -234,14 +209,6 @@ function lockTimeout(lockPath: string): Error { return new Error(`File update is locked by another process (${lockPath})`); } -function releaseLock(handle: FileHandle): void { - try { - unlock(handle.fd); - } catch { - // Closing the OS handle is the authoritative release path. - } -} - function ignoreMissing(error: unknown): void { if (!isNodeError(error, 'ENOENT')) throw error; } diff --git a/packages/storage/src/process-lifetime-owner.ts b/packages/storage/src/process-lifetime-owner.ts new file mode 100644 index 0000000000..be600c0512 --- /dev/null +++ b/packages/storage/src/process-lifetime-owner.ts @@ -0,0 +1,131 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { randomUUID } from 'node:crypto'; +import { chmod, lstat, mkdir, type FileHandle } from 'node:fs/promises'; +import { join } from 'node:path'; +import { + openStableNativeLockFile, + releaseNativeFileLock, + tryAcquireNativeFileLock, + unlinkStableNativeLockFile, +} from './native-file-lock.js'; + +const OWNER_REFERENCE_PREFIX = 'lock-v1:'; +const OWNER_REFERENCE_PATTERN = + /^lock-v1:([0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})$/; + +export interface ProcessLifetimeRecoveryClaim { + retire(): Promise; + close(): Promise; +} + +export interface ProcessLifetimeOwner { + readonly reference: string; + tryClaimReleased(reference: string): Promise; + close(): Promise; +} + +export async function acquireProcessLifetimeOwner(root: string): Promise { + const ownersRoot = join(root, 'owners'); + await mkdir(ownersRoot, { recursive: true, mode: 0o700 }); + const ownersRootStat = await lstat(ownersRoot); + if (!ownersRootStat.isDirectory() || ownersRootStat.isSymbolicLink()) { + throw new Error(`Process lifetime owner root is not a directory: ${ownersRoot}`); + } + if (process.platform !== 'win32') await chmod(ownersRoot, 0o700); + + const reference = `${OWNER_REFERENCE_PREFIX}${randomUUID()}`; + const path = ownerPath(ownersRoot, reference); + const handle = await openStableNativeLockFile(path); + if (!tryAcquireNativeFileLock(handle)) { + await handle.close(); + throw new Error(`New process lifetime owner reference is already active: ${reference}`); + } + return new ProcessLifetimeOwnerImpl(ownersRoot, reference, path, handle); +} + +class ProcessLifetimeOwnerImpl implements ProcessLifetimeOwner { + #closed = false; + + constructor( + private readonly ownersRoot: string, + readonly reference: string, + private readonly path: string, + private readonly handle: FileHandle, + ) {} + + async tryClaimReleased(reference: string): Promise { + if (this.#closed) throw new Error('Process lifetime owner is closed'); + if (reference === this.reference) return undefined; + const path = ownerPath(this.ownersRoot, reference); + const handle = await openStableNativeLockFile(path); + if (!tryAcquireNativeFileLock(handle)) { + await handle.close(); + return undefined; + } + return new ProcessLifetimeRecoveryClaimImpl(path, handle); + } + + async close(): Promise { + if (this.#closed) return; + this.#closed = true; + try { + // Removing the name while the old inode is still locked prevents a gap + // where two recovery claimants could lock two generations of the path. + await unlinkStableNativeLockFile(this.handle, this.path); + } finally { + releaseNativeFileLock(this.handle); + await this.handle.close(); + } + } +} + +class ProcessLifetimeRecoveryClaimImpl implements ProcessLifetimeRecoveryClaim { + #closed = false; + + constructor( + private readonly path: string, + private readonly handle: FileHandle, + ) {} + + async retire(): Promise { + if (this.#closed) return; + this.#closed = true; + try { + await unlinkStableNativeLockFile(this.handle, this.path); + } finally { + releaseNativeFileLock(this.handle); + await this.handle.close(); + } + } + + async close(): Promise { + if (this.#closed) return; + this.#closed = true; + releaseNativeFileLock(this.handle); + await this.handle.close(); + } +} + +function ownerPath(ownersRoot: string, reference: string): string { + const match = OWNER_REFERENCE_PATTERN.exec(reference); + if (!match) throw new Error(`Invalid process lifetime owner reference: ${reference}`); + return join(ownersRoot, `${match[1]}.lease`); +} diff --git a/packages/storage/src/session-copy-cleanup.ts b/packages/storage/src/session-copy-cleanup.ts index 1f98e84df3..4f4b1b6777 100644 --- a/packages/storage/src/session-copy-cleanup.ts +++ b/packages/storage/src/session-copy-cleanup.ts @@ -19,6 +19,10 @@ import { randomUUID } from 'node:crypto'; import { acquireOperationalStateDatabase } from './operational-state-store.js'; +import type { + ProcessLifetimeOwner, + ProcessLifetimeRecoveryClaim, +} from './process-lifetime-owner.js'; export interface SessionCopyCreationLease { sessionId: string; @@ -34,6 +38,7 @@ interface PersistedSessionCopyLease { sessionId: string; trackedAt: number; ownerProcessId?: string; + ownerLifetimeRef?: string; ownerId?: string; phase: 'creating' | 'live' | 'cleanup'; cancelRequested: boolean; @@ -46,6 +51,7 @@ interface SessionCopyCleanupStore { beginCreation( creation: SessionCopyCreationLease, ownerProcessId: string, + ownerLifetimeRef?: string, ): Promise; markLive(sessionId: string): Promise; requestCleanup(sessionId: string): Promise; @@ -75,6 +81,7 @@ export function createSessionCopyCleanupAuthority(input: { resumeSessionCopy?: (creation: Omit) => Promise; processId?: string; isOwnerProcessActive?: (ownerProcessId: string) => boolean | Promise; + processLifetimeOwner?: ProcessLifetimeOwner; }): SessionCopyCleanupAuthority { return new SessionCopyCleanupAuthorityImpl( new SqliteSessionCopyCleanupStore(input.workspaceRoot), @@ -82,6 +89,7 @@ export function createSessionCopyCleanupAuthority(input: { input.resumeSessionCopy, input.processId ?? randomUUID(), input.isOwnerProcessActive, + input.processLifetimeOwner, ); } @@ -104,6 +112,7 @@ class SessionCopyCleanupAuthorityImpl implements SessionCopyCleanupAuthority { private readonly isOwnerProcessActive: | ((ownerProcessId: string) => boolean | Promise) | undefined, + private readonly processLifetimeOwner: ProcessLifetimeOwner | undefined, ) {} ownCreation(creation: SessionCopyCreationLease, operation: () => Promise): Promise { @@ -117,7 +126,11 @@ class SessionCopyCleanupAuthorityImpl implements SessionCopyCleanupAuthority { } const task = (async () => { try { - await this.store.beginCreation(normalized, this.processId); + await this.store.beginCreation( + normalized, + this.processId, + this.processLifetimeOwner?.reference, + ); const result = await operation(); const record = await this.store.markLive(normalized.sessionId); if (record?.cancelRequested) void this.settleCleanup(normalized.sessionId); @@ -161,32 +174,78 @@ class SessionCopyCleanupAuthorityImpl implements SessionCopyCleanupAuthority { async abandonOwner(ownerId: string): Promise { const normalizedOwnerId = normalizeOwnerId(ownerId); - const owned = (await this.store.list()).filter( - (record) => record.ownerProcessId === this.processId && record.ownerId === normalizedOwnerId, - ); + const owned = (await this.store.list()).filter((record) => { + const currentIncarnation = this.processLifetimeOwner + ? record.ownerLifetimeRef === this.processLifetimeOwner.reference + : record.ownerProcessId === this.processId; + return currentIncarnation && record.ownerId === normalizedOwnerId; + }); await Promise.all(owned.map((record) => this.schedule(record.sessionId))); } async recover(): Promise { const removed: string[] = []; const failed: SessionCopyCleanupRecovery['failed'] = []; + const lifetimeGroups = new Map(); for (const record of await this.store.list()) { - const staleOwner = - record.ownerProcessId !== undefined && - record.ownerProcessId !== this.processId && - !(await this.isOwnerProcessActive?.(record.ownerProcessId)); - if (record.phase !== 'cleanup' && !record.cancelRequested && !staleOwner) continue; + if (record.phase === 'cleanup' || record.cancelRequested) { + await this.recoverRecord(record, removed, failed); + continue; + } + if (record.ownerLifetimeRef && this.processLifetimeOwner) { + if (record.ownerLifetimeRef === this.processLifetimeOwner.reference) continue; + const group = lifetimeGroups.get(record.ownerLifetimeRef) ?? []; + group.push(record); + lifetimeGroups.set(record.ownerLifetimeRef, group); + continue; + } try { - await this.store.requestCleanup(record.sessionId); - const disposition = await this.settleCleanup(record.sessionId); - if (disposition === 'removed') removed.push(record.sessionId); + const staleOwner = + record.ownerProcessId !== undefined && + record.ownerProcessId !== this.processId && + !(await this.isOwnerProcessActive?.(record.ownerProcessId)); + if (staleOwner) await this.recoverRecord(record, removed, failed); } catch (error) { failed.push({ sessionId: record.sessionId, error }); } } + for (const [reference, records] of lifetimeGroups) { + let claim: ProcessLifetimeRecoveryClaim | undefined; + try { + claim = await this.processLifetimeOwner?.tryClaimReleased(reference); + } catch (error) { + for (const record of records) failed.push({ sessionId: record.sessionId, error }); + continue; + } + if (!claim) continue; + try { + for (const record of records) await this.recoverRecord(record, removed, failed); + const ownerStillReferenced = await this.store + .list() + .then((current) => current.some((record) => record.ownerLifetimeRef === reference)) + .catch(() => true); + if (!ownerStillReferenced) await claim.retire().catch(() => undefined); + } finally { + await claim.close().catch(() => undefined); + } + } return { removed, failed }; } + private async recoverRecord( + record: PersistedSessionCopyLease, + removed: string[], + failed: SessionCopyCleanupRecovery['failed'], + ): Promise { + try { + await this.store.requestCleanup(record.sessionId); + const disposition = await this.settleCleanup(record.sessionId); + if (disposition === 'removed') removed.push(record.sessionId); + } catch (error) { + failed.push({ sessionId: record.sessionId, error }); + } + } + private settleCleanup(sessionId: string): Promise { const active = this.cleanups.get(sessionId); if (active) return active; @@ -250,6 +309,7 @@ class SqliteSessionCopyCleanupStore implements SessionCopyCleanupStore { async beginCreation( creation: SessionCopyCreationLease, ownerProcessId: string, + ownerLifetimeRef?: string, ): Promise { return this.mutate(creation.sessionId, (current) => { if (current?.creation && !samePersistedCreation(current.creation, creation)) { @@ -263,6 +323,7 @@ class SqliteSessionCopyCleanupStore implements SessionCopyCleanupStore { sessionId: creation.sessionId, trackedAt: current?.trackedAt ?? Date.now(), ownerProcessId, + ...(ownerLifetimeRef ? { ownerLifetimeRef } : {}), ownerId: creation.ownerId, phase: current?.phase ?? 'creating', cancelRequested: false, @@ -392,7 +453,8 @@ function decodeLeaseRow(row: { value.version !== 1 || value.sessionId !== row.sessionId || (value.phase !== 'creating' && value.phase !== 'live' && value.phase !== 'cleanup') || - typeof value.cancelRequested !== 'boolean' + typeof value.cancelRequested !== 'boolean' || + (value.ownerLifetimeRef !== undefined && typeof value.ownerLifetimeRef !== 'string') ) { throw new Error(`Invalid Session copy lease: ${row.sessionId}`); } From bd7ba53a3fafb73da501fe2d4a19a350d3c2ebd5 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Wed, 26 Aug 2026 12:24:34 +0800 Subject: [PATCH 2/2] fix(storage): close session cleanup recovery gaps Generated-by: Codex --- packages/cli/src/runtime-host-tui-context.ts | 4 +- .../__tests__/process-lifetime-owner.test.ts | 26 +++- .../__tests__/session-copy-cleanup.test.ts | 142 ++++++++++++++++++ .../storage/src/process-lifetime-owner.ts | 20 ++- packages/storage/src/session-copy-cleanup.ts | 42 +++++- 5 files changed, 222 insertions(+), 12 deletions(-) diff --git a/packages/cli/src/runtime-host-tui-context.ts b/packages/cli/src/runtime-host-tui-context.ts index 815476bf9a..b2e22909e2 100644 --- a/packages/cli/src/runtime-host-tui-context.ts +++ b/packages/cli/src/runtime-host-tui-context.ts @@ -207,12 +207,12 @@ async function closeRuntimeHostTuiContext( errors.push(error); } try { - await sessionCopyCleanupOwner?.close(); + await closeConnection(); } catch (error) { errors.push(error); } try { - await closeConnection(); + await sessionCopyCleanupOwner?.close(); } catch (error) { errors.push(error); } diff --git a/packages/storage/src/__tests__/process-lifetime-owner.test.ts b/packages/storage/src/__tests__/process-lifetime-owner.test.ts index 7ce56dee8e..9f5353bb5b 100644 --- a/packages/storage/src/__tests__/process-lifetime-owner.test.ts +++ b/packages/storage/src/__tests__/process-lifetime-owner.test.ts @@ -19,7 +19,7 @@ import { strict as assert } from 'node:assert'; import { fork } from 'node:child_process'; -import { mkdtemp, rm } from 'node:fs/promises'; +import { mkdtemp, readdir, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { test, type TestContext } from 'node:test'; @@ -73,6 +73,30 @@ test('releases its owner reference on graceful close', async (t) => { await claim.retire(); }); +test('retires an unreferenced owner file after an unclean process exit', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-process-lifetime-owner-')); + const child = fork( + new URL('./fixtures/process-lifetime-owner-holder.js', import.meta.url), + [root], + { stdio: ['ignore', 'ignore', 'inherit', 'ipc'] }, + ); + t.after(async () => { + if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL'); + await rm(root, { recursive: true, force: true }); + }); + + const deadReference = await waitForReference(t, child); + child.kill('SIGKILL'); + await new Promise((resolve) => child.once('exit', () => resolve())); + + const successor = await acquireProcessLifetimeOwner(root); + t.after(() => successor.close()); + await successor.retireUnreferencedReleasedOwners(new Set()); + const files = await readdir(join(root, 'owners')); + assert.deepEqual(files, [`${successor.reference.slice('lock-v1:'.length)}.lease`]); + assert.equal(files.includes(`${deadReference.slice('lock-v1:'.length)}.lease`), false); +}); + test('rejects owner references outside its versioned namespace', async (t) => { const root = await mkdtemp(join(tmpdir(), 'maka-process-lifetime-owner-')); const owner = await acquireProcessLifetimeOwner(root); diff --git a/packages/storage/src/__tests__/session-copy-cleanup.test.ts b/packages/storage/src/__tests__/session-copy-cleanup.test.ts index fc75ed38d9..3f975d5889 100644 --- a/packages/storage/src/__tests__/session-copy-cleanup.test.ts +++ b/packages/storage/src/__tests__/session-copy-cleanup.test.ts @@ -275,6 +275,122 @@ describe('session copy cleanup authority', () => { assert.deepEqual(await readPendingIds(workspaceRoot), []); }); + it('holds the released owner claim while recovering a cleanup-phase copy', async () => { + const workspaceRoot = await createWorkspace(); + const original = fakeLifetimeOwner('lock-v1:77777777-7777-4777-8777-777777777777'); + const owner = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:707', + processLifetimeOwner: original.owner, + removeSession: async () => {}, + }); + await owner.ownCreation( + { + sessionId: 'fork-cleanup-phase', + kind: 'branch', + sourceSessionId: 'source-session', + sourceTurnId: 'source-turn', + ownerId: 'tui-side', + }, + async () => 'created', + ); + updatePendingLease(workspaceRoot, 'fork-cleanup-phase', (record) => ({ + ...record, + phase: 'cleanup', + cancelRequested: true, + })); + + const successor = fakeLifetimeOwner('lock-v1:88888888-8888-4888-8888-888888888888'); + const authority = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:808', + processLifetimeOwner: successor.owner, + removeSession: async () => { + assert.equal(successor.claimClosed, false); + }, + }); + + assert.deepEqual(await authority.recover(), { + removed: ['fork-cleanup-phase'], + failed: [], + }); + assert.equal(successor.claimAttempts, 1); + assert.equal(successor.claimRetired, true); + }); + + it('falls back to process liveness for an unsupported owner reference version', async () => { + const workspaceRoot = await createWorkspace(); + const original = fakeLifetimeOwner('lock-v1:99999999-9999-4999-8999-999999999999'); + const owner = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:909', + processLifetimeOwner: original.owner, + removeSession: async () => {}, + }); + await owner.ownCreation( + { + sessionId: 'fork-future-owner', + kind: 'branch', + sourceSessionId: 'source-session', + sourceTurnId: 'source-turn', + ownerId: 'tui-side', + }, + async () => 'created', + ); + updatePendingLease(workspaceRoot, 'fork-future-owner', (record) => ({ + ...record, + ownerLifetimeRef: 'lock-v2:future-owner', + })); + + let claimAttempts = 0; + const removed: string[] = []; + const authority = createSessionCopyCleanupAuthority({ + workspaceRoot, + processId: 'tui:1001', + processLifetimeOwner: { + reference: 'lock-v1:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', + tryClaimReleased: async () => { + claimAttempts += 1; + throw new Error('unsupported owner reference must use the PID fallback'); + }, + retireUnreferencedReleasedOwners: async () => {}, + close: async () => {}, + }, + isOwnerProcessActive: () => false, + removeSession: async (sessionId) => { + removed.push(sessionId); + }, + }); + + assert.deepEqual(await authority.recover(), { + removed: ['fork-future-owner'], + failed: [], + }); + assert.equal(claimAttempts, 0); + assert.deepEqual(removed, ['fork-future-owner']); + }); + + it('asks the process owner to retire released files with no database references', async () => { + const workspaceRoot = await createWorkspace(); + let referenced: ReadonlySet | undefined; + const authority = createSessionCopyCleanupAuthority({ + workspaceRoot, + processLifetimeOwner: { + reference: 'lock-v1:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + tryClaimReleased: async () => undefined, + retireUnreferencedReleasedOwners: async (current) => { + referenced = current; + }, + close: async () => {}, + }, + removeSession: async () => {}, + }); + + assert.deepEqual(await authority.recover(), { removed: [], failed: [] }); + assert.ok(referenced); + assert.deepEqual([...referenced], []); + }); + it('fails closed when process-incarnation ownership cannot be inspected', async () => { const workspaceRoot = await createWorkspace(); const original = fakeLifetimeOwner('lock-v1:33333333-3333-4333-8333-333333333333'); @@ -304,6 +420,7 @@ describe('session copy cleanup authority', () => { tryClaimReleased: async () => { throw failure; }, + retireUnreferencedReleasedOwners: async () => {}, close: async () => {}, }, removeSession: async () => { @@ -510,6 +627,30 @@ async function readPendingIds(workspaceRoot: string): Promise { } } +function updatePendingLease( + workspaceRoot: string, + sessionId: string, + update: (record: Record) => Record, +): void { + const database = new DatabaseSync(join(workspaceRoot, 'runtime.sqlite')); + try { + const row = database + .prepare( + 'SELECT record_json AS recordJson FROM workflow_quote_companion_cleanup WHERE session_id = ?', + ) + .get(sessionId) as { recordJson: string } | undefined; + assert.ok(row); + database + .prepare('UPDATE workflow_quote_companion_cleanup SET record_json = ? WHERE session_id = ?') + .run( + JSON.stringify(update(JSON.parse(row.recordJson) as Record)), + sessionId, + ); + } finally { + database.close(); + } +} + function fakeLifetimeOwner(reference: string): { owner: ProcessLifetimeOwner; readonly claimAttempts: number; @@ -535,6 +676,7 @@ function fakeLifetimeOwner(reference: string): { claimAttempts += 1; return claim; }, + retireUnreferencedReleasedOwners: async () => {}, close: async () => {}, }, get claimAttempts() { diff --git a/packages/storage/src/process-lifetime-owner.ts b/packages/storage/src/process-lifetime-owner.ts index be600c0512..1d2021212e 100644 --- a/packages/storage/src/process-lifetime-owner.ts +++ b/packages/storage/src/process-lifetime-owner.ts @@ -18,7 +18,7 @@ */ import { randomUUID } from 'node:crypto'; -import { chmod, lstat, mkdir, type FileHandle } from 'node:fs/promises'; +import { chmod, lstat, mkdir, readdir, type FileHandle } from 'node:fs/promises'; import { join } from 'node:path'; import { openStableNativeLockFile, @@ -30,6 +30,8 @@ import { const OWNER_REFERENCE_PREFIX = 'lock-v1:'; const OWNER_REFERENCE_PATTERN = /^lock-v1:([0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})$/; +const OWNER_FILE_PATTERN = + /^([0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})\.lease$/; export interface ProcessLifetimeRecoveryClaim { retire(): Promise; @@ -39,6 +41,7 @@ export interface ProcessLifetimeRecoveryClaim { export interface ProcessLifetimeOwner { readonly reference: string; tryClaimReleased(reference: string): Promise; + retireUnreferencedReleasedOwners(referenced: ReadonlySet): Promise; close(): Promise; } @@ -61,6 +64,10 @@ export async function acquireProcessLifetimeOwner(root: string): Promise): Promise { + for (const entry of await readdir(this.ownersRoot, { withFileTypes: true })) { + const match = OWNER_FILE_PATTERN.exec(entry.name); + if (!match) continue; + const reference = `${OWNER_REFERENCE_PREFIX}${match[1]}`; + if (referenced.has(reference)) continue; + const claim = await this.tryClaimReleased(reference); + if (claim) await claim.retire(); + } + } + async close(): Promise { if (this.#closed) return; this.#closed = true; diff --git a/packages/storage/src/session-copy-cleanup.ts b/packages/storage/src/session-copy-cleanup.ts index 4f4b1b6777..46f53d39b3 100644 --- a/packages/storage/src/session-copy-cleanup.ts +++ b/packages/storage/src/session-copy-cleanup.ts @@ -19,9 +19,10 @@ import { randomUUID } from 'node:crypto'; import { acquireOperationalStateDatabase } from './operational-state-store.js'; -import type { - ProcessLifetimeOwner, - ProcessLifetimeRecoveryClaim, +import { + isProcessLifetimeOwnerReference, + type ProcessLifetimeOwner, + type ProcessLifetimeRecoveryClaim, } from './process-lifetime-owner.js'; export interface SessionCopyCreationLease { @@ -188,17 +189,24 @@ class SessionCopyCleanupAuthorityImpl implements SessionCopyCleanupAuthority { const failed: SessionCopyCleanupRecovery['failed'] = []; const lifetimeGroups = new Map(); for (const record of await this.store.list()) { - if (record.phase === 'cleanup' || record.cancelRequested) { - await this.recoverRecord(record, removed, failed); - continue; - } - if (record.ownerLifetimeRef && this.processLifetimeOwner) { + if ( + record.ownerLifetimeRef && + this.processLifetimeOwner && + isProcessLifetimeOwnerReference(record.ownerLifetimeRef) + ) { if (record.ownerLifetimeRef === this.processLifetimeOwner.reference) continue; const group = lifetimeGroups.get(record.ownerLifetimeRef) ?? []; group.push(record); lifetimeGroups.set(record.ownerLifetimeRef, group); continue; } + if ( + record.ownerLifetimeRef === undefined && + (record.phase === 'cleanup' || record.cancelRequested) + ) { + await this.recoverRecord(record, removed, failed); + continue; + } try { const staleOwner = record.ownerProcessId !== undefined && @@ -229,6 +237,24 @@ class SessionCopyCleanupAuthorityImpl implements SessionCopyCleanupAuthority { await claim.close().catch(() => undefined); } } + if (this.processLifetimeOwner) { + const referenced = await this.store + .list() + .then( + (records) => + new Set( + records.flatMap((record) => + record.ownerLifetimeRef ? [record.ownerLifetimeRef] : [], + ), + ), + ) + .catch(() => undefined); + if (referenced) { + await this.processLifetimeOwner + .retireUnreferencedReleasedOwners(referenced) + .catch(() => undefined); + } + } return { removed, failed }; }