Fix AltDA batcher (#174)

Co-authored-by: Philippe Camacho <[email protected]>

Author: QuentinI

espresso/environment/2_espresso_liveness_test.go

@@ -283,7 +283,7 @@ func TestE2eDevNetWithEspressoEspressoDegradedLivenessViaCaffNode(t *testing.T)
 		require.NoError(t, err, "failed to get safe L2 block ref")
 		finalizedL1BlockRef, err := l1RefClient.L1BlockRefByLabel(streamBlocksCtx, eth.Finalized)
 		require.NoError(t, err, "failed to get finalized L1 block ref")
-		_, err = streamer.Refresh(streamBlocksCtx, finalizedL1BlockRef, l2BlockRef.Number, l2BlockRef.L1Origin)
+		err = streamer.Refresh(streamBlocksCtx, finalizedL1BlockRef, l2BlockRef.Number, l2BlockRef.L1Origin)
 		require.NoError(t, err, "failed to refresh streamer")
 		lastTransaction := transactions[N-1]
 
@@ -312,7 +312,7 @@ func TestE2eDevNetWithEspressoEspressoDegradedLivenessViaCaffNode(t *testing.T)
 				safeL2, safeL2Error := l2RefClient.L2BlockRefByLabel(ctx, eth.Safe)
 				if finalizedL1Err == nil && safeL2Error == nil {
 					// Refresh the Streamer with the latest finalized L1 and safe L2
-					_, err := streamer.Refresh(ctx, finalizedL1, safeL2.Number, safeL2.L1Origin)
+					err := streamer.Refresh(ctx, finalizedL1, safeL2.Number, safeL2.L1Origin)
 					if have, want := err, error(nil); have != want {
 						// NOTE: we are in a go-routine here, so we are unable
 						// to fail fatally here. Instead, we'll Fail and and

espresso/streamer.go

@@ -130,24 +130,24 @@ func (s *EspressoStreamer[B]) Reset() {
 }
 
 // Handle both L1 reorgs and batcher restarts by updating our state in case it is
-// not consistent with what's on the L1. Returns true if the state was updated.
-func (s *EspressoStreamer[B]) Refresh(ctx context.Context, finalizedL1 eth.L1BlockRef, safeBatchNumber uint64, safeL1Origin eth.BlockID) (bool, error) {
+// not consistent with what's on the L1.
+func (s *EspressoStreamer[B]) Refresh(ctx context.Context, finalizedL1 eth.L1BlockRef, safeBatchNumber uint64, safeL1Origin eth.BlockID) error {
 	s.FinalizedL1 = finalizedL1
 
 	err := s.confirmEspressoBlockHeight(safeL1Origin)
 	if err != nil {
-		return false, err
+		return err
 	}
 
 	// NOTE: be sure to update s.finalizedL1 before checking this condition and returning
 	if s.fallbackBatchPos == safeBatchNumber {
 		// This means everything is in sync, no state update needed
-		return false, nil
+		return nil
 	}
 
 	s.fallbackBatchPos = safeBatchNumber
 	s.Reset()
-	return true, nil
+	return nil
 }
 
 func (s *EspressoStreamer[B]) CheckBatch(ctx context.Context, batch B) (BatchValidity, int) {

espresso/streamer_test.go

@@ -374,10 +374,7 @@ func TestStreamerSmoke(t *testing.T) {
 
 	// update the state of our streamer
 	syncStatus := state.SyncStatus()
-	updated, err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
-	if have, want := updated, false; have != want {
-		t.Fatalf("failed to refresh streamer state:\nhave:\n\t%v\nwant:\n\t%v\n", updated, want)
-	}
+	err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
 
 	if have, want := err, error(nil); have != want {
 		t.Fatalf("failed to refresh streamer state encountered error:\nhave:\n\t\"%v\"\nwant:\n\t\"%v\"\n", have, want)
@@ -415,7 +412,7 @@ func TestEspressoStreamerSimpleIncremental(t *testing.T) {
 	for i := 0; i < N; i++ {
 		// update the state of our streamer
 		syncStatus := state.SyncStatus()
-		_, err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
+		err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
 
 		if have, want := err, error(nil); have != want {
 			t.Fatalf("failed to refresh streamer state encountered error:\nhave:\n\t\"%v\"\nwant:\n\t\"%v\"\n", have, want)
@@ -477,7 +474,7 @@ func TestEspressoStreamerIncrementalDelayedConsumption(t *testing.T) {
 
 	// update the state of our streamer
 	syncStatus := state.SyncStatus()
-	_, err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
+	err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
 
 	for i := 0; i < N; i++ {
 		batch, _, _, espTxnInBlock := state.CreateEspressoTxnData(
@@ -545,7 +542,7 @@ func TestStreamerEspressoOutOfOrder(t *testing.T) {
 
 	// update the state of our streamer
 	syncStatus := state.SyncStatus()
-	_, err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
+	err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
 
 	if have, want := err, error(nil); have != want {
 		t.Fatalf("failed to refresh streamer state encountered error:\nhave:\n\t\"%v\"\nwant:\n\t\"%v\"\n", have, want)
@@ -630,7 +627,7 @@ func TestEspressoStreamerDuplicationHandling(t *testing.T) {
 
 	// update the state of our streamer
 	syncStatus := state.SyncStatus()
-	_, err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
+	err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
 
 	if have, want := err, error(nil); have != want {
 		t.Fatalf("failed to refresh streamer state encountered error:\nhave:\n\t\"%v\"\nwant:\n\t\"%v\"\n", have, want)
@@ -651,7 +648,7 @@ func TestEspressoStreamerDuplicationHandling(t *testing.T) {
 		for j := 0; j < 2; j++ {
 			// update the state of our streamer
 			syncStatus := state.SyncStatus()
-			_, err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
+			err := streamer.Refresh(ctx, syncStatus.FinalizedL1, syncStatus.SafeL2.Number, syncStatus.SafeL2.L1Origin)
 
 			require.NoError(t, err)
 

op-batcher/batcher/espresso.go

@@ -656,8 +656,10 @@ func (l *BatchSubmitter) queueBlockToEspresso(ctx context.Context, block *types.
 }
 
 func (l *BatchSubmitter) espressoSyncAndRefresh(ctx context.Context, newSyncStatus *eth.SyncStatus) {
-	shouldClearState, err := l.streamer.Refresh(ctx, newSyncStatus.FinalizedL1, newSyncStatus.SafeL2.Number, newSyncStatus.SafeL2.L1Origin)
-	shouldClearState = shouldClearState || err != nil
+	err := l.streamer.Refresh(ctx, newSyncStatus.FinalizedL1, newSyncStatus.SafeL2.Number, newSyncStatus.SafeL2.L1Origin)
+	if err != nil {
+		l.Log.Warn("Failed to refresh Espresso streamer", "err", err)
+	}
 
 	l.channelMgrMutex.Lock()
 	defer l.channelMgrMutex.Unlock()
@@ -667,10 +669,7 @@ func (l *BatchSubmitter) espressoSyncAndRefresh(ctx context.Context, newSyncStat
 		return
 	}
 	l.prevCurrentL1 = newSyncStatus.CurrentL1
-	if syncActions.clearState == nil && shouldClearState {
-		l.channelMgr.Clear(newSyncStatus.SafeL2.L1Origin)
-		l.streamer.Reset()
-	} else if syncActions.clearState != nil {
+	if syncActions.clearState != nil {
 		l.channelMgr.Clear(*syncActions.clearState)
 		l.streamer.Reset()
 	} else {

op-node/rollup/derive/attributes_queue.go

@@ -144,7 +144,7 @@ func CaffNextBatch(s *espresso.EspressoStreamer[EspressoBatch], ctx context.Cont
 		return nil, false, err
 	}
 	// Refresh the sync status
-	if _, err := s.Refresh(ctx, finalizedL1Block, parent.Number, parent.L1Origin); err != nil {
+	if err := s.Refresh(ctx, finalizedL1Block, parent.Number, parent.L1Origin); err != nil {
 		return nil, false, err
 	}