donghyeon.kang/clean-architecture-frontend-template
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix: preserve OPFS recovery authority during cleanup

Browse Source 
Repair the compensating half of the OPFS put saga.

The coordinator now owns a single abortPreparedPut() driven by a
composition-owned bounded signal instead of the caller's already aborted one,
and the worker client no longer issues a duplicate fire-and-forget abort.
Journal rows and budget reservations are released only after the physical
effect is confirmed CLEANED or ALREADY_CLEAN; a timeout, malformed response or
EFFECT_UNKNOWN keeps PREPARING/FILES_READY and returns OBJECT_RECONCILE.

New writes carry a transaction-unique physicalGenerationId through the staging
receipt, manifest path and prepared object, so a late compensation deletes only
its own transaction's directory even when a newer transaction legitimately
reuses the same logical generation. v1 paths, receipts and prepared objects stay
readable through the rollback window.

Abort and cleanup hold the origin mutation lease through physical deletion and
staging removal. A transaction that never reached staging returns ALREADY_CLEAN
without waiting for the lease, which would otherwise deadlock against the BEGIN
it is cancelling.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
DongHyeonka
2026-08-13 23:21:58 +09:00
co-authored by Claude Opus 5
parent 6d1e44f206
commit 618da9abf5
11 changed files with 678 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapters/storage/opfs/opfs-worker-runtime.ts
+238 -54
View File
@@ -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<void> {
physicalGenerationId?: OpfsPhysicalGenerationId,
): Promise<OpfsCleanupEffect> {
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<Value>(
@@ -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<void> {
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<void> {
physicalGenerationId?: OpfsPhysicalGenerationId,
): Promise<OpfsCleanupEffect> {
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<OpfsCleanupEffect> {
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<string, unknown>;
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<void> {
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/<prefix>/<id>/<generation>/`, which two
* transactions can legitimately share. v2 writes
* `objects/<prefix>/<id>/g<generation>-<token>/` 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<string, unknown>;
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<string, unknown>;
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