link the remove_collections endpoint to the garbage collections proce…

…ss (#378)

narwhal/node/src/lib.rs

@@ -115,7 +115,7 @@ impl Node {
         State: ExecutionState + Send + Sync + 'static,
     {
         let (tx_new_certificates, rx_new_certificates) = channel(Self::CHANNEL_CAPACITY);
-        let (tx_feedback, rx_feedback) = channel(Self::CHANNEL_CAPACITY);
+        let (tx_consensus, rx_consensus) = channel(Self::CHANNEL_CAPACITY);
 
         // Compute the public key of this authority.
         let name = keypair.public().clone();
@@ -132,7 +132,7 @@ impl Node {
                 parameters.clone(),
                 execution_state,
                 rx_new_certificates,
-                tx_feedback,
+                tx_consensus.clone(),
                 tx_confirmation,
             )
             .await?;
@@ -149,9 +149,10 @@ impl Node {
             store.certificate_store.clone(),
             store.payload_store.clone(),
             /* tx_consensus */ tx_new_certificates,
-            /* rx_consensus */ rx_feedback,
+            /* rx_consensus */ rx_consensus,
             /* dag */ dag,
             network_model,
+            tx_consensus,
         );
 
         Ok(primary_handle)

narwhal/primary/src/block_remover.rs

@@ -117,6 +117,7 @@ pub struct DeleteBatchMessage {
 ///
 ///     let (tx_commands, rx_commands) = channel(1);
 ///     let (tx_delete_batches, rx_delete_batches) = channel(1);
+///     let (tx_removed_certificates, _rx_removed_certificates) = channel(1);
 ///     let (tx_delete_block_result, mut rx_delete_block_result) = channel(1);
 ///
 ///     let name = Ed25519PublicKey::default();
@@ -142,6 +143,7 @@ pub struct DeleteBatchMessage {
 ///         PrimaryToWorkerNetwork::default(),
 ///         rx_commands,
 ///         rx_delete_batches,
+///         tx_removed_certificates,
 ///     );
 ///
 ///     // A dummy certificate
@@ -211,6 +213,9 @@ pub struct BlockRemover<PublicKey: VerifyingKey> {
     // TODO: Change to a oneshot channel instead of an mpsc channel
     /// Receives all the responses to the requests to delete a batch.
     rx_delete_batches: Receiver<DeleteBatchResult>,
+
+    /// Outputs all the successfully deleted certificates
+    tx_removed_certificates: Sender<Certificate<PublicKey>>,
 }
 
 impl<PublicKey: VerifyingKey> BlockRemover<PublicKey> {
@@ -224,6 +229,7 @@ impl<PublicKey: VerifyingKey> BlockRemover<PublicKey> {
         worker_network: PrimaryToWorkerNetwork,
         rx_commands: Receiver<BlockRemoverCommand>,
         rx_delete_batches: Receiver<DeleteBatchResult>,
+        removed_certificates: Sender<Certificate<PublicKey>>,
     ) -> JoinHandle<()> {
         tokio::spawn(async move {
             Self {
@@ -239,6 +245,7 @@ impl<PublicKey: VerifyingKey> BlockRemover<PublicKey> {
                 map_tx_removal_results: HashMap::new(),
                 map_tx_worker_removal_results: HashMap::new(),
                 rx_delete_batches,
+                tx_removed_certificates: removed_certificates,
             }
             .run()
             .await;
@@ -375,11 +382,20 @@ impl<PublicKey: VerifyingKey> BlockRemover<PublicKey> {
         if let Some(dag) = &self.dag {
             dag.remove(&certificate_ids).await.map_err(Either::Right)?
         }
+
         self.certificate_store
             .remove_all(certificate_ids)
             .await
             .map_err(Either::Left)?;
 
+        // Now output all the removed certificates
+        for certificate in certificates.clone() {
+            self.tx_removed_certificates
+                .send(certificate)
+                .await
+                .expect("Couldn't forward removed certificates to channel");
+        }
+
         debug!("Successfully cleaned up certificates: {:?}", certificates);
 
         Ok(())

narwhal/primary/src/primary.rs

@@ -101,6 +101,7 @@ impl Primary {
         rx_consensus: Receiver<Certificate<PublicKey>>,
         dag: Option<Arc<Dag<PublicKey>>>,
         network_model: NetworkModel,
+        tx_committed_certificates: Sender<Certificate<PublicKey>>,
     ) -> JoinHandle<()> {
         let (tx_others_digests, rx_others_digests) = channel(CHANNEL_CAPACITY);
         let (tx_our_digests, rx_our_digests) = channel(CHANNEL_CAPACITY);
@@ -244,6 +245,7 @@ impl Primary {
             PrimaryToWorkerNetwork::default(),
             rx_block_removal_commands,
             rx_batch_removal,
+            tx_committed_certificates,
         );
 
         // Responsible for finding missing blocks (certificates) and fetching

narwhal/primary/src/tests/block_remover_tests.rs

@@ -34,6 +34,7 @@ async fn test_successful_blocks_delete() {
     // GIVEN
     let (header_store, certificate_store, payload_store) = create_db_stores();
     let (_tx_consensus, rx_consensus) = channel(1);
+    let (tx_removed_certificates, mut rx_removed_certificates) = channel(10);
     let (tx_commands, rx_commands) = channel(10);
     let (tx_remove_block, mut rx_remove_block) = channel(1);
     let (tx_delete_batches, rx_delete_batches) = channel(10);
@@ -54,6 +55,7 @@ async fn test_successful_blocks_delete() {
         PrimaryToWorkerNetwork::default(),
         rx_commands,
         rx_delete_batches,
+        tx_removed_certificates,
     );
 
     let mut block_ids = Vec::new();
@@ -157,7 +159,7 @@ async fn test_successful_blocks_delete() {
             assert_eq!(block.ids.len(), block_ids.len());
 
             // ensure that certificates have been deleted from store
-            for block_id in block_ids {
+            for block_id in block_ids.clone() {
                 assert!(certificate_store.read(block_id).await.unwrap().is_none(), "Certificate shouldn't exist");
             }
 
@@ -177,6 +179,18 @@ async fn test_successful_blocks_delete() {
             panic!("Timeout, no result has been received in time")
         }
     }
+
+    // ensure deleted certificates have been populated to output channel
+    let mut total_deleted = 0;
+    while let Ok(Some(c)) = timeout(Duration::from_secs(1), rx_removed_certificates.recv()).await {
+        assert!(
+            block_ids.contains(&c.digest()),
+            "Deleted certificate not found"
+        );
+        total_deleted += 1;
+    }
+
+    assert_eq!(total_deleted, block_ids.len());
 }
 
 #[tokio::test]
@@ -187,6 +201,7 @@ async fn test_timeout() {
     let (tx_remove_block, mut rx_remove_block) = channel(1);
     let (_tx_consensus, rx_consensus) = channel(1);
     let (tx_delete_batches, rx_delete_batches) = channel(10);
+    let (tx_removed_certificates, _rx_removed_certificates) = channel(10);
 
     // AND the necessary keys
     let (name, committee) = resolve_name_and_committee();
@@ -204,6 +219,7 @@ async fn test_timeout() {
         PrimaryToWorkerNetwork::default(),
         rx_commands,
         rx_delete_batches,
+        tx_removed_certificates,
     );
 
     let mut block_ids = Vec::new();
@@ -318,6 +334,7 @@ async fn test_unlocking_pending_requests() {
     let (tx_commands, rx_commands) = channel(10);
     let (_tx_consensus, rx_consensus) = channel(1);
     let (tx_delete_batches, rx_delete_batches) = channel(10);
+    let (tx_removed_certificates, _rx_removed_certificates) = channel(10);
 
     // AND the necessary keys
     let (name, committee) = resolve_name_and_committee();
@@ -338,6 +355,7 @@ async fn test_unlocking_pending_requests() {
         map_tx_removal_results: HashMap::new(),
         map_tx_worker_removal_results: HashMap::new(),
         rx_delete_batches,
+        tx_removed_certificates,
     };
 
     let mut block_ids = Vec::new();

narwhal/primary/tests/integration_tests_configuration_api.rs

@@ -30,7 +30,7 @@ async fn test_new_epoch() {
     let store = NodeStorage::reopen(temp_dir());
 
     let (tx_new_certificates, rx_new_certificates) = channel(CHANNEL_CAPACITY);
-    let (_tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
+    let (tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
 
     Primary::spawn(
         name.clone(),
@@ -44,6 +44,7 @@ async fn test_new_epoch() {
         /* rx_consensus */ rx_feedback,
         /* dag */ Some(Arc::new(Dag::new(&committee, rx_new_certificates).1)),
         NetworkModel::Asynchronous,
+        tx_feedback,
     );
 
     // Wait for tasks to start
@@ -90,7 +91,7 @@ async fn test_new_network_info() {
     let store = NodeStorage::reopen(temp_dir());
 
     let (tx_new_certificates, rx_new_certificates) = channel(CHANNEL_CAPACITY);
-    let (_tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
+    let (tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
 
     Primary::spawn(
         name.clone(),
@@ -104,6 +105,7 @@ async fn test_new_network_info() {
         /* rx_consensus */ rx_feedback,
         /* dag */ Some(Arc::new(Dag::new(&committee, rx_new_certificates).1)),
         NetworkModel::Asynchronous,
+        /* tx_committed_certificates */ tx_feedback,
     );
 
     // Wait for tasks to start

narwhal/primary/tests/integration_tests_proposer_api.rs

@@ -69,7 +69,7 @@ async fn test_rounds_errors() {
 
     // Spawn the primary
     let (tx_new_certificates, rx_new_certificates) = channel(CHANNEL_CAPACITY);
-    let (_tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
+    let (tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
 
     // AND create a committee passed exclusively to the DAG that does not include the name public key
     // In this way, the genesis certificate is not run for that authority and is absent when we try to fetch it
@@ -108,6 +108,7 @@ async fn test_rounds_errors() {
             Dag::new(&no_name_committee, rx_new_certificates).1,
         )),
         NetworkModel::Asynchronous,
+        tx_feedback,
     );
 
     // AND Wait for tasks to start
@@ -156,7 +157,7 @@ async fn test_rounds_return_successful_response() {
 
     // Spawn the primary
     let (tx_new_certificates, rx_new_certificates) = channel(CHANNEL_CAPACITY);
-    let (_tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
+    let (tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
 
     // AND setup the DAG
     let dag = Arc::new(Dag::new(&committee, rx_new_certificates).1);
@@ -173,6 +174,7 @@ async fn test_rounds_return_successful_response() {
         /* rx_consensus */ rx_feedback,
         /* external_consensus */ Some(dag.clone()),
         NetworkModel::Asynchronous,
+        tx_feedback,
     );
 
     // AND Wait for tasks to start
@@ -286,7 +288,7 @@ async fn test_node_read_causal_signed_certificates() {
         .await
         .unwrap();
 
-    let (_tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
+    let (tx_feedback, rx_feedback) = channel(CHANNEL_CAPACITY);
 
     let primary_1_parameters = Parameters {
         batch_size: 200, // Two transactions.
@@ -308,10 +310,11 @@ async fn test_node_read_causal_signed_certificates() {
         /* rx_consensus */ rx_feedback,
         /* dag */ Some(dag.clone()),
         NetworkModel::Asynchronous,
+        tx_feedback,
     );
 
     let (tx_new_certificates_2, rx_new_certificates_2) = channel(CHANNEL_CAPACITY);
-    let (_tx_feedback_2, rx_feedback_2) = channel(CHANNEL_CAPACITY);
+    let (tx_feedback_2, rx_feedback_2) = channel(CHANNEL_CAPACITY);
 
     let primary_2_parameters = Parameters {
         batch_size: 200, // Two transactions.
@@ -334,6 +337,7 @@ async fn test_node_read_causal_signed_certificates() {
         /* external_consensus */
         Some(Arc::new(Dag::new(&committee, rx_new_certificates_2).1)),
         NetworkModel::Asynchronous,
+        tx_feedback_2,
     );
 
     // Wait for tasks to start