hotshot_types/data/

vid_disperse.rs

// Copyright (c) 2021-2024 Espresso Systems (espressosys.com)
// This file is part of the HotShot repository.

// You should have received a copy of the MIT License
// along with the HotShot repository. If not, see <https://mit-license.org/>.

//! This module provides types for VID disperse related data structures.
//!
//! We have three types of VID disperse related structs:
//!
//! 1. `ADVZ*`: VID V0, The most original VID scheme, which has guaranteed recovery but very inefficient.
//! 2. `AvidM*`: VID V1, the efficient VID scheme, where we use it after the epocn upgrade. It's more
//!    efficient but doesn't guarantee recovery. A VID V1 commitment could correspond to some junk
//!    data, there'll be a proof of incorrect encoding in this case.
//! 3. `AvidmGf2*`: VID V2, almost the same as VID V1 but we have a much more efficient recovery
//!    implementation.

use std::{
    collections::BTreeMap, fmt::Debug, hash::Hash, marker::PhantomData, ops::Range, sync::Arc,
    time::Duration,
};

use alloy_primitives::U256;
use hotshot_utils::anytrace::*;
use jf_advz::{VidDisperse as JfVidDisperse, VidScheme};
use serde::{Deserialize, Serialize};
use tokio::{task::spawn_blocking, time::Instant};

use super::ns_table::parse_ns_table;
use crate::{
    PeerConfig,
    data::{EpochNumber, ViewNumber},
    epoch_membership::{EpochMembership, EpochMembershipCoordinator},
    message::Proposal,
    simple_vote::HasEpoch,
    traits::{
        BlockPayload,
        block_contents::EncodeBytes,
        node_implementation::NodeType,
        signature_key::{SignatureKey, StakeTableEntryType},
    },
    vid::{
        advz::{ADVZCommitment, ADVZCommon, ADVZScheme, ADVZShare, advz_scheme},
        avidm::{AvidMCommitment, AvidMCommon, AvidMScheme, AvidMShare, init_avidm_param},
        avidm_gf2::{
            AvidmGf2Commitment, AvidmGf2Common, AvidmGf2Param, AvidmGf2Scheme, AvidmGf2Share,
            init_avidm_gf2_param,
        },
    },
    vote::HasViewNumber,
};

impl<NODE: NodeType> HasEpoch for ADVZDisperse<NODE> {
    fn epoch(&self) -> Option<EpochNumber> {
        self.epoch
    }
}

impl<NODE: NodeType> HasEpoch for AvidMDisperse<NODE> {
    fn epoch(&self) -> Option<EpochNumber> {
        self.epoch
    }
}

impl<NODE: NodeType> HasEpoch for AvidMDisperseShare<NODE> {
    fn epoch(&self) -> Option<EpochNumber> {
        self.epoch
    }
}

impl<NODE: NodeType> HasEpoch for AvidmGf2Disperse<NODE> {
    fn epoch(&self) -> Option<EpochNumber> {
        self.epoch
    }
}

impl<NODE: NodeType> HasEpoch for AvidmGf2DisperseShare<NODE> {
    fn epoch(&self) -> Option<EpochNumber> {
        self.epoch
    }
}

/// ADVZ dispersal data
#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
pub struct ADVZDisperse<TYPES: NodeType> {
    /// The view number for which this VID data is intended
    pub view_number: ViewNumber,
    /// Epoch the data of this proposal belongs to
    pub epoch: Option<EpochNumber>,
    /// Epoch to which the recipients of this VID belong to
    pub target_epoch: Option<EpochNumber>,
    /// VidCommitment calculated based on the number of nodes in `target_epoch`.
    pub payload_commitment: ADVZCommitment,
    /// A storage node's key and its corresponding VID share
    pub shares: BTreeMap<TYPES::SignatureKey, ADVZShare>,
    /// VID common data sent to all storage nodes
    pub common: ADVZCommon,
}

impl<TYPES: NodeType> HasViewNumber for ADVZDisperse<TYPES> {
    fn view_number(&self) -> ViewNumber {
        self.view_number
    }
}

