import fs from 'node:fs/promises'
import path from 'node:path'
import { execFile } from 'node:child_process'
import { promisify } from 'node:util'
import { fileURLToPath } from 'node:url'
import b4a from 'b4a'
import test from 'brittle'
import { blake2b256 } from '@hiverelay/blind-protocol'
import {
createBlindBoundaryScratch,
removeBlindBoundaryScratch
} from '../../../test/blind-boundary-scratch.js'
import {
openForwardHttpsAggregateQuotaV3,
mintForwardHttpsAggregateQuotaCapabilitiesV3,
closeForwardHttpsAggregateQuotaV3,
beginForwardHttpsAggregateQuotaRecoveryV3,
finishForwardHttpsAggregateQuotaRecoveryV3,
initializeForwardHttpsAggregateQuotaV3,
assertForwardHttpsAggregateQuotaOperationalV3,
failForwardHttpsAggregateQuotaWalAttemptV3,
forwardHttpsAggregateQuotaV3Status,
createForwardHttpsStoreQuotaCostPlanV3,
reserveForwardHttpsAggregateQuotaV3,
bindForwardHttpsStoreQuotaActualBuffersV3,
applyForwardHttpsAggregateQuotaWalFrameV3,
commitForwardHttpsAggregateQuotaV3,
releaseForwardHttpsAggregateQuotaV3,
adjustForwardHttpsAggregateQuotaV3,
encodeForwardHttpsRetentionPrunedV3
} from '../forward-https-replay-journal-v4.js'
import {
openForwardHttpsTargetStoreV3,
acceptForwardedHttpsTargetTurnV3,
forwardHttpsTargetStoreV3Status,
closeForwardHttpsTargetStoreV3,
FORWARD_HTTPS_TARGET_STORE_V3_STATUS,
FORWARD_HTTPS_TARGET_STORE_V3_FAULT_POINT
} from '../forward-https-target-store-v3.js'
import {
FORWARD_HTTPS_SOURCE_STORE_V3_FAULT_POINT
} from '../forward-https-source-store-v3.js'
import {
FORWARD_HTTPS_STORAGE_V3_FAULT_POINT
} from '../forward-https-storage-authority-v3.js'
const execFileAsync = promisify(execFile)
function fixed (byte) {
return b4a.alloc(32, byte)
}
const SLOT = Object.freeze({
FREE: 'FREE',
PROVISIONAL: 'PROVISIONAL',
PREFIX_ALLOCATED: 'PREFIX_ALLOCATED',
ALLOCATED: 'ALLOCATED',
ALLOCATED_WITH_PREFIX: 'ALLOCATED_WITH_PREFIX',
CONSUMED_UNPRUNED: 'CONSUMED_UNPRUNED',
CONSUMED_PRUNED: 'CONSUMED_PRUNED'
})
const IDENTITY = Object.freeze({
NEVER_SEEN: 'NEVER_SEEN',
PRESENT_PREFIX_ALLOCATED: 'PRESENT_PREFIX_ALLOCATED',
ALLOCATED_WITH_PREFIX: 'ALLOCATED_WITH_PREFIX',
PRESENT_ALLOCATED: 'PRESENT_ALLOCATED',
PRESENT_CONSUMED_UNPRUNED: 'PRESENT_CONSUMED_UNPRUNED',
PRUNED_RELEASED: 'PRUNED_RELEASED',
PRUNED_CONSUMED: 'PRUNED_CONSUMED'
})
function identityOf (slot) {
if (slot.state === SLOT.ALLOCATED || slot.state === SLOT.PROVISIONAL) return IDENTITY.PRESENT_ALLOCATED
if (slot.state === SLOT.PREFIX_ALLOCATED) return IDENTITY.PRESENT_PREFIX_ALLOCATED
if (slot.state === SLOT.ALLOCATED_WITH_PREFIX) return IDENTITY.ALLOCATED_WITH_PREFIX
if (slot.state === SLOT.CONSUMED_UNPRUNED) return IDENTITY.PRESENT_CONSUMED_UNPRUNED
if (slot.state === SLOT.CONSUMED_PRUNED) return IDENTITY.PRUNED_CONSUMED
if (slot.state === SLOT.FREE && slot.prunedReleased) return IDENTITY.PRUNED_RELEASED
return IDENTITY.NEVER_SEEN
}
function prefixPayload (id, fill = 0x52) {
return b4a.concat([b4a.from('FTS3', 'ascii'), id, b4a.alloc(118 - 36, fill)])
}
function writeU64be (output, offset, value) {
let current = BigInt(value)
for (let index = 7; index >= 0; index--) { output[offset + index] = Number(current & 0xffn); current >>= 8n }
return offset + 8
}
// Test-side FTM9 payload builder (frozen 192-byte layout; caller-supplied
// exact payloads, the requested-terminal-arm shape).
function buildFtm9 (input) {
const output = b4a.alloc(192)
let offset = 0
b4a.copy(b4a.from('FTM9', 'ascii'), output, offset); offset += 4
output[offset++] = 1
output[offset++] = 2
output.writeUInt16BE(input.flags, offset); offset += 2
b4a.copy(input.stableSessionId, output, offset); offset += 32
offset = writeU64be(output, offset, input.sequence)
for (const value of input.buckets || [0, 0, 0, 0, 0]) { output.writeUInt16BE(value, offset); offset += 2 }
output.writeUInt16BE(input.transportTurnsSpent || 0, offset); offset += 2
output.writeUInt32BE(input.transportBytesSpent || 0, offset); offset += 4
offset = writeU64be(output, offset, input.priorSessionRevision || 0n)
output.writeUInt32BE(input.newTrustedEpochHighWatermark || 0, offset); offset += 4
const reason = b4a.from(input.reason, 'ascii')
output[offset++] = reason.byteLength
b4a.copy(reason, output, offset); offset += 64
if (input.flags === 1) {
output.writeUInt32BE(input.expiresAtEpoch, offset); offset += 4
output.writeUInt32BE(input.retainedUntilEpoch, offset); offset += 4
b4a.copy(input.exactRequestCommitment, output, offset); offset += 32
}
return output
}
// Quota-API drivers: terminal append through the composite, prune through
// bind/apply/adjust, both with caller-supplied exact payloads.
async function driveWal (store, operation, frames) {
const plan = createForwardHttpsStoreQuotaCostPlanV3(store.storeQuotaCapability, {
operation,
knownInputBuffers: frames.map(frame => frame.payload),
temporaryWriteBuffers: [],
existingDestinationBytes: 0
})
const union = await reserveForwardHttpsAggregateQuotaV3(store.storeQuotaCapability, plan)
const reservation = union.reservation || union.terminalReservation
let attempted = false
try {
const { transitionAuthority } = bindForwardHttpsStoreQuotaActualBuffersV3(store.storeQuotaCapability, reservation, {
logicalRecordBuffers: [],
encryptedPlaintextBuffers: frames.slice(0, -1).map(frame => frame.payload),
finalWalMetadataBuffers: [frames[frames.length - 1].payload],
temporaryWriteBuffers: []
})
attempted = true
let claimOrHandoff = transitionAuthority
let entry = null
for (const candidate of frames) {
const applied = await applyForwardHttpsAggregateQuotaWalFrameV3(store.storeQuotaCapability, reservation, claimOrHandoff, candidate,
async frame => store.store.appendAndApply(frame, () => {}))
entry = applied.entry
claimOrHandoff = applied.transitionAuthorityHandoff
}
return { entry, reservation }
} catch (error) {
if (!attempted) await releaseForwardHttpsAggregateQuotaV3(store.storeQuotaCapability, reservation).catch(() => {})
throw error
}
}
async function driveTerminal (store, payload) {
const { entry, reservation } = await driveWal(store, 'SESSION_TERMINAL', [{
type: 117,
transactionId: b4a.alloc(32, 0x5a),
virtualBucket: 0,
payload
}])
await commitForwardHttpsAggregateQuotaV3(store.storeQuotaCapability, reservation, {
durableWalHeadSequence: store.store.walSequence,
durableWalHeadHash: store.store.walHash
})
return Object.freeze({ walSequence: entry.walSequence, walHash: b4a.from(store.store.walHash), payload })
}
async function drivePrune (store, payload) {
const { entry } = await driveWal(store, 'PRUNE', [{
type: 118,
transactionId: b4a.alloc(32, 0x5a),
virtualBucket: 0,
payload
}])
await adjustForwardHttpsAggregateQuotaV3(store.storeQuotaCapability, {
durableTombstonePayloadBuffer: payload,
durableWalHeadSequence: store.store.walSequence,
durableWalHeadHash: store.store.walHash
})
return Object.freeze({ walSequence: entry.walSequence, walHash: b4a.from(store.store.walHash), payload })
}
// The operation final carries the exact same 32-byte requestCommitment at
// body offset 0 so recovery closes the prefix run exactly once.
function finalPayload (id, commitmentFill, fill = 0x54) {
return b4a.concat([b4a.from('FTS3', 'ascii'), id, b4a.alloc(32, commitmentFill), b4a.alloc(8, fill)])
}
// Raw durable WAL append outside any quota operation, used to construct
// crashed-prefix evidence; recovery replays it exactly once on reopen.
async function rawAppend (store, type, payload) {
return store.store.appendAndApply({ type, transactionId: b4a.alloc(32, 0x66), virtualBucket: 0, payload }, () => {})
}
async function roots (t) {
const base = await createBlindBoundaryScratch('fhts3-')
t.teardown(async () => {
await removeBlindBoundaryScratch(base)
})
const names = ['source-replay', 'target-replay', 'source-store', 'target-store']
const out = {}
for (const name of names) {
out[name] = path.join(base, name)
await fs.mkdir(out[name], { mode: 0o700 })
await fs.chmod(out[name], 0o700)
}
return out
}
async function quota (t, r) {
const authority = await openForwardHttpsAggregateQuotaV3({
sourceReplayRoot: r['source-replay'],
targetReplayRoot: r['target-replay'],
sourceStoreRoot: r['source-store'],
targetStoreRoot: r['target-store'],
maximumDurableBytesPerStore: 8589934592,
maximumForwardStorageBytesAggregate: 17179869184,
monotonicMillis: () => Date.now(),
callbackTimeoutMillis: 15000,
faultInjector: null
})
t.teardown(async () => {
await closeForwardHttpsAggregateQuotaV3(authority).catch(() => {})
})
return { authority, capabilities: mintForwardHttpsAggregateQuotaCapabilitiesV3(authority) }
}
async function quota2 (t, r) {
const authority = await openForwardHttpsAggregateQuotaV3({
sourceReplayRoot: r['source-replay'],
targetReplayRoot: r['target-replay'],
sourceStoreRoot: r['source-store'],
targetStoreRoot: r['target-store'],
maximumDurableBytesPerStore: 8589934592,
maximumForwardStorageBytesAggregate: 17179869184,
monotonicMillis: () => Date.now(),
callbackTimeoutMillis: 15000,
faultInjector: null
})
return { authority, capabilities: mintForwardHttpsAggregateQuotaCapabilitiesV3(authority) }
}
// Every store operation passes the operational gate, so the quota authority
// must be OPEN: after the first store recovery completes, finish the other
// role's recovery and initialize exactly once per authority.
const initializedAuthorities = new Set()
async function openStore (authority, r, capabilities, overrides) {
const sink = beginForwardHttpsAggregateQuotaRecoveryV3(capabilities.targetStoreQuotaCapability)
const store = await openForwardHttpsTargetStoreV3(storeOptions(r, capabilities, { targetQuotaRecoverySink: sink, ...overrides }))
const targetFinal = await finishForwardHttpsAggregateQuotaRecoveryV3(sink)
if (!initializedAuthorities.has(authority)) {
initializedAuthorities.add(authority)
const sourceSink = beginForwardHttpsAggregateQuotaRecoveryV3(capabilities.sourceStoreQuotaCapability)
const sourceFinal = await finishForwardHttpsAggregateQuotaRecoveryV3(sourceSink)
await initializeForwardHttpsAggregateQuotaV3(authority, { sourceRecoveryFinalState: sourceFinal, targetRecoveryFinalState: targetFinal })
}
return store
}
async function directoryBytes (root) {
let total = 0
for (const entry of await fs.readdir(root, { withFileTypes: true })) {
const item = path.join(root, entry.name)
if (entry.isDirectory()) total += await directoryBytes(item)
else if (entry.isFile()) total += (await fs.lstat(item)).size
}
return total
}
// Raw recovery opens (no gate initialization): the recovery sink is begun by
// the caller and always finished, including when recovery rejects.
async function recoverWithSink (capabilities, options) {
const sink = beginForwardHttpsAggregateQuotaRecoveryV3(capabilities.targetStoreQuotaCapability)
try {
return await openForwardHttpsTargetStoreV3({ ...options, targetQuotaRecoverySink: sink })
} finally {
await finishForwardHttpsAggregateQuotaRecoveryV3(sink).catch(() => {})
}
}
const PLACEHOLDER_JOURNAL = Object.freeze({})
function storeOptions (r, capabilities, overrides = {}) {
return {
root: r['target-store'],
replayJournalAuthority: PLACEHOLDER_JOURNAL,
targetStoreQuotaCapability: capabilities.targetStoreQuotaCapability,
wireV3AbiHash: fixed(0x44),
privateIpcV4Hash: fixed(0x45),
signedLaunchTopologyHash: fixed(0x46),
storeId: fixed(0x41),
mapGeneration: 1n,
ownerFenceTokenHash: fixed(0x42),
durabilityContinuityHash: fixed(0x43),
targetSignerPublicKey: fixed(0x47),
targetSignerDescriptorSequence: 1n,
targetSignerDescriptorHash: fixed(0x48),
signResult: async () => b4a.alloc(64),
createResponderState: () => ({}),
advanceResponderIngress: () => {},
advanceResponderOutcome: () => {},
atRestKey: fixed(0x49),
epochSeconds: () => 1000000,
monotonicMillis: () => Date.now(),
...overrides
}
}
test('target store: fresh OPEN TURN_FINAL allocates and type113 counts independently', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x51)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
const opened = await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id, body: b4a.alloc(8, 0x51) })
t.is(opened.walSequence, 1n)
// A type113 crypto reservation is the prefix frame of a two-frame
// TURN_FINAL operation: 113 then the 112 final, both removable entries
const reserved = await acceptForwardedHttpsTargetTurnV3(store, {
stableSessionId: id,
walType: 113,
body: b4a.alloc(118 - 36, 0x52),
plannedRemovableChargeEntryCount: 2
})
t.is(reserved.payload.byteLength, 118)
const slot = store.slots.get(b4a.toString(id, 'hex'))
t.is(slot.registry.count, 3)
const status = forwardHttpsTargetStoreV3Status(store)
t.is(status.unconsumedSlots, status.slotCapacity - 1)
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
const recoveredSlot = reopened.slots.get(b4a.toString(id, 'hex'))
t.is(recoveredSlot.registry.count, 3)
})
test('target store: minimal absent-sequence terminal and terminal-only prune', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x52)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
const result = await driveTerminal(store, buildFtm9({
flags: 1,
stableSessionId: id,
sequence: 4n,
priorSessionRevision: 0n,
exactRequestCommitment: fixed(0x53),
expiresAtEpoch: 2000,
retainedUntilEpoch: 2900,
newTrustedEpochHighWatermark: 1500,
reason: 'FORWARD_HTTPS_TARGET_STORE_V3_SEQUENCE_INVALID'
}))
t.is(result.payload[5], 2) // TARGET role byte
t.is(result.payload.readUInt16BE(6), 1)
// The terminal-only FPR9 carries the FTM9-carried expiry clock and count0
await closeForwardHttpsTargetStoreV3(store)
const recoveredStore = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(recoveredStore).catch(() => {}) })
const slot = recoveredStore.slots.get(b4a.toString(id, 'hex'))
const fpr9 = encodeForwardHttpsRetentionPrunedV3({
role: 'TARGET_STORE',
stableSessionId: id,
priorSessionRevision: slot.priorRevision,
pruneEpochSeconds: 2901,
trustedEpochHighWatermark: 2901,
expiresAtEpoch: slot.expiresAtEpoch,
recoveryGraceUntilEpoch: slot.recoveryGraceUntilEpoch,
removedOrdinaryLogicalBytes: 0n,
chargeEntryCount: 0,
beforeAuthorityBitmap: slot.authorityBitmap,
allocationDisposition: 0,
terminalSlotState: 2,
chargeEntryBuffers: [],
authorityCommitments: slot.authorityCommitments
})
const pruned = await drivePrune(recoveredStore, fpr9)
t.is(pruned.payload.readUInt32BE(72), 0)
t.is(pruned.payload.readUInt32BE(76), 640)
t.is(pruned.payload[85], 2)
await closeForwardHttpsTargetStoreV3(recoveredStore)
const finalStore = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(finalStore).catch(() => {}) })
t.is(forwardHttpsTargetStoreV3Status(finalStore).consumedPrunedSlots, 1)
})
test('target store: existing-session terminalization and budget reason', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x54)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id })
const slot = store.slots.get(b4a.toString(id, 'hex'))
const terminal = await driveTerminal(store, buildFtm9({
flags: 0,
stableSessionId: id,
sequence: 9n,
priorSessionRevision: slot.priorRevision,
reason: 'BUDGET_EXHAUSTED',
newTrustedEpochHighWatermark: 5
}))
t.is(terminal.payload.byteLength, 192)
t.is(b4a.toString(terminal.payload.subarray(77, 77 + 16), 'ascii'), 'BUDGET_EXHAUSTED')
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
t.is(reopened.slots.get(b4a.toString(id, 'hex')).state, SLOT.CONSUMED_UNPRUNED)
await t.exception.all(acceptForwardedHttpsTargetTurnV3(reopened, { stableSessionId: id }), /TERMINAL/)
})
test('target store: cross-role frame recovery is INTEGRITY', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x55)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
// A source WAL type can never be appended through the target store
await t.exception.all(acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id, walType: 96 }), /not an ordinary target/)
})
test('prefix partition: FRESH type113 prefix claims exactly one PREFIX_ALLOCATED slot', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x56)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await rawAppend(store, 113, prefixPayload(id))
await rawAppend(store, 113, prefixPayload(id))
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
const slot = reopened.slots.get(b4a.toString(id, 'hex'))
t.is(slot.state, SLOT.PREFIX_ALLOCATED)
t.is(identityOf(slot), IDENTITY.PRESENT_PREFIX_ALLOCATED)
t.is(slot.registry.count, 2)
t.is(slot.orphan.entries.length, 2)
t.is(slot.orphan.removedSum, 920n)
t.is(slot.orphan.lastRevision, 2n)
const status = forwardHttpsTargetStoreV3Status(reopened)
t.is(status.unconsumedSlots, status.slotCapacity - 1, 'exactly one slot claimed')
})
test('prefix partition: EXISTING-session type113 prefix overlays ALLOCATED_WITH_PREFIX with complete preservation', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x57)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id, body: b4a.alloc(8, 0x51) })
const before = store.slots.get(b4a.toString(id, 'hex'))
t.is(before.registry.count, 1)
await rawAppend(store, 113, prefixPayload(id))
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
const slot = reopened.slots.get(b4a.toString(id, 'hex'))
t.is(slot.state, SLOT.ALLOCATED_WITH_PREFIX)
t.is(identityOf(slot), IDENTITY.ALLOCATED_WITH_PREFIX)
t.is(slot.registry.count, 2, 'pre-operation entry preserved plus the orphan entry')
t.is(slot.orphan.entries.length, 1)
t.is(slot.orphan.removedSum, 460n)
t.is(slot.orphan.lastRevision, 2n)
const status = forwardHttpsTargetStoreV3Status(reopened)
t.is(status.unconsumedSlots, status.slotCapacity - 1, 'no second slot is claimed')
})
test('prefix partition: matching final applies exactly once for both classes', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const freshId = fixed(0x58)
const existingId = fixed(0x59)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
// FRESH prefix completed by its matching final
await rawAppend(store, 113, prefixPayload(freshId))
await rawAppend(store, 112, finalPayload(freshId, 0x52))
// EXISTING prefix completed by its matching final
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: existingId })
await rawAppend(store, 113, prefixPayload(existingId))
await rawAppend(store, 114, finalPayload(existingId, 0x52, 0x55))
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
const fresh = reopened.slots.get(b4a.toString(freshId, 'hex'))
t.is(fresh.state, SLOT.ALLOCATED)
t.is(identityOf(fresh), IDENTITY.PRESENT_ALLOCATED)
t.is(fresh.orphan, null)
t.is(fresh.registry.count, 2)
const existing = reopened.slots.get(b4a.toString(existingId, 'hex'))
t.is(existing.state, SLOT.ALLOCATED)
t.is(identityOf(existing), IDENTITY.PRESENT_ALLOCATED)
t.is(existing.orphan, null)
t.is(existing.registry.count, 3)
const status = forwardHttpsTargetStoreV3Status(reopened)
t.is(status.unconsumedSlots, status.slotCapacity - 2)
})
test('per-step identity goldens: latest slot-disposing transition governs', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x5a)
// OPEN + target112 assigns PRESENT_ALLOCATED
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id })
t.is(identityOf(store.slots.get(b4a.toString(id, 'hex'))), IDENTITY.PRESENT_ALLOCATED)
// +k complete target113 (no final, no abort) assigns ALLOCATED_WITH_PREFIX
await rawAppend(store, 113, prefixPayload(id))
await rawAppend(store, 113, prefixPayload(id))
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
const slot = reopened.slots.get(b4a.toString(id, 'hex'))
t.is(identityOf(slot), IDENTITY.ALLOCATED_WITH_PREFIX)
t.is(slot.orphan.entries.length, 2)
t.is(slot.orphan.removedSum, 920n)
})
test('flags2 abort goldens: remove-exactly-orphan, vector byte-identical, PRESENT_ALLOCATED continuity and retry', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x5c)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id })
await rawAppend(store, 113, prefixPayload(id))
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
const slot = reopened.slots.get(b4a.toString(id, 'hex'))
t.is(slot.state, SLOT.ALLOCATED_WITH_PREFIX)
const beforeBitmap = slot.authorityBitmap
// A later committed operation (no 113) is admitted while the prefix is open
await acceptForwardedHttpsTargetTurnV3(reopened, { stableSessionId: id, walType: 112, body: b4a.alloc(8, 0x5d) })
t.is(slot.registry.count, 3)
// The complete flags2 abort: ordinary net admission, exact orphan removal
const fpr9 = encodeForwardHttpsRetentionPrunedV3({
role: 'TARGET_STORE',
flags: 2,
stableSessionId: id,
priorSessionRevision: slot.orphan.lastRevision,
pruneEpochSeconds: 8000,
trustedEpochHighWatermark: 8000,
expiresAtEpoch: 0,
recoveryGraceUntilEpoch: 0,
removedOrdinaryLogicalBytes: slot.orphan.removedSum,
chargeEntryCount: slot.orphan.entries.length,
beforeAuthorityBitmap: slot.authorityBitmap,
allocationDisposition: 2,
terminalSlotState: 1,
chargeEntryBuffers: slot.orphan.entries,
authorityCommitments: slot.authorityCommitments || Array.from({ length: 10 }, () => b4a.alloc(32))
})
const aborted = await drivePrune(reopened, fpr9)
t.is(aborted.payload.readUInt16BE(6), 2)
t.is(aborted.payload.readUInt32BE(56), 0)
t.is(aborted.payload.readUInt32BE(60), 0)
t.is(aborted.payload.readUInt32BE(72), 1)
t.is(aborted.payload[84], 2)
t.is(aborted.payload[85], 1)
// Carve-out proven through exact recovery: exactly the orphan entry removed,
// later entries preserved, vector byte-identical, slot retained,
// identity stays PRESENT_ALLOCATED, no PRUNED_RELEASED recorded.
await closeForwardHttpsTargetStoreV3(reopened)
const recovered = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(recovered).catch(() => {}) })
const recoveredSlot = recovered.slots.get(b4a.toString(id, 'hex'))
t.is(recoveredSlot.state, SLOT.ALLOCATED)
t.is(recoveredSlot.orphan, null)
t.is(recoveredSlot.registry.count, 2, 'ordinary open entry plus the later committed entry')
t.is(recoveredSlot.authorityBitmap, beforeBitmap)
t.absent(recoveredSlot.prunedReleased)
t.is(identityOf(recoveredSlot), IDENTITY.PRESENT_ALLOCATED)
// The mandated retry with fresh contiguous crypto revisions is admitted
const retry = await acceptForwardedHttpsTargetTurnV3(recovered, { stableSessionId: id, walType: 113, body: b4a.alloc(118 - 36, 0x5e) })
t.is(retry.payload.byteLength, 118)
t.is(recoveredSlot.registry.count, 4)
t.is(identityOf(recoveredSlot), IDENTITY.PRESENT_ALLOCATED)
// Recovery reclassifies the continuing session exactly once
await closeForwardHttpsTargetStoreV3(recovered)
const final = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(final).catch(() => {}) })
const finalSlot = final.slots.get(b4a.toString(id, 'hex'))
t.is(finalSlot.state, SLOT.ALLOCATED)
t.is(finalSlot.orphan, null)
t.is(finalSlot.registry.count, 4)
t.is(identityOf(finalSlot), IDENTITY.PRESENT_ALLOCATED)
})
test('flags1 orphan abort: PREFIX_ALLOCATED to FREE, PRUNED_RELEASED and mutation-free CONFLICT after', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x5f)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await rawAppend(store, 113, prefixPayload(id))
await rawAppend(store, 113, prefixPayload(id))
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
const slot = reopened.slots.get(b4a.toString(id, 'hex'))
t.is(slot.state, SLOT.PREFIX_ALLOCATED)
const base = {
role: 'TARGET_STORE',
stableSessionId: id,
priorSessionRevision: slot.orphan.lastRevision,
pruneEpochSeconds: 7000,
trustedEpochHighWatermark: 7000,
expiresAtEpoch: 0,
recoveryGraceUntilEpoch: 0,
removedOrdinaryLogicalBytes: slot.orphan.removedSum,
chargeEntryCount: slot.orphan.entries.length,
beforeAuthorityBitmap: 0,
chargeEntryBuffers: slot.orphan.entries,
authorityCommitments: Array.from({ length: 10 }, () => b4a.alloc(32))
}
const fpr9 = encodeForwardHttpsRetentionPrunedV3({ ...base, flags: 1, allocationDisposition: 1, terminalSlotState: 3 })
const aborted = await drivePrune(reopened, fpr9)
t.is(aborted.payload.readUInt16BE(6), 1)
t.is(aborted.payload.readUInt32BE(72), 2)
t.is(aborted.payload[84], 1)
t.is(aborted.payload[85], 3)
await closeForwardHttpsTargetStoreV3(reopened)
const recovered = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(recovered).catch(() => {}) })
const recoveredSlot = recovered.slots.get(b4a.toString(id, 'hex'))
t.is(recoveredSlot.state, SLOT.FREE)
t.is(recoveredSlot.orphan, null)
t.is(recoveredSlot.registry.count, 0)
t.ok(recoveredSlot.prunedReleased)
t.is(identityOf(recoveredSlot), IDENTITY.PRUNED_RELEASED)
const status = forwardHttpsTargetStoreV3Status(recovered)
t.is(status.unconsumedSlots, status.slotCapacity)
t.is(status.roleGlobalLogicalBytes, 736)
// PRUNED_RELEASED returns mutation-free CONFLICT, never a NEVER_SEEN wedge
let conflict = null
try { await acceptForwardedHttpsTargetTurnV3(recovered, { stableSessionId: id }) } catch (error) { conflict = error }
t.is(conflict && conflict.code, 'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_CONFLICT')
t.is(forwardHttpsTargetStoreV3Status(recovered).unconsumedSlots, status.slotCapacity, 'no mutation from the CONFLICT')
// flags1/flags2 prefix variants require their exact recovered prefix class
const headBefore = forwardHttpsTargetStoreV3Status(recovered).walHeadSequence
const repeat1 = encodeForwardHttpsRetentionPrunedV3({ ...base, flags: 1, allocationDisposition: 1, terminalSlotState: 3 })
await t.exception.all(drivePrune(recovered, repeat1), /prefix variant|prefix class|INTEGRITY/)
const repeat2 = encodeForwardHttpsRetentionPrunedV3({ ...base, flags: 2, allocationDisposition: 2, terminalSlotState: 1 })
await t.exception.all(drivePrune(recovered, repeat2), /prefix variant|prefix class|INTEGRITY/)
t.is(forwardHttpsTargetStoreV3Status(recovered).walHeadSequence, headBefore, 'rejections are pre-WAL with zero mutation')
})
test('prefix-abort ordinary admission: exact inequality equality admits, ceiling+1 CAPACITY with zero mutation', async t => {
// Each case builds its own crashed-prefix WAL so the equality admission
// cannot mutate the evidence for the ceiling+1 case.
for (const equality of [true, false]) {
const r = await roots(t)
const { authority, capabilities } = await quota2(t, r)
const id = fixed(0x60)
const store = await openStore(authority, r, capabilities)
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id })
await rawAppend(store, 113, prefixPayload(id))
await closeForwardHttpsTargetStoreV3(store)
const measured = await directoryBytes(r['target-store'])
// The abort-time role logical: the committed open-frame charge from the
// ledger plus the raw orphan type113 frame's exact 460 charge, which the
// recovery replay seeds authoritatively.
const roleLogical = forwardHttpsAggregateQuotaV3Status(authority).targetLogicalChargedBytes + 460
await closeForwardHttpsAggregateQuotaV3(authority)
// Exact boundary: current physical + 480 planned + one unconsumed identity
// at 896, and the seeded role logical plus net 276 plus the 1344
// unconsumed liability, against the per-store ceiling.
const perStore = Math.max(measured + 480 + 896, roleLogical + 276 + 1344) - (equality ? 0 : 1)
const authority2 = await openForwardHttpsAggregateQuotaV3({
sourceReplayRoot: r['source-replay'],
targetReplayRoot: r['target-replay'],
sourceStoreRoot: r['source-store'],
targetStoreRoot: r['target-store'],
maximumDurableBytesPerStore: perStore,
maximumForwardStorageBytesAggregate: 2 * perStore,
monotonicMillis: () => Date.now(),
callbackTimeoutMillis: 15000,
faultInjector: null
})
t.teardown(async () => { await closeForwardHttpsAggregateQuotaV3(authority2).catch(() => {}) })
const capabilities2 = mintForwardHttpsAggregateQuotaCapabilitiesV3(authority2)
const reopened = await openStore(authority2, r, capabilities2, {
limits: {
maximumRetainedTurnsPerRole: 1,
maximumDurableBytesPerStore: perStore,
maximumForwardStorageBytesAggregate: 2 * perStore
}
})
const slot = reopened.slots.get(b4a.toString(id, 'hex'))
t.is(slot.state, SLOT.ALLOCATED_WITH_PREFIX)
const headBefore = forwardHttpsTargetStoreV3Status(reopened).walHeadSequence
const fpr9 = encodeForwardHttpsRetentionPrunedV3({
role: 'TARGET_STORE',
flags: 2,
stableSessionId: id,
priorSessionRevision: slot.orphan.lastRevision,
pruneEpochSeconds: 9000,
trustedEpochHighWatermark: 9000,
expiresAtEpoch: 0,
recoveryGraceUntilEpoch: 0,
removedOrdinaryLogicalBytes: slot.orphan.removedSum,
chargeEntryCount: slot.orphan.entries.length,
beforeAuthorityBitmap: slot.authorityBitmap,
allocationDisposition: 2,
terminalSlotState: 1,
chargeEntryBuffers: slot.orphan.entries,
authorityCommitments: Array.from({ length: 10 }, () => b4a.alloc(32))
})
if (equality) {
const aborted = await drivePrune(reopened, fpr9)
t.is(aborted.payload.readUInt16BE(6), 2)
await closeForwardHttpsTargetStoreV3(reopened)
const recovered = await openStore(authority2, r, capabilities2)
const recoveredSlot = recovered.slots.get(b4a.toString(id, 'hex'))
t.is(recoveredSlot.state, SLOT.ALLOCATED)
t.is(identityOf(recoveredSlot), IDENTITY.PRESENT_ALLOCATED)
await closeForwardHttpsTargetStoreV3(recovered)
} else {
let denied = null
try { await drivePrune(reopened, fpr9) } catch (error) { denied = error }
t.is(denied && denied.code, 'FORWARD_HTTPS_AGGREGATE_QUOTA_V3_CAPACITY')
// Zero mutation: WAL head, slot, orphan and registry are all unchanged
t.is(forwardHttpsTargetStoreV3Status(reopened).walHeadSequence, headBefore)
t.is(slot.state, SLOT.ALLOCATED_WITH_PREFIX)
t.is(slot.registry.count, 2)
t.is(slot.orphan.entries.length, 1)
t.is(forwardHttpsTargetStoreV3Status(reopened).roleGlobalLogicalBytes, 0)
}
await closeForwardHttpsTargetStoreV3(reopened)
}
})
test('flags2 abort requires an open existing-session prefix; mixed commitment is INTEGRITY', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x61)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id })
// flags2 on a plain ALLOCATED session (no open prefix) rejects pre-WAL
const slot = store.slots.get(b4a.toString(id, 'hex'))
const fpr9 = encodeForwardHttpsRetentionPrunedV3({
role: 'TARGET_STORE',
flags: 2,
stableSessionId: id,
priorSessionRevision: slot.priorRevision,
pruneEpochSeconds: 100,
trustedEpochHighWatermark: 100,
expiresAtEpoch: 0,
recoveryGraceUntilEpoch: 0,
removedOrdinaryLogicalBytes: slot.registry.removedLogicalBytes(),
chargeEntryCount: slot.registry.count,
beforeAuthorityBitmap: 0,
allocationDisposition: 2,
terminalSlotState: 1,
chargeEntryBuffers: slot.registry.entriesAscending(),
authorityCommitments: Array.from({ length: 10 }, () => b4a.alloc(32))
})
const headBefore = forwardHttpsTargetStoreV3Status(store).walHeadSequence
await t.exception.all(drivePrune(store, fpr9), /prefix class|prefix variant|INTEGRITY/)
t.is(forwardHttpsTargetStoreV3Status(store).walHeadSequence, headBefore)
// A mixed requestCommitment inside one run is INTEGRITY in recovery
await rawAppend(store, 113, prefixPayload(id, 0x52))
await rawAppend(store, 113, prefixPayload(id, 0x53))
await closeForwardHttpsTargetStoreV3(store)
await t.exception.all(openStore(authority, r, capabilities), /mixed prefix requestCommitment|INTEGRITY/)
})
test('overlay terminalization: flags0 on ALLOCATED_WITH_PREFIX with orphan persistence and no flags2 after', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x62)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id })
await rawAppend(store, 113, prefixPayload(id))
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
const slot = reopened.slots.get(b4a.toString(id, 'hex'))
t.is(slot.state, SLOT.ALLOCATED_WITH_PREFIX)
t.is(slot.orphan.lastRevision, 2n)
// flags0 SESSION_TERMINAL uses the exact orphan-last revision floor,
// preserves the vector and moves the slot to CONSUMED_UNPRUNED
const terminal = await driveTerminal(reopened, buildFtm9({
flags: 0,
stableSessionId: id,
sequence: 7n,
priorSessionRevision: slot.priorRevision,
reason: 'CHAIN_INVALID',
newTrustedEpochHighWatermark: 9
}))
t.is(terminal.payload.readUInt16BE(6), 0)
t.is(Number(terminal.payload.readBigUInt64BE(64)), 2)
await closeForwardHttpsTargetStoreV3(reopened)
const consumed = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(consumed).catch(() => {}) })
const consumedSlot = consumed.slots.get(b4a.toString(id, 'hex'))
t.is(consumedSlot.state, SLOT.CONSUMED_UNPRUNED)
t.is(consumedSlot.orphan, null)
t.is(consumedSlot.registry.count, 2, 'orphan entries persist into the consumed registry')
t.is(identityOf(consumedSlot), IDENTITY.PRESENT_CONSUMED_UNPRUNED)
// No flags2 abort is possible after terminalization
const flags2Attempt = encodeForwardHttpsRetentionPrunedV3({
role: 'TARGET_STORE',
flags: 2,
stableSessionId: id,
priorSessionRevision: 2n,
pruneEpochSeconds: 9500,
trustedEpochHighWatermark: 9500,
expiresAtEpoch: 0,
recoveryGraceUntilEpoch: 0,
removedOrdinaryLogicalBytes: consumedSlot.registry.removedLogicalBytes(),
chargeEntryCount: consumedSlot.registry.count,
beforeAuthorityBitmap: consumedSlot.authorityBitmap,
allocationDisposition: 2,
terminalSlotState: 1,
chargeEntryBuffers: consumedSlot.registry.entriesAscending(),
authorityCommitments: consumedSlot.authorityCommitments || Array.from({ length: 10 }, () => b4a.alloc(32))
})
await t.exception.all(drivePrune(consumed, flags2Attempt), /prefix class|prefix variant|INTEGRITY/)
// The later terminal-existing FPR9 removes the persisted orphan entries
const fpr9 = encodeForwardHttpsRetentionPrunedV3({
role: 'TARGET_STORE',
stableSessionId: id,
priorSessionRevision: consumedSlot.priorRevision,
pruneEpochSeconds: 9600,
trustedEpochHighWatermark: 9600,
expiresAtEpoch: 0,
recoveryGraceUntilEpoch: 0,
removedOrdinaryLogicalBytes: consumedSlot.registry.removedLogicalBytes(),
chargeEntryCount: consumedSlot.registry.count,
beforeAuthorityBitmap: consumedSlot.authorityBitmap,
allocationDisposition: 0,
terminalSlotState: 2,
chargeEntryBuffers: consumedSlot.registry.entriesAscending(),
authorityCommitments: consumedSlot.authorityCommitments
})
const pruned = await drivePrune(consumed, fpr9)
t.is(pruned.payload.readUInt32BE(72), 2)
// Recovery reproduces the exact consumed-pruned disposition
await closeForwardHttpsTargetStoreV3(consumed)
const recovered = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(recovered).catch(() => {}) })
const recoveredSlot = recovered.slots.get(b4a.toString(id, 'hex'))
t.is(recoveredSlot.state, SLOT.CONSUMED_PRUNED)
t.is(identityOf(recoveredSlot), IDENTITY.PRUNED_CONSUMED)
})
test('store-level precedence: PRESENT_PREFIX_ALLOCATED is SESSION_CLOSED, consumed is TERMINAL, pruned is CONFLICT', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const prefixId = fixed(0x63)
const consumedId = fixed(0x64)
const prunedId = fixed(0x65)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
// PREFIX_ALLOCATED: mutation-free SESSION_CLOSED before any other work
await rawAppend(store, 113, prefixPayload(prefixId))
// consumed identity: sticky TERMINAL
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: consumedId })
const consumedSlot = store.slots.get(b4a.toString(consumedId, 'hex'))
await driveTerminal(store, buildFtm9({ flags: 0, stableSessionId: consumedId, sequence: 3n, priorSessionRevision: consumedSlot.priorRevision, reason: 'CHAIN_INVALID' }))
// pruned identity: CONFLICT
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: prunedId })
const prunedSlot = store.slots.get(b4a.toString(prunedId, 'hex'))
await drivePrune(store, encodeForwardHttpsRetentionPrunedV3({
role: 'TARGET_STORE',
stableSessionId: prunedId,
priorSessionRevision: prunedSlot.priorRevision,
pruneEpochSeconds: 500,
trustedEpochHighWatermark: 500,
expiresAtEpoch: 0,
recoveryGraceUntilEpoch: 0,
removedOrdinaryLogicalBytes: prunedSlot.registry.removedLogicalBytes(),
chargeEntryCount: prunedSlot.registry.count,
beforeAuthorityBitmap: 0,
allocationDisposition: 1,
terminalSlotState: 1,
chargeEntryBuffers: prunedSlot.registry.entriesAscending(),
authorityCommitments: Array.from({ length: 10 }, () => b4a.alloc(32))
}))
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
t.is(identityOf(reopened.slots.get(b4a.toString(prefixId, 'hex'))), IDENTITY.PRESENT_PREFIX_ALLOCATED)
const headBefore = forwardHttpsTargetStoreV3Status(reopened).walHeadSequence
let sessionClosed = null
try { await acceptForwardedHttpsTargetTurnV3(reopened, { stableSessionId: prefixId, walType: 112 }) } catch (error) { sessionClosed = error }
t.is(sessionClosed && sessionClosed.code, 'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_SESSION_CLOSED')
let terminal = null
try { await acceptForwardedHttpsTargetTurnV3(reopened, { stableSessionId: consumedId, walType: 112 }) } catch (error) { terminal = error }
t.is(terminal && terminal.code, 'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_TERMINAL')
let conflict = null
try { await acceptForwardedHttpsTargetTurnV3(reopened, { stableSessionId: prunedId, walType: 112 }) } catch (error) { conflict = error }
t.is(conflict && conflict.code, 'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_CONFLICT')
// All three rejections are mutation-free
t.is(forwardHttpsTargetStoreV3Status(reopened).walHeadSequence, headBefore)
})
test('matching-final wrong-operation: non-matching final on a fresh prefix is INTEGRITY, never admitted', async t => {
const r = await roots(t)
const { capabilities } = await quota(t, r)
const id = fixed(0x66)
const store = await recoverWithSink(capabilities, storeOptions(r, capabilities))
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await rawAppend(store, 113, prefixPayload(id))
// A final of a DIFFERENT operation (different requestCommitment) follows
await rawAppend(store, 112, finalPayload(id, 0x53))
await closeForwardHttpsTargetStoreV3(store)
// Recovery has exactly one disposition: INTEGRITY, never an admission
await t.exception.all(recoverWithSink(capabilities, storeOptions(r, capabilities)), /non-matching final|INTEGRITY/)
})
test('recovered tombstone tamper: count and commitment mismatches are INTEGRITY at recovery', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x67)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id })
const slot = store.slots.get(b4a.toString(id, 'hex'))
const { encodeForwardHttpsRetentionPrunedV3 } = await import('../forward-https-replay-journal-v4.js')
const base = {
role: 'TARGET_STORE',
stableSessionId: id,
priorSessionRevision: slot.priorRevision,
pruneEpochSeconds: 500,
trustedEpochHighWatermark: 500,
expiresAtEpoch: slot.expiresAtEpoch,
recoveryGraceUntilEpoch: slot.recoveryGraceUntilEpoch,
removedOrdinaryLogicalBytes: slot.registry.removedLogicalBytes(),
chargeEntryCount: slot.registry.count,
beforeAuthorityBitmap: slot.authorityBitmap,
allocationDisposition: 1,
terminalSlotState: 1,
chargeEntryBuffers: slot.registry.entriesAscending(),
authorityCommitments: Array.from({ length: 10 }, () => b4a.alloc(32))
}
// count=2 against a 1-entry registry (review probe)
const countTampered = encodeForwardHttpsRetentionPrunedV3({ ...base, chargeEntryCount: 1 })
countTampered.writeUInt32BE(2, 72)
await rawAppend(store, 118, countTampered)
await closeForwardHttpsTargetStoreV3(store)
await t.exception.all(recoverWithSink(capabilities, storeOptions(r, capabilities)), /independently match|INTEGRITY/)
})
test('fresh OPEN on PRESENT_PREFIX_ALLOCATED is mutation-free SESSION_CLOSED', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x68)
const store = await recoverWithSink(capabilities, storeOptions(r, capabilities))
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await rawAppend(store, 113, prefixPayload(id))
await closeForwardHttpsTargetStoreV3(store)
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
const headBefore = forwardHttpsTargetStoreV3Status(reopened).walHeadSequence
let closed = null
try { await acceptForwardedHttpsTargetTurnV3(reopened, { stableSessionId: id }) } catch (error) { closed = error }
t.is(closed && closed.code, 'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_SESSION_CLOSED')
t.is(forwardHttpsTargetStoreV3Status(reopened).walHeadSequence, headBefore)
})
test('open ABI: exact required key set enforced and fault registry carries the exact key sets', async t => {
const r = await roots(t)
const { capabilities } = await quota(t, r)
t.is(Object.keys(FORWARD_HTTPS_TARGET_STORE_V3_FAULT_POINT).length, 22)
t.is(Object.keys(FORWARD_HTTPS_SOURCE_STORE_V3_FAULT_POINT).length, 11)
t.is(Object.keys(FORWARD_HTTPS_STORAGE_V3_FAULT_POINT).length, 12)
const good = storeOptions(r, capabilities)
const store = await recoverWithSink(capabilities, good)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
// Missing each required key rejects before any mutation. The recovery sink
// key itself is probed without a sink supplied.
for (const key of ['replayJournalAuthority', 'targetStoreQuotaCapability', 'wireV3AbiHash', 'privateIpcV4Hash', 'signedLaunchTopologyHash', 'storeId', 'mapGeneration', 'ownerFenceTokenHash', 'durabilityContinuityHash', 'targetSignerPublicKey', 'targetSignerDescriptorSequence', 'targetSignerDescriptorHash', 'signResult', 'createResponderState', 'advanceResponderIngress', 'advanceResponderOutcome', 'atRestKey', 'epochSeconds', 'monotonicMillis']) {
const missing = { ...good }
delete missing[key]
await t.exception.all(recoverWithSink(capabilities, missing), new RegExp(key))
}
const noSink = { ...good }
delete noSink.targetQuotaRecoverySink
await t.exception.all(openForwardHttpsTargetStoreV3(noSink), /targetQuotaRecoverySink/)
// Unknown keys reject
await t.exception.all(recoverWithSink(capabilities, { ...good, bogusKey: 1 }), /unknown field/)
})
test('quota gate: OPEN admits, fail-WAL enters FAILED_WAL_OUTCOME_UNKNOWN_PENDING and blocks all work', async t => {
const r = await roots(t)
const authority = await openForwardHttpsAggregateQuotaV3({
sourceReplayRoot: r['source-replay'],
targetReplayRoot: r['target-replay'],
sourceStoreRoot: r['source-store'],
targetStoreRoot: r['target-store'],
maximumDurableBytesPerStore: 8589934592,
maximumForwardStorageBytesAggregate: 17179869184,
monotonicMillis: () => Date.now(),
callbackTimeoutMillis: 15000,
faultInjector: null
})
t.teardown(async () => { await closeForwardHttpsAggregateQuotaV3(authority).catch(() => {}) })
const caps = mintForwardHttpsAggregateQuotaCapabilitiesV3(authority)
// Gate before initialize: quota is not yet OPEN
await t.exception.all(() => assertForwardHttpsAggregateQuotaOperationalV3(caps.targetStoreQuotaCapability), /not operational/)
// Initialize from empty role roots through the recovery claim ABI: claims
// are sink-private; finish returns the final recovery state.
const sourceSink = beginForwardHttpsAggregateQuotaRecoveryV3(caps.sourceStoreQuotaCapability)
const sourceFinal = await finishForwardHttpsAggregateQuotaRecoveryV3(sourceSink)
const targetSink = beginForwardHttpsAggregateQuotaRecoveryV3(caps.targetStoreQuotaCapability)
const targetFinal = await finishForwardHttpsAggregateQuotaRecoveryV3(targetSink)
await initializeForwardHttpsAggregateQuotaV3(authority, { sourceRecoveryFinalState: sourceFinal, targetRecoveryFinalState: targetFinal })
// OPEN and localOperational: the gate succeeds without mutation
t.is(assertForwardHttpsAggregateQuotaOperationalV3(caps.targetStoreQuotaCapability), undefined)
t.is(forwardHttpsAggregateQuotaV3Status(authority).state, 'OPEN')
// fail-WAL: absorbing FAILED_WAL_OUTCOME_UNKNOWN_PENDING; gate and later work reject
failForwardHttpsAggregateQuotaWalAttemptV3(caps.targetStoreQuotaCapability)
const failed = forwardHttpsAggregateQuotaV3Status(authority)
t.is(failed.state, 'FAILED_WAL_OUTCOME_UNKNOWN_PENDING')
t.absent(failed.localOperational)
t.is(failed.blocker, 'FORWARD_HTTPS_AGGREGATE_QUOTA_V3_INTEGRITY')
await t.exception.all(() => assertForwardHttpsAggregateQuotaOperationalV3(caps.targetStoreQuotaCapability), /not operational/)
})
test('target store: kill after complete tombstone recovers applied exactly once', async t => {
const base = await createBlindBoundaryScratch('fhkill-test-')
t.teardown(async () => { await removeBlindBoundaryScratch(base) })
for (const name of ['source-replay', 'target-replay', 'source-store', 'target-store']) {
await fs.mkdir(path.join(base, name), { mode: 0o700 })
await fs.chmod(path.join(base, name), 0o700)
}
const fixture = new URL('./forward-https-storage-kill-fixture.mjs', import.meta.url)
const fixturePath = fileURLToPath(fixture)
const setup = await execFileAsync(process.execPath, [fixturePath, 'setup', base])
t.is(JSON.parse(setup.stdout).walSequence, '1')
await t.exception.all(execFileAsync(process.execPath, [fixturePath, 'crash', base]))
const verify = await execFileAsync(process.execPath, [fixturePath, 'verify', base])
const recovered = JSON.parse(verify.stdout)
t.is(recovered.state, 'OPEN')
t.is(recovered.walHeadSequence, '2')
t.is(recovered.slotState, 'FREE')
t.ok(recovered.prunedReleased)
t.is(recovered.unconsumedSlots, recovered.slotCapacity)
})
test('fault registry: OPEN_AFTER_RECOVERY and CLOSE_BEFORE_STORE_CLOSE fire with coded errors', async t => {
// The module-level points fire alongside the engine points on a plain
// open/close lifecycle.
const r1 = await roots(t)
const { authority: a1, capabilities: c1 } = await quota(t, r1)
const observed = []
const contexts = []
const store = await openStore(a1, r1, c1, { faultInjector: async (point, context) => { observed.push(point); contexts.push(context) } })
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
t.ok(observed.includes(FORWARD_HTTPS_TARGET_STORE_V3_FAULT_POINT.OPEN_AFTER_RECOVERY), 'open fault point fired')
await closeForwardHttpsTargetStoreV3(store)
t.ok(observed.includes(FORWARD_HTTPS_TARGET_STORE_V3_FAULT_POINT.CLOSE_BEFORE_STORE_CLOSE), 'close fault point fired')
// Only the module's frozen points reach the caller, never raw engine points
// or a context argument.
t.ok(observed.every(point => Object.values(FORWARD_HTTPS_TARGET_STORE_V3_FAULT_POINT).includes(point)), 'no raw engine points escape')
t.ok(contexts.every(context => context === undefined), 'no context is passed')
// Open fault: the injector failure is a coded INTEGRITY rejection of open
// with no live engine left behind.
const r2 = await roots(t)
const { authority: a2, capabilities: c2 } = await quota(t, r2)
await t.exception.all(openStore(a2, r2, c2, {
faultInjector: async point => {
if (point === FORWARD_HTTPS_TARGET_STORE_V3_FAULT_POINT.OPEN_AFTER_RECOVERY) throw new Error('injected open fault')
}
}), /fault injector failed at OPEN_AFTER_RECOVERY|INTEGRITY/)
// A non-undefined injector return maps to coded INTEGRITY as well.
const r2b = await roots(t)
const { authority: a2b, capabilities: c2b } = await quota(t, r2b)
await t.exception.all(openStore(a2b, r2b, c2b, {
faultInjector: async point => {
if (point === FORWARD_HTTPS_TARGET_STORE_V3_FAULT_POINT.OPEN_AFTER_RECOVERY) return 'injected'
return undefined
}
}), /returned a value|INTEGRITY/)
// Close fault: the first close reports coded INTEGRITY, the store still
// reaches CLOSED with blocker INTEGRITY and the exact-owner repeat resolves.
const r3 = await roots(t)
const { authority: a3, capabilities: c3 } = await quota(t, r3)
const store3 = await openStore(a3, r3, c3, {
faultInjector: async point => {
if (point === FORWARD_HTTPS_TARGET_STORE_V3_FAULT_POINT.CLOSE_BEFORE_STORE_CLOSE) throw new Error('injected close fault')
}
})
let closeError = null
try {
await closeForwardHttpsTargetStoreV3(store3)
} catch (error) {
closeError = error
}
t.ok(closeError, 'the first close reports the fault')
t.is(closeError.code, 'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_INTEGRITY')
t.is(store3.closed, true)
t.is(forwardHttpsTargetStoreV3Status(store3).blocker, 'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_INTEGRITY')
await closeForwardHttpsTargetStoreV3(store3)
})
test('readiness flags: the target store STATUS constant is not implementation-ready', async t => {
t.is(FORWARD_HTTPS_TARGET_STORE_V3_STATUS.implementationReady, false)
t.is(FORWARD_HTTPS_TARGET_STORE_V3_STATUS.runtimeReady, false)
t.is(FORWARD_HTTPS_TARGET_STORE_V3_STATUS.releaseReady, false)
t.is(FORWARD_HTTPS_TARGET_STORE_V3_STATUS.authorizesRelease, false)
})
test('open ABI: a recovery sink begun on a foreign quota authority rejects at open (G5)', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x69)
// Pre-populate the target-store WAL with one live session.
const store = await openStore(authority, r, capabilities)
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id, body: b4a.alloc(8, 0x51) })
await closeForwardHttpsTargetStoreV3(store)
// Store capability from the roots' authority but a sink begun on a foreign
// authority: the first recovered frame rejects the miswire with a coded
// AUTHORITY_INVALID and no state split.
const rB = await roots(t)
const { capabilities: foreign } = await quota(t, rB)
const foreignSink = beginForwardHttpsAggregateQuotaRecoveryV3(foreign.targetStoreQuotaCapability)
await t.exception.all(
openForwardHttpsTargetStoreV3(storeOptions(r, capabilities, { targetQuotaRecoverySink: foreignSink })),
/not bound to the presented store capability|AUTHORITY_INVALID/
)
await finishForwardHttpsAggregateQuotaRecoveryV3(foreignSink).catch(() => {})
// The correctly bound reopen recovers the session exactly.
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
t.is(forwardHttpsTargetStoreV3Status(reopened).walHeadSequence, 1n)
})
test('wired cap+1 terminal conversion: the 65535-entry session ends CONSUMED_UNPRUNED on reopen (required_tests[15])', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x6f)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
// Seed 65535 charge entries through raw WAL appends (group-committed).
const seedPayload = b4a.concat([b4a.from('FTS3', 'ascii'), id, b4a.alloc(8, 0x51)])
for (let chunk = 0; chunk * 500 < 65535; chunk++) {
const appends = []
for (let index = 0; index < 500 && chunk * 500 + index < 65535; index++) {
appends.push(store.store.appendAndApply({ type: 112, transactionId: b4a.alloc(32, 0x66), virtualBucket: 0, payload: seedPayload }, () => {}))
}
await Promise.all(appends)
}
await closeForwardHttpsTargetStoreV3(store)
const seeded = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(seeded).catch(() => {}) })
t.is(seeded.slots.get(b4a.toString(id, 'hex')).registry.count, 65535)
// The wired turn at cap+1 terminalizes through the ENTRY_CAP arm, then
// surfaces BUDGET_EXHAUSTED to the caller.
await t.exception.all(acceptForwardedHttpsTargetTurnV3(seeded, { stableSessionId: id, walType: 112 }), /BUDGET_EXHAUSTED|cap exceeded/)
t.is(seeded.slots.get(b4a.toString(id, 'hex')).state, SLOT.CONSUMED_UNPRUNED)
await closeForwardHttpsTargetStoreV3(seeded)
// On reopen the session is durably CONSUMED_UNPRUNED and the terminal
// identity is sticky mutation-free.
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
t.is(reopened.slots.get(b4a.toString(id, 'hex')).state, SLOT.CONSUMED_UNPRUNED)
await t.exception.all(acceptForwardedHttpsTargetTurnV3(reopened, { stableSessionId: id, walType: 112 }), /TERMINAL/)
})
test('planned hint above the exact operation row rejects at the boundary with zero mutation (probe w1)', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x71)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id, body: b4a.alloc(4, 0x51) })
const headBefore = forwardHttpsTargetStoreV3Status(store).walHeadSequence
// plannedRemovableChargeEntryCount=65536 on a healthy session: rejected at
// the boundary before any reservation or WAL mutation (REREVIEW3-P1-001
// probe (i)).
await t.exception.all(
acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id, walType: 112, plannedRemovableChargeEntryCount: 65536 }),
/exceeds the exact operation row|TypeError/
)
t.is(forwardHttpsTargetStoreV3Status(store).walHeadSequence, headBefore)
t.absent(forwardHttpsAggregateQuotaV3Status(authority).pendingReservation)
// A lawful turn with an in-row hint succeeds; close succeeds.
await acceptForwardedHttpsTargetTurnV3(store, { stableSessionId: id, walType: 112, plannedRemovableChargeEntryCount: 2 })
await closeForwardHttpsTargetStoreV3(store)
await closeForwardHttpsAggregateQuotaV3(authority)
t.is(forwardHttpsAggregateQuotaV3Status(authority).state, 'CLOSED')
})
test('cap decision mismatch releases the reservation: flags2 abort then wired cap+1 rejects cleanly, authority stays OPEN (probe w3a)', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x72)
const c1 = fixed(0xc1)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
// Seed 65533 ordinary entries plus 2 orphan prefix entries (registry and
// mirror both at 65535, EXISTING prefix) through group-committed appends.
const seedPayload = b4a.concat([b4a.from('FTS3', 'ascii'), id, b4a.alloc(8, 0x51)])
for (let chunk = 0; chunk * 500 < 65533; chunk++) {
const appends = []
for (let index = 0; index < 500 && chunk * 500 + index < 65533; index++) {
appends.push(store.store.appendAndApply({ type: 112, transactionId: b4a.alloc(32, 0x66), virtualBucket: 0, payload: seedPayload }, () => {}))
}
await Promise.all(appends)
}
const o1 = b4a.concat([b4a.from('FTS3', 'ascii'), id, c1, b4a.alloc(118 - 68, 0x01)])
const o2 = b4a.concat([b4a.from('FTS3', 'ascii'), id, c1, b4a.alloc(118 - 68, 0x02)])
await store.store.appendAndApply({ type: 113, transactionId: b4a.alloc(32, 0x66), virtualBucket: 0, payload: o1 }, () => {})
await store.store.appendAndApply({ type: 113, transactionId: b4a.alloc(32, 0x66), virtualBucket: 0, payload: o2 }, () => {})
await closeForwardHttpsTargetStoreV3(store)
const seeded = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(seeded).catch(() => {}) })
t.is(seeded.slots.get(b4a.toString(id, 'hex')).registry.count, 65535)
// A raw-quota flags2 abort removes the two orphans from the canonical
// mirror only (the store registry keeps 65535).
const abortEntry = (walSequence, payload) => {
const entry = b4a.alloc(49)
entry[0] = 113
let current = walSequence
for (let index = 8; index >= 1; index--) { entry[index] = Number(current & 0xffn); current >>= 8n }
b4a.copy(blake2b256(payload), entry, 9)
let cost = 460n
for (let index = 48; index >= 41; index--) { entry[index] = Number(cost & 0xffn); cost >>= 8n }
return entry
}
const abort = encodeForwardHttpsRetentionPrunedV3({
role: 'TARGET_STORE',
stableSessionId: id,
priorSessionRevision: 65535n,
pruneEpochSeconds: 10,
trustedEpochHighWatermark: 10,
expiresAtEpoch: 0,
recoveryGraceUntilEpoch: 0,
removedOrdinaryLogicalBytes: 920n,
chargeEntryCount: 2,
beforeAuthorityBitmap: 0,
allocationDisposition: 2,
terminalSlotState: 1,
chargeEntryBuffers: [abortEntry(65534n, o1), abortEntry(65535n, o2)],
authorityCommitments: Array.from({ length: 10 }, () => b4a.alloc(32)),
flags: 2
})
const abortPlan = createForwardHttpsStoreQuotaCostPlanV3(seeded.storeQuotaCapability, {
operation: 'PRUNE',
knownInputBuffers: [abort],
temporaryWriteBuffers: [],
existingDestinationBytes: 0
})
const abortUnion = await reserveForwardHttpsAggregateQuotaV3(seeded.storeQuotaCapability, abortPlan)
const { transitionAuthority } = bindForwardHttpsStoreQuotaActualBuffersV3(seeded.storeQuotaCapability, abortUnion.reservation, {
logicalRecordBuffers: [],
encryptedPlaintextBuffers: [],
finalWalMetadataBuffers: [abort],
temporaryWriteBuffers: []
})
const abortWalHash = blake2b256(b4a.concat([b4a.from('wal'), abort]))
await applyForwardHttpsAggregateQuotaWalFrameV3(seeded.storeQuotaCapability, abortUnion.reservation, transitionAuthority, { type: 118, payload: abort }, async () => ({ sequence: 65536n, walHash: abortWalHash, payloadHash: blake2b256(abort) }))
await adjustForwardHttpsAggregateQuotaV3(seeded.storeQuotaCapability, {
durableTombstonePayloadBuffer: abort,
durableWalHeadSequence: 65536n,
durableWalHeadHash: abortWalHash
})
// The wired turn at registry 65535: the mirror admits (65533+2) but the
// store registry cannot hold it — clean BUDGET_EXHAUSTED rejection with the
// reservation released and the authority left OPEN (REREVIEW3-P1-001 (ii)).
await t.exception.all(acceptForwardedHttpsTargetTurnV3(seeded, { stableSessionId: id, walType: 112 }), /BUDGET_EXHAUSTED|cap exceeded/)
t.absent(forwardHttpsAggregateQuotaV3Status(authority).pendingReservation)
// A follow-on lawful turn on a fresh session settles; close succeeds.
await acceptForwardedHttpsTargetTurnV3(seeded, { stableSessionId: fixed(0x73), body: b4a.alloc(4, 0x52) })
await closeForwardHttpsTargetStoreV3(seeded)
await closeForwardHttpsAggregateQuotaV3(authority)
t.is(forwardHttpsAggregateQuotaV3Status(authority).state, 'CLOSED')
})
test('mirror-ahead wired turn terminalizes through the ENTRY_CAP arm instead of killing the authority FAILED_PREWRITE (probe w3b)', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x74)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
// Seed 65534 wired entries (group-committed).
const seedPayload = b4a.concat([b4a.from('FTS3', 'ascii'), id, b4a.alloc(8, 0x51)])
for (let chunk = 0; chunk * 500 < 65534; chunk++) {
const appends = []
for (let index = 0; index < 500 && chunk * 500 + index < 65534; index++) {
appends.push(store.store.appendAndApply({ type: 112, transactionId: b4a.alloc(32, 0x66), virtualBucket: 0, payload: seedPayload }, () => {}))
}
await Promise.all(appends)
}
await closeForwardHttpsTargetStoreV3(store)
const seeded = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(seeded).catch(() => {}) })
t.is(seeded.slots.get(b4a.toString(id, 'hex')).registry.count, 65534)
// Two raw-quota PROCESSOR_PREPARED operations push the canonical mirror to
// 65536 while the store registry stays 65534 (mirror-ahead divergence).
const capability = seeded.storeQuotaCapability
for (let index = 0; index < 2; index++) {
const payload = b4a.concat([b4a.from('FTS3', 'ascii'), id, b4a.alloc(8, 0x60 + index)])
const plan = createForwardHttpsStoreQuotaCostPlanV3(capability, {
operation: 'PROCESSOR_PREPARED',
knownInputBuffers: [payload],
temporaryWriteBuffers: [],
existingDestinationBytes: 0
})
const union = await reserveForwardHttpsAggregateQuotaV3(capability, plan)
const { transitionAuthority } = bindForwardHttpsStoreQuotaActualBuffersV3(capability, union.reservation, {
logicalRecordBuffers: [],
encryptedPlaintextBuffers: [],
finalWalMetadataBuffers: [payload],
temporaryWriteBuffers: []
})
const walHash = blake2b256(b4a.concat([b4a.from('wal'), payload]))
const sequence = BigInt(65535 + index)
await applyForwardHttpsAggregateQuotaWalFrameV3(capability, union.reservation, transitionAuthority, { type: 114, payload }, async () => ({ sequence, walHash, payloadHash: blake2b256(payload) }))
await commitForwardHttpsAggregateQuotaV3(capability, union.reservation, {
durableWalHeadSequence: sequence,
durableWalHeadHash: walHash
})
}
// A lawful in-cap wired turn: the mirror proves cap+1, so the store
// terminalizes the session through the ENTRY_CAP arm — never a
// FAILED_PREWRITE authority kill (REREVIEW3-P1-002).
await t.exception.all(acceptForwardedHttpsTargetTurnV3(seeded, { stableSessionId: id, walType: 112 }), /BUDGET_EXHAUSTED|cap exceeded/)
t.is(seeded.slots.get(b4a.toString(id, 'hex')).state, SLOT.CONSUMED_UNPRUNED)
t.is(forwardHttpsAggregateQuotaV3Status(authority).state, 'OPEN')
t.absent(forwardHttpsAggregateQuotaV3Status(authority).pendingReservation)
// Follow-on lawful work settles and close succeeds.
await acceptForwardedHttpsTargetTurnV3(seeded, { stableSessionId: fixed(0x75), body: b4a.alloc(4, 0x52) })
await closeForwardHttpsTargetStoreV3(seeded)
await closeForwardHttpsAggregateQuotaV3(authority)
t.is(forwardHttpsAggregateQuotaV3Status(authority).state, 'CLOSED')
})
test('cap+1 storm with lawful below-row hint: 4 concurrent turns at 65535 — convert once, sticky TERMINAL, no double FTM9, reopen recovers (REREVIEW4 d9/d10)', async t => {
const r = await roots(t)
const { authority, capabilities } = await quota(t, r)
const id = fixed(0x7b)
const store = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(store).catch(() => {}) })
// Seed 65535 charge entries (group-committed).
const seedPayload = b4a.concat([b4a.from('FTS3', 'ascii'), id, b4a.alloc(8, 0x51)])
for (let chunk = 0; chunk * 500 < 65535; chunk++) {
const appends = []
for (let index = 0; index < 500 && chunk * 500 + index < 65535; index++) {
appends.push(store.store.appendAndApply({ type: 112, transactionId: b4a.alloc(32, 0x66), virtualBucket: 0, payload: seedPayload }, () => {}))
}
await Promise.all(appends)
}
await closeForwardHttpsTargetStoreV3(store)
const seeded = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(seeded).catch(() => {}) })
t.is(seeded.slots.get(b4a.toString(id, 'hex')).registry.count, 65535)
// Four concurrent turns with the lawful below-row hint planned=1: op0
// converts through the ENTRY_CAP arm (the quota-side row count proves
// cap+1); op1, op2 and op3 answer sticky TERMINAL on the consumed mirror —
// no ordinary commit onto a terminal session, no second FTM9.
const outcomes = await Promise.all([0, 1, 2, 3].map(() =>
acceptForwardedHttpsTargetTurnV3(seeded, { stableSessionId: id, walType: 112, plannedRemovableChargeEntryCount: 1 }).then(
() => 'committed',
error => (error && error.code) || String(error)
)
))
t.alike(outcomes, [
'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_BUDGET_EXHAUSTED',
'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_TERMINAL',
'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_TERMINAL',
'FORWARD_HTTPS_STORAGE_AUTHORITY_V3_TERMINAL'
])
t.is(seeded.slots.get(b4a.toString(id, 'hex')).state, SLOT.CONSUMED_UNPRUNED)
t.is(forwardHttpsAggregateQuotaV3Status(authority).state, 'OPEN')
t.absent(forwardHttpsAggregateQuotaV3Status(authority).pendingReservation)
// Exactly one FTM9 in the WAL: 65535 ordinary entries + the conversion.
t.is(forwardHttpsTargetStoreV3Status(seeded).walHeadSequence, 65536n)
await closeForwardHttpsTargetStoreV3(seeded)
// Fresh-authority reopen: the root recovers the exact consumed state and
// the terminal identity is sticky.
const reopened = await openStore(authority, r, capabilities)
t.teardown(async () => { await closeForwardHttpsTargetStoreV3(reopened).catch(() => {}) })
t.is(reopened.slots.get(b4a.toString(id, 'hex')).state, SLOT.CONSUMED_UNPRUNED)
await t.exception.all(acceptForwardedHttpsTargetTurnV3(reopened, { stableSessionId: id, walType: 112 }), /TERMINAL/)
})