From 0a71f60efd1c9f7a28ee1767698b611866633e58 Mon Sep 17 00:00:00 2001 From: 99Gaoxiaoqi <72881245+99Gaoxiaoqi@users.noreply.github.com> Date: Mon, 7 Sep 2026 21:07:40 +0800 Subject: [PATCH] fix(storage): batch retired Session artifact cleanup Resolve Artifact alias guards once per retirement batch and commit successful Session deletions together. Preserve independent unlink and directory-sync failures, cross-Session alias protection, and durable cleanup retry obligations. Fixes #4038 Generated-by: Codex --- .../session-retirement-coordinator.test.ts | 239 ++++++++++- .../server/session-retirement-coordinator.ts | 22 +- .../artifact-session-purge-batch.test.ts | 386 ++++++++++++++++++ .../fixtures/artifact-batch-purge-crash.ts | 49 +++ packages/storage/src/artifact-store.ts | 98 ++++- packages/storage/src/artifact-stores.ts | 5 + 6 files changed, 781 insertions(+), 18 deletions(-) create mode 100644 packages/storage/src/__tests__/artifact-session-purge-batch.test.ts create mode 100644 packages/storage/src/__tests__/fixtures/artifact-batch-purge-crash.ts diff --git a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts index ed61c44fa2..e571c13843 100644 --- a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts @@ -47,7 +47,7 @@ import { SessionAdmissionGate } from '../server/session-admission-gate.js'; import { MemoryExtractionSessionLane } from '../server/memory-extraction-session-lane.js'; import { HostSessionRetirementCoordinator } from '../server/session-retirement-coordinator.js'; import { purgeSessionSidecars } from '../server/session-sidecar-purge.js'; -import { waitFor as pollFor } from '@maka/core/test-only/async-primitives'; +import { deferred, waitFor as pollFor } from '@maka/core/test-only/async-primitives'; const CONNECTION_CONTEXT: ConnectionContext = { hostEpoch: 'retirement-test', @@ -170,9 +170,14 @@ describe('Host Session retirement coordinator', () => { test('retires only its own context refs without draining global garbage', async () => { const contextActions: string[] = []; let garbageBatches = 0; + const artifactSessions: string[] = []; await purgeSessionSidecars( { - artifacts: { purgeSessionArtifacts: async () => {} }, + artifacts: { + purgeSessionArtifacts: async (sessionId) => { + artifactSessions.push(sessionId); + }, + }, sessionTodo: { purgeSessionState: async () => {} }, contextOffload: { retireSession: async (sessionId) => { @@ -192,6 +197,7 @@ describe('Host Session retirement coordinator', () => { 'session-context', ); + assert.deepEqual(artifactSessions, ['session-context']); assert.deepEqual(contextActions, ['retire:session-context']); assert.equal(garbageBatches, 0); }); @@ -937,6 +943,148 @@ describe('Host Session retirement coordinator', () => { }); }); + test('batches real artifact cleanup and preserves independent pending work and archived child files', async (t) => { + await withHarness(async (harness) => { + const retainedChild = await createClosedSubagent(harness, harness.rootId, 0); + const successfulSession = await createClosedGraphOperator(harness, harness.rootId, 'a'); + const owner = await tryAcquireInteractiveRootOwner( + await resolveStorageRoot({ path: harness.workspaceRoot, kind: 'interactive' }), + ); + assert.ok(owner); + const artifacts = await openInteractiveArtifactStoreForWrite(owner.lease); + let statPatch: ReturnType | undefined; + let removePatch: ReturnType | undefined; + try { + const records = await Promise.all( + [...harness.familyIds, successfulSession, retainedChild].map((sessionId) => + artifacts.create({ + sessionId, + turnId: 'turn-1', + name: 'retirement.txt', + kind: 'file', + content: `artifact for ${sessionId}`, + source: 'tool_result', + }), + ), + ); + const retained = records.find((record) => record.sessionId === retainedChild)!; + const failed = records.find((record) => record.sessionId === harness.rootId)!; + const artifactPath = (relativePath: string) => + join(owner.lease.canonicalPath, 'artifacts', relativePath); + const originalLstat = fsPromises.lstat; + let retainedStats = 0; + statPatch = t.mock.method( + fsPromises, + 'lstat', + async (...args: Parameters) => { + if (args[0] === artifactPath(retained.relativePath)) retainedStats += 1; + return originalLstat(...args); + }, + ); + const originalRm = fsPromises.rm; + let failArtifact = true; + removePatch = t.mock.method( + fsPromises, + 'rm', + async (...args: Parameters) => { + if (failArtifact && args[0] === artifactPath(failed.relativePath)) { + throw new Error('injected artifact unlink failure'); + } + return originalRm(...args); + }, + ); + syncBuiltinESMExports(); + harness.purgeArtifactBatch = (sessionIds) => + artifacts.purgeSessionArtifactsBatch(sessionIds); + harness.purgeTodo = async (sessionId) => { + if (sessionId === harness.revisionId) throw new Error('injected todo failure'); + }; + const target = await harness.store.readHeaderRecordSnapshot(harness.rootId); + const removed = await harness.coordinator.handlers['session.remove']( + { sessionId: harness.rootId, expectedRevision: target.revision }, + CONNECTION_CONTEXT, + ); + assert.equal(removed.ok, true); + await waitFor( + async () => + !(await harness.store.listPendingSessionRetirementCleanupIds()).includes( + successfulSession, + ), + 'successful session should complete despite other artifact and todo failures', + ); + assert.equal(harness.actions.artifactBatches.length, 1); + assert.deepEqual( + new Set(harness.actions.artifactBatches[0]), + new Set([...harness.familyIds, successfulSession]), + ); + assert.equal(retainedStats, 1, 'one guard-file resolution for the entire Host batch'); + assert.deepEqual( + new Set(await harness.store.listPendingSessionRetirementCleanupIds()), + new Set(harness.familyIds), + ); + assert.equal((await artifacts.listPage(harness.rootId, { offset: 0, limit: 10 })).total, 1); + assert.equal( + (await artifacts.listPage(harness.revisionId, { offset: 0, limit: 10 })).total, + 0, + ); + assert.equal( + (await artifacts.listPage(successfulSession, { offset: 0, limit: 10 })).total, + 0, + ); + assert.equal((await harness.store.readHeaderSnapshot(retainedChild)).isArchived, true); + assert.deepEqual(await artifacts.readTextInSession(retainedChild, retained.id), { + ok: true, + text: `artifact for ${retainedChild}`, + }); + + failArtifact = false; + harness.purgeTodo = undefined; + await harness.coordinator.recover(); + await harness.coordinator.close(); + assert.equal(harness.actions.artifactBatches.length, 2); + assert.deepEqual(new Set(harness.actions.artifactBatches[1]), new Set(harness.familyIds)); + assert.deepEqual(await harness.store.listPendingSessionRetirementCleanupIds(), []); + assert.equal((await artifacts.listPage(harness.rootId, { offset: 0, limit: 10 })).total, 0); + assert.deepEqual(await artifacts.readTextInSession(retainedChild, retained.id), { + ok: true, + text: `artifact for ${retainedChild}`, + }); + } finally { + await harness.coordinator.close(); + statPatch?.mock.restore(); + removePatch?.mock.restore(); + syncBuiltinESMExports(); + artifacts.close(); + await owner.close(); + } + }); + }); + + for (const failure of ['missing result', 'batch rejection'] as const) { + test(`keeps cleanup pending on artifact ${failure} while still purging other sidecars`, async () => { + await withHarness(async (harness) => { + harness.purgeArtifactBatch = async () => { + if (failure === 'batch rejection') throw new Error('injected metadata failure'); + return new Map([[harness.rootId, { status: 'fulfilled', value: undefined }]]); + }; + const target = await harness.store.readHeaderRecordSnapshot(harness.rootId); + const removed = await harness.coordinator.handlers['session.remove']( + { sessionId: harness.rootId, expectedRevision: target.revision }, + CONNECTION_CONTEXT, + ); + assert.equal(removed.ok, true); + await harness.coordinator.close(); + assert.equal(harness.actions.artifactBatches.length, 1); + assert.deepEqual(new Set(harness.actions.purgedTasks), new Set(harness.familyIds)); + assert.deepEqual(new Set(harness.actions.purgedAgentGraphs), new Set(harness.familyIds)); + assert.deepEqual( + new Set(await harness.store.listPendingSessionRetirementCleanupIds()), + new Set(failure === 'missing result' ? [harness.revisionId] : harness.familyIds), + ); + }); + }); + } + test('keeps aggregate cleanup retryable without changing a committed remove result', async () => { await withHarness(async (harness) => { harness.failArtifactCleanup = true; @@ -1053,6 +1201,69 @@ describe('Host Session retirement coordinator', () => { }); }); + test('coalesces duplicate recovery and new removals into the next batch before close', async () => { + await withHarness(async (harness) => { + const entered = deferred(); + const release = deferred(); + const next = await harness.store.create(sessionInput('Next batch')); + let first = true; + harness.purgeArtifactBatch = async (sessionIds) => { + if (first) { + first = false; + entered.resolve(); + await release.promise; + } + return new Map( + sessionIds.map((sessionId) => [sessionId, { status: 'fulfilled', value: undefined }]), + ); + }; + try { + const target = await harness.store.readHeaderRecordSnapshot(harness.rootId); + assert.equal( + ( + await harness.coordinator.handlers['session.remove']( + { sessionId: harness.rootId, expectedRevision: target.revision }, + CONNECTION_CONTEXT, + ) + ).ok, + true, + ); + await entered.promise; + await harness.coordinator.recover(); + await harness.coordinator.recover(); + const nextTarget = await harness.store.readHeaderRecordSnapshot(next.id); + assert.equal( + ( + await harness.coordinator.handlers['session.remove']( + { sessionId: next.id, expectedRevision: nextTarget.revision }, + CONNECTION_CONTEXT, + ) + ).ok, + true, + ); + assert.equal(harness.actions.artifactBatches.length, 1); + let closed = false; + const closing = harness.coordinator.close().then(() => { + closed = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(closed, false); + release.resolve(); + await closing; + assert.equal(harness.actions.artifactBatches.length, 2); + assert.deepEqual(new Set(harness.actions.artifactBatches[0]), new Set(harness.familyIds)); + assert.deepEqual( + new Set(harness.actions.artifactBatches[1]), + new Set([...harness.familyIds, next.id]), + ); + assert.equal(harness.actions.artifactBatches[1]!.length, harness.familyIds.length + 1); + assert.deepEqual(await harness.store.listPendingSessionRetirementCleanupIds(), []); + } finally { + release.resolve(); + } + }); + }); + test('projects a sibling metadata race as a family operation conflict', async () => { await withHarness(async (harness) => { await harness.store.updateHeader(harness.revisionId, { @@ -1172,6 +1383,7 @@ interface RetirementActions { readonly retiredCapabilities: string[]; readonly retiredMessages: string[]; readonly purgedArtifacts: string[]; + readonly artifactBatches: string[][]; readonly retiredContext: string[]; readonly purgedTasks: string[]; readonly purgedOperationalState: string[]; @@ -1211,6 +1423,7 @@ async function withHarness( retiredCapabilities: [], retiredMessages: [], purgedArtifacts: [], + artifactBatches: [], retiredContext: [], purgedTasks: [], purgedOperationalState: [], @@ -1252,6 +1465,8 @@ async function withHarness( failRemovalPublication: false, failArtifactCleanup: false, purgeArtifact: undefined, + purgeArtifactBatch: undefined, + purgeTodo: undefined, hideRevisionFromNextFamilyRead: false, updateMetadataDuringNextDispose: false, updateSiblingBeforeRemoveCommit: false, @@ -1366,14 +1581,22 @@ async function withHarness( }, }, artifacts: { - purgeSessionArtifacts: async (sessionId) => { - if (harness.purgeArtifact) return harness.purgeArtifact(sessionId); - if (harness.failArtifactCleanup) throw new Error('injected Artifact cleanup failure'); - actions.purgedArtifacts.push(sessionId); + purgeSessionArtifactsBatch: async (sessionIds) => { + actions.artifactBatches.push([...sessionIds]); + if (harness.purgeArtifactBatch) return harness.purgeArtifactBatch(sessionIds); + const outcomes = await Promise.allSettled( + sessionIds.map(async (sessionId) => { + if (harness.purgeArtifact) return harness.purgeArtifact(sessionId); + if (harness.failArtifactCleanup) throw new Error('injected Artifact cleanup failure'); + actions.purgedArtifacts.push(sessionId); + }), + ); + return new Map(sessionIds.map((sessionId, index) => [sessionId, outcomes[index]!])); }, }, sessionTodo: { purgeSessionState: async (sessionId) => { + if (harness.purgeTodo) await harness.purgeTodo(sessionId); actions.purgedTasks.push(sessionId); }, }, @@ -1436,6 +1659,10 @@ interface RetirementHarness { failRemovalPublication: boolean; failArtifactCleanup: boolean; purgeArtifact: ((sessionId: string) => Promise) | undefined; + purgeArtifactBatch: + | ((sessionIds: readonly string[]) => Promise>>) + | undefined; + purgeTodo: ((sessionId: string) => Promise) | undefined; hideRevisionFromNextFamilyRead: boolean; updateMetadataDuringNextDispose: boolean; updateSiblingBeforeRemoveCommit: boolean; diff --git a/packages/runtime-host/src/server/session-retirement-coordinator.ts b/packages/runtime-host/src/server/session-retirement-coordinator.ts index 52b672de96..652f0f3f29 100644 --- a/packages/runtime-host/src/server/session-retirement-coordinator.ts +++ b/packages/runtime-host/src/server/session-retirement-coordinator.ts @@ -121,7 +121,7 @@ export interface HostSessionRetirementCoordinatorOptions { readonly manager: RetirementManager; readonly capabilities: RetirementCapabilities; readonly continuity: RetirementContinuity; - readonly artifacts: Pick; + readonly artifacts: Pick; readonly sessionTodo: Pick; readonly contextOffload?: Pick; readonly purgeOperationalState: (sessionId: string) => Promise; @@ -714,16 +714,30 @@ export class HostSessionRetirementCoordinator { while (this.#cleanupQueue.size > 0) { const batch = [...this.#cleanupQueue]; this.#cleanupQueue.clear(); - await Promise.allSettled(batch.map((sessionId) => this.#cleanupRetiredSession(sessionId))); + const artifactCleanup = this.#artifacts.purgeSessionArtifactsBatch(batch); + await Promise.allSettled( + batch.map((sessionId) => this.#cleanupRetiredSession(sessionId, artifactCleanup)), + ); } } - async #cleanupRetiredSession(sessionId: string): Promise { + async #cleanupRetiredSession( + sessionId: string, + artifactCleanup: Promise>>, + ): Promise { const worktree = this.#retiredWorktrees.get(sessionId); const outcomes = await Promise.allSettled([ purgeSessionSidecars( { - artifacts: this.#artifacts, + artifacts: { + purgeSessionArtifacts: async () => { + const outcome = (await artifactCleanup).get(sessionId); + if (!outcome) { + throw new Error(`Artifact cleanup returned no result for Session ${sessionId}`); + } + if (outcome.status === 'rejected') throw outcome.reason; + }, + }, sessionTodo: this.#sessionTodo, ...(this.#contextOffload ? { contextOffload: this.#contextOffload } : {}), purgeOperationalState: this.#purgeOperationalState, diff --git a/packages/storage/src/__tests__/artifact-session-purge-batch.test.ts b/packages/storage/src/__tests__/artifact-session-purge-batch.test.ts new file mode 100644 index 0000000000..1ac3eb6c3b --- /dev/null +++ b/packages/storage/src/__tests__/artifact-session-purge-batch.test.ts @@ -0,0 +1,386 @@ +/* + * 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 assert from 'node:assert/strict'; +import { fork } from 'node:child_process'; +import { once } from 'node:events'; +import fs from 'node:fs/promises'; +import { syncBuiltinESMExports } from 'node:module'; +import { tmpdir } from 'node:os'; +import { dirname, join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; +import { test, type TestContext } from 'node:test'; +import { + createSqliteArtifactStoreWriteAuthority, + type ArtifactAuthorityStore, +} from '../artifact-store.js'; + +async function withStore(run: (store: ArtifactAuthorityStore, root: string) => Promise) { + const root = await fs.realpath(await fs.mkdtemp(join(tmpdir(), 'maka-artifact-batch-'))); + const authority = createSqliteArtifactStoreWriteAuthority(root); + try { + await run(authority.store, root); + } finally { + authority.close(); + await fs.rm(root, { recursive: true, force: true }); + } +} + +function create(store: ArtifactAuthorityStore, sessionId: string, id: string, name = `${id}.txt`) { + return store.create({ + id, + sessionId, + turnId: 'turn-1', + name, + kind: 'file', + source: 'tool_result', + content: `payload ${id}`, + now: 1, + }); +} + +function fulfilled(results: ReadonlyMap>, sessionId: string) { + assert.deepEqual(results.get(sessionId), { status: 'fulfilled', value: undefined }); +} + +function rejected(results: ReadonlyMap>, sessionId: string) { + const result = results.get(sessionId); + assert.equal(result?.status, 'rejected'); + if (result?.status !== 'rejected') throw new Error('Expected rejection'); + return result.reason as Error; +} + +test('batch purge reads metadata and resolves retained payloads once, committing only target ids', async (t) => { + await withStore(async (store, root) => { + const retained = await Promise.all( + Array.from({ length: 16 }, (_, i) => create(store, 'keep', `keep-${i}`)), + ); + const targets = await Promise.all( + Array.from({ length: 8 }, (_, i) => create(store, `retire-${i}`, `target-${i}`)), + ); + const retainedPaths = new Set(retained.map((r) => join(root, 'artifacts', r.relativePath))); + const scans = new Map(); + let reads = 0; + let writes = 0; + const originalLstat = fs.lstat; + const originalPrepare = DatabaseSync.prototype.prepare; + const lstat = t.mock.method(fs, 'lstat', async (...args: Parameters) => { + const path = String(args[0]); + if (retainedPaths.has(path)) scans.set(path, (scans.get(path) ?? 0) + 1); + return originalLstat(...args); + }); + const prepare = t.mock.method( + DatabaseSync.prototype, + 'prepare', + function (this: DatabaseSync, sql: string) { + if (/SELECT record_json\s+FROM artifact_records/.test(sql)) reads++; + if (sql === 'DELETE FROM artifact_records WHERE artifact_id = ?') writes++; + return originalPrepare.call(this, sql); + }, + ); + syncBuiltinESMExports(); + try { + const ids = targets.map((r) => r.sessionId); + const results = await store.purgeSessionArtifactsBatch([...ids, ids[0]!, 'empty']); + assert.equal(results.size, ids.length + 1); + for (const id of [...ids, 'empty']) fulfilled(results, id); + assert.equal(scans.size, retainedPaths.size); + assert.ok([...scans.values()].every((n) => n === 1)); + assert.equal(reads, 1); + assert.equal(writes, 1); + const replay = await store.purgeSessionArtifactsBatch(ids); + for (const id of ids) fulfilled(replay, id); + assert.ok([...scans.values()].every((n) => n === 1)); + assert.equal(writes, 1); + assert.equal(reads, 2); + await store.purgeSessionArtifactsBatch([]); + assert.equal(reads, 2); + } finally { + lstat.mock.restore(); + prepare.mock.restore(); + syncBuiltinESMExports(); + } + for (const record of retained) + assert.equal( + await fs.readFile(join(root, 'artifacts', record.relativePath), 'utf8'), + `payload ${record.id}`, + ); + for (const record of targets) + await assert.rejects(fs.stat(join(root, 'artifacts', record.relativePath)), { + code: 'ENOENT', + }); + assert.equal((await store.listPage('keep', { offset: 0, limit: 100 })).total, 16); + }); +}); + +test('a target unlink failure retains only that Session and retries after reopen', async (t) => { + await withStore(async (store, root) => { + const first = await create(store, 'fail', 'first'); + const blocked = await create(store, 'fail', 'second'); + const good = await create(store, 'good', 'good'); + const target = join(root, 'artifacts', blocked.relativePath); + const originalRm = fs.rm; + const mock = t.mock.method(fs, 'rm', async (...args: Parameters) => { + if (args[0] === target) + throw Object.assign(new Error('injected unlink failure'), { code: 'EIO' }); + return originalRm(...args); + }); + syncBuiltinESMExports(); + try { + const results = await store.purgeSessionArtifactsBatch(['fail', 'good']); + assert.match(rejected(results, 'fail').message, /injected unlink failure/); + fulfilled(results, 'good'); + } finally { + mock.mock.restore(); + syncBuiltinESMExports(); + } + assert.equal((await store.listPage('fail', { offset: 0, limit: 10 })).total, 2); + assert.equal((await store.listPage('good', { offset: 0, limit: 10 })).total, 0); + await assert.rejects(fs.stat(join(root, 'artifacts', first.relativePath)), { code: 'ENOENT' }); + await assert.rejects(fs.stat(join(root, 'artifacts', good.relativePath)), { code: 'ENOENT' }); + store.close(); + const reopened = createSqliteArtifactStoreWriteAuthority(root); + try { + const results = await reopened.store.purgeSessionArtifactsBatch(['fail', 'good']); + fulfilled(results, 'fail'); + fulfilled(results, 'good'); + assert.equal((await reopened.store.listPage('fail', { offset: 0, limit: 10 })).total, 0); + } finally { + reopened.close(); + } + }); +}); + +test('directory sync failure keeps its Session retryable without blocking a successful sibling', async (t) => { + if (process.platform === 'win32') return t.skip('directory fsync is a POSIX durability barrier'); + await withStore(async (store, root) => { + const failed = await create(store, 'fail', 'fail'); + await create(store, 'good', 'good'); + const targetDirectory = dirname(join(root, 'artifacts', failed.relativePath)); + const originalOpen = fs.open; + const mock = t.mock.method(fs, 'open', async (...args: Parameters) => { + if (args[0] === targetDirectory) + throw Object.assign(new Error('injected directory sync failure'), { code: 'EIO' }); + return originalOpen(...args); + }); + syncBuiltinESMExports(); + try { + for (let i = 0; i < 2; i++) { + const results = await store.purgeSessionArtifactsBatch(['fail', 'good']); + assert.match(rejected(results, 'fail').message, /injected directory sync failure/); + fulfilled(results, 'good'); + assert.equal((await store.listPage('fail', { offset: 0, limit: 10 })).total, 1); + } + } finally { + mock.mock.restore(); + syncBuiltinESMExports(); + } + store.close(); + const reopened = createSqliteArtifactStoreWriteAuthority(root); + try { + fulfilled(await reopened.store.purgeSessionArtifactsBatch(['fail']), 'fail'); + assert.equal((await reopened.store.listPage('fail', { offset: 0, limit: 10 })).total, 0); + } finally { + reopened.close(); + } + }); +}); + +test('a failed metadata transaction rolls back the complete batch and can recover missing payloads', async () => { + await withStore(async (store, root) => { + const a = await create(store, 'a', 'a'); + const b = await create(store, 'b', 'b'); + const database = new DatabaseSync(join(root, 'runtime.sqlite')); + try { + database.exec(`CREATE TRIGGER fail_batch_delete BEFORE DELETE ON artifact_records + WHEN OLD.artifact_id = 'b' BEGIN SELECT RAISE(ABORT, 'injected metadata failure'); END`); + await assert.rejects( + store.purgeSessionArtifactsBatch(['a', 'b']), + /injected metadata failure/, + ); + for (const r of [a, b]) { + assert.equal((await store.listPage(r.sessionId, { offset: 0, limit: 10 })).total, 1); + await assert.rejects(fs.stat(join(root, 'artifacts', r.relativePath)), { code: 'ENOENT' }); + } + database.exec('DROP TRIGGER fail_batch_delete'); + } finally { + database.close(); + } + store.close(); + const reopened = createSqliteArtifactStoreWriteAuthority(root); + try { + const results = await reopened.store.purgeSessionArtifactsBatch(['a', 'b']); + fulfilled(results, 'a'); + fulfilled(results, 'b'); + for (const id of ['a', 'b']) + assert.equal((await reopened.store.listPage(id, { offset: 0, limit: 10 })).total, 0); + } finally { + reopened.close(); + } + }); +}); + +test('batch admission snapshots caller ids and rejects invalid ids before deleting anything', async () => { + await withStore(async (store, root) => { + const a = await create(store, 'a', 'a'); + const b = await create(store, 'b', 'b'); + await assert.rejects(store.purgeSessionArtifactsBatch(['a', '../bad'])); + assert.equal(await fs.readFile(join(root, 'artifacts', a.relativePath), 'utf8'), 'payload a'); + const ids = ['a']; + const operation = store.purgeSessionArtifactsBatch(ids); + ids[0] = 'b'; + fulfilled(await operation, 'a'); + assert.equal(await fs.readFile(join(root, 'artifacts', b.relativePath), 'utf8'), 'payload b'); + }); +}); + +async function symlinkOrSkip(t: TestContext, target: string, path: string) { + try { + await fs.symlink(target, path); + return true; + } catch (error) { + if (process.platform === 'win32' && (error as NodeJS.ErrnoException).code === 'EPERM') { + t.skip('Windows symlinks require Developer Mode or elevation'); + return false; + } + throw error; + } +} + +test('batch preserves cross-Session alias guards even when both aliases are targets', async (t) => { + await withStore(async (store, root) => { + const a = await create(store, 'a', 'shared', 'b-file.txt'); + const b = await create(store, 'b', 'shared-b', 'file.txt'); + await create(store, 'good', 'good'); + const aPath = join(root, 'artifacts', a.relativePath); + const bPath = join(root, 'artifacts', b.relativePath); + // Both canonical records have the same basename; aliasing the parent + // directories must not let either Session delete the other's payload. + await fs.rm(bPath); + await fs.rmdir(dirname(bPath)); + if (!(await symlinkOrSkip(t, dirname(aPath), dirname(bPath)))) return; + const results = await store.purgeSessionArtifactsBatch(['a', 'b', 'good']); + assert.match(rejected(results, 'a').message, /path is still referenced/); + assert.match(rejected(results, 'b').message, /path is still referenced/); + fulfilled(results, 'good'); + assert.equal(await fs.readFile(aPath, 'utf8'), `payload ${a.id}`); + for (const id of ['a', 'b']) + assert.equal((await store.listPage(id, { offset: 0, limit: 10 })).total, 1); + }); +}); + +test('batch joins all path resolutions before rejecting an unresolved guard and releases its writer queue', async (t) => { + await withStore(async (store, root) => { + const kept = await create(store, 'keep', 'keep'); + const target = await create(store, 'target', 'target'); + const keptPath = join(root, 'artifacts', kept.relativePath); + const originalLstat = fs.lstat; + let active = 0; + const mock = t.mock.method(fs, 'lstat', async (...args: Parameters) => { + active++; + try { + if (args[0] === keptPath) + throw Object.assign(new Error('injected guard resolution failure'), { code: 'EIO' }); + return await originalLstat(...args); + } finally { + active--; + } + }); + syncBuiltinESMExports(); + try { + await assert.rejects( + store.purgeSessionArtifactsBatch(['target']), + /injected guard resolution failure/, + ); + assert.equal(active, 0); + } finally { + mock.mock.restore(); + syncBuiltinESMExports(); + } + assert.equal( + await fs.readFile(join(root, 'artifacts', target.relativePath), 'utf8'), + 'payload target', + ); + fulfilled(await store.purgeSessionArtifactsBatch(['target']), 'target'); + }); +}); + +for (const phase of ['after-unlink', 'after-metadata']) { + test(`batch purge recovers after a process is killed ${phase}`, { timeout: 15_000 }, async () => { + await withStore(async (store, root) => { + const first = await create(store, 'a', 'first'); + const second = await create(store, 'a', 'second'); + const sibling = await create(store, 'b', 'sibling'); + const kept = await create(store, 'keep', 'kept'); + store.close(); + const child = fork( + new URL('./fixtures/artifact-batch-purge-crash.js', import.meta.url), + [root, phase, join(root, 'artifacts', first.relativePath)], + { + stdio: ['ignore', 'ignore', 'pipe', 'ipc'], + }, + ); + let stderr = ''; + child.stderr?.on('data', (chunk) => { + stderr += chunk; + }); + const exit = once(child, 'exit'); + try { + const message = await Promise.race([ + once(child, 'message').then(([value]) => value), + exit.then(([code, signal]) => { + throw new Error(`Child exited before crash point (${code}, ${signal}): ${stderr}`); + }), + ]); + assert.equal(message, phase); + child.kill('SIGKILL'); + const [code, signal] = await exit; + assert.ok(signal === 'SIGKILL' || (process.platform === 'win32' && code !== 0)); + const reopened = createSqliteArtifactStoreWriteAuthority(root); + try { + assert.equal( + (await reopened.store.listPage('a', { offset: 0, limit: 10 })).total, + phase === 'after-unlink' ? 2 : 0, + ); + await assert.rejects(fs.stat(join(root, 'artifacts', first.relativePath)), { + code: 'ENOENT', + }); + const result = await reopened.store.purgeSessionArtifactsBatch(['a', 'b']); + fulfilled(result, 'a'); + fulfilled(result, 'b'); + for (const r of [first, second, sibling]) + await assert.rejects(fs.stat(join(root, 'artifacts', r.relativePath)), { + code: 'ENOENT', + }); + for (const id of ['a', 'b']) + assert.equal((await reopened.store.listPage(id, { offset: 0, limit: 10 })).total, 0); + assert.equal( + await fs.readFile(join(root, 'artifacts', kept.relativePath), 'utf8'), + 'payload kept', + ); + } finally { + reopened.close(); + } + } finally { + if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL'); + await exit; + } + }); + }); +} diff --git a/packages/storage/src/__tests__/fixtures/artifact-batch-purge-crash.ts b/packages/storage/src/__tests__/fixtures/artifact-batch-purge-crash.ts new file mode 100644 index 0000000000..c813132118 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/artifact-batch-purge-crash.ts @@ -0,0 +1,49 @@ +/* + * 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 fs from 'node:fs/promises'; +import { syncBuiltinESMExports } from 'node:module'; +import { createSqliteArtifactStoreWriteAuthority } from '../../artifact-store.js'; + +const [root, phase, target] = process.argv.slice(2); +if (!root || !target || !process.send) throw new Error('Expected root, phase, target and IPC'); +async function crashPoint() { + await new Promise((resolve, reject) => + process.send!(phase, (error) => (error ? reject(error) : resolve())), + ); + // The parent kills this process while the IPC channel keeps it alive. + await new Promise(() => {}); +} +if (phase === 'after-unlink') { + const originalRm = fs.rm; + fs.rm = async (...args: Parameters) => { + await originalRm(...args); + if (args[0] === target) await crashPoint(); + }; + syncBuiltinESMExports(); +} +const authority = createSqliteArtifactStoreWriteAuthority(root); +try { + const results = await authority.store.purgeSessionArtifactsBatch(['a', 'b']); + for (const result of results.values()) if (result.status === 'rejected') throw result.reason; + if (phase !== 'after-metadata') throw new Error('Missed unlink crash point'); + await crashPoint(); +} finally { + authority.close(); +} diff --git a/packages/storage/src/artifact-store.ts b/packages/storage/src/artifact-store.ts index 6bb8dea74f..8b382cc78a 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -193,6 +193,15 @@ export interface ArtifactAuthorityStore extends DurableArtifactAttachmentReader input: ConversationArtifactCopyInput, ): Promise; purgeSessionArtifacts(sessionId: string): Promise; + /** + * Resolves the workspace's path-alias guards once for a retiring Session batch. + * Per-Session unlink/sync failures leave that Session's metadata retryable; + * a guard-resolution or metadata-commit failure rejects the whole batch. + * Callers may discharge cleanup intents only for fulfilled Session results. + */ + purgeSessionArtifactsBatch( + sessionIds: readonly string[], + ): Promise>>; reclaimUpgradeResidue(input: ArtifactUpgradeCleanupInput): Promise; deleteOwnedArtifactInSession( sessionId: string, @@ -476,12 +485,81 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { } async purgeSessionArtifacts(sessionId: string): Promise { - assertCanonicalArtifactEntityId(sessionId, 'sessionId'); - await this.enqueueMutation(async () => { + const results = await this.purgeSessionArtifactsBatch([sessionId]); + const result = results.get(sessionId)!; + if (result.status === 'rejected') throw result.reason; + } + + async purgeSessionArtifactsBatch( + sessionIds: readonly string[], + ): Promise>> { + const acceptedIds = new Set(sessionIds); + for (const sessionId of acceptedIds) assertCanonicalArtifactEntityId(sessionId, 'sessionId'); + if (acceptedIds.size === 0) return new Map(); + return this.enqueueMutation(async () => { await this.prepareMutationUnlocked(); - await this.purgeRecordsUnlocked( - this.records.filter((record) => record.sessionId === sessionId), + const sessions = new Map( + [...acceptedIds].map((sessionId) => [sessionId, []]), ); + for (const record of this.records) sessions.get(record.sessionId)?.push(record); + const results = new Map>(); + if ([...sessions.values()].every((records) => records.length === 0)) { + for (const sessionId of acceptedIds) + results.set(sessionId, { status: 'fulfilled', value: undefined }); + return results; + } + + // Resolve one stable snapshot while holding the writer lock. Two owners + // per identity suffice to preserve each Session's cross-Session alias + // guard, including aliases between two targets in this same batch. + const root = await ensureRealDirectory(this.artifactRoot); + const resolved = await this.resolveRemovalEntriesUnlocked(this.records); + const entries = new Map(); + const owners = new Map(); + for (const [index, record] of this.records.entries()) { + const entry = resolved[index]; + entries.set(record.id, entry); + if (!entry) continue; + const owner = owners.get(entry.comparisonIdentity); + if (!owner) owners.set(entry.comparisonIdentity, { first: record }); + else if (owner.first.sessionId !== record.sessionId) owner.other = record; + } + + const deletedIds = new Set(); + for (const [sessionId, records] of sessions) { + try { + const paths = new Set(); + for (const record of records) { + validateRelativeArtifactPath(record.relativePath); + const entry = entries.get(record.id); + if (!entry) continue; + if (!isInsideOrSamePath(root, dirname(entry.unlinkPath))) { + throw new Error(`Artifact ${record.id} resolves outside the artifact root`); + } + const owner = owners.get(entry.comparisonIdentity)!; + const reference = owner.first.sessionId !== sessionId ? owner.first : owner.other; + if (reference) { + throw new Error( + `Artifact ${record.id} path is still referenced by artifact ${reference.id}`, + ); + } + paths.add(entry.unlinkPath); + } + await this.removePurgePathsUnlocked([...paths]); + for (const record of records) deletedIds.add(record.id); + results.set(sessionId, { status: 'fulfilled', value: undefined }); + } catch (reason) { + results.set(sessionId, { status: 'rejected', reason }); + } + } + // Only Sessions whose unlink and directory-sync obligations succeeded + // participate in this commit. A failed commit rejects the whole batch; + // no caller may clear its retirement intent before metadata is durable. + if (deletedIds.size > 0) { + await this.writeMetadataUnlocked({ deleteIds: [...deletedIds] }); + this.records = this.records.filter((record) => !deletedIds.has(record.id)); + } + return results; }); } @@ -843,6 +921,14 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { paths: readonly string[], ): Promise { const nextRecords = this.records.filter((record) => !ids.has(record.id)); + await this.removePurgePathsUnlocked(paths); + // Keep the paths discoverable until physical cleanup is durable. Session + // retirement already owns the pending cleanup intent and retries on reopen. + await this.writeMetadataUnlocked({ deleteIds: [...ids] }); + this.records = nextRecords; + } + + private async removePurgePathsUnlocked(paths: readonly string[]): Promise { const changedDirectories = new Set(); try { for (const path of paths) { @@ -852,10 +938,6 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { } finally { for (const directory of changedDirectories) await syncDirectory(directory); } - // Keep the paths discoverable until physical cleanup is durable. Session - // retirement already owns the pending cleanup intent and retries on reopen. - await this.writeMetadataUnlocked({ deleteIds: [...ids] }); - this.records = nextRecords; } private async prepareReadInSessionUnlocked( diff --git a/packages/storage/src/artifact-stores.ts b/packages/storage/src/artifact-stores.ts index 7ffd3fac70..e3701426ac 100644 --- a/packages/storage/src/artifact-stores.ts +++ b/packages/storage/src/artifact-stores.ts @@ -72,6 +72,7 @@ export interface InteractiveArtifactStoreWriter extends DurableArtifactAttachmen input: ConversationArtifactCopyInput, ): Promise; purgeSessionArtifacts(sessionId: string): Promise; + purgeSessionArtifactsBatch: ArtifactAuthorityStore['purgeSessionArtifactsBatch']; reclaimUpgradeResidue: ArtifactAuthorityStore['reclaimUpgradeResidue']; listPage: ArtifactAuthorityStore['listPage']; listTurnArtifacts: ArtifactAuthorityStore['listTurnArtifacts']; @@ -178,6 +179,10 @@ function createWriterFacade( return run(() => store.copyConversationArtifacts(acceptedInput)); }, purgeSessionArtifacts: (sessionId) => run(() => store.purgeSessionArtifacts(sessionId)), + purgeSessionArtifactsBatch: (sessionIds) => { + const acceptedIds = Object.freeze([...sessionIds]); + return run(() => store.purgeSessionArtifactsBatch(acceptedIds)); + }, reclaimUpgradeResidue: (input) => run(() => store.reclaimUpgradeResidue(input)), deleteUserArtifactInSession: (sessionId, artifactId) => run(() => store.deleteUserArtifactInSession(sessionId, artifactId)),