fix: race condition for finalized blocks

src/analytics/mod.rs

@@ -314,7 +314,9 @@ impl<'a, I: InputSource> Slot<'a, I> {
             db,
         };
 
-        let mut block_stream = self.accepted_block_stream().await?.boxed();
+        let mut block_stream = self.finalized_block_stream().await?.boxed();
+
+        let mut counter = 0;
 
         while let Some(data) = block_stream.try_next().await? {
             if let Some((payload, metadata)) = data
@@ -332,8 +334,11 @@ impl<'a, I: InputSource> Slot<'a, I> {
                 }
             }
             self.handle_block(analytics, &data.block, &ctx).await?;
+            counter += 1;
         }
 
+        println!("analytics blocks: {counter}");
+
         influxdb
             .insert_measurement((analytics as &mut dyn DynAnalytics).take_measurement(&ctx).await?)
             .await?;

src/bin/inx-chronicle/inx/mod.rs

@@ -109,9 +109,9 @@ impl InxWorker {
         };
 
         debug!(
-            "The node has a pruning epoch index of `{}` and a latest confirmed slot index of `{}`.",
+            "The node has a pruning epoch index of `{}` and a latest finalized slot index of `{}`.",
             node_status.pruning_epoch,
-            node_status.latest_commitment.commitment_id.slot_index()
+            node_status.latest_finalized_commitment.commitment_id.slot_index()
         );
 
         let mut node_configuration = inx.get_node_configuration().await?;
@@ -303,7 +303,7 @@ impl InxWorker {
         tracing::Span::current().record("created", slot.ledger_updates().created_outputs().len());
         tracing::Span::current().record("consumed", slot.ledger_updates().consumed_outputs().len());
 
-        self.handle_accepted_blocks(&slot).await?;
+        self.handle_finalized_blocks(&slot).await?;
 
         #[cfg(feature = "influx")]
         self.update_influx(
@@ -326,8 +326,17 @@ impl InxWorker {
     }
 
     #[instrument(skip_all, err, level = "trace")]
-    async fn handle_accepted_blocks<'a>(&mut self, slot: &Slot<'a, Inx>) -> Result<()> {
-        let blocks_stream = slot.accepted_block_stream().await?;
+    async fn handle_finalized_blocks<'a>(&mut self, slot: &Slot<'a, Inx>) -> Result<()> {
+        // let mut block_stream = slot.finalized_block_stream().await?.boxed();
+
+        // let mut counter = 0;
+        // while let Some(data) = block_stream.try_next().await? {
+        //     counter += 1;
+        // }
+
+        // println!("inx blocks: {counter}");
+
+        let blocks_stream = slot.finalized_block_stream().await?;
 
         let mut tasks = blocks_stream
             .try_chunks(INSERT_BATCH_SIZE)

src/tangle/slot_stream.rs

@@ -39,13 +39,25 @@ impl<'a, I: InputSource> Slot<'a, I> {
     pub fn commitment(&self) -> &Raw<SlotCommitment> {
         &self.commitment.commitment
     }
+
+    /// Check whether the slot has been finalized.
+    pub async fn is_finalized(&self) -> bool {
+        // TODO: unwrap
+        self.commitment.commitment_id.slot_index() >= self.source.latest_finalized_slot_index().await.unwrap()
+    }
 }
 
 impl<'a, I: InputSource> Slot<'a, I> {
     /// Returns the accepted blocks of a slot.
-    pub async fn accepted_block_stream(
+    pub async fn finalized_block_stream(
         &self,
     ) -> Result<impl Stream<Item = Result<BlockWithTransactionMetadata, I::Error>> + '_, I::Error> {
+        while !self.is_finalized().await {
+            println!("not finalized: {}", self.index());
+            tokio::time::sleep(core::time::Duration::from_millis(100)).await;
+        }
+        println!("finalized: {}", self.index());
+
         Ok(self
             .source
             .accepted_blocks(self.index())

src/tangle/sources/inx.rs

@@ -90,4 +90,14 @@ impl InputSource for Inx {
 
         Ok(LedgerUpdateStore::init(consumed, created))
     }
+
+    async fn latest_finalized_slot_index(&self) -> Result<SlotIndex, Self::Error> {
+        let mut inx = self.clone();
+        Ok(inx
+            .get_node_status()
+            .await?
+            .latest_finalized_commitment
+            .commitment_id
+            .slot_index())
+    }
 }

src/tangle/sources/memory.rs

@@ -74,4 +74,8 @@ impl InputSource for BTreeMap<SlotIndex, InMemoryData> {
             .ledger_updates
             .clone())
     }
+
+    async fn latest_finalized_slot_index(&self) -> Result<SlotIndex, Self::Error> {
+        todo!()
+    }
 }

src/tangle/sources/mod.rs

@@ -30,7 +30,7 @@ pub trait InputSource: Send + Sync {
         range: impl RangeBounds<SlotIndex> + Send,
     ) -> Result<BoxStream<Result<Commitment, Self::Error>>, Self::Error>;
 
-    /// A stream of accepted blocks for a given slot index.
+    /// A stream of (at least) accepted blocks for a given slot index.
     async fn accepted_blocks(
         &self,
         index: SlotIndex,
@@ -41,4 +41,7 @@ pub trait InputSource: Send + Sync {
 
     /// Retrieves the updates to the ledger for a given range of slots.
     async fn ledger_updates(&self, index: SlotIndex) -> Result<LedgerUpdateStore, Self::Error>;
+
+    /// Retrieves the latest finalized slot index.
+    async fn latest_finalized_slot_index(&self) -> Result<SlotIndex, Self::Error>;
 }

src/tangle/sources/mongodb.rs

@@ -104,4 +104,8 @@ impl InputSource for MongoDb {
 
         Ok(LedgerUpdateStore::init(consumed, created))
     }
+
+    async fn latest_finalized_slot_index(&self) -> Result<SlotIndex, Self::Error> {
+        todo!()
+    }
 }