persist: replace file-per-entry with WAL and refactor into generic indexedWAL

sei-tendermint/internal/autobahn/avail/inner.go

@@ -9,6 +9,11 @@ import (
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
 )
 
+// TODO: when dynamic committee changes are supported, newly joined members
+// must be added to blocks, votes, nextBlockToPersist, and persistedBlockStart.
+// Currently all four are initialized once in newInner from c.Lanes().All().
+// BlockPersister creates lane WALs lazily inside PersistBlock, but the new
+// member must also appear in inner.blocks before the next persist cycle.
 type inner struct {
 	latestAppQC    utils.Option[*types.AppQC]
 	latestCommitQC utils.AtomicSend[utils.Option[*types.CommitQC]]
@@ -23,8 +28,10 @@ type inner struct {
 	// reconstructed from the blocks already on disk (see newInner).
 	//
 	// TODO: consider giving this its own AtomicSend to avoid waking unrelated
-	// inner waiters (PushVote, PushCommitQC, etc.) on every markBlockPersisted
-	// call. Only RecvBatch needs to be notified of cursor changes;
+	// inner waiters (PushVote, PushCommitQC, etc.) on markBlockPersisted calls.
+	// Now that blocks are persisted concurrently by lane (one notification per
+	// lane per batch, not per block), the frequency is lower, but still not
+	// ideal. Only RecvBatch needs to be notified of cursor changes;
 	// collectPersistBatch is in the same goroutine and reads it directly.
 	nextBlockToPersist map[types.LaneID]types.BlockNumber
 

sei-tendermint/internal/autobahn/avail/state.go

@@ -157,20 +157,18 @@ func NewState(key types.SecretKey, data *data.State, stateDir utils.Option[strin
 		return nil, err
 	}
 
-	// Delete files below the prune anchor that were filtered out by
-	// loadPersistedState. Also reset the CommitQC persister's cursor to
-	// match the post-prune range.
-	laneFirsts := make(map[types.LaneID]types.BlockNumber, len(inner.blocks))
-	for lane, q := range inner.blocks {
-		laneFirsts[lane] = q.first
-	}
-	if err := pers.blocks.DeleteBefore(laneFirsts); err != nil {
-		return nil, fmt.Errorf("prune stale block files: %w", err)
-	}
-	if err := pers.commitQCs.DeleteBefore(inner.commitQCs.first); err != nil {
-		return nil, fmt.Errorf("prune stale commitQC files: %w", err)
+	// Truncate WAL entries below the prune anchor that were filtered out by
+	// loadPersistedState.
+	if ls, ok := loaded.Get(); ok {
+		if anchor, ok := ls.pruneAnchor.Get(); ok {
+			if err := pers.blocks.DeleteBefore(anchor.CommitQC); err != nil {
+				return nil, fmt.Errorf("prune stale block WAL entries: %w", err)
+			}
+			if err := pers.commitQCs.DeleteBefore(anchor.CommitQC.Proposal().Index(), utils.Some(anchor.CommitQC)); err != nil {
+				return nil, fmt.Errorf("prune stale commitQC WAL entries: %w", err)
+			}
+		}
 	}
-	pers.commitQCs.ResetNext(inner.commitQCs.next)
 
 	return &State{
 		key:        key,
@@ -631,18 +629,20 @@ func (s *State) Run(ctx context.Context) error {
 
 // runPersist is the main loop for the persist goroutine.
 // Write order:
-//  1. Prune anchor (AppQC + CommitQC pair) — the crash-recovery watermark.
-//  2. CommitQCs in order, then publish LastCommitQC immediately
-//     so consensus can advance without waiting for block writes.
-//  3. Blocks per lane in order, markBlockPersisted after each.
-//  4. Prune old blocks and CommitQCs.
+//  1. Prune anchor (AppQC + CommitQC pair) — the crash-recovery watermark —
+//     followed by WAL truncation of blocks and CommitQCs. Truncation is
+//     co-located with the anchor write so the pruning point is derived
+//     directly from the anchor's CommitQC. Truncation must happen before
+//     writes because the WAL requires contiguous indices — if the anchor
+//     advanced past all persisted entries, DeleteBefore resets the WAL so
+//     new writes start clean.
+//  2. CommitQCs and blocks concurrently via scope.Parallel: one goroutine for
+//     CommitQCs, one goroutine per lane for blocks (sequential within each
+//     lane). Each goroutine publishes its result (markCommitQCsPersisted /
+//     markBlockPersisted) per entry so voting unblocks ASAP.
 //
 // The prune anchor is a pruning watermark: on restart we resume from it.
 //
-// Blocks are persisted one at a time with inner.nextBlockToPersist
-// updated after each write, so vote latency equals single-block write
-// time regardless of batch size.
-//
 // TODO: use a single WAL for anchor and CommitQCs to make
 // this atomic rather than relying on write order.
 func (s *State) runPersist(ctx context.Context, pers persisters) error {
@@ -654,51 +654,74 @@ func (s *State) runPersist(ctx context.Context, pers persisters) error {
 		}
 
 		// 1. Persist prune anchor first — establishes the crash-recovery watermark.
+		//    All WAL pruning is co-located here so the truncation point is
+		//    derived directly from the anchor, making the safety invariant
+		//    explicit: we only truncate entries the on-disk anchor covers.
+		//    Block WAL pruning must happen before writes because the WAL
+		//    requires contiguous indices — if the anchor advanced past all
+		//    persisted entries, DeleteBefore resets the WAL so new writes
+		//    start clean.
 		if anchor, ok := batch.pruneAnchor.Get(); ok {
 			if err := pers.pruneAnchor.Persist(PruneAnchorConv.Encode(anchor)); err != nil {
 				return fmt.Errorf("persist prune anchor: %w", err)
 			}
 			s.advancePersistedBlockStart(anchor.CommitQC)
 			lastPersistedAppQCNext = anchor.CommitQC.Proposal().Index() + 1
-		}
 
-		// 2. Persist new CommitQCs, then publish immediately so consensus
-		//    can advance without waiting for block writes or pruning.
-		for _, qc := range batch.commitQCs {
-			if err := pers.commitQCs.PersistCommitQC(qc); err != nil {
-				return fmt.Errorf("persist commitqc %d: %w", qc.Index(), err)
+			if err := pers.blocks.DeleteBefore(anchor.CommitQC); err != nil {
+				return fmt.Errorf("block deleteBefore: %w", err)
+			}
+			if err := pers.commitQCs.DeleteBefore(anchor.CommitQC.Proposal().Index(), utils.Some(anchor.CommitQC)); err != nil {
+				return fmt.Errorf("commitqc deleteBefore: %w", err)
 			}
-		}
-		if len(batch.commitQCs) > 0 {
-			s.markCommitQCsPersisted(batch.commitQCs[len(batch.commitQCs)-1])
 		}
 
-		// 3. Persist blocks (mark each individually for vote latency).
+		// 2. Persist CommitQCs and blocks concurrently. CommitQCs go in
+		//    one goroutine; blocks fan out one goroutine per lane (sequential
+		//    within each lane to preserve block-number ordering). Each
+		//    goroutine publishes its result (markCommitQCsPersisted /
+		//    markBlockPersisted) as soon as it finishes.
+		blocksByLane := make(map[types.LaneID][]*types.Signed[*types.LaneProposal])
 		for _, proposal := range batch.blocks {
-			h := proposal.Msg().Block().Header()
-			if err := pers.blocks.PersistBlock(proposal); err != nil {
-				return fmt.Errorf("persist block %s/%d: %w", h.Lane(), h.BlockNumber(), err)
+			lane := proposal.Msg().Block().Header().Lane()
+			blocksByLane[lane] = append(blocksByLane[lane], proposal)
+		}
+		if err := scope.Parallel(func(ps scope.ParallelScope) error {
+			if len(batch.commitQCs) > 0 {
+				ps.Spawn(func() error {
+					for _, qc := range batch.commitQCs {
+						if err := pers.commitQCs.PersistCommitQC(qc); err != nil {
+							return fmt.Errorf("persist commitqc %d: %w", qc.Index(), err)
+						}
+						s.markCommitQCsPersisted(qc)
+					}
+					return nil
+				})
 			}
-			s.markBlockPersisted(h.Lane(), h.BlockNumber()+1)
-		}
-
-		// 4. Prune old data.
-		if err := pers.blocks.DeleteBefore(batch.laneFirsts); err != nil {
-			return fmt.Errorf("block deleteBefore: %w", err)
-		}
-		if err := pers.commitQCs.DeleteBefore(batch.commitQCFirst); err != nil {
-			return fmt.Errorf("commitqc deleteBefore: %w", err)
+			for lane, proposals := range blocksByLane {
+				ps.Spawn(func() error {
+					for _, p := range proposals {
+						if err := pers.blocks.PersistBlock(p); err != nil {
+							h := p.Msg().Block().Header()
+							return fmt.Errorf("persist block %s/%d: %w", h.Lane(), h.BlockNumber(), err)
+						}
+						s.markBlockPersisted(lane, p.Msg().Block().Header().BlockNumber()+1)
+					}
+					return nil
+				})
+			}
+			return nil
+		}); err != nil {
+			return err
 		}
 	}
 }
 
 // persistBatch holds the data collected under lock for one persist iteration.
 type persistBatch struct {
-	blocks        []*types.Signed[*types.LaneProposal]
-	commitQCs     []*types.CommitQC
-	pruneAnchor   utils.Option[*PruneAnchor]
-	laneFirsts    map[types.LaneID]types.BlockNumber
-	commitQCFirst types.RoadIndex
+	blocks      []*types.Signed[*types.LaneProposal]
+	commitQCs   []*types.CommitQC
+	pruneAnchor utils.Option[*PruneAnchor]
 }
 
 // advancePersistedBlockStart updates the per-lane block admission watermark
@@ -717,8 +740,9 @@ func (s *State) advancePersistedBlockStart(commitQC *types.CommitQC) {
 }
 
 // markBlockPersisted advances the per-lane block persistence cursor.
-// Called after each individual block write so that RecvBatch (and therefore
-// voting) unblocks with single-block latency regardless of batch size.
+// Called after each block is persisted so that RecvBatch (and therefore
+// voting) can unblock as soon as the block is durable. Safe for concurrent
+// callers (acquires s.inner lock internally).
 func (s *State) markBlockPersisted(lane types.LaneID, next types.BlockNumber) {
 	for inner, ctrl := range s.inner.Lock() {
 		inner.nextBlockToPersist[lane] = next
@@ -759,16 +783,13 @@ func (s *State) collectPersistBatch(ctx context.Context, lastPersistedAppQCNext
 		}); err != nil {
 			return b, err
 		}
-		b.laneFirsts = make(map[types.LaneID]types.BlockNumber, len(inner.blocks))
 		for lane, q := range inner.blocks {
 			start := max(inner.nextBlockToPersist[lane], q.first)
 			for n := start; n < q.next; n++ {
 				b.blocks = append(b.blocks, q.q[n])
 			}
-			b.laneFirsts[lane] = q.first
 		}
 		commitQCNext = max(commitQCNext, inner.commitQCs.first)
-		b.commitQCFirst = inner.commitQCs.first
 		for n := commitQCNext; n < inner.commitQCs.next; n++ {
 			b.commitQCs = append(b.commitQCs, inner.commitQCs.q[n])
 		}

sei-tendermint/internal/autobahn/avail/state_test.go

@@ -291,9 +291,10 @@ func TestStateRestartFromPersisted(t *testing.T) {
 			wantAppQCIdx = appProposal.RoadIndex()
 		}
 
-		// Wait for persistence to complete. markCommitQCsPersisted fires
-		// after all blocks, commitQCs, the prune anchor, and cleanup in the
-		// batch are on disk, so this confirms all data is durable.
+		// Wait for commitQC persistence. markCommitQCsPersisted fires after
+		// all commitQCs in the batch are on disk. Block goroutines may still
+		// be in flight, but g.Wait() in runPersist ensures they complete
+		// before the next batch, so the data is durable by scope exit.
 		if err := state.waitForCommitQC(ctx, wantAppQCIdx); err != nil {
 			return fmt.Errorf("waitForCommitQC: %w", err)
 		}
@@ -582,9 +583,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 		// All 3 commitQCs should be loaded (no AppQC to skip past).
 		require.Equal(t, types.RoadIndex(0), state.FirstCommitQC())
 		// LastCommitQC should be set to the last loaded one.
-		latest, ok := state.LastCommitQC().Load().Get()
-		require.True(t, ok)
-		require.NoError(t, utils.TestDiff(qcs[2], latest))
+		require.NoError(t, utils.TestDiff(utils.Some(qcs[2]), state.LastCommitQC().Load()))
 	})
 
 	t.Run("loads persisted commitQCs with AppQC", func(t *testing.T) {
@@ -622,14 +621,11 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 		// inner.prune(appQC@1, commitQC@1) sets commitQCs.first = 1.
 		require.Equal(t, types.RoadIndex(1), state.FirstCommitQC())
-		latest, ok := state.LastCommitQC().Load().Get()
-		require.True(t, ok)
-		require.NoError(t, utils.TestDiff(qcs[4], latest))
+		require.NoError(t, utils.TestDiff(utils.Some(qcs[4]), state.LastCommitQC().Load()))
 	})
 
 	t.Run("non-contiguous commitQC files return error", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := data.NewState(&data.Config{Committee: committee}, utils.None[data.BlockStore]())
 
 		// Build 6 sequential CommitQCs (indices 0-5).
 		allQCs := make([]*types.CommitQC, 6)
@@ -649,19 +645,109 @@ func TestNewStateWithPersistence(t *testing.T) {
 			CommitQc: types.CommitQCConv.Encode(allQCs[0]),
 		}))
 
-		// Persist QCs 0, 1, 2 contiguously, then skip to 5 (simulating
-		// corruption or manual tampering). Since the anchor is persisted
-		// first, a gap should never occur normally — treat it as an error.
+		// Persist QCs 0, 1, 2 contiguously, then try to skip to 5.
+		// PersistCommitQC enforces strict sequential order, so the gap
+		// is caught at write time rather than at load time.
 		cp, _, err := persist.NewCommitQCPersister(utils.Some(dir))
 		require.NoError(t, err)
-		for i := 0; i < 3; i++ {
+		for i := range 3 {
 			require.NoError(t, cp.PersistCommitQC(allQCs[i]))
 		}
-		require.NoError(t, cp.PersistCommitQC(allQCs[5]))
-
-		_, err = NewState(keys[0], ds, utils.Some(dir))
+		err = cp.PersistCommitQC(allQCs[5])
 		require.Error(t, err)
-		require.Contains(t, err.Error(), "non-contiguous")
+		require.Contains(t, err.Error(), "out of sequence")
+		require.NoError(t, cp.Close())
+	})
+
+	t.Run("anchor past all persisted commitQCs truncates WAL", func(t *testing.T) {
+		dir := t.TempDir()
+		ds := data.NewState(&data.Config{Committee: committee}, utils.None[data.BlockStore]())
+
+		// Build a chain of 10 CommitQCs (indices 0-9).
+		qcs := make([]*types.CommitQC, 10)
+		prev := utils.None[*types.CommitQC]()
+		for i := range qcs {
+			qcs[i] = makeCommitQC(rng, committee, keys, prev, nil, utils.None[*types.AppQC]())
+			prev = utils.Some(qcs[i])
+		}
+
+		// Persist only indices 0-4 to the CommitQC WAL.
+		cp, _, err := persist.NewCommitQCPersister(utils.Some(dir))
+		require.NoError(t, err)
+		for i := range 5 {
+			require.NoError(t, cp.PersistCommitQC(qcs[i]))
+		}
+		require.NoError(t, cp.Close())
+
+		// Persist a prune anchor at index 9 — well past the persisted range.
+		appProposal := types.NewAppProposal(50, 9, types.GenAppHash(rng))
+		appQC := types.NewAppQC(makeAppVotes(keys, appProposal))
+		prunePers, _, err := persist.NewPersister[*pb.PersistedAvailPruneAnchor](utils.Some(dir), innerFile)
+		require.NoError(t, err)
+		require.NoError(t, prunePers.Persist(&pb.PersistedAvailPruneAnchor{
+			AppQc:    types.AppQCConv.Encode(appQC),
+			CommitQc: types.CommitQCConv.Encode(qcs[9]),
+		}))
+
+		// NewState should succeed: DeleteBefore truncates the stale WAL,
+		// then the re-persist loop writes the anchor's CommitQC back.
+		state, err := NewState(keys[0], ds, utils.Some(dir))
+		require.NoError(t, err)
+
+		require.Equal(t, types.RoadIndex(9), state.FirstCommitQC())
+		require.NoError(t, utils.TestDiff(utils.Some(qcs[9]), state.LastCommitQC().Load()))
+
+		got, ok := state.LastAppQC().Get()
+		require.True(t, ok)
+		require.Equal(t, types.RoadIndex(9), got.Proposal().RoadIndex())
+	})
+
+	t.Run("anchor past all persisted blocks truncates lane WAL", func(t *testing.T) {
+		dir := t.TempDir()
+		ds := data.NewState(&data.Config{Committee: committee}, utils.None[data.BlockStore]())
+		lane := keys[0].Public()
+
+		// Persist commitQCs 0-9 and blocks 0-2 for one lane.
+		qcs := make([]*types.CommitQC, 10)
+		prev := utils.None[*types.CommitQC]()
+		cp, _, err := persist.NewCommitQCPersister(utils.Some(dir))
+		require.NoError(t, err)
+		for i := range qcs {
+			qcs[i] = makeCommitQC(rng, committee, keys, prev, nil, utils.None[*types.AppQC]())
+			prev = utils.Some(qcs[i])
+			require.NoError(t, cp.PersistCommitQC(qcs[i]))
+		}
+		require.NoError(t, cp.Close())
+
+		bp, _, err := persist.NewBlockPersister(utils.Some(dir))
+		require.NoError(t, err)
+		var parent types.BlockHeaderHash
+		for n := types.BlockNumber(0); n < 3; n++ {
+			block := types.NewBlock(lane, n, parent, types.GenPayload(rng))
+			signed := types.Sign(keys[0], types.NewLaneProposal(block))
+			parent = block.Header().Hash()
+			require.NoError(t, bp.PersistBlock(signed))
+		}
+
+		// Persist a prune anchor at index 9 with a laneRange that starts past
+		// all persisted blocks — DeleteBefore will TruncateAll the block WAL.
+		appProposal := types.NewAppProposal(50, 9, types.GenAppHash(rng))
+		appQC := types.NewAppQC(makeAppVotes(keys, appProposal))
+		prunePers, _, err := persist.NewPersister[*pb.PersistedAvailPruneAnchor](utils.Some(dir), innerFile)
+		require.NoError(t, err)
+		require.NoError(t, prunePers.Persist(&pb.PersistedAvailPruneAnchor{
+			AppQc:    types.AppQCConv.Encode(appQC),
+			CommitQc: types.CommitQCConv.Encode(qcs[9]),
+		}))
+
+		// NewState should succeed: block WAL gets truncated, lane starts clean.
+		state, err := NewState(keys[0], ds, utils.Some(dir))
+		require.NoError(t, err)
+
+		require.Equal(t, types.RoadIndex(9), state.FirstCommitQC())
+		got, ok := state.LastAppQC().Get()
+		require.True(t, ok)
+		require.Equal(t, types.RoadIndex(9), got.Proposal().RoadIndex())
 	})
 
 	t.Run("corrupt AppQC data returns error", func(t *testing.T) {