refactor(l1): simplify sync logic

crates/l1/src/subscriber.rs

@@ -517,79 +517,55 @@ where
 
     /// Determine the starting block number for backfill.
     ///
-    /// The zone's persisted Tempo checkpoint is the authoritative source for
-    /// where ingestion resumes. A non-zero hash distinguishes an L1-anchored
-    /// block-zero genesis from the unanchored template.
+    /// Resume after the furthest block represented by durable Zone state or
+    /// the subscriber's in-memory queue and tracker.
     pub(crate) fn resolve_start_block(&self) -> Result<u64, L1SubscriberError> {
         let state = self.zone_provider.latest().map_err(eyre::Report::from)?;
         let local_checkpoint = state.tempo_num_hash().map_err(eyre::Report::from)?;
         if local_checkpoint.hash == B256::ZERO {
             return Err(eyre::eyre!("zone genesis is not anchored to an L1 block").into());
         }
-        let local_tempo_block_number = local_checkpoint.number;
-        info!(local_tempo_block_number, "Resuming from local zone state");
-        Ok(local_tempo_block_number + 1)
-    }
-
-    /// Resolve the first L1 block that has not already been ingested.
-    pub(crate) fn next_block_to_sync(&self) -> Result<u64, L1SubscriberError> {
-        let resolved = self.resolve_start_block()?;
-        let queued = self
-            .deposit_queue
-            .last_enqueued()
-            .map(|last| last.number.saturating_add(1));
-        let observed = self
-            .block_tracker
-            .latest()
-            .map(|last| last.number.saturating_add(1));
-
-        let next = [Some(resolved), queued, observed]
-            .into_iter()
-            .flatten()
-            .max()
-            .expect("resolved checkpoint is always present");
-
-        // Only the persisted zone checkpoint proves consumption.
-        // Queue and observation cursors are fetch high-water marks.
-        self.block_tracker
-            .initialize_consumed_through(resolved.saturating_sub(1));
-        Ok(next)
+        let next_block = local_checkpoint
+            .number
+            .saturating_add(1)
+            .max(
+                self.deposit_queue
+                    .last_enqueued()
+                    .map_or(0, |block| block.number.saturating_add(1)),
+            )
+            .max(self.block_tracker.next_observation_number().unwrap_or(0));
+        info!(
+            local_tempo_block_number = local_checkpoint.number,
+            next_block, "Resuming L1 sync"
+        );
+        Ok(next_block)
     }
 