impl<TYPES: NodeType> ADVZDisperse<TYPES> {
    /// Create VID dispersal from a specified membership for the target epoch.
    /// Uses the specified function to calculate share dispersal
    /// Allows for more complex stake table functionality
    async fn from_membership(
        view_number: ViewNumber,
        mut vid_disperse: JfVidDisperse<ADVZScheme>,
        membership: &EpochMembershipCoordinator<TYPES>,
        target_epoch: Option<EpochNumber>,
        data_epoch: Option<EpochNumber>,
    ) -> Result<Self> {
        let shares = membership
            .stake_table_for_epoch(target_epoch)?
            .stake_table()
            .map(|entry| entry.stake_table_entry.public_key())
            .map(|node| (node.clone(), vid_disperse.shares.remove(0)))
            .collect();

        Ok(Self {
            view_number,
            shares,
            common: vid_disperse.common,
            payload_commitment: vid_disperse.commit,
            epoch: data_epoch,
            target_epoch,
        })
    }

    /// Calculate the vid disperse information from the payload given a view, epoch and membership,
    /// If the sender epoch is missing, it means it's the same as the target epoch.
    ///
    /// # Errors
    /// Returns an error if the disperse or commitment calculation fails
    #[allow(clippy::panic)]
    pub async fn calculate_vid_disperse(
        payload: &TYPES::BlockPayload,
        membership: &EpochMembershipCoordinator<TYPES>,
        view: ViewNumber,
        target_epoch: Option<EpochNumber>,
        data_epoch: Option<EpochNumber>,
    ) -> Result<(Self, Duration)> {
        let num_nodes = membership
            .stake_table_for_epoch(target_epoch)?
            .total_nodes();

        let txns = payload.encode();

        let now = Instant::now();
        let vid_disperse = spawn_blocking(move || advz_scheme(num_nodes).disperse(&txns))
            .await
            .wrap()
            .context(error!("Join error"))?
            .wrap()
            .context(|err| error!("Failed to calculate VID disperse. Error: {err}"))?;
        let advz_scheme_duration = now.elapsed();

        Ok((
            Self::from_membership(view, vid_disperse, membership, target_epoch, data_epoch).await?,
            advz_scheme_duration,
        ))
    }

    /// This function splits a VID disperse into individual shares.
    pub fn to_shares(self) -> Vec<ADVZDisperseShare<TYPES>> {
        self.shares
            .into_iter()
            .map(|(recipient_key, share)| ADVZDisperseShare {
                share,
                recipient_key,
                view_number: self.view_number,
                common: self.common.clone(),
                payload_commitment: self.payload_commitment,
            })
            .collect()
    }

    /// Split a VID disperse into a share proposal for each recipient.
    pub fn to_share_proposals(
        self,
        signature: &<<TYPES as NodeType>::SignatureKey as SignatureKey>::PureAssembledSignatureType,
    ) -> Vec<Proposal<TYPES, ADVZDisperseShare<TYPES>>> {
        self.shares
            .into_iter()
            .map(|(recipient_key, share)| Proposal {
                data: ADVZDisperseShare {
                    share,
                    recipient_key,
                    view_number: self.view_number,
                    payload_commitment: self.payload_commitment,
                    common: self.common.clone(),
                },
                signature: signature.clone(),
                _pd: PhantomData,
            })
            .collect()
    }

