From 5db074dba385576368d71192cbfc4203d00a238e Mon Sep 17 00:00:00 2001 From: rblake2320 <73768949+rblake2320@users.noreply.github.com> Date: Mon, 20 Jul 2026 23:48:28 -0500 Subject: [PATCH 1/2] feat: support governed return-site activation --- CHANGELOG.md | 6 + packages/server/src/ha-control-fencing.ts | 150 ++++++++++++++++---- tsk-b-activation-drill.mts | 165 +++++++++++++++++++++- 3 files changed, 285 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ae05e7f..ccfd94a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,12 @@ ### Security +- Made source activation a signed, append-only head/history chain so repeated + promotions on one stream cannot overwrite or replay an earlier activation. +- Added governed return-site activation: a returning target supplies its exact + signed, terminal lease high-water and receives the next guard-signed lease + transition instead of an un-installable second genesis grant. Control schema + version 2 is intentionally required for this history-bearing layout. - Closed numeric HOTP rollover paths across core derivation, lookahead, atomic stores, client persistence, and replica input. Wire v1 now commits MAX only as an exhausted sentinel and never writes or derives MAX+1. diff --git a/packages/server/src/ha-control-fencing.ts b/packages/server/src/ha-control-fencing.ts index 27645f6..e34ee6e 100644 --- a/packages/server/src/ha-control-fencing.ts +++ b/packages/server/src/ha-control-fencing.ts @@ -3,7 +3,7 @@ import { createHash, createHmac, timingSafeEqual } from 'node:crypto'; import { ContractValidationError } from './ha-outbox-contract.js'; import type { PgExecutor, PgTransactor } from './tsk-hotp-outbox-pg.js'; import type { FencingStore, FenceRecord } from './promotion.js'; -import { verifySourceFrozenReceipt, verifyBFinalizedReceipt, signLeaseGrant } from './tsk-source-fence.js'; +import { verifySourceFrozenReceipt, verifyBFinalizedReceipt, signLeaseGrant, verifyLeaseGrant } from './tsk-source-fence.js'; import type { SourceFrozenReceipt, BFinalizedReceipt, SourceVerifyKeyResolver, LeaseGrant } from './tsk-source-fence.js'; import type { KeyObject } from 'node:crypto'; @@ -32,7 +32,7 @@ import type { KeyObject } from 'node:crypto'; // ── bounds + grammars (H9: exhaust safe integers; enforce id/digest grammar) ── -export const CONTROL_SCHEMA_VERSION = 1; +export const CONTROL_SCHEMA_VERSION = 2; const MAX_EPOCH = 2 ** 40; // >> any real promotion count, safely < 2^53 const MAX_SEQ = 2 ** 40; const MAX_MS = 8.64e15; // JS Date range bound @@ -150,6 +150,12 @@ const leaseMsg = (s: string, leaseId: string, holder: string, epoch: number, seq frame('tsk_ha_lease/v1', s, leaseId, holder, epoch, seq, status, maxExpiryMs, grantCmd, prev, digest); const cutMsg = (s: string, epoch: number, commandId: string, seqno: number, phase: string, evidence: string | null, prev: string | null, digest: string): Buffer => frame('tsk_ha_cutover/v1', s, epoch, commandId, seqno, phase, evidence, prev, digest); +const activationMsg = (s: string, commandId: string, epoch: number, targetKeyId: string, + receiptDigest: string, activationSeq: number, priorTargetGrantDigest: string | null, + grantDigest: string, prevActivationDigest: string | null, activationDigest: string): Buffer => + frame('tsk_ha_source_activation/v2', s, commandId, epoch, targetKeyId, + receiptDigest, activationSeq, priorTargetGrantDigest, grantDigest, + prevActivationDigest, activationDigest); const claimDigest = (r: FenceRecord): string => sha256hex(frame('tsk_ha_claim/v1', r.nodeId, r.fenceEpoch, r.expiresAt, r.commandId)); /** (§3.4 + Erratum-R4) Cross-check the Redis claim vs the SIGNED witness floor before a fence. @@ -312,10 +318,34 @@ CREATE TABLE IF NOT EXISTS tsk_ha_source_activation ( epoch bigint NOT NULL CHECK (epoch >= 1), b_key_id text NOT NULL, b_receipt_digest text NOT NULL CHECK (b_receipt_digest ~ '^[0-9a-f]{64}$'), + activation_seq bigint NOT NULL CHECK (activation_seq >= 1), + prior_target_grant_digest text CHECK (prior_target_grant_digest IS NULL OR prior_target_grant_digest ~ '^[0-9a-f]{64}$'), + prev_activation_digest text CHECK (prev_activation_digest IS NULL OR prev_activation_digest ~ '^[0-9a-f]{64}$'), + activation_digest text NOT NULL CHECK (activation_digest ~ '^[0-9a-f]{64}$'), grant_digest text NOT NULL CHECK (grant_digest ~ '^[0-9a-f]{64}$'), grant_json text NOT NULL, guard_key_id text NOT NULL, - created_at timestamptz NOT NULL DEFAULT now() + guard_signature text NOT NULL, + updated_at timestamptz NOT NULL DEFAULT now() +); +CREATE TABLE IF NOT EXISTS tsk_ha_source_activation_history ( + stream_id text NOT NULL, + command_id text NOT NULL, + epoch bigint NOT NULL CHECK (epoch >= 1), + b_key_id text NOT NULL, + b_receipt_digest text NOT NULL CHECK (b_receipt_digest ~ '^[0-9a-f]{64}$'), + activation_seq bigint NOT NULL CHECK (activation_seq >= 1), + prior_target_grant_digest text CHECK (prior_target_grant_digest IS NULL OR prior_target_grant_digest ~ '^[0-9a-f]{64}$'), + prev_activation_digest text CHECK (prev_activation_digest IS NULL OR prev_activation_digest ~ '^[0-9a-f]{64}$'), + activation_digest text NOT NULL CHECK (activation_digest ~ '^[0-9a-f]{64}$'), + grant_digest text NOT NULL CHECK (grant_digest ~ '^[0-9a-f]{64}$'), + grant_json text NOT NULL, + guard_key_id text NOT NULL, + guard_signature text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (stream_id, activation_seq), + UNIQUE (stream_id, command_id), + UNIQUE (stream_id, activation_digest) ) `.trim(); @@ -323,7 +353,7 @@ export const HA_CONTROL_TABLES = [ 'tsk_ha_schema', 'tsk_ha_provisioning', 'tsk_ha_provisioning_history', 'tsk_ha_epoch_witness', 'tsk_ha_epoch_witness_history', 'tsk_ha_lease_head', 'tsk_ha_lease_history', 'tsk_ha_cutover_head', 'tsk_ha_cutover_history', - 'tsk_ha_source_activation', + 'tsk_ha_source_activation', 'tsk_ha_source_activation_history', ] as const; const MANIFEST_TABLES = [...HA_CONTROL_TABLES]; // attest the FULL set incl tsk_ha_schema (R4-H3) // (R8-HIGH) governed tables locked ACCESS SHARE per critical tx to freeze the catalog against a @@ -420,7 +450,7 @@ async function controlManifest(exec: PgExecutor): Promise { * defeat the pin). To RE-PIN after an intentional DDL change, run provisionControlSchema against * the new schema in an OFFLINE, code-reviewed step and copy the digest reported in the attestation * error ("live catalog digest ") into this constant. */ -export const HA_CONTROL_MANIFEST_DIGEST = 'a80069c514d7450a5a2f4482d99952f9b2920ee876d14f51ba5d516ff79eefab'; +export const HA_CONTROL_MANIFEST_DIGEST = '769640c27b3310d391f61d9afff36935be84c02ef885ef4a1537a58b6b81612a'; /** Attest the live catalog hashes to the pinned expected manifest (fail-closed, NOT TOFU). */ async function attestControlSchema(exec: PgExecutor): Promise { @@ -527,10 +557,9 @@ export interface HaControlPolicy { * (CREATE/ALTER/DROP) on this schema — then drift is impossible under the operating identity and * the per-op re-attestation is redundant. */ attestEveryTx?: boolean; - /** (PR2c) The CONFIGURED source-guard signing authority used by activateSource() to mint B's promoted + /** The CONFIGURED source-guard signing authority used by activateSource() to mint the promoted * source lease. Held by the control as deployment configuration — NOT accepted per call — so no caller - * can supply an arbitrary (evil) key. `activationTtlMs` is the (deterministic, config-fixed) lease - * horizon of the genesis promotion grant. Absent → activateSource() is disabled (throws). */ + * can supply an arbitrary key. `activationTtlMs` is the config-fixed lease horizon. */ sourceGuard?: { keyId: string; privateKey: import('node:crypto').KeyObject | string; activationTtlMs: number }; } export interface FenceEvidence { @@ -1086,23 +1115,32 @@ export class HaControlFencing { } /** - * (§5 / PR2c) GOVERNED B source-authority activation. After the cutover is ACTIVE, mint B's unforgeable - * SOURCE capability: a GUARD-signed (ed25519) lease grant that lets B become the writable source AT THE + * GOVERNED source-authority activation. After the cutover is ACTIVE, mint the target's unforgeable + * SOURCE capability: a GUARD-signed (ed25519) lease grant that lets it become writable AT THE * PROMOTED EPOCH. This binds the grant to the ratified promotion — it requires the ACTIVE head for this - * exact command/epoch, re-verifies B's BFinalizedReceipt, and asserts the ACTIVE evidence pins the SAME B - * (bReceiptDigest + n + bSystemId). The grant is at leaseEpoch == targetEpoch (the exact new epoch), holder - * == B, commandId == the cutover command — so B installs a GENESIS lease at epoch N's successor while the old - * A, still at the prior epoch with a revoked lease, is denied. B verifies the grant with the guard PUBLIC - * key (unforgeable); control never sees B's private material. Returns the signed grant for B to install. + * exact command/epoch, re-verifies the target's BFinalizedReceipt, and asserts the ACTIVE evidence pins + * that same target. A never-used target receives a genesis lease. A returning target must supply its exact + * signed, terminal lease high-water; the returned grant appends to that chain. Activation itself advances + * a signed append-only control history, so an older activation head cannot be replayed. Control never sees + * the target's private material. */ - async activateSource(streamId: string, commandId: string, targetEpoch: number, bReceipt: BFinalizedReceipt, bResolver: SourceVerifyKeyResolver): Promise { + async activateSource(streamId: string, commandId: string, targetEpoch: number, + bReceipt: BFinalizedReceipt, bResolver: SourceVerifyKeyResolver, + priorTargetLease: LeaseGrant | null = null): Promise { const s = vId(streamId, STREAM_ID_RE, 'streamId'); const cmd = vId(commandId, ID_RE, 'commandId'); const target = vInt(targetEpoch, 'targetEpoch', 1, MAX_EPOCH); const guard = this.sourceGuard; if (!guard) throw new ContractValidationError('activateSource requires a configured policy.sourceGuard signing authority'); const br = frozenSnapshot(bReceipt); + const prior = priorTargetLease === null ? null : frozenSnapshot(priorTargetLease); verifyBFinalizedReceipt(bResolver, br); + if (prior !== null) { + verifyLeaseGrant(bResolver, prior); + if (prior.streamId !== s) throw new FenceAuthorityQuarantineError('prior target lease streamId != cutover stream — quarantine'); + if (prior.leaseStatus !== 'revoked') throw new FenceAuthorityQuarantineError('prior target lease must be terminally revoked — quarantine'); + if (prior.leaseEpoch >= target) throw new FenceAuthorityQuarantineError('prior target lease epoch must precede the promoted epoch — quarantine'); + } if (br.commandId !== cmd) throw new FenceAuthorityQuarantineError('BFinalizedReceipt commandId != cutover command — quarantine'); if (br.streamId !== s) throw new FenceAuthorityQuarantineError('BFinalizedReceipt streamId != cutover stream — quarantine'); if (br.epoch !== target - 1) throw new FenceAuthorityQuarantineError(`BFinalizedReceipt epoch ${br.epoch} != targetEpoch-1 (${target - 1}) — quarantine`); @@ -1116,26 +1154,78 @@ export class HaControlFencing { // the receipt must be EXACTLY the one ratified into the ACTIVE head, and the holder == that ratified B. if (rd.bReceiptDigest !== br.receiptDigest || rd.n !== br.n || rd.bSystemId !== br.bSystemId || rd.bKeyId !== br.bKeyId) throw new FenceAuthorityQuarantineError('BFinalizedReceipt is not the one ratified into the ACTIVE head — quarantine'); if (rd.bKeyId !== holder) throw new FenceAuthorityQuarantineError('holder != the ratified B identity — quarantine'); - // (recovery / idempotency) if this stream was already activated, REHYDRATE the byte-identical signed - // grant — a retry can NEVER mint a different or conflicting grant. - const existing = (await exec.query('SELECT command_id, epoch, b_key_id, b_receipt_digest, grant_json FROM tsk_ha_source_activation WHERE stream_id=$1 FOR UPDATE', [s])).rows[0]; - if (existing) { - if (String(existing.command_id) !== cmd || vInt(existing.epoch, 'epoch', 1, MAX_EPOCH) !== target || String(existing.b_key_id) !== holder || String(existing.b_receipt_digest) !== br.receiptDigest) { - throw new FenceAuthorityQuarantineError('a DIFFERENT source activation already exists for this stream — quarantine (no conflicting re-grant)'); + const cols = 'command_id,epoch,b_key_id,b_receipt_digest,activation_seq,prior_target_grant_digest,prev_activation_digest,activation_digest,grant_digest,grant_json,guard_key_id,guard_signature'; + const headRow = (await exec.query(`SELECT ${cols} FROM tsk_ha_source_activation WHERE stream_id=$1 FOR UPDATE`, [s])).rows[0]; + const historyRows = (await exec.query(`SELECT ${cols} FROM tsk_ha_source_activation_history WHERE stream_id=$1 ORDER BY activation_seq`, [s])).rows; + const linked = (row: Record): Linked => { + const seq = vInt(row.activation_seq, 'activation_seq', 1, MAX_SEQ); + const prev = vNullableDigest(row.prev_activation_digest, 'prev_activation_digest'); + const digest = vDigest(row.activation_digest, 'activation_digest'); + const priorDigest = vNullableDigest(row.prior_target_grant_digest, 'prior_target_grant_digest'); + const grantDigest = vDigest(row.grant_digest, 'grant_digest'); + const epoch = vInt(row.epoch, 'activation epoch', 1, MAX_EPOCH); + const command = vId(row.command_id, ID_RE, 'activation command_id'); + const targetKey = vId(row.b_key_id, KEY_ID_RE, 'activation target key'); + const receiptDigest = vDigest(row.b_receipt_digest, 'activation receipt digest'); + const digMsg = activationMsg(s, command, epoch, targetKey, receiptDigest, + seq, priorDigest, grantDigest, prev, ''); + return { seq, prev, digest, keyId: String(row.guard_key_id), + sig: String(row.guard_signature), digMsg, + sigMsg: activationMsg(s, command, epoch, targetKey, receiptDigest, + seq, priorDigest, grantDigest, prev, digest) }; + }; + if (headRow) verifyChain(this.resolver, historyRows.map(linked), linked(headRow)); + else if (historyRows.length !== 0) throw new FenceAuthorityQuarantineError('activation history exists without a head — quarantine'); + + const priorDigest = prior?.grantDigest ?? null; + if (headRow && String(headRow.command_id) === cmd) { + if (vInt(headRow.epoch, 'epoch', 1, MAX_EPOCH) !== target || + String(headRow.b_key_id) !== holder || + String(headRow.b_receipt_digest) !== br.receiptDigest || + vNullableDigest(headRow.prior_target_grant_digest, 'prior_target_grant_digest') !== priorDigest) { + throw new FenceAuthorityQuarantineError('source activation retry does not byte-bind the current activation — quarantine'); } - return JSON.parse(String(existing.grant_json)) as LeaseGrant; // rehydrated, byte-identical + const existingGrant = JSON.parse(String(headRow.grant_json)) as LeaseGrant; + verifyLeaseGrant(bResolver, existingGrant); + if (existingGrant.grantDigest !== String(headRow.grant_digest)) throw new FenceAuthorityQuarantineError('stored source activation grant digest mismatch — quarantine'); + return existingGrant; } - // FIRST activation: GUARD-sign B's genesis source lease at the promoted epoch using the CONFIGURED - // authority (deterministic tuple; holder == ratified B; expiry = configured horizon read in-tx). + const currentActivationSeq = headRow ? vInt(headRow.activation_seq, 'activation_seq', 1, MAX_SEQ) : 0; + const currentActivationEpoch = headRow ? vInt(headRow.epoch, 'activation epoch', 1, MAX_EPOCH) : 0; + if (target !== currentActivationEpoch + 1) throw new FenceAuthorityQuarantineError(`source activation epoch ${target} must follow current ${currentActivationEpoch} exactly — quarantine`); + + // A never-used target receives a genesis grant. A returning target supplies its exact signed, + // terminal lease high-water, so the new grant appends to that target's immutable chain. const nowMs = await this.controlNowMs(exec); const grantObj = signLeaseGrant(guard.keyId, guard.privateKey, { streamId: s, leaseEpoch: target, leaseStatus: 'active', holderNodeId: holder, - leaseId, commandId: cmd, leaseExpiresAtMs: Math.min(nowMs + guard.activationTtlMs, MAX_MS), leaseGrantSeq: 1, prevGrantDigest: null, + leaseId, commandId: cmd, leaseExpiresAtMs: Math.min(nowMs + guard.activationTtlMs, MAX_MS), + leaseGrantSeq: (prior?.leaseGrantSeq ?? 0) + 1, + prevGrantDigest: priorDigest, }); - // bind the exact genesis grant tuple/digest into durable signed activation evidence. - affectedOne(await exec.query( - 'INSERT INTO tsk_ha_source_activation (stream_id, command_id, epoch, b_key_id, b_receipt_digest, grant_digest, grant_json, guard_key_id) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)', - [s, cmd, target, holder, br.receiptDigest, grantObj.grantDigest, JSON.stringify(grantObj), guard.keyId]), 'source activation insert'); + const activationSeq = currentActivationSeq + 1; + const prevActivationDigest = headRow ? vDigest(headRow.activation_digest, 'activation_digest') : null; + const digMsg = activationMsg(s, cmd, target, holder, br.receiptDigest, + activationSeq, priorDigest, grantObj.grantDigest, prevActivationDigest, ''); + const activationDigest = digestOf(digMsg); + const activationSignature = this.signer.sign(activationMsg(s, cmd, target, holder, + br.receiptDigest, activationSeq, priorDigest, grantObj.grantDigest, + prevActivationDigest, activationDigest)); + const values = [s, cmd, target, holder, br.receiptDigest, activationSeq, + priorDigest, prevActivationDigest, activationDigest, grantObj.grantDigest, + JSON.stringify(grantObj), this.signer.id, activationSignature]; + await exec.query( + 'INSERT INTO tsk_ha_source_activation_history (stream_id,command_id,epoch,b_key_id,b_receipt_digest,activation_seq,prior_target_grant_digest,prev_activation_digest,activation_digest,grant_digest,grant_json,guard_key_id,guard_signature) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)', + values); + if (!headRow) { + affectedOne(await exec.query( + 'INSERT INTO tsk_ha_source_activation (stream_id,command_id,epoch,b_key_id,b_receipt_digest,activation_seq,prior_target_grant_digest,prev_activation_digest,activation_digest,grant_digest,grant_json,guard_key_id,guard_signature) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)', values), + 'source activation head insert'); + } else { + affectedOne(await exec.query( + 'UPDATE tsk_ha_source_activation SET command_id=$2,epoch=$3,b_key_id=$4,b_receipt_digest=$5,activation_seq=$6,prior_target_grant_digest=$7,prev_activation_digest=$8,activation_digest=$9,grant_digest=$10,grant_json=$11,guard_key_id=$12,guard_signature=$13,updated_at=now() WHERE stream_id=$1 AND activation_digest=$14', + [...values, prevActivationDigest]), 'source activation head forward-CAS'); + } return grantObj; }); } diff --git a/tsk-b-activation-drill.mts b/tsk-b-activation-drill.mts index 54f735f..ac2dc76 100644 --- a/tsk-b-activation-drill.mts +++ b/tsk-b-activation-drill.mts @@ -13,6 +13,7 @@ import assert from 'node:assert/strict'; import { generateKeyPairSync, sign as edSign } from 'node:crypto'; import pg from 'pg'; import { Redis } from 'ioredis'; +import { MemoryFencingStore } from './packages/server/dist/promotion.js'; import { TSK_OUTBOX_PG_SCHEMA, TSK_SOURCE_LEASE_SCHEMA, TSK_SOURCE_WITNESS_SCHEMA, TSK_RECEIVER_SCHEMA, TSK_RECEIVER_TABLES, provisionSchemaVersion, PgTskDurableOutbox, NodePostgresTransactor, ContractValidationError, @@ -34,10 +35,12 @@ if (!A_URL || !B_URL || !CTRL_URL || !REDIS_URL) throw new Error('need A + B + c const SCHEMA = 'public'; const HOUR = 3_600_000; const GUARD_KEY = 'guard-1'; const guard = generateKeyPairSync('ed25519'); const SOURCE_KEY = 'source-1'; const source = generateKeyPairSync('ed25519'); +const B_SOURCE_KEY = 'source-b-1'; const bSource = generateKeyPairSync('ed25519'); const HEAD_KEY = 'k1'; const headKp = generateKeyPairSync('ed25519'); const BHEAD_KEY = 'kb'; const bHeadKp = generateKeyPairSync('ed25519'); // B signs its OWN new heads const B_KEY = 'b-1'; const bKp = generateKeyPairSync('ed25519'); -const resolver: SourceVerifyKeyResolver = { resolve: (k) => (k === GUARD_KEY ? guard.publicKey : k === SOURCE_KEY ? source.publicKey : k === HEAD_KEY ? headKp.publicKey : k === BHEAD_KEY ? bHeadKp.publicKey : k === B_KEY ? bKp.publicKey : null) }; +const A_KEY = 'a-return-1'; const aKp = generateKeyPairSync('ed25519'); +const resolver: SourceVerifyKeyResolver = { resolve: (k) => (k === GUARD_KEY ? guard.publicKey : k === SOURCE_KEY ? source.publicKey : k === B_SOURCE_KEY ? bSource.publicKey : k === HEAD_KEY ? headKp.publicKey : k === BHEAD_KEY ? bHeadKp.publicKey : k === B_KEY ? bKp.publicKey : k === A_KEY ? aKp.publicKey : null) }; const CTRL_KEY = 'ctrl-1'; const ctrlSecret = Buffer.alloc(32, 0x2b); const ctrlResolver: GuardKeyResolver = { resolve: (kid) => (kid === CTRL_KEY ? ctrlSecret : null) }; const POLICY: HaControlPolicy = { minClaimRemainingMs: 5_000, sourceGuard: { keyId: GUARD_KEY, privateKey: guard.privateKey, activationTtlMs: 3_600_000 } }; @@ -51,6 +54,24 @@ const sanitizer: HotpMutationSanitizer = { let passed = 0; async function check(name: string, fn: () => Promise) { await fn(); passed++; console.log(` ok - ${name}`); } +async function installVerifiedHistory(pool: pg.Pool, streamId: string, + bundle: SourceExportBundle, fromSequence = 1): Promise { + for (const record of bundle.historyChunks.flatMap((chunk) => chunk.records)) { + if (record.sequence < fromSequence) continue; + const mutation = JSON.parse(record.payload) as { tumblerId: string; counter: number }; + await pool.query( + `INSERT INTO tsk_outbox_rows + (stream_id,source_epoch,sequence,fence_token,op_digest,tumbler_id,hotp_counter, + mutation,head_prev,head_digest,head_key_id,head_alg,head_sig,acked_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,now())`, + [streamId, record.sourceEpoch, record.sequence, record.fenceToken, + record.opDigest, mutation.tumblerId, mutation.counter, mutation, + record.prevHeadDigest, record.headDigest, record.keyId, record.alg, + record.signature], + ); + } +} + async function main() { console.log('# TSK PR2c governed B source-authority activation (B originates N+1; old A denied)'); const aPool = new pg.Pool({ connectionString: A_URL, max: 4 }); aPool.on('error', () => {}); @@ -59,8 +80,9 @@ async function main() { const aTx = new NodePostgresTransactor(aPool as never) as unknown as PgTransactor; const bTx = new NodePostgresTransactor(bPool as never) as unknown as PgTransactor; const cTx = new NodePostgresTransactor(cPool as never) as unknown as PgTransactor; - const redis = new Redis(REDIS_URL, { maxRetriesPerRequest: 2, lazyConnect: false }); redis.on('error', () => {}); - await redis.flushdb(); + const redis = REDIS_URL === 'memory://' ? null : new Redis(REDIS_URL, { maxRetriesPerRequest: 2, lazyConnect: false }); + redis?.on('error', () => {}); + await redis?.flushdb(); const OUTBOX_DROP = 'DROP TABLE IF EXISTS tsk_outbox_rows, tsk_outbox_applied, tsk_outbox_fence, tsk_outbox_source_checkpoint, tsk_outbox_receiver_checkpoint, tsk_outbox_publisher_lease, tsk_outbox_quarantine, tsk_hotp_consumed, tsk_outbox_stream_halted, tsk_outbox_meta, tsk_source_lease, tsk_source_lease_history, tsk_source_witness, tsk_source_witness_history CASCADE'; const installSource = async (pool: pg.Pool) => { @@ -69,6 +91,7 @@ async function main() { for (const s of TSK_SOURCE_LEASE_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await pool.query(s); for (const s of TSK_SOURCE_WITNESS_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await pool.query(s); }; + await aPool.query(`DROP TABLE IF EXISTS ${TSK_RECEIVER_TABLES.join(', ')} CASCADE`); await installSource(aPool); // B is BOTH a receiver (staged generation) AND, after activation, a source (outbox) — install both catalogs. await bPool.query(`DROP TABLE IF EXISTS ${TSK_RECEIVER_TABLES.join(', ')} CASCADE`); @@ -105,6 +128,9 @@ async function main() { const bReceipt = await stageAndFinalizeReceiverGeneration(bTx, SCHEMA, await assertReceiverReady(bTx, SCHEMA), 'gen-1', bundle, dual, ropts); verifyBFinalizedReceipt(resolver, bReceipt); const headAtN = bReceipt.signedHeadDigestAtN; + // The independently replayed receiver generation is now materialized as B's source ledger. + // These exact signed rows are what the next freeze/export replays; no history is synthesized. + await installVerifiedHistory(bPool, SID, bundle); // ── control drives the cutover to ACTIVE ── const ctlReady = await provisionControlSchema(cTx as never, SCHEMA); @@ -115,7 +141,7 @@ async function main() { await ctl.beginPromotionIntent(SID, CMD, TARGET); await ctl.bindSourceFenced(SID, CMD, TARGET, frozen, resolver); await ctl.writeLease({ streamId: SID, leaseId: 'l1', holderNodeId: 'A', epoch: 0, status: 'revoked', grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'a2' }); - const store = new RedisFencingStore(redis, 'tsk:fence:' + SID); + const store = redis ? new RedisFencingStore(redis, 'tsk:fence:' + SID) : new MemoryFencingStore(); const proof: FenceProof = { safetyMarginMs: 0, claimExpiresAtMs: (await ctlNow()) + HOUR }; await ctl.advanceEpoch(SID, CMD, TARGET, 'Bnode', store, proof); await ctl.markImporting(SID, CMD, TARGET); @@ -123,6 +149,7 @@ async function main() { await ctl.activate(SID, CMD, TARGET); let bGrant: Awaited>; + let bOb: InstanceType; await check('activateSource mints a GUARD-signed governed B lease at the PROMOTED epoch via the CONFIGURED authority, holder==ratified B, durable + rehydratable', async () => { bGrant = await ctl.activateSource(SID, CMD, TARGET, bReceipt, resolver); // no per-call key; guard from config verifyLeaseGrant(resolver, bGrant); // B independently verifies with the guard PUBLIC key (unforgeable) @@ -142,7 +169,7 @@ async function main() { await bTx.transaction((exec) => installLeaseGrant(exec, resolver, bGrant)); const bReadySrc = await assertSourceFenceReady(bTx, SCHEMA, resolver, { streamId: SID, holderNodeId: bGrant.holderNodeId, leaseId: bGrant.leaseId, grantDigest: bGrant.grantDigest }); const bREADY = await provisionSchemaVersion(bTx, SCHEMA); - const bOb = new PgTskDurableOutbox(bTx, bREADY, { streamId: SID, sanitizer, signer: bSigner, maxPendingRows: 100_000, backpressure: 'fail-authoritative-mutation' }, { resolver, controlToASkewBoundMs: 0, ready: bReadySrc }); + bOb = new PgTskDurableOutbox(bTx, bREADY, { streamId: SID, sanitizer, signer: bSigner, maxPendingRows: 100_000, backpressure: 'fail-authoritative-mutation' }, { resolver, controlToASkewBoundMs: 0, ready: bReadySrc }); const res = await bOb.withOutboxTx((wtx) => bOb.appendInTx(wtx, { streamId: SID, rawMutation: { tumblerId: 'T9', counter: 1 }, fenceToken: BigInt(TARGET) })); assert.equal(res.head.sequence, N + 1, 'B appended the NEXT sequence (N+1)'); assert.equal(res.head.prevHeadDigest, headAtN, 'B chained N+1 from the verified head@N'); @@ -157,7 +184,133 @@ async function main() { assert.equal(aMax, N, 'A wrote nothing past N — the old source is fenced'); }); + // ── governed B -> A return failback on the SAME stream ── + const RETURN_CMD = 'return-2'; const RETURN_TARGET = 2; + let bRev: ReturnType; + let returnedGrant: Awaited>; + let returnReceipt: Awaited>; + await check('B freezes N+1 and A independently replays the complete signed ledger for return failback', async () => { + bRev = signLeaseGrant(GUARD_KEY, guard.privateKey, { + streamId: SID, leaseEpoch: TARGET, leaseStatus: 'revoked', + holderNodeId: bGrant.holderNodeId, leaseId: bGrant.leaseId, + commandId: RETURN_CMD, leaseExpiresAtMs: bGrant.leaseExpiresAtMs, + leaseGrantSeq: 2, prevGrantDigest: bGrant.grantDigest, + }); + await bTx.transaction((exec) => installLeaseGrant(exec, resolver, bRev)); + const frozenB = await emitSourceFrozenReceipt(bTx, SCHEMA, { + sourceKeyId: B_SOURCE_KEY, sourcePrivateKey: bSource.privateKey, + leaseResolver: resolver, headResolver: resolver, + }, { streamId: SID, commandId: RETURN_CMD, epoch: TARGET, sourceNodeId: B_KEY }); + assert.equal(frozenB.n, N + 1); + const builtB = await buildSourceExportManifest(bTx, SCHEMA, { + streamId: SID, epoch: TARGET, commandId: RETURN_CMD, sourceNodeId: B_KEY, + }, { + sourceKeyId: B_SOURCE_KEY, sourcePrivateKey: bSource.privateKey, + sanitizer, leaseResolver: resolver, headResolver: resolver, + frozenReceipt: frozenB, maxChunkItems: 4, + }); + const dualB = guardCountersignSourceExport(builtB.bundle, builtB.manifest, { + guardKeyId: GUARD_KEY, guardPrivateKey: guard.privateKey, + sanitizer, sourceManifestResolver: resolver, headResolver: resolver, + frozenResolver: resolver, frozenReceipt: frozenB, + expectedCommandId: RETURN_CMD, + }); + for (const s of TSK_RECEIVER_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await aPool.query(s); + returnReceipt = await stageAndFinalizeReceiverGeneration( + aTx, SCHEMA, await assertReceiverReady(aTx, SCHEMA), 'gen-return-2', + builtB.bundle, dualB, { + sanitizer, sourceResolver: resolver, guardResolver: resolver, + headResolver: resolver, frozenResolver: resolver, bVerifyResolver: resolver, + frozenReceipt: frozenB, expectedCommandId: RETURN_CMD, + bKeyId: A_KEY, bPrivateKey: aKp.privateKey, + }, + ); + verifyBFinalizedReceipt(resolver, returnReceipt); + assert.equal(returnReceipt.n, N + 1); + + await ctl.writeLease({ streamId: SID, leaseId: bGrant.leaseId, + holderNodeId: bGrant.holderNodeId, epoch: TARGET, status: 'active', + grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'b-active-2' }); + await ctl.beginPromotionIntent(SID, RETURN_CMD, RETURN_TARGET); + await ctl.bindSourceFenced(SID, RETURN_CMD, RETURN_TARGET, frozenB, resolver); + await ctl.writeLease({ streamId: SID, leaseId: bGrant.leaseId, + holderNodeId: bGrant.holderNodeId, epoch: TARGET, status: 'revoked', + grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'b-revoke-2' }); + await ctl.advanceEpoch(SID, RETURN_CMD, RETURN_TARGET, 'Anode', store, + { safetyMarginMs: 0, claimExpiresAtMs: (await ctlNow()) + HOUR }); + await ctl.markImporting(SID, RETURN_CMD, RETURN_TARGET); + await ctl.markReady(SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver); + await ctl.activate(SID, RETURN_CMD, RETURN_TARGET); + returnedGrant = await ctl.activateSource( + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, rev, + ); + verifyLeaseGrant(resolver, returnedGrant); + assert.equal(returnedGrant.holderNodeId, A_KEY); + assert.equal(returnedGrant.leaseEpoch, RETURN_TARGET); + assert.equal(returnedGrant.leaseGrantSeq, 3); + assert.equal(returnedGrant.prevGrantDigest, rev.grantDigest); + + // A already owns 1..N. Import only B's independently verified signed N+1 row. + await installVerifiedHistory(aPool, SID, builtB.bundle, N + 1); + await aPool.query('UPDATE tsk_outbox_fence SET fence_token=$2 WHERE stream_id=$1', [SID, RETURN_TARGET]); + await aPool.query( + 'UPDATE tsk_outbox_source_checkpoint SET sequence=$2,head_digest=$3 WHERE stream_id=$1', + [SID, N + 1, returnReceipt.signedHeadDigestAtN], + ); + await aTx.transaction((exec) => installLeaseGrant(exec, resolver, returnedGrant)); + }); + + await check('returned A originates N+2 while old B and replayed activation inputs fail closed', async () => { + const returnedReady = await assertSourceFenceReady(aTx, SCHEMA, resolver, { + streamId: SID, holderNodeId: returnedGrant.holderNodeId, + leaseId: returnedGrant.leaseId, grantDigest: returnedGrant.grantDigest, + }); + const returnedOutbox = new PgTskDurableOutbox(aTx, await provisionSchemaVersion(aTx, SCHEMA), { + streamId: SID, sanitizer, signer, + maxPendingRows: 100_000, backpressure: 'fail-authoritative-mutation', + }, { resolver, controlToASkewBoundMs: 0, ready: returnedReady }); + const result = await returnedOutbox.withOutboxTx((wtx) => returnedOutbox.appendInTx(wtx, { + streamId: SID, rawMutation: { tumblerId: 'T10', counter: 2 }, + fenceToken: BigInt(RETURN_TARGET), + })); + assert.equal(result.head.sequence, N + 2); + assert.equal(result.head.prevHeadDigest, returnReceipt.signedHeadDigestAtN); + await assert.rejects(() => bOb.withOutboxTx((wtx) => bOb.appendInTx(wtx, { + streamId: SID, rawMutation: { tumblerId: 'T9', counter: 2 }, fenceToken: 1n, + })), /revoked|not writable|lease|fence/i); + await assert.rejects(() => ctl.activateSource( + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, g, + ), /prior target lease must be terminally revoked|does not byte-bind/i); + await assert.rejects(() => ctl.activateSource( + SID, CMD, TARGET, bReceipt, resolver, + ), /epoch|activation|quarantine/i); + const retry = await ctl.activateSource( + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, rev, + ); + assert.deepEqual(retry, returnedGrant); + const activationCols = 'command_id,epoch,b_key_id,b_receipt_digest,activation_seq,' + + 'prior_target_grant_digest,prev_activation_digest,activation_digest,grant_digest,' + + 'grant_json,guard_key_id,guard_signature'; + await cPool.query( + `UPDATE tsk_ha_source_activation h SET (${activationCols}) = + (SELECT ${activationCols} FROM tsk_ha_source_activation_history + WHERE stream_id=$1 AND activation_seq=1) WHERE h.stream_id=$1`, [SID], + ); + await assert.rejects(() => ctl.activateSource( + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, rev, + ), /head is not the latest|replay|rollback/i); + await cPool.query( + `UPDATE tsk_ha_source_activation h SET (${activationCols}) = + (SELECT ${activationCols} FROM tsk_ha_source_activation_history + WHERE stream_id=$1 AND activation_seq=2) WHERE h.stream_id=$1`, [SID], + ); + const activationRows = Number((await cPool.query( + 'SELECT count(*) AS n FROM tsk_ha_source_activation_history WHERE stream_id=$1', [SID], + )).rows[0].n); + assert.equal(activationRows, 2, 'A->B and B->A activations are append-only history'); + }); + console.log(`\n# ${passed} PR2c B-source-activation checks passed`); - await aPool.end(); await bPool.end(); await cPool.end(); await redis.quit(); + await aPool.end(); await bPool.end(); await cPool.end(); await redis?.quit(); } main().then(() => process.exit(0)).catch((e) => { console.error(e); process.exit(1); }); From 57d02934f0f724fd2eafe874668cf0a5c447d16d Mon Sep 17 00:00:00 2001 From: rblake2320 <73768949+rblake2320@users.noreply.github.com> Date: Tue, 21 Jul 2026 00:15:57 -0500 Subject: [PATCH 2/2] fix: close return-activation review findings --- .github/workflows/ci.yml | 7 + CHANGELOG.md | 5 + package-boundary-suite.mts | 2 + package.json | 1 + packages/server/package.json | 1 + packages/server/src/ha-control-fencing.ts | 433 +++++++++++++++--- packages/server/src/index.ts | 6 + packages/server/src/tsk-hotp-outbox-pg.ts | 263 ++++++++++- packages/server/src/tsk-source-fence.ts | 2 +- .../ha-control-activation-continuity.test.ts | 25 + tsk-b-activation-drill.mts | 194 +++++--- tsk-control-cutover-drill.mts | 4 +- tsk-ha-control-migration-drill.mts | 90 ++++ 13 files changed, 917 insertions(+), 116 deletions(-) create mode 100644 packages/server/tests/ha-control-activation-continuity.test.ts create mode 100644 tsk-ha-control-migration-drill.mts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a2dd360..b836887 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -181,6 +181,13 @@ jobs: TSK_TEST_RECEIVER_PG_URL_B: postgresql://tsk_test:tsk-test-only-password@127.0.0.1:5433/tsk_test TSK_TEST_CONTROL_PG_URL: postgresql://tsk_test:tsk-test-only-password@127.0.0.1:5434/tsk_test TSK_TEST_REDIS_URL: redis://127.0.0.1:6379 + # Governed OFFLINE control-schema v1->v2 migration: exact v1 catalog+stamp + # attestation, signed legacy activation preservation, atomic conversion, and + # fail-closed rollback on corrupt legacy authority. Real control PG16; no skip. + - run: npm run test:ha-control-migration + env: + TSK_TEST_CONTROL_PG_URL: postgresql://tsk_test:tsk-test-only-password@127.0.0.1:5434/tsk_test + - run: npm run test:ha-control-continuity -w packages/server # Enterprise credential authority: every TumblerMapStore mutation commits its # authoritative map change and a secret-free signed TSK record in one # SERIALIZABLE source-A transaction. Independent B verifies the head chain, diff --git a/CHANGELOG.md b/CHANGELOG.md index ccfd94a..f3c6924 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,11 @@ signed, terminal lease high-water and receives the next guard-signed lease transition instead of an un-installable second genesis grant. Control schema version 2 is intentionally required for this history-bearing layout. +- Added an owned, crash-atomic receiver-to-source activation authority that + rebuilds only persisted finalized staging, re-verifies both export signatures + and the complete ledger, then installs rows, fence, checkpoint, and lease in + one serializable target transaction. Manual multi-transaction import is not + an activation path. - Closed numeric HOTP rollover paths across core derivation, lookahead, atomic stores, client persistence, and replica input. Wire v1 now commits MAX only as an exhausted sentinel and never writes or derives MAX+1. diff --git a/package-boundary-suite.mts b/package-boundary-suite.mts index 1288d05..8cb0398 100644 --- a/package-boundary-suite.mts +++ b/package-boundary-suite.mts @@ -87,6 +87,8 @@ for (const workspace of workspaces) { for (const sym of ['NodePostgresTransactor', 'AmbiguousCommitError', 'PostCommitReleaseError', 'ConnectionDisposalError']) { assert(typeof (exports as Record)[sym] === 'function', `@tsk/server must export ${sym}`); } + assert(typeof (exports as Record).activateFinalizedReceiverAsSource === 'function', + '@tsk/server must export the owned atomic receiver-to-source activation authority'); for (const sym of ['PgHaTumblerMapStore', 'PgTskCredentialReceiverCheckpoint', 'assertCredentialAuthorityReady', 'provisionCredentialRuntimeMutationBoundary', 'assertCredentialRuntimeMutationBoundary', 'HmacCredentialMutationTicketSigner']) { diff --git a/package.json b/package.json index c6ad699..0780339 100644 --- a/package.json +++ b/package.json @@ -27,6 +27,7 @@ "test:control-cutover": "npm run build -w packages/server && npx tsx tsk-control-cutover-drill.mts", "test:sigkill": "npm run build -w packages/server && npx tsx tsk-cutover-sigkill-drill.mts", "test:b-activation": "npm run build -w packages/server && npx tsx tsk-b-activation-drill.mts", + "test:ha-control-migration": "npm run build -w packages/server && npx tsx tsk-ha-control-migration-drill.mts", "test:sentinel": "npm run build -w packages/server && bash ci/redis-sentinel/run.sh tsk-ha-sentinel-drill.mts", "test:partition": "npm run build -w packages/server && bash ci/redis-sentinel/run.sh tsk-ha-partition-drill.mts", "test:two-node": "npm run build -w packages/server && npx tsx tsk-two-node-drill.mts", diff --git a/packages/server/package.json b/packages/server/package.json index b52fe0d..83721d5 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -24,6 +24,7 @@ "test:hotp-outbox": "vitest run tests/tsk-hotp-outbox-pg.test.ts", "test:pg-transactor": "vitest run tests/tsk-pg-transactor.test.ts", "test:http-outbox": "vitest run tests/http-outbox-transport.test.ts", + "test:ha-control-continuity": "npx vitest run tests/ha-control-activation-continuity.test.ts", "typecheck:contract": "tsc --noEmit -p tsconfig.contract-typecheck.json" }, "dependencies": { diff --git a/packages/server/src/ha-control-fencing.ts b/packages/server/src/ha-control-fencing.ts index e34ee6e..fdc8fe6 100644 --- a/packages/server/src/ha-control-fencing.ts +++ b/packages/server/src/ha-control-fencing.ts @@ -82,6 +82,42 @@ function frozenSnapshot(x: T): T { freeze(snap); return snap; } +const LEASE_GRANT_KEYS = ['commandId', 'grantDigest', 'guardKeyId', 'guardSignature', + 'holderNodeId', 'leaseEpoch', 'leaseExpiresAtMs', 'leaseGrantSeq', 'leaseId', + 'leaseStatus', 'prevGrantDigest', 'streamId'] as const; +const MAX_PRIOR_TARGET_LEASE_CHAIN = 10_000; +function snapshotLeaseChain(value: unknown): LeaseGrant[] { + if (!Array.isArray(value) || Object.getOwnPropertySymbols(value).length !== 0 || + value.length > MAX_PRIOR_TARGET_LEASE_CHAIN) { + throw new ContractValidationError(`priorTargetLeaseChain must be an array of at most ${MAX_PRIOR_TARGET_LEASE_CHAIN} grants`); + } + const arrayDescriptors = Object.getOwnPropertyDescriptors(value); + const expectedArrayKeys = ['length', ...Array.from({ length: value.length }, (_, i) => String(i))].sort(); + const arrayKeys = Object.keys(arrayDescriptors).sort(); + if (arrayKeys.length !== expectedArrayKeys.length || arrayKeys.some((key, index) => key !== expectedArrayKeys[index]) || + Object.entries(arrayDescriptors).some(([key, descriptor]) => key !== 'length' && !('value' in descriptor))) { + throw new ContractValidationError('priorTargetLeaseChain must be a dense exact array with data elements only'); + } + const result: LeaseGrant[] = []; + for (let i = 0; i < value.length; i++) { + const grant = arrayDescriptors[String(i)].value as unknown; + if (!grant || typeof grant !== 'object' || Array.isArray(grant) || + Object.getPrototypeOf(grant) !== Object.prototype || + Object.getOwnPropertySymbols(grant).length !== 0) { + throw new ContractValidationError('priorTargetLeaseChain grants must be exact plain data objects'); + } + const descriptors = Object.getOwnPropertyDescriptors(grant); + const keys = Object.keys(descriptors).sort(); + const expected = [...LEASE_GRANT_KEYS].sort(); + if (keys.length !== expected.length || keys.some((key, index) => key !== expected[index]) || + Object.values(descriptors).some((descriptor) => !('value' in descriptor))) { + throw new ContractValidationError('priorTargetLeaseChain grant shape is invalid'); + } + const row = Object.fromEntries(LEASE_GRANT_KEYS.map((key) => [key, descriptors[key].value])) as unknown as LeaseGrant; + result.push(Object.freeze(row)); + } + return Object.freeze(result.slice()) as LeaseGrant[]; +} // ── guard signing (HMAC over length-prefixed, keyId-bound, canonical framing) ─ @@ -194,8 +230,84 @@ export function reconcileFencedRedis(r: FenceRecord | null, evidence: FenceEvide const CUTOVER_TERMINAL = new Set(['ACTIVE', 'ABORTED']); const CUTOVER_FROZEN = new Set(['PREPARING', 'SOURCE_FENCED', 'FENCED', 'IMPORTING', 'READY']); // lease grants frozen while a promotion is in-flight +/** @internal Pure continuity rule shared with hermetic rejection vectors. */ +export function assertConsumedActivationEpochs( + currentActivationEpoch: number, + targetEpoch: number, + cutoverRows: ReadonlyArray<{ epoch: number; phase: string }>, + activatedEpochs: ReadonlyArray, +): void { + if (targetEpoch <= currentActivationEpoch) throw new FenceAuthorityQuarantineError('source activation epoch does not advance current authority — quarantine'); + const activated = new Set(activatedEpochs); + const phases = new Map(); + for (const row of cutoverRows) { + const list = phases.get(row.epoch) ?? []; + list.push(row.phase); phases.set(row.epoch, list); + } + for (let epoch = currentActivationEpoch + 1; epoch < targetEpoch; epoch++) { + if (activated.has(epoch)) throw new FenceAuthorityQuarantineError(`skipped activation epoch ${epoch} already has source authority evidence — quarantine`); + const terminal = (phases.get(epoch) ?? []).filter((phase) => CUTOVER_TERMINAL.has(phase)); + if (terminal.length !== 1 || terminal[0] !== 'ABORTED') { + throw new FenceAuthorityQuarantineError(`source activation skipped epoch ${epoch} without one verified terminal ABORTED cutover — quarantine`); + } + } +} + // ── control-DB schema (executable; hardened with range/grammar CHECKs, H9) ──── +const SOURCE_ACTIVATION_V2_SCHEMA = ` +CREATE TABLE IF NOT EXISTS tsk_ha_source_activation ( + stream_id text PRIMARY KEY, + command_id text NOT NULL, + epoch bigint NOT NULL CHECK (epoch >= 1), + b_key_id text NOT NULL, + b_receipt_digest text NOT NULL CHECK (b_receipt_digest ~ '^[0-9a-f]{64}$'), + activation_seq bigint NOT NULL CHECK (activation_seq >= 1), + prior_target_grant_digest text CHECK (prior_target_grant_digest IS NULL OR prior_target_grant_digest ~ '^[0-9a-f]{64}$'), + prev_activation_digest text CHECK (prev_activation_digest IS NULL OR prev_activation_digest ~ '^[0-9a-f]{64}$'), + activation_digest text NOT NULL CHECK (activation_digest ~ '^[0-9a-f]{64}$'), + grant_digest text NOT NULL CHECK (grant_digest ~ '^[0-9a-f]{64}$'), + grant_json text NOT NULL, + guard_key_id text NOT NULL, + guard_signature text NOT NULL, + updated_at timestamptz NOT NULL DEFAULT now() +); +CREATE TABLE IF NOT EXISTS tsk_ha_source_activation_history ( + stream_id text NOT NULL, + command_id text NOT NULL, + epoch bigint NOT NULL CHECK (epoch >= 1), + b_key_id text NOT NULL, + b_receipt_digest text NOT NULL CHECK (b_receipt_digest ~ '^[0-9a-f]{64}$'), + activation_seq bigint NOT NULL CHECK (activation_seq >= 1), + prior_target_grant_digest text CHECK (prior_target_grant_digest IS NULL OR prior_target_grant_digest ~ '^[0-9a-f]{64}$'), + prev_activation_digest text CHECK (prev_activation_digest IS NULL OR prev_activation_digest ~ '^[0-9a-f]{64}$'), + activation_digest text NOT NULL CHECK (activation_digest ~ '^[0-9a-f]{64}$'), + grant_digest text NOT NULL CHECK (grant_digest ~ '^[0-9a-f]{64}$'), + grant_json text NOT NULL, + guard_key_id text NOT NULL, + guard_signature text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (stream_id, activation_seq), + UNIQUE (stream_id, command_id), + UNIQUE (stream_id, activation_digest) +) +`.trim(); + +/** Exact legacy v1 activation table used only by the governed offline v1->v2 migration. */ +export const HA_CONTROL_V1_SOURCE_ACTIVATION_SCHEMA = ` +CREATE TABLE IF NOT EXISTS tsk_ha_source_activation ( + stream_id text PRIMARY KEY, + command_id text NOT NULL, + epoch bigint NOT NULL CHECK (epoch >= 1), + b_key_id text NOT NULL, + b_receipt_digest text NOT NULL CHECK (b_receipt_digest ~ '^[0-9a-f]{64}$'), + grant_digest text NOT NULL CHECK (grant_digest ~ '^[0-9a-f]{64}$'), + grant_json text NOT NULL, + guard_key_id text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() +) +`.trim(); + export const HA_CONTROL_PG_SCHEMA = ` CREATE TABLE IF NOT EXISTS tsk_ha_schema ( id int PRIMARY KEY CHECK (id = 1), @@ -312,41 +424,7 @@ CREATE TABLE IF NOT EXISTS tsk_ha_cutover_history ( PRIMARY KEY (stream_id, seqno), UNIQUE (stream_id, state_digest) ); -CREATE TABLE IF NOT EXISTS tsk_ha_source_activation ( - stream_id text PRIMARY KEY, - command_id text NOT NULL, - epoch bigint NOT NULL CHECK (epoch >= 1), - b_key_id text NOT NULL, - b_receipt_digest text NOT NULL CHECK (b_receipt_digest ~ '^[0-9a-f]{64}$'), - activation_seq bigint NOT NULL CHECK (activation_seq >= 1), - prior_target_grant_digest text CHECK (prior_target_grant_digest IS NULL OR prior_target_grant_digest ~ '^[0-9a-f]{64}$'), - prev_activation_digest text CHECK (prev_activation_digest IS NULL OR prev_activation_digest ~ '^[0-9a-f]{64}$'), - activation_digest text NOT NULL CHECK (activation_digest ~ '^[0-9a-f]{64}$'), - grant_digest text NOT NULL CHECK (grant_digest ~ '^[0-9a-f]{64}$'), - grant_json text NOT NULL, - guard_key_id text NOT NULL, - guard_signature text NOT NULL, - updated_at timestamptz NOT NULL DEFAULT now() -); -CREATE TABLE IF NOT EXISTS tsk_ha_source_activation_history ( - stream_id text NOT NULL, - command_id text NOT NULL, - epoch bigint NOT NULL CHECK (epoch >= 1), - b_key_id text NOT NULL, - b_receipt_digest text NOT NULL CHECK (b_receipt_digest ~ '^[0-9a-f]{64}$'), - activation_seq bigint NOT NULL CHECK (activation_seq >= 1), - prior_target_grant_digest text CHECK (prior_target_grant_digest IS NULL OR prior_target_grant_digest ~ '^[0-9a-f]{64}$'), - prev_activation_digest text CHECK (prev_activation_digest IS NULL OR prev_activation_digest ~ '^[0-9a-f]{64}$'), - activation_digest text NOT NULL CHECK (activation_digest ~ '^[0-9a-f]{64}$'), - grant_digest text NOT NULL CHECK (grant_digest ~ '^[0-9a-f]{64}$'), - grant_json text NOT NULL, - guard_key_id text NOT NULL, - guard_signature text NOT NULL, - created_at timestamptz NOT NULL DEFAULT now(), - PRIMARY KEY (stream_id, activation_seq), - UNIQUE (stream_id, command_id), - UNIQUE (stream_id, activation_digest) -) +${SOURCE_ACTIVATION_V2_SCHEMA} `.trim(); export const HA_CONTROL_TABLES = [ @@ -356,6 +434,7 @@ export const HA_CONTROL_TABLES = [ 'tsk_ha_source_activation', 'tsk_ha_source_activation_history', ] as const; const MANIFEST_TABLES = [...HA_CONTROL_TABLES]; // attest the FULL set incl tsk_ha_schema (R4-H3) +const HA_CONTROL_V1_TABLES = HA_CONTROL_TABLES.filter((t) => t !== 'tsk_ha_source_activation_history'); // (R8-HIGH) governed tables locked ACCESS SHARE per critical tx to freeze the catalog against a // concurrent ALTER/DROP between attest and mutation. Names are compile-time constants (not input). const GOVERNED_LOCK_LIST = HA_CONTROL_TABLES.join(', '); @@ -407,8 +486,7 @@ async function enterCriticalTx(exec: PgExecutor, schema: string): Promise * boundary — the runtime role MUST NOT hold DDL rights (no CREATE/ALTER/DROP) on this schema; * provisioning/migration run under a separate privileged role, offline from serving. */ -async function controlManifest(exec: PgExecutor): Promise { - const tables = [...MANIFEST_TABLES]; +async function controlManifestFor(exec: PgExecutor, tables: readonly string[], version: number): Promise { const cols = (await exec.query( `SELECT table_name, ordinal_position, column_name, data_type, is_nullable, COALESCE(column_default,'') AS column_default FROM information_schema.columns WHERE table_schema = pg_catalog.current_schema() AND table_name = ANY($1) @@ -433,7 +511,7 @@ async function controlManifest(exec: PgExecutor): Promise { `SELECT tablename AS t, policyname AS n, permissive, roles::text AS roles, cmd, COALESCE(qual,'') AS qual, COALESCE(with_check,'') AS wc FROM pg_catalog.pg_policies WHERE schemaname = pg_catalog.current_schema() AND tablename = ANY($1) ORDER BY tablename, policyname`, [tables])).rows; return [ - `V${CONTROL_SCHEMA_VERSION}`, + `V${version}`, ...cols.map((r) => `C|${r.table_name}|${r.ordinal_position}|${r.column_name}|${r.data_type}|${r.is_nullable}|${r.column_default}`), ...cons.map((r) => `K|${r.t}|${r.contype}|${r.def}`), ...idx.map((r) => `I|${r.t}|${r.n}|${r.def}`), @@ -443,6 +521,10 @@ async function controlManifest(exec: PgExecutor): Promise { ].join('\n'); } +async function controlManifest(exec: PgExecutor): Promise { + return controlManifestFor(exec, MANIFEST_TABLES, CONTROL_SCHEMA_VERSION); +} + /** COMPILED expected full-catalog manifest digest — pinned in source, computed from * HA_CONTROL_PG_SCHEMA on PostgreSQL 16. Attestation compares the LIVE catalog to THIS; a * dropped/added CHECK, column, index, trigger, or policy fails closed. There is deliberately NO @@ -450,7 +532,8 @@ async function controlManifest(exec: PgExecutor): Promise { * defeat the pin). To RE-PIN after an intentional DDL change, run provisionControlSchema against * the new schema in an OFFLINE, code-reviewed step and copy the digest reported in the attestation * error ("live catalog digest ") into this constant. */ -export const HA_CONTROL_MANIFEST_DIGEST = '769640c27b3310d391f61d9afff36935be84c02ef885ef4a1537a58b6b81612a'; +export const HA_CONTROL_MANIFEST_DIGEST = 'f01ff613773b750256a6bf91a0c93d9e538a0a88de343775ac9d9bd526b59e1a'; +export const HA_CONTROL_V1_MANIFEST_DIGEST = 'a80069c514d7450a5a2f4482d99952f9b2920ee876d14f51ba5d516ff79eefab'; /** Attest the live catalog hashes to the pinned expected manifest (fail-closed, NOT TOFU). */ async function attestControlSchema(exec: PgExecutor): Promise { @@ -530,6 +613,123 @@ export async function provisionControlSchema(db: PgTransactor, schema: string): return mintReady({ db, schema, version: CONTROL_SCHEMA_VERSION, manifestDigest }); } +/** + * Explicit OFFLINE v1 -> v2 migration. Serving must be stopped: this transaction takes + * ACCESS EXCLUSIVE locks over every v1 authority table, attests the exact compiled v1 catalog + * and authority stamp, verifies any legacy activation against both its signed lease grant and + * the signed ACTIVE cutover chain, then converts it into signed v2 head+history evidence. + * The v2 authority stamp is written only after the converted live catalog attests exactly. + */ +export interface ControlSchemaMigrationDb { + transaction(fn: (exec: PgExecutor) => Promise): Promise; +} + +export async function migrateControlSchemaV1ToV2( + db: ControlSchemaMigrationDb, + schema: string, + migrationSigner: GuardSigner, + controlResolver: GuardKeyResolver, + sourceLeaseResolver: SourceVerifyKeyResolver, +): Promise { + let manifestDigest = ''; + await db.transaction(async (exec) => { + await assertSerializable(exec); + await pinSchema(exec, schema); + const resolved = (await exec.query( + `SELECT t.name, n.nspname + FROM pg_catalog.unnest($1::text[]) AS t(name) + LEFT JOIN pg_catalog.pg_class c ON c.oid = pg_catalog.to_regclass(t.name) + LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace`, [[...HA_CONTROL_V1_TABLES]])).rows; + for (const row of resolved) { + if (row.nspname !== schema) throw new ContractValidationError(`v1 governed table ${String(row.name)} does not resolve to '${schema}' — quarantine`); + } + await exec.query(`LOCK TABLE ${HA_CONTROL_V1_TABLES.join(', ')} IN ACCESS EXCLUSIVE MODE`); + + const stampRows = (await exec.query('SELECT version, catalog_manifest FROM tsk_ha_schema WHERE id=1 FOR UPDATE')).rows; + if (stampRows.length !== 1) throw new ContractValidationError('v1 control authority stamp must be exactly one row'); + if (vInt(stampRows[0].version, 'v1 schema version', 1, MAX_EPOCH) !== 1 || + String(stampRows[0].catalog_manifest) !== HA_CONTROL_V1_MANIFEST_DIGEST) { + throw new ContractValidationError('v1 control authority stamp does not match the compiled v1 authority'); + } + const v1Live = sha256hex(Buffer.from(await controlManifestFor(exec, HA_CONTROL_V1_TABLES, 1), 'utf8')); + if (v1Live !== HA_CONTROL_V1_MANIFEST_DIGEST) { + throw new ContractValidationError(`v1 control schema attestation failed: live catalog digest ${v1Live} != pinned ${HA_CONTROL_V1_MANIFEST_DIGEST}`); + } + + const legacyRows = (await exec.query( + 'SELECT stream_id,command_id,epoch,b_key_id,b_receipt_digest,grant_digest,grant_json,guard_key_id,created_at FROM tsk_ha_source_activation ORDER BY stream_id')).rows; + const converted: Array<{ values: unknown[]; createdAt: unknown }> = []; + for (const row of legacyRows) { + const streamId = vId(row.stream_id, STREAM_ID_RE, 'legacy activation stream_id'); + const commandId = vId(row.command_id, ID_RE, 'legacy activation command_id'); + const epoch = vInt(row.epoch, 'legacy activation epoch', 1, MAX_EPOCH); + const holder = vId(row.b_key_id, KEY_ID_RE, 'legacy activation target key'); + const receiptDigest = vDigest(row.b_receipt_digest, 'legacy activation receipt digest'); + const grantDigest = vDigest(row.grant_digest, 'legacy activation grant digest'); + let grant: LeaseGrant; + try { grant = frozenSnapshot(JSON.parse(String(row.grant_json)) as LeaseGrant); } + catch { throw new FenceAuthorityQuarantineError('legacy activation grant_json is malformed — quarantine'); } + verifyLeaseGrant(sourceLeaseResolver, grant); + if (grant.streamId !== streamId || grant.commandId !== commandId || grant.leaseEpoch !== epoch || + grant.holderNodeId !== holder || grant.grantDigest !== grantDigest || + grant.guardKeyId !== String(row.guard_key_id) || grant.leaseStatus !== 'active' || + grant.leaseGrantSeq !== 1 || grant.prevGrantDigest !== null) { + throw new FenceAuthorityQuarantineError('legacy activation row does not exactly bind its signed genesis grant — quarantine'); + } + + const head = (await exec.query( + 'SELECT epoch,command_id,seqno,phase,evidence,prev_state_digest,state_digest,guard_key_id,guard_signature FROM tsk_ha_cutover_head WHERE stream_id=$1', [streamId])).rows[0]; + const history = (await exec.query( + 'SELECT epoch,command_id,seqno,phase,evidence,prev_state_digest,state_digest,guard_key_id,guard_signature FROM tsk_ha_cutover_history WHERE stream_id=$1 ORDER BY seqno', [streamId])).rows; + if (!head) throw new FenceAuthorityQuarantineError('legacy activation has no signed cutover head — quarantine'); + const link = (r: Record): Linked => { + const e = vInt(r.epoch, 'cutover epoch', 0, MAX_EPOCH); + const cmd = vId(r.command_id, ID_RE, 'cutover command_id'); + const seq = vInt(r.seqno, 'cutover seqno', 1, MAX_SEQ); + const phase = String(r.phase); + const evidence = r.evidence === null || r.evidence === undefined ? null : String(r.evidence); + const prev = vNullableDigest(r.prev_state_digest, 'cutover prev digest'); + const digest = vDigest(r.state_digest, 'cutover state digest'); + return { seq, prev, digest, keyId: String(r.guard_key_id), sig: String(r.guard_signature), + digMsg: cutMsg(streamId, e, cmd, seq, phase, evidence, prev, ''), + sigMsg: cutMsg(streamId, e, cmd, seq, phase, evidence, prev, digest) }; + }; + verifyChain(controlResolver, history.map(link), link(head)); + if (String(head.phase) !== 'ACTIVE' || vInt(head.epoch, 'cutover epoch', 1, MAX_EPOCH) !== epoch || String(head.command_id) !== commandId) { + throw new FenceAuthorityQuarantineError('legacy activation is not backed by its exact terminal ACTIVE cutover — quarantine'); + } + const activeEvidence = JSON.parse(String(head.evidence ?? '{}')) as { bReceiptDigest?: string; bKeyId?: string }; + if (activeEvidence.bReceiptDigest !== receiptDigest || activeEvidence.bKeyId !== holder) { + throw new FenceAuthorityQuarantineError('legacy activation differs from signed ACTIVE evidence — quarantine'); + } + + const activationSeq = 1; + const priorTargetGrantDigest = null; + const prevActivationDigest = null; + const activationDigest = digestOf(activationMsg(streamId, commandId, epoch, holder, + receiptDigest, activationSeq, priorTargetGrantDigest, grantDigest, prevActivationDigest, '')); + const signature = migrationSigner.sign(activationMsg(streamId, commandId, epoch, holder, + receiptDigest, activationSeq, priorTargetGrantDigest, grantDigest, prevActivationDigest, activationDigest)); + converted.push({ values: [streamId, commandId, epoch, holder, receiptDigest, activationSeq, + priorTargetGrantDigest, prevActivationDigest, activationDigest, grantDigest, + JSON.stringify(grant), migrationSigner.id, signature], createdAt: row.created_at }); + } + + await exec.query('DROP TABLE tsk_ha_source_activation'); + for (const statement of SOURCE_ACTIVATION_V2_SCHEMA.split(';').map((s) => s.trim()).filter(Boolean)) await exec.query(statement); + for (const item of converted) { + await exec.query('INSERT INTO tsk_ha_source_activation_history (stream_id,command_id,epoch,b_key_id,b_receipt_digest,activation_seq,prior_target_grant_digest,prev_activation_digest,activation_digest,grant_digest,grant_json,guard_key_id,guard_signature,created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)', [...item.values, item.createdAt]); + await exec.query('INSERT INTO tsk_ha_source_activation (stream_id,command_id,epoch,b_key_id,b_receipt_digest,activation_seq,prior_target_grant_digest,prev_activation_digest,activation_digest,grant_digest,grant_json,guard_key_id,guard_signature,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)', [...item.values, item.createdAt]); + } + manifestDigest = sha256hex(Buffer.from(await controlManifest(exec), 'utf8')); + if (manifestDigest !== HA_CONTROL_MANIFEST_DIGEST) throw new ContractValidationError(`migrated v2 control schema attestation failed: live catalog digest ${manifestDigest} != pinned ${HA_CONTROL_MANIFEST_DIGEST}`); + affectedOne(await exec.query( + 'UPDATE tsk_ha_schema SET version=$2,catalog_manifest=$3,applied_at=now() WHERE id=1 AND version=$1 AND catalog_manifest=$4', + [1, CONTROL_SCHEMA_VERSION, manifestDigest, HA_CONTROL_V1_MANIFEST_DIGEST]), 'v1->v2 authority stamp'); + }); + return manifestDigest; +} + // ── public state shapes ────────────────────────────────────────────────────── export interface ProvisioningState { streamId: string; genesisMarker: string; state: 'intent' | 'incomplete' | 'provisioned'; stateSeq: number; stateDigest: string; } @@ -857,6 +1057,47 @@ export class HaControlFencing { return { streamId: s, epoch: vInt(head.epoch, 'epoch', 0, MAX_EPOCH), commandId: String(head.command_id), seqno: vInt(head.seqno, 'seqno', 1, MAX_SEQ), phase: head.phase as CutoverState['phase'], evidence: head.evidence === null || head.evidence === undefined ? null : String(head.evidence), stateDigest: vDigest(head.state_digest, 'state_digest') }; } + /** Verify the signed append-only source-activation authority and return its current epoch. + * The control signature binds the activation tuple and grant digest; target-side lease + * verification remains the responsibility of activateSource/installation. */ + private async readActivationEpoch(exec: PgExecutor, s: string): Promise { + const cols = 'command_id,epoch,b_key_id,b_receipt_digest,activation_seq,prior_target_grant_digest,prev_activation_digest,activation_digest,grant_digest,guard_key_id,guard_signature'; + const head = (await exec.query(`SELECT ${cols} FROM tsk_ha_source_activation WHERE stream_id=$1 FOR SHARE`, [s])).rows[0]; + const history = (await exec.query(`SELECT ${cols} FROM tsk_ha_source_activation_history WHERE stream_id=$1 ORDER BY activation_seq`, [s])).rows; + if (!head) { + if (history.length !== 0) throw new FenceAuthorityQuarantineError('source activation history exists without a head — quarantine'); + return 0; + } + const linked = (row: Record): Linked => { + const seq = vInt(row.activation_seq, 'activation_seq', 1, MAX_SEQ); + const prev = vNullableDigest(row.prev_activation_digest, 'prev_activation_digest'); + const digest = vDigest(row.activation_digest, 'activation_digest'); + const command = vId(row.command_id, ID_RE, 'activation command_id'); + const epoch = vInt(row.epoch, 'activation epoch', 1, MAX_EPOCH); + const targetKey = vId(row.b_key_id, KEY_ID_RE, 'activation target key'); + const receiptDigest = vDigest(row.b_receipt_digest, 'activation receipt digest'); + const priorDigest = vNullableDigest(row.prior_target_grant_digest, 'prior_target_grant_digest'); + const grantDigest = vDigest(row.grant_digest, 'activation grant_digest'); + return { seq, prev, digest, keyId: String(row.guard_key_id), sig: String(row.guard_signature), + digMsg: activationMsg(s, command, epoch, targetKey, receiptDigest, seq, priorDigest, grantDigest, prev, ''), + sigMsg: activationMsg(s, command, epoch, targetKey, receiptDigest, seq, priorDigest, grantDigest, prev, digest) }; + }; + verifyChain(this.resolver, history.map(linked), linked(head)); + return vInt(head.epoch, 'activation epoch', 1, MAX_EPOCH); + } + + private async assertActivationGap(exec: PgExecutor, s: string, currentEpoch: number, targetEpoch: number): Promise { + const skipped = targetEpoch > currentEpoch + 1 ? (await exec.query( + 'SELECT epoch,phase FROM tsk_ha_cutover_history WHERE stream_id=$1 AND epoch>$2 AND epoch<$3 ORDER BY seqno', + [s, currentEpoch, targetEpoch])).rows : []; + const activated = targetEpoch > currentEpoch + 1 ? (await exec.query( + 'SELECT epoch FROM tsk_ha_source_activation_history WHERE stream_id=$1 AND epoch>$2 AND epoch<$3 ORDER BY epoch', + [s, currentEpoch, targetEpoch])).rows : []; + assertConsumedActivationEpochs(currentEpoch, targetEpoch, + skipped.map((row) => ({ epoch: vInt(row.epoch, 'skipped cutover epoch', 1, MAX_EPOCH), phase: String(row.phase) })), + activated.map((row) => vInt(row.epoch, 'skipped activation epoch', 1, MAX_EPOCH))); + } + private async cutTransition(exec: PgExecutor, s: string, epoch: number, commandId: string, seqno: number, phase: CutoverState['phase'], evidence: string | null, prev: string | null): Promise { const digest = digestOf(cutMsg(s, epoch, commandId, seqno, phase, evidence, prev, '')); const sig = this.signer.sign(cutMsg(s, epoch, commandId, seqno, phase, evidence, prev, digest)); @@ -892,6 +1133,34 @@ export class HaControlFencing { }); } + /** + * Consume a post-fence promotion epoch that cannot complete. Only FENCED/IMPORTING/READY may + * become ABORTED: at those phases the signed witness has already advanced, so reusing the epoch + * would be unsafe. Pre-fence intents remain resumable and cannot consume an epoch through this API. + */ + async abortFencedPromotion(streamId: string, commandId: string, targetEpoch: number, reasonCode: string): Promise { + const s = vId(streamId, STREAM_ID_RE, 'streamId'); + const cmd = vId(commandId, ID_RE, 'commandId'); + const target = vInt(targetEpoch, 'targetEpoch', 1, MAX_EPOCH); + const reason = vId(reasonCode, ID_RE, 'reasonCode'); + return this.criticalTx(s, async (exec) => { + const cur = await this.readCutover(exec, s); + const evidence = JSON.stringify({ k: 'abort/v1', reason }); + if (cur?.phase === 'ABORTED' && cur.commandId === cmd && cur.epoch === target) { + if (cur.evidence !== evidence) throw new FenceAuthorityQuarantineError('ABORTED retry reason differs from signed evidence — quarantine'); + return cur; + } + if (!cur || cur.commandId !== cmd || cur.epoch !== target || !new Set(['FENCED', 'IMPORTING', 'READY']).has(cur.phase)) { + throw new FenceAuthorityQuarantineError('only a matching post-fence cutover may consume an ABORTED epoch — quarantine'); + } + const witness = await this.readWitness(exec, s); + if (!witness || witness.state !== 'provisioned' || witness.epoch !== target) { + throw new FenceAuthorityQuarantineError('cannot consume ABORTED epoch without the exact signed witness advance — quarantine'); + } + return this.cutTransition(exec, s, target, cmd, cur.seqno + 1, 'ABORTED', evidence, cur.stateDigest); + }); + } + /** * Advance the epoch (fence the old writer) for a SOURCE_FENCED intent — the source freeze MUST already be * bound (bindSourceFenced) before a control can FENCE. This makes the §4 ordering a single-chokepoint @@ -989,14 +1258,15 @@ export class HaControlFencing { verifySourceFrozenReceipt(resolver, fr); // verifies A's ed25519 signature over the frozen receipt — throws on tamper if (fr.commandId !== cmd) throw new FenceAuthorityQuarantineError('frozen receipt commandId != cutover command — quarantine'); if (fr.streamId !== s) throw new FenceAuthorityQuarantineError('frozen receipt streamId != cutover stream — quarantine'); - // the freeze at source epoch E promotes to control target epoch E+1 — reject a cross-epoch splice. - if (fr.epoch !== target - 1) throw new FenceAuthorityQuarantineError(`frozen receipt epoch ${fr.epoch} != targetEpoch-1 (${target - 1}) — cross-epoch freeze; quarantine`); // canonical bound-field set — written as evidence AND re-asserted on an idempotent retry (H3). const bound = { k: 'source_fenced/v1', frozenReceiptDigest: fr.receiptDigest, sourceEpoch: fr.epoch, n: fr.n, headAtN: fr.signedHeadDigestAtN, stateAtN: fr.sourceStateDigestAtN, sourceNodeId: fr.sourceNodeId, revokeCommandId: fr.revokeCommandId, leaseId: fr.leaseId, leaseGrantDigest: fr.leaseGrantDigest, sourceKeyId: fr.sourceKeyId }; const ev = JSON.stringify(bound); return this.criticalTx(s, async (exec) => { + const sourceEpoch = await this.readActivationEpoch(exec, s); + await this.assertActivationGap(exec, s, sourceEpoch, target); + if (fr.epoch !== sourceEpoch) throw new FenceAuthorityQuarantineError(`frozen receipt epoch ${fr.epoch} != current signed source activation epoch ${sourceEpoch} — cross-epoch freeze; quarantine`); const cur = await this.readCutover(exec, s); if (cur && cur.phase === 'SOURCE_FENCED' && cur.commandId === cmd && cur.epoch === target) { // (H3) idempotent ONLY if the incoming receipt equals the stored signed evidence, byte-for-byte. @@ -1035,7 +1305,8 @@ export class HaControlFencing { * the source/guard public-key signatures, which were checked at export/finalize), then binds it to THIS cutover: * • commandId + streamId match the active cutover command; * • frozenReceiptDigest / n / head@N / state@N match the SOURCE_FENCED-bound freeze byte-for-byte; - * • the freeze epoch is targetEpoch-1 and B's receipt epoch equals it (no cross-epoch splice); + * • the freeze/receipt epoch equals the last signed source activation epoch; any intervening + * control epochs are individually proven terminal ABORTED (no cross-epoch splice); * • B's system_identifier is DISTINCT from BOTH the signed source system_identifier (B != source) AND * control's OWN durable PG system_identifier (B != control — the capability §4 was deferred to hold). * The READY head (control-signed) records B's finalize digest + both distinct system_identifiers, so a @@ -1066,8 +1337,11 @@ export class HaControlFencing { // B's finalize MUST bind the EXACT frozen N control bound at SOURCE_FENCED (no cross-freeze splice). if (br.frozenReceiptDigest !== sf.frozenReceiptDigest) throw new FenceAuthorityQuarantineError('BFinalizedReceipt frozenReceiptDigest != the bound SOURCE_FENCED freeze — quarantine'); if (br.n !== sf.n || br.signedHeadDigestAtN !== sf.headAtN || br.sourceStateDigestAtN !== sf.stateAtN) throw new FenceAuthorityQuarantineError('BFinalizedReceipt N/head@N/state@N != the bound freeze — quarantine'); - // epoch binding: the bound freeze is at source epoch target-1, and B's receipt must be at that SAME epoch. - if (sf.sourceEpoch !== target - 1) throw new FenceAuthorityQuarantineError(`bound SOURCE_FENCED sourceEpoch ${sf.sourceEpoch} != targetEpoch-1 (${target - 1}) — quarantine`); + // Epoch binding: the freeze is at the last signed source activation epoch. One or more + // intervening control epochs are permitted only when each is a signed terminal ABORTED cutover. + const sourceEpoch = await this.readActivationEpoch(exec, s); + await this.assertActivationGap(exec, s, sourceEpoch, target); + if (sf.sourceEpoch !== sourceEpoch) throw new FenceAuthorityQuarantineError(`bound SOURCE_FENCED sourceEpoch ${sf.sourceEpoch} != current signed source activation epoch ${sourceEpoch} — quarantine`); if (br.epoch !== sf.sourceEpoch) throw new FenceAuthorityQuarantineError(`BFinalizedReceipt epoch ${br.epoch} != the bound freeze sourceEpoch ${sf.sourceEpoch} — cross-epoch splice; quarantine`); // Distinctness: B != source (signed in the receipt) AND B != control (control's OWN in-tx sysid). const controlSystemId = await this.controlSystemId(exec); @@ -1119,33 +1393,55 @@ export class HaControlFencing { * SOURCE capability: a GUARD-signed (ed25519) lease grant that lets it become writable AT THE * PROMOTED EPOCH. This binds the grant to the ratified promotion — it requires the ACTIVE head for this * exact command/epoch, re-verifies the target's BFinalizedReceipt, and asserts the ACTIVE evidence pins - * that same target. A never-used target receives a genesis lease. A returning target must supply its exact - * signed, terminal lease high-water; the returned grant appends to that chain. Activation itself advances + * that same target. A never-used target receives a genesis lease. A returning target must supply its full + * signed, contiguous lease chain through the terminal high-water; when it has a prior control activation, + * that activation grant must be present in the chain. The returned grant appends to that chain. Activation advances * a signed append-only control history, so an older activation head cannot be replayed. Control never sees * the target's private material. */ async activateSource(streamId: string, commandId: string, targetEpoch: number, bReceipt: BFinalizedReceipt, bResolver: SourceVerifyKeyResolver, - priorTargetLease: LeaseGrant | null = null): Promise { + leaseResolver: SourceVerifyKeyResolver, + priorTargetLeaseChain: LeaseGrant[] = []): Promise { const s = vId(streamId, STREAM_ID_RE, 'streamId'); const cmd = vId(commandId, ID_RE, 'commandId'); const target = vInt(targetEpoch, 'targetEpoch', 1, MAX_EPOCH); const guard = this.sourceGuard; if (!guard) throw new ContractValidationError('activateSource requires a configured policy.sourceGuard signing authority'); const br = frozenSnapshot(bReceipt); - const prior = priorTargetLease === null ? null : frozenSnapshot(priorTargetLease); + const priorChain = snapshotLeaseChain(priorTargetLeaseChain); verifyBFinalizedReceipt(bResolver, br); - if (prior !== null) { - verifyLeaseGrant(bResolver, prior); - if (prior.streamId !== s) throw new FenceAuthorityQuarantineError('prior target lease streamId != cutover stream — quarantine'); - if (prior.leaseStatus !== 'revoked') throw new FenceAuthorityQuarantineError('prior target lease must be terminally revoked — quarantine'); - if (prior.leaseEpoch >= target) throw new FenceAuthorityQuarantineError('prior target lease epoch must precede the promoted epoch — quarantine'); + if (!Array.isArray(priorChain)) throw new ContractValidationError('priorTargetLeaseChain must be an array'); + let prior: LeaseGrant | null = null; + let priorDigest: string | null = null; + let chainPrev: string | null = null; + const epochIdentities = new Map(); + for (let i = 0; i < priorChain.length; i++) { + const grant = priorChain[i]; + verifyLeaseGrant(leaseResolver, grant); + if (grant.streamId !== s) throw new FenceAuthorityQuarantineError('prior target lease chain streamId != cutover stream — quarantine'); + if (grant.leaseGrantSeq !== i + 1 || (grant.prevGrantDigest ?? null) !== chainPrev) { + throw new FenceAuthorityQuarantineError('prior target lease chain is not contiguous from genesis — quarantine'); + } + if (grant.leaseEpoch >= target) throw new FenceAuthorityQuarantineError('prior target lease epoch must precede the promoted epoch — quarantine'); + const identity = epochIdentities.get(grant.leaseEpoch); + if (identity && (identity.holderNodeId !== grant.holderNodeId || identity.leaseId !== grant.leaseId)) { + throw new FenceAuthorityQuarantineError('prior target lease chain changes holder/leaseId within an epoch — quarantine'); + } + if (identity?.revoked) throw new FenceAuthorityQuarantineError('prior target lease chain continues after a terminal revoke — quarantine'); + epochIdentities.set(grant.leaseEpoch, { holderNodeId: grant.holderNodeId, + leaseId: grant.leaseId, revoked: grant.leaseStatus === 'revoked' }); + chainPrev = grant.grantDigest; + prior = grant; } if (br.commandId !== cmd) throw new FenceAuthorityQuarantineError('BFinalizedReceipt commandId != cutover command — quarantine'); if (br.streamId !== s) throw new FenceAuthorityQuarantineError('BFinalizedReceipt streamId != cutover stream — quarantine'); - if (br.epoch !== target - 1) throw new FenceAuthorityQuarantineError(`BFinalizedReceipt epoch ${br.epoch} != targetEpoch-1 (${target - 1}) — quarantine`); // the holder identity is DERIVED from the ratified B (its signing key id), never accepted from the caller. const holder = vId(br.bKeyId, ID_RE, 'ratified B key id (holder)'); + if (prior !== null && (prior.leaseStatus !== 'revoked' || prior.holderNodeId !== holder)) { + throw new FenceAuthorityQuarantineError('prior target lease high-water must be revoked and held by the ratified target — quarantine'); + } + priorDigest = prior?.grantDigest ?? null; const leaseId = vId(`bsrc-${cmd}`, ID_RE, 'derived leaseId'); return this.criticalTx(s, async (exec) => { const cur = await this.readCutover(exec, s); @@ -1167,6 +1463,16 @@ export class HaControlFencing { const command = vId(row.command_id, ID_RE, 'activation command_id'); const targetKey = vId(row.b_key_id, KEY_ID_RE, 'activation target key'); const receiptDigest = vDigest(row.b_receipt_digest, 'activation receipt digest'); + let storedGrant: LeaseGrant; + try { storedGrant = JSON.parse(String(row.grant_json)) as LeaseGrant; } + catch { throw new FenceAuthorityQuarantineError('activation grant_json is malformed — quarantine'); } + try { verifyLeaseGrant(leaseResolver, storedGrant); } + catch (error) { throw new FenceAuthorityQuarantineError(`activation grant_json is not a valid signed lease grant — quarantine: ${error instanceof Error ? error.message : 'unknown verification error'}`); } + if (storedGrant.grantDigest !== grantDigest || storedGrant.streamId !== s || + storedGrant.commandId !== command || storedGrant.leaseEpoch !== epoch || + storedGrant.holderNodeId !== targetKey) { + throw new FenceAuthorityQuarantineError('activation grant_json does not bind its signed activation row — quarantine'); + } const digMsg = activationMsg(s, command, epoch, targetKey, receiptDigest, seq, priorDigest, grantDigest, prev, ''); return { seq, prev, digest, keyId: String(row.guard_key_id), @@ -1177,7 +1483,6 @@ export class HaControlFencing { if (headRow) verifyChain(this.resolver, historyRows.map(linked), linked(headRow)); else if (historyRows.length !== 0) throw new FenceAuthorityQuarantineError('activation history exists without a head — quarantine'); - const priorDigest = prior?.grantDigest ?? null; if (headRow && String(headRow.command_id) === cmd) { if (vInt(headRow.epoch, 'epoch', 1, MAX_EPOCH) !== target || String(headRow.b_key_id) !== holder || @@ -1186,13 +1491,27 @@ export class HaControlFencing { throw new FenceAuthorityQuarantineError('source activation retry does not byte-bind the current activation — quarantine'); } const existingGrant = JSON.parse(String(headRow.grant_json)) as LeaseGrant; - verifyLeaseGrant(bResolver, existingGrant); + verifyLeaseGrant(leaseResolver, existingGrant); if (existingGrant.grantDigest !== String(headRow.grant_digest)) throw new FenceAuthorityQuarantineError('stored source activation grant digest mismatch — quarantine'); return existingGrant; } + const priorTargetActivationRows = historyRows.filter((row) => String(row.b_key_id) === holder); + if (priorTargetActivationRows.length > 0) { + if (prior === null) throw new FenceAuthorityQuarantineError('returning target omitted its lease chain — quarantine'); + const lastTargetActivation = priorTargetActivationRows[priorTargetActivationRows.length - 1]; + const rootedGrant = JSON.parse(String(lastTargetActivation.grant_json)) as LeaseGrant; + verifyLeaseGrant(leaseResolver, rootedGrant); + const atExpectedSequence = priorChain[rootedGrant.leaseGrantSeq - 1]; + if (!atExpectedSequence || atExpectedSequence.grantDigest !== rootedGrant.grantDigest) { + throw new FenceAuthorityQuarantineError('prior target lease chain is not rooted at its last control activation at the expected sequence — quarantine'); + } + } else if (prior !== null && (priorChain[0].leaseGrantSeq !== 1 || priorChain[0].prevGrantDigest !== null)) { + throw new FenceAuthorityQuarantineError('first return target lease chain is not rooted at genesis — quarantine'); + } const currentActivationSeq = headRow ? vInt(headRow.activation_seq, 'activation_seq', 1, MAX_SEQ) : 0; const currentActivationEpoch = headRow ? vInt(headRow.epoch, 'activation epoch', 1, MAX_EPOCH) : 0; - if (target !== currentActivationEpoch + 1) throw new FenceAuthorityQuarantineError(`source activation epoch ${target} must follow current ${currentActivationEpoch} exactly — quarantine`); + await this.assertActivationGap(exec, s, currentActivationEpoch, target); + if (br.epoch !== currentActivationEpoch) throw new FenceAuthorityQuarantineError(`BFinalizedReceipt epoch ${br.epoch} != current signed source activation epoch ${currentActivationEpoch} — quarantine`); // A never-used target receives a genesis grant. A returning target supplies its exact signed, // terminal lease high-water, so the new grant appends to that target's immutable chain. diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index eb5f876..8fdac06 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -22,6 +22,7 @@ export { StreamHeadVerificationUnavailableError, PgTskDurableOutbox, UnfencedSingleNodeTskDurableOutbox, + activateFinalizedReceiverAsSource, PgTskPublisher, PgTskReceiverCheckpoint, schemaManifest, @@ -54,6 +55,7 @@ export type { TskOutboxTransport, HotpApplier, PgTskOutboxOptions, + ImportedSourceActivationOptions, SourceFenceGate, PgTskPublisherOptions, TskDrainResult, @@ -70,6 +72,8 @@ export { HA_CONTROL_PG_SCHEMA, HA_CONTROL_TABLES, HA_CONTROL_MANIFEST_DIGEST, + HA_CONTROL_V1_MANIFEST_DIGEST, + HA_CONTROL_V1_SOURCE_ACTIVATION_SCHEMA, CONTROL_SCHEMA_VERSION, HaControlFencing, GuardSigner, @@ -80,6 +84,7 @@ export { assertRedisAuthority, reconcileFencedRedis, provisionControlSchema, + migrateControlSchemaV1ToV2, assertControlSchemaReady, FenceAuthorityQuarantineError, } from './ha-control-fencing.js'; @@ -158,6 +163,7 @@ export type { export type { GuardKeyResolver, ControlSchemaReadyToken, + ControlSchemaMigrationDb, HaControlPolicy, ProvisioningState, LeaseState, diff --git a/packages/server/src/tsk-hotp-outbox-pg.ts b/packages/server/src/tsk-hotp-outbox-pg.ts index b2dd72e..1aa466f 100644 --- a/packages/server/src/tsk-hotp-outbox-pg.ts +++ b/packages/server/src/tsk-hotp-outbox-pg.ts @@ -30,6 +30,7 @@ import { assertHeaderConformant, assertStreamHeadBinds, canonicalOpDigest, + canonicalize, isTerminalTransportError, fenceTokenToDecimal, streamHeadDigest, @@ -37,6 +38,7 @@ import { type EpochTransitionAuthorizer, type FenceToken, type HotpMutationSanitizer, + type MutationSanitizer, type OutboxRecord, type OutboxRecordHeader, type PublisherBackpressure, @@ -48,8 +50,17 @@ import { type TskHotpMutation, type TskReceiverCheckpoint, } from './ha-outbox-contract.js'; -import { assertSourceLeaseWritable, requireSourceFenceReady } from './tsk-source-fence.js'; -import type { SourceVerifyKeyResolver, SourceFenceReadyToken } from './tsk-source-fence.js'; +import { + assertSourceLeaseWritable, requireSourceFenceReady, requireReceiverReady, + attestReceiver, attestSourceLease, verifyGuardCountersignedExport, + assertExportBundleBinds, verifyBFinalizedReceipt, verifyLeaseGrant, + installLeaseGrant, TSK_RECEIVER_TABLES, TSK_SOURCE_LEASE_TABLES, +} from './tsk-source-fence.js'; +import type { + SourceVerifyKeyResolver, SourceFenceReadyToken, SourceReceiverReadyToken, + SourceExportBundle, GuardCountersignedExport, SourceFrozenReceipt, + BFinalizedReceipt, LeaseGrant, +} from './tsk-source-fence.js'; /** Contract HOTP counter bound (mirrors the TSK segment counter ceiling). */ export const TSK_HOTP_MAX_COUNTER = 2_147_483_647; @@ -740,6 +751,254 @@ export class UnfencedSingleNodeTskDurableOutbox extends AbstractTskDurableOutbox } } +export interface ImportedSourceActivationOptions { + sanitizer: Pick, 'sanitize'>; + sourceResolver: SourceVerifyKeyResolver; + guardResolver: SourceVerifyKeyResolver; + headResolver: SourceVerifyKeyResolver; + frozenResolver: SourceVerifyKeyResolver; + bReceiptResolver: SourceVerifyKeyResolver; + leaseResolver: SourceVerifyKeyResolver; + frozenReceipt: SourceFrozenReceipt; + finalizedReceipt: BFinalizedReceipt; + activationLease: LeaseGrant; + targetEpoch: number; + maxSnapshotBytes?: number; +} + +function validateSnapshotBytes(value: number | undefined): number { + const n = value ?? 32 * 1024 * 1024; + if (!Number.isSafeInteger(n) || n < 1 || n > 64 * 1024 * 1024) { + throw new ContractValidationError('maxSnapshotBytes must be a safe integer in [1, 67108864]'); + } + return n; +} + +function strictPlainSnapshot(input: T, label: string, seen = new Set()): T { + if (input === null || typeof input === 'string' || typeof input === 'boolean') return input; + if (typeof input === 'number') { + if (!Number.isSafeInteger(input)) throw new ContractValidationError(`${label} contains a non-safe integer`); + return input; + } + if (typeof input !== 'object') throw new ContractValidationError(`${label} contains unsupported data`); + const object = input as unknown as object; + if (seen.has(object)) throw new ContractValidationError(`${label} contains a cycle`); + seen.add(object); + if (Object.getOwnPropertySymbols(object).length !== 0) throw new ContractValidationError(`${label} contains symbol keys`); + const proto = Object.getPrototypeOf(object); + if (proto !== Object.prototype && proto !== Array.prototype) throw new ContractValidationError(`${label} must contain only plain data`); + const out: unknown[] | Record = Array.isArray(object) ? [] : {}; + const names = Object.getOwnPropertyNames(object); + if (Array.isArray(object)) { + const expected = Array.from({ length: object.length }, (_, i) => String(i)); + if (names.filter((n) => n !== 'length').join('\0') !== expected.join('\0')) throw new ContractValidationError(`${label} contains a sparse or extra-key array`); + } + for (const name of names) { + if (name === 'length' && Array.isArray(object)) continue; + const descriptor = Object.getOwnPropertyDescriptor(object, name); + if (!descriptor || !('value' in descriptor)) throw new ContractValidationError(`${label} contains an accessor`); + const value = strictPlainSnapshot(descriptor.value, `${label}.${name}`, seen); + if (Array.isArray(out)) out[Number(name)] = value; + else out[name] = value; + } + seen.delete(object); + return Object.freeze(out) as T; +} + +function ownData(object: object, name: string, label: string): T { + const descriptor = Object.getOwnPropertyDescriptor(object, name); + if (!descriptor || !('value' in descriptor)) throw new ContractValidationError(`${label}.${name} must be an own data property`); + return descriptor.value as T; +} + +/** Atomically turn an already-finalized receiver generation into a writable source ledger. + * The bundle is rebuilt from isolated persisted staging, both export signatures and the full + * 1..N signed-head chain are re-verified, and the exact target receipt is re-verified. Only then, + * in the SAME SERIALIZABLE transaction, are missing ledger rows installed, the fence/checkpoint + * advanced, and the guard-signed activation lease appended. A crash exposes all or none. Receipt/lease + * inputs are strict plain-data snapshots taken synchronously; resolver/sanitizer objects are trusted + * capability handles captured before the first await. */ +export async function activateFinalizedReceiverAsSource( + db: PgTransactor, + outboxReady: SchemaReadyToken, + receiverReady: SourceReceiverReadyToken, + streamIdInput: string, + generationIdInput: string, + rawOptions: ImportedSourceActivationOptions, +): Promise<{ streamId: string; generationId: string; targetEpoch: number; n: number; + headDigest: string; activationGrantDigest: string }> { + const schema = requireReady(outboxReady, db); + const receiverBinding = requireReceiverReady(receiverReady, { db }); + if (receiverBinding.schema !== schema) throw new ContractValidationError('receiver/outbox readiness schema mismatch'); + const streamId = reqString(streamIdInput, 'streamId'); + const generationId = reqString(generationIdInput, 'generationId'); + if (!rawOptions || Object.getPrototypeOf(rawOptions) !== Object.prototype || Object.getOwnPropertySymbols(rawOptions).length !== 0) { + throw new ContractValidationError('source activation options must be an exact plain object'); + } + const optionKeys = Object.getOwnPropertyNames(rawOptions).sort(); + const allowedKeys = ['activationLease', 'bReceiptResolver', 'finalizedReceipt', 'frozenReceipt', 'frozenResolver', 'guardResolver', + 'headResolver', 'leaseResolver', 'maxSnapshotBytes', 'sanitizer', 'sourceResolver', 'targetEpoch']; + if (optionKeys.some((key) => !allowedKeys.includes(key)) || allowedKeys.filter((key) => key !== 'maxSnapshotBytes').some((key) => !optionKeys.includes(key))) { + throw new ContractValidationError('source activation options have missing or extra keys'); + } + const maxSnapshotBytes = validateSnapshotBytes(optionKeys.includes('maxSnapshotBytes') + ? ownData(rawOptions, 'maxSnapshotBytes', 'source activation options') : undefined); + const frozenReceipt = strictPlainSnapshot(ownData(rawOptions, 'frozenReceipt', 'source activation options'), 'frozenReceipt'); + const finalizedReceipt = strictPlainSnapshot(ownData(rawOptions, 'finalizedReceipt', 'source activation options'), 'finalizedReceipt'); + const activationLease = strictPlainSnapshot(ownData(rawOptions, 'activationLease', 'source activation options'), 'activationLease'); + const snapshotJson = JSON.stringify({ + frozenReceipt, finalizedReceipt, activationLease, + }); + if (Buffer.byteLength(snapshotJson, 'utf8') > maxSnapshotBytes) throw new ContractValidationError('source activation snapshot exceeds maxSnapshotBytes'); + const captured = { frozenReceipt, finalizedReceipt, activationLease }; + const sanitizer = ownData(rawOptions, 'sanitizer', 'source activation options'); + const sourceResolver = ownData(rawOptions, 'sourceResolver', 'source activation options'); + const guardResolver = ownData(rawOptions, 'guardResolver', 'source activation options'); + const headResolver = ownData(rawOptions, 'headResolver', 'source activation options'); + const frozenResolver = ownData(rawOptions, 'frozenResolver', 'source activation options'); + const bReceiptResolver = ownData(rawOptions, 'bReceiptResolver', 'source activation options'); + const leaseResolver = ownData(rawOptions, 'leaseResolver', 'source activation options'); + const targetEpoch = safeSeq(ownData(rawOptions, 'targetEpoch', 'source activation options'), 'targetEpoch'); + if (targetEpoch < 1 || targetEpoch > 2_147_483_647) throw new ContractValidationError('targetEpoch out of range [1, 2^31-1]'); + verifyBFinalizedReceipt(bReceiptResolver, captured.finalizedReceipt); + verifyLeaseGrant(leaseResolver, captured.activationLease); + if (captured.finalizedReceipt.streamId !== streamId || + captured.finalizedReceipt.generationId !== generationId || + captured.finalizedReceipt.bKeyId !== captured.activationLease.holderNodeId || + captured.finalizedReceipt.epoch >= targetEpoch || + captured.activationLease.streamId !== streamId || + captured.activationLease.leaseEpoch !== targetEpoch || + captured.activationLease.leaseStatus !== 'active' || + captured.activationLease.commandId !== captured.finalizedReceipt.commandId) { + throw new ContractValidationError('activation lease/finalized receipt does not bind the target generation'); + } + return db.transaction(async (exec) => { + await enterCriticalTx(exec, schema); + // Freeze every governed catalog through commit: ACCESS SHARE blocks ALTER/DROP while allowing + // normal row work. Re-attest inside this exact authority transaction, not only at token mint. + await exec.query(`LOCK TABLE ${[...TSK_OUTBOX_TABLES, ...TSK_RECEIVER_TABLES, ...TSK_SOURCE_LEASE_TABLES].join(', ')} IN ACCESS SHARE MODE`); + await attestSchema(exec); + await assertVersionInTx(exec); + await attestReceiver(exec); + await attestSourceLease(exec); + const targetSystemId = String((await exec.query('SELECT system_identifier::text AS s FROM pg_catalog.pg_control_system()')).rows[0]?.s ?? ''); + if (targetSystemId !== captured.finalizedReceipt.bSystemId) throw new ContractValidationError('activation target PostgreSQL system_identifier differs from the finalized receiver authority'); + const pointer = (await exec.query( + `SELECT p.active_generation_id,p.checkpoint_seq,p.head_digest,p.state_digest,p.b_finalized_receipt, + g.command_id,g.epoch,g.source_epoch,g.n,g.manifest_digest,g.manifest_root,g.frozen_receipt_digest + FROM tsk_receiver_pointer p JOIN tsk_receiver_generation g + ON g.stream_id=p.stream_id AND g.generation_id=p.active_generation_id + WHERE p.stream_id=$1 FOR UPDATE OF p,g`, [streamId], + )).rows[0]; + if (!pointer || String(pointer.active_generation_id) !== generationId) throw new ContractValidationError('receiver generation is not the active finalized generation'); + const storedReceipt = typeof pointer.b_finalized_receipt === 'string' + ? JSON.parse(pointer.b_finalized_receipt) : pointer.b_finalized_receipt; + if (canonicalize(storedReceipt) !== canonicalize(captured.finalizedReceipt)) throw new ContractValidationError('stored finalized receipt differs from the activation receipt'); + const stage = (await exec.query( + 'SELECT manifest,pg_catalog.octet_length(manifest::text) AS manifest_bytes FROM tsk_receiver_stage WHERE stream_id=$1 AND generation_id=$2 AND status=$3 FOR SHARE', + [streamId, generationId, 'installed'], + )).rows[0]; + if (!stage) throw new ContractValidationError('installed receiver staging manifest is missing'); + const chunkStats = (await exec.query( + 'SELECT count(*) AS n,COALESCE(sum(pg_catalog.octet_length(content::text)),0) AS bytes FROM tsk_receiver_stage_chunk WHERE stream_id=$1 AND generation_id=$2', + [streamId, generationId], + )).rows[0]; + const chunkCount = safeSeq(chunkStats.n, 'receiver staging chunk count'); + const persistedBytes = safeSeq(chunkStats.bytes, 'receiver staging bytes') + safeSeq(stage.manifest_bytes, 'receiver manifest bytes'); + if (chunkCount < 2 || chunkCount > 100_000 || persistedBytes > maxSnapshotBytes) { + throw new ContractValidationError('persisted receiver activation material exceeds configured count/byte bounds'); + } + const manifest = (typeof stage.manifest === 'string' ? JSON.parse(stage.manifest) : stage.manifest) as GuardCountersignedExport; + const chunkRows = (await exec.query( + 'SELECT ordinal,kind,seq_from,seq_to,item_count,byte_digest,content FROM tsk_receiver_stage_chunk WHERE stream_id=$1 AND generation_id=$2 ORDER BY ordinal FOR SHARE', + [streamId, generationId], + )).rows; + const historyChunks = chunkRows.filter((row) => row.kind === 'history').map((row) => ({ + ordinal: Number(row.ordinal), seqFrom: Number(row.seq_from), seqTo: Number(row.seq_to), + records: typeof row.content === 'string' ? JSON.parse(row.content) : row.content, + byteDigest: String(row.byte_digest), + })); + const stateRows = chunkRows.filter((row) => row.kind === 'state'); + if (stateRows.length !== 1) throw new ContractValidationError('receiver staging must contain exactly one state chunk'); + const stateChunk = { + ordinal: Number(stateRows[0].ordinal), + pairs: typeof stateRows[0].content === 'string' ? JSON.parse(stateRows[0].content) : stateRows[0].content, + itemCount: Number(stateRows[0].item_count), byteDigest: String(stateRows[0].byte_digest), + }; + const bundle = { historyChunks, stateChunk } as SourceExportBundle; + verifyGuardCountersignedExport(sourceResolver, guardResolver, manifest); + const replay = assertExportBundleBinds(bundle, manifest, captured.frozenReceipt, + frozenResolver, sanitizer, headResolver); + if (replay.n !== Number(pointer.n) || replay.n !== captured.finalizedReceipt.n || + replay.signedHeadDigestAtN !== String(pointer.head_digest) || + replay.sourceStateDigestAtN !== String(pointer.state_digest) || + manifest.canonicalDigest !== String(pointer.manifest_digest) || + manifest.manifestRoot !== String(pointer.manifest_root) || + captured.frozenReceipt.receiptDigest !== String(pointer.frozen_receipt_digest)) { + throw new ContractValidationError('persisted receiver generation does not bind replay outputs'); + } + const records = bundle.historyChunks.flatMap((chunk) => chunk.records); + const maxExisting = safeSeq((await exec.query( + 'SELECT COALESCE(MAX(sequence),0) AS n FROM tsk_outbox_rows WHERE stream_id=$1', [streamId], + )).rows[0].n, 'existing source max sequence'); + if (maxExisting > replay.n) throw new ContractValidationError('target source ledger is ahead of the imported generation'); + for (const record of records) { + const mutation = sanitizer.sanitize(JSON.parse(record.payload)); + const existing = (await exec.query( + 'SELECT source_epoch,fence_token::text,op_digest,mutation,head_prev,head_digest,head_key_id,head_alg,head_sig FROM tsk_outbox_rows WHERE stream_id=$1 AND sequence=$2 FOR UPDATE', + [streamId, record.sequence], + )).rows[0]; + if (existing) { + const exact = String(existing.source_epoch) === record.sourceEpoch && + String(existing.fence_token) === record.fenceToken && String(existing.op_digest) === record.opDigest && + canonicalize(existing.mutation) === canonicalize(mutation) && + String(existing.head_prev) === record.prevHeadDigest && String(existing.head_digest) === record.headDigest && + String(existing.head_key_id) === record.keyId && String(existing.head_alg) === record.alg && + String(existing.head_sig) === record.signature; + if (!exact) throw new ContractValidationError(`target source ledger conflicts at sequence ${record.sequence}`); + continue; + } + const hotp = mutation as TskHotpMutation; + affectedOne(await exec.query( + `INSERT INTO tsk_outbox_rows + (stream_id,source_epoch,sequence,fence_token,op_digest,tumbler_id,hotp_counter,mutation, + head_prev,head_digest,head_key_id,head_alg,head_sig,acked_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,now())`, + [streamId, record.sourceEpoch, record.sequence, record.fenceToken, record.opDigest, + hotp.tumblerId, hotp.counter, canonicalize(mutation), record.prevHeadDigest, + record.headDigest, record.keyId, record.alg, record.signature]), + 'imported source row insert'); + } + const fence = (await exec.query('SELECT fence_token FROM tsk_outbox_fence WHERE stream_id=$1 FOR UPDATE', [streamId])).rows[0]; + if (!fence) affectedOne(await exec.query('INSERT INTO tsk_outbox_fence(stream_id,fence_token) VALUES ($1,$2)', [streamId, targetEpoch]), 'target fence install'); + else { + const current = Number(fence.fence_token); + if (!Number.isSafeInteger(current) || current < 0 || current > targetEpoch) throw new ContractValidationError('target fence is ahead of the activation epoch'); + if (current < targetEpoch) affectedOne(await exec.query('UPDATE tsk_outbox_fence SET fence_token=$2 WHERE stream_id=$1 AND fence_token=$3', [streamId, targetEpoch, current]), 'target fence advance'); + } + const cp = (await exec.query('SELECT source_epoch,sequence,head_digest FROM tsk_outbox_source_checkpoint WHERE stream_id=$1 FOR UPDATE', [streamId])).rows[0]; + if (!cp) affectedOne(await exec.query( + 'INSERT INTO tsk_outbox_source_checkpoint(stream_id,source_epoch,sequence,head_digest) VALUES ($1,$2,$3,$4)', + [streamId, manifest.sourceEpoch, replay.n, replay.signedHeadDigestAtN]), 'target source checkpoint install'); + else { + if (Number(cp.sequence) > replay.n) throw new ContractValidationError('target source checkpoint is ahead of imported N'); + if (Number(cp.sequence) === replay.n) { + if (String(cp.source_epoch) !== manifest.sourceEpoch || String(cp.head_digest) !== replay.signedHeadDigestAtN) { + throw new ContractValidationError('target source checkpoint conflicts at imported N'); + } + } else { + affectedOne(await exec.query( + 'UPDATE tsk_outbox_source_checkpoint SET source_epoch=$2,sequence=$3,head_digest=$4 WHERE stream_id=$1 AND sequence=$5', + [streamId, manifest.sourceEpoch, replay.n, replay.signedHeadDigestAtN, Number(cp.sequence)]), 'target source checkpoint install'); + } + } + await installLeaseGrant(exec, leaseResolver, captured.activationLease); + return { streamId, generationId, targetEpoch, n: replay.n, + headDigest: replay.signedHeadDigestAtN, + activationGrantDigest: captured.activationLease.grantDigest }; + }); +} + export interface PgTskPublisherOptions { leaseMs: number; scopeDeadlineMs?: number } export interface TskDrainResult { published: number; acked: number; quarantined: number; retriable: boolean; halted: boolean } diff --git a/packages/server/src/tsk-source-fence.ts b/packages/server/src/tsk-source-fence.ts index 7daca19..fcdba08 100644 --- a/packages/server/src/tsk-source-fence.ts +++ b/packages/server/src/tsk-source-fence.ts @@ -719,7 +719,7 @@ async function sourceFenceManifest(exec: PgExecutor, tables: readonly string[], export const SOURCE_LEASE_MANIFEST_DIGEST = '548ad3d75a8d9dc612afe2f6e3734d5660fb79ce768a011d5a2de3cfca4c04f3'; export const SOURCE_WITNESS_MANIFEST_DIGEST = 'ec71067986cb82d06fc8fa3e4e9f934b4a961b0759a92199789de33c19ce0a82'; -async function attestSourceLease(exec: PgExecutor): Promise { +export async function attestSourceLease(exec: PgExecutor): Promise { const digest = sha256hex(Buffer.from(await sourceFenceManifest(exec, TSK_SOURCE_LEASE_TABLES, 'Vsource_lease/1'), 'utf8')); if (digest !== SOURCE_LEASE_MANIFEST_DIGEST) throw new ContractValidationError(`source lease attestation failed: live catalog digest ${digest} != pinned ${SOURCE_LEASE_MANIFEST_DIGEST}`); } diff --git a/packages/server/tests/ha-control-activation-continuity.test.ts b/packages/server/tests/ha-control-activation-continuity.test.ts new file mode 100644 index 0000000..0f8caeb --- /dev/null +++ b/packages/server/tests/ha-control-activation-continuity.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from 'vitest'; +import { assertConsumedActivationEpochs } from '../src/ha-control-fencing.js'; + +describe('source activation epoch continuity', () => { + it('accepts multiple consecutive signed terminal ABORTED epochs', () => { + expect(() => assertConsumedActivationEpochs(1, 4, [ + { epoch: 2, phase: 'PREPARING' }, { epoch: 2, phase: 'FENCED' }, { epoch: 2, phase: 'ABORTED' }, + { epoch: 3, phase: 'PREPARING' }, { epoch: 3, phase: 'READY' }, { epoch: 3, phase: 'ABORTED' }, + ], [])).not.toThrow(); + }); + + it.each([ + ['missing epoch', [{ epoch: 2, phase: 'ABORTED' }], []], + ['ACTIVE epoch', [{ epoch: 2, phase: 'ACTIVE' }, { epoch: 3, phase: 'ABORTED' }], []], + ['equivocal terminal epoch', [{ epoch: 2, phase: 'ABORTED' }, { epoch: 2, phase: 'ACTIVE' }, { epoch: 3, phase: 'ABORTED' }], []], + ['already activated epoch', [{ epoch: 2, phase: 'ABORTED' }, { epoch: 3, phase: 'ABORTED' }], [2]], + ])('fails closed for %s', (_name, rows, activated) => { + expect(() => assertConsumedActivationEpochs(1, 4, rows, activated)).toThrow(/quarantine/i); + }); + + it('rejects reuse and rollback', () => { + expect(() => assertConsumedActivationEpochs(2, 2, [], [])).toThrow(/does not advance/i); + expect(() => assertConsumedActivationEpochs(2, 1, [], [])).toThrow(/does not advance/i); + }); +}); diff --git a/tsk-b-activation-drill.mts b/tsk-b-activation-drill.mts index ac2dc76..92eb880 100644 --- a/tsk-b-activation-drill.mts +++ b/tsk-b-activation-drill.mts @@ -20,6 +20,7 @@ import { signLeaseGrant, installLeaseGrant, assertSourceFenceReady, emitSourceFrozenReceipt, buildSourceExportManifest, guardCountersignSourceExport, assertReceiverReady, stageAndFinalizeReceiverGeneration, verifyBFinalizedReceipt, + activateFinalizedReceiverAsSource, HA_CONTROL_PG_SCHEMA, HA_CONTROL_TABLES, HaControlFencing, GuardSigner, RedisFencingStore, provisionControlSchema, verifyLeaseGrant, type PgTransactor, type StreamHeadSigner, type HotpMutationSanitizer, type SanitizedMutation, type TskHotpMutation, type SourceVerifyKeyResolver, type SourceExportBundle, type GuardCountersignedExport, @@ -54,24 +55,6 @@ const sanitizer: HotpMutationSanitizer = { let passed = 0; async function check(name: string, fn: () => Promise) { await fn(); passed++; console.log(` ok - ${name}`); } -async function installVerifiedHistory(pool: pg.Pool, streamId: string, - bundle: SourceExportBundle, fromSequence = 1): Promise { - for (const record of bundle.historyChunks.flatMap((chunk) => chunk.records)) { - if (record.sequence < fromSequence) continue; - const mutation = JSON.parse(record.payload) as { tumblerId: string; counter: number }; - await pool.query( - `INSERT INTO tsk_outbox_rows - (stream_id,source_epoch,sequence,fence_token,op_digest,tumbler_id,hotp_counter, - mutation,head_prev,head_digest,head_key_id,head_alg,head_sig,acked_at) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,now())`, - [streamId, record.sourceEpoch, record.sequence, record.fenceToken, - record.opDigest, mutation.tumblerId, mutation.counter, mutation, - record.prevHeadDigest, record.headDigest, record.keyId, record.alg, - record.signature], - ); - } -} - async function main() { console.log('# TSK PR2c governed B source-authority activation (B originates N+1; old A denied)'); const aPool = new pg.Pool({ connectionString: A_URL, max: 4 }); aPool.on('error', () => {}); @@ -111,36 +94,34 @@ async function main() { const nowA = async () => Number((await aPool.query('SELECT (extract(epoch from clock_timestamp())*1000)::bigint AS ms')).rows[0].ms); await aPool.query('INSERT INTO tsk_outbox_fence (stream_id, fence_token) VALUES ($1,0)', [SID]); await aPool.query('INSERT INTO tsk_outbox_source_checkpoint (stream_id, source_epoch, sequence) VALUES ($1,$2,0)', [SID, SRC_EPOCH]); - const g = signLeaseGrant(GUARD_KEY, guard.privateKey, { streamId: SID, leaseEpoch: 0, leaseStatus: 'active', holderNodeId: 'A', leaseId: 'l1', commandId: 'grant-1', leaseExpiresAtMs: (await nowA()) + HOUR, leaseGrantSeq: 1, prevGrantDigest: null }); + const g = signLeaseGrant(GUARD_KEY, guard.privateKey, { streamId: SID, leaseEpoch: 0, leaseStatus: 'active', holderNodeId: A_KEY, leaseId: 'l1', commandId: 'grant-1', leaseExpiresAtMs: (await nowA()) + HOUR, leaseGrantSeq: 1, prevGrantDigest: null }); await aTx.transaction((exec) => installLeaseGrant(exec, resolver, g)); - const aReady = await assertSourceFenceReady(aTx, SCHEMA, resolver, { streamId: SID, holderNodeId: 'A', leaseId: 'l1', grantDigest: g.grantDigest }); + const aReady = await assertSourceFenceReady(aTx, SCHEMA, resolver, { streamId: SID, holderNodeId: A_KEY, leaseId: 'l1', grantDigest: g.grantDigest }); const aOb = new PgTskDurableOutbox(aTx, READY, { streamId: SID, sanitizer, signer, maxPendingRows: 100_000, backpressure: 'fail-authoritative-mutation' }, { resolver, controlToASkewBoundMs: 0, ready: aReady }); for (const [t, c] of [['T1', 1], ['T2', 5], ['T1', 2], ['T3', 9]] as [string, number][]) await aOb.withOutboxTx((wtx) => aOb.appendInTx(wtx, { streamId: SID, rawMutation: { tumblerId: t, counter: c }, fenceToken: 0n })); const N = 4; - const rev = signLeaseGrant(GUARD_KEY, guard.privateKey, { streamId: SID, leaseEpoch: 0, leaseStatus: 'revoked', holderNodeId: 'A', leaseId: 'l1', commandId: CMD, leaseExpiresAtMs: (await nowA()) + HOUR, leaseGrantSeq: 2, prevGrantDigest: g.grantDigest }); + const rev = signLeaseGrant(GUARD_KEY, guard.privateKey, { streamId: SID, leaseEpoch: 0, leaseStatus: 'revoked', holderNodeId: A_KEY, leaseId: 'l1', commandId: CMD, leaseExpiresAtMs: (await nowA()) + HOUR, leaseGrantSeq: 2, prevGrantDigest: g.grantDigest }); await aTx.transaction((exec) => installLeaseGrant(exec, resolver, rev)); - const frozen = await emitSourceFrozenReceipt(aTx, SCHEMA, { sourceKeyId: SOURCE_KEY, sourcePrivateKey: source.privateKey, leaseResolver: resolver, headResolver: resolver }, { streamId: SID, commandId: CMD, epoch: 0, sourceNodeId: 'A' }); + const frozen = await emitSourceFrozenReceipt(aTx, SCHEMA, { sourceKeyId: SOURCE_KEY, sourcePrivateKey: source.privateKey, leaseResolver: resolver, headResolver: resolver }, { streamId: SID, commandId: CMD, epoch: 0, sourceNodeId: A_KEY }); assert.equal(frozen.n, N); - const built = await buildSourceExportManifest(aTx, SCHEMA, { streamId: SID, epoch: 0, commandId: CMD, sourceNodeId: 'A' }, { sourceKeyId: SOURCE_KEY, sourcePrivateKey: source.privateKey, sanitizer, leaseResolver: resolver, headResolver: resolver, frozenReceipt: frozen, maxChunkItems: 4 }); + const built = await buildSourceExportManifest(aTx, SCHEMA, { streamId: SID, epoch: 0, commandId: CMD, sourceNodeId: A_KEY }, { sourceKeyId: SOURCE_KEY, sourcePrivateKey: source.privateKey, sanitizer, leaseResolver: resolver, headResolver: resolver, frozenReceipt: frozen, maxChunkItems: 4 }); const bundle: SourceExportBundle = built.bundle; const dual: GuardCountersignedExport = guardCountersignSourceExport(bundle, built.manifest, { guardKeyId: GUARD_KEY, guardPrivateKey: guard.privateKey, sanitizer, sourceManifestResolver: resolver, headResolver: resolver, frozenResolver: resolver, frozenReceipt: frozen, expectedCommandId: CMD }); const ropts = { sanitizer, sourceResolver: resolver, guardResolver: resolver, headResolver: resolver, frozenResolver: resolver, bVerifyResolver: resolver, frozenReceipt: frozen, expectedCommandId: CMD, bKeyId: B_KEY, bPrivateKey: bKp.privateKey }; - const bReceipt = await stageAndFinalizeReceiverGeneration(bTx, SCHEMA, await assertReceiverReady(bTx, SCHEMA), 'gen-1', bundle, dual, ropts); + const bReceiverReady = await assertReceiverReady(bTx, SCHEMA); + const bReceipt = await stageAndFinalizeReceiverGeneration(bTx, SCHEMA, bReceiverReady, 'gen-1', bundle, dual, ropts); verifyBFinalizedReceipt(resolver, bReceipt); const headAtN = bReceipt.signedHeadDigestAtN; - // The independently replayed receiver generation is now materialized as B's source ledger. - // These exact signed rows are what the next freeze/export replays; no history is synthesized. - await installVerifiedHistory(bPool, SID, bundle); // ── control drives the cutover to ACTIVE ── const ctlReady = await provisionControlSchema(cTx as never, SCHEMA); const ctl = new HaControlFencing(cTx as never, new GuardSigner(CTRL_KEY, ctrlSecret), ctrlResolver, ctlReady, POLICY); const ctlNow = async () => Number((await cPool.query('SELECT (extract(epoch from clock_timestamp())*1000)::bigint AS ms')).rows[0].ms); await ctl.provision(SID, 'g-bact'); - await ctl.writeLease({ streamId: SID, leaseId: 'l1', holderNodeId: 'A', epoch: 0, status: 'active', grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'a1' }); + await ctl.writeLease({ streamId: SID, leaseId: 'l1', holderNodeId: A_KEY, epoch: 0, status: 'active', grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'a1' }); await ctl.beginPromotionIntent(SID, CMD, TARGET); await ctl.bindSourceFenced(SID, CMD, TARGET, frozen, resolver); - await ctl.writeLease({ streamId: SID, leaseId: 'l1', holderNodeId: 'A', epoch: 0, status: 'revoked', grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'a2' }); + await ctl.writeLease({ streamId: SID, leaseId: 'l1', holderNodeId: A_KEY, epoch: 0, status: 'revoked', grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'a2' }); const store = redis ? new RedisFencingStore(redis, 'tsk:fence:' + SID) : new MemoryFencingStore(); const proof: FenceProof = { safetyMarginMs: 0, claimExpiresAtMs: (await ctlNow()) + HOUR }; await ctl.advanceEpoch(SID, CMD, TARGET, 'Bnode', store, proof); @@ -151,22 +132,79 @@ async function main() { let bGrant: Awaited>; let bOb: InstanceType; await check('activateSource mints a GUARD-signed governed B lease at the PROMOTED epoch via the CONFIGURED authority, holder==ratified B, durable + rehydratable', async () => { - bGrant = await ctl.activateSource(SID, CMD, TARGET, bReceipt, resolver); // no per-call key; guard from config + bGrant = await ctl.activateSource(SID, CMD, TARGET, bReceipt, resolver, resolver); // no per-call key; guard from config verifyLeaseGrant(resolver, bGrant); // B independently verifies with the guard PUBLIC key (unforgeable) assert.equal(bGrant.leaseEpoch, TARGET); assert.equal(bGrant.holderNodeId, B_KEY, 'holder is the ratified B signing identity'); assert.equal(bGrant.leaseStatus, 'active'); assert.equal(bGrant.commandId, CMD); // (recovery) a RETRY rehydrates the byte-identical grant — never a different or conflicting one. - const again = await ctl.activateSource(SID, CMD, TARGET, bReceipt, resolver); + const again = await ctl.activateSource(SID, CMD, TARGET, bReceipt, resolver, resolver); assert.equal(again.grantDigest, bGrant.grantDigest); assert.deepEqual(again, bGrant); // a receipt that was NOT ratified into the ACTIVE head is refused. const bogus = { ...bReceipt, n: 999 }; - await assert.rejects(() => ctl.activateSource(SID, CMD, TARGET, bogus as typeof bReceipt, resolver), /verif|not the one ratified|digest mismatch/i); + await assert.rejects(() => ctl.activateSource(SID, CMD, TARGET, bogus as typeof bReceipt, resolver, resolver), /verif|not the one ratified|digest mismatch/i); }); await check('B installs the governed lease, seeds its checkpoint to (N, head@N), and ORIGINATES sequence N+1 through the fenced outbox', async () => { - // B becomes the source: fence epoch = the promoted epoch; checkpoint continues from the verified (N, head@N). - await bPool.query('INSERT INTO tsk_outbox_fence (stream_id, fence_token) VALUES ($1,$2)', [SID, TARGET]); - await bPool.query('INSERT INTO tsk_outbox_source_checkpoint (stream_id, source_epoch, sequence, head_digest) VALUES ($1,$2,$3,$4)', [SID, SRC_EPOCH, N, headAtN]); - await bTx.transaction((exec) => installLeaseGrant(exec, resolver, bGrant)); + let failBeforeCommit = false; + const crashTx: PgTransactor = { + transaction: (fn, opts) => bTx.transaction(fn, { + ...opts, + onBeforeCommit: async () => { if (failBeforeCommit) throw new Error('simulated crash before commit'); }, + }), + }; + const crashOutboxReady = await provisionSchemaVersion(crashTx, SCHEMA); + const crashReceiverReady = await assertReceiverReady(crashTx, SCHEMA); + failBeforeCommit = true; + await assert.rejects(() => activateFinalizedReceiverAsSource( + crashTx, crashOutboxReady, crashReceiverReady, SID, 'gen-1', { + sanitizer, sourceResolver: resolver, guardResolver: resolver, + headResolver: resolver, frozenResolver: resolver, + bReceiptResolver: resolver, leaseResolver: resolver, + frozenReceipt: frozen, finalizedReceipt: bReceipt, + activationLease: bGrant, targetEpoch: TARGET, + }, + ), /simulated crash before commit/); + const partial = await Promise.all([ + bPool.query('SELECT count(*)::int AS n FROM tsk_outbox_rows WHERE stream_id=$1', [SID]), + bPool.query('SELECT count(*)::int AS n FROM tsk_outbox_fence WHERE stream_id=$1', [SID]), + bPool.query('SELECT count(*)::int AS n FROM tsk_outbox_source_checkpoint WHERE stream_id=$1', [SID]), + bPool.query('SELECT count(*)::int AS n FROM tsk_source_lease WHERE stream_id=$1', [SID]), + ]); + assert.deepEqual(partial.map((r) => Number(r.rows[0].n)), [0, 0, 0, 0], 'failed activation exposed no partial source authority'); + const bOutboxReady = await provisionSchemaVersion(bTx, SCHEMA); + // ONE owned SERIALIZABLE authority imports the persisted, independently-verified generation, + // advances fence/checkpoint, and installs the lease. A crash exposes all or none. + const imported = await activateFinalizedReceiverAsSource( + bTx, bOutboxReady, bReceiverReady, SID, 'gen-1', { + sanitizer, sourceResolver: resolver, guardResolver: resolver, + headResolver: resolver, frozenResolver: resolver, + bReceiptResolver: resolver, leaseResolver: resolver, + frozenReceipt: frozen, finalizedReceipt: bReceipt, + activationLease: bGrant, targetEpoch: TARGET, + }, + ); + assert.equal(imported.n, N); assert.equal(imported.headDigest, headAtN); + // Lost reply/restart is byte-identically idempotent and cannot duplicate the lease or ledger. + const importedAgain = await activateFinalizedReceiverAsSource( + bTx, bOutboxReady, bReceiverReady, SID, 'gen-1', { + sanitizer, sourceResolver: resolver, guardResolver: resolver, + headResolver: resolver, frozenResolver: resolver, + bReceiptResolver: resolver, leaseResolver: resolver, + frozenReceipt: frozen, finalizedReceipt: bReceipt, + activationLease: bGrant, targetEpoch: TARGET, + }, + ); + assert.deepEqual(importedAgain, imported); + const accessorReceipt = { ...bReceipt }; + Object.defineProperty(accessorReceipt, 'commandId', { enumerable: true, get: () => CMD }); + await assert.rejects(() => activateFinalizedReceiverAsSource( + bTx, bOutboxReady, bReceiverReady, SID, 'gen-1', { + sanitizer, sourceResolver: resolver, guardResolver: resolver, + headResolver: resolver, frozenResolver: resolver, + bReceiptResolver: resolver, leaseResolver: resolver, + frozenReceipt: frozen, finalizedReceipt: accessorReceipt, + activationLease: bGrant, targetEpoch: TARGET, + }, + ), /contains an accessor/); const bReadySrc = await assertSourceFenceReady(bTx, SCHEMA, resolver, { streamId: SID, holderNodeId: bGrant.holderNodeId, leaseId: bGrant.leaseId, grantDigest: bGrant.grantDigest }); const bREADY = await provisionSchemaVersion(bTx, SCHEMA); bOb = new PgTskDurableOutbox(bTx, bREADY, { streamId: SID, sanitizer, signer: bSigner, maxPendingRows: 100_000, backpressure: 'fail-authoritative-mutation' }, { resolver, controlToASkewBoundMs: 0, ready: bReadySrc }); @@ -185,11 +223,12 @@ async function main() { }); // ── governed B -> A return failback on the SAME stream ── - const RETURN_CMD = 'return-2'; const RETURN_TARGET = 2; + const ABORT_CMD = 'aborted-handoff-2'; + const RETURN_CMD = 'return-3'; const RETURN_TARGET = 3; let bRev: ReturnType; let returnedGrant: Awaited>; let returnReceipt: Awaited>; - await check('B freezes N+1 and A independently replays the complete signed ledger for return failback', async () => { + await check('after a signed terminal ABORTED epoch, B freezes N+1 and A independently replays the complete signed ledger for return failback', async () => { bRev = signLeaseGrant(GUARD_KEY, guard.privateKey, { streamId: SID, leaseEpoch: TARGET, leaseStatus: 'revoked', holderNodeId: bGrant.holderNodeId, leaseId: bGrant.leaseId, @@ -216,8 +255,9 @@ async function main() { expectedCommandId: RETURN_CMD, }); for (const s of TSK_RECEIVER_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await aPool.query(s); + const aReceiverReady = await assertReceiverReady(aTx, SCHEMA); returnReceipt = await stageAndFinalizeReceiverGeneration( - aTx, SCHEMA, await assertReceiverReady(aTx, SCHEMA), 'gen-return-2', + aTx, SCHEMA, aReceiverReady, 'gen-return-2', builtB.bundle, dualB, { sanitizer, sourceResolver: resolver, guardResolver: resolver, headResolver: resolver, frozenResolver: resolver, bVerifyResolver: resolver, @@ -228,21 +268,39 @@ async function main() { verifyBFinalizedReceipt(resolver, returnReceipt); assert.equal(returnReceipt.n, N + 1); + const frozenAbort = await emitSourceFrozenReceipt(bTx, SCHEMA, { + sourceKeyId: B_SOURCE_KEY, sourcePrivateKey: bSource.privateKey, + leaseResolver: resolver, headResolver: resolver, + }, { streamId: SID, commandId: ABORT_CMD, epoch: TARGET, sourceNodeId: B_KEY }); await ctl.writeLease({ streamId: SID, leaseId: bGrant.leaseId, holderNodeId: bGrant.holderNodeId, epoch: TARGET, status: 'active', grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'b-active-2' }); - await ctl.beginPromotionIntent(SID, RETURN_CMD, RETURN_TARGET); - await ctl.bindSourceFenced(SID, RETURN_CMD, RETURN_TARGET, frozenB, resolver); + await ctl.beginPromotionIntent(SID, ABORT_CMD, 2); + await ctl.bindSourceFenced(SID, ABORT_CMD, 2, frozenAbort, resolver); await ctl.writeLease({ streamId: SID, leaseId: bGrant.leaseId, holderNodeId: bGrant.holderNodeId, epoch: TARGET, status: 'revoked', grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'b-revoke-2' }); + await ctl.advanceEpoch(SID, ABORT_CMD, 2, 'unused-node', store, + { safetyMarginMs: 0, claimExpiresAtMs: (await ctlNow()) + HOUR }); + await ctl.abortFencedPromotion(SID, ABORT_CMD, 2, 'target-unavailable'); + + // Epoch 2 is consumed but never receives source authority. Control records a terminal + // no-writer lease state, then the later promotion proves the signed ABORTED gap. + await ctl.writeLease({ streamId: SID, leaseId: bGrant.leaseId, + holderNodeId: bGrant.holderNodeId, epoch: 2, status: 'active', + grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'b-empty-active-3' }); + await ctl.writeLease({ streamId: SID, leaseId: bGrant.leaseId, + holderNodeId: bGrant.holderNodeId, epoch: 2, status: 'revoked', + grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'b-empty-revoke-3' }); + await ctl.beginPromotionIntent(SID, RETURN_CMD, RETURN_TARGET); + await ctl.bindSourceFenced(SID, RETURN_CMD, RETURN_TARGET, frozenB, resolver); await ctl.advanceEpoch(SID, RETURN_CMD, RETURN_TARGET, 'Anode', store, { safetyMarginMs: 0, claimExpiresAtMs: (await ctlNow()) + HOUR }); await ctl.markImporting(SID, RETURN_CMD, RETURN_TARGET); await ctl.markReady(SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver); await ctl.activate(SID, RETURN_CMD, RETURN_TARGET); returnedGrant = await ctl.activateSource( - SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, rev, + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, resolver, [g, rev], ); verifyLeaseGrant(resolver, returnedGrant); assert.equal(returnedGrant.holderNodeId, A_KEY); @@ -250,14 +308,17 @@ async function main() { assert.equal(returnedGrant.leaseGrantSeq, 3); assert.equal(returnedGrant.prevGrantDigest, rev.grantDigest); - // A already owns 1..N. Import only B's independently verified signed N+1 row. - await installVerifiedHistory(aPool, SID, builtB.bundle, N + 1); - await aPool.query('UPDATE tsk_outbox_fence SET fence_token=$2 WHERE stream_id=$1', [SID, RETURN_TARGET]); - await aPool.query( - 'UPDATE tsk_outbox_source_checkpoint SET sequence=$2,head_digest=$3 WHERE stream_id=$1', - [SID, N + 1, returnReceipt.signedHeadDigestAtN], + const returned = await activateFinalizedReceiverAsSource( + aTx, await provisionSchemaVersion(aTx, SCHEMA), aReceiverReady, SID, 'gen-return-2', { + sanitizer, sourceResolver: resolver, guardResolver: resolver, + headResolver: resolver, frozenResolver: resolver, + bReceiptResolver: resolver, leaseResolver: resolver, + frozenReceipt: frozenB, finalizedReceipt: returnReceipt, + activationLease: returnedGrant, targetEpoch: RETURN_TARGET, + }, ); - await aTx.transaction((exec) => installLeaseGrant(exec, resolver, returnedGrant)); + assert.equal(returned.n, N + 1); + assert.equal(returned.headDigest, returnReceipt.signedHeadDigestAtN); }); await check('returned A originates N+2 while old B and replayed activation inputs fail closed', async () => { @@ -279,13 +340,23 @@ async function main() { streamId: SID, rawMutation: { tumblerId: 'T9', counter: 2 }, fenceToken: 1n, })), /revoked|not writable|lease|fence/i); await assert.rejects(() => ctl.activateSource( - SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, g, - ), /prior target lease must be terminally revoked|does not byte-bind/i); + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, resolver, [g], + ), /high-water must be revoked|does not byte-bind/i); + const extraChain = [g, rev] as (typeof g)[] & { extra?: string }; + extraChain.extra = 'not-authority'; + await assert.rejects(() => ctl.activateSource( + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, resolver, extraChain, + ), /dense exact array|data elements/i); + const accessorChain = [g, rev]; + Object.defineProperty(accessorChain, '0', { enumerable: true, configurable: true, get: () => g }); await assert.rejects(() => ctl.activateSource( - SID, CMD, TARGET, bReceipt, resolver, + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, resolver, accessorChain, + ), /dense exact array|data elements/i); + await assert.rejects(() => ctl.activateSource( + SID, CMD, TARGET, bReceipt, resolver, resolver, ), /epoch|activation|quarantine/i); const retry = await ctl.activateSource( - SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, rev, + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, resolver, [g, rev], ); assert.deepEqual(retry, returnedGrant); const activationCols = 'command_id,epoch,b_key_id,b_receipt_digest,activation_seq,' + @@ -297,13 +368,28 @@ async function main() { WHERE stream_id=$1 AND activation_seq=1) WHERE h.stream_id=$1`, [SID], ); await assert.rejects(() => ctl.activateSource( - SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, rev, + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, resolver, [g, rev], ), /head is not the latest|replay|rollback/i); await cPool.query( `UPDATE tsk_ha_source_activation h SET (${activationCols}) = (SELECT ${activationCols} FROM tsk_ha_source_activation_history WHERE stream_id=$1 AND activation_seq=2) WHERE h.stream_id=$1`, [SID], ); + const savedGrantJson = String((await cPool.query( + 'SELECT grant_json FROM tsk_ha_source_activation_history WHERE stream_id=$1 AND activation_seq=1', + [SID], + )).rows[0].grant_json); + await cPool.query( + "UPDATE tsk_ha_source_activation_history SET grant_json='{}' WHERE stream_id=$1 AND activation_seq=1", + [SID], + ); + await assert.rejects(() => ctl.activateSource( + SID, RETURN_CMD, RETURN_TARGET, returnReceipt, resolver, resolver, [g, rev], + ), /activation grant_json|lease grant/i); + await cPool.query( + 'UPDATE tsk_ha_source_activation_history SET grant_json=$2 WHERE stream_id=$1 AND activation_seq=1', + [SID, savedGrantJson], + ); const activationRows = Number((await cPool.query( 'SELECT count(*) AS n FROM tsk_ha_source_activation_history WHERE stream_id=$1', [SID], )).rows[0].n); diff --git a/tsk-control-cutover-drill.mts b/tsk-control-cutover-drill.mts index 886ecf5..bc22958 100644 --- a/tsk-control-cutover-drill.mts +++ b/tsk-control-cutover-drill.mts @@ -236,14 +236,14 @@ async function main() { }); // ── negative: a freeze at the WRONG source epoch cannot bind (cross-epoch splice) ─ - await check('bindSourceFenced REJECTS a freeze whose epoch != targetEpoch-1 (foreign-epoch freeze)', async () => { + await check('bindSourceFenced REJECTS a freeze outside the current signed source activation epoch', async () => { const SIDE = 'tsk:pair:epoch/v1'; const frE1 = await genesisFreezeAtEpoch(SIDE, 'promote-epoch', 1); // frozen at source epoch 1 await ctl.provision(SIDE, 'g-epoch'); const past = (await ctrlNowMs()) - 5_000; await ctl.writeLease({ streamId: SIDE, leaseId: 'l1', holderNodeId: 'A', epoch: 0, status: 'active', grantedMaxExpiryMs: past, grantCommandId: 'a-epoch' }); await ctl.beginPromotionIntent(SIDE, 'promote-epoch', 1); // target 1 → target-1 = 0, but the freeze is epoch 1 - await assert.rejects(() => ctl.bindSourceFenced(SIDE, 'promote-epoch', 1, frE1, resolver), /epoch 1 != targetEpoch-1|cross-epoch/); + await assert.rejects(() => ctl.bindSourceFenced(SIDE, 'promote-epoch', 1, frE1, resolver), /current signed source activation epoch|cross-epoch/); }); // ── negative: the ORDERING gate is enforced AT advanceEpoch (single chokepoint) ─ diff --git a/tsk-ha-control-migration-drill.mts b/tsk-ha-control-migration-drill.mts new file mode 100644 index 0000000..e78f89e --- /dev/null +++ b/tsk-ha-control-migration-drill.mts @@ -0,0 +1,90 @@ +import assert from 'node:assert/strict'; +import { createHash, generateKeyPairSync } from 'node:crypto'; +import pg from 'pg'; +import { + HA_CONTROL_PG_SCHEMA, HA_CONTROL_TABLES, HA_CONTROL_V1_SOURCE_ACTIVATION_SCHEMA, + HA_CONTROL_V1_MANIFEST_DIGEST, HA_CONTROL_MANIFEST_DIGEST, + GuardSigner, NodePostgresTransactor, migrateControlSchemaV1ToV2, + assertControlSchemaReady, signLeaseGrant, + type GuardKeyResolver, type SourceVerifyKeyResolver, type ControlSchemaMigrationDb, +} from './packages/server/dist/index.js'; + +const URL = process.env['TSK_TEST_CONTROL_PG_URL']; +if (!URL) throw new Error('TSK_TEST_CONTROL_PG_URL is required (real PostgreSQL 16; no skip)'); +const SCHEMA = 'public'; const STREAM = 'migration-stream'; const CMD = 'activate-1'; +const CONTROL_KEY = 'control-1'; const CONTROL_SECRET = Buffer.alloc(32, 0x4d); +const SOURCE_GUARD_KEY = 'source-guard-1'; const sourceGuard = generateKeyPairSync('ed25519'); +const signer = new GuardSigner(CONTROL_KEY, CONTROL_SECRET); +const controlResolver: GuardKeyResolver = { resolve: (keyId: string) => keyId === CONTROL_KEY ? CONTROL_SECRET : null }; +const sourceResolver: SourceVerifyKeyResolver = { resolve: (keyId: string) => keyId === SOURCE_GUARD_KEY ? sourceGuard.publicKey : null }; + +const frame = (...parts: (string | number | null)[]): Buffer => Buffer.concat(parts.flatMap((part) => { + if (part === null) return [Buffer.from([0])]; + const value = Buffer.from(String(part)); const length = Buffer.alloc(4); length.writeUInt32BE(value.length); + return [Buffer.from([1]), length, value]; +})); +const digest = (value: Buffer): string => createHash('sha256').update(value).digest('hex'); +const cutMsg = (stream: string, epoch: number, command: string, seq: number, phase: string, + evidence: string | null, prev: string | null, stateDigest: string): Buffer => + frame('tsk_ha_cutover/v1', stream, epoch, command, seq, phase, evidence, prev, stateDigest); + +async function installV1(pool: pg.Pool, tamperGrant = false): Promise { + await pool.query(`DROP TABLE IF EXISTS ${HA_CONTROL_TABLES.join(', ')} CASCADE`); + for (const statement of HA_CONTROL_PG_SCHEMA.split(';').map((s) => s.trim()).filter(Boolean)) { + if (!/^CREATE TABLE IF NOT EXISTS tsk_ha_source_activation(?:_history)?\s*\(/.test(statement)) await pool.query(statement); + } + await pool.query(HA_CONTROL_V1_SOURCE_ACTIVATION_SCHEMA); + await pool.query('INSERT INTO tsk_ha_schema(id,version,catalog_manifest) VALUES(1,1,$1)', [HA_CONTROL_V1_MANIFEST_DIGEST]); + + const receiptDigest = 'a'.repeat(64); const holder = 'node-b'; + const activeEvidence = JSON.stringify({ k: 'active/v1', bReceiptDigest: receiptDigest, bKeyId: holder }); + const stateDigest = digest(cutMsg(STREAM, 1, CMD, 1, 'ACTIVE', activeEvidence, null, '')); + const signature = signer.sign(cutMsg(STREAM, 1, CMD, 1, 'ACTIVE', activeEvidence, null, stateDigest)); + const cutValues = [STREAM, 1, CMD, 1, 'ACTIVE', activeEvidence, null, stateDigest, CONTROL_KEY, signature]; + await pool.query('INSERT INTO tsk_ha_cutover_history(stream_id,epoch,command_id,seqno,phase,evidence,prev_state_digest,state_digest,guard_key_id,guard_signature) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)', cutValues); + await pool.query('INSERT INTO tsk_ha_cutover_head(stream_id,epoch,command_id,seqno,phase,evidence,prev_state_digest,state_digest,guard_key_id,guard_signature) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)', cutValues); + + const grant = signLeaseGrant(SOURCE_GUARD_KEY, sourceGuard.privateKey, { + streamId: STREAM, leaseEpoch: 1, leaseStatus: 'active', holderNodeId: holder, + leaseId: 'bsrc-activate-1', commandId: CMD, leaseExpiresAtMs: Date.now() + 60_000, + leaseGrantSeq: 1, prevGrantDigest: null, + }); + await pool.query('INSERT INTO tsk_ha_source_activation(stream_id,command_id,epoch,b_key_id,b_receipt_digest,grant_digest,grant_json,guard_key_id) VALUES($1,$2,$3,$4,$5,$6,$7,$8)', + [STREAM, CMD, 1, holder, receiptDigest, tamperGrant ? 'b'.repeat(64) : grant.grantDigest, JSON.stringify(grant), SOURCE_GUARD_KEY]); +} + +async function main(): Promise { + const pool = new pg.Pool({ connectionString: URL, max: 3 }); pool.on('error', () => {}); + const tx = new NodePostgresTransactor(pool as never); + const migrationDb: ControlSchemaMigrationDb = { async transaction(fn: (exec: { query(sql: string, params?: unknown[]): Promise<{ rows: Record[]; rowCount: number }> }) => Promise): Promise { + const client = await pool.connect(); + try { + await client.query('BEGIN ISOLATION LEVEL SERIALIZABLE'); + const result = await fn({ query: async (sql, params = []) => { + const value = await client.query(sql, params); return { rows: value.rows, rowCount: value.rowCount ?? 0 }; + } }); + await client.query('COMMIT'); return result; + } catch (error) { await client.query('ROLLBACK').catch(() => {}); throw error; } + finally { client.release(); } + } }; + await installV1(pool); + const migratedDigest = await migrateControlSchemaV1ToV2(migrationDb, SCHEMA, signer, controlResolver, sourceResolver); + await assertControlSchemaReady(tx, SCHEMA); + assert.equal(migratedDigest, HA_CONTROL_MANIFEST_DIGEST); + const stamp = (await pool.query('SELECT version,catalog_manifest FROM tsk_ha_schema WHERE id=1')).rows[0]; + assert.equal(Number(stamp.version), 2); assert.equal(stamp.catalog_manifest, HA_CONTROL_MANIFEST_DIGEST); + const migrated = (await pool.query('SELECT * FROM tsk_ha_source_activation_history WHERE stream_id=$1', [STREAM])).rows; + assert.equal(migrated.length, 1); assert.equal(Number(migrated[0].activation_seq), 1); + assert.equal(migrated[0].prev_activation_digest, null); + assert.equal(migrated[0].grant_digest, JSON.parse(migrated[0].grant_json).grantDigest); + console.log(' ok - exact v1 authority + signed legacy activation migrate atomically to attested v2'); + + await installV1(pool, true); + await assert.rejects(() => migrateControlSchemaV1ToV2(migrationDb, SCHEMA, signer, controlResolver, sourceResolver), /exactly bind|quarantine/i); + const rolledBack = (await pool.query('SELECT version,catalog_manifest FROM tsk_ha_schema WHERE id=1')).rows[0]; + assert.equal(Number(rolledBack.version), 1); assert.equal(rolledBack.catalog_manifest, HA_CONTROL_V1_MANIFEST_DIGEST); + assert.equal((await pool.query("SELECT pg_catalog.to_regclass('tsk_ha_source_activation_history') AS r")).rows[0].r, null); + console.log(' ok - invalid legacy authority fails closed and the one-transaction migration rolls back'); + await pool.end(); +} +main().catch((error) => { console.error(error); process.exitCode = 1; });