fix: address PR review feedback on batch queue durability

- fail Load on datastore read errors instead of dropping txs
- detach rollback WAL cleanup from drain context so it runs on shutdown
- raise queue load timeout for large WAL backlogs
- merge postponed-tx collection into the filter loop
- document DropIncluded aliasing, SetPostponed contract, txSeen bound
- add tests: executor ack retry, reconcile crash recovery, bulk-prepend rollback

Author: chatton

block/internal/executing/executor_logic_test.go

@@ -122,6 +122,61 @@ func TestPendingLimit_SkipsProduction(t *testing.T) {
 	assert.Equal(t, h1, h2, "height should not change when production is skipped")
 }
 
+func TestProduceBlock_RetriesFailedAckBeforeNextBlock(t *testing.T) {
+	fx := setupTestExecutor(t, 1000)
+	defer fx.Cancel()
+
+	var ackCalls int
+	fx.Exec.SetOnBatchCommitted(func(ctx context.Context) error {
+		ackCalls++
+		if ackCalls <= 2 {
+			return assert.AnError
+		}
+		return nil
+	})
+
+	fx.MockSeq.EXPECT().GetNextBatch(mock.Anything, mock.AnythingOfType("sequencer.GetNextBatchRequest")).
+		RunAndReturn(func(ctx context.Context, req coreseq.GetNextBatchRequest) (*coreseq.GetNextBatchResponse, error) {
+			return &coreseq.GetNextBatchResponse{Batch: &coreseq.Batch{Transactions: nil}, Timestamp: time.Now()}, nil
+		}).Once()
+	fx.MockExec.EXPECT().ExecuteTxs(mock.Anything, mock.Anything, uint64(1), mock.AnythingOfType("time.Time"), fx.InitStateRoot).
+		Return([]byte("root1"), nil).Once()
+	fx.MockSeq.EXPECT().GetDAHeight().Return(uint64(0)).Once()
+
+	// block 1 commits, but the post-commit ack fails (non-fatal)
+	require.NoError(t, fx.Exec.ProduceBlock(fx.Exec.ctx))
+	assert.Equal(t, 1, ackCalls)
+	assert.True(t, fx.Exec.pendingBatchAck.Load())
+
+	// the pending ack is retried before draining a new batch; if the retry
+	// fails again block production aborts without calling GetNextBatch
+	err := fx.Exec.ProduceBlock(fx.Exec.ctx)
+	require.Error(t, err)
+	assert.Equal(t, 2, ackCalls)
+	assert.True(t, fx.Exec.pendingBatchAck.Load())
+
+	h, err := fx.MemStore.Height(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, uint64(1), h, "no block should be produced while the ack retry fails")
+
+	// once the retry succeeds production resumes: retry ack + post-commit ack
+	fx.MockSeq.EXPECT().GetNextBatch(mock.Anything, mock.AnythingOfType("sequencer.GetNextBatchRequest")).
+		RunAndReturn(func(ctx context.Context, req coreseq.GetNextBatchRequest) (*coreseq.GetNextBatchResponse, error) {
+			return &coreseq.GetNextBatchResponse{Batch: &coreseq.Batch{Transactions: nil}, Timestamp: time.Now()}, nil
+		}).Once()
+	fx.MockExec.EXPECT().ExecuteTxs(mock.Anything, mock.Anything, uint64(2), mock.AnythingOfType("time.Time"), []byte("root1")).
+		Return([]byte("root2"), nil).Once()
+	fx.MockSeq.EXPECT().GetDAHeight().Return(uint64(0)).Once()
+
+	require.NoError(t, fx.Exec.ProduceBlock(fx.Exec.ctx))
+	assert.Equal(t, 4, ackCalls)
+	assert.False(t, fx.Exec.pendingBatchAck.Load())
+
+	h, err = fx.MemStore.Height(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, uint64(2), h)
+}
+
 func TestExecutor_executeTxsWithRetry(t *testing.T) {
 	t.Parallel()
 

pkg/sequencers/single/queue.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"strconv"
 	"sync"
+	"time"
 
 	ds "github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/query"
@@ -56,6 +57,8 @@ type BatchQueue struct {
 	// txSeen is an in-memory dedup set keyed by sha256 hash of each tx.
 	// hashes are added in AddBatch and removed on successful Ack.
 	// prevents the reaper from enqueuing the same tx multiple scrape cycles.
+	// its size tracks queued + in-flight txs, so it is bounded indirectly
+	// by maxQueueSize (unbounded when maxQueueSize is 0).
 	txSeen map[[32]byte]struct{}
 
 	// totalEnqueued counts batches ever enqueued via AddBatch. Monotonic,
@@ -134,7 +137,7 @@ func (bq *BatchQueue) AddBatch(ctx context.Context, batch coresequencer.Batch) e
 	key := seqToKey(bq.nextAddSeq)
 	if err := bq.persistBatch(ctx, batch, key); err != nil {
 		bq.rollbackSeenLocked(hashes)
-		return err
+		return fmt.Errorf("failed to persist batch %s to WAL: %w", key, err)
 	}
 	bq.nextAddSeq++
 
@@ -198,8 +201,11 @@ func (bq *BatchQueue) Drain(ctx context.Context, maxBytes uint64) (*coresequence
 }
 
 // SetPostponed records txs that should be requeued on the next Ack.
-// Must be called between Drain and Ack. The queue owns this state so
-// it is only cleared on successful Ack — no data loss on failure.
+// Must be called at most once between a Drain and its Ack — a later call in
+// the same cycle would overwrite the recorded txs. The postponedItem guard
+// makes a repeated call after a failed Ack a no-op, so the already-persisted
+// entry is not replaced. The queue owns this state so it is only cleared on
+// successful Ack — no data loss on failure.
 func (bq *BatchQueue) SetPostponed(txs [][]byte) {
 	bq.mu.Lock()
 	defer bq.mu.Unlock()
@@ -305,7 +311,11 @@ func (bq *BatchQueue) rollbackInFlightLocked(ctx context.Context) {
 	}
 
 	if bq.postponedItem != nil {
-		if err := bq.db.Delete(ctx, ds.NewKey(bq.postponedItem.Key)); err != nil {
+		// detach from the caller's context so this best-effort cleanup still
+		// runs during graceful shutdown when the drain context is cancelled
+		delCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+		defer cancel()
+		if err := bq.db.Delete(delCtx, ds.NewKey(bq.postponedItem.Key)); err != nil {
 			bq.logger.Warn().Err(err).Str("key", bq.postponedItem.Key).
 				Msg("failed to delete rolled-back postponed WAL entry")
 		}
@@ -395,6 +405,9 @@ func (bq *BatchQueue) DropIncluded(ctx context.Context, included [][]byte) (int,
 	}
 
 	var dropped int
+	// in-place compaction: kept aliases bq.queue's backing array. This is safe
+	// because range evaluates the slice header once and writes always trail
+	// reads (at most one item is appended per item iterated).
 	kept := bq.queue[:bq.head]
 	for _, item := range bq.queue[bq.head:] {
 		remaining := make([][]byte, 0, len(item.Batch.Transactions))
@@ -456,8 +469,9 @@ func (bq *BatchQueue) Load(ctx context.Context) error {
 	var legacyItems []queuedItem
 	for result := range results.Next() {
 		if result.Error != nil {
-			bq.logger.Error().Err(result.Error).Msg("failed to read entry from datastore")
-			continue
+			// a datastore read failure means the WAL cannot be trusted as
+			// loaded — fail startup rather than silently dropping txs.
+			return fmt.Errorf("failed to read WAL entry from datastore: %w", result.Error)
 		}
 		// We care about the last part of the key (the sequence number)
 		// ds.Key usually has a leading slash.

pkg/sequencers/single/queue_test.go

@@ -735,6 +735,38 @@ func TestBatchQueue_Drain_RollsBackUnackedInFlight(t *testing.T) {
 	assert.Equal(t, []byte("tx2"), batch2.Transactions[1])
 }
 
+func TestBatchQueue_Drain_RollbackBulkPrependAfterCompact(t *testing.T) {
+	ctx := context.Background()
+	db := ds.NewMapDatastore()
+	queue := NewBatchQueue(db, "test-drain-rollback-bulk", 0, zerolog.Nop())
+	require.NoError(t, queue.Load(ctx))
+
+	// drain >100 entries so compactLocked resets head to 0 while they
+	// are in flight, forcing the bulk-prepend rollback path on next drain
+	const n = 150
+	for i := range n {
+		tx := []byte(fmt.Sprintf("tx-%03d", i))
+		require.NoError(t, queue.AddBatch(ctx, coresequencer.Batch{Transactions: [][]byte{tx}}))
+	}
+
+	batch1, err := queue.Drain(ctx, 0)
+	require.NoError(t, err)
+	require.Len(t, batch1.Transactions, n)
+	require.Equal(t, 0, queue.head, "compact should reset head while entries are in flight")
+
+	// enqueue one more so the rollback has a tail to prepend in front of
+	require.NoError(t, queue.AddBatch(ctx, coresequencer.Batch{Transactions: [][]byte{[]byte("tx-new")}}))
+
+	// second drain without ack rolls back via the bulk-prepend path
+	batch2, err := queue.Drain(ctx, 0)
+	require.NoError(t, err)
+	require.Len(t, batch2.Transactions, n+1)
+	for i := range n {
+		assert.Equal(t, []byte(fmt.Sprintf("tx-%03d", i)), batch2.Transactions[i])
+	}
+	assert.Equal(t, []byte("tx-new"), batch2.Transactions[n])
+}
+
 func TestBatchQueue_Ack_DeletesWALEntries(t *testing.T) {
 	ctx := context.Background()
 	db := ds.NewMapDatastore()

pkg/sequencers/single/sequencer.go

@@ -94,7 +94,9 @@ func NewSequencer(
 	s.SetDAHeight(genesis.DAStartHeight) // default value, will be overridden by executor or submitter
 	s.daStartHeight.Store(genesis.DAStartHeight)
 
-	loadCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	// generous timeout: loading and reconciling a large WAL backlog
+	// (e.g. after a burst followed by a crash) can take a while
+	loadCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
 	defer cancel()
 
 	// Load batch queue from DB
@@ -321,6 +323,7 @@ func (c *Sequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextB
 	// TxIndex tracks consumed txs from the start of the epoch, so we must process in order.
 	var validForcedTxs [][]byte
 	var validMempoolTxs [][]byte
+	var postponedMempoolTxs [][]byte
 	var forcedTxConsumedCount uint64
 	var forcedTxPostponed bool
 
@@ -347,19 +350,15 @@ func (c *Sequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextB
 			switch status {
 			case execution.FilterOK:
 				validMempoolTxs = append(validMempoolTxs, allTxs[i])
-			case execution.FilterPostpone, execution.FilterRemove:
-				// Mempool txs that are postponed/removed are handled separately
+			case execution.FilterPostpone:
+				// requeued at ack time (after block commit)
+				postponedMempoolTxs = append(postponedMempoolTxs, allTxs[i])
+			case execution.FilterRemove:
+				// dropped permanently
 			}
 		}
 	}
 
-	// collect postponed mempool txs to be requeued at ack time (after block commit)
-	var postponedMempoolTxs [][]byte
-	for i, status := range filterStatuses {
-		if i >= forcedTxCount && status == execution.FilterPostpone {
-			postponedMempoolTxs = append(postponedMempoolTxs, allTxs[i])
-		}
-	}
 	c.queue.SetPostponed(postponedMempoolTxs)
 
 	// Update checkpoint after consuming forced inclusion transactions.

pkg/sequencers/single/sequencer_test.go

@@ -19,8 +19,10 @@ import (
 	"github.com/evstack/ev-node/pkg/config"
 	datypes "github.com/evstack/ev-node/pkg/da/types"
 	"github.com/evstack/ev-node/pkg/genesis"
+	"github.com/evstack/ev-node/pkg/store"
 	"github.com/evstack/ev-node/test/mocks"
 	"github.com/evstack/ev-node/test/testda"
+	"github.com/evstack/ev-node/types"
 )
 
 // MockFullDAClient combines MockClient and MockVerifier to implement FullDAClient
@@ -2309,3 +2311,35 @@ func TestSequencer_GetNextBatch_GasFilteringPreservesUnprocessedTxs(t *testing.T
 	assert.True(t, txFound[string(tx2)], "tx2 should have been processed (must not be lost)")
 	assert.True(t, txFound[string(tx3)], "tx3 should have been processed (must not be lost)")
 }
+
+func TestSequencer_ReconcileQueueWithLastBlock_CrashRecovery(t *testing.T) {
+	ctx := context.Background()
+	db := ds.NewMapDatastore()
+
+	// commit a block at height 1 containing two txs
+	header, data := types.GetRandomBlock(1, 2, "test")
+	evStore := store.New(store.NewEvNodeKVStore(db))
+	batch, err := evStore.NewBatch(ctx)
+	require.NoError(t, err)
+	require.NoError(t, batch.SaveBlockData(header, data, &header.Signature))
+	require.NoError(t, batch.SetHeight(1))
+	require.NoError(t, batch.Commit())
+
+	committed1 := []byte(data.Txs[0])
+	committed2 := []byte(data.Txs[1])
+	pending := []byte("tx-still-pending")
+
+	// simulate a crash between block commit and queue ack: WAL still holds
+	// the committed txs alongside a tx that was never included
+	queue := NewBatchQueue(db, "batches", 0, zerolog.Nop())
+	require.NoError(t, queue.AddBatch(ctx, coresequencer.Batch{Transactions: [][]byte{committed1, committed2}}))
+	require.NoError(t, queue.AddBatch(ctx, coresequencer.Batch{Transactions: [][]byte{pending}}))
+
+	// restart: NewSequencer loads the queue and reconciles against the last block
+	seq := newTestSequencer(t, db, newDummyDA(100_000_000))
+
+	drained, err := seq.queue.Drain(ctx, 0)
+	require.NoError(t, err)
+	require.Len(t, drained.Transactions, 1, "committed txs should have been dropped on reconcile")
+	assert.Equal(t, pending, drained.Transactions[0])
+}