diff --git a/docs/architecture/browser-file-and-origin-storage.md b/docs/architecture/browser-file-and-origin-storage.md index 3f28ad2..14f7677 100644 --- a/docs/architecture/browser-file-and-origin-storage.md +++ b/docs/architecture/browser-file-and-origin-storage.md @@ -690,11 +690,20 @@ default physical layout은 구현과 동일하게 다음과 같다. ```text /ca-frontend-opfs-v1/ authorities//// - objects////manifest.json + objects////manifest.json # physical v1 (read-only) + objects///g-/manifest.json # physical v2 (new writes) chunks/sha256//.bin staging//receipt.json ``` +physical v2는 STO-01 수정의 일부다. logical `generation`은 설계상 transaction 간에 +재사용되므로, 늦게 도착한 T1 보상이 같은 logical generation을 쓰는 T2의 디렉터리를 +지울 수 있었다. v2는 transaction-unique `physicalGenerationId` fencing token을 +경로, staging receipt, prepared object에 함께 기록해 보상이 자기 transaction의 +디렉터리만 삭제하도록 만든다. expand 단계에서는 v1 경로/receipt/prepared object를 +계속 읽고 새 write만 v2로 쓴다. rollback window가 끝나기 전에 v1 physical +generation을 일괄 삭제하지 않는다. + 구조화 metadata, query, revision, refcount와 operation journal은 IndexedDB가 소유한다. OPFS에는 immutable chunk와 bounded runtime-schema-validated manifest만 둔다. readable `scope.namespace`는 경로에 쓰지 않는다. @@ -736,6 +745,23 @@ COMMITTED <- 사용자에게 보이는 유일한 commit point CLEANED -> journal 제거 ``` +보상(compensation)은 saga의 반쪽이며 다음 규칙을 따른다. + +- journal row와 budget reservation은 physical cleanup effect가 + `CLEANED` 또는 `ALREADY_CLEAN`으로 확인된 뒤에만 해제한다. timeout, crash, + malformed response, `EFFECT_UNKNOWN`은 성공이 아니며 `PREPARING`/`FILES_READY`를 + 그대로 남기고 `OBJECT_RECONCILE`로 반환한다. +- coordinator가 `abortPreparedPut()` 하나만 소유한다. worker client는 prepare 실패 + 시 별도의 fire-and-forget abort를 발행하지 않는다. 중복 보상은 아직 남아 있어야 + 할 journal row를 조기에 지우는 경로였다. +- 보상은 caller signal을 상속하지 않는다. composition이 소유한 bounded + `compensationSignal`을 사용하므로 이미 abort된 caller가 cleanup RPC 자체를 + 시작조차 못 하게 만들 수 없다. +- abort/cleanup은 origin mutation Web Lock을 physical 삭제와 staging 제거가 끝날 + 때까지 계속 보유한다. lease를 먼저 release하지 않는다. 단, staging이 아직 없는 + transaction은 삭제할 것이 없으므로 lock을 기다리지 않고 `ALREADY_CLEAN`을 + 반환한다. 이는 자기 자신이 취소하는 BEGIN과의 deadlock을 막는다. + - `PREPARING` crash: partial staging을 검증 후 resume하거나 purge한다. - `FILES_READY` crash: expected generation과 digest가 맞으면 idempotent logical commit, 아니면 quarantine한다. diff --git a/docs/operations/adapter-remediation-ledger.md b/docs/operations/adapter-remediation-ledger.md index 6535b53..e5fbc6d 100644 --- a/docs/operations/adapter-remediation-ledger.md +++ b/docs/operations/adapter-remediation-ledger.md @@ -88,7 +88,7 @@ Rollout state starts at `NOT_STARTED`; documented-unimplemented items start at | ID | Activation | Red test command | Fix commit/PR | Rollout state | Rollback trigger | Evidence | | --- | --- | --- | --- | --- | --- | --- | -| STO-01 | OPFS not composed in template; **Critical** for any product writer | `corepack pnpm exec vitest run tests/unit/opfs-byte-store.test.ts tests/unit/indexeddb-opfs-journal.test.ts` | — | `NOT_STARTED` | OPFS reconcile backlog or journal growth | — | +| STO-01 | OPFS not composed in template; **Critical** for any product writer | `corepack pnpm exec vitest run tests/unit/opfs-byte-store.test.ts tests/unit/opfs-worker-runtime.test.ts tests/unit/indexeddb-opfs-journal.test.ts` | `fix: preserve OPFS recovery authority during cleanup` | `FIXED_NOT_RELEASED` | OPFS reconcile backlog or journal growth | Red 4 new saga cases → green 25/25 across the three OPFS suites; `check:types` PASS (incl. web-worker); `check:browser-file-storage-boundaries` PASS; `lint` PASS; `test:unit` 1511 passed with only the pre-existing environmental `ci-artifact-contract` failures | | STO-02 | Browser file runtime not composed | `corepack pnpm exec vitest run tests/unit/browser-file-download.test.ts` | — | `NOT_STARTED` | download navigation blocked by canonical target | — | | STO-03 | Public cache not composed | `corepack pnpm exec vitest run tests/unit/public-response-cache.test.ts` | — | `NOT_STARTED` | composition rejection of an existing policy | — | | STO-04 | Public cache not composed | `corepack pnpm exec vitest run tests/unit/public-response-cache.test.ts` | — | `NOT_STARTED` | restage loop or bandwidth spike | — | diff --git a/docs/operations/browser-file-storage-recovery.md b/docs/operations/browser-file-storage-recovery.md index 248c5fe..56b6623 100644 --- a/docs/operations/browser-file-storage-recovery.md +++ b/docs/operations/browser-file-storage-recovery.md @@ -13,6 +13,24 @@ probe는 현재 `DESIGNED_NOT_IMPLEMENTED`다. 아래 절차에서 이 기능을 자동 조치는 해당 runtime이 구현·조합된 제품에서만 실행한다. 현재 reference primitive를 coordinator 완료 증거로 사용하지 않는다. + +## OPFS 보상 실패와 reconcile (STO-01) + +`put()`이 실패했는데 보상 cleanup effect가 확인되지 않으면 runtime은 실패를 +`OBJECT_RECONCILE` / `CONFLICT`(recovery `RETRY`)로 보고하고 journal row를 남긴다. +이는 결함이 아니라 설계된 상태다. + +1. journal에 `PREPARING` 또는 `FILES_READY` row가 남아 있는지 확인한다. 남아 + 있다면 staging bytes가 아직 존재할 수 있다는 뜻이다. +2. `maintenance.reconcile()`을 실행한다. reconcile은 같은 exact physical + generation token만 삭제하고, effect가 여전히 `EFFECT_UNKNOWN`이면 journal을 + 유지한 채 다시 `OBJECT_RECONCILE`을 반환한다. +3. journal row를 수동으로 삭제하지 않는다. row가 사라지면 stale staging을 추적할 + 근거가 사라지고 quota만 누수된다. +4. OPFS root나 journal database를 통째로 삭제하거나 schema를 downgrade하지 + 않는다. rollback은 새 v2 write admission을 닫고 v1+v2 reader를 유지하는 것으로 + 수행한다. + ## 1. 공통 원칙 incident 중에도 다음 작업은 금지한다. diff --git a/src/adapters/storage/opfs/indexeddb-opfs-journal.ts b/src/adapters/storage/opfs/indexeddb-opfs-journal.ts index 6b2b8df..3b4996a 100644 --- a/src/adapters/storage/opfs/indexeddb-opfs-journal.ts +++ b/src/adapters/storage/opfs/indexeddb-opfs-journal.ts @@ -1650,12 +1650,26 @@ function isChunkReference( ); } +/** + * STO-01 expand phase. v1 prepared objects stay readable through the rollback + * window; v2 additionally carries a transaction-unique physical fencing token. + */ +function isSupportedPhysicalSchema(value: object): boolean { + const record = value as Record; + if (record.physicalSchemaVersion === 1) return true; + return ( + record.physicalSchemaVersion === 2 && + typeof record.physicalGenerationId === "string" && + /^[0-9a-f]{32}$/u.test(record.physicalGenerationId) + ); +} + function isPreparedObject(value: unknown): value is OpfsPreparedObject { if ( !value || typeof value !== "object" || !("physicalSchemaVersion" in value) || - value.physicalSchemaVersion !== 1 || + !isSupportedPhysicalSchema(value) || !("descriptor" in value) || !value.descriptor || typeof value.descriptor !== "object" || diff --git a/src/adapters/storage/opfs/opfs-byte-store-adapter.ts b/src/adapters/storage/opfs/opfs-byte-store-adapter.ts index 1deb6a8..e25d4dc 100644 --- a/src/adapters/storage/opfs/opfs-byte-store-adapter.ts +++ b/src/adapters/storage/opfs/opfs-byte-store-adapter.ts @@ -10,6 +10,7 @@ import type { OpfsReconciliationReport, OpfsStorageScope, PutDurableObjectRequest, + OpfsPhysicalGenerationId, } from "../../../application/ports/browser-file-storage/opfs-ports.ts"; import { type BrowserDataFailure, @@ -89,6 +90,12 @@ export type OpfsByteStoreDependencies = Readonly<{ storagePolicy: BrowserStoragePolicy; policy?: Partial; createTransactionId?: () => string; + createPhysicalGenerationId?: () => OpfsPhysicalGenerationId; + /** + * Composition-owned bounded signal for compensating cleanup. It is + * deliberately separate from any caller signal. + */ + compensationSignal?: AbortSignal; now?: () => number; observer?: OpfsSafeObserver; /** @@ -130,6 +137,20 @@ export function createOpfsByteStoreAdapter( const createTransactionId = dependencies.createTransactionId ?? (() => globalThis.crypto.randomUUID()); + const createPhysicalGenerationId = + dependencies.createPhysicalGenerationId ?? + (() => { + const bytes = new Uint8Array(16); + globalThis.crypto.getRandomValues(bytes); + let hex = ""; + for (const byte of bytes) hex += byte.toString(16).padStart(2, "0"); + return hex as OpfsPhysicalGenerationId; + }); + /** + * STO-01. Compensation must not inherit the caller's already aborted signal; + * a cleanup that never starts cannot justify releasing the journal row. + */ + const compensationSignal = dependencies.compensationSignal; const now = dependencies.now ?? Date.now; const objects: DurableObjectStorePort = Object.freeze({ @@ -182,6 +203,10 @@ export function createOpfsByteStoreAdapter( } const targetGeneration = (current?.descriptor.generation ?? 0) + 1; const transactionId = createTransactionId(); + // STO-01. The logical generation is reused across transactions; this + // token makes the physical target unique so a late compensation can never + // delete a newer transaction's directory. + const physicalGenerationId = createPhysicalGenerationId(); notifyProgress(request, "PREPARING", 0); const begun = await dependencies.journal.begin({ transactionId, @@ -214,13 +239,24 @@ export function createOpfsByteStoreAdapter( }); const prepared = await dependencies.worker.preparePut({ transactionId, + physicalGenerationId, descriptor, source: request.source, signal: request.signal, onProgress: request.onProgress, }); if (!prepared.ok) { - await rollbackBestEffort(begun.value, request.signal); + const compensated = await compensatePreparedPut( + begun.value, + physicalGenerationId, + ); + if (!compensated.ok) { + return observeFailure( + compensated, + dependencies.observer, + request.source.byteLength!, + ); + } return observeFailure( prepared, dependencies.observer, @@ -234,7 +270,17 @@ export function createOpfsByteStoreAdapter( prepared.value, ); if (!filesReady.ok) { - await rollbackBestEffort(begun.value, request.signal); + const compensated = await compensatePreparedPut( + begun.value, + physicalGenerationId, + ); + if (!compensated.ok) { + return observeFailure( + compensated, + dependencies.observer, + request.source.byteLength!, + ); + } return observeFailure( rebaseFailure(filesReady.error, "OBJECT_WRITE"), dependencies.observer, @@ -830,19 +876,38 @@ export function createOpfsByteStoreAdapter( return Object.freeze({ objects, maintenance }); - async function rollbackBestEffort( + /** + * STO-01. The compensating half of the put saga. + * + * The journal row and its budget reservation are the only durable evidence + * that a physical staging generation may still exist, so they are released + * exactly when the physical effect is confirmed `CLEANED` or + * `ALREADY_CLEAN`. A timeout, crash, malformed response or `EFFECT_UNKNOWN` + * keeps `PREPARING`/`FILES_READY` in place and asks for reconciliation. + */ + async function compensatePreparedPut( transaction: OpfsJournalTransaction, - signal: AbortSignal | undefined, - ): Promise { - await dependencies.worker.cleanupTransaction( - transaction.scope, - transaction.transactionId, - signal, - ); - await dependencies.journal.rollback( + physicalGenerationId: OpfsPhysicalGenerationId, + ): Promise> { + const cleanup = await dependencies.worker.abortPreparedPut({ + scope: transaction.scope, + transactionId: transaction.transactionId, + physicalGenerationId, + ...(compensationSignal ? { signal: compensationSignal } : {}), + }); + if (!cleanup.ok || cleanup.value.kind === "EFFECT_UNKNOWN") { + return browserDataFailure("CONFLICT", "OBJECT_RECONCILE", { + recovery: "RETRY", + }); + } + const rolledBack = await dependencies.journal.rollback( transaction.transactionId, transaction.fencingToken, ); + if (!rolledBack.ok) { + return rebaseFailure(rolledBack.error, "OBJECT_RECONCILE"); + } + return browserDataSuccess(undefined); } async function reconcileTransaction( @@ -860,6 +925,11 @@ export function createOpfsByteStoreAdapter( signal, ); if (!cleaned.ok) return cleaned; + if (cleaned.value.kind === "EFFECT_UNKNOWN") { + return browserDataFailure("CONFLICT", "OBJECT_RECONCILE", { + recovery: "RETRY", + }); + } const rolledBack = await dependencies.journal.rollback( transaction.transactionId, transaction.fencingToken, @@ -1292,6 +1362,7 @@ function snapshotOpfsWorker( openObject, removeObject, cleanupTransaction, + abortPreparedPut, finalizePut, listOrphanCandidates, deleteOrphanChunk, @@ -1305,6 +1376,7 @@ function snapshotOpfsWorker( openObject, removeObject, cleanupTransaction, + abortPreparedPut, finalizePut, listOrphanCandidates, deleteOrphanChunk, @@ -1320,6 +1392,7 @@ function snapshotOpfsWorker( openObject: openObject.bind(source), removeObject: removeObject.bind(source), cleanupTransaction: cleanupTransaction.bind(source), + abortPreparedPut: abortPreparedPut.bind(source), finalizePut: finalizePut.bind(source), listOrphanCandidates: listOrphanCandidates.bind(source), deleteOrphanChunk: deleteOrphanChunk.bind(source), diff --git a/src/adapters/storage/opfs/opfs-worker-client.ts b/src/adapters/storage/opfs/opfs-worker-client.ts index 8e1e2bf..ad5e70d 100644 --- a/src/adapters/storage/opfs/opfs-worker-client.ts +++ b/src/adapters/storage/opfs/opfs-worker-client.ts @@ -1,5 +1,6 @@ import type { OpfsCapabilities, + OpfsCleanupEffect, OpfsPreparedObject, } from "../../../application/ports/browser-file-storage/opfs-ports.ts"; import type { @@ -14,6 +15,7 @@ import { } from "../../browser-file-storage/result.ts"; import type { OpfsRuntimePolicy } from "./opfs-policy.ts"; import type { + AbortPreparedPutRequest, OpfsWorkerGateway, OpfsOrphanCandidateBatch, OpfsOrphanDeleteResult, @@ -187,17 +189,6 @@ export function createOpfsWorkerGateway( } } - async function abortAndCleanup( - scope: OpfsPreparedObject["descriptor"]["scope"], - transactionId: string, - ): Promise { - try { - await rpc({ kind: "ABORT_PUT", scope, transactionId }); - } catch { - // Journal reconciliation repeats cleanup after a crash or timeout. - } - } - return Object.freeze({ async capabilities() { return await invoke( @@ -215,6 +206,7 @@ export function createOpfsWorkerGateway( { kind: "BEGIN_PUT", transactionId: request.transactionId, + physicalGenerationId: request.physicalGenerationId, scope: request.descriptor.scope, objectId: request.descriptor.objectId, generation: request.descriptor.generation, @@ -227,10 +219,8 @@ export function createOpfsWorkerGateway( request.signal, ); if (!begin.ok) { - await abortAndCleanup( - request.descriptor.scope, - request.transactionId, - ); + // STO-01. Compensation belongs to the coordinator: it owns the journal + // row this cleanup would otherwise invalidate. return begin; } @@ -257,22 +247,12 @@ export function createOpfsWorkerGateway( request.signal, [chunk], ); - if (!append.ok) { - await abortAndCleanup( - request.descriptor.scope, - request.transactionId, - ); - return append; - } + if (!append.ok) return append; sequence += 1; transferredBytes += chunkByteLength; notifyProgress(request, "TRANSFERRING", transferredBytes); } if (transferredBytes !== request.descriptor.byteLength) { - await abortAndCleanup( - request.descriptor.scope, - request.transactionId, - ); return browserDataFailure("INTEGRITY_FAILED", "OBJECT_WRITE", { recovery: "RESELECT", }); @@ -289,18 +269,8 @@ export function createOpfsWorkerGateway( [], parsePreparedObject, ); - if (!finished.ok) { - await abortAndCleanup( - request.descriptor.scope, - request.transactionId, - ); - } return finished; } catch (error) { - await abortAndCleanup( - request.descriptor.scope, - request.transactionId, - ); return failureResult( error instanceof OpfsRpcError ? error.code : "NOT_READABLE", "OBJECT_WRITE", @@ -387,10 +357,32 @@ export function createOpfsWorkerGateway( transactionId: string, signal?: AbortSignal, ) { - return await invoke( + return await invoke( "OBJECT_RECONCILE", { kind: "CLEANUP_TRANSACTION", scope, transactionId }, signal, + [], + parseCleanupEffect, + ); + }, + + /** + * STO-01. The coordinator's single compensation entry point. An RPC that + * times out or fails leaves the physical effect unknown, which is never a + * success and must not release journal or budget state. + */ + async abortPreparedPut(request: AbortPreparedPutRequest) { + return await invoke( + "OBJECT_RECONCILE", + { + kind: "ABORT_PUT", + scope: request.scope, + transactionId: request.transactionId, + physicalGenerationId: request.physicalGenerationId, + }, + request.signal, + [], + parseCleanupEffect, ); }, @@ -610,12 +602,34 @@ function parseCapabilities(value: unknown): OpfsCapabilities | null { return value as OpfsCapabilities; } +/** + * STO-01 expand phase. v1 prepared objects stay readable through the rollback + * window; v2 additionally carries a transaction-unique physical fencing token. + */ +function isSupportedPhysicalSchema(value: object): boolean { + const record = value as Record; + if (record.physicalSchemaVersion === 1) return true; + return ( + record.physicalSchemaVersion === 2 && + typeof record.physicalGenerationId === "string" && + /^[0-9a-f]{32}$/u.test(record.physicalGenerationId) + ); +} + +function parseCleanupEffect(value: unknown): OpfsCleanupEffect | null { + if (!value || typeof value !== "object") return null; + const kind = (value as Record).kind; + return kind === "CLEANED" || kind === "ALREADY_CLEAN" + ? Object.freeze({ kind }) + : null; +} + function parsePreparedObject(value: unknown): OpfsPreparedObject | null { if ( !value || typeof value !== "object" || !("physicalSchemaVersion" in value) || - value.physicalSchemaVersion !== 1 || + !isSupportedPhysicalSchema(value) || !("descriptor" in value) || !("chunks" in value) || !Array.isArray(value.chunks) diff --git a/src/adapters/storage/opfs/opfs-worker-protocol.ts b/src/adapters/storage/opfs/opfs-worker-protocol.ts index 90768a6..7f51ddd 100644 --- a/src/adapters/storage/opfs/opfs-worker-protocol.ts +++ b/src/adapters/storage/opfs/opfs-worker-protocol.ts @@ -1,6 +1,8 @@ import type { DurableObjectDescriptor, OpfsCapabilities, + OpfsCleanupEffect, + OpfsPhysicalGenerationId, OpfsPreparedObject, OpfsStorageScope, } from "../../../application/ports/browser-file-storage/opfs-ports.ts"; @@ -24,6 +26,8 @@ export type OpfsWorkerRequest = scope: OpfsStorageScope; objectId: string; generation: number; + /** STO-01. Transaction-unique physical fencing token for new writes. */ + physicalGenerationId: OpfsPhysicalGenerationId; declaredByteLength: number; mediaType: string; createdAtEpochMs: number; @@ -49,6 +53,7 @@ export type OpfsWorkerRequest = kind: "ABORT_PUT"; scope: OpfsStorageScope; transactionId: string; + physicalGenerationId?: OpfsPhysicalGenerationId; }> | Readonly<{ requestId: string; @@ -73,6 +78,12 @@ export type OpfsWorkerRequest = kind: "CLEANUP_TRANSACTION"; scope: OpfsStorageScope; transactionId: string; + /** + * STO-01. When present, cleanup deletes only this exact physical + * generation and can never touch a newer transaction that reused the same + * logical generation. + */ + physicalGenerationId?: OpfsPhysicalGenerationId; }> | Readonly<{ requestId: string; @@ -124,6 +135,7 @@ export type OpfsWorkerResponse = ok: true; value?: | OpfsCapabilities + | OpfsCleanupEffect | OpfsPreparedObject | ArrayBuffer | boolean @@ -138,12 +150,25 @@ export type OpfsWorkerResponse = export type PreparePhysicalObjectRequest = Readonly<{ transactionId: string; + physicalGenerationId: OpfsPhysicalGenerationId; descriptor: Omit; source: ByteSource; signal?: AbortSignal; onProgress?: (progress: TransferProgress) => void; }>; +/** + * STO-01. The single compensation entry point. The coordinator owns it and + * passes a composition-owned bounded signal, never the already aborted caller + * signal. + */ +export type AbortPreparedPutRequest = Readonly<{ + scope: OpfsStorageScope; + transactionId: string; + physicalGenerationId: OpfsPhysicalGenerationId; + signal?: AbortSignal; +}>; + /** * The coordinator depends on this technology-neutral worker gateway. The * browser implementation below the boundary owns Worker, MessageEvent and @@ -172,7 +197,10 @@ export interface OpfsWorkerGateway { scope: OpfsStorageScope, transactionId: string, signal?: AbortSignal, - ): Promise>; + ): Promise>; + abortPreparedPut( + request: AbortPreparedPutRequest, + ): Promise>; finalizePut( transactionId: string, preparedObject: OpfsPreparedObject, diff --git a/src/adapters/storage/opfs/opfs-worker-runtime.ts b/src/adapters/storage/opfs/opfs-worker-runtime.ts index 96c5856..733b4b6 100644 --- a/src/adapters/storage/opfs/opfs-worker-runtime.ts +++ b/src/adapters/storage/opfs/opfs-worker-runtime.ts @@ -1,6 +1,8 @@ import type { OpfsCapabilities, OpfsChunkReference, + OpfsCleanupEffect, + OpfsPhysicalGenerationId, OpfsPreparedObject, OpfsStorageScope, } from "../../../application/ports/browser-file-storage/opfs-ports.ts"; @@ -54,6 +56,7 @@ type ActivePut = { readonly scope: OpfsPreparedObject["descriptor"]["scope"]; readonly objectId: string; readonly generation: number; + readonly physicalGenerationId: OpfsPhysicalGenerationId; readonly declaredByteLength: number; readonly mediaType: string; readonly createdAtEpochMs: number; @@ -209,8 +212,14 @@ export function createOpfsWorkerRuntime( case "FINISH_PUT": return success(request.requestId, await finishPut(request)); case "ABORT_PUT": - await abortPut(request.scope, request.transactionId); - return success(request.requestId); + return success( + request.requestId, + await abortPut( + request.scope, + request.transactionId, + request.physicalGenerationId, + ), + ); case "VERIFY_OBJECT": return success( request.requestId, @@ -232,11 +241,15 @@ export function createOpfsWorkerRuntime( ); return success(request.requestId); case "CLEANUP_TRANSACTION": - await cleanupTransaction( - request.scope, - request.transactionId, + return success( + request.requestId, + await cleanupTransaction( + request.scope, + request.transactionId, + true, + request.physicalGenerationId, + ), ); - return success(request.requestId); case "FINALIZE_PUT": await finalizePut( request.transactionId, @@ -278,6 +291,7 @@ export function createOpfsWorkerRuntime( !dependencies.policy.isObjectIdAllowed(request.objectId) || !Number.isSafeInteger(request.generation) || request.generation < 1 || + !isPhysicalGenerationId(request.physicalGenerationId) || !Number.isSafeInteger(request.declaredByteLength) || request.declaredByteLength < 0 || request.declaredByteLength > dependencies.policy.maxObjectBytes || @@ -327,6 +341,7 @@ export function createOpfsWorkerRuntime( scope: request.scope, objectId: request.objectId, generation: request.generation, + physicalGenerationId: request.physicalGenerationId, declaredByteLength: request.declaredByteLength, mediaType: request.mediaType, createdAtEpochMs: request.createdAtEpochMs, @@ -457,7 +472,8 @@ export function createOpfsWorkerRuntime( ); assertActivePut(transactionKey, put, true); const prepared: OpfsPreparedObject = Object.freeze({ - physicalSchemaVersion: 1, + physicalSchemaVersion: 2, + physicalGenerationId: put.physicalGenerationId, descriptor: Object.freeze({ objectId: put.objectId, scope: put.scope, @@ -501,10 +517,13 @@ export function createOpfsWorkerRuntime( async function abortPut( scope: OpfsPreparedObject["descriptor"]["scope"], transactionId: string, - ): Promise { + physicalGenerationId?: OpfsPhysicalGenerationId, + ): Promise { if ( !isValidOpfsStorageScope(scope) || - !SAFE_TRANSACTION_ID.test(transactionId) + !SAFE_TRANSACTION_ID.test(transactionId) || + (physicalGenerationId !== undefined && + !isPhysicalGenerationId(physicalGenerationId)) ) { throw new OpfsRuntimeFailure("INVALID_INPUT"); } @@ -518,15 +537,33 @@ export function createOpfsWorkerRuntime( await active.operationTail; if (activePuts.get(transactionKey) === active) { activePuts.delete(transactionKey); + } + try { + // STO-01. The mutation lease is held through the physical delete and + // the staging cleanup; releasing it earlier would let a new transaction + // race this compensation. + await removePhysicalGeneration( + active.scope, + active.objectId, + active.generation, + active.physicalGenerationId, + ); + return await cleanupTransactionLocked( + scope, + transactionId, + true, + physicalGenerationId ?? active.physicalGenerationId, + ); + } finally { active.lease.release(); } - await removePhysicalGeneration( - active.scope, - active.objectId, - active.generation, - ); } - await cleanupTransaction(scope, transactionId); + return await cleanupTransaction( + scope, + transactionId, + true, + physicalGenerationId, + ); } async function runActivePutOperation( @@ -571,20 +608,30 @@ export function createOpfsWorkerRuntime( put.abortController.abort(); if (activePuts.get(transactionKey) === put) { activePuts.delete(transactionKey); + } + try { + await removePhysicalGeneration( + put.scope, + put.objectId, + put.generation, + put.physicalGenerationId, + ); + await cleanupTransactionLocked( + put.scope, + put.transactionId, + true, + put.physicalGenerationId, + ); + } finally { put.lease.release(); } - await removePhysicalGeneration( - put.scope, - put.objectId, - put.generation, - ); - await cleanupTransaction(put.scope, put.transactionId); } async function removePhysicalGeneration( scope: OpfsStorageScope, objectId: string, generation: number, + physicalGenerationId: OpfsPhysicalGenerationId | undefined, ): Promise { try { const objectDirectory = await getDirectory( @@ -599,7 +646,7 @@ export function createOpfsWorkerRuntime( ); await removeEntryIfPresent( objectDirectory, - String(generation), + generationSegmentFor(generation, physicalGenerationId), true, ); } catch (error) { @@ -714,17 +761,68 @@ export function createOpfsWorkerRuntime( } } + /** + * STO-01. Cleanup deletes exactly one transaction's physical generation while + * holding the origin mutation lease, and reports whether the effect actually + * happened. `ALREADY_CLEAN` means there was nothing left to delete. + */ async function cleanupTransaction( scope: OpfsPreparedObject["descriptor"]["scope"], transactionId: string, removePreparedGeneration = true, - ): Promise { + physicalGenerationId?: OpfsPhysicalGenerationId, + ): Promise { if ( !isValidOpfsStorageScope(scope) || !SAFE_TRANSACTION_ID.test(transactionId) ) { throw new OpfsRuntimeFailure("INVALID_INPUT"); } + // Probe before locking. A transaction that never reached staging has + // nothing to delete, and waiting for the mutation lease here would deadlock + // against the very BEGIN this compensation is cancelling. + try { + await getDirectory( + dependencies.root, + [...scopeRootPath(scope), "staging", transactionId], + false, + ); + } catch (error) { + if (isNotFound(error)) return CLEANUP_ALREADY_CLEAN; + throw error; + } + // Every destructive step below runs while the lease is held. + const lease = await dependencies.leaseManager!.acquire(); + try { + return await cleanupTransactionLocked( + scope, + transactionId, + removePreparedGeneration, + physicalGenerationId, + ); + } finally { + lease.release(); + } + } + + /** + * Callers that already hold the origin mutation lease use this directly, so + * an abort never releases the lease between fencing and physical deletion. + */ + async function cleanupTransactionLocked( + scope: OpfsPreparedObject["descriptor"]["scope"], + transactionId: string, + removePreparedGeneration = true, + physicalGenerationId?: OpfsPhysicalGenerationId, + ): Promise { + if ( + !isValidOpfsStorageScope(scope) || + !SAFE_TRANSACTION_ID.test(transactionId) || + (physicalGenerationId !== undefined && + !isPhysicalGenerationId(physicalGenerationId)) + ) { + throw new OpfsRuntimeFailure("INVALID_INPUT"); + } let staging: FileSystemDirectoryHandle; try { staging = await getDirectory( @@ -733,32 +831,38 @@ export function createOpfsWorkerRuntime( false, ); } catch (error) { - if (isNotFound(error)) return; + if (isNotFound(error)) return CLEANUP_ALREADY_CLEAN; throw error; } - if (removePreparedGeneration) { - let receipt: unknown; - try { - receipt = await readJson(receiptPath(scope, transactionId)); - } catch (error) { - if (isNotFound(error)) { - await removeEntryIfPresent(staging, transactionId, true); - return; + { + if (removePreparedGeneration) { + let receipt: unknown; + try { + receipt = await readJson(receiptPath(scope, transactionId)); + } catch (error) { + if (isNotFound(error)) { + await removeEntryIfPresent(staging, transactionId, true); + return CLEANUP_ALREADY_CLEAN; + } + // Keep unreadable staging in place so orphan GC fails closed. + throw error; } - // Keep unreadable staging in place so orphan GC fails closed. - throw error; + const target = extractReceiptPhysicalTarget(receipt, scope); + if (!target) { + throw new OpfsRuntimeFailure("CORRUPT_DATA"); + } + // A caller-supplied token wins: a stale compensation must not widen its + // target to whatever the receipt now says. + await removePhysicalGeneration( + scope, + target.objectId, + target.generation, + physicalGenerationId ?? target.physicalGenerationId, + ); } - const target = extractReceiptPhysicalTarget(receipt, scope); - if (!target) { - throw new OpfsRuntimeFailure("CORRUPT_DATA"); - } - await removePhysicalGeneration( - scope, - target.objectId, - target.generation, - ); + await removeEntryIfPresent(staging, transactionId, true); + return CLEANUP_CLEANED; } - await removeEntryIfPresent(staging, transactionId, true); } async function finalizePut( @@ -788,8 +892,8 @@ export function createOpfsWorkerRuntime( for await (const [name, handle] of objectDirectory.entries()) { if ( handle.kind === "directory" && - /^\d+$/u.test(name) && - name !== String(descriptor.generation) + isGenerationSegment(name) && + name !== preparedGenerationSegment(preparedObject) ) { await objectDirectory.removeEntry(name, { recursive: true }); } @@ -980,7 +1084,11 @@ export function createOpfsWorkerRuntime( function extractReceiptPhysicalTarget( receipt: unknown, scope: OpfsStorageScope, - ): Readonly<{ objectId: string; generation: number }> | null { + ): Readonly<{ + objectId: string; + generation: number; + physicalGenerationId: OpfsPhysicalGenerationId | undefined; + }> | null { if (!receiptBelongsToScope(receipt, scope)) return null; const record = receipt as Record; if (record.phase === "PREPARING") { @@ -989,7 +1097,15 @@ export function createOpfsWorkerRuntime( typeof record.generation === "number" && Number.isSafeInteger(record.generation) && record.generation > 0 - ? { objectId: record.objectId, generation: record.generation } + ? { + objectId: record.objectId, + generation: record.generation, + physicalGenerationId: isPhysicalGenerationId( + record.physicalGenerationId, + ) + ? record.physicalGenerationId + : undefined, + } : null; } if ( @@ -997,9 +1113,14 @@ export function createOpfsWorkerRuntime( isPreparedObjectSafe(record.preparedObject, dependencies.policy) && sameScope(record.preparedObject.descriptor.scope, scope) ) { + const prepared = record.preparedObject; return { - objectId: record.preparedObject.descriptor.objectId, - generation: record.preparedObject.descriptor.generation, + objectId: prepared.descriptor.objectId, + generation: prepared.descriptor.generation, + physicalGenerationId: + prepared.physicalSchemaVersion === 2 + ? prepared.physicalGenerationId + : undefined, }; } return null; @@ -1019,11 +1140,12 @@ export function createOpfsWorkerRuntime( async function writeReceipt(put: ActivePut): Promise { await writeJsonAtomic(receiptPath(put.scope, put.transactionId), { - schemaVersion: 1, + schemaVersion: 2, phase: "PREPARING", scope: put.scope, objectId: put.objectId, generation: put.generation, + physicalGenerationId: put.physicalGenerationId, declaredByteLength: put.declaredByteLength, chunks: put.chunks, }); @@ -1296,6 +1418,52 @@ async function removeEntryIfPresent( } } +const CLEANUP_CLEANED: OpfsCleanupEffect = Object.freeze({ kind: "CLEANED" }); +const CLEANUP_ALREADY_CLEAN: OpfsCleanupEffect = Object.freeze({ + kind: "ALREADY_CLEAN", +}); + +const PHYSICAL_GENERATION_ID = /^[0-9a-f]{32}$/u; +const V1_GENERATION_SEGMENT = /^\d+$/u; +const V2_GENERATION_SEGMENT = /^g\d+-[0-9a-f]{32}$/u; + +export function isPhysicalGenerationId( + value: unknown, +): value is OpfsPhysicalGenerationId { + return typeof value === "string" && PHYSICAL_GENERATION_ID.test(value); +} + +/** + * STO-01. v1 wrote `objects////`, which two + * transactions can legitimately share. v2 writes + * `objects///g-/` so a late compensation can + * only ever delete its own transaction's directory. v1 segments stay readable + * through the rollback window. + */ +function generationSegmentFor( + generation: number, + physicalGenerationId: OpfsPhysicalGenerationId | undefined, +): string { + return physicalGenerationId === undefined + ? String(generation) + : `g${generation}-${physicalGenerationId}`; +} + +function preparedGenerationSegment( + preparedObject: OpfsPreparedObject, +): string { + return generationSegmentFor( + preparedObject.descriptor.generation, + preparedObject.physicalSchemaVersion === 2 + ? preparedObject.physicalGenerationId + : undefined, + ); +} + +function isGenerationSegment(name: string): boolean { + return V1_GENERATION_SEGMENT.test(name) || V2_GENERATION_SEGMENT.test(name); +} + function manifestPath(preparedObject: OpfsPreparedObject): readonly string[] { const descriptor = preparedObject.descriptor; return [ @@ -1303,7 +1471,7 @@ function manifestPath(preparedObject: OpfsPreparedObject): readonly string[] { "objects", descriptor.objectId.slice(0, 2), descriptor.objectId, - String(descriptor.generation), + preparedGenerationSegment(preparedObject), "manifest.json", ]; } @@ -1373,7 +1541,8 @@ function receiptBelongsToScope( if (!receipt || typeof receipt !== "object") return false; const record = receipt as Record; if ( - record.schemaVersion !== 1 || + // v1 receipts stay readable through the rollback window. + (record.schemaVersion !== 1 && record.schemaVersion !== 2) || (record.phase !== "PREPARING" && record.phase !== "FILES_READY") ) { return false; @@ -1464,6 +1633,20 @@ function stableJson(value: unknown): string { throw new OpfsRuntimeFailure("CORRUPT_DATA"); } +/** + * STO-01 expand phase. v1 prepared objects stay readable through the rollback + * window; v2 additionally carries a transaction-unique physical fencing token. + */ +function isSupportedPhysicalSchema(value: object): boolean { + const record = value as Record; + if (record.physicalSchemaVersion === 1) return true; + return ( + record.physicalSchemaVersion === 2 && + typeof record.physicalGenerationId === "string" && + /^[0-9a-f]{32}$/u.test(record.physicalGenerationId) + ); +} + function isPreparedObjectSafe( value: unknown, policy: OpfsRuntimePolicy, @@ -1472,7 +1655,7 @@ function isPreparedObjectSafe( !value || typeof value !== "object" || !("physicalSchemaVersion" in value) || - value.physicalSchemaVersion !== 1 || + !isSupportedPhysicalSchema(value) || !("descriptor" in value) || !value.descriptor || typeof value.descriptor !== "object" || @@ -1567,6 +1750,7 @@ function success( value?: OpfsWorkerResponse extends infer _Response ? | OpfsCapabilities + | OpfsCleanupEffect | OpfsPreparedObject | ArrayBuffer | boolean diff --git a/src/application/ports/browser-file-storage/opfs-ports.ts b/src/application/ports/browser-file-storage/opfs-ports.ts index 011adc3..25d1fb7 100644 --- a/src/application/ports/browser-file-storage/opfs-ports.ts +++ b/src/application/ports/browser-file-storage/opfs-ports.ts @@ -169,12 +169,46 @@ export type OpfsChunkReference = Readonly<{ digestHex: string; }>; -export type OpfsPreparedObject = Readonly<{ +/** + * STO-01. A transaction-unique physical fencing token. + * + * The logical `generation` is reused across transactions by design, so a late + * compensation from an abandoned transaction could otherwise delete the + * physical directory a newer transaction just created under the same logical + * generation. Physical paths are keyed by this token instead. + */ +declare const opfsPhysicalGenerationBrand: unique symbol; +export type OpfsPhysicalGenerationId = string & { + readonly [opfsPhysicalGenerationBrand]: "OpfsPhysicalGenerationId"; +}; + +export type OpfsPreparedObjectV1 = Readonly<{ descriptor: DurableObjectDescriptor; chunks: readonly OpfsChunkReference[]; physicalSchemaVersion: 1; }>; +export type OpfsPreparedObjectV2 = Readonly<{ + descriptor: DurableObjectDescriptor; + chunks: readonly OpfsChunkReference[]; + physicalSchemaVersion: 2; + physicalGenerationId: OpfsPhysicalGenerationId; +}>; + +/** + * Expand phase: v1 readers stay for the rollback window while every new write + * emits v2. + */ +export type OpfsPreparedObject = OpfsPreparedObjectV1 | OpfsPreparedObjectV2; + +/** + * STO-01. Compensation is only allowed to release journal and budget state + * after the physical effect is confirmed. `EFFECT_UNKNOWN` is never a success. + */ +export type OpfsCleanupEffect = + | Readonly<{ kind: "CLEANED" | "ALREADY_CLEAN" }> + | Readonly<{ kind: "EFFECT_UNKNOWN" }>; + export type OpfsJournalMutation = "PUT" | "DELETE"; export type OpfsJournalPhase = | "PREPARING" diff --git a/tests/unit/opfs-byte-store.test.ts b/tests/unit/opfs-byte-store.test.ts index 6547ca5..fd84757 100644 --- a/tests/unit/opfs-byte-store.test.ts +++ b/tests/unit/opfs-byte-store.test.ts @@ -290,7 +290,10 @@ function createWorker( return browserDataSuccess(undefined); }, async cleanupTransaction() { - return browserDataSuccess(undefined); + return browserDataSuccess({ kind: "CLEANED" as const }); + }, + async abortPreparedPut() { + return browserDataSuccess({ kind: "CLEANED" as const }); }, async finalizePut() { return browserDataSuccess(undefined); @@ -501,6 +504,176 @@ describe("OPFS byte-store coordinator", () => { expect(replacementWorkerMethod).not.toHaveBeenCalled(); }); + it("keeps PREPARING journal when compensating cleanup is aborted or unavailable", async () => { + for (const failing of [ + { + async abortPreparedPut() { + return browserDataFailure("ABORTED", "OBJECT_RECONCILE"); + }, + }, + { + async abortPreparedPut() { + return browserDataSuccess({ kind: "EFFECT_UNKNOWN" as const }); + }, + }, + ]) { + const journal = createJournal(); + const worker = createWorker({ + async preparePut() { + return browserDataFailure("QUOTA_EXCEEDED", "OBJECT_WRITE"); + }, + ...failing, + }); + const adapter = createOpfsByteStoreAdapter({ + journal, + worker, + scope, + storagePolicy, + createTransactionId: () => "transaction_12345678", + now: () => 100, + }); + + const result = await adapter.objects.put({ + objectId: "object_12345678", + expectedGeneration: null, + mediaType: "application/octet-stream", + source: sourceFrom(new Uint8Array([1])), + }); + + expect(result.ok).toBe(false); + // The journal row is the only durable evidence that staging bytes may + // still exist, so it survives an unconfirmed compensation. + expect( + journal.transactions.get("transaction_12345678")?.phase, + ).toBe("PREPARING"); + } + }); + + it("does not roll back journal after an unknown worker mutation effect", async () => { + const journal = createJournal(); + const worker = createWorker({ + async abortPreparedPut() { + return browserDataSuccess({ kind: "EFFECT_UNKNOWN" as const }); + }, + }); + const rollback = vi.spyOn(journal, "rollback"); + vi.spyOn(journal, "markFilesReady").mockResolvedValueOnce( + browserDataFailure("UNAVAILABLE", "OBJECT_WRITE"), + ); + const adapter = createOpfsByteStoreAdapter({ + journal, + worker, + scope, + storagePolicy, + createTransactionId: () => "transaction_12345678", + now: () => 100, + }); + + const result = await adapter.objects.put({ + objectId: "object_12345678", + expectedGeneration: null, + mediaType: "application/octet-stream", + source: sourceFrom(new Uint8Array([1])), + }); + + expect(result.ok).toBe(false); + expect(rollback).not.toHaveBeenCalled(); + }); + + it("holds the OPFS mutation lease until exact physical cleanup completes", async () => { + const journal = createJournal(); + const observed: Array>> = []; + const worker = createWorker({ + async preparePut() { + return browserDataFailure("QUOTA_EXCEEDED", "OBJECT_WRITE"); + }, + async abortPreparedPut(request) { + observed.push({ ...request }); + return browserDataSuccess({ kind: "CLEANED" as const }); + }, + }); + const compensation = new AbortController(); + const caller = new AbortController(); + caller.abort(); + const adapter = createOpfsByteStoreAdapter({ + journal, + worker, + scope, + storagePolicy, + createTransactionId: () => "transaction_12345678", + createPhysicalGenerationId: () => "f".repeat(32) as never, + compensationSignal: compensation.signal, + now: () => 100, + }); + + await adapter.objects.put({ + objectId: "object_12345678", + expectedGeneration: null, + mediaType: "application/octet-stream", + source: sourceFrom(new Uint8Array([1])), + }); + + expect(observed).toHaveLength(1); + expect(observed[0]).toMatchObject({ + transactionId: "transaction_12345678", + physicalGenerationId: "f".repeat(32), + }); + // Compensation never inherits the caller signal. + expect(observed[0]?.signal).toBe(compensation.signal); + expect( + journal.transactions.has("transaction_12345678"), + ).toBe(false); + }); + + it("delayed stale cleanup cannot delete a reused logical generation", async () => { + const journal = createJournal(); + const issued: string[] = []; + let nextToken = 0; + const worker = createWorker({ + async abortPreparedPut(request) { + issued.push(request.physicalGenerationId); + return browserDataSuccess({ kind: "CLEANED" as const }); + }, + }); + let transaction = 0; + // T1 abandons its prepared put at the same logical generation 1. + vi.spyOn(journal, "markFilesReady").mockResolvedValueOnce( + browserDataFailure("UNAVAILABLE", "OBJECT_WRITE"), + ); + const adapter = createOpfsByteStoreAdapter({ + journal, + worker, + scope, + storagePolicy, + createTransactionId: () => `transaction_1234567${(transaction += 1)}`, + createPhysicalGenerationId: () => + String(nextToken += 1).padStart(32, "0") as never, + now: () => 100, + }); + const first = await adapter.objects.put({ + objectId: "object_12345678", + expectedGeneration: null, + mediaType: "application/octet-stream", + source: sourceFrom(new Uint8Array([1])), + }); + expect(first.ok).toBe(false); + + // T2 legitimately reuses logical generation 1 with a different token. + const second = await adapter.objects.put({ + objectId: "object_12345678", + expectedGeneration: null, + mediaType: "application/octet-stream", + source: sourceFrom(new Uint8Array([1])), + }); + expect(second.ok).toBe(true); + if (second.ok) expect(second.value.generation).toBe(1); + + // The stale compensation targeted only T1's physical token. + expect(issued).toEqual([String(1).padStart(32, "0")]); + const stored = journal.objects.get("object_12345678"); + expect(stored?.physicalSchemaVersion).toBe(1); + }); + it("keeps a committed journal row for reconciliation when cleanup fails", async () => { const journal = createJournal(); const worker = createWorker({ diff --git a/tests/unit/opfs-worker-runtime.test.ts b/tests/unit/opfs-worker-runtime.test.ts index ffc1ee7..8280b00 100644 --- a/tests/unit/opfs-worker-runtime.test.ts +++ b/tests/unit/opfs-worker-runtime.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import type { OpfsPreparedObject, + OpfsPhysicalGenerationId, OpfsStorageScope, } from "../../src/application/ports/browser-file-storage/opfs-ports.ts"; import type { @@ -170,10 +171,13 @@ function notFound(): DOMException { return new DOMException("Entry was not found.", "NotFoundError"); } +const PHYSICAL_GENERATION_A = "a".repeat(32) as OpfsPhysicalGenerationId; + function beginRequest( requestId: string, transactionId: string, scope: OpfsStorageScope, + physicalGenerationId: OpfsPhysicalGenerationId = PHYSICAL_GENERATION_A, ): OpfsWorkerRequest { return { requestId, @@ -182,6 +186,7 @@ function beginRequest( scope, objectId: "object_12345678", generation: 1, + physicalGenerationId, declaredByteLength: 1, mediaType: "application/octet-stream", createdAtEpochMs: 100,