Struct narwhal_primary::BlockRemover

pub struct BlockRemover { /* private fields */ }

BlockRemover is responsible for removing blocks identified by their certificate id (digest) from across our system. On high level It will make sure that the DAG is updated, internal storage where there certificates and headers are stored, and the corresponding batches as well.

Example

Basic setup of the BlockRemover

This example shows the basic setup of the BlockRemover module. It showcases the necessary components that have to be used (e.x channels, datastore etc) and how a request (command) should be issued to delete a list of blocks and receive the result of it.

#[tokio::main(flavor = "current_thread")]
    const CERTIFICATES_CF: &str = "certificates";
    const CERTIFICATE_ID_BY_ROUND_CF: &str = "certificate_id_by_round";
    const HEADERS_CF: &str = "headers";
    const PAYLOAD_CF: &str = "payload";

    let temp_dir = tempdir().expect("Failed to open temporary directory").into_path();

    // Basic setup: datastore, channels & BlockWaiter
    let rocksdb = rocks::open_cf(temp_dir, None, &[CERTIFICATES_CF, CERTIFICATE_ID_BY_ROUND_CF, HEADERS_CF, PAYLOAD_CF])
        .expect("Failed creating database");

    let (certificate_map, certificate_id_by_round_map, headers_map, payload_map) = reopen!(&rocksdb,
            CERTIFICATES_CF;<CertificateDigest, Certificate>,
            CERTIFICATE_ID_BY_ROUND_CF;<(Round, CertificateDigest), CertificateToken>,
            HEADERS_CF;<HeaderDigest, Header>,
            PAYLOAD_CF;<(BatchDigest, WorkerId), PayloadToken>);
    let certificate_store = CertificateStore::new(certificate_map, certificate_id_by_round_map);
    let headers_store = Store::new(headers_map);
    let payload_store = Store::new(payload_map);

    let (tx_commands, rx_commands) = test_utils::test_channel!(1);
    let (tx_delete_batches, rx_delete_batches) = test_utils::test_channel!(1);
    let (tx_removed_certificates, _rx_removed_certificates) = test_utils::test_channel!(1);
    let (tx_delete_block_result, mut rx_delete_block_result) = channel(1);

    let name = PublicKey::default();
    let committee = Committee{ epoch: 0, authorities: BTreeMap::new() };
    let worker_cache: SharedWorkerCache = WorkerCache{ epoch: 0, workers: BTreeMap::new() }.into();
    let (_tx_reconfigure, rx_reconfigure) = watch::channel(ReconfigureNotification::NewEpoch(committee.clone()));
    let consensus_metrics = Arc::new(ConsensusMetrics::new(&Registry::new()));
    // A dag with genesis for the committee
    let (tx_new_certificates, rx_new_certificates) = test_utils::test_channel!(1);
    let dag = Arc::new(Dag::new(&committee, rx_new_certificates, consensus_metrics).1);
    // Populate genesis in the Dag
    join_all(
      Certificate::genesis(&committee)
      .iter()
      .map(|cert| dag.insert(cert.clone())),
    )
    .await;

    let _ = BlockRemover::spawn(
        name,
        committee,
        worker_cache,
        certificate_store.clone(),
        headers_store.clone(),
        payload_store.clone(),
        Some(dag),
        network::P2pNetwork::new(test_utils::random_network()),
        rx_reconfigure,
        rx_commands,
        rx_delete_batches,
        tx_removed_certificates,
    );

    // A dummy certificate
    let certificate = Certificate::default();

    // Send a command to receive a block
    tx_commands
        .send(BlockRemoverCommand::RemoveBlocks {
            ids: vec![certificate.clone().digest()],
            sender: tx_delete_block_result,
        })
        .await;

    // Dummy - we expect to receive the deleted batches responses via another component
    // and get fed via the tx_delete_batches channel.
    tx_delete_batches.send(Ok(DeleteBatchMessage{ ids: vec![BatchDigest::default()] })).await;

    // Wait to receive the blocks delete output to the provided sender channel
    match rx_delete_block_result.recv().await {
        Some(Ok(result)) => {
            println!("Successfully received a delete blocks response");
        }
        Some(Err(err)) => {
            println!("Received an error {:?}", err);
        }
        _ => {
            println!("Nothing received");
        }
    }

Implementations

impl BlockRemover

pub fn spawn(
    name: PublicKey,
    committee: Committee,
    worker_cache: SharedWorkerCache,
    certificate_store: CertificateStore,
    header_store: Store<HeaderDigest, Header>,
    payload_store: Store<(BatchDigest, WorkerId), PayloadToken>,
    dag: Option<Arc<Dag>>,
    worker_network: P2pNetwork,
    rx_reconfigure: Receiver<ReconfigureNotification>,
    rx_commands: Receiver<BlockRemoverCommand>,
    rx_delete_batches: Receiver<Result<DeleteBatchMessage, DeleteBatchMessage>>,
    removed_certificates: Sender<Certificate>
) -> JoinHandle<()>