    /// Returns the payload length in bytes.
    pub fn payload_byte_len(&self) -> u32 {
        ADVZScheme::get_payload_byte_len(&self.common)
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
/// ADVZ share and associated metadata for a single node
pub struct ADVZDisperseShare<TYPES: NodeType> {
    /// The view number for which this VID data is intended
    pub view_number: ViewNumber,
    /// Block payload commitment
    pub payload_commitment: ADVZCommitment,
    /// A storage node's key and its corresponding VID share
    pub share: ADVZShare,
    /// VID common data sent to all storage nodes
    pub common: ADVZCommon,
    /// a public key of the share recipient
    pub recipient_key: TYPES::SignatureKey,
}

impl<NODE: NodeType> HasEpoch for ADVZDisperseShare<NODE> {
    fn epoch(&self) -> Option<EpochNumber> {
        None
    }
}

impl<TYPES: NodeType> HasViewNumber for ADVZDisperseShare<TYPES> {
    fn view_number(&self) -> ViewNumber {
        self.view_number
    }
}

impl<TYPES: NodeType> ADVZDisperseShare<TYPES> {
    /// Consume `self` and return a `Proposal`
    pub fn to_proposal(
        self,
        private_key: &<TYPES::SignatureKey as SignatureKey>::PrivateKey,
    ) -> Option<Proposal<TYPES, Self>> {
        let Ok(signature) =
            TYPES::SignatureKey::sign(private_key, self.payload_commitment.as_ref())
        else {
            tracing::error!("VID: failed to sign dispersal share payload");
            return None;
        };
        Some(Proposal {
            signature,
            _pd: PhantomData,
            data: self,
        })
    }

    /// Create `VidDisperse` out of an iterator to `VidDisperseShare`s
    pub fn to_advz_disperse<'a, I>(mut it: I) -> Option<ADVZDisperse<TYPES>>
    where
        I: Iterator<Item = &'a Self>,
    {
        let first_vid_disperse_share = it.next()?.clone();
        let mut share_map = BTreeMap::new();
        share_map.insert(
            first_vid_disperse_share.recipient_key,
            first_vid_disperse_share.share,
        );
        let mut vid_disperse = ADVZDisperse {
            view_number: first_vid_disperse_share.view_number,
            epoch: None,
            target_epoch: None,
            payload_commitment: first_vid_disperse_share.payload_commitment,
            common: first_vid_disperse_share.common,
            shares: share_map,
        };
        it.for_each(|vid_disperse_share| {
            vid_disperse.shares.insert(
                vid_disperse_share.recipient_key.clone(),
                vid_disperse_share.share.clone(),
            );
        });
        Some(vid_disperse)
    }

    /// Check if vid common is consistent with the commitment.
    pub fn is_consistent(&self) -> bool {
        ADVZScheme::is_consistent(&self.payload_commitment, &self.common).is_ok()
    }

    /// Verify share assuming common data is already verified consistent.
    /// Caller MUST call `is_consistent()` first.
    pub fn verify_with_verified_common(&self) -> bool {
        let total_weight = ADVZScheme::get_num_storage_nodes(&self.common) as usize;
        advz_scheme(total_weight)
            .verify_share(&self.share, &self.common, &self.payload_commitment)
            .is_ok_and(|r| r.is_ok())
    }

    /// Internally verify the share given necessary information
    pub fn verify(&self, _total_weight: usize) -> bool {
        self.is_consistent() && self.verify_with_verified_common()
    }

    /// Returns the payload length in bytes.
    pub fn payload_byte_len(&self) -> u32 {
        ADVZScheme::get_payload_byte_len(&self.common)
    }
}

/// AvidM dispersal data
#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
pub struct AvidMDisperse<TYPES: NodeType> {
    /// The view number for which this VID data is intended
    pub view_number: ViewNumber,
    /// Epoch the data of this proposal belongs to
    pub epoch: Option<EpochNumber>,
    /// Epoch to which the recipients of this VID belong to
    pub target_epoch: Option<EpochNumber>,
    /// VidCommitment calculated based on the number of nodes in `target_epoch`.
    pub payload_commitment: AvidMCommitment,
    /// A storage node's key and its corresponding VID share
    pub shares: BTreeMap<TYPES::SignatureKey, AvidMShare>,
    /// Length of payload in bytes
    pub payload_byte_len: usize,
    /// VID common data sent to all storage nodes
    pub common: AvidMCommon,
}

impl<TYPES: NodeType> HasViewNumber for AvidMDisperse<TYPES> {
    fn view_number(&self) -> ViewNumber {
        self.view_number
    }
}

/// The target total stake to scale to for VID.
pub const VID_TARGET_TOTAL_STAKE: u32 = 1000;

/// The weights and total weight used in VID calculations
struct Weights {
    // weights, in stake table order
    weights: Vec<u32>,

    // total weight
    total_weight: usize,
}

pub fn vid_total_weight<'a, T, I>(stake_table: I, epoch: Option<EpochNumber>) -> usize
where
    T: NodeType,
    I: Iterator<Item = &'a PeerConfig<T>>,
{
    if epoch.is_none() {
        stake_table
            .fold(U256::ZERO, |acc, entry| {
                acc + entry.stake_table_entry.stake()
            })
            .to::<usize>()
    } else {
        let stake_table = stake_table.cloned().collect::<Vec<_>>();
        approximate_weights(&stake_table[..]).total_weight
    }
}

fn approximate_weights<TYPES: NodeType>(stake_table: &[PeerConfig<TYPES>]) -> Weights {
    let total_stake = stake_table.iter().fold(U256::ZERO, |acc, entry| {
        acc + entry.stake_table_entry.stake()
    });

    let mut total_weight: usize = 0;

    // don't attempt to scale if the total stake is small enough
    if total_stake <= U256::from(VID_TARGET_TOTAL_STAKE) {
        let weights = stake_table
            .iter()
            .map(|entry| entry.stake_table_entry.stake().to::<u32>())
            .collect();

        // Note: this panics if `total_stake` exceeds `usize::MAX`, but this shouldn't happen.
        total_weight = total_stake.to::<usize>();

        Weights {
            weights,
            total_weight,
        }
    } else {
        let weights = stake_table
            .iter()
            .map(|entry| {
                let weight: U256 = ((entry.stake_table_entry.stake()
                    * U256::from(VID_TARGET_TOTAL_STAKE))
                    / total_stake)
                    + U256::ONE;

                // Note: this panics if `weight` exceeds `usize::MAX`, but this shouldn't happen.
                total_weight += weight.to::<usize>();

                // Note: this panics if `weight` exceeds `u32::MAX`, but this shouldn't happen
                // and would likely cause a stack overflow in the VID calculation anyway
                weight.to::<u32>()
            })
            .collect();

        Weights {
            weights,
            total_weight,
        }
    }
}

impl<TYPES: NodeType> AvidMDisperse<TYPES> {
    /// Create VID dispersal from a specified membership for the target epoch.
    /// Uses the specified function to calculate share dispersal
    /// Allows for more complex stake table functionality
    async fn from_membership(
        view_number: ViewNumber,
        commit: AvidMCommitment,
        shares: &[AvidMShare],
        common: AvidMCommon,
        membership: &EpochMembership<TYPES>,
        target_epoch: Option<EpochNumber>,
        data_epoch: Option<EpochNumber>,
    ) -> Result<Self> {
        let payload_byte_len = shares[0].payload_byte_len();
        let shares = membership
            .coordinator
            .stake_table_for_epoch(target_epoch)?
            .stake_table()
            .map(|entry| entry.stake_table_entry.public_key())
            .zip(shares)
            .map(|(node, share)| (node.clone(), share.clone()))
            .collect();

        Ok(Self {
            view_number,
            shares,
            payload_commitment: commit,
            epoch: data_epoch,
            target_epoch,
            payload_byte_len,
            common,
        })
    }

    /// Calculate the vid disperse information from the payload given a view, epoch and membership,
    /// If the sender epoch is missing, it means it's the same as the target epoch.
    ///
    /// # Errors
    /// Returns an error if the disperse or commitment calculation fails
    #[allow(clippy::panic)]
    #[allow(clippy::single_range_in_vec_init)]
    pub async fn calculate_vid_disperse(
        payload: &TYPES::BlockPayload,
        membership: &EpochMembershipCoordinator<TYPES>,
        view: ViewNumber,
        target_epoch: Option<EpochNumber>,
        data_epoch: Option<EpochNumber>,
        metadata: &<TYPES::BlockPayload as BlockPayload<TYPES>>::Metadata,
    ) -> Result<(Self, Duration)> {
        let target_mem = membership.stake_table_for_epoch(target_epoch)?;
        let stake_table: Vec<_> = target_mem.stake_table().cloned().collect();
        let approximate_weights = approximate_weights(&stake_table);

        let txns = payload.encode();
        let num_txns = txns.len();

        let avidm_param = init_avidm_param(approximate_weights.total_weight)?;
        let common = avidm_param.clone();

        let ns_table = parse_ns_table(num_txns, &metadata.encode());
        let ns_table_clone = ns_table.clone();

        let now = Instant::now();
        let (commit, shares) = spawn_blocking(move || {
            AvidMScheme::ns_disperse(
                &avidm_param,
                &approximate_weights.weights,
                &txns,
                ns_table_clone,
            )
        })
        .await
        .wrap()
        .context(error!("Join error"))?
        .wrap()
        .context(|err| error!("Failed to calculate VID disperse. Error: {err}"))?;
        let ns_disperse_duration = now.elapsed();

        Ok((
            Self::from_membership(
                view,
                commit,
                &shares,
                common,
                &target_mem,
                target_epoch,
                data_epoch,
            )
            .await?,
            ns_disperse_duration,
        ))
    }

    /// This function splits a VID disperse into individual shares.
    pub fn to_shares(self) -> Vec<AvidMDisperseShare<TYPES>> {
        self.shares
            .into_iter()
            .map(|(recipient_key, share)| AvidMDisperseShare {
                share,
                recipient_key,
                view_number: self.view_number,
                payload_commitment: self.payload_commitment,
                epoch: self.epoch,
                target_epoch: self.target_epoch,
                common: self.common.clone(),
            })
            .collect()
    }

    /// Split a VID disperse into a share proposal for each recipient.
    pub fn to_share_proposals(
        self,
        signature: &<<TYPES as NodeType>::SignatureKey as SignatureKey>::PureAssembledSignatureType,
    ) -> Vec<Proposal<TYPES, AvidMDisperseShare<TYPES>>> {
        self.shares
            .into_iter()
            .map(|(recipient_key, share)| Proposal {
                data: AvidMDisperseShare {
                    share,
                    recipient_key,
                    view_number: self.view_number,
                    payload_commitment: self.payload_commitment,
                    epoch: self.epoch,
                    target_epoch: self.target_epoch,
                    common: self.common.clone(),
                },
                signature: signature.clone(),
                _pd: PhantomData,
            })
            .collect()
    }

    /// Construct a VID disperse from an iterator of disperse shares.
    pub fn try_from_shares<'a, I>(mut it: I) -> Option<Self>
    where
        I: Iterator<Item = &'a AvidMDisperseShare<TYPES>>,
    {
        let first_vid_disperse_share = it.next()?.clone();
        let payload_byte_len = first_vid_disperse_share.share.payload_byte_len();
        let mut share_map = BTreeMap::new();
        share_map.insert(
            first_vid_disperse_share.recipient_key,
            first_vid_disperse_share.share,
        );
        let mut vid_disperse = Self {
            view_number: first_vid_disperse_share.view_number,
            epoch: first_vid_disperse_share.epoch,
            target_epoch: first_vid_disperse_share.target_epoch,
            payload_commitment: first_vid_disperse_share.payload_commitment,
            shares: share_map,
            payload_byte_len,
            common: first_vid_disperse_share.common,
        };
        it.for_each(|vid_disperse_share| {
            vid_disperse.shares.insert(
                vid_disperse_share.recipient_key.clone(),
                vid_disperse_share.share.clone(),
            );
        });
        Some(vid_disperse)
    }

    /// Returns the payload length in bytes.
    pub fn payload_byte_len(&self) -> u32 {
        self.payload_byte_len as u32
    }
}

#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
/// VID share and associated metadata for a single node
pub struct AvidMDisperseShare<TYPES: NodeType> {
    /// The view number for which this VID data is intended
    pub view_number: ViewNumber,
    /// The epoch number for which this VID data belongs to
    pub epoch: Option<EpochNumber>,
    /// The epoch number to which the recipient of this VID belongs to
    pub target_epoch: Option<EpochNumber>,
    /// Block payload commitment
    pub payload_commitment: AvidMCommitment,
    /// A storage node's key and its corresponding VID share
    pub share: AvidMShare,
    /// a public key of the share recipient
    pub recipient_key: TYPES::SignatureKey,
    /// VID common data sent to all storage nodes
    pub common: AvidMCommon,
}

impl<TYPES: NodeType> HasViewNumber for AvidMDisperseShare<TYPES> {
    fn view_number(&self) -> ViewNumber {
        self.view_number
    }
}

impl<TYPES: NodeType> AvidMDisperseShare<TYPES> {
    /// Consume `self` and return a `Proposal`
    pub fn to_proposal(
        self,
        private_key: &<TYPES::SignatureKey as SignatureKey>::PrivateKey,
    ) -> Option<Proposal<TYPES, Self>> {
        let Ok(signature) =
            TYPES::SignatureKey::sign(private_key, self.payload_commitment.as_ref())
        else {
            tracing::error!("VID: failed to sign dispersal share payload");
            return None;
        };
        Some(Proposal {
            signature,
            _pd: PhantomData,
            data: self,
        })
    }

    /// Returns the payload length in bytes.
    pub fn payload_byte_len(&self) -> u32 {
        self.share.payload_byte_len() as u32
    }

    /// Check if vid common is consistent with the commitment.
    /// For AvidM, ns_commits is inside the share, so there's no separate consistency check.
    pub fn is_consistent(&self) -> bool {
        true
    }

    /// Verify share assuming common data is already verified consistent.
    /// For AvidM, this is equivalent to the full verify since there's
    /// no separate consistency check (ns_commits is inside the share).
    pub fn verify_with_verified_common(&self) -> bool {
        AvidMScheme::verify_share(&self.common, &self.payload_commitment, &self.share)
            .is_ok_and(|r| r.is_ok())
    }

    /// Internally verify the share given necessary information
    pub fn verify(&self, _total_weight: usize) -> bool {
        self.is_consistent() && self.verify_with_verified_common()
    }
}

/// AvidmGf2 dispersal data
#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
pub struct AvidmGf2Disperse<TYPES: NodeType> {
    /// The view number for which this VID data is intended
    pub view_number: ViewNumber,
    /// Epoch the data of this proposal belongs to
    pub epoch: Option<EpochNumber>,
    /// Epoch to which the recipients of this VID belong to
    pub target_epoch: Option<EpochNumber>,
    /// VidCommitment calculated based on the number of nodes in `target_epoch`.
    pub payload_commitment: AvidmGf2Commitment,
    /// A storage node's key and its corresponding VID share
    pub shares: BTreeMap<TYPES::SignatureKey, AvidmGf2Share>,
    /// Length of payload in bytes
    pub payload_byte_len: usize,
    /// VID common data sent to all storage nodes
    pub common: AvidmGf2Common,
}

impl<TYPES: NodeType> HasViewNumber for AvidmGf2Disperse<TYPES> {
    fn view_number(&self) -> ViewNumber {
        self.view_number
    }
}

impl<TYPES: NodeType> AvidmGf2Disperse<TYPES> {
    /// Create VID dispersal from a specified membership for the target epoch.
    /// Uses the specified function to calculate share dispersal
    /// Allows for more complex stake table functionality
    fn from_membership(
        view_number: ViewNumber,
        commit: AvidmGf2Commitment,
        shares: &[AvidmGf2Share],
        common: AvidmGf2Common,
        membership: &EpochMembership<TYPES>,
        target_epoch: Option<EpochNumber>,
        data_epoch: Option<EpochNumber>,
    ) -> Result<Self> {
        let payload_byte_len = common.payload_byte_len();
        let shares = membership
            .coordinator
            .stake_table_for_epoch(target_epoch)?
            .stake_table()
            .map(|entry| entry.stake_table_entry.public_key())
            .zip(shares)
            .map(|(node, share)| (node.clone(), share.clone()))
            .collect();

        Ok(Self {
            view_number,
            shares,
            payload_commitment: commit,
            epoch: data_epoch,
            target_epoch,
            payload_byte_len,
            common,
        })
    }

    /// Calculate the vid disperse information from the payload given a view, epoch and membership,
    /// If the sender epoch is missing, it means it's the same as the target epoch.
    ///
    /// # Errors
    /// Returns an error if the disperse or commitment calculation fails
    pub async fn calculate_vid_disperse(
        payload: &TYPES::BlockPayload,
        membership: &EpochMembershipCoordinator<TYPES>,
        view: ViewNumber,
        target_epoch: Option<EpochNumber>,
        data_epoch: Option<EpochNumber>,
        metadata: &<TYPES::BlockPayload as BlockPayload<TYPES>>::Metadata,
    ) -> Result<(Self, Duration)> {
        let target_mem = membership.stake_table_for_epoch(target_epoch)?;
        let stake_table: Vec<_> = target_mem.stake_table().cloned().collect();
        let approximate_weights = approximate_weights(&stake_table);

        let txns = payload.encode();
        let num_txns = txns.len();

        let avidm_param = init_avidm_gf2_param(approximate_weights.total_weight)?;

        let ns_table = parse_ns_table(num_txns, &metadata.encode());
        let ns_table_clone = ns_table.clone();

        let now = Instant::now();
        let (commit, common, shares) = spawn_blocking(move || {
            AvidmGf2Scheme::ns_disperse(
                &avidm_param,
                &approximate_weights.weights,
                &txns,
                ns_table_clone,
            )
        })
        .await
        .wrap()
        .context(error!("Join error"))?
        .wrap()
        .context(|err| error!("Failed to calculate VID disperse. Error: {err}"))?;
        let ns_disperse_duration = now.elapsed();

        Ok((
            Self::from_membership(
                view,
                commit,
                &shares,
                common,
                &target_mem,
                target_epoch,
                data_epoch,
            )?,
            ns_disperse_duration,
        ))
    }

    /// This function splits a VID disperse into individual shares.
    pub fn to_shares(self) -> Vec<AvidmGf2DisperseShare<TYPES>> {
        self.shares
            .into_iter()
            .map(|(recipient_key, share)| AvidmGf2DisperseShare {
                share,
                recipient_key,
                view_number: self.view_number,
                payload_commitment: self.payload_commitment,
                epoch: self.epoch,
                target_epoch: self.target_epoch,
                common: self.common.clone(),
            })
            .collect()
    }

    /// Split a VID disperse into a share proposal for each recipient.
    pub fn to_share_proposals(
        self,
        signature: &<<TYPES as NodeType>::SignatureKey as SignatureKey>::PureAssembledSignatureType,
    ) -> Vec<Proposal<TYPES, AvidmGf2DisperseShare<TYPES>>> {
        self.shares
            .into_iter()
            .map(|(recipient_key, share)| Proposal {
                data: AvidmGf2DisperseShare {
                    share,
                    recipient_key,
                    view_number: self.view_number,
                    payload_commitment: self.payload_commitment,
                    epoch: self.epoch,
                    target_epoch: self.target_epoch,
                    common: self.common.clone(),
                },
                signature: signature.clone(),
                _pd: PhantomData,
            })
            .collect()
    }

    /// Construct a VID disperse from an iterator of disperse shares.
    pub fn try_from_shares<'a, I>(mut it: I) -> Option<Self>
    where
        I: Iterator<Item = &'a AvidmGf2DisperseShare<TYPES>>,
    {
        let first_vid_disperse_share = it.next()?.clone();
        let payload_byte_len = first_vid_disperse_share.common.payload_byte_len();
        let mut share_map = BTreeMap::new();
        share_map.insert(
            first_vid_disperse_share.recipient_key,
            first_vid_disperse_share.share,
        );
        let mut vid_disperse = Self {
            view_number: first_vid_disperse_share.view_number,
            epoch: first_vid_disperse_share.epoch,
            target_epoch: first_vid_disperse_share.target_epoch,
            payload_commitment: first_vid_disperse_share.payload_commitment,
            shares: share_map,
            payload_byte_len,
            common: first_vid_disperse_share.common,
        };
        it.for_each(|vid_disperse_share| {
            vid_disperse.shares.insert(
                vid_disperse_share.recipient_key.clone(),
                vid_disperse_share.share.clone(),
            );
        });
        Some(vid_disperse)
    }

    /// Returns the payload length in bytes.
    pub fn payload_byte_len(&self) -> u32 {
        self.payload_byte_len as u32
    }

    /// Resolve the inputs needed to disperse `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.
    pub fn disperse_params(
        payload: &TYPES::BlockPayload,
        membership: &EpochMembershipCoordinator<TYPES>,
        target_epoch: Option<EpochNumber>,
        metadata: &<TYPES::BlockPayload as BlockPayload<TYPES>>::Metadata,
    ) -> Result<AvidmGf2DisperseParams<TYPES>> {
        let target_mem = membership.stake_table_for_epoch(target_epoch)?;
        let stake_table: Vec<_> = target_mem.stake_table().cloned().collect();
        let Weights {
            weights,
            total_weight,
        } = approximate_weights(&stake_table);
        let recipients = stake_table
            .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 param = init_avidm_gf2_param(total_weight)?;
        Ok(AvidmGf2DisperseParams {
            param,
            weights,
            recipients,
            ns_table,
            payload,
        })
    }
}

/// VID share and associated metadata for a single node
#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
pub struct AvidmGf2DisperseShare<TYPES: NodeType> {
    /// The view number for which this VID data is intended
    pub view_number: ViewNumber,
    /// The epoch number for which this VID data belongs to
    pub epoch: Option<EpochNumber>,
    /// The epoch number to which the recipient of this VID belongs to
    pub target_epoch: Option<EpochNumber>,
    /// Block payload commitment
    pub payload_commitment: AvidmGf2Commitment,
    /// A storage node's key and its corresponding VID share
    pub share: AvidmGf2Share,
    /// a public key of the share recipient
    pub recipient_key: TYPES::SignatureKey,
    /// VID common data sent to all storage nodes
    pub common: AvidmGf2Common,
}

impl<TYPES: NodeType> HasViewNumber for AvidmGf2DisperseShare<TYPES> {
    fn view_number(&self) -> ViewNumber {
        self.view_number
    }
}

impl<TYPES: NodeType> AvidmGf2DisperseShare<TYPES> {
    /// Consume `self` and return a `Proposal`
    pub fn to_proposal(
        self,
        private_key: &<TYPES::SignatureKey as SignatureKey>::PrivateKey,
    ) -> Option<Proposal<TYPES, Self>> {
        let Ok(signature) =
            TYPES::SignatureKey::sign(private_key, self.payload_commitment.as_ref())
        else {
            tracing::error!("VID: failed to sign dispersal share payload");
            return None;
        };
        Some(Proposal {
            signature,
            _pd: PhantomData,
            data: self,
        })
    }
    /// Returns the payload length in bytes.
    pub fn payload_byte_len(&self) -> u32 {
        self.common.payload_byte_len() as u32
    }
    /// Check if vid common is consistent with the commitment.
    pub fn is_consistent(&self) -> bool {
        AvidmGf2Scheme::is_consistent(&self.payload_commitment, &self.common)
    }

    /// Verify share assuming common data is already verified consistent.
    /// Caller MUST call `is_consistent()` first.
    pub fn verify_with_verified_common(&self) -> bool {
        AvidmGf2Scheme::verify_share_with_verified_common(&self.common, &self.share)
            .is_ok_and(|r| r.is_ok())
    }

    /// Internally verify the share given necessary information
    pub fn verify(&self, total_weight: usize) -> bool {
        // A share's commitment hash-binds its `ns_commits` (via `is_consistent`)
        // but not its `param`, so check `param` against the committee-derived
        // expectation; otherwise a forged param would pass verification.
        init_avidm_gf2_param(total_weight).is_ok_and(|expected| self.common.param == expected)
            && self.is_consistent()
            && self.verify_with_verified_common()
    }
}

/// VID shares consist of fragments as the unit of transmission.
#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
pub struct AvidmGf2DisperseShareFragment<T: NodeType> {
    /// The view number for which this VID data is intended.
    pub view_number: ViewNumber,
    /// The epoch number for which this VID data belongs to.
    pub epoch: Option<EpochNumber>,
    /// The epoch number to which the recipient of this VID belongs to.
    pub target_epoch: Option<EpochNumber>,
    /// Block payload commitment (the aggregate over all namespaces).
    pub payload_commitment: AvidmGf2Commitment,
    /// A public key of the share recipient.
    pub recipient_key: T::SignatureKey,
    /// VID erasure parameters; identical across a view's fragments.
    pub param: AvidmGf2Param,
    /// Total number of namespaces in the block.
    pub num_namespaces: usize,
    /// Namespace share fragment pieces.
    pub namespaces: Vec<AvidmGf2NamespacePiece>,
}

impl<T: NodeType> HasViewNumber for AvidmGf2DisperseShareFragment<T> {
    fn view_number(&self) -> ViewNumber {
        self.view_number
    }
}

impl<T: NodeType> HasEpoch for AvidmGf2DisperseShareFragment<T> {
    fn epoch(&self) -> Option<EpochNumber> {
        self.epoch
    }
}

/// VID share fragments hold pieces.
///
/// A piece is the smallest unit. Fragments group multiple pieces together,
/// depending on the cumulative length of their payload lengths.
#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq, Hash)]
pub struct AvidmGf2NamespacePiece {
    /// This namespace's index in the namespace table, in `0..num_namespaces`.
    pub ns_index: usize,
    /// Byte length of this namespace's slice of the payload.
    pub ns_payload_byte_len: usize,
    /// Commitment to this namespace's shards.
    pub ns_commit: vid::avidm_gf2::AvidmGf2Commit,
    /// This recipient's share of this namespace.
    pub ns_share: vid::avidm_gf2::AvidmGf2Share,
}

/// Parameters for a per-namespace AvidmGf2 dispersal.
///
/// [`AvidmGf2Disperse::disperse_params`] resolves the VID parameters and
/// recipient ordering up front so a caller can send each namespace's
/// fragments as they are produced.
///
/// `recipients[i]` is the storage node whose weight is `weights[i]` and
/// whose share is the `i`th entry of each namespace's dispersal.
#[non_exhaustive]
pub struct AvidmGf2DisperseParams<T: NodeType> {
    /// VID erasure parameters.
    pub param: AvidmGf2Param,
    /// Per-node weights, in stake-table order.
    pub weights: Vec<u32>,
    /// Recipients, in the same order as `weights`.
    pub recipients: Vec<T::SignatureKey>,
    /// Namespace byte ranges over the encoded payload.
    pub ns_table: Vec<Range<usize>>,
    /// The encoded payload.
    pub payload: Arc<[u8]>,
}