-    /// Return the block number referenced by the L1 `finalized` tag.
-    async fn finalized_block_number(
+    /// Synchronize all missing blocks through the current finalized L1 head.
+    ///
+    /// The cursor advances after each block is fully applied.
+    pub(crate) async fn sync_finalized(
         &self,
         l1_provider: &impl Provider<TempoNetwork>,
-    ) -> Result<u64, L1SubscriberError> {
-        Ok(l1_provider
+        next_block: &mut u64,
+    ) -> Result<(), L1SubscriberError> {
+        let finalized = l1_provider
             .get_header_by_number(BlockNumberOrTag::Finalized)
             .await
             .inspect_err(|_| self.subscriber_metrics.fetch_failures.increment(1))?
             .map(|header| header.number())
-            .ok_or_eyre("L1 finalized block is not available")?)
-    }
+            .ok_or_eyre("L1 finalized block is not available")?;
 
-    /// Synchronize all missing blocks through the current finalized L1 head.
-    ///
-    /// Callers provide the next block number and receive the next cursor after
-    /// a successful sync.
-    pub(crate) async fn sync_finalized_once(
-        &self,
-        l1_provider: &impl Provider<TempoNetwork>,
-        next_block: u64,
-    ) -> Result<u64, L1SubscriberError> {
-        let finalized = self.finalized_block_number(l1_provider).await?;
-        if next_block > finalized {
-            self.record_seen_block(finalized, 0);
-            return Ok(next_block);
+        let pending_blocks = finalized.saturating_sub(next_block.saturating_sub(1));
+        self.record_seen_block(finalized, pending_blocks);
+        if pending_blocks == 0 {
+            return Ok(());
         }
 
-        let blocks = finalized - next_block + 1;
-        self.record_seen_block(finalized, blocks);
         info!(
-            from = next_block,
+            from = *next_block,
             to = finalized,
-            blocks,
+            blocks = pending_blocks,
             "Synchronizing finalized L1 blocks"
         );
 
@@ -599,29 +575,7 @@ where
             .backfill_duration_seconds
             .record(start.elapsed().as_secs_f64());
         self.subscriber_metrics.current_l1_lag_blocks.set(0.0);
-        Ok(finalized.saturating_add(1))
-    }
-
-    /// Follow finalized L1 using transport-specific head notifications as wakeups.
-    ///
-    /// Header contents are intentionally ignored. Canonical block selection is
-    /// always based on the `finalized` tag read by [`Self::sync_finalized_once`].
-    pub(crate) async fn follow_finalized(
-        &self,
-        l1_provider: &impl Provider<TempoNetwork>,
-        mut stream: HeaderStream,
-    ) -> Result<(), L1SubscriberError> {
-        let mut next_block = self.next_block_to_sync()?;
-
-        // Subscribe before the initial sync so a head published while catching
-        // up remains queued in the stream.
-        next_block = self.sync_finalized_once(l1_provider, next_block).await?;
-
-        while stream.next().await.is_some() {
-            next_block = self.sync_finalized_once(l1_provider, next_block).await?;
-        }
-
-        Err(eyre::eyre!("L1 head notification stream ended").into())
+        Ok(())
     }
 
     /// Backfill L1 blocks from `from..=to` with pipelined RPC fetching.
@@ -630,15 +584,16 @@ where
     /// parallel, then processes them sequentially (event extraction and enqueue).
     /// Receipts are fetched by the corresponding block
     /// hash and validated against the header's receipts root before processing.
-    #[instrument(skip(self, l1_provider), fields(from, to))]
+    #[instrument(skip(self, l1_provider, next_block), fields(from = *next_block, to))]
     async fn backfill(
         &self,
         l1_provider: &impl Provider<TempoNetwork>,
-        from: u64,
+        next_block: &mut u64,
         to: u64,
     ) -> Result<(), L1SubscriberError> {
         use futures::stream;
 
+        let from = *next_block;
         let concurrency = self.config.l1_fetch_concurrency.max(1);
         let subscriber_metrics = self.subscriber_metrics.clone();
         let block_tracker = self.block_tracker.clone();
@@ -766,6 +721,7 @@ where
             // configured retention sink and the contiguous observation tracker.
             self.apply_enabled_token_events(&events);
             self.update_l1_state_anchor(block_number, &invalidated);
+            *next_block = block_number.saturating_add(1);
             if appended {
                 self.subscriber_metrics.blocks_enqueued.increment(1);
             }
@@ -804,16 +760,28 @@ where
     ///
     /// Transport and ordinary failures reconnect after the configured retry interval.
     /// Deterministic failures while applying a receipt-verified finalized block are fatal.
-    pub async fn run(self) {
+    pub async fn run(self) -> Result<(), L1SubscriberError> {
         loop {
-            let result = async {
+            let result: Result<(), L1SubscriberError> = async {
+                let mut next_block = self.resolve_start_block()?;
+                self.block_tracker
+                    .initialize_consumed_through(next_block.saturating_sub(1));
+
                 let provider = self.connect().await?;
-                let header_stream = self.subscribe_block_headers(&provider).await?;
+                let mut header_stream = self.subscribe_block_headers(&provider).await?;
                 info!(
                     portal = %self.config.portal_address,
                     "Following finalized L1 blocks"
                 );
-                self.follow_finalized(&provider, header_stream).await
+
+                // Subscribe before the initial sync so a head published while catching
+                // up remains queued in the stream.
+                self.sync_finalized(&provider, &mut next_block).await?;
+                while let Some(()) = header_stream.next().await {
+                    self.sync_finalized(&provider, &mut next_block).await?;
+                }
+
+                Err(eyre::eyre!("L1 head notification stream ended").into())
             }
             .await;
 
@@ -829,7 +797,7 @@ where
                     );
                     tokio::time::sleep(retry_interval).await;
                 } else {
-                    panic!("{error}");
+                    return Err(error);
                 }
             }
         }

crates/l1/src/tests.rs

@@ -727,6 +727,22 @@ fn test_resolve_start_block_reads_live_local_state_each_time() {
     assert_eq!(subscriber.resolve_start_block().unwrap(), 12);
 }
 
+#[test]
+fn test_resolve_start_block_does_not_rewind_in_memory_progress() {
+    let subscriber = test_subscriber(9);
+
+    subscriber
+        .deposit_queue
+        .enqueue(make_test_header(10), L1PortalEvents::default());
+    assert_eq!(subscriber.resolve_start_block().unwrap(), 11);
+
+    subscriber
+        .block_tracker
+        .record(NumHash::new(11, B256::with_last_byte(11)))
+        .unwrap();
+    assert_eq!(subscriber.resolve_start_block().unwrap(), 12);
+}
+
 #[test]
 fn test_resolve_start_block_accepts_block_zero_with_nonzero_hash() {
     let subscriber = test_subscriber(0);
@@ -746,7 +762,7 @@ fn test_resolve_start_block_rejects_unanchored_genesis() {
 }
 
 #[tokio::test]
-async fn test_follow_finalized_uses_new_heads_to_sync_missing_finalized_range() {
+async fn test_sync_finalized_ingests_missing_finalized_range() {
     let subscriber = test_subscriber(9);
     let asserter = Asserter::new();
     let l1_provider =
@@ -767,11 +783,16 @@ async fn test_follow_finalized_uses_new_heads_to_sync_missing_finalized_range()
     push_header_and_empty_receipts(&asserter, header_11);
     push_header_and_empty_receipts(&asserter, header_12);
 
-    let err = subscriber
-        .follow_finalized(&l1_provider, Box::pin(futures::stream::iter([()])))
+    let mut next_block = 10;
+    subscriber
+        .sync_finalized(&l1_provider, &mut next_block)
         .await
-        .expect_err("finite header stream should end the subscriber");
-    assert!(err.to_string().contains("head notification stream ended"));
+        .unwrap();
+    subscriber
+        .sync_finalized(&l1_provider, &mut next_block)
+        .await
+        .unwrap();
+    assert_eq!(next_block, 13);
 
     let blocks = subscriber.deposit_queue.drain();
     assert_eq!(
@@ -813,23 +834,64 @@ async fn test_subscribe_block_headers_falls_back_to_http_block_filter() {
 }
 
 #[tokio::test]
-async fn test_sync_finalized_once_does_not_refetch_current_cursor() {
+async fn test_sync_finalized_does_not_refetch_current_cursor() {
     let subscriber = test_subscriber(10);
     let asserter = Asserter::new();
     let l1_provider =
         ProviderBuilder::new_with_network::<TempoNetwork>().connect_mocked_client(asserter.clone());
     asserter.push_success(&Some(header_response(make_test_header(10))));
 
-    let next = subscriber
-        .sync_finalized_once(&l1_provider, 11)
+    let mut next_block = 11;
+    subscriber
+        .sync_finalized(&l1_provider, &mut next_block)
         .await
         .unwrap();
 
-    assert_eq!(next, 11);
+    assert_eq!(next_block, 11);
     assert!(subscriber.deposit_queue.drain().is_empty());
     assert!(asserter.read_q().is_empty());
 }
 
+#[tokio::test]
+async fn test_sync_finalized_preserves_progress_after_partial_failure() {
+    let subscriber = test_subscriber(9);
+    let asserter = Asserter::new();
+    let l1_provider =
+        ProviderBuilder::new_with_network::<TempoNetwork>().connect_mocked_client(asserter.clone());
+    let header_10 = make_test_header(10);
+    let header_11 = make_chained_header(11, header_hash(&header_10));
+
+    asserter.push_success(&Some(header_response(header_11.clone())));
+    push_header_and_empty_receipts(&asserter, header_10);
+    asserter.push_failure_msg("temporary block fetch failure");
+
+    let mut next_block = 10;
+    subscriber
+        .sync_finalized(&l1_provider, &mut next_block)
+        .await
+        .expect_err("the first attempt should fail while fetching block 11");
+    assert_eq!(next_block, 11);
+
+    asserter.push_success(&Some(header_response(header_11.clone())));
+    push_header_and_empty_receipts(&asserter, header_11);
+    subscriber
+        .sync_finalized(&l1_provider, &mut next_block)
+        .await
+        .unwrap();
+
+    assert_eq!(next_block, 12);
+    assert_eq!(
+        subscriber
+            .deposit_queue
+            .drain()
+            .iter()
+            .map(|block| block.header.number())
+            .collect::<Vec<_>>(),
+        vec![10, 11]
+    );
+    assert!(asserter.read_q().is_empty());
+}
+
 #[test]
 fn test_push_log_decodes_withdrawal_bounce_back() {
     let portal_address = address!("0x0000000000000000000000000000000000000ABC");
@@ -1730,8 +1792,9 @@ async fn sync_classifies_corrupt_recognized_portal_log_as_fatal() {
     asserter.push_success(&Some(header_response(header_10)));
     asserter.push_success(&Some(vec![receipt]));
 
+    let mut next_block = 10;
     let err = subscriber
-        .sync_finalized_once(&l1_provider, 10)
+        .sync_finalized(&l1_provider, &mut next_block)
         .await
         .unwrap_err();
     assert!(matches!(
@@ -1797,13 +1860,12 @@ async fn sync_applies_leadership_transition_before_enqueueing_the_activation_blo
     asserter.push_success(&Some(header_response(header_10.clone())));
     asserter.push_success(&Some(vec![receipt]));
 
-    assert_eq!(
-        subscriber
-            .sync_finalized_once(&l1_provider, 10)
-            .await
-            .unwrap(),
-        11
-    );
+    let mut next_block = 10;
+    subscriber
+        .sync_finalized(&l1_provider, &mut next_block)
+        .await
+        .unwrap();
+    assert_eq!(next_block, 11);
 
     let seen = sink.seen.lock();
     assert_eq!(seen.len(), 1);
@@ -1843,8 +1905,9 @@ async fn sync_fails_fatally_when_the_leadership_sink_rejects_the_transition() {
     asserter.push_success(&Some(header_response(header_10.clone())));
     asserter.push_success(&Some(vec![receipt]));
 
+    let mut next_block = 10;
     let err = subscriber
-        .sync_finalized_once(&l1_provider, 10)
+        .sync_finalized(&l1_provider, &mut next_block)
         .await
         .unwrap_err();
     assert!(matches!(

crates/node/src/node.rs

@@ -664,7 +664,15 @@ where
             self.encryption_keys.clone(),
         );
         let task_executor = ctx.node.task_executor().clone();
-        task_executor.spawn_critical_task("l1-block-subscriber", Box::pin(l1_subscriber.run()));
+        task_executor.spawn_critical_task(
+            "l1-block-subscriber",
+            Box::pin(async move {
+                l1_subscriber
+                    .run()
+                    .await
+                    .unwrap_or_else(|error| panic!("{error}"));
+            }),
+        );
         info!(target: "reth::cli", "L1 subscriber started with deposit enqueueing");
 
         // Start the Commonware network and the long-lived event router