Merge pull request #725 from EspressoSystems/jb/fallible-update

Make `update` fallible again

src/data_source/extension.rs

@@ -433,7 +433,7 @@ mod impl_testable_data_source {
         }
 
         async fn handle_event(&self, event: &Event<MockTypes>) {
-            self.update(event).await;
+            self.update(event).await.unwrap();
         }
     }
 }

src/data_source/fs.rs

@@ -138,7 +138,9 @@ pub use super::storage::fs::Transaction;
 ///         let mut events = hotshot.event_stream();
 ///         while let Some(event) = events.next().await {
 ///             let mut state = state.write().await;
-///             state.hotshot_qs.update(&event).await;
+///             if state.hotshot_qs.update(&event).await.is_err() {
+///                 continue;
+///             }
 ///
 ///             // Update other modules' states based on `event`.
 ///             let mut tx = state.hotshot_qs.write().await.unwrap();
@@ -269,7 +271,7 @@ mod impl_testable_data_source {
         }
 
         async fn handle_event(&self, event: &Event<MockTypes>) {
-            self.update(event).await;
+            self.update(event).await.unwrap();
         }
     }
 }

src/data_source/sql.rs

@@ -267,7 +267,9 @@ impl Config {
 ///     spawn(async move {
 ///         let mut events = hotshot.event_stream();
 ///         while let Some(event) = events.next().await {
-///             state.hotshot_qs.update(&event).await;
+///             if state.hotshot_qs.update(&event).await.is_err() {
+///                 continue;
+///             }
 ///
 ///             let mut tx = state.hotshot_qs.write().await.unwrap();
 ///             // Update other modules' states based on `event`, using `tx` to access the database.
@@ -334,7 +336,7 @@ pub mod testing {
         }
 
         async fn handle_event(&self, event: &Event<MockTypes>) {
-            self.update(event).await;
+            self.update(event).await.unwrap();
         }
     }
 }

src/data_source/storage/no_storage.rs

@@ -383,7 +383,7 @@ pub mod testing {
         }
 
         async fn handle_event(&self, event: &Event<MockTypes>) {
-            self.update(event).await;
+            self.update(event).await.unwrap();
         }
     }
 

src/data_source/update.rs

@@ -56,7 +56,13 @@ pub trait UpdateDataSource<Types: NodeType>: UpdateAvailabilityData<Types> {
     ///
     /// If you want to update the data source with an untrusted event, for example one received from
     /// a peer over the network, you must authenticate it first.
-    async fn update(&self, event: &Event<Types>);
+    ///
+    /// # Returns
+    ///
+    /// If all provided data is successfully inserted into the database, returns `Ok(())`. If any
+    /// error occurred, the error is logged, and the return value is the height of the first leaf
+    /// which failed to be inserted.
+    async fn update(&self, event: &Event<Types>) -> Result<(), u64>;
 }
 
 #[async_trait]
@@ -66,7 +72,7 @@ where
     Payload<Types>: QueryablePayload<Types>,
     <Types as NodeType>::InstanceState: Default,
 {
-    async fn update(&self, event: &Event<Types>) {
+    async fn update(&self, event: &Event<Types>) -> Result<(), u64> {
         if let EventType::Decide { leaf_chain, qc, .. } = &event.event {
             // `qc` justifies the first (most recent) leaf...
             let qcs = once((**qc).clone())
@@ -95,7 +101,7 @@ where
                             ?qc,
                             "inconsistent leaf; cannot append leaf information: {err:#}"
                         );
-                        continue;
+                        return Err(leaf.block_header().block_number());
                     }
                 };
 
@@ -137,10 +143,12 @@ where
                     .append(BlockInfo::new(leaf_data, block_data, vid_common, vid_share))
                     .await
                 {
-                    tracing::warn!(height, "failed to append leaf information: {err:#}");
+                    tracing::error!(height, "failed to append leaf information: {err:#}");
+                    return Err(leaf.block_header().block_number());
                 }
             }
         }
+        Ok(())
     }
 }
 

src/lib.rs

@@ -77,7 +77,7 @@
 //! let mut events = hotshot.event_stream();
 //! while let Some(event) = events.next().await {
 //!     // Update the query data based on this event.
-//!     data_source.update(&event).await;
+//!     data_source.update(&event).await.ok();
 //! }
 //! # Ok(())
 //! # }
@@ -553,8 +553,10 @@ where
 
     // Update query data using HotShot events.
     while let Some(event) = events.next().await {
-        // Update the query data based on this event.
-        data_source.update(&event).await;
+        // Update the query data based on this event. It is safe to ignore errors here; the error
+        // just returns the failed block height for use in garbage collection, but this simple
+        // implementation isn't doing any kind of garbage collection.
+        data_source.update(&event).await.ok();
     }
 
     Ok(())