persist: review feedback — disable fsync, harden ReadAll, fix Close

- Disable WAL fsync; the prune anchor (A/B files with fsync) already
  provides the crash-recovery watermark.
- Use Count() == 0 instead of firstIdx == 0 for emptiness checks in
  Write and ReadAll for robustness.
- Add post-replay count check in ReadAll to detect silent data loss.
- Use errors.Join in BlockPersister.Close to close all lane WALs.
- Rename laneDir → lanePath to avoid shadowing the laneDir() function.
- Add TestIndexedWAL_ReadAllDetectsStaleNextIdx.

Made-with: Cursor

Author: wen-coding

sei-tendermint/internal/autobahn/consensus/persist/blocks.go

@@ -2,6 +2,7 @@ package persist
 
 import (
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -97,17 +98,17 @@ func NewBlockPersister(stateDir utils.Option[string]) (*BlockPersister, map[type
 			logger.Warn("skipping lane dir with invalid key", "name", e.Name(), "err", err)
 			continue
 		}
-		laneDir := filepath.Join(dir, e.Name())
-		lw, err := newLaneWAL(laneDir)
+		lanePath := filepath.Join(dir, e.Name())
+		lw, err := newLaneWAL(lanePath)
 		if err != nil {
 			_ = bp.Close()
-			return nil, nil, fmt.Errorf("open lane WAL in %s: %w", laneDir, err)
+			return nil, nil, fmt.Errorf("open lane WAL in %s: %w", lanePath, err)
 		}
 		loaded, err := lw.loadAll()
 		if err != nil {
 			_ = lw.Close()
 			_ = bp.Close()
-			return nil, nil, fmt.Errorf("load lane WAL in %s: %w", laneDir, err)
+			return nil, nil, fmt.Errorf("load lane WAL in %s: %w", lanePath, err)
 		}
 		bp.lanes[lane] = lw
 		if len(loaded) > 0 {
@@ -201,12 +202,13 @@ func (bp *BlockPersister) Close() error {
 	if _, ok := bp.dir.Get(); !ok {
 		return nil
 	}
+	var errs []error
 	for _, lw := range bp.lanes {
 		if err := lw.Close(); err != nil {
-			return err
+			errs = append(errs, err)
 		}
 	}
-	return nil
+	return errors.Join(errs...)
 }
 
 // loadAll reads all entries from the lane WAL and returns the loaded blocks.

sei-tendermint/internal/autobahn/consensus/persist/wal.go

@@ -26,9 +26,12 @@ type indexedWAL[T any] struct {
 	nextIdx  uint64   // WAL index that the next Write will be assigned
 }
 
-// openIndexedWAL creates (or opens) a WAL in dir with synchronous, unbatched,
-// fsync-enabled writes. Initializes index tracking from the WAL's stored
-// offsets so the caller can immediately Write, ReadAll, or TruncateBefore.
+// openIndexedWAL creates (or opens) a WAL in dir with synchronous, unbatched
+// writes. Fsync is disabled because the prune anchor (persisted via A/B files
+// with fsync) is the crash-recovery watermark — on power loss we restart from
+// the anchor and re-sync any lost WAL entries from peers.
+// Initializes index tracking from the WAL's stored offsets so the caller can
+// immediately Write, ReadAll, or TruncateBefore.
 func openIndexedWAL[T any](dir string, codec codec[T]) (*indexedWAL[T], error) {
 	if err := os.MkdirAll(dir, 0700); err != nil {
 		return nil, fmt.Errorf("create dir %s: %w", dir, err)
@@ -42,7 +45,6 @@ func openIndexedWAL[T any](dir string, codec codec[T]) (*indexedWAL[T], error) {
 		dbwal.Config{
 			WriteBufferSize: 0, // synchronous writes
 			WriteBatchSize:  1, // no batching
-			FsyncEnabled:    true,
 		},
 	)
 	if err != nil {
@@ -73,7 +75,7 @@ func (w *indexedWAL[T]) Write(entry T) error {
 	if err := w.wal.Write(entry); err != nil {
 		return err
 	}
-	if w.firstIdx == 0 {
+	if w.Count() == 0 {
 		w.firstIdx = w.nextIdx
 	}
 	w.nextIdx++
@@ -101,7 +103,7 @@ func (w *indexedWAL[T]) TruncateBefore(walIdx uint64, verify func(T) error) erro
 
 // ReadAll returns all entries in the WAL. Returns nil if empty.
 func (w *indexedWAL[T]) ReadAll() ([]T, error) {
-	if w.firstIdx == 0 {
+	if w.Count() == 0 {
 		return nil, nil
 	}
 	entries := make([]T, 0, w.Count())
@@ -112,10 +114,15 @@ func (w *indexedWAL[T]) ReadAll() ([]T, error) {
 	if err != nil {
 		return nil, err
 	}
+	if uint64(len(entries)) != w.Count() {
+		return nil, fmt.Errorf("WAL replay returned %d entries, expected %d (possible silent data loss)", len(entries), w.Count())
+	}
 	return entries, nil
 }
 
 // Count returns the number of entries in the WAL.
+// firstIdx == 0 is the empty sentinel because tidwall/wal returns 0 for both
+// FirstIndex and LastIndex on an empty log, and real indices start at 1.
 func (w *indexedWAL[T]) Count() uint64 {
 	if w.firstIdx == 0 {
 		return 0

sei-tendermint/internal/autobahn/consensus/persist/wal_test.go

@@ -206,6 +206,28 @@ func TestIndexedWAL_Reset(t *testing.T) {
 	require.NoError(t, iw2.Close())
 }
 
+func TestIndexedWAL_ReadAllDetectsStaleNextIdx(t *testing.T) {
+	dir := t.TempDir()
+	iw, err := openIndexedWAL(dir, stringCodec{})
+	require.NoError(t, err)
+
+	require.NoError(t, iw.Write("a"))
+	require.NoError(t, iw.Write("b"))
+	require.Equal(t, uint64(2), iw.Count())
+
+	// Simulate stale internal state: advance nextIdx so Count() reports more
+	// entries than the WAL actually contains. ReadAll must return an error
+	// (either from Replay failing to read the missing entry, or from the
+	// post-replay count check).
+	iw.nextIdx++
+
+	_, err = iw.ReadAll()
+	require.Error(t, err)
+
+	iw.nextIdx--
+	require.NoError(t, iw.Close())
+}
+
 func TestIndexedWAL_WriteAfterTruncate(t *testing.T) {
 	dir := t.TempDir()