feat(data): integrate LittDB-backed BlockDB into autobahn data layer (CON-272)

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

@@ -1,16 +1,25 @@
 package avail
 
 import (
-	pb "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pb"
 	"testing"
 
+	pb "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pb"
+
 	"github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/consensus/persist"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/data"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
 	"github.com/stretchr/testify/require"
 )
 
+func newTestDataState(t testing.TB, cfg *data.Config) *data.State {
+	s, err := data.NewState(cfg, utils.None[types.BlockDB]())
+	if err != nil {
+		t.Fatalf("data.NewState: %v", err)
+	}
+	return s
+}
+
 func TestPruneMismatchedIndices(t *testing.T) {
 	rng := utils.TestRng()
 	committee, keys := types.GenCommittee(rng, 4)
@@ -35,17 +44,17 @@ func TestPruneMismatchedIndices(t *testing.T) {
 	qc1 := makeCommitQC(utils.Some(qc0))
 
 	t.Logf("test State.PushAppQC")
-	ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
-	state, err := NewState(keys[0], ds, utils.None[string]())
+	ds := newTestDataState(t, &data.Config{Committee: committee})
+	state, err := NewState(keys[0], ds, utils.Some(t.TempDir()))
 	require.NoError(t, err)
 	require.Error(t, state.PushAppQC(makeAppQC(qc0, qc0), qc1), "bad range, bad index should fail")
 	require.Error(t, state.PushAppQC(makeAppQC(qc1, qc0), qc1), "good range, bad index should fail")
 	require.Error(t, state.PushAppQC(makeAppQC(qc0, qc1), qc1), "bad range, good index should fail")
 	require.NoError(t, state.PushAppQC(makeAppQC(qc1, qc1), qc1), "good range, good index should succeed")
 
 	t.Logf("test inner.prune")
-	ds = utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
-	state, err = NewState(keys[0], ds, utils.None[string]())
+	ds = newTestDataState(t, &data.Config{Committee: committee})
+	state, err = NewState(keys[0], ds, utils.Some(t.TempDir()))
 	require.NoError(t, err)
 	for inner := range state.inner.Lock() {
 		_, err := inner.prune(committee, makeAppQC(qc1, qc0), qc1)

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

@@ -271,9 +271,8 @@ func (s *State) PushCommitQC(ctx context.Context, qc *types.CommitQC) error {
 		}
 		inner.commitQCs.pushBack(qc)
 		metrics.ObserveCommitQC(committee, qc)
-		// The persist goroutine publishes latestCommitQC after writing to disk
-		// (or immediately for no-op persisters), so consensus won't advance
-		// until the CommitQC is durable.
+		// The persist goroutine publishes latestCommitQC after writing to disk,
+		// so consensus won't advance until the CommitQC is durable.
 		ctrl.Updated()
 		return nil
 	}

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

@@ -105,9 +105,7 @@ func testState(t *testing.T, stateDir utils.Option[string]) {
 	committee, keys := types.GenCommittee(rng, 3)
 
 	if err := scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
-		ds := utils.OrPanic1(data.NewState(&data.Config{
-			Committee: committee,
-		}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 		s.SpawnBgNamed("data.State.Run()", func() error {
 			return utils.IgnoreCancel(ds.Run(ctx))
 		})
@@ -236,7 +234,7 @@ func TestStateRestartFromPersisted(t *testing.T) {
 	var wantNextBlocks map[types.LaneID]types.BlockNumber
 
 	require.NoError(t, scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error {
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 		s.SpawnBgNamed("data.Run", func() error {
 			return utils.IgnoreCancel(ds.Run(ctx))
 		})
@@ -311,7 +309,7 @@ func TestStateRestartFromPersisted(t *testing.T) {
 	}))
 
 	// Phase 2: Restart from the same directory.
-	ds2 := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+	ds2 := newTestDataState(t, &data.Config{Committee: committee})
 	state2, err := NewState(keys[0], ds2, utils.Some(dir))
 	require.NoError(t, err)
 
@@ -335,10 +333,8 @@ func TestStateMismatchedQCs(t *testing.T) {
 	committee, keys := types.GenCommittee(rng, 4)
 	initialBlock := committee.FirstBlock()
 
-	ds := utils.OrPanic1(data.NewState(&data.Config{
-		Committee: committee,
-	}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
-	state, err := NewState(keys[0], ds, utils.None[string]())
+	ds := newTestDataState(t, &data.Config{Committee: committee})
+	state, err := NewState(keys[0], ds, utils.Some(t.TempDir()))
 	require.NoError(t, err)
 
 	// Helper to create a CommitQC for a specific index
@@ -393,10 +389,8 @@ func TestPushBlockRejectsBadParentHash(t *testing.T) {
 	rng := utils.TestRng()
 	committee, keys := types.GenCommittee(rng, 3)
 
-	ds := utils.OrPanic1(data.NewState(&data.Config{
-		Committee: committee,
-	}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
-	state := utils.OrPanic1(NewState(keys[0], ds, utils.None[string]()))
+	ds := newTestDataState(t, &data.Config{Committee: committee})
+	state := utils.OrPanic1(NewState(keys[0], ds, utils.Some(t.TempDir())))
 
 	// Produce a valid first block on our lane.
 	_, err := state.ProduceLocalBlock(state.NextBlock(keys[0].Public()), types.GenPayload(rng))
@@ -418,10 +412,8 @@ func TestPushBlockRejectsWrongSigner(t *testing.T) {
 	rng := utils.TestRng()
 	committee, keys := types.GenCommittee(rng, 3)
 
-	ds := utils.OrPanic1(data.NewState(&data.Config{
-		Committee: committee,
-	}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
-	state := utils.OrPanic1(NewState(keys[0], ds, utils.None[string]()))
+	ds := newTestDataState(t, &data.Config{Committee: committee})
+	state := utils.OrPanic1(NewState(keys[0], ds, utils.Some(t.TempDir())))
 
 	// Create a block on keys[0]'s lane but sign it with keys[1].
 	lane := keys[0].Public()
@@ -439,7 +431,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 	t.Run("empty dir loads fresh state", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 
 		state, err := NewState(keys[0], ds, utils.Some(dir))
 		require.NoError(t, err)
@@ -452,7 +444,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 	t.Run("loads persisted AppQC", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 
 		roadIdx := types.RoadIndex(7)
 		globalNum := types.GlobalBlockNumber(50)
@@ -493,7 +485,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 	t.Run("loads persisted blocks", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 		lane := keys[0].Public()
 
 		// Persist blocks using BlockPersister.
@@ -517,7 +509,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 	t.Run("loads persisted AppQC and blocks together", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 		lane := keys[0].Public()
 
 		roadIdx := types.RoadIndex(2)
@@ -570,7 +562,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 	t.Run("loads persisted commitQCs", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 
 		// Persist CommitQCs to disk.
 		cp, _, err := persist.NewCommitQCPersister(utils.Some(dir))
@@ -595,7 +587,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 	t.Run("loads persisted commitQCs with AppQC", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 
 		// Persist AppQC at road index 1.
 		roadIdx := types.RoadIndex(1)
@@ -668,7 +660,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 	t.Run("anchor past all persisted commitQCs truncates WAL", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 
 		// Build a chain of 10 CommitQCs (indices 0-9).
 		qcs := make([]*types.CommitQC, 10)
@@ -711,7 +703,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 	t.Run("anchor past all persisted blocks truncates lane WAL", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 		lane := keys[0].Public()
 
 		// Persist commitQCs 0-9 and blocks 0-2 for one lane.
@@ -759,7 +751,7 @@ func TestNewStateWithPersistence(t *testing.T) {
 
 	t.Run("corrupt AppQC data returns error", func(t *testing.T) {
 		dir := t.TempDir()
-		ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+		ds := newTestDataState(t, &data.Config{Committee: committee})
 
 		// Create a throwaway persister to discover the A/B filenames,
 		// then corrupt them so NewState fails on load.

sei-tendermint/internal/autobahn/consensus/inner_test.go

@@ -3,9 +3,12 @@ package consensus
 import (
 	"context"
 	"errors"
+	"fmt"
+	"path/filepath"
 	"testing"
 	"time"
 
+	"github.com/sei-protocol/sei-chain/sei-db/ledger_db/block/littblock"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/consensus/persist"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/data"
@@ -15,6 +18,28 @@ import (
 	"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require"
 )
 
+func newTestBlockDB(t *testing.T, dir string) types.BlockDB {
+	t.Helper()
+	cfg, err := littblock.DefaultConfig(dir)
+	if err != nil {
+		t.Fatalf("littblock.DefaultConfig: %v", err)
+	}
+	db, err := littblock.NewBlockDB(cfg)
+	if err != nil {
+		t.Fatalf("littblock.NewBlockDB: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	return db
+}
+
+func newTestDataState(committee *types.Committee) *data.State {
+	s, err := data.NewState(&data.Config{Committee: committee}, utils.None[types.BlockDB]())
+	if err != nil {
+		panic(fmt.Sprintf("data.NewState: %v", err))
+	}
+	return s
+}
+
 // seedPersistedInner is a test helper that persists a persistedInner using the public API.
 func seedPersistedInner(dir string, state *persistedInner) {
 	p, _, err := persist.NewPersister[*pb.PersistedInner](utils.Some(dir), innerFile)
@@ -923,15 +948,21 @@ func (f failPersister[T]) Persist(T) error { return f.err }
 func TestRunOutputsPersistErrorPropagates(t *testing.T) {
 	// Verify that a persist error in runOutputs propagates
 	// and terminates the consensus component (instead of panicking).
+	dir := t.TempDir()
 	rng := utils.TestRng()
 	committee, keys := types.GenCommittee(rng, 4)
-	ds := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee))))
+	db := newTestBlockDB(t, filepath.Join(dir, "blockdb"))
+	ds, err := data.NewState(&data.Config{Committee: committee}, utils.Some[types.BlockDB](db))
+	if err != nil {
+		t.Fatalf("data.NewState: %v", err)
+	}
 
 	wantErr := errors.New("disk on fire")
 	pers := utils.Some[persist.Persister[*pb.PersistedInner]](failPersister[*pb.PersistedInner]{err: wantErr})
 	cs, err := newState(&Config{
-		Key:         keys[0],
-		ViewTimeout: func(types.View) time.Duration { return time.Hour },
+		Key:                keys[0],
+		ViewTimeout:        func(types.View) time.Duration { return time.Hour },
+		PersistentStateDir: utils.Some(dir),
 	}, ds, pers, utils.None[*pb.PersistedInner]())
 	require.NoError(t, err)