Make update fallible again

A recent commit made `update` infallible, reasoning that if we fail
to store a new leaf, it is better to at least try storing the next
one, rather than failing completely. However, this is not accurate:
the client uses the success/failure of `udpate` to determine when
it is safe to garbage collect data from temporary storage. Thus,
returning success when we didn't actually store all the leaves can
lead to data loss.

This change fixes the situation by again returning an error whenever
we fail to insert a leaf. It also tries to make the return value more
useful, by returning on error the height of the first leaf we failed
to insert (which is useful for garbage collection) instead of just an
opaque error message which the caller can't do much with.

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(())