fix(publisher): Don't defer error logging

FuelLabs · Dec 12, 2024 · b1699d8 · b1699d8
1 parent c165b19
commit b1699d8
Showing 1 changed file with 31 additions and 28 deletions.
diff --git a/crates/fuel-streams-publisher/src/publisher/mod.rs b/crates/fuel-streams-publisher/src/publisher/mod.rs
@@ -133,35 +133,37 @@ impl Publisher {
 
         loop {
             tokio::select! {
-                Some(sealed_block) = blocks_stream.next() => {
-                    let sealed_block = sealed_block.context("block streams failed to produce sealed block")?;
-
-                        tracing::info!("Processing blocks stream");
-
-                        let fuel_core = &self.fuel_core;
-                        let (block, block_producer) =
-                            fuel_core.get_block_and_producer(sealed_block);
-
-                        // TODO: Avoid awaiting Offchain DB sync for all streams by grouping in their own service
-                        fuel_core
-                            .await_offchain_db_sync(&block.id())
-                            .await
-                            .context("Failed to await Offchain DB sync")?;
-
-                        if let Err(err) = self.publish(&block, &block_producer).await {
-                            tracing::error!("Failed to publish block data: {}", err);
-                            self.telemetry.record_failed_publishing(self.fuel_core.chain_id(), &block_producer);
+                    Some(sealed_block) = blocks_stream.next() => {
+                        let sealed_block = sealed_block.context("block streams failed to produce sealed block")?;
+
+                            tracing::info!("Processing blocks stream");
+
+                            let fuel_core = &self.fuel_core;
+                            let (block, block_producer) =
+                                fuel_core.get_block_and_producer(sealed_block);
+
+                            // TODO: Avoid awaiting Offchain DB sync for all streams by grouping in their own service
+                            fuel_core
+                                .await_offchain_db_sync(&block.id())
+                                .await
+                                .context("Failed to await Offchain DB sync")?;
+
+                            if let Err(err) = self.publish(&block, &block_producer).await {
+            tracing::error!("Publisher encountered an error: {:?}", err);
+
+                                tracing::error!("Failed to publish block data: {}", err);
+                                self.telemetry.record_failed_publishing(self.fuel_core.chain_id(), &block_producer);
+                            }
+
+                    },
+                    shutdown = shutdown_token.wait_for_shutdown() => {
+                        if shutdown {
+                            tracing::info!("Shutdown signal received. Stopping services ...");
+                            self.shutdown_services_with_timeout().await?;
+                            break;
                         }
-
-                },
-                shutdown = shutdown_token.wait_for_shutdown() => {
-                    if shutdown {
-                        tracing::info!("Shutdown signal received. Stopping services ...");
-                        self.shutdown_services_with_timeout().await?;
-                        break;
-                    }
-                },
-            };
+                    },
+                };
         }
 
         tracing::info!("Publishing stopped successfully!");
@@ -281,6 +283,7 @@ pub fn publish<S: Streamable + 'static>(
                 Ok(())
             }
             Err(e) => {
+                tracing::error!("Failed to publish: {:?}", e);
                 telemetry.log_error(&e.to_string());
                 telemetry.update_publisher_error_metrics(
                     wildcard,