diff --git a/.env.example b/.env.example index 532c66f0..e4ade59d 100644 --- a/.env.example +++ b/.env.example @@ -26,6 +26,13 @@ CONSENSUS_TIME=10 # rebroadcast with one bounded aggregate from the block sender. Keep disabled # outside an isolated devnet until convergence and adversarial tests pass. BLOCK_SYNC_AGGREGATION_ENABLED=false +# Aggregation wire/dissemination version: 1 = secretary-only JSON list, +# 2 = committee-partitioned block delivery + bitmap partial aggregates. +BLOCK_SYNC_AGGREGATION_VERSION=2 +# First block height at which the aggregation send path activates. 0 = now. +# Any change to ENABLED or VERSION on a live fleet must ship with a fresh +# future height, identical on every node, so the whole fleet flips together. +BLOCK_SYNC_AGGREGATION_ACTIVATION_HEIGHT=0 # Genesis-state network parameters. Governance proposals can override these # at runtime; these values are the chain's bootstrap defaults. diff --git a/docs/poc/block-sync-aggregation.md b/docs/poc/block-sync-aggregation.md index c5ec8d78..40da1cb2 100644 --- a/docs/poc/block-sync-aggregation.md +++ b/docs/poc/block-sync-aggregation.md @@ -1,5 +1,10 @@ # Block sync aggregation POC +Version 1 (below) proved the linear-call model. Version 2, documented in +[Version 2: bitmap aggregates and partitioned dissemination](#version-2-bitmap-aggregates-and-partitioned-dissemination), +addresses the three follow-ups the v1 review named: aggregate byte size, the +relay trust rule, and mixed-version activation. + ## Purpose This POC tests one narrow change to the existing PoRBFT network path. It does @@ -163,3 +168,263 @@ OmniProtocol connections, cryptographic acknowledgement signatures, or WAN kernel scheduling. The 18.1 MB sender burst at 500 peers also motivates the documented bitmap/tree follow-up even though the burst completed in under one second here. + +## Version 2: bitmap aggregates and partitioned dissemination + +Version 2 is selected with `BLOCK_SYNC_AGGREGATION_VERSION=2` (the default +when aggregation is enabled) and changes two things relative to v1 while +keeping every consensus rule untouched: how acknowledgements are encoded, and +who sends what. + +### Bitmap wire format + +The v2 aggregate replaces the JSON identity list with a bitmap over the +*canonical acknowledgement index*: the block's committed peerlist identities, +normalized to lowercase, deduplicated and sorted. The committed peerlist is +part of the block-hash preimage (`serializeBlockContent` covers +`content.peerlist`), so every node holding a block derives the identical +index locally and no identities travel on the wire: + +```jsonc +{ + "version": 2, + "blockNumber": 12345, + "blockHash": "…", + "peerlistSize": 500, // cross-check against the local index length + "ackBits": "…" // base64, bit i = index entry i acknowledged +} +``` + +At 500 peers the bitmap is 63 bytes (~84 base64 characters), reducing the +aggregate body from 34.4 KB to roughly a quarter of a kilobyte. + +The index deliberately does **not** include `validation_data.signatures`: +the signature map is merged incrementally per node, is outside the block-hash +preimage, and is therefore not guaranteed identical across nodes. Building a +bitmap over it would make the same aggregate decode differently on different +nodes. Signers drawn from the committed peerlist (the normal case — the +committee is drawn from it) are representable; a signer absent from the +committed peerlist simply cannot be acknowledged in v2, which is a liveness +hint loss only. + +Receivers fail closed on decoding: an aggregate is rejected unless its +`peerlistSize` equals the locally derived index length, its base64 payload is +canonical and exactly `ceil(size / 8)` bytes, and every bit beyond the index +is zero. Both wire versions remain admissible on the receive path +indefinitely, so mixed v1/v2 fleets converge. + +### Partitioned block delivery and partial aggregates + +v1 made the secretary (`committee[0]`) the sole publisher of both the block +and the aggregate. That concentrated an `O(N)` byte burst on one node and +created a single point of failure — made worse by the fact that the drawn +committee order is liveness-filtered and view-dependent, so nodes can +disagree about who `committee[0]` even is (zero-publisher rounds). + +In v2 every *signing* committee member publishes: + +- **Block delivery** is partitioned. Peer `p` belongs to the slice + `H(p, blockHash) mod S`, owned by the member at that position of the + *sorted signing committee*. The assignment depends only on the peer + identity, the signing committee and the block hash — never on peerlist + ordering, which differs between nodes — so divergent local views degrade + to duplicate deliveries (deduplicated by the receiver's existing + `handleNewBlock` short-circuits) or missed deliveries (repaired by the + retained fastSync/anti-entropy path, now affecting ~1/S of peers instead + of all of them). Salting with the block hash rotates slice ownership every + block, so a withholding or crashed member cannot starve the same peers + round after round. Only members whose signature is on the block are + excluded from slices: a member that aborted mid-round holds neither the + block nor a signature and stays an ordinary delivery target. Total + deliveries stay `N − S`; per-sender block bytes divide by `S`. +- **Acknowledgement aggregation** is per-member. Each member builds one + partial bitmap aggregate from its own slice's delivery responses, applies + it locally through the admission path, and broadcasts it to every other + node. Receivers admit each partial through the same fail-closed path and + union the results; the union is idempotent and order-independent, so no + merge protocol or extra message type is needed. +- **Race buffering.** A member with a fast slice publishes its partial while + slower slices are still delivering, so partials routinely reach a peer + just before that peer's own block delivery. Receivers buffer aggregates + addressed to `lastBlockNumber + 1` (bounded to 64 entries, 60 s TTL) and + replay them through the same admission path once the block lands, instead + of rejecting them permanently. + +Modeled post-block request counts for a four-member committee: + +```text +legacy: S*(N-S) + N*(N-S) + N*S +v1: (N-S) + (N-1) +v2: (N-S) + S*(N-1) +``` + +| Nodes | Legacy | v1 | v2 | v2 reduction vs legacy | +| ----: | ------: | ----: | ----: | ---------------------: | +| 6 | 44 | 7 | 22 | 50.0% | +| 20 | 464 | 35 | 92 | 80.2% | +| 50 | 2,684 | 95 | 242 | 91.0% | +| 500 | 251,984 | 995 | 2,492 | 99.0% | + +v2 spends more requests than v1 (each of the `S` members broadcasts its own +partial) and buys three things with them: aggregate bytes shrink by two +orders of magnitude, per-sender block-delivery load divides by `S`, and no +single node is a required publisher for either the block or the hints. + +Measured with the same loopback emulator, knobs and pass criteria as the v1 +run (2026-08-19, five bursts per size, 20–100 ms jitter, 5% slow peers, 5% +transient failures, bounded retries; `--aggregate-version=2`): + +| Peers | Legacy calls | v2 calls | Reduction | Mean burst | Partial aggregate | Total wire/block | Peak RSS | Event-loop p99 | +| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| 100 | 10,384 | 492 | 95.3% | 573 ms | 119 B | 0.08 MB | 95 MB | 4.0 ms | +| 250 | 63,484 | 1,242 | 98.0% | 754 ms | 143 B | 0.22 MB | 105 MB | 5.9 ms | +| 500 | 251,984 | 2,492 | 99.0% | 1,294 ms | 183 B | 0.54 MB | 206 MB | 28.0 ms | + +Observed calls matched `estimatePostBlockTraffic(n, s, true, 2)` exactly at +every size; every partial was admitted everywhere, every non-signer received +the block exactly once, and every receiver's accepted union was exactly the +committed peerlist minus itself. Total post-block wire volume at 500 peers +drops from 18.1 MB (v1) to 0.54 MB — roughly 34× less — while the burst +stays under 1.3 s through a single emulator process. The same v1 safety +cases plus four v2-specific rejection cases (non-signer sender, wrong block, +index-size mismatch, trailing bitmap bit) all fail closed. One integration +requirement surfaced by the emulator: a committee member never receives its +own partial over the wire, so the builder must apply its own partial locally +through the same admission path — `broadcastNewBlock` does exactly that +before publishing. + +### Sync-hint monotonicity + +v1's `applySyncAggregate` mutated live `Peer.sync` hints without the +monotonicity guard the legacy `updateSyncData` path gets from +`PeerManager.addPeer`, so an aggregate for an older (still locally verified) +block could regress a fresher hint. Those hints feed sync-source selection, +the forge quorum pre-check and the network-ahead veto. v2 adds the guard: an +aggregate never lowers `peer.sync.block`; a same-height aggregate may still +correct a conflicting hash to the locally verified one. + +### Trust rule (v1 review gate 1) + +The aggregate relay operates under the following rule, which replaces the +per-acknowledgement signature requirement for this feature *only*: + +> A sender that signed block `B` (present in the receiver's stored +> `validation_data.signatures` for `B`) is trusted to relay *liveness-only* +> acknowledgement observations about `B`. + +This is sound because admission is fail-closed on every axis that could +affect consensus or state: + +- an aggregate is only accepted for a block the receiver itself already + verified and stored — it can never announce, deliver, or advance a chain; +- it can only reference identities already committed in that block's + hash-covered peerlist (or its signers) *and* already known locally — it + can never add a peer; +- it never marks a peer online and never regresses a hint (monotonicity + guard) — it can only advance a known peer's sync pointer to a block the + receiver holds; +- the sender must hold a consensus identity that signed the block, verified + via the authenticated RPC envelope. + +The residual power of a malicious block signer is to *falsely advance* a +peer's hint to the current block. The reachable consequences are bounded to +liveness: a stale sync-source choice (retried against the next peer), or an +inflated forge-quorum pre-check that lets a round start which then fails to +gather real signatures. No state, no finality, and no block content can be +influenced, because none of the consumers of `peer.sync` feed block +validation. Appendix A documents the upgrade path to per-acknowledgement +detached signatures if a later review rejects this rule. + +### Activation protocol (v1 review gate 2) + +`BLOCK_SYNC_AGGREGATION_ACTIVATION_HEIGHT` coordinates a mixed-version +fleet. The send-side behaviour switch is evaluated per block height, not at +process start: + +1. Roll out the build with `BLOCK_SYNC_AGGREGATION_ENABLED=true`, + `BLOCK_SYNC_AGGREGATION_VERSION=2` and an agreed future activation + height `H` on every node, restarting nodes at operator convenience. +2. Below `H` every node keeps byte-identical legacy behaviour (all-member + block broadcast, receiver status rebroadcasts). +3. From the first block with `number >= H`, upgraded committee members + switch to partitioned delivery and partial aggregates in the same round. +4. Receivers admit v1 and v2 aggregates regardless of height, and legacy + `updateSyncData` handlers remain in place, so laggard nodes degrade to + the anti-entropy path instead of diverging. Nodes that never received + the flag keep full legacy behaviour and interoperate. + +Rollback is the reverse: unset the flag (or raise `H`) and restart; there is +no persistent state to migrate because aggregates only touch in-memory sync +hints. + +**Rollout invariant:** any change to `BLOCK_SYNC_AGGREGATION_ENABLED` or +`BLOCK_SYNC_AGGREGATION_VERSION` on a live fleet must ship with a fresh +activation height beyond the expected rollout completion. Nodes disagreeing +about the active mode inside one committee cannot corrupt anything (all +paths fail closed and receivers admit both versions), but a mixed committee +where a v2 member delivers only its slice while v1 members expect a sole +secretary publisher leaves some peers waiting on fastSync for the whole +mismatch window. The activation height exists precisely so that window never +opens; it is per-node configuration, so operators — not the protocol — +enforce agreement today (see finding 6). + +### Review findings recorded for maintainers + +Discovered while building v2; pre-existing on `stabilisation` unless noted: + +1. `BroadcastManager.handleUpdatePeerSyncData` returns + `peerman.addPeer(peer) ? 200 : 400`, but `addPeer` returns a + `[boolean, string]` tuple — always truthy, so rejected peers still get + 200 "Sync data updated". +2. The RPC auth header signs only `sha256(identity:timestamp)`; request + bodies are unbound. Before the `nonceEnforcement` fork activates, a + captured header is a replayable bearer token, and `gcr_routine` is not in + `PROTECTED_ENDPOINTS`. The aggregate path fails closed on content, but + body-binding signatures are worth revisiting network-wide. +3. Hello gossip piggybacks on `isNetworkAhead('mainLoop')` and messages + every known peer roughly every 2 s — an O(N²)-per-interval background + load that dwarfs the post-block burst at scale and is the natural next + target after this change. +4. `peerGossip`/`peerRoutine` are dead code in the live loop (commented out + of `mainLoopCycle`), although POC comments name peer gossip as a recovery + path; actual recovery flows through hello + fastSync. +5. The duplicate-block short-circuit in `manageGCRRoutines` returned a bare + string without `syncData`, so already-synced peers could not be counted + in aggregates (fixed in this branch by returning the standard + `handleNewBlock` response shape). +6. `BLOCK_SYNC_AGGREGATION_ACTIVATION_HEIGHT` is per-node configuration + with no in-band enforcement; committing it as a governance/network + parameter would remove the operator-coordination requirement. +7. Aggregate admission trusts the locally merged + `validation_data.signatures` map, which is not hash-covered; a valid + partial can be rejected 403 on a receiver that has not yet merged the + sender's signature. Fail-closed, hint-only loss, repaired by + anti-entropy; verifying the sender's signature cryptographically against + the block hash would remove the race at the cost of a worker-pool verify + per aggregate. +8. `updateSyncAggregate` (like `updateSyncData`) has no per-peer rate + limit; a block signer can replay valid aggregates cheaply. Admission work + is bounded (one DB fetch plus one sort of ≤ committed-peerlist size), but + a token bucket on the route would close the amplification avenue. + +### Appendix A: detached acknowledgement signatures (designed, not enabled) + +If a future review requires cryptographic acknowledgements instead of the +trust rule, the primitives already exist: recipients would sign with the +same worker pool that produces block signatures +(`TxValidatorPool.getInstance().sign`) and return the signature alongside +`syncData` in the `syncNewBlock` response; publishers would verify before +setting a bit and attach the signature set; receivers would verify per bit. + +Two constraints for that design, learned here: + +- **Domain separation is mandatory.** Block signatures sign the raw block + hash bytes (`createBlock.ts`). An acknowledgement must sign a + domain-separated message such as `demos-sync-ack:v1::` — + signing the bare hash would make every acknowledgement a forgeable block + signature and vice versa. +- **Bytes revert to O(N).** 500 ed25519 signatures ≈ 32 KB — the same order + as the v1 JSON aggregate — plus ~500 verifications per receiver per + block. A practical middle ground is carrying the signature set only on + demand (audit pull) or sampling; a compact certificate needs a BLS-class + scheme, which is not in the dependency tree today. diff --git a/src/config/defaults.ts b/src/config/defaults.ts index 926f87af..f9a2b589 100644 --- a/src/config/defaults.ts +++ b/src/config/defaults.ts @@ -68,6 +68,8 @@ export const DEFAULT_CONFIG: AppConfig = { blockWatchdogEnabled: false, blockWatchdogTimeoutSeconds: 600, blockSyncAggregationEnabled: false, + blockSyncAggregationVersion: 2, + blockSyncAggregationActivationHeight: 0, }, tlsnotary: { diff --git a/src/config/envKeys.ts b/src/config/envKeys.ts index 87685062..ad88daee 100644 --- a/src/config/envKeys.ts +++ b/src/config/envKeys.ts @@ -54,6 +54,9 @@ export const EnvKey = { BLOCK_WATCHDOG_ENABLED: "BLOCK_WATCHDOG_ENABLED", BLOCK_WATCHDOG_TIMEOUT_SECONDS: "BLOCK_WATCHDOG_TIMEOUT_SECONDS", BLOCK_SYNC_AGGREGATION_ENABLED: "BLOCK_SYNC_AGGREGATION_ENABLED", + BLOCK_SYNC_AGGREGATION_VERSION: "BLOCK_SYNC_AGGREGATION_VERSION", + BLOCK_SYNC_AGGREGATION_ACTIVATION_HEIGHT: + "BLOCK_SYNC_AGGREGATION_ACTIVATION_HEIGHT", // --- TLSNotary --- TLSNOTARY_ENABLED: "TLSNOTARY_ENABLED", diff --git a/src/config/loader.ts b/src/config/loader.ts index 96feeb74..ac64ecda 100644 --- a/src/config/loader.ts +++ b/src/config/loader.ts @@ -44,6 +44,19 @@ function envBool(key: string, fallback: boolean): boolean { return raw === "true" || raw === "1" } +function envAggregationVersion(fallback: 1 | 2): 1 | 2 { + const raw = process.env[EnvKey.BLOCK_SYNC_AGGREGATION_VERSION] + if (raw === undefined || raw === "") return fallback + if (raw === "1") return 1 + if (raw === "2") return 2 + // A silently coerced version would produce divergent dissemination + // behaviour inside one committee, so refuse anything unrecognised. + console.warn( + `Invalid ${EnvKey.BLOCK_SYNC_AGGREGATION_VERSION}="${raw}" (expected 1 or 2); using ${fallback}`, + ) + return fallback +} + function envList(key: string, fallback: string[] = []): string[] { const raw = process.env[key] if (!raw) return fallback @@ -163,6 +176,16 @@ export function loadConfig(): Readonly { EnvKey.BLOCK_SYNC_AGGREGATION_ENABLED, d.core.blockSyncAggregationEnabled, ), + blockSyncAggregationVersion: envAggregationVersion( + d.core.blockSyncAggregationVersion, + ), + blockSyncAggregationActivationHeight: Math.max( + 0, + envInt( + EnvKey.BLOCK_SYNC_AGGREGATION_ACTIVATION_HEIGHT, + d.core.blockSyncAggregationActivationHeight, + ), + ), }, tlsnotary: { diff --git a/src/config/types.ts b/src/config/types.ts index f699a23c..830f67ad 100644 --- a/src/config/types.ts +++ b/src/config/types.ts @@ -71,6 +71,20 @@ export interface CoreConfig { * block-signer aggregate. Disabled by default until multi-node validation. */ blockSyncAggregationEnabled: boolean + /** + * Aggregate wire/dissemination version when aggregation is enabled. + * 1 = secretary-only publisher with a JSON identity list. + * 2 = committee-partitioned block delivery plus a bitmap partial + * aggregate from every committee member. Receivers accept both. + */ + blockSyncAggregationVersion: 1 | 2 + /** + * First block height at which the aggregation send path activates. + * Below it nodes keep the legacy broadcast behaviour even when the flag + * is on, so a mixed-version fleet flips together at one coordinated + * block. 0 activates immediately. + */ + blockSyncAggregationActivationHeight: number rpcFee: number networkFee: number /** Per-tx burn — sat/lamport-style integer for now. diff --git a/src/libs/communications/broadcastManager.ts b/src/libs/communications/broadcastManager.ts index b144dfc1..697b838e 100644 --- a/src/libs/communications/broadcastManager.ts +++ b/src/libs/communications/broadcastManager.ts @@ -11,9 +11,13 @@ import { Mutex } from "async-mutex" import { Config } from "src/config" import { MetricsService } from "src/features/metrics/MetricsService" import { - BlockSyncAggregateV1, + BlockSyncAggregate, admitSyncAggregate, + blockDeliveryPartition, buildSyncAggregate, + buildSyncAggregateV2, + shouldPublishBlock, + syncAggregationActiveAt, } from "./syncAggregation" /** @@ -21,25 +25,104 @@ import { * Manages the broadcasting of messages to the network */ export class BroadcastManager { - private static syncAggregationEnabled(): boolean { - return Config.getInstance().core.blockSyncAggregationEnabled + /** + * Post-block dissemination mode for a given block height. + * 0 = legacy broadcast, 1 = secretary aggregate POC, 2 = partitioned + * bitmap aggregation. Receivers admit aggregates regardless of mode. + */ + private static syncAggregationModeFor(blockNumber: number): 0 | 1 | 2 { + const core = Config.getInstance().core + if ( + !syncAggregationActiveAt( + core.blockSyncAggregationEnabled, + core.blockSyncAggregationActivationHeight, + blockNumber, + ) + ) { + return 0 + } + return core.blockSyncAggregationVersion + } + + /** + * Post-consensus publication entry point. Every committee member calls + * this; the active mode decides who actually sends what. + */ + static async publishBlock(block: Block, committeeIdentities: string[]) { + const mode = this.syncAggregationModeFor(block.number) + if ( + mode === 1 && + !shouldPublishBlock( + true, + getSharedState.publicKeyHex, + committeeIdentities, + ) + ) { + // Version 1 keeps the POC's single designated publisher. + return false + } + return this.broadcastNewBlock(block, committeeIdentities) } /** * Broadcasts a new block to the network * * @param block The new block to broadcast + * @param committeeIdentities Current committee; only used by mode 2 to + * derive this node's deterministic delivery slice. */ - static async broadcastNewBlock(block: Block) { + static async broadcastNewBlock( + block: Block, + committeeIdentities: string[], + ) { + const mode = this.syncAggregationModeFor(block.number) const peerlist = PeerManager.getInstance().getPeers() // filter by block signers - const peers = peerlist.filter( + let peers = peerlist.filter( peer => block.validation_data.signatures[peer.identity] == undefined, ) - if (peers.length === 0) { + if (mode === 2) { + // Each SIGNING committee member delivers only its deterministic + // slice. A member that aborted mid-round holds no signature (and + // no block), so it must stay a delivery target rather than a + // deliverer. The assignment depends on the peer identity, the + // signing committee and the block hash (rotating slice ownership + // every block), so divergent local peer views cost at most + // duplicate or missed deliveries, both repaired by dedupe and + // anti-entropy. + const signerIds = new Set( + Object.keys(block.validation_data.signatures ?? {}).map( + identity => identity.toLowerCase(), + ), + ) + const signingCommittee = committeeIdentities.filter(identity => + signerIds.has(identity.toLowerCase()), + ) + const slice = blockDeliveryPartition( + getSharedState.publicKeyHex, + signingCommittee, + peers.map(peer => peer.identity), + block.hash, + ) + if (slice === null) { + log.warning( + `[broadcastNewBlock] Asked to publish block ${block.number} without a partition slot (not a signing committee member)`, + ) + } + const allowed = new Set( + (slice ?? []).map(identity => identity.toLowerCase()), + ) + peers = peers.filter(peer => + allowed.has(peer.identity.toLowerCase()), + ) + } + + // Mode 2 still publishes its partial aggregate (it carries our own + // acknowledgement) even when the delivery slice is empty. + if (peers.length === 0 && mode !== 2) { return } @@ -80,12 +163,23 @@ export class BroadcastManager { .map(r => r.value) const successful = responses.filter(res => res.result.result === 200) - if (this.syncAggregationEnabled()) { - const aggregate = buildSyncAggregate( - block, - getSharedState.publicKeyHex, - responses, - ) + if (mode !== 0) { + // Mode 2 encodes acknowledgements as a bitmap over the block's + // hash-committed peerlist; blocks without a usable committed + // peerlist fall back to the bounded version-1 identity list. + const aggregate: BlockSyncAggregate = + (mode === 2 + ? buildSyncAggregateV2( + block, + getSharedState.publicKeyHex, + responses, + ) + : null) ?? + buildSyncAggregate( + block, + getSharedState.publicKeyHex, + responses, + ) // Apply the same aggregate locally before publishing it so the // block sender and recipients converge through one code path. this.applySyncAggregate( @@ -211,11 +305,19 @@ export class BroadcastManager { const peer = peerman.getPeer(sender) const res = await syncBlock(block, peer) + if (res) { + // Partial aggregates that raced this block's delivery were + // buffered; replay them now through the same admission path. + this.drainPendingSyncAggregates(block) + } + // Legacy behaviour fans each recipient's status back out to every - // peer. The POC returns the same syncData in this response and lets - // the block sender publish one aggregate instead. Existing hello and - // peer-gossip routines remain the anti-entropy recovery path. - if (!this.syncAggregationEnabled()) { + // peer. The aggregation path returns the same syncData in this + // response and lets the block deliverer publish an aggregate + // instead. Existing hello and peer-gossip routines remain the + // anti-entropy recovery path. Gated per block height so a fleet + // waiting on an activation height keeps legacy semantics below it. + if (this.syncAggregationModeFor(block.number) === 0) { await this.broadcastOurSyncData("receiver_post_block") } @@ -291,7 +393,7 @@ export class BroadcastManager { } /** Publish one compact acknowledgement set for a consensus-approved block. */ - static async broadcastSyncAggregate(aggregate: BlockSyncAggregateV1) { + static async broadcastSyncAggregate(aggregate: BlockSyncAggregate) { const peerlist = PeerManager.getInstance() .getPeers() .filter( @@ -337,6 +439,46 @@ export class BroadcastManager { ).length } + /** + * Partitioned committee members publish their partial aggregates as soon + * as their own slice settles, so a partial routinely reaches a peer + * moments before that peer's own block delivery. Buffering those + * next-block aggregates briefly (instead of rejecting them outright) + * preserves the acknowledgements they carry; each entry is replayed + * through the same fail-closed admission path once the block lands. + */ + private static readonly MAX_PENDING_SYNC_AGGREGATES = 64 + private static readonly PENDING_SYNC_AGGREGATE_TTL_MS = 60_000 + private static pendingSyncAggregates: { + sender: string + value: unknown + blockNumber: number + receivedAt: number + }[] = [] + + private static prunePendingSyncAggregates() { + const cutoff = Date.now() - this.PENDING_SYNC_AGGREGATE_TTL_MS + this.pendingSyncAggregates = this.pendingSyncAggregates.filter( + entry => + entry.receivedAt >= cutoff && + entry.blockNumber > getSharedState.lastBlockNumber, + ) + } + + /** Replay buffered aggregates for a block that just finished syncing. */ + private static drainPendingSyncAggregates(block: Block) { + const matching = this.pendingSyncAggregates.filter( + entry => entry.blockNumber === block.number, + ) + this.pendingSyncAggregates = this.pendingSyncAggregates.filter( + entry => entry.blockNumber !== block.number, + ) + for (const entry of matching) { + this.applySyncAggregate(entry.sender, entry.value, block) + } + this.prunePendingSyncAggregates() + } + /** * Apply a bounded block-signer observation. This POC deliberately treats * the aggregate as a liveness hint: it can only advance known peers to an @@ -357,6 +499,32 @@ export class BroadcastManager { typeof (value as { blockNumber?: unknown }).blockNumber === "number" ? (value as { blockNumber: number }).blockNumber : -1 + if ( + Number.isSafeInteger(rawBlockNumber) && + rawBlockNumber === getSharedState.lastBlockNumber + 1 + ) { + this.prunePendingSyncAggregates() + if ( + this.pendingSyncAggregates.length < + this.MAX_PENDING_SYNC_AGGREGATES + ) { + this.pendingSyncAggregates.push({ + sender, + value, + blockNumber: rawBlockNumber, + receivedAt: Date.now(), + }) + return { + result: 200, + message: "Sync aggregate buffered until the block arrives", + accepted: 0, + syncData: PeerManager.getInstance().ourSyncDataString, + } + } + log.debug( + "[handleSyncAggregate] Pending aggregate buffer full, dropping next-block aggregate", + ) + } const blockNumber = Number.isSafeInteger(rawBlockNumber) && rawBlockNumber >= 0 && @@ -402,6 +570,18 @@ export class BroadcastManager { .getPeers() .find(peer => peer.identity.toLowerCase() === identity) if (!existing) continue + // Monotonicity: an aggregate for an older (locally verified) + // block must never regress a fresher hint. The legacy + // updateSyncData path enforces this inside PeerManager.addPeer; + // this path mutates live Peer objects directly, so it guards + // here. Same-height aggregates may still correct a conflicting + // hash to the locally verified one. + if ( + typeof existing.sync.block === "number" && + existing.sync.block > block.number + ) { + continue + } const changed = !existing.sync.status || existing.sync.block !== block.number || diff --git a/src/libs/communications/syncAggregation.test.ts b/src/libs/communications/syncAggregation.test.ts index 97e12483..83325519 100644 --- a/src/libs/communications/syncAggregation.test.ts +++ b/src/libs/communications/syncAggregation.test.ts @@ -1,15 +1,26 @@ import { describe, expect, test } from "bun:test" import { + MAX_SYNC_AGGREGATE_BITMAP_BYTES, MAX_SYNC_AGGREGATE_IDENTITIES, MAX_SYNC_AGGREGATE_IDENTITY_LENGTH, admitSyncAggregate, + blockDeliveryPartition, buildSyncAggregate, + buildSyncAggregateV2, + canonicalAckIndex, + decodeAckBits, + encodeAckBits, estimatePostBlockTraffic, isBlockSyncAggregateV1, + isBlockSyncAggregateV2, + partitionIndexFor, shouldPublishBlock, + syncAggregationActiveAt, syncDataMatchesBlock, } from "./syncAggregation" +type V2SourceBlock = Parameters[0] + describe("block sync aggregation", () => { test("collects only exact-block acknowledgements deterministically", () => { const aggregate = buildSyncAggregate( @@ -269,3 +280,573 @@ describe("block sync aggregation", () => { }, ) }) + +describe("block sync aggregation v2", () => { + test.each([1, 8, 9, 500])( + "round-trips a %i-bit acknowledgement bitmap losslessly", + size => { + const flags = Array.from( + { length: size }, + (_, index) => index % 3 === 0, + ) + + expect(decodeAckBits(encodeAckBits(flags), size)).toEqual(flags) + }, + ) + + test("accepts the maximum bitmap and rejects every non-canonical payload", () => { + const maxSize = MAX_SYNC_AGGREGATE_BITMAP_BYTES * 8 + const maxFlags = Array.from( + { length: maxSize }, + (_, index) => index % 7 === 0, + ) + expect(decodeAckBits(encodeAckBits(maxFlags), maxSize)).toEqual( + maxFlags, + ) + + // Wrong byte length: two encoded bytes against a one-byte expectation. + expect(decodeAckBits(encodeAckBits(new Array(16).fill(true)), 8)).toBe( + null, + ) + // Set bit beyond expectedSize: bit 7 is valid at size 8, junk at 7. + expect(decodeAckBits("gA==", 8)).toEqual([ + false, + false, + false, + false, + false, + false, + false, + true, + ]) + expect(decodeAckBits("gA==", 7)).toBe(null) + // Junk base64: illegal alphabet and truncated quantum. + expect(decodeAckBits("####", 4)).toBe(null) + expect(decodeAckBits("AQA", 4)).toBe(null) + // Non-canonical re-encoding: "AR==" decodes to the same byte as + // "AQ==" but was not produced by the canonical encoder. + expect(decodeAckBits("AQ==", 1)).toEqual([true]) + expect(decodeAckBits("AR==", 1)).toBe(null) + // Out-of-range expected sizes. + expect(decodeAckBits(encodeAckBits([true]), 0)).toBe(null) + expect(decodeAckBits(encodeAckBits(maxFlags), maxSize + 1)).toBe(null) + }) + + test("derives the canonical ack index from both committed peerlist shapes", () => { + const index = canonicalAckIndex({ + number: 7, + hash: "block-7", + content: { + peerlist: [ + "PEER-B", + { identity: "peer-a" }, + "peer-b", + { identity: "PEER-A" }, + "y".repeat(MAX_SYNC_AGGREGATE_IDENTITY_LENGTH), + "x".repeat(MAX_SYNC_AGGREGATE_IDENTITY_LENGTH + 1), + "", + { notIdentity: "peer-c" }, + 42, + ], + }, + }) + + expect(index).toEqual([ + "peer-a", + "peer-b", + "y".repeat(MAX_SYNC_AGGREGATE_IDENTITY_LENGTH), + ]) + expect(canonicalAckIndex({ number: 7, hash: "block-7" })).toEqual([]) + expect( + canonicalAckIndex({ + number: 7, + hash: "block-7", + content: { peerlist: "not-an-array" }, + }), + ).toEqual([]) + }) + + test("builds the exact bitmap wire object from delivery responses", () => { + const block = { + number: 7, + hash: "block-7", + content: { + peerlist: [ + "peer-c", + { identity: "PEER-A" }, + "peer-b", + { identity: "sender-node" }, + ], + }, + } as unknown as V2SourceBlock + + // Canonical index: peer-a(0) peer-b(1) peer-c(2) sender-node(3). + // Sender bit 3 plus acked bit 1 = byte 0b00001010 = base64 "Cg==". + const aggregate = buildSyncAggregateV2(block, "SENDER-NODE", [ + { + pubkey: "PEER-B", + result: { + result: 200, + response: { syncData: "1:7:block-7" }, + }, + }, + { + pubkey: "peer-c", + result: { + result: 200, + response: { syncData: "1:6:block-6" }, + }, + }, + { + pubkey: "outside-the-index", + result: { + result: 200, + response: { syncData: "1:7:block-7" }, + }, + }, + ]) + + expect(aggregate).toEqual({ + version: 2, + blockNumber: 7, + blockHash: "block-7", + peerlistSize: 4, + ackBits: "Cg==", + }) + expect(isBlockSyncAggregateV2(aggregate)).toBe(true) + expect( + buildSyncAggregateV2( + { + number: 7, + hash: "block-7", + content: { peerlist: [] }, + } as unknown as V2SourceBlock, + "SENDER-NODE", + [], + ), + ).toBe(null) + }) + + test("validates the version-2 wire shape at every boundary", () => { + const maxAckBitsLength = + 4 * Math.ceil(MAX_SYNC_AGGREGATE_BITMAP_BYTES / 3) + const aggregate = { + version: 2, + blockNumber: 42, + blockHash: "block-42", + peerlistSize: 4, + ackBits: "Cg==", + } + + expect(isBlockSyncAggregateV2(aggregate)).toBe(true) + expect(isBlockSyncAggregateV2(null)).toBe(false) + expect(isBlockSyncAggregateV2([])).toBe(false) + expect(isBlockSyncAggregateV2({ ...aggregate, version: 1 })).toBe(false) + expect(isBlockSyncAggregateV2({ ...aggregate, version: 3 })).toBe(false) + expect(isBlockSyncAggregateV2({ ...aggregate, blockNumber: -1 })).toBe( + false, + ) + expect( + isBlockSyncAggregateV2({ ...aggregate, blockNumber: 2 ** 53 }), + ).toBe(false) + expect(isBlockSyncAggregateV2({ ...aggregate, blockHash: "" })).toBe( + false, + ) + expect(isBlockSyncAggregateV2({ ...aggregate, peerlistSize: 0 })).toBe( + false, + ) + expect( + isBlockSyncAggregateV2({ + ...aggregate, + peerlistSize: MAX_SYNC_AGGREGATE_BITMAP_BYTES * 8, + }), + ).toBe(true) + expect( + isBlockSyncAggregateV2({ + ...aggregate, + peerlistSize: MAX_SYNC_AGGREGATE_BITMAP_BYTES * 8 + 1, + }), + ).toBe(false) + expect( + isBlockSyncAggregateV2({ + ...aggregate, + ackBits: "A".repeat(maxAckBitsLength), + }), + ).toBe(true) + expect( + isBlockSyncAggregateV2({ + ...aggregate, + ackBits: "A".repeat(maxAckBitsLength + 1), + }), + ).toBe(false) + }) + + test("admits a bitmap aggregate excluding the local and unknown identities", () => { + const block = { + number: 42, + hash: "block-42", + validation_data: { signatures: { secretary: {} } }, + content: { + peerlist: ["secretary", "peer-a", "peer-b", "unknown-peer"], + }, + } + // Index: peer-a(0) peer-b(1) secretary(2) unknown-peer(3); all four + // bits set = byte 0b00001111 = base64 "Dw==". + const aggregate = { + version: 2, + blockNumber: 42, + blockHash: "block-42", + peerlistSize: 4, + ackBits: "Dw==", + } + + expect( + admitSyncAggregate(aggregate, block, "SECRETARY", "peer-b", [ + "secretary", + "peer-a", + "peer-b", + ]), + ).toEqual({ + ok: true, + acceptedPeerIds: ["peer-a", "secretary"], + }) + }) + + test("rejects bitmap aggregates for the wrong chain, sender, index or bits", () => { + const block = { + number: 42, + hash: "block-42", + validation_data: { signatures: { secretary: {} } }, + content: { + peerlist: ["secretary", "peer-a", "peer-b", "unknown-peer"], + }, + } + const aggregate = { + version: 2, + blockNumber: 42, + blockHash: "block-42", + peerlistSize: 4, + ackBits: "Dw==", + } + const knownPeers = ["secretary", "peer-a", "peer-b"] + + expect( + admitSyncAggregate( + aggregate, + { ...block, hash: "different-block" }, + "secretary", + "local", + knownPeers, + ), + ).toEqual({ + ok: false, + status: 400, + message: "Sync aggregate does not match the local chain", + }) + expect( + admitSyncAggregate( + aggregate, + { ...block, number: 43 }, + "secretary", + "local", + knownPeers, + ), + ).toEqual({ + ok: false, + status: 400, + message: "Sync aggregate does not match the local chain", + }) + expect( + admitSyncAggregate(aggregate, block, "peer-a", "local", knownPeers), + ).toEqual({ + ok: false, + status: 403, + message: "Sync aggregate sender did not sign the block", + }) + expect( + admitSyncAggregate( + { ...aggregate, peerlistSize: 5 }, + block, + "secretary", + "local", + knownPeers, + ), + ).toEqual({ + ok: false, + status: 400, + message: "Sync aggregate peerlist index mismatch", + }) + // Byte 0b00010000 = "EA==" sets only bit 4, past the 4-entry index. + expect( + admitSyncAggregate( + { ...aggregate, ackBits: "EA==" }, + block, + "secretary", + "local", + knownPeers, + ), + ).toEqual({ + ok: false, + status: 400, + message: "Invalid sync aggregate bitmap", + }) + }) + + test("admits a version-1 aggregate identically to its bitmap equivalent", () => { + const block = { + number: 42, + hash: "block-42", + validation_data: { signatures: { secretary: {} } }, + content: { + peerlist: ["secretary", "peer-a", "peer-b", "unknown-peer"], + }, + } + const knownPeers = ["secretary", "peer-a", "peer-b"] + const v2Admission = admitSyncAggregate( + { + version: 2, + blockNumber: 42, + blockHash: "block-42", + peerlistSize: 4, + ackBits: "Dw==", + }, + block, + "secretary", + "peer-b", + knownPeers, + ) + const v1Admission = admitSyncAggregate( + { + version: 1, + blockNumber: 42, + blockHash: "block-42", + syncedPeerIds: ["peer-a", "peer-b", "secretary", "unknown-peer"], + }, + block, + "secretary", + "peer-b", + knownPeers, + ) + + expect(v1Admission).toEqual(v2Admission) + expect(v1Admission).toEqual({ + ok: true, + acceptedPeerIds: ["peer-a", "secretary"], + }) + }) + + test("assigns a stable in-range partition slot for hex and non-hex identities", () => { + expect(partitionIndexFor("0xabcdef12", 4)).toBe(2) + expect(partitionIndexFor("0xABCDEF12", 4)).toBe( + partitionIndexFor("0xabcdef12", 4), + ) + expect(partitionIndexFor("not-hex-identity", 4)).toBe( + partitionIndexFor("not-hex-identity", 4), + ) + expect(partitionIndexFor("not-hex-identity", 4)).toBeLessThan(4) + expect(partitionIndexFor("not-hex-identity", 4)).toBeGreaterThanOrEqual( + 0, + ) + expect(partitionIndexFor("0xabcdef12", 0)).toBe(0) + + const slots = new Set() + for (let i = 0; i < 100; i++) { + const identity = `0x${i.toString(16).padStart(8, "0")}` + const slot = partitionIndexFor(identity, 5) + expect(slot).toBe(i % 5) + slots.add(slot) + } + expect([...slots].sort()).toEqual([0, 1, 2, 3, 4]) + }) + + test("partitions delivery duty disjointly over exactly the non-committee peers", () => { + const peers = Array.from( + { length: 40 }, + (_, index) => `0x${index.toString(16).padStart(40, "0")}`, + ) + const committee = peers.slice(0, 4) + const shuffledCommittee = [...committee] + .reverse() + .map(identity => identity.toUpperCase()) + + const slices = committee.map(member => + blockDeliveryPartition(member, committee, peers), + ) + const shuffledSlices = committee.map(member => + blockDeliveryPartition( + member.toUpperCase(), + shuffledCommittee, + peers, + ), + ) + + expect(shuffledSlices).toEqual(slices) + const union: string[] = [] + for (const slice of slices) { + expect(slice).not.toBe(null) + for (const identity of slice ?? []) { + expect(union).not.toContain(identity) + expect(committee).not.toContain(identity) + union.push(identity) + } + } + expect(union.sort()).toEqual(peers.slice(4)) + + // Peer entries keep their input casing in the returned slice. + const mixedCasePeer = `0x${"AbCdEf00".repeat(5)}` + expect( + blockDeliveryPartition(committee[0], committee, [mixedCasePeer]), + ).toEqual([mixedCasePeer]) + expect( + blockDeliveryPartition( + `0x${"9".repeat(40)}`, + committee, + peers, + ), + ).toBe(null) + }) + + test.each([ + [6, 44, 22, 50], + [20, 464, 92, 80], + [50, 2684, 242, 90.9], + [500, 251984, 2492, 99], + ])( + "%i nodes reduces the version-2 burst from %i to %i requests", + (nodeCount, legacyRequests, v2Requests, minimumReduction) => { + const legacy = estimatePostBlockTraffic(nodeCount, 4, false) + const v2 = estimatePostBlockTraffic(nodeCount, 4, true, 2) + const reduction = + ((legacy.totalRequests - v2.totalRequests) / + legacy.totalRequests) * + 100 + + expect(legacy.totalRequests).toBe(legacyRequests) + expect(v2).toEqual({ + nodeCount, + signerCount: 4, + blockPublishers: 4, + blockDeliveries: nodeCount - 4, + receiverSyncBroadcasts: 0, + senderSyncBroadcasts: 0, + aggregateBroadcasts: 4 * (nodeCount - 1), + totalRequests: v2Requests, + }) + expect(reduction).toBeGreaterThanOrEqual(minimumReduction) + }, + ) + + test("defaults the aggregate model to version 1 unchanged", () => { + const defaulted = estimatePostBlockTraffic(50, 4, true) + + expect(defaulted).toEqual(estimatePostBlockTraffic(50, 4, true, 1)) + expect(defaulted.totalRequests).toBe(95) + }) + + test("activates aggregation only when enabled and at or past the height", () => { + expect(syncAggregationActiveAt(false, 0, 100)).toBe(false) + expect(syncAggregationActiveAt(false, 50, 100)).toBe(false) + expect(syncAggregationActiveAt(true, 0, 0)).toBe(true) + expect(syncAggregationActiveAt(true, 0, 100)).toBe(true) + expect(syncAggregationActiveAt(true, 50, 49)).toBe(false) + expect(syncAggregationActiveAt(true, 50, 50)).toBe(true) + expect(syncAggregationActiveAt(true, 50, 51)).toBe(true) + expect(syncAggregationActiveAt(true, -5, 0)).toBe(true) + expect(syncAggregationActiveAt(true, 2.5, 1)).toBe(true) + expect(syncAggregationActiveAt(true, Number.NaN, 0)).toBe(true) + }) + + test.each([20, 50, 500])( + "%i receivers converge from four partitioned partial bitmap aggregates", + nodeCount => { + const identities = Array.from( + { length: nodeCount }, + (_, index) => `node-${index.toString().padStart(3, "0")}`, + ) + const signers = identities.slice(0, 4) + const block = { + number: 42, + hash: "block-42", + validation_data: { + signatures: Object.fromEntries( + signers.map(identity => [identity, {}]), + ), + }, + content: { peerlist: identities }, + } + const partials = signers.map(signer => { + const slice = blockDeliveryPartition( + signer, + signers, + identities, + ) + expect(slice).not.toBe(null) + const aggregate = buildSyncAggregateV2( + block as unknown as V2SourceBlock, + signer, + (slice ?? []).map(pubkey => ({ + pubkey, + result: { + result: 200, + response: { syncData: "1:42:block-42" }, + }, + })), + ) + expect(aggregate).not.toBe(null) + return { signer, aggregate } + }) + + for (const localIdentity of identities) { + const acknowledged = new Set() + for (const { signer, aggregate } of partials) { + const admission = admitSyncAggregate( + aggregate, + block, + signer, + localIdentity, + identities, + ) + expect(admission.ok).toBe(true) + if ("status" in admission) continue + for (const identity of admission.acceptedPeerIds) { + acknowledged.add(identity) + } + } + expect([...acknowledged].sort()).toEqual( + identities.filter(identity => identity !== localIdentity), + ) + } + }, + ) + + test("seeded partition slots are deterministic and rotate across seeds", () => { + const committee = ["0xaa", "0xbb", "0xcc", "0xdd"] + const peers = Array.from( + { length: 64 }, + (_, i) => `0x${(i + 256).toString(16).padStart(4, "0")}`, + ) + + for (const seed of ["", "block-hash-1", "block-hash-2"]) { + const slices = committee.map(member => + blockDeliveryPartition(member, committee, peers, seed), + ) + expect(slices.flatMap(slice => slice ?? []).sort()).toEqual( + [...peers].sort(), + ) + expect( + blockDeliveryPartition(committee[0], committee, peers, seed), + ).toEqual(slices[0]) + } + + const bySeed = ["block-hash-1", "block-hash-2"].map(seed => + peers.map(peer => partitionIndexFor(peer, committee.length, seed)), + ) + expect(bySeed[0]).not.toEqual(bySeed[1]) + for (const assignment of bySeed) { + for (const slot of assignment) { + expect(slot).toBeGreaterThanOrEqual(0) + expect(slot).toBeLessThan(committee.length) + } + } + }) +}) diff --git a/src/libs/communications/syncAggregation.ts b/src/libs/communications/syncAggregation.ts index b1325710..33f8dcb6 100644 --- a/src/libs/communications/syncAggregation.ts +++ b/src/libs/communications/syncAggregation.ts @@ -1,8 +1,12 @@ import type Block from "../blockchain/block" export const SYNC_AGGREGATE_VERSION = 1 as const +export const SYNC_AGGREGATE_V2_VERSION = 2 as const export const MAX_SYNC_AGGREGATE_IDENTITIES = 1000 export const MAX_SYNC_AGGREGATE_IDENTITY_LENGTH = 256 +// 4096 bytes covers 32,768 committed peerlist entries; anything larger is +// rejected before decoding so a hostile aggregate cannot force allocation. +export const MAX_SYNC_AGGREGATE_BITMAP_BYTES = 4096 export interface BlockSyncAggregateV1 { version: typeof SYNC_AGGREGATE_VERSION @@ -11,6 +15,27 @@ export interface BlockSyncAggregateV1 { syncedPeerIds: string[] } +/** + * Version 2 encodes acknowledgements as a bitmap over the canonical index of + * the block's committed peerlist instead of a JSON identity list. The + * peerlist is inside the block-hash preimage, so every node holding the block + * derives the identical index locally and no identities travel on the wire. + * `validation_data.signatures` is deliberately NOT part of the index: the + * signature map is merged incrementally per node and is not hash-covered, so + * it can differ between nodes holding the same block. + */ +export interface BlockSyncAggregateV2 { + version: typeof SYNC_AGGREGATE_V2_VERSION + blockNumber: number + blockHash: string + /** Expected canonical index length; cheap cross-check before decoding. */ + peerlistSize: number + /** Base64 bitmap, bit i = canonical index entry i acknowledged the block. */ + ackBits: string +} + +export type BlockSyncAggregate = BlockSyncAggregateV1 | BlockSyncAggregateV2 + export interface PostBlockTrafficEstimate { nodeCount: number signerCount: number @@ -167,10 +192,155 @@ export function isBlockSyncAggregateV1( ) } +/** + * Canonical acknowledgement index for a block: the committed peerlist + * identities, normalized, deduplicated and sorted. The committed peerlist is + * part of the block-hash preimage, so every node holding the block computes + * the identical index. Handles both committed entry shapes (identity strings + * and legacy `{ identity }` objects). + */ +export function canonicalAckIndex(block: SyncAggregateBlockView): string[] { + const committedPeerlist = Array.isArray(block.content?.peerlist) + ? block.content.peerlist + : [] + const identities = new Set() + for (const entry of committedPeerlist) { + let identity: string | null = null + if (typeof entry === "string") { + identity = entry + } else if ( + entry && + typeof entry === "object" && + typeof (entry as { identity?: unknown }).identity === "string" + ) { + identity = (entry as { identity: string }).identity + } + if (identity && isBoundedIdentity(identity)) { + identities.add(normalizeIdentity(identity)) + } + } + return [...identities].sort() +} + +/** Pack acknowledgement flags LSB-first into a base64 bitmap. */ +export function encodeAckBits(flags: boolean[]): string { + const bytes = new Uint8Array(Math.ceil(flags.length / 8)) + for (let i = 0; i < flags.length; i++) { + if (flags[i]) bytes[i >> 3] |= 1 << (i & 7) + } + return Buffer.from(bytes).toString("base64") +} + +/** + * Decode a base64 acknowledgement bitmap. Returns null unless the payload is + * canonical: strict base64, exactly ceil(expectedSize / 8) bytes, and every + * bit beyond expectedSize cleared. + */ +export function decodeAckBits( + ackBits: string, + expectedSize: number, +): boolean[] | null { + if ( + typeof ackBits !== "string" || + expectedSize <= 0 || + expectedSize > MAX_SYNC_AGGREGATE_BITMAP_BYTES * 8 || + ackBits.length % 4 !== 0 || + !/^[A-Za-z0-9+/]*={0,2}$/.test(ackBits) + ) { + return null + } + const bytes = Buffer.from(ackBits, "base64") + if (bytes.length !== Math.ceil(expectedSize / 8)) return null + // Reject non-canonical re-encodings (base64 that decodes to the right + // bytes but was not produced by Buffer's encoder, e.g. junk in padding). + if (Buffer.from(bytes).toString("base64") !== ackBits) return null + const flags: boolean[] = new Array(expectedSize) + for (let i = 0; i < expectedSize; i++) { + flags[i] = (bytes[i >> 3] & (1 << (i & 7))) !== 0 + } + for (let i = expectedSize; i < bytes.length * 8; i++) { + if ((bytes[i >> 3] & (1 << (i & 7))) !== 0) return null + } + return flags +} + +/** + * Build the version-2 bitmap aggregate from block delivery responses. + * Returns null when the block commits no usable peerlist (or one too large + * to represent); callers should fall back to the version-1 list shape. + */ +export function buildSyncAggregateV2( + block: Pick, + senderIdentity: string, + responses: SyncResponseLike[], +): BlockSyncAggregateV2 | null { + const index = canonicalAckIndex(block as unknown as SyncAggregateBlockView) + if ( + index.length === 0 || + index.length > MAX_SYNC_AGGREGATE_BITMAP_BYTES * 8 + ) { + return null + } + const position = new Map(index.map((identity, i) => [identity, i])) + const flags: boolean[] = new Array(index.length).fill(false) + + const senderPosition = position.get(normalizeIdentity(senderIdentity)) + if (senderPosition !== undefined) flags[senderPosition] = true + + for (const response of responses) { + const syncData = extractSyncData(response) + if ( + !syncData || + !syncDataMatchesBlock(syncData, block.number, block.hash) + ) { + continue + } + const claimedPosition = position.get(normalizeIdentity(response.pubkey)) + if (claimedPosition !== undefined) flags[claimedPosition] = true + } + + return { + version: SYNC_AGGREGATE_V2_VERSION, + blockNumber: block.number, + blockHash: block.hash, + peerlistSize: index.length, + ackBits: encodeAckBits(flags), + } +} + +// Base64 of MAX_SYNC_AGGREGATE_BITMAP_BYTES is 4 * ceil(4096 / 3) characters. +const MAX_ACK_BITS_LENGTH = 4 * Math.ceil(MAX_SYNC_AGGREGATE_BITMAP_BYTES / 3) + +export function isBlockSyncAggregateV2( + value: unknown, +): value is BlockSyncAggregateV2 { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return false + } + + const aggregate = value as Record + return ( + aggregate.version === SYNC_AGGREGATE_V2_VERSION && + typeof aggregate.blockNumber === "number" && + Number.isSafeInteger(aggregate.blockNumber) && + aggregate.blockNumber >= 0 && + typeof aggregate.blockHash === "string" && + aggregate.blockHash.length > 0 && + aggregate.blockHash.length <= 256 && + typeof aggregate.peerlistSize === "number" && + Number.isSafeInteger(aggregate.peerlistSize) && + aggregate.peerlistSize > 0 && + aggregate.peerlistSize <= MAX_SYNC_AGGREGATE_BITMAP_BYTES * 8 && + typeof aggregate.ackBits === "string" && + aggregate.ackBits.length <= MAX_ACK_BITS_LENGTH + ) +} + /** * Validate a relayed acknowledgement set against a block already verified by * the receiver. The aggregate cannot add an identity to the peer set, mark an * identity online, or describe a block absent from the receiver's chain. + * Accepts both wire versions so mixed fleets converge during rollout. */ export function admitSyncAggregate( value: unknown, @@ -179,22 +349,56 @@ export function admitSyncAggregate( localIdentity: string, knownPeerIdentities: string[], ): SyncAggregateAdmission { - if (!isBlockSyncAggregateV1(value)) { - return { - ok: false, - status: 400, - message: "Invalid sync aggregate", + let claimedIdentities: string[] + let cachedIndex: string[] | null = null + if (isBlockSyncAggregateV2(value)) { + if ( + !block || + block.number !== value.blockNumber || + block.hash !== value.blockHash + ) { + return { + ok: false, + status: 400, + message: "Sync aggregate does not match the local chain", + } } - } - if ( - !block || - block.number !== value.blockNumber || - block.hash !== value.blockHash - ) { + const index = canonicalAckIndex(block) + cachedIndex = index + if (index.length === 0 || index.length !== value.peerlistSize) { + return { + ok: false, + status: 400, + message: "Sync aggregate peerlist index mismatch", + } + } + const flags = decodeAckBits(value.ackBits, index.length) + if (!flags) { + return { + ok: false, + status: 400, + message: "Invalid sync aggregate bitmap", + } + } + claimedIdentities = index.filter((_, i) => flags[i]) + } else if (isBlockSyncAggregateV1(value)) { + if ( + !block || + block.number !== value.blockNumber || + block.hash !== value.blockHash + ) { + return { + ok: false, + status: 400, + message: "Sync aggregate does not match the local chain", + } + } + claimedIdentities = value.syncedPeerIds + } else { return { ok: false, status: 400, - message: "Sync aggregate does not match the local chain", + message: "Invalid sync aggregate", } } @@ -212,28 +416,15 @@ export function admitSyncAggregate( } const eligibleIds = new Set(signerIds) - const committedPeerlist = Array.isArray(block.content?.peerlist) - ? block.content.peerlist - : [] - for (const entry of committedPeerlist) { - if (typeof entry === "string") { - eligibleIds.add(normalizeIdentity(entry)) - } else if ( - entry && - typeof entry === "object" && - typeof (entry as { identity?: unknown }).identity === "string" - ) { - eligibleIds.add( - normalizeIdentity((entry as { identity: string }).identity), - ) - } + for (const identity of cachedIndex ?? canonicalAckIndex(block)) { + eligibleIds.add(identity) } const local = normalizeIdentity(localIdentity) const knownIds = new Set(knownPeerIdentities.map(normalizeIdentity)) const acceptedPeerIds: string[] = [] const seen = new Set() - for (const claimedIdentity of value.syncedPeerIds) { + for (const claimedIdentity of claimedIdentities) { const identity = normalizeIdentity(claimedIdentity) if ( seen.has(identity) || @@ -250,6 +441,94 @@ export function admitSyncAggregate( return { ok: true, acceptedPeerIds } } +/** + * Stable partition slot for an identity. The assignment depends only on the + * identity, the partition count and the optional seed — never on peerlist + * ordering, which differs between nodes. Passing the block hash as the seed + * rotates slot ownership every block, so a faulty or withholding committee + * member cannot starve the same delivery slice round after round. + */ +export function partitionIndexFor( + identity: string, + partitionCount: number, + seed = "", +): number { + if (partitionCount <= 0) return 0 + const normalized = normalizeIdentity(identity) + if (seed === "") { + const hexTail = normalized.replace(/^0x/, "").slice(-8) + if (/^[0-9a-f]{1,8}$/.test(hexTail)) { + return Number.parseInt(hexTail, 16) % partitionCount + } + } + const keyed = seed === "" ? normalized : `${normalized}:${seed}` + let acc = 0 + for (let i = 0; i < keyed.length; i++) { + acc = (acc * 31 + keyed.charCodeAt(i)) >>> 0 + } + return acc % partitionCount +} + +/** + * The delivery slice a committee member owns under version-2 partitioned + * block publication. Callers must pass only the committee members that + * signed the block: a member that aborted mid-round holds neither the block + * nor a delivery duty and must stay eligible to RECEIVE the block through + * someone's slice. Passed members are excluded from every slice (consensus + * already gave them the block). Returns null when the local node has no + * delivery duty. Every member computes only its own slice, so divergent + * peer views degrade to duplicate or missed deliveries that the receiver + * dedupe and the retained anti-entropy path already handle. + */ +export function blockDeliveryPartition( + localIdentity: string, + committeeIdentities: string[], + peerIdentities: string[], + seed = "", +): string[] | null { + const committee = [ + ...new Set( + committeeIdentities + .filter( + identity => + typeof identity === "string" && + isBoundedIdentity(identity), + ) + .map(normalizeIdentity), + ), + ].sort() + const localIndex = committee.indexOf(normalizeIdentity(localIdentity)) + if (localIndex === -1) return null + + const committeeSet = new Set(committee) + return peerIdentities.filter(identity => { + const normalized = normalizeIdentity(identity) + return ( + !committeeSet.has(normalized) && + partitionIndexFor(normalized, committee.length, seed) === + localIndex + ) + }) +} + +/** + * Whether the aggregation path is active for a given block height. The + * activation height lets a mixed-version fleet flip behaviour at one + * coordinated block instead of on process restart. + */ +export function syncAggregationActiveAt( + enabled: boolean, + activationHeight: number, + blockNumber: number, +): boolean { + if (!enabled) return false + const height = + Number.isSafeInteger(activationHeight) && activationHeight > 0 + ? activationHeight + : 0 + return blockNumber >= height +} + /** * Model the post-block request burst. This deliberately excludes periodic * anti-entropy because it is not triggered once per block. @@ -258,11 +537,35 @@ export function estimatePostBlockTraffic( nodeCount: number, signerCount: number, aggregateEnabled: boolean, + aggregateVersion: 1 | 2 = 1, ): PostBlockTrafficEstimate { const nodes = Math.max(0, Math.floor(nodeCount)) const signers = Math.min(nodes, Math.max(0, Math.floor(signerCount))) const recipients = nodes - signers + if (aggregateEnabled && aggregateVersion === 2) { + // Every signer delivers the block to its deterministic slice, so the + // delivery total is unchanged while per-sender load divides by the + // committee size. Each signer then broadcasts its own tiny partial + // bitmap aggregate to every other node. + const blockPublishers = signers + const blockDeliveries = signers > 0 ? recipients : 0 + const aggregateBroadcasts = + blockDeliveries > 0 || signers > 1 + ? signers * Math.max(0, nodes - 1) + : 0 + return { + nodeCount: nodes, + signerCount: signers, + blockPublishers, + blockDeliveries, + receiverSyncBroadcasts: 0, + senderSyncBroadcasts: 0, + aggregateBroadcasts, + totalRequests: blockDeliveries + aggregateBroadcasts, + } + } + if (aggregateEnabled) { const blockPublishers = signers > 0 ? 1 : 0 const blockDeliveries = blockPublishers * recipients diff --git a/src/libs/consensus/v2/PoRBFT.ts b/src/libs/consensus/v2/PoRBFT.ts index 31f88283..a7eb14b9 100644 --- a/src/libs/consensus/v2/PoRBFT.ts +++ b/src/libs/consensus/v2/PoRBFT.ts @@ -45,8 +45,6 @@ import { readNonces, } from "@/libs/debug/nonceTrace" import { computeMergedPeerlist } from "./routines/peerlistMerge" -import { Config } from "src/config" -import { shouldPublishBlock } from "src/libs/communications/syncAggregation" export type { FailedTranscation } from "./routines/mempoolFilters" @@ -410,25 +408,19 @@ export async function consensusRoutine(): Promise { ) } - const aggregationEnabled = - Config.getInstance().core.blockSyncAggregationEnabled - if ( - shouldPublishBlock( - aggregationEnabled, - getSharedState.publicKeyHex, - manager.shard.members.map(member => member.identity), + // The active sync-aggregation mode decides who publishes what: + // legacy and mode 2 publish from every member (mode 2 only to a + // deterministic slice), mode 1 from the designated secretary. + BroadcastManager.publishBlock( + block, + manager.shard.members.map(member => member.identity), + ).catch(error => { + log.error( + `[consensusRoutine] Block broadcast failed: ${ + error instanceof Error ? error.message : String(error) + }`, ) - ) { - BroadcastManager.broadcastNewBlock(block).catch(error => { - log.error( - `[consensusRoutine] Block broadcast failed: ${ - error instanceof Error - ? error.message - : String(error) - }`, - ) - }) - } + }) DTRManager.releaseDTRWaiter(block) // Apply pending L2PS proofs to L1 state diff --git a/src/libs/network/manageGCRRoutines.ts b/src/libs/network/manageGCRRoutines.ts index 4aa48380..ce99f17d 100644 --- a/src/libs/network/manageGCRRoutines.ts +++ b/src/libs/network/manageGCRRoutines.ts @@ -10,6 +10,7 @@ import { NomisIdentityProvider } from "@/libs/identity/providers/nomisIdentityPr import HumanPassportProvider from "@/libs/identity/tools/humanpassport" import { EthosIdentityProvider } from "@/libs/identity/providers/ethosIdentityProvider" import { BroadcastManager } from "../communications/broadcastManager" +import { PeerManager } from "../peer" import { GCRStorageProgramRoutines } from "../blockchain/gcr/gcr_routines/GCRStorageProgramRoutines" import Datasource from "@/model/datasource" import { GCRStorageProgram } from "@/model/entities/GCRv2/GCR_StorageProgram" @@ -264,7 +265,14 @@ export default async function manageGCRRoutines( if (block.number <= getSharedState.lastBlockNumber) { response.result = 200 - response.response = "Block is already processed" + // Keep the handleNewBlock response shape: without syncData + // the block deliverer cannot count an already-synced peer in + // its acknowledgement aggregate. + response.response = { + result: 200, + message: "Block is already processed", + syncData: PeerManager.getInstance().ourSyncDataString, + } break } diff --git a/testing/devnet/scripts/run-sync-scale-emulator.ts b/testing/devnet/scripts/run-sync-scale-emulator.ts index 496c14c2..95ea5891 100644 --- a/testing/devnet/scripts/run-sync-scale-emulator.ts +++ b/testing/devnet/scripts/run-sync-scale-emulator.ts @@ -1,15 +1,18 @@ import { performance } from "node:perf_hooks" import { admitSyncAggregate, + blockDeliveryPartition, buildSyncAggregate, + buildSyncAggregateV2, estimatePostBlockTraffic, - type BlockSyncAggregateV1, + type BlockSyncAggregate, type SyncAggregateBlockView, } from "../../../src/libs/communications/syncAggregation" interface EmulatorConfig { nodeCounts: number[] iterations: number + aggregateVersion: 1 | 2 signerCount: number baseLatencyMs: number jitterMs: number @@ -25,6 +28,10 @@ interface ActiveRound { identities: string[] secretary: string config: EmulatorConfig + /** Successful /block deliveries per peer index (v2 exactly-once check). */ + blockDeliveredCounts: number[] + /** Per receiving peer, union of acceptedPeerIds across admitted partials. */ + acceptedUnions: Set[] } interface AttemptResult { @@ -51,12 +58,21 @@ interface IterationResult { blockPhaseMs: number aggregatePhaseMs: number requestLatenciesMs: number[] + /** v2 only: v1 aggregate size for the same responses as the partial. */ + v1AggregateBytes?: number + /** v2 only: every non-signer received the block exactly once. */ + blockDeliveredExactlyOnce?: boolean + /** v2 only: every partial was admitted (HTTP ok + result 200) everywhere. */ + allPartialsAdmitted?: boolean + /** v2 only: every receiver's accepted union is all identities but itself. */ + coverageExact?: boolean } interface ScenarioResult { nodeCount: number signerCount: number iterations: number + aggregateVersion: 1 | 2 legacyCallsPerBlock: number aggregateCallsPerBlock: number modeledReductionPercent: number @@ -87,6 +103,11 @@ interface ScenarioResult { max: number } allDeliveriesAdmitted: boolean + coverageExact: boolean + v1VsV2AggregateBytes: { + v1AggregateBytes: number | null + v2PartialAggregateBytes: number | null + } } const encoder = new TextEncoder() @@ -105,6 +126,20 @@ function numericArgument(name: string, fallback: number): number { return parsed } +function parseAggregateVersion(): 1 | 2 { + const raw = + process.argv + .find(value => value.startsWith("--aggregate-version=")) + ?.slice("--aggregate-version=".length) ?? + process.env.AGGREGATE_VERSION ?? + "2" + const parsed = Number(raw) + if (parsed !== 1 && parsed !== 2) { + throw new Error("--aggregate-version must be 1 or 2") + } + return parsed +} + function parseConfig(): EmulatorConfig { const rawCounts = process.argv @@ -123,6 +158,7 @@ function parseConfig(): EmulatorConfig { return { nodeCounts, iterations: numericArgument("iterations", 5), + aggregateVersion: parseAggregateVersion(), signerCount: numericArgument("signers", 4), baseLatencyMs: numericArgument("base-latency-ms", 20), jitterMs: numericArgument("jitter-ms", 80), @@ -225,6 +261,7 @@ const server = Bun.serve({ ) if (phaseName === "block") { + round.blockDeliveredCounts[peerIndex] += 1 return json({ result: 200, response: { @@ -233,13 +270,30 @@ const server = Bun.serve({ }) } + // v2 partials arrive from every committee member; the sender index + // travels as a query parameter so the handler admits with the real + // sender identity. Absent (v1 flow), the secretary remains the sender. + const senderParam = url.searchParams.get("sender") + let senderIdentity = round.secretary + if (senderParam !== null) { + const senderIndex = Number(senderParam) + if ( + !Number.isSafeInteger(senderIndex) || + senderIndex < 0 || + senderIndex >= round.identities.length + ) { + return json({ error: "unknown-sender" }, 404) + } + senderIdentity = round.identities[senderIndex] + } + const payload = (await request.json()) as { - aggregate?: BlockSyncAggregateV1 + aggregate?: BlockSyncAggregate } const admission = admitSyncAggregate( payload.aggregate, round.block, - round.secretary, + senderIdentity, round.identities[peerIndex], round.identities, ) @@ -249,6 +303,9 @@ const server = Bun.serve({ admission.status, ) } + for (const identity of admission.acceptedPeerIds) { + round.acceptedUnions[peerIndex].add(identity) + } return json({ result: 200, accepted: admission.acceptedPeerIds.length, @@ -324,11 +381,11 @@ function rounded(value: number): number { return Number(value.toFixed(3)) } -async function runIteration( +function makeRoundFixture( nodeCount: number, blockNumber: number, config: EmulatorConfig, -): Promise { +): ActiveRound { const identities = Array.from({ length: nodeCount }, (_, index) => peerIdentity(index), ) @@ -344,7 +401,27 @@ async function runIteration( }, content: { peerlist: identities }, } - activeRound = { block, identities, secretary, config } + return { + block, + identities, + secretary, + config, + blockDeliveredCounts: new Array(nodeCount).fill(0), + acceptedUnions: identities.map(() => new Set()), + } +} + +async function runIteration( + nodeCount: number, + blockNumber: number, + config: EmulatorConfig, +): Promise { + if (config.aggregateVersion === 2) { + return runIterationV2(nodeCount, blockNumber, config) + } + const round = makeRoundFixture(nodeCount, blockNumber, config) + const { block, identities, secretary } = round + activeRound = round const started = performance.now() const blockStarted = performance.now() @@ -425,6 +502,216 @@ async function runIteration( } } +async function runIterationV2( + nodeCount: number, + blockNumber: number, + config: EmulatorConfig, +): Promise { + const round = makeRoundFixture(nodeCount, blockNumber, config) + const { block, identities } = round + const signers = identities.slice(0, config.signerCount) + const nonSigners = identities.slice(config.signerCount) + const indexOf = new Map( + identities.map((identity, index) => [identity, index] as const), + ) + + // Every committee member computes its own delivery slice with the real + // partition function; together the slices must cover every non-signer + // exactly once. + const slices = signers.map(member => { + const slice = blockDeliveryPartition(member, signers, identities) + if (slice === null) { + throw new Error( + `committee member ${member} has no v2 delivery slice`, + ) + } + return { member, slice } + }) + const coveredCounts = new Map() + for (const { slice } of slices) { + for (const identity of slice) { + coveredCounts.set(identity, (coveredCounts.get(identity) ?? 0) + 1) + } + } + if ( + coveredCounts.size !== nonSigners.length || + nonSigners.some(identity => coveredCounts.get(identity) !== 1) + ) { + throw new Error( + "v2 delivery slices do not cover the non-signers exactly once", + ) + } + + activeRound = round + + const started = performance.now() + const blockStarted = performance.now() + const blockBody = { blockNumber: block.number, blockHash: block.hash } + const memberDeliveries = await Promise.all( + slices.map(async ({ member, slice }) => { + const results = await Promise.all( + slice.map(identity => { + const peerIndex = indexOf.get(identity) + if (peerIndex === undefined) { + throw new Error(`unknown delivery target ${identity}`) + } + return postWithRetry( + attempt => `/block/${peerIndex}/${attempt}`, + blockBody, + config, + ).then(result => ({ identity, result })) + }), + ) + return { member, results } + }), + ) + const blockPhaseMs = performance.now() - blockStarted + const blockResults = memberDeliveries.flatMap(entry => entry.results) + + // One partial bitmap aggregate per committee member, built from that + // member's own slice responses with the real builder. + const partials = memberDeliveries.map(({ member, results }) => { + const responses = results + .filter(entry => entry.result.ok) + .map(entry => ({ + pubkey: entry.identity, + result: entry.result.body as { + result: number + response?: unknown + }, + })) + const partial = buildSyncAggregateV2( + block as unknown as Parameters[0], + member, + responses, + ) + if (!partial) { + throw new Error( + "buildSyncAggregateV2 returned null for a committed peerlist", + ) + } + return { member, responses, partial } + }) + // v1 aggregate for the same responses, built in memory purely for the + // byte comparison — never sent. + const v1Comparison = buildSyncAggregate( + { number: block.number, hash: block.hash }, + partials[0].member, + partials[0].responses, + ) + + // Each builder applies its own partial locally (a node never posts to + // itself), through the real admission path, so its accepted union also + // carries its own slice. + for (const { member, partial } of partials) { + const selfIndex = indexOf.get(member) + if (selfIndex === undefined) { + throw new Error(`unknown committee member ${member}`) + } + const selfAdmission = admitSyncAggregate( + partial, + block, + member, + member, + identities, + ) + if (!selfAdmission.ok) { + throw new Error( + `local self-admission failed for committee member ${member}`, + ) + } + for (const identity of selfAdmission.acceptedPeerIds) { + round.acceptedUnions[selfIndex].add(identity) + } + } + + const aggregateStarted = performance.now() + const aggregateResults = ( + await Promise.all( + partials.map(({ member, partial }) => { + const senderIndex = indexOf.get(member) + if (senderIndex === undefined) { + throw new Error(`unknown committee member ${member}`) + } + const aggregateBody = { aggregate: partial } + return Promise.all( + identities + .map((_, peerIndex) => peerIndex) + .filter(peerIndex => peerIndex !== senderIndex) + .map(peerIndex => + postWithRetry( + attempt => + `/aggregate/${peerIndex}/${attempt}?sender=${senderIndex}`, + aggregateBody, + config, + ), + ), + ) + }), + ) + ).flat() + const aggregatePhaseMs = performance.now() - aggregateStarted + activeRound = null + + const blockDeliveredExactlyOnce = round.blockDeliveredCounts.every( + (count, index) => count === (index < config.signerCount ? 0 : 1), + ) + const allPartialsAdmitted = aggregateResults.every(result => { + if (!result.ok) return false + const body = result.body as { result?: unknown } | null + return body !== null && body?.result === 200 + }) + // Admission excludes the receiver's own identity, so every receiver must + // end with exactly the committed peerlist minus itself. + const coverageExact = identities.every((identity, index) => { + const union = round.acceptedUnions[index] + return ( + union.size === identities.length - 1 && + !union.has(identity) && + identities.every( + other => other === identity || union.has(other), + ) + ) + }) + + const allResults = [ + ...blockResults.map(entry => entry.result), + ...aggregateResults, + ] + return { + nodeCount, + blockNumber, + aggregateIdentities: partials[0].responses.length + 1, + aggregateBytes: bodyBytes(partials[0].partial), + v1AggregateBytes: bodyBytes(v1Comparison), + blockDeliverySuccesses: blockResults.filter(entry => entry.result.ok) + .length, + aggregateDeliverySuccesses: aggregateResults.filter(result => result.ok) + .length, + blockDeliveredExactlyOnce, + allPartialsAdmitted, + coverageExact, + logicalCalls: allResults.length, + httpAttempts: allResults.reduce( + (total, result) => total + result.attempts, + 0, + ), + retryAttempts: allResults.reduce( + (total, result) => total + result.attempts - 1, + 0, + ), + wireBytes: allResults.reduce( + (total, result) => + total + result.requestBytes + result.responseBytes, + 0, + ), + elapsedMs: performance.now() - started, + blockPhaseMs, + aggregatePhaseMs, + requestLatenciesMs: allResults.map(result => result.elapsedMs), + } +} + async function runScenario( nodeCount: number, config: EmulatorConfig, @@ -467,6 +754,7 @@ async function runScenario( nodeCount, config.signerCount, true, + config.aggregateVersion, ) const elapsed = iterations.map(result => result.elapsedMs) const requestLatencies = iterations.flatMap( @@ -485,6 +773,7 @@ async function runScenario( nodeCount, signerCount: config.signerCount, iterations: config.iterations, + aggregateVersion: config.aggregateVersion, legacyCallsPerBlock: legacy.totalRequests, aggregateCallsPerBlock: aggregate.totalRequests, modeledReductionPercent: rounded( @@ -529,14 +818,48 @@ async function runScenario( p99: rounded(percentile(eventLoopDelaysMs, 0.99)), max: rounded(Math.max(0, ...eventLoopDelaysMs)), }, - allDeliveriesAdmitted: iterations.every( - result => - result.blockDeliverySuccesses === - nodeCount - config.signerCount && - result.aggregateDeliverySuccesses === nodeCount - 1 && - result.aggregateIdentities === - nodeCount - config.signerCount + 1, - ), + allDeliveriesAdmitted: + config.aggregateVersion === 2 + ? iterations.every( + result => + result.blockDeliverySuccesses === + nodeCount - config.signerCount && + result.aggregateDeliverySuccesses === + config.signerCount * (nodeCount - 1) && + result.blockDeliveredExactlyOnce === true && + result.allPartialsAdmitted === true, + ) + : iterations.every( + result => + result.blockDeliverySuccesses === + nodeCount - config.signerCount && + result.aggregateDeliverySuccesses === + nodeCount - 1 && + result.aggregateIdentities === + nodeCount - config.signerCount + 1, + ), + coverageExact: + config.aggregateVersion === 2 + ? iterations.every(result => result.coverageExact === true) + : true, + v1VsV2AggregateBytes: + config.aggregateVersion === 2 + ? { + v1AggregateBytes: Math.max( + ...iterations.map( + result => result.v1AggregateBytes ?? 0, + ), + ), + v2PartialAggregateBytes: Math.max( + ...iterations.map(result => result.aggregateBytes), + ), + } + : { + v1AggregateBytes: Math.max( + ...iterations.map(result => result.aggregateBytes), + ), + v2PartialAggregateBytes: null, + }, } } @@ -586,6 +909,63 @@ function validateSafetyCases(): Record { identities[1], identities, ) + const v2Responses = identities.slice(4).map(pubkey => ({ + pubkey, + result: { + result: 200, + response: { syncData: "1:42:block-42" }, + }, + })) + const aggregateV2 = buildSyncAggregateV2( + block as unknown as Parameters[0], + identities[0], + v2Responses, + ) + if (!aggregateV2) { + return { v2AggregateBuilt: false } + } + const validV2 = admitSyncAggregate( + aggregateV2, + block, + identities[0], + identities[1], + identities, + ) + const v2NonSigner = admitSyncAggregate( + aggregateV2, + block, + identities[5], + identities[1], + identities, + ) + const v2WrongBlock = admitSyncAggregate( + aggregateV2, + { ...block, hash: "wrong-block" }, + identities[0], + identities[1], + identities, + ) + const v2PeerlistMismatch = admitSyncAggregate( + { ...aggregateV2, peerlistSize: aggregateV2.peerlistSize + 1 }, + block, + identities[0], + identities[1], + identities, + ) + // Craft a bitmap with a set bit beyond the canonical index: decode the + // valid ackBits, set a high trailing bit and re-encode. + const tamperedBytes = Buffer.from(aggregateV2.ackBits, "base64") + tamperedBytes[tamperedBytes.length - 1] |= 0x80 + const v2TrailingBit = admitSyncAggregate( + { + ...aggregateV2, + ackBits: Buffer.from(tamperedBytes).toString("base64"), + }, + block, + identities[0], + identities[1], + identities, + ) return { validAggregateAccepted: valid.ok, nonSignerRejected: @@ -594,6 +974,24 @@ function validateSafetyCases(): Record { !wrongBlock.ok && "status" in wrongBlock && wrongBlock.status === 400, + v2AggregateBuilt: true, + v2AggregateAccepted: validV2.ok, + v2NonSignerRejected: + !v2NonSigner.ok && + "status" in v2NonSigner && + v2NonSigner.status === 403, + v2WrongBlockRejected: + !v2WrongBlock.ok && + "status" in v2WrongBlock && + v2WrongBlock.status === 400, + v2PeerlistSizeMismatchRejected: + !v2PeerlistMismatch.ok && + "status" in v2PeerlistMismatch && + v2PeerlistMismatch.status === 400, + v2TrailingBitRejected: + !v2TrailingBit.ok && + "status" in v2TrailingBit && + v2TrailingBit.status === 400, } } @@ -623,6 +1021,7 @@ async function main(): Promise { results.every( result => result.allDeliveriesAdmitted && + result.coverageExact && result.observedLogicalCallsPerBlock === result.aggregateCallsPerBlock, ) && Object.values(safety).every(Boolean) diff --git a/testing/devnet/scripts/run-sync-scale-vps.sh b/testing/devnet/scripts/run-sync-scale-vps.sh index 3b06fd64..526eb371 100755 --- a/testing/devnet/scripts/run-sync-scale-vps.sh +++ b/testing/devnet/scripts/run-sync-scale-vps.sh @@ -4,6 +4,7 @@ set -euo pipefail ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" NODE_COUNTS="${NODE_COUNTS:-100,250,500}" ITERATIONS="${ITERATIONS:-5}" +AGGREGATE_VERSION="${AGGREGATE_VERSION:-2}" MIN_AVAILABLE_KIB="${MIN_AVAILABLE_KIB:-2097152}" RESULTS_DIR="${RESULTS_DIR:-${ROOT_DIR}/.poc-results}" TIMESTAMP="$(date -u +%Y%m%dT%H%M%SZ)" @@ -47,13 +48,14 @@ if (( AVAILABLE_KIB < MIN_AVAILABLE_KIB )); then fi mkdir -p "${RESULTS_DIR}" -echo "Running bounded sync emulator: nodes=${NODE_COUNTS} iterations=${ITERATIONS}" >&2 +echo "Running bounded sync emulator: nodes=${NODE_COUNTS} iterations=${ITERATIONS} aggregate-version=${AGGREGATE_VERSION}" >&2 echo "Six-node POC paused=${#PAUSED_NODES[@]}; live DACS remains active" >&2 cd "${ROOT_DIR}" bun testing/devnet/scripts/run-sync-scale-emulator.ts \ --nodes="${NODE_COUNTS}" \ - --iterations="${ITERATIONS}" >"${REPORT_PATH}" & + --iterations="${ITERATIONS}" \ + --aggregate-version="${AGGREGATE_VERSION}" >"${REPORT_PATH}" & EMULATOR_PID=$! while kill -0 "${EMULATOR_PID}" 2>/dev/null; do