From 0dcbe78e920b8912a2cfc4a1d7e2d007e8cd7965 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Wed, 2 Sep 2026 12:17:44 -0600 Subject: [PATCH] fix(publication): authenticate rotation handoffs --- scripts/lib/pylon-bounded-file.mjs | 80 +++- scripts/lib/pylon-consumer-lock.mjs | 484 +++++++++++++++++--- scripts/pylon-publication.test.mjs | 680 +++++++++++++++++++++++++++- 3 files changed, 1163 insertions(+), 81 deletions(-) diff --git a/scripts/lib/pylon-bounded-file.mjs b/scripts/lib/pylon-bounded-file.mjs index 2932290b1a..3489e7a94c 100644 --- a/scripts/lib/pylon-bounded-file.mjs +++ b/scripts/lib/pylon-bounded-file.mjs @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import { closeSync, constants, @@ -11,10 +12,33 @@ export const PYLON_PUBLICATION_MANIFEST_MAX_BYTES = 64 * 1024; export const PYLON_STABLE_HISTORY_MAX_MANIFESTS = 4096; export const PYLON_STABLE_HISTORY_MAX_BYTES = 32 * 1024 * 1024; +export class BoundedFileUnlinkedDuringReadError extends Error { + constructor(path, description, bytes, expectedSha256) { + super(`${description} changed while it was read because the same opened inode was removed.`); + this.name = "BoundedFileUnlinkedDuringReadError"; + this.path = path; + this.description = description; + this.bytes = Buffer.from(bytes); + this.expectedSha256 = expectedSha256; + } +} -function sameStat(left, right) { +function sameInodeReadBounds(left, right) { return left.dev === right.dev && left.ino === right.ino && left.size === right.size && - left.mtimeMs === right.mtimeMs && left.ctimeMs === right.ctimeMs; + left.mtimeMs === right.mtimeMs; +} + +function sameStat(left, right) { + return sameInodeReadBounds(left, right) && left.ctimeMs === right.ctimeMs && left.nlink === right.nlink; +} + +function isPinnedHandleRemoval(pathEntry, before, after, extraBytes, finalPathMissing) { + return finalPathMissing && extraBytes === 0 && sameStat(pathEntry, before) && + sameInodeReadBounds(before, after) && before.nlink > 0 && after.nlink === 0; +} + +function exactSha256(bytes, expectedSha256) { + return expectedSha256 !== null && createHash("sha256").update(bytes).digest("hex") === expectedSha256; } export async function readBoundedRegularFile( @@ -27,9 +51,13 @@ export async function readBoundedRegularFile( lstatEntry = lstat, validateHandle, hooks, + expectedSha256 = null, } = {}, ) { - if (!Number.isSafeInteger(maxBytes) || maxBytes < 1 || !Number.isSafeInteger(minBytes) || minBytes < 0 || minBytes > maxBytes) { + if ( + !Number.isSafeInteger(maxBytes) || maxBytes < 1 || !Number.isSafeInteger(minBytes) || minBytes < 0 || minBytes > maxBytes || + !(expectedSha256 === null || /^[0-9a-f]{64}$/.test(expectedSha256)) + ) { throw new Error("Bounded file limits are invalid."); } let pathEntry; @@ -56,6 +84,7 @@ export async function readBoundedRegularFile( let before = await handle.stat(); if (!before.isFile()) throw new Error(`${description} is not one regular non-symlink file.`); if (validateHandle) before = await validateHandle(handle, before, description); + if (!sameStat(pathEntry, before)) throw new Error(`${description} changed while it was read.`); if (before.size < minBytes || before.size > maxBytes) throw new Error(`${description} exceeds its format byte limit or is malformed.`); await hooks?.afterInitialStat?.({ path, handle, stat: before }); const bytes = Buffer.alloc(before.size); @@ -69,7 +98,24 @@ export async function readBoundedRegularFile( const { bytesRead: extraBytes } = await handle.read(extra, 0, 1, bytes.length); await hooks?.beforeFinalStat?.({ path, handle, bytes }); const after = await handle.stat(); - if (extraBytes !== 0 || !sameStat(before, after)) throw new Error(`${description} changed while it was read.`); + let finalPathEntry; + let finalPathMissing = false; + try { + finalPathEntry = await lstatEntry(path); + } catch (error) { + if (error?.code !== "ENOENT") throw error; + finalPathMissing = true; + } + if ( + isPinnedHandleRemoval(pathEntry, before, after, extraBytes, finalPathMissing) && + exactSha256(bytes, expectedSha256) + ) { + throw new BoundedFileUnlinkedDuringReadError(path, description, bytes, expectedSha256); + } + if ( + extraBytes !== 0 || finalPathMissing || finalPathEntry.isSymbolicLink?.() || !finalPathEntry.isFile() || + !sameStat(before, after) || !sameStat(after, finalPathEntry) + ) throw new Error(`${description} changed while it was read.`); return bytes; } finally { await handle.close(); @@ -89,9 +135,13 @@ export function readBoundedRegularFileSync( readFile = readSync, closeFile = closeSync, hooks, + expectedSha256 = null, } = {}, ) { - if (!Number.isSafeInteger(maxBytes) || maxBytes < 1 || !Number.isSafeInteger(minBytes) || minBytes < 0 || minBytes > maxBytes) { + if ( + !Number.isSafeInteger(maxBytes) || maxBytes < 1 || !Number.isSafeInteger(minBytes) || minBytes < 0 || minBytes > maxBytes || + !(expectedSha256 === null || /^[0-9a-f]{64}$/.test(expectedSha256)) + ) { throw new Error("Bounded file limits are invalid."); } let pathEntry; @@ -115,6 +165,7 @@ export function readBoundedRegularFileSync( try { const before = statFile(descriptor); if (!before.isFile()) throw new Error(`${description} is not one regular non-symlink file.`); + if (!sameStat(pathEntry, before)) throw new Error(`${description} changed while it was read.`); if (before.size < minBytes || before.size > maxBytes) throw new Error(`${description} exceeds its format byte limit or is malformed.`); hooks?.afterInitialStat?.({ path, descriptor, stat: before }); const bytes = Buffer.alloc(before.size); @@ -128,7 +179,24 @@ export function readBoundedRegularFileSync( const extraBytes = readFile(descriptor, extra, 0, 1, bytes.length); hooks?.beforeFinalStat?.({ path, descriptor, bytes }); const after = statFile(descriptor); - if (extraBytes !== 0 || !sameStat(before, after)) throw new Error(`${description} changed while it was read.`); + let finalPathEntry; + let finalPathMissing = false; + try { + finalPathEntry = lstatEntry(path); + } catch (error) { + if (error?.code !== "ENOENT") throw error; + finalPathMissing = true; + } + if ( + isPinnedHandleRemoval(pathEntry, before, after, extraBytes, finalPathMissing) && + exactSha256(bytes, expectedSha256) + ) { + throw new BoundedFileUnlinkedDuringReadError(path, description, bytes, expectedSha256); + } + if ( + extraBytes !== 0 || finalPathMissing || finalPathEntry.isSymbolicLink?.() || !finalPathEntry.isFile() || + !sameStat(before, after) || !sameStat(after, finalPathEntry) + ) throw new Error(`${description} changed while it was read.`); return bytes; } finally { closeFile(descriptor); diff --git a/scripts/lib/pylon-consumer-lock.mjs b/scripts/lib/pylon-consumer-lock.mjs index 384b671ad3..04aafcb7d0 100644 --- a/scripts/lib/pylon-consumer-lock.mjs +++ b/scripts/lib/pylon-consumer-lock.mjs @@ -3,13 +3,24 @@ import { constants } from "node:fs"; import { link, lstat, mkdir, open, readdir, rename, rm } from "node:fs/promises"; import { basename, dirname, join, parse, relative, resolve, sep } from "node:path"; -import { readBoundedRegularFile } from "./pylon-bounded-file.mjs"; +import { + BoundedFileUnlinkedDuringReadError, + readBoundedRegularFile, +} from "./pylon-bounded-file.mjs"; + +class ConsumerEpochAdvancedError extends Error { + constructor() { + super("Consumer high-water journal epoch changed and fenced a paused writer."); + this.name = "ConsumerEpochAdvancedError"; + } +} export const PYLON_CONSUMER_LOCK_STALE_MS = 30_000; export const PYLON_CONSUMER_LOCK_UPDATE_MS = 10_000; export const PYLON_CONSUMER_ROTATE_CLAIM_TRIGGER = 60_000; export const PYLON_CONSUMER_ROTATE_TRANSITION_TRIGGER = 3_800; const LOCK_SCHEMA_VERSION = 2; +const CLAIM_INDEX_SCHEMA_VERSION = 1; const LEGACY_LOCK_SCHEMA_VERSION = 1; const TRANSACTION_SCHEMA_VERSION = 1; const CHECKPOINT_SCHEMA_VERSION = 2; @@ -31,7 +42,9 @@ const TEMPORARY_DIRECTORY_NAME = ".owned-temporaries-v2"; const LEGACY_RETIREMENT_MARKER_NAME = ".pylon-consumer-v1-retired.json"; const uuidSource = "[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}"; const uuidPattern = new RegExp(`^${uuidSource}$`); -const claimPattern = /^claim-([0-9]{16})\.json$/; +const claimPattern = /^claim-([0-9]{16})-([0-9a-f]{64})\.json$/; +const claimIndexPattern = /^claim-index-([0-9]{16})\.json$/; +const undigestedClaimPattern = /^claim-([0-9]{16})\.json$/; const transitionPattern = /^transition-([0-9a-f]{64})\.json$/; const legacyTransitionPattern = /^([0-9a-f]{64})\.json$/; const checkpointPattern = new RegExp(`^checkpoint-([0-9]{16})-(${uuidSource})\\.json$`); @@ -76,8 +89,12 @@ function epochName(checkpoint) { return `epoch-${generationName(checkpoint.epoch)}-${checkpoint.epochId}`; } -function claimPath(context, generation) { - return join(context.epochDirectory, `claim-${generationName(generation)}.json`); +function claimPath(context, claim) { + return join(context.epochDirectory, `claim-${generationName(claim.generation)}-${digest(metadataBytes(claim))}.json`); +} + +function claimIndexPath(context, generation) { + return join(context.epochDirectory, `claim-index-${generationName(generation)}.json`); } function heartbeatPath(context, claim) { @@ -120,6 +137,23 @@ function validateClaim(value, context, stateMaxBytes) { return value; } +function claimIndexFor(claim) { + return { + schemaVersion: CLAIM_INDEX_SCHEMA_VERSION, + generation: claim.generation, + claimSha256: digest(metadataBytes(claim)), + }; +} + +function validateClaimIndex(value, generation) { + if ( + !exactKeys(value, ["schemaVersion", "generation", "claimSha256"]) || + value.schemaVersion !== CLAIM_INDEX_SCHEMA_VERSION || value.generation !== generation || + !/^[0-9a-f]{64}$/.test(value.claimSha256 ?? "") + ) throw new Error("Consumer high-water claim index is malformed."); + return value; +} + function validateHeartbeat(value, claim) { if ( claim.type !== "normal" || @@ -431,7 +465,7 @@ async function ensureDirectory(path, description, options) { await options.syncDirectory(dirname(path)); } -async function readSecureFile(path, maxBytes, description, options, minBytes = 1, hooks) { +async function readSecureFile(path, maxBytes, description, options, minBytes = 1, hooks, expectedSha256 = null) { return readBoundedRegularFile(path, { maxBytes, minBytes, @@ -439,12 +473,21 @@ async function readSecureFile(path, maxBytes, description, options, minBytes = 1 openFile: options.openFile, lstatEntry: options.lstatEntry, hooks, + expectedSha256, validateHandle: (handle, stat) => secureHandle(handle, stat, description, "file", options), }); } -async function readExactMetadata(path, maxBytes, validate, description, options, budget) { - const bytes = await readSecureFile(path, maxBytes, description, options); +async function readExactMetadata(path, maxBytes, validate, description, options, budget, expectedSha256 = null) { + const bytes = await readSecureFile( + path, + maxBytes, + description, + options, + 1, + options.hooks?.metadataRead, + expectedSha256, + ); if (bytes === null) return null; if (budget) { budget.bytes += bytes.length; @@ -495,7 +538,7 @@ async function inspectTemporary(path, options) { } const kind = match[6]; const allowedKinds = new Set([ - "checkpoint", "projection", "transition", "claim", "initial-heartbeat", "heartbeat", + "checkpoint", "projection", "transition", "claim", "claim-index", "initial-heartbeat", "heartbeat", "terminal-released", "terminal-retired", "terminal-commit", "applied", "legacy-guard", "legacy-retirement", ]); @@ -512,7 +555,157 @@ async function inspectTemporary(path, options) { }; } -async function revalidateAuthority(context, operation, options) { +function isImmediateSuccessorCheckpoint(context, checkpoint) { + return checkpoint.epoch === context.checkpoint.epoch + 1 && + checkpoint.epochId === deterministicUuid( + `pylon-consumer-rotation-v2:${context.checkpointDigest}:${checkpoint.anchorDigest}`, + ) && + checkpoint.previousCheckpointSha256 === context.checkpointDigest && + checkpoint.previousTipSha256 === checkpoint.anchorDigest && + checkpoint.retiredEpochDirectory === basename(context.epochDirectory) && + checkpoint.sourceAuthoritySha256 === context.checkpoint.sourceAuthoritySha256 && + checkpoint.sourceAuthorityTipDigest === context.checkpoint.sourceAuthorityTipDigest && + checkpoint.sourceAuthorityTipBase64 === context.checkpoint.sourceAuthorityTipBase64 && + checkpoint.historySha256 === digest(Buffer.from( + `${context.checkpoint.historySha256}:${context.checkpointDigest}:${checkpoint.anchorDigest}`, + )); +} + +async function confirmAuthenticatedSuccessorRoot(scan, context, options, invalidRoot) { + const rootNames = new Set([ + TEMPORARY_DIRECTORY_NAME, + ...scan.checkpointEntries.map((entry) => entry.name), + ...scan.epochEntries.map((entry) => entry.name), + ...scan.temporaries + .filter((temporary) => dirname(temporary.path) === context.journalDirectory) + .map((temporary) => basename(temporary.path)), + ]); + const headEpoch = scan.head.checkpoint.epoch; + const optionalNames = new Set([ + ...scan.checkpointEntries.filter((entry) => entry.checkpoint.epoch < headEpoch).map((entry) => entry.name), + ...scan.epochEntries.filter((entry) => entry.epoch < headEpoch).map((entry) => entry.name), + ...scan.temporaries + .filter((temporary) => dirname(temporary.path) === context.journalDirectory) + .map((temporary) => basename(temporary.path)), + ]); + const sameRootAuthority = (names) => { + const actual = new Set(names); + return [...actual].every((name) => rootNames.has(name)) && + [...rootNames].every((name) => optionalNames.has(name) || actual.has(name)); + }; + if (!sameRootAuthority(await options.readDirectory(context.journalDirectory))) throw invalidRoot(); + for (const entry of scan.checkpointEntries) { + let checkpoint; + try { + checkpoint = await readExactMetadata( + entry.path, + options.metadataMaxBytes, + (value) => validateCheckpoint(value, options.stateMaxBytes).value, + "Consumer high-water journal checkpoint", + options, + undefined, + entry.digest, + ); + } catch (error) { + if (optionalNames.has(entry.name) && error instanceof BoundedFileUnlinkedDuringReadError) continue; + throw error; + } + if (checkpoint === null && optionalNames.has(entry.name)) continue; + if (checkpoint === null || digest(metadataBytes(checkpoint)) !== entry.digest) throw invalidRoot(); + } + for (const entry of scan.epochEntries) { + try { + await secureDirectory(entry.path, "Consumer high-water epoch directory", options); + } catch (error) { + if (optionalNames.has(entry.name) && error?.code === "ENOENT") continue; + throw error; + } + } + if (!sameRootAuthority(await options.readDirectory(context.journalDirectory))) throw invalidRoot(); +} + +async function authenticateChangedRoot( + context, + options, + inProgressCheckpoint = null, + allowInProgressDiscovery = false, +) { + const scan = await scanJournalRoot(context.statePath, context.journalDirectory, options, 0, true); + const invalidRoot = () => new Error( + "Consumer high-water journal root changed without one exact current or immediate-successor authority.", + ); + const currentCheckpoint = scan.checkpointEntries.find((entry) => entry.path === context.checkpointPath); + if (currentCheckpoint && currentCheckpoint.digest !== context.checkpointDigest) throw invalidRoot(); + + if (inProgressCheckpoint !== null && scan.checkpointEntries.length === 1) { + const checkpoint = validateCheckpoint(inProgressCheckpoint, options.stateMaxBytes).value; + const nextEpochPath = join(context.journalDirectory, epochName(checkpoint)); + if ( + !isImmediateSuccessorCheckpoint(context, checkpoint) || + scan.checkpointEntries.length !== 1 || scan.head?.path !== context.checkpointPath || scan.missingHeadEpoch || + scan.epochEntries.length !== 2 || + scan.epochEntries.some((entry) => ![context.epochDirectory, nextEpochPath].includes(entry.path)) || + !scan.epochEntries.some((entry) => entry.path === nextEpochPath) || + (await options.readDirectory(nextEpochPath)).length !== 0 + ) throw invalidRoot(); + if ((await options.readDirectory(nextEpochPath)).length !== 0) throw invalidRoot(); + return false; + } + + const discoveredNextEpoch = scan.epochEntries.find((entry) => entry.path !== context.epochDirectory); + if ( + allowInProgressDiscovery && inProgressCheckpoint === null && + scan.checkpointEntries.length === 1 && scan.head?.path === context.checkpointPath && !scan.missingHeadEpoch && + scan.epochEntries.length === 2 && discoveredNextEpoch?.epoch === context.checkpoint.epoch + 1 && + (await options.readDirectory(discoveredNextEpoch.path)).length === 0 + ) { + if ((await options.readDirectory(discoveredNextEpoch.path)).length !== 0) throw invalidRoot(); + return discoveredNextEpoch.path; + } + + const retainedCheckpoint = scan.checkpointEntries.find((entry) => entry.path !== context.checkpointPath); + const retiredEpochPath = context.checkpoint.retiredEpochDirectory === null + ? null + : join(context.journalDirectory, context.checkpoint.retiredEpochDirectory); + if ( + currentCheckpoint && scan.head?.path === context.checkpointPath && !scan.missingHeadEpoch && + scan.checkpointEntries.length <= 2 && scan.epochEntries.length <= 2 && + (!retainedCheckpoint || ( + retainedCheckpoint.digest === context.checkpoint.previousCheckpointSha256 && + epochName(retainedCheckpoint.checkpoint) === context.checkpoint.retiredEpochDirectory + )) && + scan.epochEntries.every((entry) => [context.epochDirectory, retiredEpochPath].includes(entry.path)) + ) { + return false; + } + + const successor = scan.head; + const expectedCheckpointPath = successor + ? join(context.journalDirectory, checkpointName(successor.checkpoint)) + : null; + const expectedEpochPath = successor + ? join(context.journalDirectory, epochName(successor.checkpoint)) + : null; + if ( + !successor || scan.missingHeadEpoch || successor.path !== expectedCheckpointPath || + !isImmediateSuccessorCheckpoint(context, successor.checkpoint) || + scan.checkpointEntries.length < 1 || scan.checkpointEntries.length > 2 || + scan.epochEntries.length < 1 || scan.epochEntries.length > 2 || + scan.checkpointEntries.some((entry) => ![context.checkpointPath, expectedCheckpointPath].includes(entry.path)) || + scan.epochEntries.some((entry) => ![context.epochDirectory, expectedEpochPath].includes(entry.path)) || + !scan.epochEntries.some((entry) => entry.path === expectedEpochPath) + ) throw invalidRoot(); + await confirmAuthenticatedSuccessorRoot(scan, context, options, invalidRoot); + return true; +} + +async function revalidateAuthority( + context, + operation, + options, + inProgressCheckpoint = options.inProgressCheckpoint ?? null, + allowInProgressDiscovery = false, +) { await options.hooks?.beforePathOperation?.({ operation, statePath: context.statePath, @@ -526,17 +719,33 @@ async function revalidateAuthority(context, operation, options) { await secureDirectory(dirname(context.statePath), "Consumer high-water state directory", options); await secureDirectory(context.journalDirectory, "Consumer high-water journal directory", options); await secureDirectory(context.temporaryDirectory, "Consumer high-water temporary directory", options); - await secureDirectory(context.epochDirectory, "Consumer high-water epoch directory", options); + let oldEpochError = null; + try { + await secureDirectory(context.epochDirectory, "Consumer high-water epoch directory", options); + } catch (error) { + if (error?.code !== "ENOENT") throw error; + oldEpochError = error; + } const entries = await options.readDirectory(context.journalDirectory); if (entries.length > MAX_JOURNAL_ROOT_ENTRIES + MAX_TEMPORARY_ENTRIES) { throw new Error("Consumer high-water journal root exceeds its safe allocation bound."); } - const checkpoints = entries.map((name) => ({ name, match: checkpointPattern.exec(name) })).filter((entry) => entry.match); - if (checkpoints.length < 1 || checkpoints.length > 2) throw new Error("Consumer high-water journal checkpoint set is malformed."); - checkpoints.sort((left, right) => Number(left.match[1]) - Number(right.match[1])); - if (checkpoints.at(-1).name !== basename(context.checkpointPath)) { - throw new Error("Consumer high-water journal epoch changed and fenced a paused writer."); + const expectedRootNames = new Set([ + TEMPORARY_DIRECTORY_NAME, + basename(context.checkpointPath), + basename(context.epochDirectory), + ]); + let changedRoot = false; + if (entries.some((name) => !expectedRootNames.has(name))) { + changedRoot = await authenticateChangedRoot( + context, + options, + inProgressCheckpoint, + allowInProgressDiscovery, + ); + if (changedRoot === true) throw new ConsumerEpochAdvancedError(); } + if (oldEpochError) throw oldEpochError; const current = await readExactMetadata( context.checkpointPath, options.metadataMaxBytes, @@ -547,6 +756,7 @@ async function revalidateAuthority(context, operation, options) { if (digest(metadataBytes(current)) !== context.checkpointDigest) { throw new Error("Consumer high-water journal checkpoint changed and fenced a paused writer."); } + return typeof changedRoot === "string" ? changedRoot : null; } async function publishImmutable({ @@ -559,8 +769,9 @@ async function publishImmutable({ options, revalidate = true, beforeLink, + inProgressCheckpoint = null, }) { - if (revalidate) await revalidateAuthority(context, kind, options); + if (revalidate) await revalidateAuthority(context, kind, options, inProgressCheckpoint); const temporary = join(context.temporaryDirectory, temporaryName(path, kind, writer, context)); let handle; let linked = false; @@ -573,7 +784,7 @@ async function publishImmutable({ handle = undefined; await options.hooks?.afterFileSync?.({ kind, path, temporary }); await beforeLink?.(); - if (revalidate) await revalidateAuthority(context, `${kind}-link`, options); + if (revalidate) await revalidateAuthority(context, `${kind}-link`, options, inProgressCheckpoint); try { await options.linkFile(temporary, path); linked = true; @@ -631,7 +842,13 @@ function genesisCheckpoint(statePath) { }; } -async function scanJournalRoot(statePath, journalDirectory, options, replacementRetries = PROJECTION_RETRY_LIMIT) { +async function scanJournalRoot( + statePath, + journalDirectory, + options, + replacementRetries = PROJECTION_RETRY_LIMIT, + allowVanishedEarlierEntries = false, +) { await secureDirectory(journalDirectory, "Consumer high-water journal directory", options); await options.syncDirectory(journalDirectory); const names = await options.readDirectory(journalDirectory); @@ -678,8 +895,9 @@ async function scanJournalRoot(statePath, journalDirectory, options, replacement return match && Number(match[1]) > epoch; }); if (hasLaterCheckpoint && replacementRetries > 0) { - return scanJournalRoot(statePath, journalDirectory, options, replacementRetries - 1); + return scanJournalRoot(statePath, journalDirectory, options, replacementRetries - 1, allowVanishedEarlierEntries); } + if (hasLaterCheckpoint && allowVanishedEarlierEntries) continue; throw new Error("Consumer high-water journal lost its current checkpoint during an authenticated scan."); } if (checkpointName(checkpoint) !== name || checkpoint.epoch !== Number(checkpointMatch[1])) { @@ -700,8 +918,9 @@ async function scanJournalRoot(statePath, journalDirectory, options, replacement return match && Number(match[1]) > epoch; }); if (error?.code === "ENOENT" && hasLaterEpoch && replacementRetries > 0) { - return scanJournalRoot(statePath, journalDirectory, options, replacementRetries - 1); + return scanJournalRoot(statePath, journalDirectory, options, replacementRetries - 1, allowVanishedEarlierEntries); } + if (error?.code === "ENOENT" && hasLaterEpoch && allowVanishedEarlierEntries) continue; throw error; } if (!entry.isDirectory() || entry.isSymbolicLink?.()) throw new Error("Consumer high-water epoch entry must be one real directory."); @@ -714,8 +933,9 @@ async function scanJournalRoot(statePath, journalDirectory, options, replacement return match && Number(match[1]) > epoch; }); if (error?.code === "ENOENT" && hasLaterEpoch && replacementRetries > 0) { - return scanJournalRoot(statePath, journalDirectory, options, replacementRetries - 1); + return scanJournalRoot(statePath, journalDirectory, options, replacementRetries - 1, allowVanishedEarlierEntries); } + if (error?.code === "ENOENT" && hasLaterEpoch && allowVanishedEarlierEntries) continue; throw error; } epochEntries.push({ name, path, epoch: Number(epochMatch[1]), epochId: epochMatch[2] }); @@ -900,10 +1120,7 @@ function isProjectionReplacementTransient(error) { } function isCommitHelperReplacementTransient(error) { - return isProjectionReplacementTransient(error) || [ - "Consumer high-water journal epoch changed and fenced a paused writer.", - "Consumer high-water journal checkpoint changed and fenced a paused writer.", - ].includes(error?.message); + return isProjectionReplacementTransient(error) || error instanceof ConsumerEpochAdvancedError; } async function repairProjection(context, initialTip, options, writer = options.activeWriter) { @@ -961,13 +1178,21 @@ async function publishTransition(context, transaction, claim, options) { } async function scanEpoch(context, options) { - await revalidateAuthority(context, "scan-claims", options); + const discoveredNextEpoch = await revalidateAuthority( + context, + "scan-claims", + options, + options.inProgressCheckpoint ?? null, + true, + ); await options.syncDirectory(context.epochDirectory); const names = await options.readDirectory(context.epochDirectory); if (names.length > options.maxJournalEntries + MAX_TEMPORARY_ENTRIES) { throw new Error("Consumer high-water epoch exceeds its safe allocation bound."); } - const claimNames = new Map(); + const claimContentNames = new Map(); + const claimIndexNames = new Map(); + const legacyClaimNames = new Map(); const heartbeatNames = new Map(); const terminalNames = new Map(); const appliedNames = new Map(); @@ -976,8 +1201,20 @@ async function scanEpoch(context, options) { for (const name of names) { let match; if ((match = claimPattern.exec(name))) { - if (claimNames.has(Number(match[1]))) throw new Error("Consumer high-water lock contains a duplicate claim."); - claimNames.set(Number(match[1]), name); + const generation = Number(match[1]); + const contents = claimContentNames.get(generation) ?? new Map(); + contents.set(match[2], name); + claimContentNames.set(generation, contents); + authoritativeEntryCount += 1; + } else if ((match = claimIndexPattern.exec(name))) { + const generation = Number(match[1]); + if (claimIndexNames.has(generation)) throw new Error("Consumer high-water lock contains a duplicate claim index."); + claimIndexNames.set(generation, name); + authoritativeEntryCount += 1; + } else if ((match = undigestedClaimPattern.exec(name))) { + const generation = Number(match[1]); + if (legacyClaimNames.has(generation)) throw new Error("Consumer high-water lock contains a duplicate legacy claim."); + legacyClaimNames.set(generation, name); authoritativeEntryCount += 1; } else if ((match = heartbeatPattern.exec(name))) { heartbeatNames.set(`${Number(match[1])}:${match[2]}`, name); @@ -1007,21 +1244,72 @@ async function scanEpoch(context, options) { const budget = { bytes: 0 }; const claims = []; const byKey = new Map(); - for (const [generation, name] of [...claimNames].sort((left, right) => left[0] - right[0])) { - const claim = await readExactMetadata( - join(context.epochDirectory, name), - options.metadataMaxBytes, - (value) => validateClaim(value, context, options.stateMaxBytes), - "Consumer high-water operation claim", - options, - budget, - ); - if (claim.generation !== generation || name !== `claim-${generationName(generation)}.json`) { - throw new Error("Consumer high-water lock claim name differs from its exact generation."); + const referencedClaimContents = new Set(); + const generations = new Set([...claimIndexNames.keys(), ...legacyClaimNames.keys()]); + for (const generation of [...generations].sort((left, right) => left - right)) { + if (claimIndexNames.has(generation) && legacyClaimNames.has(generation)) { + throw new Error("Consumer high-water lock contains competing indexed and legacy claims."); + } + let claim; + if (claimIndexNames.has(generation)) { + const index = await readExactMetadata( + join(context.epochDirectory, claimIndexNames.get(generation)), + options.metadataMaxBytes, + (value) => validateClaimIndex(value, generation), + "Consumer high-water claim index", + options, + budget, + ); + const name = claimContentNames.get(generation)?.get(index.claimSha256); + if (!name) throw new Error("Consumer high-water claim index lacks its exact digest-bound claim bytes."); + referencedClaimContents.add(name); + claim = await readExactMetadata( + join(context.epochDirectory, name), + options.metadataMaxBytes, + (value) => validateClaim(value, context, options.stateMaxBytes), + "Consumer high-water operation claim", + options, + budget, + index.claimSha256, + ); + if ( + claim.generation !== generation || digest(metadataBytes(claim)) !== index.claimSha256 || + name !== basename(claimPath(context, claim)) + ) throw new Error("Consumer high-water claim index differs from its exact canonical claim bytes."); + } else { + const name = legacyClaimNames.get(generation); + claim = await readExactMetadata( + join(context.epochDirectory, name), + options.metadataMaxBytes, + (value) => validateClaim(value, context, options.stateMaxBytes), + "Consumer high-water legacy operation claim", + options, + budget, + ); + if (claim.generation !== generation || name !== `claim-${generationName(generation)}.json`) { + throw new Error("Consumer high-water legacy claim name differs from its exact generation."); + } } claims.push(claim); byKey.set(`${generation}:${claim.token}`, claim); } + for (const [generation, contents] of [...claimContentNames].sort((left, right) => left[0] - right[0])) { + for (const [claimSha256, name] of [...contents].sort((left, right) => left[0].localeCompare(right[0]))) { + if (referencedClaimContents.has(name)) continue; + const claim = await readExactMetadata( + join(context.epochDirectory, name), + options.metadataMaxBytes, + (value) => validateClaim(value, context, options.stateMaxBytes), + "Consumer high-water unindexed claim content", + options, + budget, + ); + if ( + claim.generation !== generation || digest(metadataBytes(claim)) !== claimSha256 || + name !== basename(claimPath(context, claim)) + ) throw new Error("Consumer high-water unindexed claim content differs from its exact canonical bytes."); + } + } if (claims.length > MAX_OPERATION_GENERATIONS) throw new Error("Consumer high-water operation generation bound is exhausted."); for (let index = 0; index < claims.length; index += 1) { if (claims[index].generation !== index + 1) throw new Error("Consumer high-water lock generations are not contiguous."); @@ -1073,6 +1361,20 @@ async function scanEpoch(context, options) { throw new Error("Consumer high-water operation generations crossed an unresolved earlier slot."); } } + if (discoveredNextEpoch !== null) { + const latest = claims.at(-1); + if (latest?.type !== "rotation") { + throw new Error("Consumer high-water in-progress next epoch lacks its exact published rotation intent."); + } + const intent = validateRotationIntent(latest.intent, context, options.stateMaxBytes); + if ( + !isImmediateSuccessorCheckpoint(context, intent.checkpoint) || + discoveredNextEpoch !== join(context.journalDirectory, epochName(intent.checkpoint)) + ) throw new Error("Consumer high-water in-progress next epoch differs from its exact published rotation intent."); + if (await authenticateChangedRoot(context, options, intent.checkpoint)) { + throw new ConsumerEpochAdvancedError(); + } + } return { claims, terminals, temporaries }; } @@ -1383,6 +1685,7 @@ async function finishRotationCheckpoint(context, checkpoint, writer, options) { context, writer, options, + inProgressCheckpoint: checkpoint, beforeLink: () => scanRotationPublicationSet(context, checkpoint, options, false), }); await options.hooks?.afterRotationCheckpoint?.({ checkpoint: structuredClone(checkpoint), nextPath }); @@ -1438,6 +1741,27 @@ async function resolveOperationFrontier(context, options) { return { scan, frontier: latest, rotated: false, active: false }; } +async function tryPublishClaim(context, claim, options) { + const contentPath = claimPath(context, claim); + const contentResult = await publishMetadata(contentPath, claim, "claim", context, claim, options); + const existingClaim = validateClaim(contentResult.value, context, options.stateMaxBytes); + if (!metadataBytes(existingClaim).equals(metadataBytes(claim))) { + throw new Error("Consumer high-water claim content lost its exact digest-bound publication."); + } + const index = claimIndexFor(claim); + const indexResult = await publishMetadata( + claimIndexPath(context, claim.generation), + index, + "claim-index", + context, + claim, + options, + ); + const existingIndex = validateClaimIndex(indexResult.value, claim.generation); + if (!metadataBytes(existingIndex).equals(metadataBytes(index))) return false; + return indexResult.created; +} + async function tryCreateNormalClaim(context, generation, options) { const claim = { schemaVersion: LOCK_SCHEMA_VERSION, @@ -1447,9 +1771,7 @@ async function tryCreateNormalClaim(context, generation, options) { ownerPid: process.pid, createdAtMs: options.now(), }; - const result = await publishMetadata(claimPath(context, generation), claim, "claim", context, claim, options); - if (!result.created) return null; - validateClaim(result.value, context, options.stateMaxBytes); + if (!(await tryPublishClaim(context, claim, options))) return null; const heartbeat = { schemaVersion: LOCK_SCHEMA_VERSION, generation, @@ -1464,9 +1786,7 @@ async function tryCreateNormalClaim(context, generation, options) { async function tryCreateRotationClaim(context, generation, tip, options) { const claim = rotationClaimFor(context, generation, tip); await options.hooks?.beforeRotationDecision?.({ intent: structuredClone(claim.intent), claim: structuredClone(claim) }); - const result = await publishMetadata(claimPath(context, generation), claim, "claim", context, claim, options); - if (!result.created) return null; - validateClaim(result.value, context, options.stateMaxBytes); + if (!(await tryPublishClaim(context, claim, options))) return null; await options.hooks?.afterRotationIntent?.({ intent: structuredClone(claim.intent), claim: structuredClone(claim) }); return claim; } @@ -1611,7 +1931,9 @@ async function cleanupAuthority( const temporary = await inspectTemporary(path, options); if (temporary) retiredTemporaries.push(temporary); } else if ( - !claimPattern.test(name) && !heartbeatPattern.test(name) && !terminalPattern.test(name) && + !claimPattern.test(name) && !claimIndexPattern.test(name) && !undigestedClaimPattern.test(name) && + !heartbeatPattern.test(name) && + !terminalPattern.test(name) && !appliedPattern.test(name) && !transitionPattern.test(name) ) { throw new Error("Consumer high-water retired epoch contains an unexpected entry."); @@ -1660,34 +1982,35 @@ async function cleanupAuthority( async function helpRotationOperation(context, claim, options) { if (claim.type !== "rotation") throw new Error("Consumer high-water rotation helper requires one rotation operation slot."); const intent = validateRotationIntent(claim.intent, context, options.stateMaxBytes); - const completedBeforeHelp = await completedRotationResult(context, intent, options).catch(() => null); + const helperOptions = { ...options, inProgressCheckpoint: intent.checkpoint }; + const completedBeforeHelp = await completedRotationResult(context, intent, helperOptions).catch(() => null); if (completedBeforeHelp) return true; try { - const scan = await scanEpoch(context, options); + const scan = await scanEpoch(context, helperOptions); const latest = scan.claims.at(-1); if (operationIdentity(latest) !== operationIdentity(claim) || !metadataBytes(latest).equals(metadataBytes(claim))) { throw new Error("Consumer high-water rotation operation is not the unique latest slot."); } - const tip = await effectiveTip(context, options); + const tip = await effectiveTip(context, helperOptions); if (tip.tipDigest !== intent.tipSha256) { throw new Error("Consumer high-water rotation operation no longer matches its exact authoritative tip."); } - const rootScan = await scanJournalRoot(context.statePath, context.journalDirectory, options); + const rootScan = await scanJournalRoot(context.statePath, context.journalDirectory, helperOptions); const nextEpochName = epochName(intent.checkpoint); await cleanupAuthority( context, claim, rootScan, scan.temporaries, - options, + helperOptions, true, scan, nextEpochName, ); - await finishRotationCheckpoint(context, intent.checkpoint, claim, options); + await finishRotationCheckpoint(context, intent.checkpoint, claim, helperOptions); return true; } catch (error) { - const completed = await completedRotationResult(context, intent, options).catch(() => null); + const completed = await completedRotationResult(context, intent, helperOptions).catch(() => null); if (completed) return true; throw error; } @@ -1792,7 +2115,7 @@ function normalizeOptions({ maxTransactionDepth, maxLockGenerations, maxJournalBytes, - maxJournalEntries: MAX_OPERATION_GENERATIONS * 4 + MAX_TRANSACTION_DEPTH + 32, + maxJournalEntries: MAX_OPERATION_GENERATIONS * 5 + MAX_TRANSACTION_DEPTH + 32, metadataMaxBytes: stateMaxBytes * 3 + 8192, now, startHeartbeat, @@ -1818,7 +2141,7 @@ function normalizeRotationOptions(rawOptions) { options.maxTransactionDepth = MAX_TRANSACTION_DEPTH; options.maxLockGenerations = MAX_LOCK_GENERATIONS; options.maxJournalBytes = MAX_JOURNAL_BYTES; - options.maxJournalEntries = MAX_OPERATION_GENERATIONS * 4 + MAX_TRANSACTION_DEPTH + 32; + options.maxJournalEntries = MAX_OPERATION_GENERATIONS * 5 + MAX_TRANSACTION_DEPTH + 32; options.metadataMaxBytes = MAX_STATE_BYTES * 3 + 8192; return options; } @@ -1917,7 +2240,7 @@ async function readLegacyAuthority(statePath, lockDirectory, transactionDirector ); continue; } - if ((match = claimPattern.exec(name))) claimNames.set(Number(match[1]), name); + if ((match = undigestedClaimPattern.exec(name))) claimNames.set(Number(match[1]), name); else if ((match = heartbeatPattern.exec(name))) heartbeatNames.set(`${Number(match[1])}:${match[2]}`, name); else if ((match = terminalPattern.exec(name))) terminalNames.set(`${Number(match[1])}:${match[2]}`, name); else if ((match = appliedPattern.exec(name))) appliedNames.set(`${Number(match[1])}:${match[2]}`, name); @@ -2797,6 +3120,28 @@ async function runNormalLocked(statePath, action, rawOptions) { } } +function isExpectedRemovedClaimRead(error, context, options) { + if ( + !(error instanceof BoundedFileUnlinkedDuringReadError) || + error.description !== "Consumer high-water operation claim" || typeof error.path !== "string" || + !Buffer.isBuffer(error.bytes) || error.bytes.length < 1 || error.bytes.length > options.metadataMaxBytes + ) return false; + const name = basename(error.path); + const match = claimPattern.exec(name); + if ( + !match || error.expectedSha256 !== match[2] || digest(error.bytes) !== match[2] || + error.path !== join(context.epochDirectory, name) || dirname(error.path) !== context.epochDirectory + ) return false; + let claim; + try { + claim = validateClaim(JSON.parse(error.bytes), context, options.stateMaxBytes); + } catch { + return false; + } + return claim.generation === Number(match[1]) && metadataBytes(claim).equals(error.bytes) && + error.path === claimPath(context, claim); +} + async function completedRotationResult(context, intent, options) { const scan = await scanJournalRoot(context.statePath, context.journalDirectory, options); if (scan.head && metadataBytes(scan.head.checkpoint).equals(metadataBytes(intent.checkpoint))) { @@ -2820,17 +3165,28 @@ async function recoverCompletedCurrentRotation(context, scan, options) { async function runRotation(statePath, rawOptions) { const options = normalizeRotationOptions(rawOptions); for (;;) { - const { context } = await prepareContext(statePath, options); - await ensureLegacyGuard(context, { generation: 0, token: context.checkpoint.epochId, type: "rotation" }, options); - const initialScan = await scanEpoch(context, options); - const completed = await recoverCompletedCurrentRotation(context, initialScan, options); - if (completed) return completed; - const expectedIntent = rotationIntentFor(context, await effectiveTip(context, options)); + let context; + let expectedIntent; + try { + ({ context } = await prepareContext(statePath, options)); + await ensureLegacyGuard(context, { generation: 0, token: context.checkpoint.epochId, type: "rotation" }, options); + const initialScan = await scanEpoch(context, options); + const latest = initialScan.claims.at(-1); + const preparationOptions = latest?.type === "rotation" + ? { ...options, inProgressCheckpoint: validateRotationIntent(latest.intent, context, options.stateMaxBytes).checkpoint } + : options; + const completed = await recoverCompletedCurrentRotation(context, initialScan, preparationOptions); + if (completed) return completed; + expectedIntent = rotationIntentFor(context, await effectiveTip(context, preparationOptions)); + } catch (error) { + if (error instanceof ConsumerEpochAdvancedError) continue; + throw error; + } let frontier; try { frontier = await resolveOperationFrontier(context, options); } catch (error) { - if (error?.message !== "Consumer high-water journal epoch changed and fenced a paused writer.") throw error; + if (!(error instanceof ConsumerEpochAdvancedError) && !isExpectedRemovedClaimRead(error, context, options)) throw error; const completedResult = await completedRotationResult(context, expectedIntent, options); if (completedResult) return completedResult; throw error; diff --git a/scripts/pylon-publication.test.mjs b/scripts/pylon-publication.test.mjs index 37e359b446..d0eb95f751 100644 --- a/scripts/pylon-publication.test.mjs +++ b/scripts/pylon-publication.test.mjs @@ -6,6 +6,8 @@ import { closeSync, existsSync, fsyncSync, + fstatSync, + lstatSync, mkdirSync, mkdtempSync, openSync, @@ -17,13 +19,15 @@ import { statSync, symlinkSync, truncateSync, + utimesSync, writeFileSync, writeSync, } from "node:fs"; import { tmpdir } from "node:os"; -import { join, resolve } from "node:path"; +import { basename, join, resolve } from "node:path"; import { pathToFileURL } from "node:url"; import { createRequire } from "node:module"; +import { lstat as lstatFile, open as openFileHandle, readdir as readDirectoryEntries } from "node:fs/promises"; import { test } from "node:test"; import { @@ -87,9 +91,11 @@ import { withConsumerStateLock, } from "./lib/pylon-consumer-lock.mjs"; import { + BoundedFileUnlinkedDuringReadError, PYLON_PUBLICATION_MANIFEST_MAX_BYTES, PYLON_STABLE_HISTORY_MAX_MANIFESTS, readBoundedRegularFile, + readBoundedRegularFileSync, } from "./lib/pylon-bounded-file.mjs"; import { isExactWithdrawalReplay, selectStableHistoryReleases } from "./prepare-pylon-stable-manifest.mjs"; import { recoverStableDraft } from "./recover-pylon-stable-manifest.mjs"; @@ -733,10 +739,185 @@ test("consumer stable high-water pins and bounds every manifest before parsing", }, }, }), - /changed while it was read/, + (error) => { + assert.equal(error instanceof BoundedFileUnlinkedDuringReadError, false); + assert.match(error.message, /changed while it was read/); + return true; + }, ); assert.equal(swapped, true, "the descriptor re-stat detects a pathname swap without reading the replacement"); + const replacedBeforeOpen = join(fixture, "replaced-before-open.json"); + const replacedBeforeOpenMoved = join(fixture, "replaced-before-open-original.json"); + writeFileSync(replacedBeforeOpen, "original"); + let replacedOnOpen = false; + await assert.rejects( + () => readBoundedRegularFile(replacedBeforeOpen, { + maxBytes: 1024, + description: "Replaced bounded input", + openFile: async (path, flags) => { + if (!replacedOnOpen) { + replacedOnOpen = true; + renameSync(path, replacedBeforeOpenMoved); + writeFileSync(path, "replacement"); + } + return openFileHandle(path, flags); + }, + }), + (error) => { + assert.equal(error instanceof BoundedFileUnlinkedDuringReadError, false); + assert.match(error.message, /changed while it was read/); + return true; + }, + ); + assert.equal(replacedOnOpen, true, "an inode replaced between lstat and open is terminal, not convergence"); + + const overwrittenAsync = join(fixture, "overwritten-unlinked-async.bin"); + const originalAsyncBytes = Buffer.from("original-async"); + const replacementAsyncBytes = Buffer.from("replaced-async"); + assert.equal(originalAsyncBytes.length, replacementAsyncBytes.length); + writeFileSync(overwrittenAsync, originalAsyncBytes); + const originalAsyncStat = statSync(overwrittenAsync); + await assert.rejects( + () => readBoundedRegularFile(overwrittenAsync, { + maxBytes: 1024, + description: "Overwritten async bounded input", + expectedSha256: sha256Bytes(originalAsyncBytes), + hooks: { + afterInitialStat: () => { + writeFileSync(overwrittenAsync, replacementAsyncBytes); + utimesSync(overwrittenAsync, originalAsyncStat.atime, originalAsyncStat.mtime); + rmSync(overwrittenAsync); + }, + }, + }), + (error) => { + assert.equal(error instanceof BoundedFileUnlinkedDuringReadError, false); + assert.match(error.message, /changed while it was read/); + return true; + }, + ); + + const unanchoredRemovalPath = join(fixture, "unanchored-removal.bin"); + writeFileSync(unanchoredRemovalPath, "legacy"); + await assert.rejects( + () => readBoundedRegularFile(unanchoredRemovalPath, { + maxBytes: 1024, + description: "Unanchored bounded input", + hooks: { afterInitialStat: () => rmSync(unanchoredRemovalPath) }, + }), + (error) => { + assert.equal(error instanceof BoundedFileUnlinkedDuringReadError, false); + assert.match(error.message, /changed while it was read/); + return true; + }, + ); + + const overwrittenSync = join(fixture, "overwritten-unlinked-sync.bin"); + const originalSyncBytes = Buffer.from("original-sync"); + const replacementSyncBytes = Buffer.from("replaced-sync"); + assert.equal(originalSyncBytes.length, replacementSyncBytes.length); + writeFileSync(overwrittenSync, originalSyncBytes); + const originalSyncStat = statSync(overwrittenSync); + assert.throws( + () => readBoundedRegularFileSync(overwrittenSync, { + maxBytes: 1024, + description: "Overwritten sync bounded input", + expectedSha256: sha256Bytes(originalSyncBytes), + hooks: { + afterInitialStat: () => { + writeFileSync(overwrittenSync, replacementSyncBytes); + utimesSync(overwrittenSync, originalSyncStat.atime, originalSyncStat.mtime); + rmSync(overwrittenSync); + }, + }, + }), + (error) => { + assert.equal(error instanceof BoundedFileUnlinkedDuringReadError, false); + assert.match(error.message, /changed while it was read/); + return true; + }, + ); + + const asyncFstatFailurePath = join(fixture, "async-fstat-failure.bin"); + writeFileSync(asyncFstatFailurePath, "fstat"); + const asyncFstatFailure = new Error("injected async final fstat failure"); + await assert.rejects( + () => readBoundedRegularFile(asyncFstatFailurePath, { + maxBytes: 1024, + description: "Async fstat failure input", + expectedSha256: sha256Bytes(Buffer.from("fstat")), + openFile: async (path, flags) => { + const handle = await openFileHandle(path, flags); + let statCalls = 0; + return { + close: handle.close.bind(handle), + read: handle.read.bind(handle), + stat: async () => { + statCalls += 1; + if (statCalls === 2) throw asyncFstatFailure; + return handle.stat(); + }, + }; + }, + }), + (error) => error === asyncFstatFailure, + ); + + const asyncFinalLstatFailurePath = join(fixture, "async-final-lstat-failure.bin"); + writeFileSync(asyncFinalLstatFailurePath, "lstat"); + const asyncFinalLstatFailure = new Error("injected async final lstat failure"); + let asyncLstatCalls = 0; + await assert.rejects( + () => readBoundedRegularFile(asyncFinalLstatFailurePath, { + maxBytes: 1024, + description: "Async lstat failure input", + expectedSha256: sha256Bytes(Buffer.from("lstat")), + lstatEntry: async (path) => { + asyncLstatCalls += 1; + if (asyncLstatCalls === 2) throw asyncFinalLstatFailure; + return lstatFile(path); + }, + }), + (error) => error === asyncFinalLstatFailure, + ); + + const syncFstatFailurePath = join(fixture, "sync-fstat-failure.bin"); + writeFileSync(syncFstatFailurePath, "fstat"); + const syncFstatFailure = new Error("injected sync final fstat failure"); + let syncFstatCalls = 0; + assert.throws( + () => readBoundedRegularFileSync(syncFstatFailurePath, { + maxBytes: 1024, + description: "Sync fstat failure input", + expectedSha256: sha256Bytes(Buffer.from("fstat")), + statFile: (descriptor) => { + syncFstatCalls += 1; + if (syncFstatCalls === 2) throw syncFstatFailure; + return fstatSync(descriptor); + }, + }), + (error) => error === syncFstatFailure, + ); + + const syncFinalLstatFailurePath = join(fixture, "sync-final-lstat-failure.bin"); + writeFileSync(syncFinalLstatFailurePath, "lstat"); + const syncFinalLstatFailure = new Error("injected sync final lstat failure"); + let syncLstatCalls = 0; + assert.throws( + () => readBoundedRegularFileSync(syncFinalLstatFailurePath, { + maxBytes: 1024, + description: "Sync lstat failure input", + expectedSha256: sha256Bytes(Buffer.from("lstat")), + lstatEntry: (path) => { + syncLstatCalls += 1; + if (syncLstatCalls === 2) throw syncFinalLstatFailure; + return lstatSync(path); + }, + }), + (error) => error === syncFinalLstatFailure, + ); + const maximum = join(fixture, "maximum.bin"); writeFileSync(maximum, Buffer.alloc(PYLON_PUBLICATION_MANIFEST_MAX_BYTES, 0x61)); assert.equal( @@ -2167,7 +2348,7 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat assert.match(recoveries.find((result) => result.status === "rejected").reason.message, /actively locked/); oldRelease.resolve(); await oldOwnerRejected; - const raceClaims = readdirSync(consumerJournal(racePath).epoch).filter((name) => name.startsWith("claim-")); + const raceClaims = readdirSync(consumerJournal(racePath).epoch).filter((name) => name.startsWith("claim-index-")); assert.equal(raceClaims.length, 2); const fencedPath = join(fixture, "fenced.json"); @@ -2303,6 +2484,9 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat ["afterFileSync", "claim"], ["afterMetadataLink", "claim"], ["afterMetadataDirectorySync", "claim"], + ["afterFileSync", "claim-index"], + ["afterMetadataLink", "claim-index"], + ["afterMetadataDirectorySync", "claim-index"], ["afterFileSync", "terminal-commit"], ["afterMetadataLink", "terminal-commit"], ["afterMetadataDirectorySync", "terminal-commit"], @@ -2434,7 +2618,7 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat await transaction.commitState(bytes("normal-winner")); }, manualRuntime({ value: 2 }, { afterMetadataLink: async ({ kind }) => { - if (kind !== "claim") return; + if (kind !== "claim-index") return; normalClaimLinked.resolve(); await releaseNormalClaim.promise; }, @@ -2462,7 +2646,7 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat const releaseRotationClaim = deferred(); const rotationWinner = rotateConsumerStateJournal(rotationWinsPath, manualRuntime({ value: 2 }, { afterMetadataLink: async ({ kind }) => { - if (kind !== "claim") return; + if (kind !== "claim-index") return; rotationClaimLinked.resolve(); await releaseRotationClaim.promise; }, @@ -2470,7 +2654,7 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat await rotationClaimLinked.promise; const pendingRotationJournal = consumerJournal(rotationWinsPath); const pendingRotationClaim = readdirSync(pendingRotationJournal.epoch) - .filter((name) => name.startsWith("claim-")) + .filter((name) => /^claim-[0-9]{16}-[0-9a-f]{64}\.json$/.test(name)) .map((name) => JSON.parse(readFileSync(join(pendingRotationJournal.epoch, name)))) .at(-1); assert.equal(pendingRotationClaim.type, "rotation"); @@ -2530,7 +2714,7 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat const finalCrashCheckpoint = JSON.parse(readFileSync(finalCrashJournal.checkpoint)); const finalCrashClaim = JSON.parse(readFileSync(join( finalCrashJournal.epoch, - readdirSync(finalCrashJournal.epoch).find((name) => name === "claim-0000000000000002.json"), + readdirSync(finalCrashJournal.epoch).find((name) => /^claim-0000000000000002-[0-9a-f]{64}\.json$/.test(name)), ))); const finalCrashTemporaryDirectory = join(finalCrashJournal.journal, ".owned-temporaries-v2"); for (let index = 0; index < 17; index += 1) { @@ -2557,7 +2741,7 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat }); const floodJournal = consumerJournal(temporaryFloodPath); const floodCheckpoint = JSON.parse(readFileSync(floodJournal.checkpoint)); - const floodClaimName = readdirSync(floodJournal.epoch).find((name) => name.startsWith("claim-")); + const floodClaimName = readdirSync(floodJournal.epoch).find((name) => /^claim-[0-9]{16}-[0-9a-f]{64}\.json$/.test(name)); const floodClaim = JSON.parse(readFileSync(join(floodJournal.epoch, floodClaimName))); const floodTemporaryDirectory = join(floodJournal.journal, ".owned-temporaries-v2"); const temporaryFileName = ({ pid, generation, token, attempt, kind }) => @@ -2607,7 +2791,7 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat () => rotateConsumerStateJournal(liveTemporaryPath, manualRuntime({ value: 2 })), /rotation operation is pending.*temporary writer quiesces/, ); - assert.equal(readdirSync(liveJournal.epoch).filter((name) => name.startsWith("claim-")).some( + assert.equal(readdirSync(liveJournal.epoch).filter((name) => /^claim-[0-9]{16}-[0-9a-f]{64}\.json$/.test(name)).some( (name) => JSON.parse(readFileSync(join(liveJournal.epoch, name))).type === "rotation", ), true); await assert.rejects( @@ -2652,6 +2836,473 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat tipSha256: sha256Bytes(bytes("frontier-handoff-anchor")), }); + const removedClaimHandoffPath = join(fixture, "removed-claim-completion-handoff.json"); + await withConsumerStateLock(removedClaimHandoffPath, async (_path, transaction) => { + await transaction.commitState(bytes("removed-claim-handoff-anchor")); + }, manualRuntime({ value: 1 })); + const removedClaimJournal = consumerJournal(removedClaimHandoffPath); + const removedClaimIndexName = readdirSync(removedClaimJournal.epoch).find((name) => name.startsWith("claim-index-")); + const removedClaimIndex = JSON.parse(readFileSync(join(removedClaimJournal.epoch, removedClaimIndexName))); + const removedClaimContentName = readdirSync(removedClaimJournal.epoch) + .find((name) => /^claim-[0-9]{16}-[0-9a-f]{64}\.json$/.test(name)); + const removedClaimContentBytes = readFileSync(join(removedClaimJournal.epoch, removedClaimContentName)); + assert.equal(removedClaimIndex.claimSha256, sha256Bytes(removedClaimContentBytes)); + assert.match(removedClaimContentName, new RegExp(`${removedClaimIndex.claimSha256}\\.json$`)); + const removedClaimReadReached = deferred(); + const releaseRemovedClaimRead = deferred(); + let claimReads = 0; + let openedClaimPath; + const removedClaimStale = rotateConsumerStateJournal(removedClaimHandoffPath, manualRuntime({ value: 2 }, { + metadataRead: { + afterInitialStat: async ({ path }) => { + if (!/^claim-[0-9]{16}-[0-9a-f]{64}\.json$/.test(basename(path))) return; + claimReads += 1; + if (claimReads !== 2) return; + openedClaimPath = path; + removedClaimReadReached.resolve(); + await releaseRemovedClaimRead.promise; + }, + }, + })); + await removedClaimReadReached.promise; + const removedClaimWinner = await rotateConsumerStateJournal(removedClaimHandoffPath, manualRuntime({ value: 3 })); + const removedClaimCleanup = await rotateConsumerStateJournal(removedClaimHandoffPath, manualRuntime({ value: 4 })); + assert.deepEqual(removedClaimCleanup, removedClaimWinner); + assert.equal(existsSync(openedClaimPath), false, "the authenticated old claim inode is removed while its read handle is pinned"); + releaseRemovedClaimRead.resolve(); + const removedClaimResumed = await removedClaimStale; + assert.equal(Buffer.from(JSON.stringify(removedClaimResumed)).equals( + Buffer.from(JSON.stringify(removedClaimWinner)), + ), true, "a same-inode old-claim removal converges only to the byte-identical winner receipt"); + assert.deepEqual(removedClaimWinner, { + epoch: 2, + tipSha256: sha256Bytes(bytes("removed-claim-handoff-anchor")), + }); + + const earlyFenceHandoffPath = join(fixture, "early-fence-completion-handoff.json"); + await withConsumerStateLock(earlyFenceHandoffPath, async (_path, transaction) => { + await transaction.commitState(bytes("early-fence-handoff-anchor")); + }, manualRuntime({ value: 1 })); + const earlyFenceReached = deferred(); + const releaseEarlyFence = deferred(); + let preIntentWalks = 0; + const earlyFenceStale = rotateConsumerStateJournal(earlyFenceHandoffPath, manualRuntime({ value: 2 }, { + beforePathOperation: async ({ operation }) => { + if (operation !== "walk-transactions") return; + preIntentWalks += 1; + if (preIntentWalks !== 1) return; + earlyFenceReached.resolve(); + await releaseEarlyFence.promise; + }, + })); + await earlyFenceReached.promise; + const earlyFenceWinner = await rotateConsumerStateJournal(earlyFenceHandoffPath, manualRuntime({ value: 3 })); + releaseEarlyFence.resolve(); + const earlyFenceResumed = await earlyFenceStale; + assert.equal(Buffer.from(JSON.stringify(earlyFenceResumed)).equals( + Buffer.from(JSON.stringify(earlyFenceWinner)), + ), true, "a pre-intent epoch fence reauthenticates and returns the byte-identical winner receipt"); + assert.deepEqual(earlyFenceWinner, { + epoch: 2, + tipSha256: sha256Bytes(bytes("early-fence-handoff-anchor")), + }); + + const malformedSuccessorPath = join(fixture, "malformed-successor-fence.json"); + await withConsumerStateLock(malformedSuccessorPath, async (_path, transaction) => { + await transaction.commitState(bytes("malformed-successor-anchor")); + }, manualRuntime({ value: 1 })); + const malformedSuccessorReached = deferred(); + const releaseMalformedSuccessor = deferred(); + let malformedSuccessorWalks = 0; + const malformedSuccessorStale = rotateConsumerStateJournal(malformedSuccessorPath, manualRuntime({ value: 2 }, { + beforePathOperation: async ({ operation }) => { + if (operation !== "walk-transactions") return; + malformedSuccessorWalks += 1; + if (malformedSuccessorWalks !== 1) return; + malformedSuccessorReached.resolve(); + await releaseMalformedSuccessor.promise; + }, + })); + await malformedSuccessorReached.promise; + const malformedSuccessorJournal = consumerJournal(malformedSuccessorPath); + writePrivate( + join(malformedSuccessorJournal.journal, `checkpoint-0000000000000002-${randomUUID()}.json`), + "{}\n", + ); + releaseMalformedSuccessor.resolve(); + await assert.rejects(malformedSuccessorStale, /journal checkpoint is malformed/); + + const skippedSuccessorPath = join(fixture, "skipped-successor-fence.json"); + await withConsumerStateLock(skippedSuccessorPath, async (_path, transaction) => { + await transaction.commitState(bytes("skipped-successor-anchor")); + }, manualRuntime({ value: 1 })); + const skippedSuccessorReached = deferred(); + const releaseSkippedSuccessor = deferred(); + let skippedSuccessorWalks = 0; + const skippedSuccessorStale = rotateConsumerStateJournal(skippedSuccessorPath, manualRuntime({ value: 2 }, { + beforePathOperation: async ({ operation }) => { + if (operation !== "walk-transactions") return; + skippedSuccessorWalks += 1; + if (skippedSuccessorWalks !== 1) return; + skippedSuccessorReached.resolve(); + await releaseSkippedSuccessor.promise; + }, + })); + await skippedSuccessorReached.promise; + let skippedCheckpoint; + const captureSkippedCheckpoint = new Error("capture skipped successor checkpoint"); + await assert.rejects( + () => rotateConsumerStateJournal(skippedSuccessorPath, manualRuntime({ value: 3 }, { + beforeRotationDecision: async ({ claim }) => { + skippedCheckpoint = { ...claim.intent.checkpoint, epoch: claim.intent.checkpoint.epoch + 1 }; + throw captureSkippedCheckpoint; + }, + })), + (error) => error === captureSkippedCheckpoint, + ); + const skippedSuccessorJournal = consumerJournal(skippedSuccessorPath); + writePrivate( + join( + skippedSuccessorJournal.journal, + `checkpoint-${String(skippedCheckpoint.epoch).padStart(16, "0")}-${skippedCheckpoint.epochId}.json`, + ), + metadata(skippedCheckpoint), + ); + releaseSkippedSuccessor.resolve(); + await assert.rejects(skippedSuccessorStale, /journal checkpoints are not contiguous/); + + const competingSuccessorPath = join(fixture, "competing-successor-fence.json"); + await withConsumerStateLock(competingSuccessorPath, async (_path, transaction) => { + await transaction.commitState(bytes("competing-successor-anchor")); + }, manualRuntime({ value: 1 })); + const competingSuccessorReached = deferred(); + const releaseCompetingSuccessor = deferred(); + let competingSuccessorWalks = 0; + const competingSuccessorStale = rotateConsumerStateJournal(competingSuccessorPath, manualRuntime({ value: 2 }, { + beforePathOperation: async ({ operation }) => { + if (operation !== "walk-transactions") return; + competingSuccessorWalks += 1; + if (competingSuccessorWalks !== 1) return; + competingSuccessorReached.resolve(); + await releaseCompetingSuccessor.promise; + }, + })); + await competingSuccessorReached.promise; + let competingCheckpoint; + const captureCompetingCheckpoint = new Error("capture competing successor checkpoint"); + await assert.rejects( + () => rotateConsumerStateJournal(competingSuccessorPath, manualRuntime({ value: 3 }, { + beforeRotationDecision: async ({ claim }) => { + competingCheckpoint = claim.intent.checkpoint; + throw captureCompetingCheckpoint; + }, + })), + (error) => error === captureCompetingCheckpoint, + ); + const competingSuccessorJournal = consumerJournal(competingSuccessorPath); + const alternateCompetingCheckpoint = { ...competingCheckpoint, epochId: randomUUID() }; + for (const checkpoint of [competingCheckpoint, alternateCompetingCheckpoint]) { + writePrivate( + join( + competingSuccessorJournal.journal, + `checkpoint-${String(checkpoint.epoch).padStart(16, "0")}-${checkpoint.epochId}.json`, + ), + metadata(checkpoint), + ); + } + releaseCompetingSuccessor.resolve(); + await assert.rejects( + competingSuccessorStale, + /journal (?:checkpoint set is malformed|root contains unbounded checkpoint metadata)/, + ); + + const publicationEntries = (journal) => readdirSync(journal.epoch) + .filter((name) => /^(?:claim(?:-index)?-|terminal-|transition-)/.test(name)) + .sort(); + const sameEpochCheckpointPath = join(fixture, "same-epoch-checkpoint-root-competitor.json"); + await withConsumerStateLock(sameEpochCheckpointPath, async (_path, transaction) => { + await transaction.commitState(bytes("same-epoch-checkpoint-anchor")); + }, manualRuntime({ value: 1 })); + const sameEpochCheckpointJournal = consumerJournal(sameEpochCheckpointPath); + const sameEpochCheckpointBaseline = publicationEntries(sameEpochCheckpointJournal); + const maliciousCheckpointName = `checkpoint-0000000000000001-${randomUUID()}.json`; + const sameEpochCheckpointReached = deferred(); + const releaseSameEpochCheckpoint = deferred(); + let sameEpochCheckpointInjected = false; + const sameEpochCheckpointWriter = withConsumerStateLock(sameEpochCheckpointPath, async () => {}, { + ...manualRuntime({ value: 2 }, { + beforePathOperation: async ({ operation }) => { + if (operation !== "claim") return; + sameEpochCheckpointReached.resolve(); + await releaseSameEpochCheckpoint.promise; + }, + }), + readDirectory: async (path) => { + const names = await readDirectoryEntries(path); + if (!sameEpochCheckpointInjected || path !== sameEpochCheckpointJournal.journal) return names; + return [maliciousCheckpointName, ...names.filter((name) => name !== maliciousCheckpointName)]; + }, + }); + await sameEpochCheckpointReached.promise; + writePrivate(join(sameEpochCheckpointJournal.journal, maliciousCheckpointName), "{}\n"); + sameEpochCheckpointInjected = true; + releaseSameEpochCheckpoint.resolve(); + await assert.rejects(sameEpochCheckpointWriter, /journal checkpoint is malformed/); + assert.deepEqual( + publicationEntries(sameEpochCheckpointJournal), + sameEpochCheckpointBaseline, + "a malicious same-epoch checkpoint cannot publish a claim, terminal, or transition", + ); + + const sameEpochDirectoryPath = join(fixture, "same-epoch-directory-root-competitor.json"); + await withConsumerStateLock(sameEpochDirectoryPath, async (_path, transaction) => { + await transaction.commitState(bytes("same-epoch-directory-anchor")); + }, manualRuntime({ value: 1 })); + const sameEpochDirectoryJournal = consumerJournal(sameEpochDirectoryPath); + const sameEpochDirectoryBaseline = publicationEntries(sameEpochDirectoryJournal); + const sameEpochDirectoryReached = deferred(); + const releaseSameEpochDirectory = deferred(); + const sameEpochDirectoryWriter = withConsumerStateLock(sameEpochDirectoryPath, async () => {}, manualRuntime( + { value: 2 }, + { + beforePathOperation: async ({ operation }) => { + if (operation !== "claim") return; + sameEpochDirectoryReached.resolve(); + await releaseSameEpochDirectory.promise; + }, + }, + )); + await sameEpochDirectoryReached.promise; + mkdirSync( + join(sameEpochDirectoryJournal.journal, `epoch-0000000000000001-${randomUUID()}`), + { mode: 0o700 }, + ); + releaseSameEpochDirectory.resolve(); + await assert.rejects(sameEpochDirectoryWriter, /competing epoch directories for one parent epoch/); + assert.deepEqual( + publicationEntries(sameEpochDirectoryJournal), + sameEpochDirectoryBaseline, + "a sibling same-epoch directory cannot publish a claim, terminal, or transition", + ); + + const unindexedOrderPath = join(fixture, "unindexed-claim-validation-order.json"); + await withConsumerStateLock(unindexedOrderPath, async () => {}, manualRuntime({ value: 1 })); + const unindexedOrderJournal = consumerJournal(unindexedOrderPath); + const indexedClaimName = readdirSync(unindexedOrderJournal.epoch) + .find((name) => /^claim-[0-9]{16}-[0-9a-f]{64}\.json$/.test(name)); + const indexedClaimBytes = readFileSync(join(unindexedOrderJournal.epoch, indexedClaimName)); + const unorderedClaims = [ + `claim-0000000000000003-${"0".repeat(64)}.json`, + `claim-0000000000000002-${"f".repeat(64)}.json`, + `claim-0000000000000002-${"0".repeat(64)}.json`, + ]; + for (const name of unorderedClaims) writePrivate(join(unindexedOrderJournal.epoch, name), indexedClaimBytes); + const unindexedReads = []; + await assert.rejects( + () => rotateConsumerStateJournal(unindexedOrderPath, { + ...manualRuntime({ value: 2 }, { + metadataRead: { + afterInitialStat: ({ path }) => { + if (unorderedClaims.includes(basename(path))) unindexedReads.push(basename(path)); + }, + }, + }), + readDirectory: async (path) => { + const names = await readDirectoryEntries(path); + if (path !== unindexedOrderJournal.epoch) return names; + return [...unorderedClaims, ...names.filter((name) => !unorderedClaims.includes(name))]; + }, + }), + /unindexed claim content differs from its exact canonical bytes/, + ); + assert.deepEqual( + unindexedReads, + [`claim-0000000000000002-${"0".repeat(64)}.json`], + "unindexed claims validate by numeric generation and then lexical content digest", + ); + + const missingSuccessorEpochPath = join(fixture, "missing-successor-epoch-fence.json"); + await withConsumerStateLock(missingSuccessorEpochPath, async (_path, transaction) => { + await transaction.commitState(bytes("missing-successor-epoch-anchor")); + }, manualRuntime({ value: 1 })); + const missingSuccessorEpochReached = deferred(); + const releaseMissingSuccessorEpoch = deferred(); + let missingSuccessorEpochWalks = 0; + const missingSuccessorEpochStale = rotateConsumerStateJournal(missingSuccessorEpochPath, manualRuntime({ value: 2 }, { + beforePathOperation: async ({ operation }) => { + if (operation !== "walk-transactions") return; + missingSuccessorEpochWalks += 1; + if (missingSuccessorEpochWalks !== 1) return; + missingSuccessorEpochReached.resolve(); + await releaseMissingSuccessorEpoch.promise; + }, + })); + await missingSuccessorEpochReached.promise; + let missingEpochCheckpoint; + const captureMissingEpoch = new Error("capture successor without publishing it"); + await assert.rejects( + () => rotateConsumerStateJournal(missingSuccessorEpochPath, manualRuntime({ value: 3 }, { + beforeRotationDecision: async ({ claim }) => { + missingEpochCheckpoint = claim.intent.checkpoint; + throw captureMissingEpoch; + }, + })), + (error) => error === captureMissingEpoch, + ); + const missingSuccessorJournal = consumerJournal(missingSuccessorEpochPath); + writePrivate( + join( + missingSuccessorJournal.journal, + `checkpoint-${String(missingEpochCheckpoint.epoch).padStart(16, "0")}-${missingEpochCheckpoint.epochId}.json`, + ), + metadata(missingEpochCheckpoint), + ); + releaseMissingSuccessorEpoch.resolve(); + await assert.rejects( + missingSuccessorEpochStale, + /journal root changed without one exact current or immediate-successor authority/, + ); + + const successorIoPath = join(fixture, "successor-scan-io-fence.json"); + await withConsumerStateLock(successorIoPath, async (_path, transaction) => { + await transaction.commitState(bytes("successor-io-anchor")); + }, manualRuntime({ value: 1 })); + const successorIoJournal = consumerJournal(successorIoPath); + const successorIoReached = deferred(); + const releaseSuccessorIo = deferred(); + const successorIoFailure = new Error("injected successor scan I/O failure"); + let successorIoWalks = 0; + let injectSuccessorIo = false; + let successorIoRootReads = 0; + const successorIoStale = rotateConsumerStateJournal(successorIoPath, { + ...manualRuntime({ value: 2 }, { + beforePathOperation: async ({ operation }) => { + if (operation !== "walk-transactions") return; + successorIoWalks += 1; + if (successorIoWalks !== 1) return; + successorIoReached.resolve(); + await releaseSuccessorIo.promise; + }, + }), + readDirectory: async (path) => { + if (injectSuccessorIo && path === successorIoJournal.journal) { + successorIoRootReads += 1; + if (successorIoRootReads === 2) throw successorIoFailure; + } + return readDirectoryEntries(path); + }, + }); + await successorIoReached.promise; + await rotateConsumerStateJournal(successorIoPath, manualRuntime({ value: 3 })); + injectSuccessorIo = true; + releaseSuccessorIo.resolve(); + await assert.rejects(successorIoStale, (error) => error === successorIoFailure); + + const falseFencePath = join(fixture, "false-typed-fence.json"); + await withConsumerStateLock(falseFencePath, async (_path, transaction) => { + await transaction.commitState(bytes("false-fence-anchor")); + }, manualRuntime({ value: 1 })); + const falseFenceReached = deferred(); + const releaseFalseFence = deferred(); + const falseFence = new Error("Consumer high-water journal epoch changed and fenced a paused writer."); + let falseFenceScans = 0; + const falseFenceStale = rotateConsumerStateJournal(falseFencePath, manualRuntime({ value: 2 }, { + beforePathOperation: async ({ operation }) => { + if (operation !== "scan-claims") return; + falseFenceScans += 1; + if (falseFenceScans !== 2) return; + falseFenceReached.resolve(); + await releaseFalseFence.promise; + throw falseFence; + }, + })); + await falseFenceReached.promise; + await rotateConsumerStateJournal(falseFencePath, manualRuntime({ value: 3 })); + releaseFalseFence.resolve(); + await assert.rejects(falseFenceStale, (error) => { + assert.equal(error, falseFence); + assert.equal(error.name, "Error"); + return true; + }); + + const escapedRemovalPath = join(fixture, "escaped-removal-signal.json"); + await withConsumerStateLock(escapedRemovalPath, async (_path, transaction) => { + await transaction.commitState(bytes("escaped-removal-anchor")); + }, manualRuntime({ value: 1 })); + const escapedRemovalJournal = consumerJournal(escapedRemovalPath); + const escapedClaimName = readdirSync(escapedRemovalJournal.epoch) + .find((name) => /^claim-[0-9]{16}-[0-9a-f]{64}\.json$/.test(name)); + const escapedClaimBytes = readFileSync(join(escapedRemovalJournal.epoch, escapedClaimName)); + const escapedClaimSha256 = /-([0-9a-f]{64})\.json$/.exec(escapedClaimName)[1]; + const escapedRemovalReached = deferred(); + const releaseEscapedRemoval = deferred(); + let escapedRemovalScans = 0; + let escapedRemoval; + const escapedRemovalStale = rotateConsumerStateJournal(escapedRemovalPath, manualRuntime({ value: 2 }, { + beforePathOperation: async ({ operation, transactionDirectory }) => { + if (operation !== "scan-claims") return; + escapedRemovalScans += 1; + if (escapedRemovalScans !== 2) return; + escapedRemoval = new BoundedFileUnlinkedDuringReadError( + join(transactionDirectory, "..", escapedClaimName), + "Consumer high-water operation claim", + escapedClaimBytes, + escapedClaimSha256, + ); + escapedRemovalReached.resolve(); + await releaseEscapedRemoval.promise; + throw escapedRemoval; + }, + })); + await escapedRemovalReached.promise; + await rotateConsumerStateJournal(escapedRemovalPath, manualRuntime({ value: 3 })); + releaseEscapedRemoval.resolve(); + await assert.rejects(escapedRemovalStale, (error) => { + assert.equal(error, escapedRemoval); + return true; + }); + + const changedClaimPath = join(fixture, "changed-claim-is-terminal.json"); + await withConsumerStateLock(changedClaimPath, async (_path, transaction) => { + await transaction.commitState(bytes("changed-claim-anchor")); + }, manualRuntime({ value: 1 })); + const changedClaimReadReached = deferred(); + const releaseChangedClaimRead = deferred(); + let changedClaimReads = 0; + let openedChangedClaimPath; + const changedClaimStale = rotateConsumerStateJournal(changedClaimPath, manualRuntime({ value: 2 }, { + metadataRead: { + afterInitialStat: async ({ path }) => { + if (!/^claim-[0-9]{16}-[0-9a-f]{64}\.json$/.test(basename(path))) return; + changedClaimReads += 1; + if (changedClaimReads !== 2) return; + openedChangedClaimPath = path; + changedClaimReadReached.resolve(); + await releaseChangedClaimRead.promise; + }, + }, + })); + await changedClaimReadReached.promise; + await rotateConsumerStateJournal(changedClaimPath, manualRuntime({ value: 3 })); + writePrivate(openedChangedClaimPath, '{"malformed":true}\n'); + releaseChangedClaimRead.resolve(); + await assert.rejects(changedClaimStale, (error) => { + assert.equal(error instanceof BoundedFileUnlinkedDuringReadError, false); + assert.match(error.message, /changed while it was read/); + return true; + }); + + const malformedClaimPath = join(fixture, "malformed-claim-is-terminal.json"); + await withConsumerStateLock(malformedClaimPath, async () => {}, manualRuntime({ value: 1 })); + const malformedClaimJournal = consumerJournal(malformedClaimPath); + const malformedClaimName = readdirSync(malformedClaimJournal.epoch) + .find((name) => /^claim-[0-9]{16}-[0-9a-f]{64}\.json$/.test(name)); + writePrivate(join(malformedClaimJournal.epoch, malformedClaimName), "{}\n"); + await assert.rejects( + () => rotateConsumerStateJournal(malformedClaimPath, manualRuntime({ value: 2 })), + /operation claim is malformed/, + ); + const concurrentRotationPath = join(fixture, "concurrent-rotation.json"); await withConsumerStateLock(concurrentRotationPath, async (_path, transaction) => { await transaction.commitState(bytes("concurrent-anchor")); @@ -2766,9 +3417,12 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat ["afterFileSync", "claim"], ["afterMetadataLink", "claim"], ["afterMetadataDirectorySync", "claim"], + ["afterFileSync", "claim-index"], + ["afterMetadataLink", "claim-index"], + ["afterMetadataDirectorySync", "claim-index"], ["afterRotationIntent", null], ]) { - const intentCrashPath = join(fixture, `rotation-operation-crash-${hookName}.json`); + const intentCrashPath = join(fixture, `rotation-operation-crash-${hookName}-${wantedKind ?? "rotation-decision"}.json`); await withConsumerStateLock(intentCrashPath, async (_path, transaction) => { await transaction.commitState(bytes("intent-anchor")); }, manualRuntime({ value: 1 })); @@ -2783,7 +3437,11 @@ test("consumer state locking, recovery, transaction fencing, durability, and pat })), /simulated rotation operation crash/, ); - assert.equal((await rotateConsumerStateJournal(intentCrashPath, manualRuntime({ value: 3 }))).epoch, 2); + assert.equal( + (await rotateConsumerStateJournal(intentCrashPath, manualRuntime({ value: 3 }))).epoch, + 2, + `${hookName}/${wantedKind ?? "rotation-decision"} must recover the same rotation epoch`, + ); } for (const [hookName, wantedKind] of [