diff --git a/agent/CRATES_ARCHITECTURE.md b/agent/CRATES_ARCHITECTURE.md index 3dd5a60207..c59d8c753c 100644 --- a/agent/CRATES_ARCHITECTURE.md +++ b/agent/CRATES_ARCHITECTURE.md @@ -437,6 +437,10 @@ terminating the process. Process-level health supervision would need to detect t but the current runtime does not provide that guarantee. A restart, when it occurs, treats the event log as authoritative and reconciles missing derived index rows. +History collectors use the same `MAILBOX_LIMIT_LARGE` capacity to reduce mailbox saturation while +they record bus events for history inspection. The larger mailbox does not make live fanout lossless; +do not use it as a replacement for acknowledged delivery or replay. + Those replay guarantees bound local replay memory and file-descriptor use, but they do not make the whole persistence path synchronously acknowledged. Live publication and the sequencer/store response path still contain `do_send` edges. Snapshot replay also forwards with `do_send`, and `BatchRouter` @@ -568,8 +572,8 @@ sequenceDiagram P->>Z: folded/recursive aggregation proof work P->>W: aggregated public key Chain->>K: ciphertext outputs - K->>Z: C6 decryption-share proofs - K->>T: share + proof per output and party + K->>Z: C6 decryption-share proofs for the canonical H roster + K->>T: share + proof per output and honest party T->>Z: C7 aggregation proof T->>W: plaintext output ``` diff --git a/agent/flow-trace/00_INDEX.md b/agent/flow-trace/00_INDEX.md index 27b80d8926..bd95044371 100644 --- a/agent/flow-trace/00_INDEX.md +++ b/agent/flow-trace/00_INDEX.md @@ -72,9 +72,9 @@ 12. COMPUTE Data encrypted with aggregate PK, computation runs → Ciphertext output published on-chain -13. DECRYPT Committee members produce decryption shares - → C6 proof per share (proves share correctly derived) - → broadcast to all committee members for buffering +13. DECRYPT Canonical honest committee members produce decryption shares + → C6 proof per share (proves share correctly derived) + → broadcast to all committee members for buffering 14. AGGREGATE Active aggregator combines M+1 shares → plaintext → C7 proof (proves reconstruction correct) diff --git a/agent/flow-trace/04_DKG_AND_COMPUTATION.md b/agent/flow-trace/04_DKG_AND_COMPUTATION.md index ca0d9ad8c6..4fdfef1788 100644 --- a/agent/flow-trace/04_DKG_AND_COMPUTATION.md +++ b/agent/flow-trace/04_DKG_AND_COMPUTATION.md @@ -499,12 +499,15 @@ ThresholdKeyshare receives AllThresholdSharesCollected │ │ │ → Stored encrypted locally for later decryption │ │ │ └─────────────────────────────────────────────────────┘ │ -├─ 2b. CANONICAL H ROSTER (when H < N): +├─ 2b. C4 ROSTER (when H < N): │ Before C4 witness layout, merge external honest party_ids with own_party_id, │ sort ascending, and keep the lowest H — same rule as PublicKeyAggregator C5 cap -│ (`e3_zk_helpers::canonical_honest_party_ids_with_own`). Persisted as `honest_parties`. -│ Parties outside the lowest H still complete KeyshareCreated but are not in the -│ aggregator's NodeFold / `honest_committee_addresses` roster. +│ (`e3_zk_helpers::canonical_honest_party_ids_with_own`). +│ +│ After C5, PublicKeyAggregated carries the authoritative H-address subset. Each +│ keyshare maps those addresses through the full N committee order and replaces its +│ persisted `honest_parties` set before C6. Parties outside that global H roster +│ ignore CiphertextOutputPublished and do not generate C6 work. │ ├─ 3. PUBLISH C4 PROOF REQUESTS: │ DecryptionShareProofsPending { @@ -904,7 +907,11 @@ active E3's deadlines. --- -## Phase 4: Decryption Share Generation (Each Committee Member, with C6 Proof) +## Phase 4: Decryption Share Generation (Canonical Honest Members, with C6 Proof) + +Only members in the persisted canonical honest roster of size `H` generate a decryption share and +C6 proof. Other committee members ignore `CiphertextOutputPublished` for this phase. The active +aggregator verifies and folds the resulting `H` share/proof bundles. Before proof verification, the BFV wrapper requires every public input to use its canonical BN254 field representation. Message coefficients must also fit exactly in 64 bits. This second check is diff --git a/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md b/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md index 247d657084..a2b858e57f 100644 --- a/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md +++ b/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md @@ -1136,8 +1136,9 @@ Slash Reasons (derived from ProofType for Lane A): │ Complete Proof-to-Slash Pipeline │ ├─────────────────────────────────────────────────────────────────┤ │ │ -│ 1. PROOF GENERATION (each committee member) │ +│ 1. PROOF GENERATION (per phase-eligible producer) │ │ ProofRequestActor generates & signs C0-C7 proofs │ +│ → C6 generation is limited to the canonical H roster │ │ → Broadcasts signed proofs via P2P gossip │ │ │ │ 2. PROOF VERIFICATION (each receiving committee member) │ diff --git a/circuits/benchmarks/scripts/generate_report.sh b/circuits/benchmarks/scripts/generate_report.sh index 2a41d7806e..de7934d169 100755 --- a/circuits/benchmarks/scripts/generate_report.sh +++ b/circuits/benchmarks/scripts/generate_report.sh @@ -771,6 +771,12 @@ if [ -n "$INTEGRATION_BLOB" ]; then [ -z "$name" ] && continue echo "| $name | $(format_s "$avgr") | $runs | $(format_s "$tot") |" >> "$OUTPUT_FILE" done < <(jq -r '.operation_timings[]? | [.name, .avg_seconds, .runs, .total_seconds] | @tsv' <<<"$INTEGRATION_BLOB") + c6_h=$(jq -r '.benchmark_config.committee_h // empty' <<<"$INTEGRATION_BLOB") + c6_n=$(jq -r '.benchmark_config.committee_n // empty' <<<"$INTEGRATION_BLOB") + if [ -n "$c6_h" ] && [ -n "$c6_n" ]; then + echo "" >> "$OUTPUT_FILE" + echo "_Run-count note: DKG preparation rows are N-scoped and remain \`N=${c6_n}\` for this committee. C6 producer rows are H-scoped: \`ZkThresholdShareDecryption\` should run once for each eligible honest party, so its expected request count is \`H=${c6_h}\`. C7 aggregation is one job per ciphertext output and consumes H C6 inputs. The \`runs\` value counts proof-generation or aggregation jobs, as applicable._" >> "$OUTPUT_FILE" + fi ott=$(jq -r '.operation_timings_total_seconds // empty' <<<"$INTEGRATION_BLOB") if [ -n "$ott" ] && [ "$ott" != "null" ]; then echo "" >> "$OUTPUT_FILE" diff --git a/crates/events/src/eventbus.rs b/crates/events/src/eventbus.rs index 99076d3489..80b3cd0130 100644 --- a/crates/events/src/eventbus.rs +++ b/crates/events/src/eventbus.rs @@ -7,7 +7,7 @@ use crate::traits::{ErrorEvent, Event}; use crate::{EventBusBarrier, EventType}; use actix::prelude::*; -use e3_utils::{colorize, Color, MAILBOX_LIMIT, MAILBOX_LIMIT_LARGE}; +use e3_utils::{colorize, Color, MAILBOX_LIMIT_LARGE}; use futures_util::future::join_all; use std::collections::{HashMap, HashSet, VecDeque}; use std::fmt; @@ -546,7 +546,7 @@ impl HistoryCollector { impl Actor for HistoryCollector { type Context = Context; fn started(&mut self, ctx: &mut Self::Context) { - ctx.set_mailbox_capacity(MAILBOX_LIMIT); + ctx.set_mailbox_capacity(MAILBOX_LIMIT_LARGE); } } diff --git a/crates/keyshare/src/threshold_keyshare/actor.rs b/crates/keyshare/src/threshold_keyshare/actor.rs index 68dc797b9c..570c413ead 100644 --- a/crates/keyshare/src/threshold_keyshare/actor.rs +++ b/crates/keyshare/src/threshold_keyshare/actor.rs @@ -57,9 +57,10 @@ use crate::actors::threshold_share_collector::{ use crate::domain::timeout_policy::{resolve_timeout, DkgTimeoutPhase}; use crate::domain::{ build_decryption_key_plan, build_shares_generated_plan, generate_bfv_keypair, - AggregatingDecryptionKey, BfvKeypairMaterial, CollectingEncryptionKeysData, Decrypting, - DecryptionKeyPlan, GeneratingDecryptionProof, GeneratingThresholdShareData, KeyshareState, - ProofRequestData, ReadyForDecryption, ReceivedShareProofs, ThresholdKeyshareState, + party_ids_for_honest_addresses, AggregatingDecryptionKey, BfvKeypairMaterial, + CollectingEncryptionKeysData, Decrypting, DecryptionKeyPlan, GeneratingDecryptionProof, + GeneratingThresholdShareData, KeyshareState, ProofRequestData, ReadyForDecryption, + ReceivedShareProofs, ThresholdKeyshareState, }; #[path = "recovery_state.rs"] diff --git a/crates/keyshare/src/threshold_keyshare/effects/create_decryption_share.rs b/crates/keyshare/src/threshold_keyshare/effects/create_decryption_share.rs index b6ec7ed05a..3767325eb5 100644 --- a/crates/keyshare/src/threshold_keyshare/effects/create_decryption_share.rs +++ b/crates/keyshare/src/threshold_keyshare/effects/create_decryption_share.rs @@ -11,6 +11,23 @@ impl ThresholdKeyshare { msg: TypedEvent, ) -> Result<()> { let (msg, ec) = msg.into_components(); + let state = self.state.try_get()?; + let Some(honest_parties) = state.honest_parties.as_ref() else { + info!( + e3_id = %state.e3_id, + party_id = state.party_id, + "Skipping decryption-share generation because the canonical honest roster is unavailable" + ); + return Ok(()); + }; + if !honest_parties.contains(&state.party_id) { + info!( + e3_id = %state.e3_id, + party_id = state.party_id, + "Skipping decryption-share generation for a party outside the canonical honest roster" + ); + return Ok(()); + } let ciphertext_output = msg.ciphertext_output; // If we are already in Decrypting (or beyond), this is a duplicate diff --git a/crates/keyshare/src/threshold_keyshare/effects/route_events.rs b/crates/keyshare/src/threshold_keyshare/effects/route_events.rs index 4554d9f560..5472203d3b 100644 --- a/crates/keyshare/src/threshold_keyshare/effects/route_events.rs +++ b/crates/keyshare/src/threshold_keyshare/effects/route_events.rs @@ -16,11 +16,22 @@ impl Handler for ThresholdKeyshare { self.notify_sync(ctx, TypedEvent::new(data, ec)) } InterfoldEventData::PublicKeyAggregated(data) => { + let honest_party_ids = match party_ids_for_honest_addresses( + &data.committee_addresses, + &data.honest_committee_addresses, + ) { + Ok(ids) => ids, + Err(err) => { + self.bus.err(EType::KeyGeneration, err); + return; + } + }; let committee_hash = e3_committee_hash::hash_committee_addresses(&data.committee_addresses); let pk = ArcBytes::from_bytes(&data.pubkey); let _ = self.state.try_mutate(&ec, |mut s| { s.aggregated_pk = Some(pk); + s.honest_parties = Some(honest_party_ids.clone()); s.decryption_domain = Some(e3_committee_hash::DecryptionDomainContext { interfold_address: self.interfold_address, committee_hash, @@ -28,6 +39,11 @@ impl Handler for ThresholdKeyshare { }); Ok(s) }); + info!( + e3_id = %data.e3_id, + honest_party_ids = ?honest_party_ids, + "Stored the canonical honest roster for decryption-share generation" + ); } InterfoldEventData::ThresholdShareCreated(data) => { let _ = diff --git a/crates/keyshare/src/threshold_keyshare/state.rs b/crates/keyshare/src/threshold_keyshare/state.rs index 4231a544dd..1bfd37eef0 100644 --- a/crates/keyshare/src/threshold_keyshare/state.rs +++ b/crates/keyshare/src/threshold_keyshare/state.rs @@ -11,6 +11,7 @@ //! It contains NO actix, persistence, bus or timer dependencies — only plain //! synchronous data and transition logic, which makes it directly unit-testable. +use alloy::primitives::Address; use anyhow::{anyhow, Result}; use e3_committee_hash::DecryptionDomainContext; use e3_crypto::SensitiveBytes; @@ -30,6 +31,37 @@ use std::{ use crate::domain::timeout_policy::now_unix_secs; +/// Map the authoritative honest addresses to their party IDs in full committee order. +pub(crate) fn party_ids_for_honest_addresses( + committee_addresses: &[Address], + honest_committee_addresses: &[Address], +) -> Result> { + if honest_committee_addresses.is_empty() { + return Err(anyhow!( + "PublicKeyAggregated contained an empty honest committee" + )); + } + + let mut party_ids = BTreeSet::new(); + for address in honest_committee_addresses { + let index = committee_addresses + .iter() + .position(|candidate| candidate == address) + .ok_or_else(|| { + anyhow!("Honest committee address {address} is not in the full committee") + })?; + let party_id = u64::try_from(index) + .map_err(|_| anyhow!("Committee index {index} does not fit in a party ID"))?; + if !party_ids.insert(party_id) { + return Err(anyhow!( + "Honest committee contains duplicate address {address}" + )); + } + } + + Ok(party_ids) +} + #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] pub struct CollectingEncryptionKeysData { pub(crate) sk_bfv: SensitiveBytes, diff --git a/crates/keyshare/src/threshold_keyshare/state_tests.rs b/crates/keyshare/src/threshold_keyshare/state_tests.rs index 3eaf2f62c8..7513439c4d 100644 --- a/crates/keyshare/src/threshold_keyshare/state_tests.rs +++ b/crates/keyshare/src/threshold_keyshare/state_tests.rs @@ -5,6 +5,7 @@ // or FITNESS FOR A PARTICULAR PURPOSE. use super::*; +use alloy::primitives::Address; use e3_events::E3id; fn arc(bytes: &[u8]) -> ArcBytes { @@ -37,6 +38,28 @@ fn new_initialises_defaults_and_records_dkg_start() { assert_eq!(s.get_address(), "0xabc"); } +#[test] +fn honest_addresses_map_to_full_committee_party_ids() { + let committee = vec![ + Address::from([1u8; 20]), + Address::from([2u8; 20]), + Address::from([3u8; 20]), + ]; + + assert_eq!( + party_ids_for_honest_addresses(&committee, &[committee[0], committee[2]]) + .expect("valid honest subset"), + BTreeSet::from([0, 2]) + ); +} + +#[test] +fn honest_address_outside_full_committee_is_rejected() { + let committee = vec![Address::from([1u8; 20])]; + + assert!(party_ids_for_honest_addresses(&committee, &[Address::from([2u8; 20])]).is_err()); +} + #[test] fn same_branch_transition_is_always_valid() { // Re-entering the same phase variant must be accepted (idempotent mutations). diff --git a/crates/keyshare/src/threshold_keyshare/tests.rs b/crates/keyshare/src/threshold_keyshare/tests.rs index d93915cd3b..6f76c57182 100644 --- a/crates/keyshare/src/threshold_keyshare/tests.rs +++ b/crates/keyshare/src/threshold_keyshare/tests.rs @@ -14,11 +14,11 @@ use e3_data::{AutoPersist, DataStore, InMemStore, Persistable, Repository}; use e3_events::{ hlc_factory::HlcFactory, BusHandle, ComputeRequestKind, E3Stage, E3id, EffectsEnabled, EventBus, EventBusConfig, EventSource, FailureReason, HistoryCollector, InterfoldEvent, - InterfoldEventData, Sequencer, StoreEventRequested, StoreEventResponse, TakeEvents, - Unsequenced, + InterfoldEventData, PublicKeyAggregated, Sequencer, StoreEventRequested, StoreEventResponse, + TakeEvents, Unsequenced, }; use e3_fhe_params::DEFAULT_BFV_PRESET; -use std::sync::Arc; +use std::{collections::BTreeSet, sync::Arc}; #[derive(Default)] struct TestEventStore { @@ -109,6 +109,43 @@ async fn start_actor() -> Result<( start_actor_with_state(KeyshareState::Init).await } +#[actix::test] +async fn public_key_aggregated_replaces_local_roster_with_global_honest_set() -> Result<()> { + let (actor, _history, e3_id, repo) = start_actor().await?; + let committee_addresses = vec![ + Address::from([1u8; 20]), + Address::from([2u8; 20]), + Address::from([3u8; 20]), + ]; + let event = PublicKeyAggregated { + pubkey: ArcBytes::from_bytes(b"public-key"), + e3_id, + nodes: Default::default(), + committee_addresses: committee_addresses.clone(), + honest_committee_addresses: vec![committee_addresses[0], committee_addresses[2]], + pk_commitment: [0u8; 32], + dkg_aggregator_proof: None, + dkg_attestation_bundle: None, + }; + + actor + .send( + InterfoldEvent::::new_with_timestamp( + event.into(), + None, + 1, + None, + EventSource::Local, + ) + .into_sequenced(1), + ) + .await?; + + let state = repo.read().await?.expect("persisted keyshare state"); + assert_eq!(state.honest_parties, Some(BTreeSet::from([0, 2]))); + Ok(()) +} + async fn next_event(history: &Addr>) -> Result { let mut result = history.send(TakeEvents::::new(1)).await?; assert!(!result.timed_out, "timed out waiting for an event"); diff --git a/crates/tests/tests/integration.rs b/crates/tests/tests/integration.rs index 3a0c56090b..a74d72ebda 100644 --- a/crates/tests/tests/integration.rs +++ b/crates/tests/tests/integration.rs @@ -16,9 +16,9 @@ use e3_events::{ hlc::HlcTimestamp, prelude::*, AggregateConfig, AggregateId, BusHandle, CiphertextOutputPublished, CommitteeFinalized, CommitteeRequested, ComputeRequestKind, ComputeResponseKind, ConfigurationUpdated, DkgFoldAttestationContextEstablished, E3Requested, - E3id, InterfoldEvent, InterfoldEventData, OperatorActivationChanged, PlaintextAggregated, - ProofType, Seed, TakeEvents, TicketBalanceUpdated, VerificationKind, ZkRequest, ZkResponse, - DKG_FOLD_ATTESTATION_CONTEXT_SCHEMA_VERSION, + E3id, EventId, EventSource, InterfoldEvent, InterfoldEventData, OperatorActivationChanged, + PlaintextAggregated, ProofType, Seed, TakeEvents, TicketBalanceUpdated, VerificationKind, + ZkRequest, ZkResponse, DKG_FOLD_ATTESTATION_CONTEXT_SCHEMA_VERSION, }; use e3_fhe_params::DEFAULT_BFV_PRESET; use e3_fhe_params::{encode_bfv_params, BfvParamSet, BfvPreset}; @@ -51,6 +51,8 @@ use tokio::{ time::sleep, }; +const BENCHMARK_POST_BARRIER_SETTLE_TIMEOUT: Duration = Duration::from_secs(60); + #[derive(Debug, Clone, Copy)] struct BenchmarkParams { /// Noir artifact preset directory name under `circuits/bin/` (e.g. `secure-8192`). @@ -111,7 +113,6 @@ fn select_benchmark_params() -> BenchmarkParams { } else { Duration::from_secs(1_000) }; - BenchmarkParams { preset_subdir, bfv_preset, @@ -154,6 +155,26 @@ fn benchmark_multithread_concurrent_jobs() -> usize { .unwrap_or(1) } +/// Logical CPUs that the shared Rayon pool must not use, so actor runtimes keep headroom +/// (`BENCHMARK_RESERVE_THREADS`, default `max(1, min(max(2, cores / 4), cores.saturating_sub(1)))`). +/// +/// The harness runs all nodes, their event buses, and their history collectors on one tokio +/// runtime. A pool sized near the core count starves those actors. EventBus fanout then passes +/// its accept deadline and silently drops events from node histories. +fn benchmark_multithread_reserve_threads() -> usize { + std::env::var("BENCHMARK_RESERVE_THREADS") + .ok() + .and_then(|s| s.parse().ok()) + .filter(|&threads| threads >= 1) + .unwrap_or_else(|| { + let cores = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(1); + let floor = (cores / 4).max(2); + floor.min(cores.saturating_sub(1)).max(1) + }) +} + static NEXT_BENCHMARK_NODE_RNG_SALT: AtomicU64 = AtomicU64::new(1); /// One ChaCha20 mutex per ciphernode in `test_trbfv_actor` (see `derive_shared_rng`). @@ -191,20 +212,19 @@ fn uses_dist_preset_artifacts(preset_subdir: &str, committee_str: &str) -> bool .exists() } -/// Preset stamp path — under the new layout each `{preset}/{committee}` dir has its own stamp. -fn resolve_preset_stamp_path(preset_subdir: &str, committee_str: &str) -> PathBuf { - if uses_dist_preset_artifacts(preset_subdir, committee_str) { - repo_root() - .join("dist/circuits") - .join(preset_subdir) - .join(committee_str) - .join(".build-stamp.json") - } else { - repo_root() - .join("circuits") - .join("bin") - .join(".active-preset.json") - } +fn preset_stamp_matches( + stamp_path: &std::path::Path, + preset_subdir: &str, + committee_str: &str, +) -> bool { + let Ok(raw) = std::fs::read_to_string(stamp_path) else { + return false; + }; + let Ok(stamp) = serde_json::from_str::(&raw) else { + return false; + }; + stamp.get("preset").and_then(|value| value.as_str()) == Some(preset_subdir) + && stamp.get("committee").and_then(|value| value.as_str()) == Some(committee_str) } /// Reads the active committee from `circuits/bin/.active-preset.json`, which is written by every @@ -387,21 +407,23 @@ async fn setup_test_zk_backend( let preset_out = circuits_dir.join(preset_subdir).join(committee_str); let circuits_bin_marker = repo_root.join("circuits/bin/dkg/target/pk.json"); - // `circuits/bin` is preset-agnostic on disk — only `.active-preset.json` records which - // preset+committee the most recent local build targeted. Without it we cannot tell - // whether `circuits/bin` matches `preset_subdir`, and copying wrong artifacts would - // silently produce invalid proofs. - let preset_build_stamp = resolve_preset_stamp_path(preset_subdir, committee_str); - - if uses_dist_preset_artifacts(preset_subdir, committee_str) { + // `circuits/bin` is preset-agnostic on disk. Require the stamp to match both the requested + // preset and committee before copying local artifacts into the test backend. + let dist_stamp = dist_preset.join(".build-stamp.json"); + let active_bin_stamp = repo_root.join("circuits/bin/.active-preset.json"); + let dist_artifacts_ready = uses_dist_preset_artifacts(preset_subdir, committee_str) + && preset_stamp_matches(&dist_stamp, preset_subdir, committee_str); + let bin_artifacts_ready = circuits_bin_marker.exists() + && preset_stamp_matches(&active_bin_stamp, preset_subdir, committee_str); + + if dist_artifacts_ready { copy_dir_recursive(&dist_preset, &preset_out).await?; - } else if !circuits_bin_marker.exists() || !preset_build_stamp.exists() { - // Either no local build exists, or the local build cannot be proven to match - // the requested preset; download the pinned release tarball instead. + } else if !bin_artifacts_ready { + // Either no local build exists, or the local build does not match the requested + // preset and committee. Download the pinned release tarball instead. println!( - "No verifiable local circuit fixtures for preset `{}/{}` \ - (need either dist/circuits/{}/{}/recursive/dkg/pk/pk.json \ - or circuits/bin + circuits/bin/.active-preset.json); \ + "No matching local circuit fixtures for preset `{}/{}` \ + (checked dist/circuits/{}/{}/.build-stamp.json and circuits/bin/.active-preset.json); \ downloading release circuits via ensure_installed()...", preset_subdir, committee_str, preset_subdir, committee_str ); @@ -933,6 +955,71 @@ fn count_projected_events(projected: &[&str], event_type: &str) -> usize { projected.iter().filter(|seen| **seen == event_type).count() } +fn keyshare_parties(history: &[InterfoldEvent], e3_id: &E3id) -> HashSet { + history + .iter() + .filter_map(|event| match event.get_data() { + InterfoldEventData::KeyshareCreated(data) if data.e3_id == *e3_id => { + Some(data.party_id) + } + _ => None, + }) + .collect() +} + +fn aggregator_role_event_at( + history: &[InterfoldEvent], + e3_id: &E3id, + event_index: usize, +) -> Option<(usize, bool)> { + history + .iter() + .take(event_index + 1) + .enumerate() + .rev() + .find_map(|(index, event)| match event.get_data() { + InterfoldEventData::AggregatorChanged(data) if data.e3_id == *e3_id => { + Some((index, data.is_aggregator)) + } + _ => None, + }) +} + +/// Return the local aggregator role at a history event. +fn local_aggregator_role_at( + history: &[InterfoldEvent], + e3_id: &E3id, + event_index: usize, +) -> Option { + aggregator_role_event_at(history, e3_id, event_index).map(|(_, is_aggregator)| is_aggregator) +} + +fn collect_local_aggregate_events( + history: &[InterfoldEvent], + e3_id: &E3id, + node_index: usize, + phase: &str, + mut is_aggregate: F, +) -> Result> +where + F: FnMut(&InterfoldEventData) -> bool, +{ + let mut aggregate_events = Vec::new(); + for (event_index, event) in history.iter().enumerate() { + if event.source() != EventSource::Local || !is_aggregate(event.get_data()) { + continue; + } + if local_aggregator_role_at(history, e3_id, event_index) != Some(true) { + bail!( + "{phase}: node {node_index} recorded {} while inactive at history index {event_index}", + event.event_type() + ); + } + aggregate_events.push((event_index, event.event_id())); + } + Ok(aggregate_events) +} + /// Scan a node history for slashing, accusation, and protocol-fault signals that must not /// appear on an all-honest benchmark run. Catches regressions such as spurious C2→C4 /// commitment mismatches when N > H that completion-only assertions would miss. @@ -1026,43 +1113,6 @@ fn assert_honest_run_safeguards(history: &[InterfoldEvent], e3_id: &E3id, contex ); } -async fn wait_for_history_match( - nodes: &CiphernodeSystem, - node_index: usize, - after: usize, - description: &str, - total_timeout: Duration, - matches: F, -) -> Result -where - F: Fn(&InterfoldEventData) -> bool, -{ - let start = Instant::now(); - loop { - let history = nodes.get_history(node_index).await?; - if history - .iter() - .skip(after) - .any(|event| matches(event.get_data())) - { - return Ok(history); - } - if start.elapsed() >= total_timeout { - let observed_events = history - .iter() - .skip(after) - .map(InterfoldEvent::event_type) - .collect::>(); - bail!( - "Timed out after {:?} while waiting for {description} on node {node_index}; observed events after offset {after}: {:?}", - start.elapsed(), - observed_events - ); - } - sleep(Duration::from_millis(50)).await; - } -} - /// Wall seconds between first `start_when` and last `end_when` event in `history` (HLC physical time). fn history_wall_seconds_between( history: &[InterfoldEvent], @@ -1164,11 +1214,8 @@ fn plaintext_aggregator_marker(data: &InterfoldEventData, e3_id: &E3id) -> Optio if data.e3_id == *e3_id && matches!( &data.request, - ComputeRequestKind::Zk(ZkRequest::VerifyShareProofs(_)) - | ComputeRequestKind::TrBFV(TrBFVRequest::CalculateThresholdDecryption(_)) + ComputeRequestKind::TrBFV(TrBFVRequest::CalculateThresholdDecryption(_)) | ComputeRequestKind::Zk(ZkRequest::DecryptedSharesAggregation(_)) - | ComputeRequestKind::Zk(ZkRequest::NodeDkgFold { .. }) - | ComputeRequestKind::Zk(ZkRequest::DkgAggregation { .. }) | ComputeRequestKind::Zk(ZkRequest::DecryptionAggregation { .. }) ) => { @@ -1178,13 +1225,8 @@ fn plaintext_aggregator_marker(data: &InterfoldEventData, e3_id: &E3id) -> Optio if data.e3_id == *e3_id && matches!( &data.response, - ComputeResponseKind::Zk(ZkResponse::VerifyShareProofs(_)) - | ComputeResponseKind::TrBFV(TrBFVResponse::CalculateThresholdDecryption( - _ - )) + ComputeResponseKind::TrBFV(TrBFVResponse::CalculateThresholdDecryption(_)) | ComputeResponseKind::Zk(ZkResponse::DecryptedSharesAggregation(_)) - | ComputeResponseKind::Zk(ZkResponse::NodeDkgFold(_)) - | ComputeResponseKind::Zk(ZkResponse::DkgAggregation(_)) | ComputeResponseKind::Zk(ZkResponse::DecryptionAggregation(_)) ) => { @@ -1213,6 +1255,56 @@ fn plaintext_aggregator_marker(data: &InterfoldEventData, e3_id: &E3id) -> Optio } } +/// `VerifyShareProofs` is shared by C1 and C6; a C6 proof identifies a C6 request. +fn is_c6_verify_request(request: &ComputeRequestKind) -> bool { + let ComputeRequestKind::Zk(ZkRequest::VerifyShareProofs(request)) = request else { + return false; + }; + + request.party_proofs.iter().any(|party| { + party + .signed_proofs + .iter() + .any(|proof| proof.payload.proof_type == ProofType::C6ThresholdShareDecryption) + }) +} + +fn plaintext_aggregator_events(history: &[InterfoldEvent], e3_id: &E3id) -> Vec<&'static str> { + let mut c6_correlations = HashSet::new(); + + history + .iter() + .filter_map(|event| match event.get_data() { + InterfoldEventData::ComputeRequest(data) + if data.e3_id == *e3_id && is_c6_verify_request(&data.request) => + { + c6_correlations.insert(data.correlation_id); + Some("C6VerifyRequest") + } + InterfoldEventData::ComputeResponse(data) + if data.e3_id == *e3_id + && c6_correlations.contains(&data.correlation_id) + && matches!( + &data.response, + ComputeResponseKind::Zk(ZkResponse::VerifyShareProofs(_)) + ) => + { + Some("C6VerifyResponse") + } + data => plaintext_aggregator_marker(data, e3_id), + }) + .collect() +} + +fn is_c6_share_verification_dispatched(event: &InterfoldEvent, e3_id: &E3id) -> bool { + matches!( + event.get_data(), + InterfoldEventData::ShareVerificationDispatched(data) + if data.e3_id == *e3_id + && data.kind == VerificationKind::ThresholdDecryptionProofs + ) +} + async fn setup_score_sortition_environment( bus: &BusHandle, eth_addrs: &Vec, @@ -1417,8 +1509,9 @@ async fn test_trbfv_actor() -> Result<()> { // Actor system setup let concurrent_jobs = benchmark_multithread_concurrent_jobs(); + let reserve_threads = benchmark_multithread_reserve_threads(); let slashing_manager_addr = benchmark_slashing_manager_address(); - let max_threadroom = Multithread::get_max_threads_minus(1); + let max_threadroom = Multithread::get_max_threads_minus(reserve_threads); let pool_threads = concurrent_jobs.min(max_threadroom).max(1); let task_pool = Multithread::create_taskpool(pool_threads, concurrent_jobs); let multithread_report = MultithreadReport::new(pool_threads, concurrent_jobs).start(); @@ -1675,21 +1768,6 @@ async fn test_trbfv_actor() -> Result<()> { // keyshares into PublicKeyAggregated; DKGRecursiveAggregationComplete is one per member (N). let ks_n: Vec<&'static str> = vec!["KeyshareCreated"; threshold_n]; let dkg_n: Vec<&'static str> = vec!["DKGRecursiveAggregationComplete"; threshold_n]; - let mut active_aggregator_c1_c5: Vec<&'static str> = vec![ - "ShareVerificationDispatched", - "CommitmentConsistencyCheckRequested", - "CommitmentConsistencyCheckComplete", - ]; - // C1 verification dispatches ALL N submitted keyshare proofs (the protocol needs to know - // who's dishonest before it can pick the H honest set), so N ProofVerificationPassed events - // fire. The aggregator subsequently truncates to H for C5 input only. - active_aggregator_c1_c5.extend(std::iter::repeat_n("ProofVerificationPassed", threshold_n)); - active_aggregator_c1_c5.extend_from_slice(&[ - "ShareVerificationComplete", - "PkAggregationProofPending", - "PkAggregationProofSigned", - ]); - let mut expected_events: Vec<&'static str> = vec!["AggregatorChanged"]; if proof_aggregation_enabled { expected_events.extend_from_slice(&ks_n); @@ -1737,22 +1815,21 @@ async fn test_trbfv_actor() -> Result<()> { _ => None, }) .collect(); - // Unlike KeyshareCreated (cheap, gossiped early — all N arrive before aggregation), the - // per-node recursive fold proof (DKGRecursiveAggregationComplete) is expensive and late. // `PublicKeyAggregated` fires once the aggregator selects and aggregates the honest set, so - // only the H honest folds are guaranteed to have reached node 0 by this barrier; the extra - // N-H members' folds race against it (and may land afterward). Assert the guaranteed floor - // and that every observed fold party is a committee member, not the racy `== N`. + // only the H honest inputs are guaranteed to have reached node 0 by this barrier. Both + // gossiped event types can race with the barrier, so assert the guaranteed floor and the + // committee-party bounds instead of requiring all N events. assert!( dkg_parties.len() >= committee_h && dkg_parties.len() <= threshold_n && dkg_parties.iter().all(|p| (*p as usize) < threshold_n), "node 0: expected DKGRecursiveAggregationComplete from {committee_h}..={threshold_n} committee members before PublicKeyAggregated (only the H honest folds are guaranteed by this barrier), got parties {dkg_parties:?}" ); - assert_eq!( - ks_parties.len(), - threshold_n, - "node 0: expected KeyshareCreated from each committee member (N={threshold_n}), got parties {ks_parties:?}" + assert!( + ks_parties.len() >= committee_h + && ks_parties.len() <= threshold_n + && ks_parties.iter().all(|p| (*p as usize) < threshold_n), + "node 0: expected KeyshareCreated from {committee_h}..={threshold_n} committee members before PublicKeyAggregated, got parties {ks_parties:?}" ); let pk_agg = h .iter() @@ -1769,49 +1846,127 @@ async fn test_trbfv_actor() -> Result<()> { ); let active_aggregator_history = nodes.get_history(active_aggregator_index).await?; - let active_aggregator_pubkey_history_len = active_aggregator_history.len(); - let mut expected_active_aggregator_pubkey_events = vec![ - "CommitteeFinalized", - "CiphernodeSelected", - "AggregatorChanged", - ]; - expected_active_aggregator_pubkey_events.extend_from_slice(&ks_n); - expected_active_aggregator_pubkey_events.extend_from_slice(&active_aggregator_c1_c5); - expected_active_aggregator_pubkey_events.push("PublicKeyAggregated"); - - // The active aggregator is also a selected committee member, so its node history contains - // local ThresholdKeyshare DKG work in addition to the public-key aggregation stage. Project - // only the deterministic pubkey-aggregation signals instead of comparing the whole raw node bus. - // - // KeyshareCreated and DKGRecursiveAggregationComplete events are produced independently - // by each committee member and gossiped in parallel with the active aggregator's own - // C1→C5 verification flow, so their positions relative to the C1→C5 sub-sequence are - // non-deterministic. Compare as a multiset plus boundary events rather than strict order. - let active_aggregator_pubkey_events = project_history(&active_aggregator_history, |data| { - publickey_aggregator_marker(data, &e3_id) - }); - let mut actual_sorted = active_aggregator_pubkey_events.clone(); - actual_sorted.retain(|event| *event != "DKGRecursiveAggregationComplete"); - let mut expected_sorted = expected_active_aggregator_pubkey_events.clone(); - actual_sorted.sort(); - expected_sorted.sort(); - assert_eq!( - actual_sorted, expected_sorted, - "Active aggregator public-key flow: event multiset mismatch" - ); - assert_eq!( - active_aggregator_pubkey_events.first().copied(), - Some("CommitteeFinalized"), - "Active aggregator: first event must be CommitteeFinalized" + let active_keyshare_party_ids = keyshare_parties(&active_aggregator_history, &e3_id); + assert!( + active_keyshare_party_ids.len() >= committee_h + && active_keyshare_party_ids.len() <= threshold_n, + "Active aggregator: expected KeyshareCreated from {committee_h}..={threshold_n} committee parties, got {active_keyshare_party_ids:?}" ); - assert_eq!( - active_aggregator_pubkey_events.last().copied(), - Some("PublicKeyAggregated"), - "Active aggregator: last event must be PublicKeyAggregated" + assert!( + active_keyshare_party_ids + .iter() + .all(|party_id| (*party_id as usize) < threshold_n), + "Active aggregator: KeyshareCreated contains a party outside 0..{threshold_n}: {active_keyshare_party_ids:?}" ); + // The winning member must record its own C1→C5 events while it has the aggregator role. + // Other members can contribute gossiped keyshares and public-key results to the same history. + let pubkey_deadline = Instant::now() + BENCHMARK_POST_BARRIER_SETTLE_TIMEOUT; + let mut pubkey_candidate_deadline = None; + let mut pubkey_flow_winner: Option<(usize, Vec)> = None; + loop { + let mut histories = Vec::with_capacity(nodes.len()); + let mut pubkey_producer_ids = HashSet::new(); + for index in 0..nodes.len() { + let history = nodes.get_history(index).await?; + let local_events = collect_local_aggregate_events( + &history, + &e3_id, + index, + "public-key aggregation", + |data| { + matches!( + data, + InterfoldEventData::PublicKeyAggregated(data) if data.e3_id == e3_id + ) + }, + )?; + pubkey_producer_ids.extend(local_events.iter().map(|(_, event_id)| *event_id)); + histories.push((index, history, local_events)); + } + + if pubkey_producer_ids.len() > 1 { + bail!( + "public-key aggregation produced multiple logical local results: {pubkey_producer_ids:?}" + ); + } + + if let Some(&producer_id) = pubkey_producer_ids.iter().next() { + let candidate_deadline = pubkey_candidate_deadline + .get_or_insert_with(|| Instant::now() + BENCHMARK_POST_BARRIER_SETTLE_TIMEOUT); + for (index, history, local_events) in &histories { + if *index == 0 { + continue; + } + let Some(public_key_index) = + local_events + .iter() + .rev() + .find_map(|(event_index, event_id)| { + (*event_id == producer_id).then_some(*event_index) + }) + else { + continue; + }; + let Some((active_role_index, true)) = + aggregator_role_event_at(history, &e3_id, public_key_index) + else { + continue; + }; + let keyshare_party_ids = keyshare_parties(history, &e3_id); + if keyshare_party_ids.len() != threshold_n + || keyshare_party_ids + .iter() + .any(|party_id| (*party_id as usize) >= threshold_n) + { + continue; + } + + let local_pubkey_events: Vec<&str> = history[active_role_index..=public_key_index] + .iter() + .filter(|event| event.source() == EventSource::Local) + .filter_map(|event| publickey_aggregator_marker(event.get_data(), &e3_id)) + .collect(); + let required_local_markers = [ + "ShareVerificationDispatched", + "CommitmentConsistencyCheckRequested", + "CommitmentConsistencyCheckComplete", + "ShareVerificationComplete", + "PkAggregationProofPending", + "PkAggregationProofSigned", + ]; + let local_c1_proof_count = + count_projected_events(&local_pubkey_events, "ProofVerificationPassed"); + if required_local_markers + .iter() + .all(|marker| local_pubkey_events.contains(marker)) + && local_c1_proof_count >= threshold_n + && Instant::now() >= *candidate_deadline + { + pubkey_flow_winner = Some((*index, history.to_vec())); + break; + } + } + } + if pubkey_flow_winner.is_some() { + break; + } + let now = Instant::now(); + let deadline = pubkey_candidate_deadline.unwrap_or(pubkey_deadline); + if now >= deadline { + break; + } + sleep(Duration::from_secs(5)).await; + } + let (_pubkey_flow_winner_index, pubkey_flow_winner_history) = + pubkey_flow_winner.ok_or_else(|| { + anyhow::anyhow!( + "No committee member recorded a local C1→C5 public-key flow after the public-key barrier within {BENCHMARK_POST_BARRIER_SETTLE_TIMEOUT:?}" + ) + })?; + if let Some(secs) = history_wall_seconds_between( - &active_aggregator_history, + &pubkey_flow_winner_history, |d| { matches!( d, @@ -1902,25 +2057,82 @@ async fn test_trbfv_actor() -> Result<()> { println!("CiphertextOutputPublished event has been dispatched!"); - // PlaintextAggregated is a local publication intent and is not gossiped. Wait for it on the - // active aggregator, then inspect the observer for the shared ciphertext and decryption shares. + // PlaintextAggregated is a local publication intent and is not gossiped. Poll committee + // members and continue with the node that recorded the event while it held the aggregator role. println!( - "[bench-progress] waiting for PlaintextAggregated on active aggregator node {active_aggregator_index}" + "[bench-progress] polling all nodes for a local PlaintextAggregated (score-picked node {active_aggregator_index})" ); - let active_aggregator_history = wait_for_history_match( - &nodes, - active_aggregator_index, - active_aggregator_pubkey_history_len, - "PlaintextAggregated", - plaintext_flow_timeout, - |data| matches!(data, InterfoldEventData::PlaintextAggregated(event) if event.e3_id == e3_id), - ) - .await - .map_err(|e| { + let plaintext_deadline = Instant::now() + plaintext_flow_timeout; + let mut plaintext_candidate_deadline = None; + let (plaintext_node_index, active_aggregator_history) = loop { + let mut histories = Vec::with_capacity(nodes.len()); + let mut plaintext_producer_ids = HashSet::new(); + for index in 0..nodes.len() { + let history = nodes.get_history(index).await?; + let local_events = collect_local_aggregate_events( + &history, + &e3_id, + index, + "plaintext aggregation", + |data| { + matches!( + data, + InterfoldEventData::PlaintextAggregated(data) if data.e3_id == e3_id + ) + }, + )?; + plaintext_producer_ids.extend(local_events.iter().map(|(_, event_id)| *event_id)); + histories.push((index, history, local_events)); + } + + if plaintext_producer_ids.len() > 1 { + bail!( + "plaintext aggregation produced multiple logical local results: {plaintext_producer_ids:?}" + ); + } + + let mut found = None; + if let Some(&producer_id) = plaintext_producer_ids.iter().next() { + let candidate_deadline = plaintext_candidate_deadline + .get_or_insert_with(|| Instant::now() + BENCHMARK_POST_BARRIER_SETTLE_TIMEOUT); + if Instant::now() >= *candidate_deadline { + found = histories + .into_iter() + .find_map(|(index, history, local_events)| { + (index != 0 + && local_events + .iter() + .any(|(_, event_id)| *event_id == producer_id)) + .then_some((index, history)) + }); + } + } + if let Some(found) = found { + break found; + } + let now = Instant::now(); + let deadline = plaintext_candidate_deadline.unwrap_or(plaintext_deadline); + if now >= deadline { + bail!("No node recorded a local PlaintextAggregated within {plaintext_flow_timeout:?}"); + } + sleep(Duration::from_secs(5)).await; + }; + // The decryption phase starts at this node's own CiphertextOutputPublished observation; + // everything before it belongs to the public-key phase. + let ciphertext_output_index = active_aggregator_history + .iter() + .position(|event| { + matches!( + event.get_data(), + InterfoldEventData::CiphertextOutputPublished(data) if data.e3_id == e3_id + ) + }) + .ok_or_else(|| { anyhow::anyhow!( - "FAILURE on active aggregator node {active_aggregator_index} plaintext flow: {e}" + "plaintext node {plaintext_node_index}: missing CiphertextOutputPublished" ) })?; + println!("[bench-progress] PlaintextAggregated observed on node {plaintext_node_index}"); let observer_history = nodes.get_history(0).await?; let actual_types = project_history(&observer_history, |data| match data { @@ -1952,21 +2164,18 @@ async fn test_trbfv_actor() -> Result<()> { _ => None, }) .collect(); - // All N committee members that received share material attempt decryption and gossip; the - // aggregator consumes only H. So the number of distinct senders observed at the collector - // sits in [H, N]. - assert!( - unique_ds_parties.len() >= committee_h && unique_ds_parties.len() <= threshold_n, - "collector: expected DecryptionshareCreated from {committee_h}..={threshold_n} distinct parties, got {} parties {unique_ds_parties:?}", + // Only the canonical H honest members generate decryption shares. The collector still sees + // the shares through the event bus, but no non-roster member should publish one. + assert_eq!( + unique_ds_parties.len(), + committee_h, + "collector: expected DecryptionshareCreated from exactly H={committee_h} honest parties, got {} parties {unique_ds_parties:?}", unique_ds_parties.len() ); - println!( - "[bench-progress] PlaintextAggregated observed on active aggregator node {active_aggregator_index}" - ); - let active_aggregator_plaintext_events = project_history( - &active_aggregator_history[active_aggregator_pubkey_history_len..], - |data| plaintext_aggregator_marker(data, &e3_id), + let active_aggregator_plaintext_events = plaintext_aggregator_events( + &active_aggregator_history[ciphertext_output_index..], + &e3_id, ); // C6 head layout: @@ -1984,46 +2193,69 @@ async fn test_trbfv_actor() -> Result<()> { .iter() .position(|e| *e == "ShareVerificationDispatched") .expect("ShareVerificationDispatched should be present in plaintext flow"); + let c6_svd_history_index = active_aggregator_history + .iter() + .enumerate() + .skip(ciphertext_output_index + 1) + .find_map(|(index, event)| { + is_c6_share_verification_dispatched(event, &e3_id).then_some(index) + }) + .expect("C6 ShareVerificationDispatched should be present in plaintext history"); let pre_svd = &active_aggregator_plaintext_events[1..svd_index]; assert!( pre_svd.iter().all(|e| *e == "DecryptionshareCreated"), "active aggregator: only DecryptionshareCreated allowed between CiphertextOutputPublished and ShareVerificationDispatched, got {pre_svd:?}" ); let unique_ds_parties_agg: HashSet = active_aggregator_history - [active_aggregator_pubkey_history_len..] + [ciphertext_output_index..c6_svd_history_index] .iter() - .take_while(|e| { - !matches!( - e.get_data(), - InterfoldEventData::ShareVerificationDispatched(_) - ) - }) .filter_map(|e| match e.get_data() { InterfoldEventData::DecryptionshareCreated(d) if d.e3_id == e3_id => Some(d.party_id), _ => None, }) .collect(); - assert!( - unique_ds_parties_agg.len() >= committee_h && unique_ds_parties_agg.len() <= threshold_n, - "active aggregator: expected DecryptionshareCreated from {committee_h}..={threshold_n} distinct parties before ShareVerificationDispatched, got {} parties {unique_ds_parties_agg:?}", + assert_eq!( + unique_ds_parties_agg.len(), + committee_h, + "active aggregator: expected DecryptionshareCreated from exactly H={committee_h} parties before ShareVerificationDispatched, got {} parties {unique_ds_parties_agg:?}", unique_ds_parties_agg.len() ); + let c6_head_end = active_aggregator_plaintext_events + .iter() + .enumerate() + .skip(svd_index) + .filter(|(_, event)| **event != "DecryptionshareCreated") + .nth(2) + .map(|(index, _)| index + 1) + .expect("C6 verification head should contain three markers"); + let c6_head: Vec<&str> = active_aggregator_plaintext_events[svd_index..c6_head_end] + .iter() + .copied() + .filter(|event| *event != "DecryptionshareCreated") + .collect(); assert_eq!( - &active_aggregator_plaintext_events[svd_index..svd_index + 3], - &[ + c6_head, + vec![ "ShareVerificationDispatched", "CommitmentConsistencyCheckRequested", "CommitmentConsistencyCheckComplete", - ][..], + ], "Unexpected active aggregator C6 head after ShareVerificationDispatched" ); - let c6_head_end = svd_index + 3; let aggregation_pending_index = active_aggregator_plaintext_events .iter() .position(|event| *event == "AggregationProofPending") .expect("AggregationProofPending should be present"); let c6_body = &active_aggregator_plaintext_events[c6_head_end..aggregation_pending_index]; + assert!( + count_projected_events(c6_body, "C6VerifyRequest") >= 1, + "expected at least one C6 verification request before aggregation" + ); + assert!( + count_projected_events(c6_body, "C6VerifyResponse") >= 1, + "expected at least one paired C6 verification response before aggregation" + ); assert_eq!( count_projected_events(c6_body, "ShareVerificationComplete"), 1, @@ -2093,7 +2325,7 @@ async fn test_trbfv_actor() -> Result<()> { ); if let Some(secs) = history_wall_seconds_between( - &active_aggregator_history[active_aggregator_pubkey_history_len..], + &active_aggregator_history[ciphertext_output_index..], |d| { matches!( d,