diff --git a/crates/hotshot/new-protocol/bench/src/metrics.rs b/crates/hotshot/new-protocol/bench/src/metrics.rs index 195c687f3c2..4f6add5ba8b 100644 --- a/crates/hotshot/new-protocol/bench/src/metrics.rs +++ b/crates/hotshot/new-protocol/bench/src/metrics.rs @@ -13,7 +13,6 @@ pub struct ViewMetrics { pub is_leader: bool, pub header_created_ns: Option, pub block_built_ns: Option, - pub vid_disperse_ns: Option, pub proposal_sent_ns: Option, pub proposal_recv_ns: Option, pub state_validated_ns: Option, @@ -65,10 +64,6 @@ impl MetricsCollector { let v = **view; self.view_mut(v).block_built_ns = Some(ts); }, - ConsensusInput::VidDisperseCreated(view, _) => { - let v = **view; - self.view_mut(v).vid_disperse_ns = Some(ts); - }, // Replica: proposal processing ConsensusInput::Proposal(_sender, p) => { let v = *p.view_number(); diff --git a/crates/hotshot/new-protocol/bench/src/node.rs b/crates/hotshot/new-protocol/bench/src/node.rs index cf3cc6e625c..6948b52b9ec 100644 --- a/crates/hotshot/new-protocol/bench/src/node.rs +++ b/crates/hotshot/new-protocol/bench/src/node.rs @@ -1,7 +1,7 @@ use std::{path::PathBuf, sync::Arc}; use anyhow::Result; -use hotshot::{traits::BlockPayload, types::BLSPubKey}; +use hotshot::types::BLSPubKey; use hotshot_example_types::{ block_types::{TestBlockHeader, TestBlockPayload, TestMetadata, TestTransaction}, node_types::{TEST_VERSIONS, TestTypes}, @@ -9,29 +9,34 @@ use hotshot_example_types::{ storage_types::TestStorage, }; use hotshot_new_protocol::{ - block::{BlockBuilder, BlockBuilderConfig}, + block::{BlockAndHeaderRequest, BlockBuilder, BlockBuilderConfig}, cert_verifier::CertVerifiers, client::CoordinatorClient, consensus::{Consensus, ConsensusInput, ConsensusOutput}, - coordinator::{Coordinator, timer::Timer}, + coordinator::{Coordinator, error::Severity, timer::Timer}, epoch::EpochManager, helpers::proposal_commitment, - network::Cliquenet, + network::{Cliquenet, Sender}, outbox::Outbox, proposal::{ProposalValidator, VidShareValidator}, state::StateManager, - vid::{VidDisperser, VidReconstructor}, + vid::{VidReconstructor, fanout}, vote::VoteCollector, }; use hotshot_types::{ PeerConnectInfo, addr::NetAddr, - data::{EpochNumber, Leaf2, ViewNumber}, + data::{EpochNumber, Leaf2, VidCommitment, VidDisperse2, ViewNumber}, epoch_membership::EpochMembershipCoordinator, message::UpgradeLock, - traits::{metrics::NoMetrics, node_implementation::NodeType, signature_key::SignatureKey}, + traits::{ + EncodeBytes, metrics::NoMetrics, node_implementation::NodeType, signature_key::SignatureKey, + }, + utils::BuilderCommitment, + vid::avidm_gf2::{AvidmGf2Commitment, AvidmGf2Scheme}, x25519::Keypair, }; +use tokio::task::JoinSet; use tracing::{error, info, warn}; use versions::{NEW_PROTOCOL_VERSION, Upgrade}; @@ -39,6 +44,19 @@ use crate::{config::NodeConfig, membership::make_membership, metrics::MetricsCol type BenchCoordinator = Coordinator>; +/// State the bench keeps to disperse injected blocks itself. +/// +/// The bench injects synthetic blocks straight as `BlockBuilt`, bypassing the +/// real `BlockBuilder`. Since VID dispersal now lives in the builder, the bench +/// must fan the shares out on its own — this bundles what `fan_out` needs. +#[derive(Clone)] +struct BenchDisperser { + network: Sender, + membership: EpochMembershipCoordinator, + public_key: BLSPubKey, + private_key: ::PrivateKey, +} + /// Build and run a single benchmark node. pub async fn run(cfg: NodeConfig) -> Result<()> { let (public_key, private_key) = BLSPubKey::generated_from_seed_indexed([0u8; 32], cfg.node_id); @@ -47,10 +65,17 @@ pub async fn run(cfg: NodeConfig) -> Result<()> { let (membership, client) = make_membership(cfg.total_nodes, public_key).await; let network = create_network(cfg.node_id, &public_key, &private_key, &cfg).await?; + let disperser = BenchDisperser { + network: network.sender().clone(), + membership: membership.clone(), + public_key, + private_key: private_key.clone(), + }; + let coordinator = build_coordinator(public_key, private_key, membership, network, client, &cfg).await; - run_instrumented(coordinator, &cfg).await + run_instrumented(coordinator, &cfg, disperser).await } async fn create_network( @@ -141,20 +166,15 @@ async fn build_coordinator( let epoch_manager = EpochManager::new(epoch_height, membership.clone()); - let vid_disperser = VidDisperser::new( - membership.clone(), - network.sender().clone(), - public_key, - private_key.clone(), - ); - let vid_reconstructor = VidReconstructor::new(); - let block_config = BlockBuilderConfig::default(); let block_builder = BlockBuilder::new( instance.clone(), membership.clone(), - block_config, + network.sender().clone(), + public_key, + private_key.clone(), + BlockBuilderConfig::default(), upgrade_lock.clone(), ); @@ -203,7 +223,6 @@ async fn build_coordinator( .timeout_one_honest_collector(timeout_one_honest_collector) .epoch_root_collector(epoch_root_collector) .cert_verifiers(CertVerifiers::new(membership.clone(), upgrade_lock.clone())) - .vid_disperser(vid_disperser) .vid_reconstructor(vid_reconstructor) .epoch_manager(epoch_manager) .block_builder(block_builder) @@ -234,7 +253,11 @@ async fn build_coordinator( } /// Run coordinator with metrics instrumentation and block injection. -async fn run_instrumented(mut coordinator: BenchCoordinator, cfg: &NodeConfig) -> Result<()> { +async fn run_instrumented( + mut coordinator: BenchCoordinator, + cfg: &NodeConfig, + disperser: BenchDisperser, +) -> Result<()> { let mut metrics = MetricsCollector::new(cfg.node_id); let output_path = PathBuf::from(&cfg.output_file); @@ -244,63 +267,56 @@ async fn run_instrumented(mut coordinator: BenchCoordinator, cfg: &NodeConfig) - "entering event loop" ); + // Test blocks are built off the event loop (as the real BlockBuilder is) and + // injected once ready, so the loop keeps serving consensus — votes, certs, + // messages — while a block is being built, rather than blocking on it. + let mut builds: JoinSet = JoinSet::new(); + loop { - match coordinator.next_consensus_input().await { - Ok(input) => { - metrics.on_input(&input); - coordinator.apply_consensus(input); - }, - Err(err) - if err.severity == hotshot_new_protocol::coordinator::error::Severity::Critical => - { - error!(%err, "critical error in consensus input"); - metrics.write_csv(&output_path)?; - return Err(anyhow::anyhow!("{err}")); + tokio::select! { + biased; + Some(built) = builds.join_next() => match built { + Ok(built) => inject_test_block(&mut coordinator, &mut metrics, built), + Err(err) => error!(%err, "test block build task panicked"), }, - Err(err) => { - warn!(%err, "recoverable error in consensus input"); - continue; + result = coordinator.next_consensus_input() => match result { + Ok(input) => { + metrics.on_input(&input); + coordinator.apply_consensus(input); + }, + Err(err) if err.severity == Severity::Critical => { + error!(%err, "critical error in consensus input"); + metrics.write_csv(&output_path)?; + return Err(anyhow::anyhow!("{err}")); + }, + Err(err) => { + warn!(%err, "recoverable error in consensus input"); + continue; + }, }, } while let Some(output) = coordinator.outbox_mut().pop_front() { metrics.on_output(&output); - // Intercept block requests and inject test block (bypassing BlockBuilder). + // Intercept block requests: build + erasure-code on a blocking task + // (bypassing the real BlockBuilder), then inject the result once ready + // via the `builds` JoinSet (see `inject_test_block`). if let ConsensusOutput::RequestBlockAndHeader(ref req) = output && cfg.block_size > 0 { - let block = build_test_block(cfg.block_size, cfg.total_nodes); - let parent_leaf = req.parent_proposal.clone().into(); - let version = bench_upgrade_lock().version_infallible(req.view); - let header = TestBlockHeader::new::( - &parent_leaf, - block.payload_commitment, - block.builder_commitment, - block.metadata, - version, - ); - let header_input = ConsensusInput::HeaderCreated( - req.view, - proposal_commitment(&req.parent_proposal), - header, - ); - metrics.on_input(&header_input); - coordinator.apply_consensus(header_input); - let block_input = ConsensusInput::BlockBuilt { - view: req.view, - epoch: req.epoch, - payload: block.block, - metadata: block.metadata, - payload_commitment: block.payload_commitment, - }; - metrics.on_input(&block_input); - coordinator.apply_consensus(block_input); + let disperser = disperser.clone(); + let size = cfg.block_size; + let req = req.clone(); + builds.spawn_blocking(move || { + let td = build_test_block(size, &disperser, req.view, req.epoch); + BuiltBlock { req, td } + }); continue; // skip process_consensus_output for this one } if let Err(err) = coordinator.process_consensus_output(output) { - if err.severity == hotshot_new_protocol::coordinator::error::Severity::Critical { + if err.severity == Severity::Critical { error!(%err, "critical error processing output"); metrics.write_csv(&output_path)?; return Err(anyhow::anyhow!("{err}")); @@ -323,17 +339,71 @@ async fn run_instrumented(mut coordinator: BenchCoordinator, cfg: &NodeConfig) - } } -/// Build a test block with a single transaction of the given size. +/// A test block of the given size and its payload commitment. +/// +/// [`build_test_block`] mirrors the real `BlockBuilder`: it erasure-codes the +/// block once, fans the shares out to the committee, and returns only what the +/// caller needs to inject the block — the payload, metadata, and the commitment +/// derived from that same dispersal. struct TestBlock { block: TestBlockPayload, metadata: TestMetadata, - payload_commitment: hotshot_types::data::VidCommitment, - builder_commitment: hotshot_types::utils::BuilderCommitment, + commitment: AvidmGf2Commitment, +} + +/// A test block built off the event loop, paired with the request it answers, +/// sent back to the loop for injection. +struct BuiltBlock { + req: BlockAndHeaderRequest, + td: TestBlock, } -fn build_test_block(size: usize, num_nodes: usize) -> TestBlock { - use hotshot_types::traits::EncodeBytes; +/// Inject a built test block as if the `BlockBuilder` had produced it, feeding +/// the header and the block into the coordinator. +fn inject_test_block( + coordinator: &mut BenchCoordinator, + metrics: &mut MetricsCollector, + built: BuiltBlock, +) { + let BuiltBlock { req, td } = built; + let payload = Arc::new(td.block); + // `builder_commitment` is being deprecated and the bench never checks it, so + // inject a dummy rather than computing it from the payload. + let builder_commitment = BuilderCommitment::from_bytes([]); + let payload_commitment = VidCommitment::V2(td.commitment); + + let parent_leaf = req.parent_proposal.clone().into(); + let version = bench_upgrade_lock().version_infallible(req.view); + let header = TestBlockHeader::new::( + &parent_leaf, + payload_commitment, + builder_commitment, + td.metadata, + version, + ); + let header_input = + ConsensusInput::HeaderCreated(req.view, proposal_commitment(&req.parent_proposal), header); + metrics.on_input(&header_input); + coordinator.apply_consensus(header_input); + + let block_input = ConsensusInput::BlockBuilt { + view: req.view, + epoch: req.epoch, + payload, + payload_commitment, + }; + metrics.on_input(&block_input); + coordinator.apply_consensus(block_input); +} +/// Build and disperse a test block. Synchronous and CPU-bound (erasure coding); +/// callers run it on a blocking task so it stays off the async runtime. +fn build_test_block( + size: usize, + disperser: &BenchDisperser, + view: ViewNumber, + epoch: EpochNumber, +) -> TestBlock { let tx = TestTransaction::new(vec![0u8; size]); let block = TestBlockPayload { transactions: vec![tx], @@ -341,21 +411,56 @@ fn build_test_block(size: usize, num_nodes: usize) -> TestBlock { let metadata = TestMetadata { num_transactions: 1, }; - // Use the actual committee size so the commitment matches what - // VidDisperse::calculate_vid_disperse will produce. - let payload_commitment = hotshot_types::data::vid_commitment( - &block.encode(), - &metadata.encode(), - num_nodes, - versions::NEW_PROTOCOL_VERSION, - ); - let builder_commitment = - >::builder_commitment(&block, &metadata); + + // Erasure-code the block, deriving the commitment from that computation. + let params = VidDisperse2::::disperse_params( + block.encode(), + metadata.encode().as_ref(), + &disperser.membership, + Some(epoch), + ) + .expect("resolve dispersal params"); + let (commitment, common, shares) = AvidmGf2Scheme::ns_disperse( + ¶ms.param, + ¶ms.weights, + ¶ms.payload, + params.ns_table.iter().cloned(), + ) + .expect("erasure-code test block"); + + // Fan the shares out on background tasks, including the leader's own share + // via loopback (as the production BlockBuilder does). A watcher surfaces + // failures and panics; the build does not wait for the fanout, so the + // proposal is not gated on it. + let recipients = params.recipients; + let network = disperser.network.clone(); + let public_key = disperser.public_key; + let private_key = disperser.private_key.clone(); + let fanout_handle = tokio::task::spawn_blocking(move || { + fanout::fan_out::( + shares, + common, + commitment, + recipients, + view, + epoch, + network, + public_key, + private_key, + ) + }); + tokio::spawn(async move { + match fanout_handle.await { + Ok(Ok(())) => {}, + Ok(Err(err)) => error!(%view, %err, "bench vid fanout failed"), + Err(err) => error!(%view, %err, "bench vid fanout task panicked"), + } + }); + TestBlock { block, metadata, - payload_commitment, - builder_commitment, + commitment, } } diff --git a/crates/hotshot/new-protocol/src/block.rs b/crates/hotshot/new-protocol/src/block.rs index 26aab497609..d76e41d978b 100644 --- a/crates/hotshot/new-protocol/src/block.rs +++ b/crates/hotshot/new-protocol/src/block.rs @@ -7,9 +7,8 @@ use std::{ use committable::{Commitment, Committable}; use hotshot::traits::{BlockPayload, ValidatedState as _}; use hotshot_types::{ - consensus::PayloadWithMetadata, data::{ - EpochNumber, Leaf2, VidCommitment, ViewNumber, vid_commitment, + EpochNumber, Leaf2, VidCommitment, VidDisperse2, ViewNumber, vid_commitment, vid_disperse::vid_total_weight, }, epoch_membership::EpochMembershipCoordinator, @@ -18,21 +17,25 @@ use hotshot_types::{ EncodeBytes, block_contents::{BuilderFee, Transaction}, node_implementation::NodeType, - signature_key::BuilderSignatureKey, + signature_key::{BuilderSignatureKey, SignatureKey}, }, utils::BuilderCommitment, + vid::avidm_gf2::AvidmGf2Scheme, }; use tokio::{ - task::{AbortHandle, JoinSet}, + task::{AbortHandle, JoinSet, spawn_blocking}, time::sleep, }; use tracing::{error, warn}; +use versions::NEW_PROTOCOL_VERSION; use crate::{ consensus::ConsensusInput, helpers::proposal_commitment, message::{DedupManifest, Proposal, TransactionMessage}, + network::Sender, state::HeaderRequest, + vid::fanout, }; #[derive(Debug, thiserror::Error)] @@ -45,6 +48,8 @@ pub enum BlockError { #[error("builder signature failed")] BuilderSignature, + #[error("vid dispersal failed: {0}")] + VidDisperse(String), } #[derive(Clone, Eq, PartialEq, Debug)] @@ -57,7 +62,8 @@ pub struct BlockAndHeaderRequest { pub struct BlockBuilderOutput { pub view: ViewNumber, pub epoch: EpochNumber, - pub payload: PayloadWithMetadata, + pub payload: Arc, + pub metadata: >::Metadata, pub parent_proposal: Proposal, pub builder_commitment: BuilderCommitment, pub builder_fee: BuilderFee, @@ -92,6 +98,9 @@ struct RetryEntry { pub struct BlockBuilder { instance: Arc, membership: EpochMembershipCoordinator, + network: Sender, + public_key: T::SignatureKey, + private_key: ::PrivateKey, retry_pending: HashMap, RetryEntry>, retry_total_bytes: u64, leader_buffer: HashMap, T::Transaction>, @@ -109,15 +118,22 @@ pub struct BlockBuilder { } impl BlockBuilder { + #[allow(clippy::too_many_arguments)] pub fn new( instance: Arc, membership: EpochMembershipCoordinator, + network: Sender, + public_key: T::SignatureKey, + private_key: ::PrivateKey, config: BlockBuilderConfig, upgrade_lock: UpgradeLock, ) -> Self { Self { instance, membership, + network, + public_key, + private_key, config, upgrade_lock, retry_pending: HashMap::new(), @@ -146,6 +162,9 @@ impl BlockBuilder { self.leader_total_bytes = 0; let instance = self.instance.clone(); let membership = self.membership.clone(); + let network = self.network.clone(); + let public_key = self.public_key.clone(); + let private_key = self.private_key.clone(); let handle = self.tasks.spawn(async move { // Throttle empty block production: when no transactions are pending, @@ -167,18 +186,75 @@ impl BlockBuilder { T::BlockPayload::from_transactions(txs, &validated_state, &instance) .await .map_err(|e| BlockError::PayloadConstruction(e.to_string()))?; - let payload: PayloadWithMetadata = PayloadWithMetadata { payload, metadata }; - let payload_bytes = payload.payload.encode(); - let metadata_bytes = payload.metadata.encode(); + let payload_bytes = payload.encode(); + let metadata_bytes = metadata.encode(); + let block_size = payload_bytes.len() as u64; - let total_weight = { - let target_mem = membership - .stake_table_for_epoch(Some(epoch)) - .map_err(|_| BlockError::StakeTableUnavailable)?; - vid_total_weight(target_mem.stake_table(), Some(epoch)) - }; - let payload_commitment = { + // Only the new protocol (>= V0_6) disperses V2/AvidmGf2 shares. During + // the cutover period this builder also produces pre-V0_6 blocks, which + // must carry a version-appropriate commitment so their headers match + // the legacy builder's; those are not dispersed here (the `BlockBuilt` + // handler ignores non-V2 commitments). + let payload_commitment = if version >= NEW_PROTOCOL_VERSION { + // Erasure-code every namespace exactly once and derive the payload + // commitment from that same computation. Runs on a blocking thread; + // the proposal is not gated on the fanout that follows. + let (commitment, common, shares, recipients) = + spawn_blocking(move || -> Result<_, BlockError> { + let params = VidDisperse2::::disperse_params( + payload_bytes, + metadata_bytes.as_ref(), + &membership, + Some(epoch), + ) + .map_err(|e| BlockError::VidDisperse(e.to_string()))?; + let (commitment, common, shares) = AvidmGf2Scheme::ns_disperse( + ¶ms.param, + ¶ms.weights, + ¶ms.payload, + params.ns_table.iter().cloned(), + ) + .map_err(|e| BlockError::VidDisperse(e.to_string()))?; + Ok((commitment, common, shares, params.recipients)) + }) + .await + .map_err(|e| BlockError::VidDisperse(e.to_string()))??; + + // Fan the shares out in the background, including the leader's own + // share (delivered via unicast loopback, which is how the leader + // obtains the share it votes with). A critical send failure is + // logged, not fatal. + let fanout_handle = spawn_blocking(move || { + fanout::fan_out::( + shares, + common, + commitment, + recipients, + view, + epoch, + network, + public_key, + private_key, + ) + }); + // Surface fanout failures and panics; a detached blocking task + // would otherwise swallow them silently. + tokio::spawn(async move { + match fanout_handle.await { + Ok(Ok(())) => {}, + Ok(Err(err)) => error!(%view, %err, "vid share fanout failed"), + Err(err) => error!(%view, %err, "vid share fanout task panicked"), + } + }); + VidCommitment::V2(commitment) + } else { + let total_weight = { + let target_mem = membership + .stake_table_for_epoch(Some(epoch)) + .map_err(|_| BlockError::StakeTableUnavailable)?; + vid_total_weight(target_mem.stake_table(), Some(epoch)) + }; vid_commitment( payload_bytes.as_ref(), metadata_bytes.as_ref(), @@ -187,10 +263,9 @@ impl BlockBuilder { ) }; - let builder_commitment = payload.payload.builder_commitment(&payload.metadata); + let builder_commitment = payload.builder_commitment(&metadata); let (builder_key, builder_private_key) = T::BuilderSignatureKey::generated_from_seed_indexed([0u8; 32], 0); - let block_size = payload_bytes.len() as u64; let offered_fee = block_size; let builder_fee = BuilderFee { fee_amount: offered_fee, @@ -198,14 +273,15 @@ impl BlockBuilder { fee_signature: T::BuilderSignatureKey::sign_fee( &builder_private_key, offered_fee, - &payload.metadata, + &metadata, ) .map_err(|_| BlockError::BuilderSignature)?, }; Ok(BlockBuilderOutput { view, epoch, - payload, + payload: Arc::new(payload), + metadata, parent_proposal: request.parent_proposal, builder_commitment, builder_fee, @@ -369,7 +445,7 @@ impl From<&BlockBuilderOutput> for HeaderRequest { parent_proposal: output.parent_proposal.clone(), payload_commitment: output.payload_commitment, builder_commitment: output.builder_commitment.clone(), - metadata: output.payload.metadata.clone(), + metadata: output.metadata.clone(), builder_fee: output.builder_fee.clone(), } } @@ -380,8 +456,7 @@ impl From> for ConsensusInput { ConsensusInput::BlockBuilt { view: output.view, epoch: output.epoch, - payload: output.payload.payload, - metadata: output.payload.metadata, + payload: output.payload, payload_commitment: output.payload_commitment, } } diff --git a/crates/hotshot/new-protocol/src/consensus.rs b/crates/hotshot/new-protocol/src/consensus.rs index ebe893f2708..cf492698c8b 100644 --- a/crates/hotshot/new-protocol/src/consensus.rs +++ b/crates/hotshot/new-protocol/src/consensus.rs @@ -6,7 +6,6 @@ use std::{ }; use committable::{Commitment, CommitmentBoundsArkless, Committable}; -use hotshot::traits::BlockPayload; use hotshot_contract_adapter::light_client::derive_signed_state_digest; use hotshot_types::{ data::{ @@ -83,8 +82,7 @@ pub enum ConsensusInput { BlockBuilt { view: ViewNumber, epoch: EpochNumber, - payload: T::BlockPayload, - metadata: >::Metadata, + payload: Arc, payload_commitment: VidCommitment, }, BlockReconstructed(ViewNumber, VidCommitment2), @@ -116,7 +114,6 @@ pub enum ConsensusInput { Timeout(ViewNumber, EpochNumber), TimeoutCertificate(ValidCert>), TimeoutOneHonest(ViewNumber, EpochNumber), - VidDisperseCreated(ViewNumber, VidCommitment2), DrbResult(EpochNumber, DrbResult), } @@ -139,13 +136,6 @@ pub enum ConsensusOutput { /// from votes can still decide. Mirrors `SendCertificate1`. SendCertificate2(Certificate2), SendEpochChange(EpochChangeMessage), - RequestVidDisperse { - view: ViewNumber, - epoch: EpochNumber, - payload: T::BlockPayload, - metadata: >::Metadata, - payload_commitment: VidCommitment2, - }, LeafDecided { leaves: Vec>, /// Certificate1 (QC) that certifies the most recent (first) leaf in the chain. @@ -208,7 +198,7 @@ pub struct Consensus { unpaired_vid_shares: UnpairedVidShares, states_verified: BTreeMap>>, blocks_reconstructed: BTreeSet<(ViewNumber, VidCommitment2)>, - blocks: BTreeMap<(ViewNumber, VidCommitment2), T::BlockPayload>, + blocks: BTreeMap<(ViewNumber, VidCommitment2), Arc>, certs: BTreeMap>, certs2: BTreeMap>, timeout_certs: BTreeMap>, @@ -783,29 +773,20 @@ impl Consensus { view, epoch, payload, - metadata, payload_commitment, } => { debug!(%view, %epoch, "apply: block built"); if let VidCommitment::V2(payload_commitment) = payload_commitment { - outbox.push_back(ConsensusOutput::RequestVidDisperse { - view, - epoch, - payload: payload.clone(), - metadata, - payload_commitment, - }); + // We built this block and are dispersing its shares from the + // build task, so the payload is available locally without + // reconstruction (formerly signalled by VidDisperseCreated). + self.blocks_reconstructed.insert((view, payload_commitment)); self.blocks.insert((view, payload_commitment), payload); } else { warn!(%view, %epoch, "block built with non-V2 payload commitment; ignoring"); } Protocol::Continue }, - ConsensusInput::VidDisperseCreated(view, payload_commitment) => { - debug!(%view, "apply: vid disperse created"); - self.blocks_reconstructed.insert((view, payload_commitment)); - Protocol::Continue - }, ConsensusInput::DrbResult(epoch, drb_result) => { info!(%epoch, "apply: drb result"); self.drb_results.insert(epoch, drb_result); @@ -1806,7 +1787,7 @@ impl Consensus { if let VidCommitment::V2(pc) = proposal.block_header.payload_commitment() && let Some(payload) = self.blocks.get(&(view, pc)) { - leaf.fill_block_payload_unchecked(payload.clone()); + leaf.fill_block_payload_unchecked((**payload).clone()); } let mut decided = vec![leaf]; let mut vid_shares = vec![self.signed_vid_share(view)]; @@ -1827,7 +1808,7 @@ impl Consensus { if let VidCommitment::V2(pc) = proposal.block_header.payload_commitment() && let Some(payload) = self.blocks.get(&(parent_view, pc)) { - leaf.fill_block_payload_unchecked(payload.clone()); + leaf.fill_block_payload_unchecked((**payload).clone()); } vid_shares.push(self.signed_vid_share(parent_view)); decided.push(leaf); @@ -2531,7 +2512,6 @@ impl ConsensusInput { // processing is for the next view cert.view_number() + 1 }, - ConsensusInput::VidDisperseCreated(view, _) => *view, // DRB results arrive asynchronously and don't belong to any // particular view; `apply` handles routing by using // `current_view` for this variant. diff --git a/crates/hotshot/new-protocol/src/coordinator.rs b/crates/hotshot/new-protocol/src/coordinator.rs index b67fb7cdb1d..b9b38beff2a 100644 --- a/crates/hotshot/new-protocol/src/coordinator.rs +++ b/crates/hotshot/new-protocol/src/coordinator.rs @@ -14,7 +14,7 @@ use hotshot::{HotShotInitializer, traits::BlockPayload, types::SignatureKey}; use hotshot_types::{ consensus::{ConsensusMetricsValue, ParticipationTracker}, data::{ - EpochNumber, Leaf2, VidCommitment, VidCommitment2, ViewNumber, + EpochNumber, Leaf2, VidCommitment, VidCommitment2, VidDisperseShare2, ViewNumber, vid_disperse::vid_total_weight, }, epoch_membership::EpochMembershipCoordinator, @@ -52,10 +52,10 @@ use crate::{ }, network::Cliquenet, outbox::Outbox, - proposal::{ProposalValidator, VidShareValidator}, + proposal::{ProposalValidator, ValidatedProposal, VidShareValidator}, state::{HeaderRequest, StateEntry, StateManager, StateManagerOutput}, storage::{NewProtocolStorage, Storage}, - vid::{VidDisperseRequest, VidDisperser, VidFragmentAccumulator, VidReconstructor}, + vid::{VidFragmentAccumulator, VidReconstructor}, vote::{EpochRootTally, SimpleTally, VoteCollector}, }; @@ -96,7 +96,6 @@ pub struct Coordinator { state_manager: StateManager, #[builder(default)] client: CoordinatorClient, - vid_disperser: VidDisperser, vid_reconstructor: VidReconstructor, #[builder(default)] vid_fragment_accumulator: VidFragmentAccumulator, @@ -124,6 +123,10 @@ pub struct Coordinator { pending_proposal_fetches: PendingProposalFetches, #[builder(skip)] requested_missing_proposals: HashSet>, + #[builder(default)] + cached_validated_proposals: BTreeMap<(ViewNumber, VidCommitment2), ValidatedProposal>, + #[builder(default)] + cached_vid_shares: BTreeMap<(ViewNumber, VidCommitment2), VidDisperseShare2>, #[builder(skip)] da_payloads: BTreeMap<(ViewNumber, VidCommitment2), PendingDa>, metrics: Option, @@ -274,26 +277,17 @@ where consensus.seed_state_cert(state_cert); } - let participation = ParticipationTracker::new(&membership_coordinator, anchor_epoch); + // Capture a network sender for the block builder's VID fanout before + // `network` is moved into the coordinator below. + let block_builder_sender = network.sender().clone(); - let vid_disperser = VidDisperser::new( - membership_coordinator.clone(), - network.sender().clone(), - public_key.clone(), - private_key.clone(), - ) - .with_metrics( - coordinator_metrics - .as_ref() - .map(|m| m.consensus.vid_disperse_duration.clone().into()), - ); + let participation = ParticipationTracker::new(&membership_coordinator, anchor_epoch); let lock = upgrade_lock.clone(); Self::builder() .consensus(consensus) .network(network) .state_manager(state_manager) - .vid_disperser(vid_disperser) .vid_reconstructor(VidReconstructor::new()) .vote1_collector(VoteCollector::new( membership_coordinator.clone(), @@ -326,6 +320,9 @@ where .block_builder(BlockBuilder::new( Arc::new(initializer.instance_state.clone()), membership_coordinator.clone(), + block_builder_sender, + public_key.clone(), + private_key.clone(), BlockBuilderConfig::default(), upgrade_lock.clone(), )) @@ -377,7 +374,7 @@ where self.storage.append_da( ViewNumber::genesis(), EpochNumber::genesis(), - payload, + Arc::new(payload), metadata, genesis_leaf.payload_commitment(), ); @@ -555,8 +552,8 @@ where (block.view, commit), PendingDa { epoch: block.epoch, - payload: block.payload.payload.clone(), - metadata: block.payload.metadata.clone(), + payload: block.payload.clone(), + metadata: block.metadata.clone(), }, ); } else { @@ -575,14 +572,6 @@ where return Err(CoordinatorError::regular(err).context("block building")) } }, - Some(item) = self.vid_disperser.next() => match item { - Ok(out) => { - return Ok(ConsensusInput::VidDisperseCreated(out.view, out.payload_commitment)) - } - Err(err) => { - return Err(CoordinatorError::from(err).context("vid disperse")) - } - }, Some(item) = self.vid_reconstructor.next() => match item { Ok(out) => { self.payload_txn_bytes.insert(out.view, out.payload.txn_bytes()); @@ -590,7 +579,7 @@ where self.storage.append_da( out.view, out.epoch, - out.payload.clone(), + Arc::new(out.payload.clone()), out.metadata.clone(), VidCommitment::V2(out.payload_commitment), ); @@ -668,22 +657,6 @@ where ); self.state_manager.request_state(state_request); }, - ConsensusOutput::RequestVidDisperse { - view, - epoch, - payload, - metadata, - payload_commitment, - } => { - debug!(%node, %view, %epoch, "request vid disperse"); - self.vid_disperser.request_vid_disperse(VidDisperseRequest { - view, - epoch, - block: payload, - metadata, - payload_commitment, - }); - }, ConsensusOutput::RequestDrbResult(epoch) => { debug!(%node, %epoch, "request drb result"); self.epoch_manager.request_drb_result(epoch); @@ -1789,8 +1762,11 @@ where self.consensus.gc(scope); match scope { GcScope::Local(view) => { + let vc = VidCommitment2::default(); self.block_builder.gc(view); - self.vid_disperser.gc(view); + self.cached_validated_proposals = + self.cached_validated_proposals.split_off(&(view, vc)); + self.cached_vid_shares = self.cached_vid_shares.split_off(&(view, vc)); self.vid_fragment_accumulator.gc(view); // When we enter a new view, we do not want to GC certain data // for the previous view yet: @@ -1965,7 +1941,7 @@ pub enum GcScope { /// A payload built locally and awaiting DA persistence. struct PendingDa { epoch: EpochNumber, - payload: T::BlockPayload, + payload: Arc, metadata: >::Metadata, } diff --git a/crates/hotshot/new-protocol/src/coordinator/error.rs b/crates/hotshot/new-protocol/src/coordinator/error.rs index 39e7c321814..51d4fc60ee3 100644 --- a/crates/hotshot/new-protocol/src/coordinator/error.rs +++ b/crates/hotshot/new-protocol/src/coordinator/error.rs @@ -1,11 +1,8 @@ use std::fmt; use crate::{ - block::BlockError, - epoch::EpochManagerError, - network::NetworkError, - proposal::ValidationError, - vid::{VidDisperseError, VidReconstructError}, + block::BlockError, epoch::EpochManagerError, network::NetworkError, proposal::ValidationError, + vid::VidReconstructError, }; #[derive(Debug, thiserror::Error)] @@ -93,9 +90,6 @@ pub enum ErrorSource { #[error("vid reconstruction error: {0}")] VidReconstruct(String), - - #[error("vid disperse error: {0}")] - VidDisperse(#[from] VidDisperseError), } impl From for CoordinatorError { @@ -108,16 +102,6 @@ impl From for CoordinatorError { } } -impl From for CoordinatorError { - fn from(e: VidDisperseError) -> Self { - if e.is_critical() { - Self::critical(e) - } else { - Self::regular(e) - } - } -} - // `VidReconstructError` is generic over the signature key type, but // `ErrorSource` is not parameterized by it, so flatten to the error's // `Display` (view + kind). The attributable `bad_share_keys` are consumed by diff --git a/crates/hotshot/new-protocol/src/storage.rs b/crates/hotshot/new-protocol/src/storage.rs index d80ac621a88..7f11c5e1e35 100644 --- a/crates/hotshot/new-protocol/src/storage.rs +++ b/crates/hotshot/new-protocol/src/storage.rs @@ -179,7 +179,7 @@ impl> Storage { &mut self, view_number: ViewNumber, epoch: EpochNumber, - block_payload: T::BlockPayload, + block_payload: Arc, metadata: >::Metadata, vid_commit: VidCommitment, ) { diff --git a/crates/hotshot/new-protocol/src/tests/block.rs b/crates/hotshot/new-protocol/src/tests/block.rs index e023be23114..a40d1d7a419 100644 --- a/crates/hotshot/new-protocol/src/tests/block.rs +++ b/crates/hotshot/new-protocol/src/tests/block.rs @@ -1,15 +1,22 @@ -use std::sync::Arc; +use std::{net::Ipv4Addr, sync::Arc}; use committable::Committable; +use hotshot::types::BLSPubKey; use hotshot_example_types::{ block_types::TestTransaction, node_types::TestTypes, state_types::TestInstanceState, }; -use hotshot_types::data::{EpochNumber, ViewNumber}; +use hotshot_types::{ + addr::NetAddr, + data::{EpochNumber, ViewNumber}, + traits::{metrics::NoMetrics, signature_key::SignatureKey}, + x25519::Keypair, +}; use crate::{ block::{BlockBuilder, BlockBuilderConfig}, helpers::test_upgrade_lock, message::{DedupManifest, TransactionMessage}, + network::Cliquenet, tests::common::utils::mock_membership, }; @@ -41,18 +48,42 @@ fn small_config() -> BlockBuilderConfig { } } -fn builder() -> BlockBuilder { +async fn builder_with_config(config: BlockBuilderConfig) -> BlockBuilder { + let (public_key, private_key) = BLSPubKey::generated_from_seed_indexed([0; 32], 0); + let upgrade_lock = test_upgrade_lock(); + let keypair = + Keypair::derive_from::(&private_key).expect("keypair derivation should succeed"); + let port = test_utils::reserve_tcp_port().expect("OS should have ephemeral ports available"); + let addr = NetAddr::Inet(Ipv4Addr::LOCALHOST.into(), port); + let network = Cliquenet::create( + "test-block-builder", + public_key, + keypair, + addr, + vec![], + upgrade_lock.clone(), + Box::new(NoMetrics), + ) + .await + .expect("cliquenet creation should succeed"); BlockBuilder::new( Arc::new(TestInstanceState::default()), mock_membership(), - small_config(), - test_upgrade_lock(), + network.sender().clone(), + public_key, + private_key, + config, + upgrade_lock, ) } +async fn builder() -> BlockBuilder { + builder_with_config(small_config()).await +} + #[tokio::test] async fn test_retry_buffer() { - let mut b = builder(); + let mut b = builder().await; let t1 = tx(1); let t2 = tx(2); b.on_submit_transaction(t1.clone()); @@ -75,7 +106,7 @@ async fn test_retry_buffer() { #[tokio::test] async fn test_leader_buffer_drain() { - let mut b = builder(); + let mut b = builder().await; b.on_transactions(tx_msg(view(1), vec![tx(1), tx(2)])); let (mut txns, manifest) = b.drain(view(1), epoch()); txns.sort_by_key(|t| t.bytes().clone()); @@ -112,7 +143,7 @@ async fn test_request_block_same_view_different_parent_both_produce_output() { block::BlockAndHeaderRequest, helpers::proposal_commitment, tests::common::utils::TestData, }; - let mut b = builder(); + let mut b = builder().await; let test_data = TestData::new(3).await; let parent_a = test_data.views[0].proposal.data.clone(); @@ -150,7 +181,7 @@ async fn test_request_block_same_view_different_parent_both_produce_output() { async fn test_request_block_dedups_same_view_same_parent() { use crate::{block::BlockAndHeaderRequest, tests::common::utils::TestData}; - let mut b = builder(); + let mut b = builder().await; let test_data = TestData::new(2).await; let parent = test_data.views[0].proposal.data.clone(); @@ -169,15 +200,11 @@ async fn test_request_block_dedups_same_view_same_parent() { #[tokio::test] async fn test_dedup_window() { - let mut b = BlockBuilder::new( - Arc::new(TestInstanceState::default()), - mock_membership(), - BlockBuilderConfig { - dedup_window_size: 2, - ..small_config() - }, - test_upgrade_lock(), - ); + let mut b = builder_with_config(BlockBuilderConfig { + dedup_window_size: 2, + ..small_config() + }) + .await; let t = tx(1); b.on_dedup_manifest(DedupManifest { diff --git a/crates/hotshot/new-protocol/src/tests/common/assertions.rs b/crates/hotshot/new-protocol/src/tests/common/assertions.rs index 31ef7ffcbad..0273322d53c 100644 --- a/crates/hotshot/new-protocol/src/tests/common/assertions.rs +++ b/crates/hotshot/new-protocol/src/tests/common/assertions.rs @@ -55,10 +55,6 @@ pub(crate) fn is_persist_proposal(output: &ConsensusOutput) -> bool { matches!(output, ConsensusOutput::PersistProposal(_)) } -pub(crate) fn is_request_vid_disperse(output: &ConsensusOutput) -> bool { - matches!(output, ConsensusOutput::RequestVidDisperse { .. }) -} - pub(crate) fn is_view_changed(output: &ConsensusOutput) -> bool { matches!(output, ConsensusOutput::ViewChanged(..)) } @@ -94,10 +90,6 @@ pub(crate) fn is_cert2(input: &ConsensusInput) -> bool { matches!(input, ConsensusInput::Certificate2(_)) } -pub(crate) fn is_vid_disperse(input: &ConsensusInput) -> bool { - matches!(input, ConsensusInput::VidDisperseCreated(..)) -} - pub(crate) fn is_block_reconstructed(input: &ConsensusInput) -> bool { matches!(input, ConsensusInput::BlockReconstructed(..)) } diff --git a/crates/hotshot/new-protocol/src/tests/common/coordinator_builder.rs b/crates/hotshot/new-protocol/src/tests/common/coordinator_builder.rs index f7ee9090bd9..9f422686825 100644 --- a/crates/hotshot/new-protocol/src/tests/common/coordinator_builder.rs +++ b/crates/hotshot/new-protocol/src/tests/common/coordinator_builder.rs @@ -32,7 +32,7 @@ use crate::{ outbox::Outbox, proposal::{ProposalValidator, VidShareValidator}, state::StateManager, - vid::{VidDisperser, VidReconstructor}, + vid::VidReconstructor, vote::VoteCollector, }; @@ -86,17 +86,14 @@ pub async fn build_test_coordinator( epoch_height, ); - let vid_disperser = VidDisperser::new( - membership.clone(), - network.sender().clone(), - public_key, - private_key.clone(), - ); let vid_reconstructor = VidReconstructor::new(); let block_builder = BlockBuilder::new( instance.clone(), membership.clone(), + network.sender().clone(), + public_key, + private_key.clone(), BlockBuilderConfig::default(), upgrade_lock.clone(), ); @@ -250,7 +247,6 @@ pub async fn build_test_coordinator( .timeout_one_honest_collector(timeout_one_honest_collector) .epoch_root_collector(epoch_root_collector) .cert_verifiers(CertVerifiers::new(membership.clone(), upgrade_lock.clone())) - .vid_disperser(vid_disperser) .vid_reconstructor(vid_reconstructor) .epoch_manager(epoch_manager) .block_builder(block_builder) diff --git a/crates/hotshot/new-protocol/src/tests/common/harness.rs b/crates/hotshot/new-protocol/src/tests/common/harness.rs index d63655ed053..e26f0d5d84d 100644 --- a/crates/hotshot/new-protocol/src/tests/common/harness.rs +++ b/crates/hotshot/new-protocol/src/tests/common/harness.rs @@ -25,7 +25,7 @@ use crate::{ proposal::{ProposalValidator, VidShareValidator}, state::StateManager, tests::common::mock::MockCoordinator, - vid::{VidDisperser, VidReconstructor}, + vid::VidReconstructor, vote::VoteCollector, }; @@ -90,14 +90,6 @@ impl TestHarness { let vid_reconstruction_task = VidReconstructor::new(); - let block_config = BlockBuilderConfig::default(); - let block_builder = BlockBuilder::new( - instance.clone(), - membership.clone(), - block_config, - upgrade_lock.clone(), - ); - let mut state_manager = StateManager::new(instance.clone(), upgrade_lock.clone()); state_manager.seed_state(ViewNumber::genesis(), Arc::new(genesis_state), genesis_leaf); @@ -123,11 +115,14 @@ impl TestHarness { .await .expect("cliquenet creation should succeed"); - let vid_disperse_task = VidDisperser::new( + let block_builder = BlockBuilder::new( + instance.clone(), membership.clone(), network.sender().clone(), public_key, private_key.clone(), + BlockBuilderConfig::default(), + upgrade_lock.clone(), ); let coordinator = MockCoordinator::builder() @@ -140,7 +135,6 @@ impl TestHarness { .timeout_one_honest_collector(timeout_one_honest_collector) .epoch_root_collector(epoch_root_collector) .cert_verifiers(CertVerifiers::new(membership.clone(), upgrade_lock.clone())) - .vid_disperser(vid_disperse_task) .vid_reconstructor(vid_reconstruction_task) .epoch_manager(epoch_manager) .block_builder(block_builder) diff --git a/crates/hotshot/new-protocol/src/tests/common/utils.rs b/crates/hotshot/new-protocol/src/tests/common/utils.rs index ca03cff4222..e88457bb85d 100644 --- a/crates/hotshot/new-protocol/src/tests/common/utils.rs +++ b/crates/hotshot/new-protocol/src/tests/common/utils.rs @@ -26,8 +26,8 @@ use hotshot_testing::{ use hotshot_types::{ PeerConnectInfo, data::{ - EpochNumber, Leaf2, VidCommitment, VidCommitment2, VidDisperse, VidDisperse2, - VidDisperseShare2, ViewNumber, vid_commitment, + EpochNumber, Leaf2, VidCommitment, VidCommitment2, VidDisperse2, VidDisperseShare2, + ViewNumber, vid_commitment, vid_disperse::{AvidmGf2DisperseShareFragment, AvidmGf2NamespacePiece}, }, epoch_membership::EpochMembershipCoordinator, @@ -110,7 +110,7 @@ impl TestView { /// messages the leader unicasts to `recipient_key` for this view — the /// production wire form, where a node reassembles its share from the /// fragments. The leader's signature is over the share's - /// `payload_commitment`, matching what the `VidDisperser` produces. + /// `payload_commitment`, matching what the block builder's fanout produces. /// /// Pair this with `proposal_input` when simulating a leader's send to a /// replica. @@ -979,8 +979,7 @@ pub fn reconstructed_blocks( /// Lightweight consensus-only test harness. Wraps a single [`Consensus`] /// instance and auto-responds to outputs that consensus expects feedback for -/// (`RequestState`, `RequestBlockAndHeader`, `RequestVidDisperse`, -/// `RequestDrbResult`). +/// (`RequestState`, `RequestBlockAndHeader`, `RequestDrbResult`). pub(crate) struct ConsensusHarness { pub consensus: Consensus, pub membership_coordinator: EpochMembershipCoordinator, @@ -1162,37 +1161,12 @@ impl ConsensusHarness { ConsensusInput::BlockBuilt { view: req.view, epoch: req.epoch, - payload: mock_block.block, - metadata: mock_block.metadata, + payload: std::sync::Arc::new(mock_block.block), payload_commitment: mock_block.payload_commitment, }, outbox, ); }, - ConsensusOutput::RequestVidDisperse { - view, - epoch, - payload, - metadata, - payload_commitment: _, - } => { - let vid_disperse = VidDisperse::calculate_vid_disperse( - payload, - &self.membership_coordinator, - *view, - Some(*epoch), - Some(*epoch), - metadata, - &test_upgrade_lock(), - ) - .await - .unwrap(); - let VidDisperse::V2(vid) = vid_disperse.disperse else { - panic!("VidDisperse is not a V2"); - }; - let input = ConsensusInput::VidDisperseCreated(*view, vid.payload_commitment); - self.consensus.apply(input, outbox); - }, ConsensusOutput::RequestDrbResult(epoch) => { self.consensus .apply(ConsensusInput::DrbResult(*epoch, TEST_DRB_RESULT), outbox); diff --git a/crates/hotshot/new-protocol/src/tests/consensus.rs b/crates/hotshot/new-protocol/src/tests/consensus.rs index ab687e76db7..d10d26430ca 100644 --- a/crates/hotshot/new-protocol/src/tests/consensus.rs +++ b/crates/hotshot/new-protocol/src/tests/consensus.rs @@ -984,8 +984,7 @@ async fn test_bridged_legacy_qc_adopts_lock_and_reproposes() { ConsensusInput::BlockBuilt { view: ViewNumber::new(3), epoch: request_epoch, - payload: mock_block.block, - metadata: mock_block.metadata, + payload: std::sync::Arc::new(mock_block.block), payload_commitment: mock_block.payload_commitment, }, &mut outbox, diff --git a/crates/hotshot/new-protocol/src/tests/integration.rs b/crates/hotshot/new-protocol/src/tests/integration.rs index 76557a89220..e9a9f6ee087 100644 --- a/crates/hotshot/new-protocol/src/tests/integration.rs +++ b/crates/hotshot/new-protocol/src/tests/integration.rs @@ -11,10 +11,9 @@ use crate::{ tests::common::assertions::{ any, count_matching, is_block_built, is_block_reconstructed, is_cert1, is_cert2, is_drb_result, is_header_created, is_header_created_for_view, is_leaf_decided, is_proposal, - is_proposal_for_view, is_request_block_and_header, is_request_vid_disperse, is_send_cert1, - is_send_epoch_change, is_send_timeout_vote, is_state_validated, is_timeout, - is_timeout_cert, is_timeout_one_honest, is_vid_disperse, is_view_changed, is_vote1, - is_vote2, node_index_for_key, + is_proposal_for_view, is_request_block_and_header, is_send_cert1, is_send_epoch_change, + is_send_timeout_vote, is_state_validated, is_timeout, is_timeout_cert, + is_timeout_one_honest, is_view_changed, is_vote1, is_vote2, node_index_for_key, }, }; @@ -216,12 +215,11 @@ async fn test_leader_proposal() { && any(inputs, is_state_validated) && any(inputs, is_block_built) && any(inputs, is_header_created) - && any(inputs, is_vid_disperse) }) .await; - // SendProposal proves the CPU VidDisperseTask computed the VID - // disperse — consensus cannot send a proposal without it. + // The leader proposes once the block is built and the header is created; + // VID dispersal fans out in the background and does not gate the proposal. harness .process_until_output(|outputs| any(outputs, is_proposal)) .await; @@ -302,9 +300,7 @@ async fn test_leader_proposes_after_timeout() { harness .process_until(|inputs| { assert!(!any(inputs, is_timeout)); - any(inputs, is_vid_disperse) - && any(inputs, is_block_built) - && any(inputs, is_header_created) + any(inputs, is_block_built) && any(inputs, is_header_created) }) .await; @@ -313,10 +309,6 @@ async fn test_leader_proposes_after_timeout() { "Leader should request block and header after TC" ); - assert!( - any(harness.outputs(), is_request_vid_disperse), - "Leader should request VID disperse after TC" - ); // Proposal with timeout view change evidence, released once stored. harness .process_until_output(|outputs| any(outputs, is_proposal)) diff --git a/crates/hotshot/new-protocol/src/tests/legacy_cutover.rs b/crates/hotshot/new-protocol/src/tests/legacy_cutover.rs index 2c4a988c91e..17eafb678ed 100644 --- a/crates/hotshot/new-protocol/src/tests/legacy_cutover.rs +++ b/crates/hotshot/new-protocol/src/tests/legacy_cutover.rs @@ -231,7 +231,7 @@ async fn build_cutover_coordinator( proposal::{ProposalValidator, VidShareValidator}, state::StateManager, tests::common::coordinator_builder::{build_genesis_cert1, build_genesis_proposal}, - vid::{VidDisperser, VidReconstructor}, + vid::VidReconstructor, vote::VoteCollector, }; @@ -281,6 +281,9 @@ async fn build_cutover_coordinator( let block_builder = BlockBuilder::new( instance.clone(), membership.clone(), + network.sender().clone(), + public_key, + private_key.clone(), BlockBuilderConfig::default(), upgrade_lock.clone(), ); @@ -290,13 +293,6 @@ async fn build_cutover_coordinator( let share_validator = VidShareValidator::new(membership.clone(), epoch_height, upgrade_lock.clone()); - let vid_disperser = VidDisperser::new( - membership.clone(), - network.sender().clone(), - public_key, - private_key.clone(), - ); - Coordinator::builder() .consensus(consensus) .network(network) @@ -307,7 +303,6 @@ async fn build_cutover_coordinator( .timeout_one_honest_collector(VoteCollector::new(membership.clone(), upgrade_lock.clone())) .epoch_root_collector(VoteCollector::new(membership.clone(), upgrade_lock.clone())) .cert_verifiers(CertVerifiers::new(membership.clone(), upgrade_lock.clone())) - .vid_disperser(vid_disperser) .vid_reconstructor(VidReconstructor::new()) .epoch_manager(EpochManager::new(epoch_height, membership.clone())) .block_builder(block_builder) diff --git a/crates/hotshot/new-protocol/src/vid.rs b/crates/hotshot/new-protocol/src/vid.rs index dc3553f6d36..ae0efe326e6 100644 --- a/crates/hotshot/new-protocol/src/vid.rs +++ b/crates/hotshot/new-protocol/src/vid.rs @@ -5,14 +5,15 @@ //! whose shards cover the recovery threshold. This module owns the three stages //! of that lifecycle, one per submodule: //! -//! - [`disperse`] -- the leader side. [`VidDisperser`] erasure-codes each -//! namespace, coalesces namespaces into size-balanced buckets, and unicasts -//! to every node a stream of [`AvidmGf2DisperseShareFragment`] messages -//! (one per bucket), each carrying that node's shares for the bucket's -//! namespaces. +//! - [`fanout`] -- the leader side. The block builder erasure-codes the block +//! once (via `NsAvidmGf2Scheme::ns_disperse`, which yields the commitment) and +//! [`fan_out`] coalesces namespaces into size-balanced buckets and unicasts to +//! every node — including the leader itself, via loopback — a stream of +//! [`AvidmGf2DisperseShareFragment`] messages (one per bucket), each carrying +//! that node's shares for the bucket's namespaces. //! //! - [`fragments`] -- the receive side of dispersal, the mirror of -//! [`VidDisperser`]. [`VidFragmentAccumulator`] buffers the fragments a node +//! [`fanout`]. [`VidFragmentAccumulator`] buffers the fragments a node //! receives for its *own* share and, once every namespace has arrived, //! reassembles them into a complete [`VidDisperseShare2`]. That share is then //! verified, attached to this node's vote, and fed to the reconstructor. @@ -24,12 +25,12 @@ //! //! [`AvidmGf2DisperseShareFragment`]: hotshot_types::data::vid_disperse::AvidmGf2DisperseShareFragment //! [`VidDisperseShare2`]: hotshot_types::data::VidDisperseShare2 +//! [`fan_out`]: fanout::fan_out -mod disperse; +pub mod fanout; mod fragments; mod reconstruct; -pub use disperse::{VidDisperseError, VidDisperseOutput, VidDisperseRequest, VidDisperser}; pub use fragments::{VidFragmentAccumulator, VidFragmentError}; pub use reconstruct::{ VidReconstructError, VidReconstructErrorKind, VidReconstructOutput, VidReconstructor, diff --git a/crates/hotshot/new-protocol/src/vid/disperse.rs b/crates/hotshot/new-protocol/src/vid/disperse.rs deleted file mode 100644 index ac7ff360896..00000000000 --- a/crates/hotshot/new-protocol/src/vid/disperse.rs +++ /dev/null @@ -1,352 +0,0 @@ -use std::{ - collections::BTreeMap, - mem, - ops::Range, - sync::{Arc, LazyLock}, -}; - -use hotshot::traits::BlockPayload; -use hotshot_types::{ - data::{ - EpochNumber, VidCommitment2, VidDisperse2, ViewNumber, - vid_disperse::{AvidmGf2DisperseShareFragment, AvidmGf2NamespacePiece}, - }, - epoch_membership::EpochMembershipCoordinator, - message::Proposal as SignedProposal, - traits::{metrics::Histogram, node_implementation::NodeType, signature_key::SignatureKey}, - vid::avidm_gf2::AvidmGf2Scheme, -}; -use hotshot_utils::anytrace::{self, Wrap}; -use rayon::prelude::*; -use tokio::task::{AbortHandle, JoinSet}; -use tracing::{error, warn}; - -use crate::{ - coordinator::metrics::{Measurement, finish_measurement}, - message::{ConsensusMessage, Message, MessageType}, - network::{NetworkError, Sender}, -}; - -static NUM_THREADS: LazyLock = LazyLock::new(|| rayon::current_num_threads().max(1)); - -#[derive(Clone, Eq, PartialEq, Debug)] -pub struct VidDisperseRequest { - pub view: ViewNumber, - pub epoch: EpochNumber, - pub block: T::BlockPayload, - pub metadata: >::Metadata, - pub payload_commitment: VidCommitment2, -} - -pub struct VidDisperseOutput { - pub view: ViewNumber, - pub payload_commitment: VidCommitment2, -} - -pub struct VidDisperser { - calculations: BTreeMap<(ViewNumber, VidCommitment2), AbortHandle>, - epoch_membership_coordinator: EpochMembershipCoordinator, - network: Sender, - public_key: T::SignatureKey, - private_key: ::PrivateKey, - tasks: JoinSet>, - duration_metric: Option>, -} - -impl VidDisperser { - pub fn new( - epoch_membership_coordinator: EpochMembershipCoordinator, - network: Sender, - public_key: T::SignatureKey, - private_key: ::PrivateKey, - ) -> Self { - Self { - calculations: BTreeMap::new(), - epoch_membership_coordinator, - network, - public_key, - private_key, - tasks: JoinSet::new(), - duration_metric: None, - } - } - - pub fn with_metrics(mut self, hist: Option>) -> Self { - self.duration_metric = hist; - self - } - - pub fn request_vid_disperse(&mut self, vid_disperse_request: VidDisperseRequest) { - let key = ( - vid_disperse_request.view, - vid_disperse_request.payload_commitment, - ); - if self.calculations.contains_key(&key) { - return; - } - let membership = self.epoch_membership_coordinator.clone(); - let network = self.network.clone(); - let public_key = self.public_key.clone(); - let private_key = self.private_key.clone(); - let duration_metric = self.duration_metric.clone(); - let handle = self.tasks.spawn_blocking(move || { - let measurement = duration_metric.map(Measurement::start); - let result = handle_vid_disperse_request( - membership, - network, - public_key, - private_key, - vid_disperse_request, - ); - finish_measurement(measurement); - result - }); - self.calculations.insert(key, handle); - } - - pub async fn next(&mut self) -> Option> { - loop { - match self.tasks.join_next().await { - Some(Ok(result)) => return Some(result), - Some(Err(err)) => { - if err.is_panic() { - error!(%err, "vid disperse task panic"); - } - continue; - }, - None => return None, - } - } - } - - pub fn gc(&mut self, view_number: ViewNumber) { - let keep = self - .calculations - .split_off(&(view_number, VidCommitment2::default())); - for handle in self.calculations.values_mut() { - handle.abort(); - } - self.calculations = keep; - } -} - -fn handle_vid_disperse_request( - epoch_membership_coordinator: EpochMembershipCoordinator, - network: Sender, - public_key: T::SignatureKey, - private_key: ::PrivateKey, - vid_disperse_request: VidDisperseRequest, -) -> Result { - let view = vid_disperse_request.view; - let epoch = vid_disperse_request.epoch; - let payload_commitment = vid_disperse_request.payload_commitment; - - let params = VidDisperse2::::disperse_params( - &vid_disperse_request.block, - &epoch_membership_coordinator, - Some(epoch), - &vid_disperse_request.metadata, - ) - .map_err(VidDisperseError::Vid)?; - - let signature = T::SignatureKey::sign(&private_key, payload_commitment.as_ref()) - .map_err(|err| VidDisperseError::Sign(err.into()))?; - - let num_namespaces = params.ns_table.len(); - - // Coalesce namespaces into size-balanced buckets, roughly one per core, - // but never below the minimum size, so a block of many tiny namespaces - // goes out as a few balanced messages per recipient instead of one tiny - // message each. Each bucket is one parallel unit and one message per - // recipient. - let buckets: Vec> = { - // We want buckets to be at least 256 KiB, otherwise the per-message - // overhead is too large. Larger payloads are divided by the number - // of available threads to fully utilise rayon. - let threshold = params.payload.len().div_ceil(*NUM_THREADS).max(256 * 1024); - bucketize(¶ms.ns_table, threshold) - }; - - buckets.par_iter().try_for_each_with( - network, - |network, bucket| -> Result<(), VidDisperseError> { - let mut pieces = vec![Vec::new(); params.recipients.len()]; - - for &ns_index in bucket { - let dispersal = AvidmGf2Scheme::ns_disperse_one( - ¶ms.param, - ¶ms.weights, - ¶ms.payload[params.ns_table[ns_index].clone()], - ns_index, - ) - .wrap() - .map_err(VidDisperseError::Vid)?; - - let ns_payload_byte_len = dispersal.payload_byte_len; - let ns_commit = dispersal.commit; - for (pieces, ns_share) in pieces.iter_mut().zip(dispersal.shares) { - pieces.push(AvidmGf2NamespacePiece { - ns_index: dispersal.ns_index, - ns_payload_byte_len, - ns_commit, - ns_share, - }); - } - } - - for (recipient, pieces) in params.recipients.iter().zip(pieces) { - let fragment = AvidmGf2DisperseShareFragment { - view_number: view, - epoch: Some(epoch), - target_epoch: Some(epoch), - payload_commitment, - recipient_key: recipient.clone(), - param: params.param.clone(), - num_namespaces, - namespaces: pieces, - }; - let message = Message { - sender: public_key.clone(), - message_type: MessageType::Consensus(ConsensusMessage::VidShareFragment( - SignedProposal::new(fragment, signature.clone()), - )), - }; - if let Err(err) = network.unicast(view, recipient, &message) { - if err.is_critical() { - return Err(err.into()); - } - warn!(%err, "network error while sending vid share fragment"); - } - } - Ok(()) - }, - )?; - - Ok(VidDisperseOutput { - view, - payload_commitment, - }) -} - -/// Group namespace indices into buckets whose payload sizes are each at least -/// `threshold` bytes (except possibly the last), coalescing small namespaces so -/// the disperser sends one message per bucket rather than one per namespace. -/// -/// A `threshold` of 0 puts every namespace in its own bucket (no coalescing); a -/// threshold larger than the whole payload yields a single bucket. A namespace -/// at least `threshold` bytes on its own seals its bucket immediately, so large -/// namespaces stay separate while small ones accumulate. -fn bucketize(ns_table: &[Range], threshold: usize) -> Vec> { - let mut buckets = Vec::new(); - let mut current = Vec::new(); - let mut current_bytes = 0usize; - for (ns_index, range) in ns_table.iter().enumerate() { - current.push(ns_index); - current_bytes += range.len(); - if current_bytes >= threshold { - buckets.push(mem::take(&mut current)); - current_bytes = 0; - } - } - if !current.is_empty() { - buckets.push(current); - } - buckets -} - -#[derive(Debug, thiserror::Error)] -pub enum VidDisperseError { - #[error("network error: {0}")] - Net(#[from] NetworkError), - - #[error("vid error: {0}")] - Vid(#[source] anytrace::Error), - - #[error("sign error: {0}")] - Sign(#[source] Box), -} - -impl VidDisperseError { - pub fn is_critical(&self) -> bool { - match self { - Self::Net(e) => e.is_critical(), - Self::Vid(_) | Self::Sign(_) => false, - } - } -} - -#[cfg(test)] -mod tests { - use std::ops::Range; - - use quickcheck::{TestResult, quickcheck}; - - use super::bucketize; - - /// Build a contiguous namespace table from per-namespace byte lengths. - /// Only the lengths matter to `bucketize`; the offsets are incidental. - fn ns_ranges(lens: &[u8]) -> Vec> { - let mut start = 0; - lens.iter() - .map(|&len| { - let range = start..start + usize::from(len); - start += usize::from(len); - range - }) - .collect() - } - - /// Total payload bytes of the namespaces at `indices`. - fn bytes(indices: &[usize], ns_table: &[Range]) -> usize { - indices.iter().map(|&i| ns_table[i].len()).sum() - } - - quickcheck! { - /// The buckets partition the namespace indices: every index appears - /// exactly once, contiguously, and in namespace-table order. This pins - /// completeness, disjointness, and ordering in a single property. - fn prop_buckets_partition(lens: Vec, threshold: u16) -> bool { - let ns_table = ns_ranges(&lens); - bucketize(&ns_table, threshold.into()) - .into_iter() - .flatten() - .eq(0..ns_table.len()) - } - - /// No bucket is ever empty. - fn prop_buckets_non_empty(lens: Vec, threshold: u16) -> bool { - let ns_table = ns_ranges(&lens); - bucketize(&ns_table, threshold.into()) - .iter() - .all(|bucket| !bucket.is_empty()) - } - - /// Every bucket except the last reaches the threshold: small namespaces - /// are coalesced until the bucket is large enough. - fn prop_non_last_buckets_meet_threshold(lens: Vec, threshold: u16) -> bool { - let ns_table = ns_ranges(&lens); - let buckets = bucketize(&ns_table, threshold.into()); - let non_last = buckets.len().saturating_sub(1); - buckets - .iter() - .take(non_last) - .all(|bucket| bytes(bucket, &ns_table) >= usize::from(threshold)) - } - - /// Buckets are minimal: dropping a non-last bucket's final namespace - /// leaves it below the threshold, so the disperser seals as soon as it - /// reaches the cutoff rather than over-coalescing. (Vacuous at - /// `threshold == 0`, where every namespace is its own bucket.) - fn prop_non_last_buckets_are_minimal(lens: Vec, threshold: u16) -> TestResult { - if threshold == 0 { - return TestResult::discard(); - } - let ns_table = ns_ranges(&lens); - let buckets = bucketize(&ns_table, threshold.into()); - let non_last = buckets.len().saturating_sub(1); - TestResult::from_bool(buckets.iter().take(non_last).all(|bucket| { - bytes(&bucket[..bucket.len() - 1], &ns_table) < usize::from(threshold) - })) - } - } -} diff --git a/crates/hotshot/new-protocol/src/vid/fanout.rs b/crates/hotshot/new-protocol/src/vid/fanout.rs new file mode 100644 index 00000000000..90e0bf3af83 --- /dev/null +++ b/crates/hotshot/new-protocol/src/vid/fanout.rs @@ -0,0 +1,208 @@ +//! Leader-side VID dispersal: erasure-code a block once and fan the shares out. +//! +//! The block builder Reed-Solomon-encodes every namespace a single time (via +//! `NsAvidmGf2Scheme::ns_disperse`, which also yields the payload commitment) and +//! then hands the shares to [`fan_out`] (on a background task) which coalesces +//! namespaces into size-balanced buckets and unicasts every node — including the +//! leader itself, via loopback — a stream of [`AvidmGf2DisperseShareFragment`] +//! messages (one per bucket). + +use std::{mem, sync::LazyLock}; + +use hotshot_types::{ + data::{ + EpochNumber, ViewNumber, + vid_disperse::{AvidmGf2DisperseShareFragment, AvidmGf2NamespacePiece}, + }, + message::Proposal as SignedProposal, + traits::{node_implementation::NodeType, signature_key::SignatureKey}, + vid::avidm_gf2::{AvidmGf2Commitment, AvidmGf2Common, AvidmGf2Share}, +}; +use rayon::prelude::*; +use tracing::warn; + +use crate::{ + message::{ConsensusMessage, Message, MessageType}, + network::{NetworkError, Sender}, +}; + +static NUM_THREADS: LazyLock = LazyLock::new(|| rayon::current_num_threads().max(1)); + +/// Fan the encoded shares out to every recipient. +/// +/// Coalesces namespaces into size-balanced buckets and assembles one +/// [`AvidmGf2DisperseShareFragment`] per (bucket, recipient), then unicasts it. +/// The recipient list includes the leader itself; the loopback send is how the +/// leader obtains its own share to vote. Parallel over recipients, overlapping +/// serialization with the network sends of others. +#[allow(clippy::too_many_arguments)] +pub fn fan_out( + shares: Vec, + common: AvidmGf2Common, + payload_commitment: AvidmGf2Commitment, + recipients: Vec, + view: ViewNumber, + epoch: EpochNumber, + network: Sender, + public_key: T::SignatureKey, + private_key: ::PrivateKey, +) -> Result<(), FanoutError> { + let signature = T::SignatureKey::sign(&private_key, payload_commitment.as_ref()) + .map_err(|err| FanoutError::Sign(err.into()))?; + + let num_namespaces = common.ns_lens.len(); + // We want buckets to be at least 256 KiB, otherwise the per-message overhead + // is too large. Larger payloads are divided by the number of available + // threads to fully utilise rayon. + let threshold = common + .payload_byte_len() + .div_ceil(*NUM_THREADS) + .max(256 * 1024); + let buckets = bucketize(&common.ns_lens, threshold); + + shares.into_par_iter().zip(recipients).try_for_each_with( + network, + |network, (share, recipient)| -> Result<(), FanoutError> { + // Buckets partition the namespaces contiguously in order, so draining + // this recipient's shares in lockstep with the buckets pairs each + // share with its namespace. + let mut ns_shares = share.into_ns_shares().into_iter(); + for bucket in &buckets { + let namespaces: Vec<_> = bucket + .iter() + .zip(ns_shares.by_ref()) + .map(|(&ns_index, ns_share)| AvidmGf2NamespacePiece { + ns_index, + ns_payload_byte_len: common.ns_lens[ns_index], + ns_commit: common.ns_commits[ns_index], + ns_share, + }) + .collect(); + let fragment = AvidmGf2DisperseShareFragment { + view_number: view, + epoch: Some(epoch), + target_epoch: Some(epoch), + payload_commitment, + recipient_key: recipient.clone(), + param: common.param.clone(), + num_namespaces, + namespaces, + }; + let message = Message { + sender: public_key.clone(), + message_type: MessageType::Consensus(ConsensusMessage::VidShareFragment( + SignedProposal::new(fragment, signature.clone()), + )), + }; + if let Err(err) = network.unicast(view, &recipient, &message) { + if err.is_critical() { + return Err(err.into()); + } + warn!(%err, "network error while sending vid share fragment"); + } + } + Ok(()) + }, + ) +} + +/// Group namespace indices into buckets whose payload sizes are each at least +/// `threshold` bytes (except possibly the last), coalescing small namespaces so +/// the disperser sends one message per bucket rather than one per namespace. +/// +/// A `threshold` of 0 puts every namespace in its own bucket (no coalescing); a +/// threshold larger than the whole payload yields a single bucket. A namespace +/// at least `threshold` bytes on its own seals its bucket immediately, so large +/// namespaces stay separate while small ones accumulate. +fn bucketize(ns_lens: &[usize], threshold: usize) -> Vec> { + let mut buckets = Vec::new(); + let mut current = Vec::new(); + let mut current_bytes = 0usize; + for (ns_index, &len) in ns_lens.iter().enumerate() { + current.push(ns_index); + current_bytes += len; + if current_bytes >= threshold { + buckets.push(mem::take(&mut current)); + current_bytes = 0; + } + } + if !current.is_empty() { + buckets.push(current); + } + buckets +} + +#[derive(Debug, thiserror::Error)] +pub enum FanoutError { + #[error("network error: {0}")] + Net(#[from] NetworkError), + + #[error("sign error: {0}")] + Sign(#[source] Box), +} + +#[cfg(test)] +mod tests { + use quickcheck::{TestResult, quickcheck}; + + use super::bucketize; + + /// Per-namespace byte lengths as `usize`. + fn ns_lens(lens: &[u8]) -> Vec { + lens.iter().map(|&len| usize::from(len)).collect() + } + + /// Total payload bytes of the namespaces at `indices`. + fn bytes(indices: &[usize], ns_lens: &[usize]) -> usize { + indices.iter().map(|&i| ns_lens[i]).sum() + } + + quickcheck! { + /// The buckets partition the namespace indices: every index appears + /// exactly once, contiguously, and in namespace-table order. This pins + /// completeness, disjointness, and ordering in a single property. + fn prop_buckets_partition(lens: Vec, threshold: u16) -> bool { + let ns_lens = ns_lens(&lens); + bucketize(&ns_lens, threshold.into()) + .into_iter() + .flatten() + .eq(0..ns_lens.len()) + } + + /// No bucket is ever empty. + fn prop_buckets_non_empty(lens: Vec, threshold: u16) -> bool { + let ns_lens = ns_lens(&lens); + bucketize(&ns_lens, threshold.into()) + .iter() + .all(|bucket| !bucket.is_empty()) + } + + /// Every bucket except the last reaches the threshold: small namespaces + /// are coalesced until the bucket is large enough. + fn prop_non_last_buckets_meet_threshold(lens: Vec, threshold: u16) -> bool { + let ns_lens = ns_lens(&lens); + let buckets = bucketize(&ns_lens, threshold.into()); + let non_last = buckets.len().saturating_sub(1); + buckets + .iter() + .take(non_last) + .all(|bucket| bytes(bucket, &ns_lens) >= usize::from(threshold)) + } + + /// Buckets are minimal: dropping a non-last bucket's final namespace + /// leaves it below the threshold, so the disperser seals as soon as it + /// reaches the cutoff rather than over-coalescing. (Vacuous at + /// `threshold == 0`, where every namespace is its own bucket.) + fn prop_non_last_buckets_are_minimal(lens: Vec, threshold: u16) -> TestResult { + if threshold == 0 { + return TestResult::discard(); + } + let ns_lens = ns_lens(&lens); + let buckets = bucketize(&ns_lens, threshold.into()); + let non_last = buckets.len().saturating_sub(1); + TestResult::from_bool(buckets.iter().take(non_last).all(|bucket| { + bytes(&bucket[..bucket.len() - 1], &ns_lens) < usize::from(threshold) + })) + } + } +} diff --git a/crates/hotshot/types/src/data/vid_disperse.rs b/crates/hotshot/types/src/data/vid_disperse.rs index f5c6c8e32d5..3ce4917c1ce 100644 --- a/crates/hotshot/types/src/data/vid_disperse.rs +++ b/crates/hotshot/types/src/data/vid_disperse.rs @@ -830,17 +830,19 @@ impl AvidmGf2Disperse { self.payload_byte_len as u32 } - /// Resolve the inputs needed to disperse `payload` to `target_epoch`'s - /// committee one namespace at a time. + /// Resolve the inputs needed to disperse an already-encoded payload to + /// `target_epoch`'s committee one namespace at a time. /// - /// Mirrors the setup in [`Self::calculate_vid_disperse`] but returns the - /// owned parameters instead of performing the dispersal, leaving the (heavy, - /// per-namespace) computation to the caller. + /// Mirrors the setup in [`Self::calculate_vid_disperse`] but takes the + /// encoded `payload`/`metadata_bytes` (so a caller that already encoded them + /// need not do so again) and returns the owned parameters instead of + /// performing the dispersal, leaving the (heavy, per-namespace) computation + /// to the caller. pub fn disperse_params( - payload: &TYPES::BlockPayload, + payload: Arc<[u8]>, + metadata_bytes: &[u8], membership: &EpochMembershipCoordinator, target_epoch: Option, - metadata: &>::Metadata, ) -> Result> { let target_mem = membership.stake_table_for_epoch(target_epoch)?; let stake_table: Vec<_> = target_mem.stake_table().cloned().collect(); @@ -852,8 +854,7 @@ impl AvidmGf2Disperse { .iter() .map(|entry| entry.stake_table_entry.public_key()) .collect(); - let payload = payload.encode(); - let ns_table = parse_ns_table(payload.len(), &metadata.encode()); + let ns_table = parse_ns_table(payload.len(), metadata_bytes); let param = init_avidm_gf2_param(total_weight)?; Ok(AvidmGf2DisperseParams { param, diff --git a/vid/src/avidm_gf2/namespaced.rs b/vid/src/avidm_gf2/namespaced.rs index ab3d865b7be..4ef685e7885 100644 --- a/vid/src/avidm_gf2/namespaced.rs +++ b/vid/src/avidm_gf2/namespaced.rs @@ -71,6 +71,14 @@ impl NsAvidmGf2Share { self.0.get(ns_index).cloned() } + /// Consume the share, yielding its per-namespace shares in namespace order. + /// + /// Lets a caller move each namespace's share out (e.g. to build per-namespace + /// dispersal fragments) without cloning. + pub fn into_ns_shares(self) -> Vec { + self.0 + } + /// The shard range this share covers, identical across all namespaces. /// `None` if the share is empty or its namespaces disagree on the range. pub fn range(&self) -> Option<&Range> { @@ -105,7 +113,7 @@ impl NsAvidmGf2Scheme { let ns_table = ns_table.into_iter().collect::>(); let ns_lens = ns_table.iter().map(|r| r.len()).collect::>(); let ns_commits = ns_table - .into_iter() + .into_par_iter() .map(|ns_range| AvidmGf2Scheme::commit(param, &payload[ns_range])) .collect::, _>>()?; let common = NsAvidmGf2Common { @@ -113,10 +121,20 @@ impl NsAvidmGf2Scheme { ns_commits, ns_lens, }; - let commit = MerkleTree::from_elems(None, common.ns_commits.iter().map(|c| c.commit)) + let commit = Self::aggregate_commit(&common.ns_commits)?; + Ok((commit, common)) + } + + /// Aggregate per-namespace commitments into the top-level namespaced + /// commitment (the Merkle root over the namespace commits). + /// + /// Lets a caller that has already computed each namespace's commit (e.g. via + /// [`Self::ns_disperse_one`]) form the block commitment without re-encoding. + pub fn aggregate_commit(ns_commits: &[AvidmGf2Commit]) -> VidResult { + let commit = MerkleTree::from_elems(None, ns_commits.iter().map(|c| c.commit)) .map_err(|err| VidError::Internal(err.into()))? .commitment(); - Ok((NsAvidmGf2Commit { commit }, common)) + Ok(NsAvidmGf2Commit { commit }) } /// Check whether the namespaced commitment is consistent with the common data @@ -161,11 +179,7 @@ impl NsAvidmGf2Scheme { ns_commits, ns_lens, }; - let commit = NsAvidmGf2Commit { - commit: MerkleTree::from_elems(None, common.ns_commits.iter().map(|c| c.commit)) - .map_err(|err| VidError::Internal(err.into()))? - .commitment(), - }; + let commit = Self::aggregate_commit(&common.ns_commits)?; let mut shares = vec![NsAvidmGf2Share::default(); num_storage_nodes]; disperses.into_iter().for_each(|ns_disperse| { shares @@ -227,11 +241,7 @@ impl NsAvidmGf2Scheme { ns_commits, ns_lens, }; - let commit = NsAvidmGf2Commit { - commit: MerkleTree::from_elems(None, common.ns_commits.iter().map(|c| c.commit)) - .map_err(|err| VidError::Internal(err.into()))? - .commitment(), - }; + let commit = Self::aggregate_commit(&common.ns_commits)?; let mut shares = vec![NsAvidmGf2Share::default(); num_storage_nodes]; disperses.into_iter().for_each(|ns_disperse| { shares @@ -510,4 +520,37 @@ pub mod tests { let payload_recovered = NsAvidmGf2Scheme::recover(&common, &shares[..cut_index]).unwrap(); assert_eq!(payload_recovered, payload); } + + /// The block builder encodes each namespace with `ns_disperse_one` and folds + /// the per-namespace commits with `aggregate_commit`; that must yield the + /// same commitment as the monolithic `commit`, so the proposal and the + /// dispersed shares agree. + #[test] + fn aggregate_commit_matches_commit() { + let num_storage_nodes = 9; + let ns_table = [(0usize..15), (15..48), (48..49)]; + let weights = vec![1u32; num_storage_nodes]; + let params = NsAvidmGf2Scheme::setup(3, num_storage_nodes).unwrap(); + let payload: Vec = (0..49).map(|i| i as u8).collect(); + + let ns_commits: Vec<_> = ns_table + .iter() + .enumerate() + .map(|(ns_index, range)| { + NsAvidmGf2Scheme::ns_disperse_one( + ¶ms, + &weights, + &payload[range.clone()], + ns_index, + ) + .unwrap() + .commit + }) + .collect(); + let aggregated = NsAvidmGf2Scheme::aggregate_commit(&ns_commits).unwrap(); + + let (expected, _) = + NsAvidmGf2Scheme::commit(¶ms, &payload, ns_table.iter().cloned()).unwrap(); + assert_eq!(aggregated, expected); + } }