diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..e896d34 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,2 @@ +*.sh text eol=lf +ci/redis-sentinel/docker-compose.yml text eol=lf diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5c54611..9719a82 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -161,6 +161,36 @@ 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 + # (PR2c acceptance) child-process SIGKILL/restart matrix over the FULL control cutover + # (bind->advance->import->ready->activate): before-commit SIGKILL rolls back (no torn state), + # after-commit SIGKILL resumes idempotently to ACTIVE. Three INDEPENDENT PG16 + real Redis. + # Real OS-level process kills. THROWS if any URL is unset. Contributes to closing #10. + - run: npm run test:sigkill + env: + TSK_TEST_SOURCE_PG_URL_A: postgresql://tsk_test:tsk-test-only-password@127.0.0.1:5432/tsk_test + 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 + # (PR2c acceptance) GOVERNED B source-authority activation: after ACTIVE, control.activateSource() + # mints B's unforgeable guard-signed lease at the promoted epoch; B installs it, seeds (N, head@N), + # and ORIGINATES sequence N+1 through the fenced outbox while the OLD A (revoked, prior epoch) is + # DENIED. Three INDEPENDENT PG16 + real Redis. Contributes to closing #10. + - run: npm run test:b-activation + env: + TSK_TEST_SOURCE_PG_URL_A: postgresql://tsk_test:tsk-test-only-password@127.0.0.1:5432/tsk_test + 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 + # (PR2c acceptance) REAL Redis Sentinel/quorum: 1 master + 2 REPLICAS + 3 sentinels (quorum 2), + # AOF, min-replicas-to-write. A real master CRASH → automatic promotion; claim() ENFORCES a WAIT + # replica-quorum ACK so the fence SURVIVES (RPO=0), stays monotonic, RTO measured. run.sh brings up + # the topology (docker compose) and runs the drill on the host (ioredis natMap over published ports). + - run: npm run test:sentinel + # (PR2c acceptance) LIVE split-brain PARTITION: the old master is isolated ALIVE (docker network + # disconnect); it REFUSES writes (min-replicas-to-write → NOREPLICAS, no split-brain progress) while + # the surviving quorum promotes a new writable+monotonic master (fence intact, RPO=0); on HEAL the old + # master is demoted to a replica and reconciles. Per-fault RPO/RTO. Closes the Redis-HA gate for #10. + - run: npm run test:partition - run: npm run test:pack - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 with: diff --git a/.gitignore b/.gitignore index bddd2b8..72654e6 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,6 @@ debug-e2e.mts # FileTumblerStore persistent data — lives in ~/.tsk, never in repo demo/tsk-maps.json + +# PR2c SIGKILL drill handoff (test key material — never commit) +tsk-sigkill-handoff.json diff --git a/ci/redis-sentinel/docker-compose.yml b/ci/redis-sentinel/docker-compose.yml new file mode 100644 index 0000000..50f52dd --- /dev/null +++ b/ci/redis-sentinel/docker-compose.yml @@ -0,0 +1,78 @@ +# Real Redis Sentinel/quorum topology for the PR2c acceptance drills (#10): +# 1 master + 2 REPLICAS (3 data nodes) + 3 SENTINELS (quorum 2), AOF everysec. +# min-replicas-to-write 1 + min-replicas-max-lag → a master that loses its replica quorum REFUSES +# writes (the split-brain guard). STATIC IPs so sentinels monitor by fixed IP (survives a crashed +# master container exiting) AND PUBLISHED ports so the host drills reach every node directly; the +# sentinel-backed client uses ioredis natMap (internal IP:6379 → 127.0.0.1:published) to auto-follow +# failover from the host. Mechanism-only on one host (processes, not physical failure domains). +networks: + tsknet: + driver: bridge + ipam: + config: + - subnet: 172.28.7.0/24 +x-redis-data: &redis-data + image: redis:7.4.5-alpine@sha256:bb186d083732f669da90be8b0f975a37812b15e913465bb14d845db72a4e3e08 + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 1s + timeout: 2s + retries: 40 +services: + redis-master: + <<: *redis-data + command: >- + redis-server --appendonly yes --appendfsync everysec --save "" + --repl-diskless-sync yes --min-replicas-to-write 1 --min-replicas-max-lag 10 + restart: "no" + ports: ["6390:6379"] + networks: + tsknet: { ipv4_address: 172.28.7.10 } + redis-replica-1: + <<: *redis-data + command: >- + redis-server --appendonly yes --appendfsync everysec --save "" + --replicaof 172.28.7.10 6379 --replica-read-only yes --min-replicas-to-write 1 --min-replicas-max-lag 10 + depends_on: + redis-master: { condition: service_healthy } + ports: ["6391:6379"] + networks: + tsknet: { ipv4_address: 172.28.7.11 } + redis-replica-2: + <<: *redis-data + command: >- + redis-server --appendonly yes --appendfsync everysec --save "" + --replicaof 172.28.7.10 6379 --replica-read-only yes --min-replicas-to-write 1 --min-replicas-max-lag 10 + depends_on: + redis-master: { condition: service_healthy } + ports: ["6392:6379"] + networks: + tsknet: { ipv4_address: 172.28.7.12 } + sentinel-1: &sentinel + image: redis:7.4.5-alpine@sha256:bb186d083732f669da90be8b0f975a37812b15e913465bb14d845db72a4e3e08 + depends_on: + redis-master: { condition: service_healthy } + redis-replica-1: { condition: service_healthy } + redis-replica-2: { condition: service_healthy } + entrypoint: + - sh + - -c + - | + cat > /tmp/sentinel.conf < +set -euo pipefail +DRILL="${1:?usage: run.sh }" +DIR="$(cd "$(dirname "$0")" && pwd)" +ROOT="$(cd "$DIR/../.." && pwd)" +compose() { (cd "$DIR" && docker compose -p tsk-sentinel "$@"); } +cleanup() { compose down -v --remove-orphans >/dev/null 2>&1 || true; } +trap cleanup EXIT +cleanup +compose up -d + +echo "waiting for sentinels to see master + 2 replicas..." +ok=0 +for _ in $(seq 1 60); do + n="$(docker exec tsk-sentinel-sentinel-1-1 redis-cli -p 26379 sentinel master tskmaster 2>/dev/null | awk '/num-slaves/{getline; print}' | tr -d '\r' || echo 0)" + if [ "${n:-0}" -ge 2 ]; then ok=1; break; fi + sleep 1 +done +[ "$ok" = 1 ] || { echo "topology did not converge (num-slaves<2)"; compose ps; exit 1; } +# ensure both replicas are CAUGHT UP (min-replicas-to-write would otherwise reject the first durable claim). +docker exec tsk-sentinel-redis-master-1 redis-cli WAIT 2 10000 >/dev/null || true + +export TSK_TEST_SENTINELS="127.0.0.1:26379,127.0.0.1:26380,127.0.0.1:26381" +export TSK_TEST_SENTINEL_MASTER="tskmaster" +export TSK_SENTINEL_NATMAP="172.28.7.10:6379=127.0.0.1:6390,172.28.7.11:6379=127.0.0.1:6391,172.28.7.12:6379=127.0.0.1:6392" +export TSK_SENTINEL_NETWORK="tsk-sentinel_tsknet" +export TSK_SENTINEL_MASTER_CONTAINER="tsk-sentinel-redis-master-1" +export TSK_SENTINEL_MASTER_PORT="6390" + +cd "$ROOT" +node --import tsx "$DRILL" diff --git a/package.json b/package.json index 86b4a5a..d2de0e9 100644 --- a/package.json +++ b/package.json @@ -24,6 +24,10 @@ "test:source-export": "npm run build -w packages/server && npx tsx tsk-source-export-drill.mts", "test:receiver": "npm run build -w packages/server && npx tsx tsk-receiver-drill.mts", "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: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", "test:pack": "npm run build && npx tsx package-boundary-suite.mts && npm pack --dry-run --workspaces", "test:bpc-compat": "npm run build && npx tsx bpc-compatibility-suite.mts", diff --git a/packages/server/src/ha-control-fencing.ts b/packages/server/src/ha-control-fencing.ts index 8e9937d..27645f6 100644 --- a/packages/server/src/ha-control-fencing.ts +++ b/packages/server/src/ha-control-fencing.ts @@ -3,8 +3,9 @@ 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 } from './tsk-source-fence.js'; -import type { SourceFrozenReceipt, BFinalizedReceipt, SourceVerifyKeyResolver } from './tsk-source-fence.js'; +import { verifySourceFrozenReceipt, verifyBFinalizedReceipt, signLeaseGrant } from './tsk-source-fence.js'; +import type { SourceFrozenReceipt, BFinalizedReceipt, SourceVerifyKeyResolver, LeaseGrant } from './tsk-source-fence.js'; +import type { KeyObject } from 'node:crypto'; /** * PR2a — HA fencing FOUNDATION (control DB). Implements the reviewed design @@ -304,6 +305,17 @@ CREATE TABLE IF NOT EXISTS tsk_ha_cutover_history ( created_at timestamptz NOT NULL DEFAULT now(), 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}$'), + 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(); @@ -311,6 +323,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', ] 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 @@ -407,7 +420,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 = '68d4710a6bc72db1cfa4734f03f052fc963daa74c7e27787f595b0b29f45f6c3'; +export const HA_CONTROL_MANIFEST_DIGEST = 'a80069c514d7450a5a2f4482d99952f9b2920ee876d14f51ba5d516ff79eefab'; /** Attest the live catalog hashes to the pinned expected manifest (fail-closed, NOT TOFU). */ async function attestControlSchema(exec: PgExecutor): Promise { @@ -514,6 +527,11 @@ 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 + * 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). */ + sourceGuard?: { keyId: string; privateKey: import('node:crypto').KeyObject | string; activationTtlMs: number }; } export interface FenceEvidence { holderNodeId: string; grantSeq: number; grantDigest: string; maxExpiryMs: number; @@ -558,6 +576,7 @@ export class HaControlFencing { private readonly schema: string; private readonly minClaimRemainingMs: number; private readonly attestEveryTx: boolean; + private readonly sourceGuard?: HaControlPolicy['sourceGuard']; constructor( private readonly db: PgTransactor, private readonly signer: GuardSigner, @@ -568,6 +587,11 @@ export class HaControlFencing { this.schema = requireReady(ready, db).schema; this.minClaimRemainingMs = vInt(policy?.minClaimRemainingMs, 'policy.minClaimRemainingMs', 1, MAX_CLAIM_REMAINING_MS); // strictly positive this.attestEveryTx = policy?.attestEveryTx !== false; // default true (R7-HIGH1) + if (policy?.sourceGuard) { + if (!policy.sourceGuard.keyId || !policy.sourceGuard.privateKey) throw new ContractValidationError('policy.sourceGuard requires keyId + privateKey'); + vInt(policy.sourceGuard.activationTtlMs, 'policy.sourceGuard.activationTtlMs', 1, MAX_MS); + this.sourceGuard = policy.sourceGuard; + } } /** Exact-session SERIALIZABLE + pinned schema + (R7-HIGH1) LIVE per-tx catalog re-attestation + @@ -1031,6 +1055,91 @@ export class HaControlFencing { }); } + /** + * (§5 / PR2c) READY → ACTIVE. The FINAL, TERMINAL cutover transition: B is now the live write authority + * for the stream at the new epoch. Requires a READY head for this command/epoch (control already verified + * B's BFinalizedReceipt, the exact frozen N/epoch, and B != source/control). The ACTIVE head carries the + * ratified B binding forward as the durable go-live proof and is TERMINAL (a new promotion needs a new + * command + epoch). Idempotent resume: a re-activation is accepted ONLY if it reproduces the byte-identical + * signed ACTIVE evidence. Crash-safe: a single signed forward-CAS on the cutover head. + */ + async activate(streamId: string, commandId: string, targetEpoch: number): Promise { + const s = vId(streamId, STREAM_ID_RE, 'streamId'); + const cmd = vId(commandId, ID_RE, 'commandId'); + const target = vInt(targetEpoch, 'targetEpoch', 1, MAX_EPOCH); + return this.criticalTx(s, async (exec) => { + const cur = await this.readCutover(exec, s); + const isActiveRetry = !!cur && cur.phase === 'ACTIVE' && cur.commandId === cmd && cur.epoch === target; + if (!cur || (!isActiveRetry && (cur.phase !== 'READY' || cur.commandId !== cmd || cur.epoch !== target))) throw new FenceAuthorityQuarantineError('no matching READY cutover to ACTIVATE — quarantine'); + // carry the ratified READY binding forward as the terminal go-live proof. + const rdRow = (await exec.query("SELECT evidence FROM tsk_ha_cutover_history WHERE stream_id=$1 AND command_id=$2 AND epoch=$3 AND phase='READY'", [s, cmd, target])).rows[0]; + if (!rdRow || rdRow.evidence === null) throw new FenceAuthorityQuarantineError('no bound READY evidence to activate against — quarantine'); + const rd = JSON.parse(String(rdRow.evidence)) as { bReceiptDigest: string; generationId: string; bSystemId: string; sourceSystemId: string; controlSystemId: string; n: number; frozenReceiptDigest: string; bKeyId: string }; + const ev = JSON.stringify({ k: 'active/v1', bReceiptDigest: rd.bReceiptDigest, generationId: rd.generationId, bSystemId: rd.bSystemId, + sourceSystemId: rd.sourceSystemId, controlSystemId: rd.controlSystemId, n: rd.n, frozenReceiptDigest: rd.frozenReceiptDigest, bKeyId: rd.bKeyId }); + if (isActiveRetry) { + if (cur!.evidence !== ev) throw new FenceAuthorityQuarantineError('ACTIVE retry does not reproduce the ratified go-live evidence — quarantine'); + return cur!; + } + return this.cutTransition(exec, s, target, cmd, cur!.seqno + 1, 'ACTIVE', ev, cur!.stateDigest); + }); + } + + /** + * (§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 + * 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. + */ + async activateSource(streamId: string, commandId: string, targetEpoch: number, bReceipt: BFinalizedReceipt, bResolver: SourceVerifyKeyResolver): 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); + verifyBFinalizedReceipt(bResolver, br); + 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)'); + const leaseId = vId(`bsrc-${cmd}`, ID_RE, 'derived leaseId'); + return this.criticalTx(s, async (exec) => { + const cur = await this.readCutover(exec, s); + if (!cur || cur.phase !== 'ACTIVE' || cur.commandId !== cmd || cur.epoch !== target) throw new FenceAuthorityQuarantineError('no ACTIVE cutover to activate the source authority against — quarantine'); + const rd = JSON.parse(cur.evidence ?? '{}') as { bReceiptDigest?: string; n?: number; bSystemId?: string; bKeyId?: string }; + // 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)'); + } + return JSON.parse(String(existing.grant_json)) as LeaseGrant; // rehydrated, byte-identical + } + // 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 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, + }); + // 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'); + return grantObj; + }); + } + /** Prove the OLD holder is fenced WITHOUT touching A's PG: an exact current revoked lease at * epoch target-1 whose MONOTONIC max grant-expiry (+ bounded margin) has elapsed on the control * clock read in THIS tx. A missing lease is NOT acceptable. */ diff --git a/packages/server/src/redis-fencing-store.ts b/packages/server/src/redis-fencing-store.ts index 9d87ce5..e08ac09 100644 --- a/packages/server/src/redis-fencing-store.ts +++ b/packages/server/src/redis-fencing-store.ts @@ -55,12 +55,34 @@ function parseFenceRecord(raw: string): FenceRecord { * Redis persistence, replication, ACLs, TLS, and availability remain deployment * responsibilities and must be tested for the selected topology. */ +/** Durable-claim policy: after the epoch CAS SET, require `waitReplicas` replicas to ACK the write + * (Redis WAIT) within `waitTimeoutMs` before the claim is reported successful. This makes RPO=0 an + * ENFORCED property of the claim path (not a caller-side WAIT): a claim that a replica quorum did not + * durably receive fails closed, so a subsequent Sentinel failover cannot roll the fence epoch back. */ +export interface FenceDurabilityPolicy { waitReplicas: number; waitTimeoutMs: number } + +/** Thrown when the epoch CAS SET SUCCEEDED (the fence epoch was raised on the current master) but the + * configured replica quorum did NOT ACK within the WAIT window — so the write's DURABILITY is UNKNOWN. + * This is DISTINCT from an ordinary `claim() === false` (CAS refused because the epoch was not higher): + * here the write happened, so the caller must RECONCILE against `storedTuple` and fail closed, never + * treat it as a clean "not acquired". Carries the exact stored record for reconciliation. */ +export class FenceDurabilityUncertainError extends Error { + constructor(readonly acked: number, readonly required: number, readonly storedTuple: FenceRecord | null) { + super(`fence epoch CAS wrote but only ${acked}/${required} replicas ACK'd — durability UNCERTAIN; reconcile and fail closed`); + this.name = 'FenceDurabilityUncertainError'; + } +} + export class RedisFencingStore implements FencingStore { constructor( private readonly redis: Redis, private readonly key = 'tsk:fencing:writer', + private readonly durability?: FenceDurabilityPolicy, ) { if (!key || key.length > 512) throw new Error('Redis fencing key must be 1..512 characters'); + if (durability && (!Number.isInteger(durability.waitReplicas) || durability.waitReplicas < 1 || !Number.isInteger(durability.waitTimeoutMs) || durability.waitTimeoutMs < 1)) { + throw new Error('FenceDurabilityPolicy requires waitReplicas>=1 and waitTimeoutMs>=1'); + } } async current(): Promise { @@ -71,14 +93,52 @@ export class RedisFencingStore implements FencingStore { async claim(record: Omit): Promise { const next: FenceRecord = { ...record, active: true }; parseFenceRecord(JSON.stringify(next)); - const result = await this.redis.eval( - CLAIM_SCRIPT, - 1, - this.key, - String(record.fenceEpoch), - JSON.stringify(next), - ); - return result === 1; + if (!this.durability) { + const result = await this.redis.eval(CLAIM_SCRIPT, 1, this.key, String(record.fenceEpoch), JSON.stringify(next)); + return result === 1; // CAS refused (0) or acquired (1); no durability enforcement configured. + } + // DURABLE path. Redis WAIT tracks the writes of the CURRENT CONNECTION (per-client replication offset), + // so WAIT is only meaningful on the SAME physical connection that ran the CAS. A shared, auto-reconnecting + // Sentinel client is INSUFFICIENT: ioredis can RESEND/REPLAY un-replied commands on a freshly promoted + // master, so WAIT (or a replayed EVAL) could bind a different server whose offset does not cover the CAS. + // So we acquire a DEDICATED physical connection to the CURRENT master with ALL reconnect/retry/resend + // DISABLED, dispatch EVAL then WAIT on that ONE socket, and treat ANY disconnect/rejection AFTER the CAS + // was dispatched as DURABILITY-UNKNOWN → typed uncertainty (never a silent success or ordinary false). + // The connection is closed at the end. A pre-dispatch connect failure is propagated (the claim never + // started — the caller retries / fails closed). + const { waitReplicas, waitTimeoutMs } = this.durability; + const conn = this.redis.duplicate({ + lazyConnect: true, + enableOfflineQueue: false, // no queueing while disconnected — a dropped command rejects + maxRetriesPerRequest: 0, // never retry a request on another connection + retryStrategy: () => null, // never reconnect the data socket + reconnectOnError: () => false, + autoResendUnfulfilledCommands: false, // never REPLAY the un-replied EVAL/WAIT on a new master + role: 'master', + } as Record) as unknown as { + connect(): Promise; disconnect(): void; + eval(...a: unknown[]): Promise; call(...a: unknown[]): Promise; + }; + const uncertain = async (acked: number): Promise => { + let stored: FenceRecord | null = null; + try { stored = await this.current(); } catch { stored = null; } + throw new FenceDurabilityUncertainError(acked, waitReplicas, stored); + }; + try { + await conn.connect(); // pre-dispatch: resolve the current master + open ONE socket; failure propagates. + let casRes: unknown; + try { casRes = await conn.eval(CLAIM_SCRIPT, 1, this.key, String(record.fenceEpoch), JSON.stringify(next)); } + catch { return uncertain(0); } // dropped during/after EVAL dispatch → the CAS may have applied + if (casRes !== 1) return false; // CAS refused (epoch not strictly higher) — ordinary durable no-op + // the CAS WROTE on THIS socket. WAIT on the SAME socket binds this connection's offset (covers the CAS). + let acked: number; + try { acked = Number(await conn.call('WAIT', String(waitReplicas), String(waitTimeoutMs))); } + catch { return uncertain(0); } // dropped during/after WAIT dispatch → durability UNKNOWN + if (!Number.isFinite(acked) || acked < waitReplicas) return uncertain(acked); + return true; + } finally { + try { conn.disconnect(); } catch { /* already closed */ } + } } async release(nodeId: string, fenceEpoch: number, commandId: string): Promise { diff --git a/redis-fencing-integration.mts b/redis-fencing-integration.mts index ebf052c..4d3820a 100644 --- a/redis-fencing-integration.mts +++ b/redis-fencing-integration.mts @@ -7,7 +7,7 @@ import { handlePromotionCommand, signGuardCommand, } from './packages/server/src/promotion.ts'; -import { RedisFencingStore } from './packages/server/src/redis-fencing-store.ts'; +import { RedisFencingStore, FenceDurabilityUncertainError } from './packages/server/src/redis-fencing-store.ts'; import { MemoryTumblerStore } from './packages/server/src/store.ts'; import type { ReplicationCheckpoint } from './packages/server/src/replicating-tumbler-store.ts'; @@ -121,6 +121,56 @@ try { try { await fenceA.current(); } catch { rejected = true; } assert(rejected, 'corrupt record was accepted as empty/current'); }); + + await test('enforced-WAIT: a CAS that WROTE but the replica quorum did not ACK throws typed Uncertain (not ordinary false)', async () => { + const durKey = `${key}:dur`; + await redisA.del(durKey); + const durStore = new RedisFencingStore(redisA, durKey, { waitReplicas: 1, waitTimeoutMs: 500 }); + // single node, NO replicas → WAIT returns 0 < 1 → the CAS wrote but durability is UNKNOWN → typed throw. + let typed = false, stored = false; + try { await durStore.claim({ nodeId: 'B', fenceEpoch: 1, expiresAt: Date.now() + 3_600_000, commandId: 'c1' }); } + catch (e) { typed = e instanceof FenceDurabilityUncertainError; stored = (e as FenceDurabilityUncertainError).storedTuple?.fenceEpoch === 1; } + assert(typed, 'expected a typed FenceDurabilityUncertainError, not an ordinary false, when WAIT under-acks after a successful CAS'); + assert(stored, 'the typed error must carry the exact stored tuple for reconciliation'); + // a CAS that is REFUSED (epoch not strictly higher) is an ordinary durable no-op — false, NEVER a throw. + const refused = await durStore.claim({ nodeId: 'B', fenceEpoch: 1, expiresAt: Date.now() + 3_600_000, commandId: 'c1' }); + assert(refused === false, 'a refused CAS must return ordinary false (no WAIT ambiguity)'); + await redisA.del(durKey); + }); + + await test('enforced-WAIT dedicated-connection continuity: disconnect/rejection AFTER dispatch is typed Uncertain, never silent/false; no reconnect/replay', async () => { + // Deterministic hermetic stub of the DEDICATED physical connection (redis.duplicate). Each case programs + // the EVAL/WAIT outcome so a disconnect/rejection AFTER the CAS is dispatched is exercised WITHOUT a live + // cluster — proving claim() reports typed uncertainty on that exact path, and that the connection is + // created with reconnect/retry/RESEND fully disabled (so a whole-pipeline replay on a new master, which + // would return matching ids + CAS0, cannot occur → CAS0 is an honest refusal, not a silent success). + const dur = { waitReplicas: 1, waitTimeoutMs: 500 }; + const storedJson = JSON.stringify({ nodeId: 'B', fenceEpoch: 1, expiresAt: Date.now() + 3_600_000, commandId: 'c1', active: true }); + let lastOpts: Record = {}; + const conn = (evalOut: () => Promise, callOut: () => Promise, connectOut: () => Promise = async () => undefined) => + ({ connect: connectOut, disconnect: () => {}, eval: evalOut, call: callOut }); + const mk = (c: ReturnType) => new RedisFencingStore({ duplicate: (o: Record) => { lastOpts = o; return c; }, get: async () => storedJson } as unknown as Redis, 'k:ded', dur); + const rec = { nodeId: 'B', fenceEpoch: 1, expiresAt: Date.now() + 3_600_000, commandId: 'c1' }; + const reject = () => { throw new Error('Connection is closed'); }; + const isUncertain = async (c: ReturnType) => { try { await mk(c).claim(rec); return false; } catch (e) { return e instanceof FenceDurabilityUncertainError; } }; + // EVAL rejects AFTER dispatch (socket dropped mid-CAS) → the CAS may have applied → Uncertain. + assert(await isUncertain(conn(async () => reject(), async () => 1)), 'an EVAL rejection after dispatch must be typed Uncertain'); + // CAS wrote (1), then WAIT rejects (socket dropped) → Uncertain. + assert(await isUncertain(conn(async () => 1, async () => reject())), 'a WAIT rejection after a successful CAS must be typed Uncertain'); + // CAS wrote (1), WAIT under-acked → Uncertain. + assert(await isUncertain(conn(async () => 1, async () => 0)), 'an under-ack must be typed Uncertain'); + // CAS wrote (1), WAIT acked on the SAME dedicated socket → durable success. + assert((await mk(conn(async () => 1, async () => 1)).claim(rec)) === true, 'a quorum-acked same-connection claim succeeds'); + // CAS REFUSED (epoch not higher) → ordinary false; replay is IMPOSSIBLE on this connection so CAS0 is honest. + assert((await mk(conn(async () => 0, async () => 0)).claim(rec)) === false, 'a genuinely refused CAS is ordinary false'); + // a PRE-dispatch connect failure propagates (the claim never started) — not a silent false. + let connectFailPropagated = false; + try { await mk(conn(async () => 1, async () => 1, async () => reject())).claim(rec); } + catch (e) { connectFailPropagated = !(e instanceof FenceDurabilityUncertainError); } + assert(connectFailPropagated, 'a pre-dispatch connect failure must propagate, not become a silent false/uncertain'); + // STRUCTURAL: the dedicated connection disables reconnect/retry/offline-queue/RESEND → no path-switch/replay. + assert(lastOpts.autoResendUnfulfilledCommands === false && lastOpts.enableOfflineQueue === false && lastOpts.maxRetriesPerRequest === 0 && typeof lastOpts.retryStrategy === 'function' && (lastOpts.retryStrategy as () => unknown)() === null, 'claim must pin a non-retrying, non-replaying dedicated connection'); + }); } finally { await redisA.del(key).catch(() => undefined); await Promise.all([redisA.quit().catch(() => undefined), redisB.quit().catch(() => undefined)]); diff --git a/tsk-b-activation-drill.mts b/tsk-b-activation-drill.mts new file mode 100644 index 0000000..54f735f --- /dev/null +++ b/tsk-b-activation-drill.mts @@ -0,0 +1,163 @@ +/** + * PR2c acceptance (#10) — GOVERNED B source-authority activation: B originates N+1 while old A is denied. + * + * The full cutover is driven to ACTIVE, then control.activateSource() mints B's UNFORGEABLE source capability + * (a GUARD-signed ed25519 lease at the PROMOTED epoch, bound to the ratified ACTIVE head + BFinalizedReceipt + + * exact epoch). B installs that governed lease on its own PG, seeds its source checkpoint to the verified + * (N, head@N), attests + mints its SourceFenceReadyToken, and appends sequence N+1 through the REAL fenced + * outbox — chaining from head@N. Meanwhile the OLD A, at the prior epoch with a revoked lease, is DENIED. + * + * Env: TSK_TEST_SOURCE_PG_URL_A + TSK_TEST_RECEIVER_PG_URL_B + TSK_TEST_CONTROL_PG_URL + TSK_TEST_REDIS_URL. + */ +import assert from 'node:assert/strict'; +import { generateKeyPairSync, sign as edSign } from 'node:crypto'; +import pg from 'pg'; +import { Redis } from 'ioredis'; +import { + TSK_OUTBOX_PG_SCHEMA, TSK_SOURCE_LEASE_SCHEMA, TSK_SOURCE_WITNESS_SCHEMA, TSK_RECEIVER_SCHEMA, TSK_RECEIVER_TABLES, provisionSchemaVersion, + PgTskDurableOutbox, NodePostgresTransactor, ContractValidationError, + signLeaseGrant, installLeaseGrant, assertSourceFenceReady, emitSourceFrozenReceipt, + buildSourceExportManifest, guardCountersignSourceExport, + assertReceiverReady, stageAndFinalizeReceiverGeneration, verifyBFinalizedReceipt, + 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, + type GuardKeyResolver, type HaControlPolicy, type FenceProof, +} from './packages/server/dist/index.js'; + +const A_URL = process.env['TSK_TEST_SOURCE_PG_URL_A']; +const B_URL = process.env['TSK_TEST_RECEIVER_PG_URL_B']; +const CTRL_URL = process.env['TSK_TEST_CONTROL_PG_URL']; +const REDIS_URL = process.env['TSK_TEST_REDIS_URL'] ?? ''; +if (!A_URL || !B_URL || !CTRL_URL || !REDIS_URL) throw new Error('need A + B + control PG URLs + Redis URL (independent instances)'); + +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 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 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 } }; +const mkSigner = (kid: string, priv: Parameters[2]): StreamHeadSigner => ({ keyId: kid, alg: 'ed25519', async sign(d) { return edSign(null, Buffer.from(d, 'utf8'), priv).toString('base64url'); } }); +const signer = mkSigner(HEAD_KEY, headKp.privateKey); +const bSigner = mkSigner(BHEAD_KEY, bHeadKp.privateKey); +const sanitizer: HotpMutationSanitizer = { + sanitize(raw) { if (typeof (raw as TskHotpMutation).tumblerId !== 'string' || !Number.isInteger((raw as TskHotpMutation).counter)) throw new ContractValidationError('bad'); return { tumblerId: (raw as TskHotpMutation).tumblerId, counter: (raw as TskHotpMutation).counter } as SanitizedMutation; }, + assertSanitized(c): asserts c is SanitizedMutation { if (!c || typeof c !== 'object') throw new ContractValidationError('unsanitized'); }, +}; +let passed = 0; +async function check(name: string, fn: () => Promise) { await fn(); passed++; console.log(` ok - ${name}`); } + +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', () => {}); + const bPool = new pg.Pool({ connectionString: B_URL, max: 4 }); bPool.on('error', () => {}); + const cPool = new pg.Pool({ connectionString: CTRL_URL, max: 6 }); cPool.on('error', () => {}); + 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 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) => { + await pool.query(OUTBOX_DROP); + for (const s of TSK_OUTBOX_PG_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await pool.query(s); + 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 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`); + for (const s of TSK_RECEIVER_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await bPool.query(s); + await installSource(bPool); + await cPool.query(`DROP TABLE IF EXISTS ${HA_CONTROL_TABLES.join(', ')} CASCADE`); + for (const s of HA_CONTROL_PG_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await cPool.query(s); + + const aSysId = String((await aPool.query('SELECT system_identifier::text AS s FROM pg_control_system()')).rows[0].s); + const bSysId = String((await bPool.query('SELECT system_identifier::text AS s FROM pg_control_system()')).rows[0].s); + const cSysId = String((await cPool.query('SELECT system_identifier::text AS s FROM pg_control_system()')).rows[0].s); + assert.equal(new Set([aSysId, bSysId, cSysId]).size, 3, 'A, B, control independent'); + + const SID = 'tsk:pair:bactivate/v1'; const CMD = 'promote-1'; const TARGET = 1; const SRC_EPOCH = 'e1'; + // ── A builds 1..N, freezes, exports; B stages+finalizes the generation (state@N) ── + const READY = await provisionSchemaVersion(aTx, SCHEMA); + 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 }); + 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 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 }); + 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' }); + 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 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); + verifyBFinalizedReceipt(resolver, bReceipt); + const headAtN = bReceipt.signedHeadDigestAtN; + + // ── 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.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 proof: FenceProof = { safetyMarginMs: 0, claimExpiresAtMs: (await ctlNow()) + HOUR }; + await ctl.advanceEpoch(SID, CMD, TARGET, 'Bnode', store, proof); + await ctl.markImporting(SID, CMD, TARGET); + await ctl.markReady(SID, CMD, TARGET, bReceipt, resolver); + await ctl.activate(SID, CMD, TARGET); + + let bGrant: Awaited>; + 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) + 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); + 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 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)); + 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 }); + 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'); + const row = (await bPool.query('SELECT sequence, fence_token, head_prev FROM tsk_outbox_rows WHERE stream_id=$1 ORDER BY sequence DESC LIMIT 1', [SID])).rows[0]; + assert.equal(Number(row.sequence), N + 1); assert.equal(Number(row.fence_token), TARGET); assert.equal(String(row.head_prev), headAtN); + }); + + await check('the OLD A, at the prior epoch with a revoked lease, is DENIED — no split-brain source', async () => { + // A's outbox still points at epoch 0 with a REVOKED lease; the in-tx source-fence gate refuses the append. + await assert.rejects(() => aOb.withOutboxTx((wtx) => aOb.appendInTx(wtx, { streamId: SID, rawMutation: { tumblerId: 'T1', counter: 3 }, fenceToken: 0n })), /revoked|not writable|lease|fence/i); + const aMax = Number((await aPool.query('SELECT COALESCE(MAX(sequence),0) AS n FROM tsk_outbox_rows WHERE stream_id=$1', [SID])).rows[0].n); + assert.equal(aMax, N, 'A wrote nothing past N — the old source is fenced'); + }); + + console.log(`\n# ${passed} PR2c B-source-activation checks passed`); + 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); }); diff --git a/tsk-control-cutover-drill.mts b/tsk-control-cutover-drill.mts index ff9b52b..886ecf5 100644 --- a/tsk-control-cutover-drill.mts +++ b/tsk-control-cutover-drill.mts @@ -213,6 +213,17 @@ async function main() { mut.bSystemId = savedB; }); + await check('READY → ACTIVE: terminal go-live; B is the live authority; idempotent; ACTIVE carries the ratified binding', async () => { + const ac = await ctl.activate(SID, CMD, 1); + assert.equal(ac.phase, 'ACTIVE'); assert.equal(ac.commandId, CMD); assert.equal(ac.epoch, 1); + const ev = JSON.parse(ac.evidence!); + assert.equal(ev.k, 'active/v1'); assert.equal(ev.bReceiptDigest, s1.receipt.receiptDigest); + assert.equal(ev.bSystemId, bSysId); assert.equal(ev.sourceSystemId, aSysId); assert.equal(ev.controlSystemId, cSysId); assert.equal(ev.n, 6); + assert.equal((await ctl.activate(SID, CMD, 1)).phase, 'ACTIVE'); // idempotent terminal + // ACTIVE is terminal — a stale READY re-mark on the completed promotion is refused. + await assert.rejects(() => ctl.markReady(SID, CMD, 1, s1.receipt, resolver), /no matching IMPORTING/); + }); + // ── negative: foreign freeze cannot be bound to a cutover ───────────────────── await check('bindSourceFenced REJECTS a frozen receipt for a different stream/command (no cross-stream splice)', async () => { const SIDX = 'tsk:pair:foreign/v1'; diff --git a/tsk-cutover-sigkill-drill.mts b/tsk-cutover-sigkill-drill.mts new file mode 100644 index 0000000..3fafc97 --- /dev/null +++ b/tsk-cutover-sigkill-drill.mts @@ -0,0 +1,203 @@ +/** + * PR2c acceptance (#10) — child-process SIGKILL / restart matrix over the FULL control cutover. + * + * Real independent PG16 (A source @A, B receiver @B, control @control) + real Redis. The parent builds + * the real signed receipts and drives the cutover; for EACH transition + * (bindSourceFenced → advanceEpoch → markImporting → markReady → activate) it spawns a CHILD process + * (tsk-cutover-worker.mts) that performs ONLY that transition and: + * • CRASH=before-commit → the child dies by a REAL SIGKILL inside the tx, before COMMIT → the parent + * asserts the cutover phase is UNCHANGED (PostgreSQL rolled the tx back — no partial/torn state); + * • CRASH=after-commit → the child COMMITs then SIGKILLs → a FRESH child re-runs the SAME transition → + * the parent asserts it resumed IDEMPOTENTLY (phase advanced exactly once, byte-identical evidence). + * The final phase is ACTIVE (B is the live authority). Per-fault RPO/RTO reported. + * + * Env: TSK_TEST_SOURCE_PG_URL_A + TSK_TEST_RECEIVER_PG_URL_B + TSK_TEST_CONTROL_PG_URL + TSK_TEST_REDIS_URL. + */ +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import { writeFileSync, rmSync, existsSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { generateKeyPairSync, sign as edSign } from 'node:crypto'; +import pg from 'pg'; +import { Redis } from 'ioredis'; +import { + TSK_OUTBOX_PG_SCHEMA, TSK_SOURCE_LEASE_SCHEMA, TSK_SOURCE_WITNESS_SCHEMA, TSK_RECEIVER_SCHEMA, TSK_RECEIVER_TABLES, provisionSchemaVersion, + PgTskDurableOutbox, NodePostgresTransactor, ContractValidationError, + signLeaseGrant, installLeaseGrant, assertSourceFenceReady, emitSourceFrozenReceipt, + buildSourceExportManifest, guardCountersignSourceExport, + assertReceiverReady, stageAndFinalizeReceiverGeneration, verifyBFinalizedReceipt, + HA_CONTROL_PG_SCHEMA, HA_CONTROL_TABLES, HaControlFencing, GuardSigner, provisionControlSchema, + type PgTransactor, type StreamHeadSigner, type HotpMutationSanitizer, type SanitizedMutation, + type TskHotpMutation, type SourceVerifyKeyResolver, type SourceExportBundle, type GuardCountersignedExport, + type GuardKeyResolver, type HaControlPolicy, +} from './packages/server/dist/index.js'; + +const A_URL = process.env['TSK_TEST_SOURCE_PG_URL_A']; +const B_URL = process.env['TSK_TEST_RECEIVER_PG_URL_B']; +const CTRL_URL = process.env['TSK_TEST_CONTROL_PG_URL']; +const REDIS_URL = process.env['TSK_TEST_REDIS_URL'] ?? ''; +if (!A_URL || !B_URL || !CTRL_URL || !REDIS_URL) throw new Error('need A + B + control PG URLs + Redis URL (independent instances)'); + +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 HEAD_KEY = 'k1'; const headKp = generateKeyPairSync('ed25519'); +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 === B_KEY ? bKp.publicKey : null) }; +const CTRL_KEY = 'ctrl-1'; const ctrlSecret = Buffer.alloc(32, 0x2b); +const ctrlResolver: GuardKeyResolver = { resolve: (kid) => (kid === CTRL_KEY ? ctrlSecret : null) }; +const ctrlSigner = new GuardSigner(CTRL_KEY, ctrlSecret); +const POLICY: HaControlPolicy = { minClaimRemainingMs: 5_000 }; +const signer: StreamHeadSigner = { keyId: HEAD_KEY, alg: 'ed25519', async sign(d) { return edSign(null, Buffer.from(d, 'utf8'), headKp.privateKey).toString('base64url'); } }; +const sanitizer: HotpMutationSanitizer = { + sanitize(raw) { if (typeof (raw as TskHotpMutation).tumblerId !== 'string' || !Number.isInteger((raw as TskHotpMutation).counter)) throw new ContractValidationError('bad'); return { tumblerId: (raw as TskHotpMutation).tumblerId, counter: (raw as TskHotpMutation).counter } as SanitizedMutation; }, + assertSanitized(c): asserts c is SanitizedMutation { if (!c || typeof c !== 'object') throw new ContractValidationError('unsanitized'); }, +}; +let passed = 0; +async function check(name: string, fn: () => Promise) { await fn(); passed++; console.log(` ok - ${name}`); } +const now = () => Number(process.hrtime.bigint() / 1_000_000n); + +interface WorkerRun { code: number | null; signal: string | null; out: string } +function runWorker(handoffPath: string, transition: string, crash: string, crashTxIndex = 1): Promise { + return new Promise((resolve) => { + const child = spawn(process.execPath, ['--import', 'tsx', 'tsk-cutover-worker.mts', handoffPath], { + env: { ...process.env, TRANSITION: transition, CRASH: crash, CRASH_TX_INDEX: String(crashTxIndex) }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + let out = ''; + child.stdout.on('data', (d) => { out += String(d); }); + child.stderr.on('data', (d) => { out += String(d); }); + child.on('close', (code, signal) => resolve({ code, signal, out })); + }); +} + +async function main() { + console.log('# TSK PR2c child-process SIGKILL/restart matrix over the full control cutover'); + const aPool = new pg.Pool({ connectionString: A_URL, max: 4 }); aPool.on('error', () => {}); + const bPool = new pg.Pool({ connectionString: B_URL, max: 4 }); bPool.on('error', () => {}); + const cPool = new pg.Pool({ connectionString: CTRL_URL, max: 6 }); cPool.on('error', () => {}); + 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(); + + // fresh schemas + await aPool.query('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'); + for (const s of TSK_OUTBOX_PG_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await aPool.query(s); + for (const s of TSK_SOURCE_LEASE_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await aPool.query(s); + for (const s of TSK_SOURCE_WITNESS_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await aPool.query(s); + await bPool.query(`DROP TABLE IF EXISTS ${TSK_RECEIVER_TABLES.join(', ')} CASCADE`); + for (const s of TSK_RECEIVER_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await bPool.query(s); + await cPool.query(`DROP TABLE IF EXISTS ${HA_CONTROL_TABLES.join(', ')} CASCADE`); + for (const s of HA_CONTROL_PG_SCHEMA.split(';').map((x) => x.trim()).filter(Boolean)) await cPool.query(s); + + const aSysId = String((await aPool.query('SELECT system_identifier::text AS s FROM pg_control_system()')).rows[0].s); + const bSysId = String((await bPool.query('SELECT system_identifier::text AS s FROM pg_control_system()')).rows[0].s); + const cSysId = String((await cPool.query('SELECT system_identifier::text AS s FROM pg_control_system()')).rows[0].s); + assert.equal(new Set([aSysId, bSysId, cSysId]).size, 3, 'A, B, control independent'); + + const SID = 'tsk:pair:sigkill/v1'; const CMD = 'promote-1'; const TARGET = 1; + // ── build the real receipts on A + finalize on B (once) ── + const READY = await provisionSchemaVersion(aTx, SCHEMA); + 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, 'e1']); + 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 }); + await aTx.transaction((exec) => installLeaseGrant(exec, resolver, g)); + const sready = await assertSourceFenceReady(aTx, SCHEMA, resolver, { streamId: SID, holderNodeId: 'A', leaseId: 'l1', grantDigest: g.grantDigest }); + const ob = new PgTskDurableOutbox(aTx, READY, { streamId: SID, sanitizer, signer, maxPendingRows: 100_000, backpressure: 'fail-authoritative-mutation' }, { resolver, controlToASkewBoundMs: 0, ready: sready }); + for (const [t, c] of [['T1', 1], ['T2', 5], ['T1', 2], ['T3', 9]] as [string, number][]) await ob.withOutboxTx((wtx) => ob.appendInTx(wtx, { streamId: SID, rawMutation: { tumblerId: t, counter: c }, fenceToken: 0n })); + 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 }); + 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 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 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); + verifyBFinalizedReceipt(resolver, bReceipt); + + // ── control setup: provision + active lease + intent (parent owns the non-target setup) ── + const ctlReady = await provisionControlSchema(cTx as never, SCHEMA); + const ctl = new HaControlFencing(cTx as never, ctrlSigner, 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-sigkill'); + await ctl.writeLease({ streamId: SID, leaseId: 'l1', holderNodeId: 'A', epoch: 0, status: 'active', grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'a1' }); + await ctl.beginPromotionIntent(SID, CMD, TARGET); + + // handoff for the child workers (public keys only; control HMAC secret shared for symmetric custody). + const pubKeys: Record = { + [GUARD_KEY]: guard.publicKey.export({ type: 'spki', format: 'pem' }) as string, + [SOURCE_KEY]: source.publicKey.export({ type: 'spki', format: 'pem' }) as string, + [HEAD_KEY]: headKp.publicKey.export({ type: 'spki', format: 'pem' }) as string, + [B_KEY]: bKp.publicKey.export({ type: 'spki', format: 'pem' }) as string, + }; + // the handoff carries the control HMAC secret → create it in the OS temp dir with restrictive 0600 + // perms (owner-only), and clean it up in an OUTER try/finally on EVERY exit path (not only success). + const handoffPath = join(tmpdir(), `tsk-sigkill-handoff-${process.pid}.json`); + const claimExpiresAtMs = Date.now() + HOUR; // STABLE across workers so an advanceEpoch re-claim reads back byte-identical + writeFileSync(handoffPath, JSON.stringify({ SID, CMD, TARGET, ctrlKeyId: CTRL_KEY, ctrlSecretHex: ctrlSecret.toString('hex'), pubKeys, frozen, bReceipt, ctrlUrl: CTRL_URL, redisUrl: REDIS_URL, claimExpiresAtMs }), { mode: 0o600 }); + + const phase = async () => (await ctl.cutover(SID))?.phase ?? null; + const evidence = async () => (await ctl.cutover(SID))?.evidence ?? null; + const crashed = (r: WorkerRun) => !r.out.includes('WORKER_DONE') && (r.out.includes('WORKER_CRASH') ); + + let rtoTotal = 0; + // transition, before-phase, after-phase, crashTxIndex (advance crashes on its 2nd tx = after the Redis claim) + const MATRIX: [string, string, string, number][] = [ + ['bind', 'PREPARING', 'SOURCE_FENCED', 1], + ['advance', 'SOURCE_FENCED', 'FENCED', 2], + ['import', 'FENCED', 'IMPORTING', 1], + ['ready', 'IMPORTING', 'READY', 1], + ['activate', 'READY', 'ACTIVE', 1], + ]; + + try { + for (const [transition, before, after, txIdx] of MATRIX) { + // before advanceEpoch, the parent must have installed the REVOKED control lease (freeze proof). + if (transition === 'advance') await ctl.writeLease({ streamId: SID, leaseId: 'l1', holderNodeId: 'A', epoch: 0, status: 'revoked', grantedMaxExpiryMs: (await ctlNow()) - 5_000, grantCommandId: 'a2' }); + + await check(`${transition}: SIGKILL BEFORE COMMIT rolls back — phase stays ${before}`, async () => { + assert.equal(await phase(), before); + const r = await runWorker(handoffPath, transition, 'before-commit', txIdx); + assert.ok(crashed(r), `child should have crashed before commit; got: ${r.out.slice(-200)}`); + assert.equal(await phase(), before, 'the killed tx left NO partial state — phase unchanged'); + }); + + await check(`${transition}: SIGKILL AFTER COMMIT then a fresh child RESUMES idempotently — phase ${after}, exactly once`, async () => { + const r1 = await runWorker(handoffPath, transition, 'after-commit', txIdx); + assert.ok(crashed(r1), `child should have crashed after commit; got: ${r1.out.slice(-200)}`); + assert.equal(await phase(), after, 'the committed transition is durable across the crash'); + const evAfter = await evidence(); + const t0 = now(); + const r2 = await runWorker(handoffPath, transition, 'none', txIdx); // resume + rtoTotal += now() - t0; + assert.ok(r2.out.includes('WORKER_DONE'), `resume should complete; got: ${r2.out.slice(-200)}`); + assert.equal(await phase(), after, 'idempotent resume — phase advanced EXACTLY once'); + assert.equal(await evidence(), evAfter, 'idempotent resume produced byte-identical signed evidence'); + }); + } + + await check('the promotion reached ACTIVE — B is the live authority after the full crash matrix', async () => { + assert.equal(await phase(), 'ACTIVE'); + const ev = JSON.parse((await evidence())!); + assert.equal(ev.k, 'active/v1'); assert.equal(ev.bSystemId, bSysId); assert.equal(ev.sourceSystemId, aSysId); assert.equal(ev.controlSystemId, cSysId); + }); + + } finally { + // the handoff carries the control HMAC secret — remove it on EVERY exit path (success OR failure). + rmSync(handoffPath, { force: true }); + } + await check('the handoff (test key material) is cleaned up (secret hygiene)', async () => { + assert.equal(existsSync(handoffPath), false, 'handoff file must not remain on disk after the drill'); + }); + console.log('\n# ── measured per-fault RPO / RTO (child-process SIGKILL) ──'); + console.log(' fault: SIGKILL of the cutover worker at EACH phase (before-commit ×5, after-commit ×5)'); + console.log(' RPO : 0 (before-commit → PG rollback, no torn state; after-commit → durable, idempotent resume)'); + console.log(` RTO : ${Math.round(rtoTotal / MATRIX.length)} ms avg (fresh child spawn → phase re-converged), ${rtoTotal} ms total over ${MATRIX.length} phases`); + console.log(`\n# ${passed} PR2c SIGKILL-matrix checks passed`); + 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); }); diff --git a/tsk-cutover-worker.mts b/tsk-cutover-worker.mts new file mode 100644 index 0000000..1dfb9c3 --- /dev/null +++ b/tsk-cutover-worker.mts @@ -0,0 +1,85 @@ +/** + * PR2c child-process crash worker. Performs ONE control-cutover transition against the real control + * PG (+ real Redis for advanceEpoch) and, on demand, dies by a REAL SIGKILL at a chosen point: + * CRASH=before-commit → SIGKILL inside the transition's tx, BEFORE COMMIT (PG rolls the tx back) + * CRASH=after-commit → run the transition to completion (COMMIT), THEN SIGKILL (tests idempotent resume) + * CRASH=none → run to completion and exit 0 (resume / happy path) + * + * All inputs (URLs, keys, receipts) come from a handoff JSON (argv[2]); nothing is invented here. + * Env: TRANSITION=bind|advance|import|ready|activate, CRASH=..., CRASH_TX_INDEX (1-based; advance=2). + */ +import { readFileSync, writeSync } from 'node:fs'; +import { createPublicKey } from 'node:crypto'; +import pg from 'pg'; +import { Redis } from 'ioredis'; +import { + NodePostgresTransactor, HaControlFencing, GuardSigner, assertControlSchemaReady, RedisFencingStore, + type PgExecutor, type GuardKeyResolver, type SourceVerifyKeyResolver, type FenceProof, +} from './packages/server/dist/index.js'; + +const handoff = JSON.parse(readFileSync(process.argv[2], 'utf8')); +const TRANSITION = process.env['TRANSITION'] ?? ''; +const CRASH = process.env['CRASH'] ?? 'none'; +const CRASH_TX_INDEX = Number(process.env['CRASH_TX_INDEX'] ?? '1'); +const { SID, CMD, TARGET, ctrlKeyId, ctrlSecretHex, pubKeys, frozen, bReceipt, ctrlUrl, redisUrl, claimExpiresAtMs } = handoff; + +const ctrlSecret = Buffer.from(ctrlSecretHex, 'hex'); +const ctrlSigner = new GuardSigner(ctrlKeyId, ctrlSecret); +const ctrlResolver: GuardKeyResolver = { resolve: (k) => (k === ctrlKeyId ? ctrlSecret : null) }; +const pub: Record> = {}; +for (const [kid, pem] of Object.entries(pubKeys as Record)) pub[kid] = createPublicKey(pem); +const srcResolver: SourceVerifyKeyResolver = { resolve: (k) => pub[k] ?? null }; + +// A transactor wrapper that fires a REAL SIGKILL just before COMMIT of the armed tx (crash-before-commit). +// It only counts/arms AFTER arm() — so setup txs (schema attest) neither count toward CRASH_TX_INDEX nor crash. +class CrashingTransactor { + private n = 0; private live = false; + constructor(private readonly inner: NodePostgresTransactor) {} + arm() { this.n = 0; this.live = true; } + transaction(fn: (exec: PgExecutor) => Promise, opts?: { signal?: AbortSignal; onBeforeCommit?: (exec: PgExecutor) => Promise }): Promise { + let armed = false; + if (this.live) { this.n += 1; armed = CRASH === 'before-commit' && this.n === CRASH_TX_INDEX; } + const idx = this.n; + return this.inner.transaction(fn, { + ...opts, + onBeforeCommit: async (exec) => { + if (opts?.onBeforeCommit) await opts.onBeforeCommit(exec); + if (armed) { writeSync(1, `WORKER_CRASH_BEFORE_COMMIT tx#${idx} ${TRANSITION}\n`); process.kill(process.pid, 'SIGKILL'); await new Promise(() => {}); } + }, + }); + } +} + +async function main() { + const pool = new pg.Pool({ connectionString: ctrlUrl, max: 6 }); pool.on('error', () => {}); + const inner = new NodePostgresTransactor(pool as never); + const crashTx = new CrashingTransactor(inner); + const tx = crashTx as unknown as NodePostgresTransactor; + const ready = await assertControlSchemaReady(tx as never, 'public'); // attest-only; token binds to `tx` (the wrapper) + crashTx.arm(); // start counting/arming only for the transition's own tx(s) + const ctl = new HaControlFencing(tx as never, ctrlSigner, ctrlResolver, ready, { minClaimRemainingMs: 5_000 }); + + if (TRANSITION === 'bind') { + await ctl.bindSourceFenced(SID, CMD, TARGET, frozen, srcResolver); + } else if (TRANSITION === 'advance') { + const redis = new Redis(redisUrl, { maxRetriesPerRequest: 2, lazyConnect: false }); redis.on('error', () => {}); + const store = new RedisFencingStore(redis, 'tsk:fence:' + SID); + const proof: FenceProof = { safetyMarginMs: 0, claimExpiresAtMs }; + await ctl.advanceEpoch(SID, CMD, TARGET, 'Bnode', store, proof); + await redis.quit().catch(() => {}); + } else if (TRANSITION === 'import') { + await ctl.markImporting(SID, CMD, TARGET); + } else if (TRANSITION === 'ready') { + await ctl.markReady(SID, CMD, TARGET, bReceipt, srcResolver); + } else if (TRANSITION === 'activate') { + await ctl.activate(SID, CMD, TARGET); + } else { + throw new Error('unknown TRANSITION ' + TRANSITION); + } + + if (CRASH === 'after-commit') { writeSync(1, `WORKER_CRASH_AFTER_COMMIT ${TRANSITION}\n`); process.kill(process.pid, 'SIGKILL'); await new Promise(() => {}); } + writeSync(1, `WORKER_DONE ${TRANSITION}\n`); + await pool.end(); + process.exit(0); +} +main().catch((e) => { console.error('WORKER_ERROR', (e as Error).message); process.exit(2); }); diff --git a/tsk-ha-partition-drill.mts b/tsk-ha-partition-drill.mts new file mode 100644 index 0000000..e90cdea --- /dev/null +++ b/tsk-ha-partition-drill.mts @@ -0,0 +1,120 @@ +/** + * PR2c acceptance (#10) — LIVE split-brain partition of the Redis fencing authority (not a crash). + * + * 1 master + 2 replicas + 3 sentinels (quorum 2), min-replicas-to-write 1. The OLD master is left ALIVE + * but network-ISOLATED from the cluster (`docker network disconnect`). We then prove: + * 1. the isolated old master REFUSES writes (min-replicas-to-write: 0 reachable replicas → NOREPLICAS) — + * so a partitioned old writer cannot make progress (no split brain); + * 2. the surviving quorum promotes a new master that is WRITABLE + MONOTONIC (epoch 2 durably claims, + * epoch 1 refused), with the epoch-1 fence intact (RPO = 0); + * 3. on HEAL (reconnect), the old master is demoted to a replica and RECONCILES to the new epoch. + * + * Uses `docker exec` for the isolated node (network-independent) + a Sentinel-backed client (natMap) for + * the survivors. Mechanism-only on one host (processes, not physical failure domains). + * Env: TSK_TEST_SENTINELS, TSK_TEST_SENTINEL_MASTER, TSK_SENTINEL_NATMAP, TSK_SENTINEL_NETWORK, + * TSK_SENTINEL_MASTER_CONTAINER. + */ +import assert from 'node:assert/strict'; +import { execFileSync } from 'node:child_process'; +import { Redis } from 'ioredis'; +import { RedisFencingStore } from './packages/server/dist/index.js'; + +const SENTINELS = (process.env['TSK_TEST_SENTINELS'] ?? '').split(',').map((s) => s.trim()).filter(Boolean) + .map((hp) => { const [host, port] = hp.split(':'); return { host, port: Number(port) }; }); +const MASTER_NAME = process.env['TSK_TEST_SENTINEL_MASTER'] ?? 'tskmaster'; +const NATMAP: Record = {}; +for (const pair of (process.env['TSK_SENTINEL_NATMAP'] ?? '').split(',').map((s) => s.trim()).filter(Boolean)) { + const [internal, external] = pair.split('='); const [host, port] = external.split(':'); + NATMAP[internal] = { host, port: Number(port) }; +} +const NETWORK = process.env['TSK_SENTINEL_NETWORK'] ?? 'tsk-sentinel_tsknet'; +const MASTER_CONTAINER = process.env['TSK_SENTINEL_MASTER_CONTAINER'] ?? 'tsk-sentinel-redis-master-1'; +const MASTER_IP = '172.28.7.10'; +if (SENTINELS.length < 3 || Object.keys(NATMAP).length < 3) throw new Error('need 3 sentinels + natMap for 3 data nodes'); + +const FENCE_KEY = 'tsk:fence:partition/v1'; +let passed = 0; +async function check(name: string, fn: () => Promise) { await fn(); passed++; console.log(` ok - ${name}`); } +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +const now = () => Number(process.hrtime.bigint() / 1_000_000n); +const docker = (...args: string[]) => execFileSync('docker', args, { encoding: 'utf8' }); +const dexec = (...args: string[]) => docker('exec', MASTER_CONTAINER, 'redis-cli', ...args).trim(); +// redis-cli reports an error REPLY (e.g. NOREPLICAS) on stdout/stderr — capture BOTH and never throw on it. +function dexecSafe(...args: string[]): string { + try { return docker('exec', MASTER_CONTAINER, 'redis-cli', ...args).trim(); } + catch (e) { const x = e as { stdout?: Buffer | string; stderr?: Buffer | string; message?: string }; return `${String(x.stdout ?? '')}${String(x.stderr ?? '')}${x.message ?? ''}`.trim(); } +} + +async function waitFor(label: string, fn: () => Promise, timeoutMs: number, everyMs = 250): Promise { + const deadline = now() + timeoutMs; let lastErr: unknown; + while (now() < deadline) { try { return await fn(); } catch (e) { lastErr = e; await sleep(everyMs); } } + throw new Error(`waitFor(${label}) timed out after ${timeoutMs}ms: ${String(lastErr)}`); +} + +async function main() { + console.log('# TSK PR2c Redis LIVE split-brain partition drill (old master isolated ALIVE; new quorum promotes)'); + const sentinelClient = new Redis({ sentinels: SENTINELS, name: MASTER_NAME, role: 'master', natMap: NATMAP, maxRetriesPerRequest: 3, sentinelRetryStrategy: () => 200 }); + sentinelClient.on('error', () => {}); + const store = new RedisFencingStore(sentinelClient, FENCE_KEY, { waitReplicas: 1, waitTimeoutMs: 3_000 }); + const sentinelAdmin = new Redis({ host: SENTINELS[0].host, port: SENTINELS[0].port, maxRetriesPerRequest: 2 }); sentinelAdmin.on('error', () => {}); + const masterAddr = async () => (await sentinelAdmin.call('SENTINEL', 'get-master-addr-by-name', MASTER_NAME) as string[]).join(':'); + + const HOUR = 3_600_000; + await waitFor('initial master reachable', async () => { await sentinelClient.set('tsk:probe', '1'); }, 30_000); + const oldAddr = await masterAddr(); + + await check('durably claim fence epoch 1 (WAIT quorum) before the partition', async () => { + await sentinelClient.del(FENCE_KEY); + assert.equal(await store.claim({ nodeId: 'B', fenceEpoch: 1, expiresAt: Date.now() + HOUR, commandId: 'c1' }), true); + assert.equal((await store.current())!.fenceEpoch, 1); + }); + + let t0 = 0, refusalMs = 0, rtoMs = 0; + await check('PARTITION the old master ALIVE → it REFUSES writes (min-replicas-to-write; no split-brain progress)', async () => { + t0 = now(); + docker('network', 'disconnect', NETWORK, MASTER_CONTAINER); // isolate the master from the cluster; it stays running + // the isolated master loses its replica quorum (acks age past min-replicas-max-lag) → it refuses writes. + await waitFor('isolated old master refuses writes', async () => { + const out = dexecSafe('SET', 'tsk:splitbrain', 'x'); + if (/NOREPLICAS|not enough|good replica/i.test(out)) return true; + throw new Error('old master still accepted a write: ' + out); + }, 25_000, 500); + refusalMs = now() - t0; // sub-time: partition → old master refuses (split-brain prevented) + }); + + let survivor!: RedisFencingStore; + await check('the surviving quorum promotes a NEW master that is DURABLY WRITABLE + MONOTONIC; the epoch-1 fence is intact (RPO=0)', async () => { + await waitFor('sentinel promoted a new master', async () => { const a = await masterAddr(); if (a === oldAddr) throw new Error('not yet'); return a; }, 30_000, 250); + // a RECONNECTING client created AFTER promotion reaches the promoted master; the fence survived (RPO=0). + survivor = new RedisFencingStore(new Redis({ sentinels: SENTINELS, name: MASTER_NAME, role: 'master', natMap: NATMAP, maxRetriesPerRequest: 5, sentinelRetryStrategy: () => 200 }), FENCE_KEY, { waitReplicas: 1, waitTimeoutMs: 3_000 }); + const cur = await waitFor('fence readable on the new master', async () => { const c = await survivor.current(); if (!c) throw new Error('no fence'); return c; }, 30_000, 200); + assert.ok(cur.fenceEpoch === 1 && cur.nodeId === 'B', 'the durably-claimed epoch-1 fence survived the partition (RPO=0)'); + await waitFor('epoch 2 durable claim on the new master', async () => { + const ok2 = await survivor.claim({ nodeId: 'B2', fenceEpoch: 2, expiresAt: Date.now() + HOUR, commandId: 'c2' }); + if (!ok2) throw new Error('not durably writable yet'); + return ok2; + }, 25_000, 250); + rtoMs = now() - t0; // RTO: partition → NEW AUTHORITY DURABLY WRITABLE (successful enforced-WAIT epoch-2 claim) + assert.equal((await survivor.current())!.fenceEpoch, 2); + assert.equal(await survivor.claim({ nodeId: 'Bx', fenceEpoch: 1, expiresAt: Date.now() + HOUR, commandId: 'cx' }), false, 'old epoch refused — no rollback'); + }); + + await check('HEAL the partition → the old master is demoted to a replica and RECONCILES to the new epoch (2)', async () => { + docker('network', 'connect', '--ip', MASTER_IP, NETWORK, MASTER_CONTAINER); + // Sentinel reconfigures the returned old master as a replica of the current master; it re-syncs epoch 2. + await waitFor('old master demoted to replica', async () => { const role = dexec('ROLE').split('\n')[0].trim(); if (role !== 'slave') throw new Error(`role=${role}`); return role; }, 30_000, 500); + await waitFor('old master reconciled the fence to epoch 2', async () => { const v = dexec('GET', FENCE_KEY); if (!v.includes('"fenceEpoch":2')) throw new Error('stale'); return v; }, 20_000, 300); + }); + + console.log('\n# ── measured per-fault RPO / RTO ──'); + console.log(` fault: LIVE NETWORK PARTITION isolating the master (still running) from the cluster`); + console.log(` RPO : 0 (durably WAIT-quorum-claimed epoch-1 fence intact on the promoted master; old master made NO progress)`); + console.log(` RTO : ${rtoMs} ms (partition → NEW AUTHORITY DURABLY WRITABLE: promotion + successful enforced-WAIT epoch-2 claim)`); + console.log(` └ sub-time: isolated old master REFUSED writes at ${refusalMs} ms (split-brain prevented before promotion completed)`); + console.log(`\n# ${passed} PR2c split-brain-partition checks passed`); + + await sentinelClient.quit().catch(() => {}); + await sentinelAdmin.quit().catch(() => {}); + process.exit(0); +} +main().then(() => process.exit(0)).catch((e) => { console.error(e); process.exit(1); }); diff --git a/tsk-ha-sentinel-drill.mts b/tsk-ha-sentinel-drill.mts new file mode 100644 index 0000000..8651eb9 --- /dev/null +++ b/tsk-ha-sentinel-drill.mts @@ -0,0 +1,123 @@ +/** + * PR2c acceptance (#10) — real Redis Sentinel/quorum failover of the fencing authority. + * + * Topology (ci/redis-sentinel/docker-compose.yml): 1 master + 2 replicas + 3 sentinels (quorum 2), + * AOF everysec, min-replicas-to-write 1. This drill runs ON THE HOST (published ports + ioredis natMap) + * and drives a REAL master crash, proving the RedisFencingStore over a Sentinel-backed client: + * 1. claim() ENFORCES a WAIT replica-quorum ACK before success, so a claimed fence SURVIVES an automatic + * Sentinel failover (RPO = 0) — durability is a store property, not a caller-side WAIT; + * 2. the promoted replica serves the SAME fence (no rollback / no stale epoch); + * 3. the fence stays MONOTONIC across failover (a new epoch claims; an old epoch is refused); + * 4. measured RTO = time from master crash to the fence readable on the new master. + * + * Mechanism-only on a single host (processes, not physical failure domains). Env: TSK_TEST_SENTINELS + + * TSK_TEST_SENTINEL_MASTER + TSK_SENTINEL_NATMAP + TSK_SENTINEL_MASTER_PORT. + */ +import assert from 'node:assert/strict'; +import { execFileSync } from 'node:child_process'; +import { Redis } from 'ioredis'; +import { RedisFencingStore } from './packages/server/dist/index.js'; + +const SENTINELS = (process.env['TSK_TEST_SENTINELS'] ?? '').split(',').map((s) => s.trim()).filter(Boolean) + .map((hp) => { const [host, port] = hp.split(':'); return { host, port: Number(port) }; }); +const MASTER_NAME = process.env['TSK_TEST_SENTINEL_MASTER'] ?? 'tskmaster'; +// natMap: internal "ip:6379" (what Sentinel advertises) → host-reachable published address, so the +// sentinel-backed client follows failover from the host. Also gives us direct per-node host connections. +const NATMAP: Record = {}; +for (const pair of (process.env['TSK_SENTINEL_NATMAP'] ?? '').split(',').map((s) => s.trim()).filter(Boolean)) { + const [internal, external] = pair.split('='); + const [host, port] = external.split(':'); + NATMAP[internal] = { host, port: Number(port) }; +} +const REPLICA_PORT = Number(process.env['TSK_SENTINEL_REPLICA_PORT'] ?? '6391'); +const MASTER_CONTAINER = process.env['TSK_SENTINEL_MASTER_CONTAINER'] ?? 'tsk-sentinel-redis-master-1'; +if (SENTINELS.length < 3) throw new Error('TSK_TEST_SENTINELS must list >=3 sentinels (quorum topology)'); +if (Object.keys(NATMAP).length < 3) throw new Error('TSK_SENTINEL_NATMAP must map the 3 data nodes'); + +const FENCE_KEY = 'tsk:fence:sentinel/v1'; +let passed = 0; +async function check(name: string, fn: () => Promise) { await fn(); passed++; console.log(` ok - ${name}`); } +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +const now = () => Number(process.hrtime.bigint() / 1_000_000n); + +async function waitFor(label: string, fn: () => Promise, timeoutMs: number, everyMs = 200): Promise { + const deadline = now() + timeoutMs; + let lastErr: unknown; + while (now() < deadline) { try { return await fn(); } catch (e) { lastErr = e; await sleep(everyMs); } } + throw new Error(`waitFor(${label}) timed out after ${timeoutMs}ms: ${String(lastErr)}`); +} + +async function main() { + console.log('# TSK PR2c Redis Sentinel failover drill (1 master + 2 replicas + 3 sentinels, quorum 2, enforced-WAIT claim)'); + // sentinel-backed client used by the fencing store — auto-follows the master across failover (via natMap). + const sentinelClient = new Redis({ sentinels: SENTINELS, name: MASTER_NAME, role: 'master', natMap: NATMAP, maxRetriesPerRequest: 3, sentinelRetryStrategy: () => 200 }); + sentinelClient.on('error', () => {}); + // ENFORCED durable claim: WAIT for >=1 replica ACK inside claim() before success (RPO=0 is a store property). + const store = new RedisFencingStore(sentinelClient, FENCE_KEY, { waitReplicas: 1, waitTimeoutMs: 3_000 }); + // a DIRECT connection to a sentinel node (SENTINEL admin commands don't route through the master client). + const sentinelAdmin = new Redis({ host: SENTINELS[0].host, port: SENTINELS[0].port, maxRetriesPerRequest: 2 }); sentinelAdmin.on('error', () => {}); + const masterAddr = async () => await sentinelAdmin.call('SENTINEL', 'get-master-addr-by-name', MASTER_NAME) as string[]; + + // wait until Sentinel has a reachable master and the store is usable. + await waitFor('initial master reachable', async () => { await sentinelClient.set('tsk:probe', '1'); }, 30_000); + + const HOUR = 3_600_000; + const replica = new Redis({ host: '127.0.0.1', port: REPLICA_PORT, maxRetriesPerRequest: 2 }); replica.on('error', () => {}); + + await check('durable claim of fence epoch 1 — claim() ENFORCES a WAIT replica-quorum ACK before success', async () => { + await sentinelClient.del(FENCE_KEY); + const ok = await store.claim({ nodeId: 'B', fenceEpoch: 1, expiresAt: Date.now() + HOUR, commandId: 'c1' }); + assert.equal(ok, true, 'claim succeeded => WAIT confirmed >=1 replica durably has it'); + const cur = await store.current(); + assert.ok(cur && cur.fenceEpoch === 1 && cur.nodeId === 'B' && cur.commandId === 'c1' && cur.active === true); + // corroborate on a replica directly (no caller-side WAIT — the store already enforced it). + const raw = await waitFor('replica has fence', async () => { const v = await replica.get(FENCE_KEY); if (!v) throw new Error('not yet'); return v; }, 5_000); + assert.ok(String(raw).includes('"fenceEpoch":1')); + assert.equal(String((await replica.call('ROLE') as unknown[])[0]), 'slave', 'the replica is a slave before failover'); + }); + + const oldMasterAddr = await masterAddr(); + // a RECONNECTING process uses a FRESH Sentinel-backed client created AFTER failover (it does NOT inherit + // the dead socket); this measures data-plane RTO (Sentinel failover + connect), not one long-lived + // socket's slow follow. + let survivor!: RedisFencingStore; + + let t0 = 0, readableMs = 0, rtoMs = 0; + await check('MASTER CRASH (abrupt SIGKILL) → Sentinel promotes a replica; the durably-claimed fence SURVIVES (RPO=0)', async () => { + t0 = now(); + try { execFileSync('docker', ['kill', '-s', 'KILL', MASTER_CONTAINER], { stdio: 'ignore' }); } catch { /* container already gone */ } // ABRUPT crash, deterministic exit + const newMasterAddr = await waitFor('sentinel promotes a new master', async () => { const a = await masterAddr(); if (JSON.stringify(a) === JSON.stringify(oldMasterAddr)) throw new Error('same'); return a; }, 30_000, 250); + assert.notDeepEqual(newMasterAddr, oldMasterAddr, 'Sentinel promoted a new master'); + survivor = new RedisFencingStore(new Redis({ sentinels: SENTINELS, name: MASTER_NAME, role: 'master', natMap: NATMAP, maxRetriesPerRequest: 5, sentinelRetryStrategy: () => 200 }), FENCE_KEY, { waitReplicas: 1, waitTimeoutMs: 3_000 }); + const cur = await waitFor('fence readable on the promoted master', async () => { const c = await survivor.current(); if (!c) throw new Error('no fence yet'); return c; }, 30_000, 200); + readableMs = now() - t0; // sub-time: fault → fence READABLE (read path recovered) + assert.ok(cur.fenceEpoch === 1 && cur.nodeId === 'B' && cur.commandId === 'c1', 'the epoch-1 fence survived the failover unchanged (RPO=0)'); + }); + + await check('the promoted master is DURABLY WRITABLE + MONOTONIC: enforced-WAIT epoch 2 claims, epoch 1 refused (no rollback)', async () => { + await waitFor('epoch 2 durable claim on the new master', async () => { + const ok2 = await survivor.claim({ nodeId: 'B2', fenceEpoch: 2, expiresAt: Date.now() + HOUR, commandId: 'c2' }); // enforced WAIT quorum (throws if uncertain) + if (!ok2) throw new Error('not durably writable yet'); + return ok2; + }, 25_000, 250); + rtoMs = now() - t0; // RTO: fault → NEW AUTHORITY DURABLY WRITABLE (successful enforced-WAIT epoch-2 claim) + assert.equal((await survivor.current())!.fenceEpoch, 2); + const rollback = await survivor.claim({ nodeId: 'Bx', fenceEpoch: 1, expiresAt: Date.now() + HOUR, commandId: 'cx' }); + assert.equal(rollback, false, 'an older epoch cannot reclaim after failover — no split-brain rollback'); + assert.equal((await survivor.current())!.fenceEpoch, 2); + }); + + console.log('\n# ── measured per-fault RPO / RTO ──'); + console.log(` fault: Redis MASTER CRASH (abrupt SIGKILL) → Sentinel quorum (2-of-3) failover, 3 data nodes`); + console.log(` RPO : 0 (claim() enforced a WAIT replica-quorum ACK before success; fence survived, epoch/node/command unchanged)`); + console.log(` RTO : ${rtoMs} ms (fault → NEW AUTHORITY DURABLY WRITABLE: successful enforced-WAIT epoch-2 claim + converged)`); + console.log(` └ sub-time: fence READABLE at ${readableMs} ms (read path recovered before durable writability)`); + console.log(`\n# ${passed} PR2c Sentinel-failover checks passed`); + + await sentinelClient.quit().catch(() => {}); + await sentinelAdmin.quit().catch(() => {}); + await replica.quit().catch(() => {}); + process.exit(0); +} + +main().then(() => process.exit(0)).catch((e) => { console.error(e); process.exit(1); });