diff --git a/sei-tendermint/internal/autobahn/avail/inner_test.go b/sei-tendermint/internal/autobahn/avail/inner_test.go index 257641de63..9b52cc6e88 100644 --- a/sei-tendermint/internal/autobahn/avail/inner_test.go +++ b/sei-tendermint/internal/autobahn/avail/inner_test.go @@ -1,9 +1,10 @@ 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" @@ -11,6 +12,14 @@ import ( "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,8 +44,8 @@ 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") @@ -44,8 +53,8 @@ func TestPruneMismatchedIndices(t *testing.T) { 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) diff --git a/sei-tendermint/internal/autobahn/avail/state.go b/sei-tendermint/internal/autobahn/avail/state.go index d6e49ebb18..c751a13eaa 100644 --- a/sei-tendermint/internal/autobahn/avail/state.go +++ b/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 } diff --git a/sei-tendermint/internal/autobahn/avail/state_test.go b/sei-tendermint/internal/autobahn/avail/state_test.go index 4bb2d3e2fa..cfe880d21a 100644 --- a/sei-tendermint/internal/autobahn/avail/state_test.go +++ b/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. diff --git a/sei-tendermint/internal/autobahn/consensus/inner_test.go b/sei-tendermint/internal/autobahn/consensus/inner_test.go index 761e528d36..dec517c1c4 100644 --- a/sei-tendermint/internal/autobahn/consensus/inner_test.go +++ b/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) diff --git a/sei-tendermint/internal/autobahn/consensus/persist/fullcommitqcs.go b/sei-tendermint/internal/autobahn/consensus/persist/fullcommitqcs.go deleted file mode 100644 index f4ff1df34e..0000000000 --- a/sei-tendermint/internal/autobahn/consensus/persist/fullcommitqcs.go +++ /dev/null @@ -1,166 +0,0 @@ -package persist - -import ( - "fmt" - "path/filepath" - - "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" -) - -const fullCommitQCsDir = "fullcommitqcs" - -// fullCommitQCState is the mutable state protected by FullCommitQCPersister's mutex. -type fullCommitQCState struct { - iw utils.Option[*indexedWAL[*types.FullCommitQC]] - committee *types.Committee - next types.GlobalBlockNumber // next expected GlobalRange().First == last QC's GlobalRange().Next - loaded []*types.FullCommitQC -} - -func (s *fullCommitQCState) persistQC(qc *types.FullCommitQC) error { - gr := qc.QC().GlobalRange(s.committee) - if gr.First < s.next { - return nil - } - if gr.First > s.next { - return fmt.Errorf("full commitqc %d out of sequence (next=%d)", gr.First, s.next) - } - if iw, ok := s.iw.Get(); ok { - if err := iw.Write(qc); err != nil { - return fmt.Errorf("persist full commitqc at %d: %w", gr.First, err) - } - } - s.next = gr.Next - return nil -} - -func (s *fullCommitQCState) truncateBefore(n types.GlobalBlockNumber) error { - if n == 0 { - return nil - } - if n > s.next { - s.next = n - } - iw, ok := s.iw.Get() - if !ok || iw.Count() == 0 { - return nil - } - // Remove QCs whose range is fully before n. A QC is stale when - // GlobalRange().Next <= n. In practice the scan visits 0–1 entries - // per prune call because pruning advances one block at a time while - // each QC covers many blocks. - if err := iw.TruncateWhile(func(entry *types.FullCommitQC) bool { - return entry.QC().GlobalRange(s.committee).Next <= n - }); err != nil { - return fmt.Errorf("truncate full commitqc WAL: %w", err) - } - return nil -} - -// FullCommitQCPersister manages persistence of FullCommitQCs using a WAL. -// Each entry is one FullCommitQC covering a range of global block numbers. -// When stateDir is None, all disk I/O is skipped (no-op mode). -// All public methods are safe for concurrent use. -type FullCommitQCPersister struct { - state utils.Mutex[*fullCommitQCState] -} - -// NewFullCommitQCPersister opens (or creates) a WAL in the fullcommitqcs/ -// subdir and replays all persisted entries. Loaded QCs are available via -// ConsumeLoaded. When stateDir is None, returns a no-op persister. -func NewFullCommitQCPersister(stateDir utils.Option[string], committee *types.Committee) (*FullCommitQCPersister, error) { - sd, ok := stateDir.Get() - if !ok { - return &FullCommitQCPersister{state: utils.NewMutex(&fullCommitQCState{committee: committee, next: committee.FirstBlock()})}, nil - } - dir := filepath.Join(sd, fullCommitQCsDir) - iw, err := openIndexedWAL(dir, types.FullCommitQCConv) - if err != nil { - return nil, fmt.Errorf("open full commitqc WAL in %s: %w", dir, err) - } - - s := &fullCommitQCState{iw: utils.Some(iw), committee: committee, next: committee.FirstBlock()} - loaded, err := s.loadAll() - if err != nil { - _ = iw.Close() - return nil, err - } - if len(loaded) > 0 { - s.next = loaded[len(loaded)-1].QC().GlobalRange(s.committee).Next - } - s.loaded = loaded - return &FullCommitQCPersister{ - state: utils.NewMutex(s), - }, nil -} - -// Next returns the next GlobalBlockNumber expected by the persister -// (i.e., the next QC's GlobalRange().First). -func (gp *FullCommitQCPersister) Next() types.GlobalBlockNumber { - for s := range gp.state.Lock() { - return s.next - } - panic("unreachable") -} - -// LoadedFirst returns the first global block number of the first loaded QC, -// or committee.FirstBlock() if empty. -func (gp *FullCommitQCPersister) LoadedFirst() types.GlobalBlockNumber { - for s := range gp.state.Lock() { - if len(s.loaded) > 0 { - return s.loaded[0].QC().GlobalRange(s.committee).First - } - return s.committee.FirstBlock() - } - panic("unreachable") -} - -// ConsumeLoaded returns QCs loaded from the WAL during construction -// and nils the internal slice so the data is not retained. -func (gp *FullCommitQCPersister) ConsumeLoaded() []*types.FullCommitQC { - for s := range gp.state.Lock() { - loaded := s.loaded - s.loaded = nil - return loaded - } - panic("unreachable") -} - -// PersistQC appends a FullCommitQC to the WAL. Duplicates are silently ignored. -// Gaps return an error. -func (gp *FullCommitQCPersister) PersistQC(qc *types.FullCommitQC) error { - for s := range gp.state.Lock() { - return s.persistQC(qc) - } - panic("unreachable") -} - -// TruncateBefore removes all QC entries whose range is fully before n. -func (gp *FullCommitQCPersister) TruncateBefore(n types.GlobalBlockNumber) error { - for s := range gp.state.Lock() { - return s.truncateBefore(n) - } - panic("unreachable") -} - -// Close shuts down the WAL. -func (gp *FullCommitQCPersister) Close() error { - for s := range gp.state.Lock() { - iw, ok := s.iw.Get() - if !ok { - return nil - } - s.iw = utils.None[*indexedWAL[*types.FullCommitQC]]() - return iw.Close() - } - panic("unreachable") -} - -func (s *fullCommitQCState) loadAll() ([]*types.FullCommitQC, error) { - iw, ok := s.iw.Get() - if !ok { - return nil, nil - } - return iw.ReadAll() -} diff --git a/sei-tendermint/internal/autobahn/consensus/persist/fullcommitqcs_test.go b/sei-tendermint/internal/autobahn/consensus/persist/fullcommitqcs_test.go deleted file mode 100644 index 8932bcab76..0000000000 --- a/sei-tendermint/internal/autobahn/consensus/persist/fullcommitqcs_test.go +++ /dev/null @@ -1,293 +0,0 @@ -package persist - -import ( - "testing" - "time" - - "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" -) - -func makeSequentialFullCommitQCs( - rng utils.Rng, - committee *types.Committee, - keys []types.SecretKey, - n int, -) []*types.FullCommitQC { - qcs := make([]*types.FullCommitQC, n) - prev := utils.None[*types.CommitQC]() - for i := range n { - blocks := map[types.LaneID][]*types.Block{} - for range 3 { - lane := committee.Lanes().At(rng.Intn(committee.Lanes().Len())) - var b *types.Block - if bs := blocks[lane]; len(bs) > 0 { - parent := bs[len(bs)-1] - b = types.NewBlock(lane, parent.Header().Next(), parent.Header().Hash(), types.GenPayload(rng)) - } else { - b = types.NewBlock( - lane, - types.LaneRangeOpt(prev, lane).Next(), - types.GenBlockHeaderHash(rng), - types.GenPayload(rng), - ) - } - blocks[lane] = append(blocks[lane], b) - } - laneQCs := map[types.LaneID]*types.LaneQC{} - var headers []*types.BlockHeader - for lane := range committee.Lanes().All() { - if bs := blocks[lane]; len(bs) > 0 { - votes := make([]*types.Signed[*types.LaneVote], 0, len(keys)) - for _, k := range keys { - votes = append(votes, types.Sign(k, types.NewLaneVote(bs[len(bs)-1].Header()))) - } - laneQCs[lane] = types.NewLaneQC(votes) - for _, b := range bs { - headers = append(headers, b.Header()) - } - } - } - viewSpec := types.ViewSpec{CommitQC: prev} - leader := committee.Leader(viewSpec.View()) - var leaderKey types.SecretKey - for _, k := range keys { - if k.Public() == leader { - leaderKey = k - break - } - } - proposal := utils.OrPanic1(types.NewProposal( - leaderKey, - committee, - viewSpec, - time.Now(), - laneQCs, - utils.None[*types.AppQC](), - )) - votes := make([]*types.Signed[*types.CommitVote], 0, len(keys)) - for _, k := range keys { - votes = append(votes, types.Sign(k, types.NewCommitVote(proposal.Proposal().Msg()))) - } - cqc := types.NewCommitQC(votes) - qcs[i] = types.NewFullCommitQC(cqc, headers) - prev = utils.Some(cqc) - } - return qcs -} - -func TestNewFullCommitQCPersisterEmptyDir(t *testing.T) { - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - dir := t.TempDir() - gp, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.NotNil(t, gp) - require.Equal(t, 0, len(gp.ConsumeLoaded())) - require.Equal(t, committee.FirstBlock(), gp.Next()) - require.NoError(t, gp.Close()) -} - -func TestNewFullCommitQCPersisterNoop(t *testing.T) { - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - qcs := makeSequentialFullCommitQCs(rng, committee, keys, 5) - - gp, err := NewFullCommitQCPersister(utils.None[string](), committee) - require.NoError(t, err) - require.NotNil(t, gp) - require.Equal(t, 0, len(gp.ConsumeLoaded())) - - for _, qc := range qcs { - require.NoError(t, gp.PersistQC(qc)) - } - lastNext := qcs[len(qcs)-1].QC().GlobalRange(committee).Next - require.Equal(t, lastNext, gp.Next()) - - // Truncate past everything in no-op mode advances cursor. - futureN := lastNext + 100 - require.NoError(t, gp.TruncateBefore(futureN)) - require.Equal(t, futureN, gp.Next()) - require.NoError(t, gp.Close()) -} - -func TestFullCommitQCPersistAndReload(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - qcs := makeSequentialFullCommitQCs(rng, committee, keys, 5) - - gp, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - for _, qc := range qcs { - require.NoError(t, gp.PersistQC(qc)) - } - lastNext := qcs[len(qcs)-1].QC().GlobalRange(committee).Next - require.Equal(t, lastNext, gp.Next()) - require.NoError(t, gp.Close()) - - gp2, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - loaded := gp2.ConsumeLoaded() - require.Equal(t, len(qcs), len(loaded)) - for i, lqc := range loaded { - require.Equal(t, qcs[i].QC().GlobalRange(committee).First, lqc.QC().GlobalRange(committee).First) - } - require.Equal(t, lastNext, gp2.Next()) - require.NoError(t, gp2.Close()) -} - -func TestFullCommitQCTruncateAndReload(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - qcs := makeSequentialFullCommitQCs(rng, committee, keys, 5) - - gp, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - for _, qc := range qcs { - require.NoError(t, gp.PersistQC(qc)) - } - // Truncate before the third QC's range start, which should remove - // all QCs whose range is fully below that point. - truncPoint := qcs[2].QC().GlobalRange(committee).First - require.NoError(t, gp.TruncateBefore(truncPoint)) - require.NoError(t, gp.Close()) - - gp2, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - loaded := gp2.ConsumeLoaded() - // QCs 0 and 1 should be gone (their ranges are fully before truncPoint). - // QC 2 should be the first one remaining. - require.GreaterOrEqual(t, len(loaded), 1) - require.Equal(t, qcs[2].QC().GlobalRange(committee).First, loaded[0].QC().GlobalRange(committee).First) - require.NoError(t, gp2.Close()) -} - -func TestFullCommitQCTruncateAll(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - qcs := makeSequentialFullCommitQCs(rng, committee, keys, 3) - - gp, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - for _, qc := range qcs { - require.NoError(t, gp.PersistQC(qc)) - } - lastNext := qcs[len(qcs)-1].QC().GlobalRange(committee).Next - require.NoError(t, gp.TruncateBefore(lastNext+100)) - require.NoError(t, gp.Close()) - - gp2, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 0, len(gp2.ConsumeLoaded())) - require.NoError(t, gp2.Close()) -} - -func TestFullCommitQCDuplicateIgnored(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - qcs := makeSequentialFullCommitQCs(rng, committee, keys, 2) - - gp, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.NoError(t, gp.PersistQC(qcs[0])) - require.NoError(t, gp.PersistQC(qcs[0])) // duplicate - require.Equal(t, qcs[0].QC().GlobalRange(committee).Next, gp.Next()) - require.NoError(t, gp.Close()) -} - -func TestFullCommitQCGapError(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - qcs := makeSequentialFullCommitQCs(rng, committee, keys, 3) - - gp, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - // Skip qcs[0] and try to persist qcs[1] directly. - err = gp.PersistQC(qcs[1]) - require.Error(t, err) - require.Contains(t, err.Error(), "out of sequence") - require.NoError(t, gp.Close()) -} - -func TestFullCommitQCTruncateBeforeNoop(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - qcs := makeSequentialFullCommitQCs(rng, committee, keys, 3) - - gp, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - for _, qc := range qcs { - require.NoError(t, gp.PersistQC(qc)) - } - // TruncateBefore(0) is a no-op. - require.NoError(t, gp.TruncateBefore(0)) - require.NoError(t, gp.Close()) - - gp2, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 3, len(gp2.ConsumeLoaded())) - require.NoError(t, gp2.Close()) -} - -func TestFullCommitQCContinueAfterReload(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - qcs := makeSequentialFullCommitQCs(rng, committee, keys, 6) - - gp, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - for _, qc := range qcs[:3] { - require.NoError(t, gp.PersistQC(qc)) - } - require.NoError(t, gp.Close()) - - gp2, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 3, len(gp2.ConsumeLoaded())) - for _, qc := range qcs[3:] { - require.NoError(t, gp2.PersistQC(qc)) - } - require.Equal(t, qcs[5].QC().GlobalRange(committee).Next, gp2.Next()) - require.NoError(t, gp2.Close()) - - gp3, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 6, len(gp3.ConsumeLoaded())) - require.NoError(t, gp3.Close()) -} - -func TestFullCommitQCTruncateMidRange(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - qcs := makeSequentialFullCommitQCs(rng, committee, keys, 5) - - gp, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - for _, qc := range qcs { - require.NoError(t, gp.PersistQC(qc)) - } - // Truncate at a point inside the first QC's range. - // The first QC should be kept because its range extends past truncPoint. - gr0 := qcs[0].QC().GlobalRange(committee) - if gr0.Len() > 1 { - midPoint := gr0.First + types.GlobalBlockNumber(gr0.Len()/2) - require.NoError(t, gp.TruncateBefore(midPoint)) - require.NoError(t, gp.Close()) - - gp2, err := NewFullCommitQCPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 5, len(gp2.ConsumeLoaded())) - require.NoError(t, gp2.Close()) - } else { - require.NoError(t, gp.Close()) - } -} diff --git a/sei-tendermint/internal/autobahn/consensus/persist/globalblocks.go b/sei-tendermint/internal/autobahn/consensus/persist/globalblocks.go deleted file mode 100644 index 9471994e3c..0000000000 --- a/sei-tendermint/internal/autobahn/consensus/persist/globalblocks.go +++ /dev/null @@ -1,231 +0,0 @@ -package persist - -import ( - "encoding/binary" - "fmt" - "path/filepath" - - "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" -) - -const globalBlocksDir = "globalblocks" - -// LoadedGlobalBlock is a block loaded from disk during state restoration. -// Also used as the internal WAL entry type. Block doesn't carry its -// GlobalBlockNumber (that's assigned by the ordering layer), so we embed -// it in each WAL entry to make entries self-describing. -type LoadedGlobalBlock struct { - Number types.GlobalBlockNumber - Block *types.Block -} - -// loadedGlobalBlockCodec serializes LoadedGlobalBlock as [8-byte LE number][proto block]. -type loadedGlobalBlockCodec struct{} - -func (loadedGlobalBlockCodec) Marshal(e LoadedGlobalBlock) []byte { - proto := types.BlockConv.Marshal(e.Block) - out := make([]byte, 8+len(proto)) - binary.LittleEndian.PutUint64(out, uint64(e.Number)) - copy(out[8:], proto) - return out -} - -func (loadedGlobalBlockCodec) Unmarshal(raw []byte) (LoadedGlobalBlock, error) { - if len(raw) < 8 { - return LoadedGlobalBlock{}, fmt.Errorf("global block entry too short: %d bytes", len(raw)) - } - n := types.GlobalBlockNumber(binary.LittleEndian.Uint64(raw[:8])) - block, err := types.BlockConv.Unmarshal(raw[8:]) - if err != nil { - return LoadedGlobalBlock{}, fmt.Errorf("unmarshal block %d: %w", n, err) - } - return LoadedGlobalBlock{Number: n, Block: block}, nil -} - -// globalBlockState is the mutable state protected by GlobalBlockPersister's mutex. -type globalBlockState struct { - iw utils.Option[*indexedWAL[LoadedGlobalBlock]] - committee *types.Committee - next types.GlobalBlockNumber - loaded []LoadedGlobalBlock -} - -func (s *globalBlockState) persistBlock(n types.GlobalBlockNumber, block *types.Block) error { - if n < s.next { - return nil - } - if n > s.next { - return fmt.Errorf("global block %d out of sequence (next=%d)", n, s.next) - } - if iw, ok := s.iw.Get(); ok { - if err := iw.Write(LoadedGlobalBlock{Number: n, Block: block}); err != nil { - return fmt.Errorf("persist global block %d: %w", n, err) - } - } - s.next = n + 1 - return nil -} - -func (s *globalBlockState) truncateBefore(n types.GlobalBlockNumber) error { - if n == 0 { - return nil - } - if n > s.next { - s.next = n - } - iw, ok := s.iw.Get() - if !ok || iw.Count() == 0 { - return nil - } - firstGlobal := s.next - types.GlobalBlockNumber(iw.Count()) - if n <= firstGlobal { - return nil - } - walIdx := iw.FirstIdx() + uint64(n-firstGlobal) - if err := iw.TruncateBefore(walIdx, func(entry LoadedGlobalBlock) error { - if entry.Number != n { - return fmt.Errorf("global block at WAL index %d has number %d, expected %d (index mapping broken)", walIdx, entry.Number, n) - } - return nil - }); err != nil { - return fmt.Errorf("truncate global block WAL before %d: %w", n, err) - } - return nil -} - -// GlobalBlockPersister manages persistence of globally-ordered blocks using a WAL. -// Each entry embeds its GlobalBlockNumber since Block doesn't carry it. -// When stateDir is None, all disk I/O is skipped (no-op mode). -// All public methods are safe for concurrent use. -type GlobalBlockPersister struct { - state utils.Mutex[*globalBlockState] -} - -// NewGlobalBlockPersister opens (or creates) a WAL in the globalblocks/ subdir -// and replays all persisted entries. Loaded blocks are available via -// ConsumeLoaded. When stateDir is None, returns a no-op persister. -func NewGlobalBlockPersister(stateDir utils.Option[string], committee *types.Committee) (*GlobalBlockPersister, error) { - sd, ok := stateDir.Get() - if !ok { - return &GlobalBlockPersister{state: utils.NewMutex(&globalBlockState{committee: committee, next: committee.FirstBlock()})}, nil - } - dir := filepath.Join(sd, globalBlocksDir) - iw, err := openIndexedWAL(dir, loadedGlobalBlockCodec{}) - if err != nil { - return nil, fmt.Errorf("open global block WAL in %s: %w", dir, err) - } - - s := &globalBlockState{iw: utils.Some(iw), committee: committee, next: committee.FirstBlock()} - // TODO: avoid loading all blocks on startup; cache only the last N blocks - // (e.g. 1000) in memory instead. - loaded, err := s.loadAll() - if err != nil { - _ = iw.Close() - return nil, err - } - if len(loaded) > 0 { - s.next = loaded[len(loaded)-1].Number + 1 - } - s.loaded = loaded - return &GlobalBlockPersister{ - state: utils.NewMutex(s), - }, nil -} - -// Next returns the next GlobalBlockNumber expected by the persister. -func (gp *GlobalBlockPersister) Next() types.GlobalBlockNumber { - for s := range gp.state.Lock() { - return s.next - } - panic("unreachable") -} - -// LoadedFirst returns the first loaded block number, or committee.FirstBlock() if empty. -func (gp *GlobalBlockPersister) LoadedFirst() types.GlobalBlockNumber { - for s := range gp.state.Lock() { - if len(s.loaded) > 0 { - return s.loaded[0].Number - } - return s.committee.FirstBlock() - } - panic("unreachable") -} - -// ConsumeLoaded returns blocks loaded from the WAL during construction -// and nils the internal slice so the data is not retained. -func (gp *GlobalBlockPersister) ConsumeLoaded() []LoadedGlobalBlock { - for s := range gp.state.Lock() { - loaded := s.loaded - s.loaded = nil - return loaded - } - panic("unreachable") -} - -// PersistBlock appends a block to the WAL. Duplicates are silently ignored. -// Gaps return an error. -func (gp *GlobalBlockPersister) PersistBlock(n types.GlobalBlockNumber, block *types.Block) error { - for s := range gp.state.Lock() { - return s.persistBlock(n, block) - } - panic("unreachable") -} - -// TruncateBefore removes all entries before n from the WAL. -func (gp *GlobalBlockPersister) TruncateBefore(n types.GlobalBlockNumber) error { - for s := range gp.state.Lock() { - return s.truncateBefore(n) - } - panic("unreachable") -} - -// TruncateAfter removes block entries >= n from the WAL, updates the -// persister cursor, and trims loaded data. Called by DataWAL.reconcile -// to remove blocks persisted without corresponding QCs. -func (gp *GlobalBlockPersister) TruncateAfter(n types.GlobalBlockNumber) error { - for s := range gp.state.Lock() { - iw, ok := s.iw.Get() - if ok && iw.Count() > 0 && n < s.next { - firstGlobal := s.next - types.GlobalBlockNumber(iw.Count()) - walIdx := iw.FirstIdx() - if n > firstGlobal { - walIdx += uint64(n - firstGlobal) - } - if err := iw.TruncateAfter(walIdx); err != nil { - return fmt.Errorf("truncate global block WAL after %d: %w", n, err) - } - s.next = firstGlobal + types.GlobalBlockNumber(iw.Count()) - } - trimmed := s.loaded[:0] - for _, lb := range s.loaded { - if lb.Number < n { - trimmed = append(trimmed, lb) - } - } - s.loaded = trimmed - return nil - } - panic("unreachable") -} - -// Close shuts down the WAL. -func (gp *GlobalBlockPersister) Close() error { - for s := range gp.state.Lock() { - iw, ok := s.iw.Get() - if !ok { - return nil - } - s.iw = utils.None[*indexedWAL[LoadedGlobalBlock]]() - return iw.Close() - } - panic("unreachable") -} - -func (s *globalBlockState) loadAll() ([]LoadedGlobalBlock, error) { - iw, ok := s.iw.Get() - if !ok { - return nil, nil - } - return iw.ReadAll() -} diff --git a/sei-tendermint/internal/autobahn/consensus/persist/globalblocks_test.go b/sei-tendermint/internal/autobahn/consensus/persist/globalblocks_test.go deleted file mode 100644 index 85e50feb7c..0000000000 --- a/sei-tendermint/internal/autobahn/consensus/persist/globalblocks_test.go +++ /dev/null @@ -1,296 +0,0 @@ -package persist - -import ( - "testing" - - "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" -) - -func makeGlobalBlocks(rng utils.Rng, n int) []*types.Block { - blocks := make([]*types.Block, n) - for i := range blocks { - blocks[i] = types.GenBlock(rng) - } - return blocks -} - -func TestNewGlobalBlockPersisterEmptyDir(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.NotNil(t, gp) - require.Equal(t, 0, len(gp.ConsumeLoaded())) - require.Equal(t, fb, gp.Next()) - require.NoError(t, gp.Close()) -} - -func TestNewGlobalBlockPersisterNoop(t *testing.T) { - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - blocks := makeGlobalBlocks(rng, 5) - - gp, err := NewGlobalBlockPersister(utils.None[string](), committee) - require.NoError(t, err) - require.NotNil(t, gp) - require.Equal(t, 0, len(gp.ConsumeLoaded())) - - for i, b := range blocks { - require.NoError(t, gp.PersistBlock(fb+types.GlobalBlockNumber(i), b)) - } - require.Equal(t, fb+5, gp.Next()) - - // Truncate in no-op mode. Before the fix, truncateBefore wouldn't - // advance s.next, so subsequent PersistBlock calls at higher numbers - // would fail with "out of sequence". - require.NoError(t, gp.TruncateBefore(fb+10)) - require.Equal(t, fb+10, gp.Next()) - - newBlock := types.GenBlock(rng) - require.NoError(t, gp.PersistBlock(fb+10, newBlock)) - require.Equal(t, fb+11, gp.Next()) - require.NoError(t, gp.Close()) -} - -func TestGlobalBlockPersistAndReload(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - blocks := makeGlobalBlocks(rng, 5) - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - for i, b := range blocks { - require.NoError(t, gp.PersistBlock(fb+types.GlobalBlockNumber(i), b)) - } - require.Equal(t, fb+5, gp.Next()) - require.NoError(t, gp.Close()) - - gp2, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - loaded := gp2.ConsumeLoaded() - require.Equal(t, 5, len(loaded)) - for i, lb := range loaded { - require.Equal(t, fb+types.GlobalBlockNumber(i), lb.Number) - } - require.Equal(t, fb+5, gp2.Next()) - require.NoError(t, gp2.Close()) -} - -func TestGlobalBlockTruncateAndReload(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - blocks := makeGlobalBlocks(rng, 10) - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - for i, b := range blocks { - require.NoError(t, gp.PersistBlock(fb+types.GlobalBlockNumber(i), b)) - } - require.NoError(t, gp.TruncateBefore(fb+5)) - require.NoError(t, gp.Close()) - - gp2, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - loaded := gp2.ConsumeLoaded() - require.Equal(t, 5, len(loaded)) - require.Equal(t, fb+5, loaded[0].Number) - require.Equal(t, fb+9, loaded[4].Number) - require.Equal(t, fb+10, gp2.Next()) - require.NoError(t, gp2.Close()) -} - -func TestGlobalBlockTruncateAll(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - blocks := makeGlobalBlocks(rng, 5) - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - for i, b := range blocks { - require.NoError(t, gp.PersistBlock(fb+types.GlobalBlockNumber(i), b)) - } - require.NoError(t, gp.TruncateBefore(fb+10)) - require.Equal(t, fb+10, gp.Next()) - require.NoError(t, gp.Close()) - - gp2, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 0, len(gp2.ConsumeLoaded())) - require.Equal(t, fb, gp2.Next()) - require.NoError(t, gp2.Close()) -} - -func TestGlobalBlockDuplicateIgnored(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - block := types.GenBlock(rng) - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.NoError(t, gp.PersistBlock(fb, block)) - require.NoError(t, gp.PersistBlock(fb, block)) - require.Equal(t, fb+1, gp.Next()) - require.NoError(t, gp.Close()) -} - -func TestGlobalBlockGapError(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - block := types.GenBlock(rng) - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - err = gp.PersistBlock(fb+2, block) - require.Error(t, err) - require.Contains(t, err.Error(), "out of sequence") - require.NoError(t, gp.Close()) -} - -func TestGlobalBlockTruncateBeforeNoop(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - blocks := makeGlobalBlocks(rng, 5) - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - for i, b := range blocks { - require.NoError(t, gp.PersistBlock(fb+types.GlobalBlockNumber(i), b)) - } - require.NoError(t, gp.TruncateBefore(0)) - require.Equal(t, fb+5, gp.Next()) - require.NoError(t, gp.Close()) - - gp2, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 5, len(gp2.ConsumeLoaded())) - require.NoError(t, gp2.Close()) -} - -func TestGlobalBlockContinueAfterReload(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - blocks := makeGlobalBlocks(rng, 10) - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - for i := range 5 { - require.NoError(t, gp.PersistBlock(fb+types.GlobalBlockNumber(i), blocks[i])) - } - require.NoError(t, gp.Close()) - - gp2, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 5, len(gp2.ConsumeLoaded())) - for i := 5; i < 10; i++ { - require.NoError(t, gp2.PersistBlock(fb+types.GlobalBlockNumber(i), blocks[i])) - } - require.Equal(t, fb+10, gp2.Next()) - require.NoError(t, gp2.Close()) - - gp3, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 10, len(gp3.ConsumeLoaded())) - require.NoError(t, gp3.Close()) -} - -func TestGlobalBlockTruncateAfterMiddle(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - blocks := makeGlobalBlocks(rng, 5) - - // First session: persist 5 blocks. - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - for i, b := range blocks { - require.NoError(t, gp.PersistBlock(fb+types.GlobalBlockNumber(i), b)) - } - require.NoError(t, gp.Close()) - - // Second session: reload, then TruncateAfter trims WAL and loaded. - gp2, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.NoError(t, gp2.TruncateAfter(fb+3)) - require.Equal(t, fb+3, gp2.Next()) - loaded := gp2.ConsumeLoaded() - require.Equal(t, 3, len(loaded)) - require.Equal(t, fb, loaded[0].Number) - require.Equal(t, fb+2, loaded[2].Number) - require.NoError(t, gp2.Close()) - - // Third session: verify WAL persistence. - gp3, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - loaded = gp3.ConsumeLoaded() - require.Equal(t, 3, len(loaded)) - require.Equal(t, fb, loaded[0].Number) - require.Equal(t, fb+2, loaded[2].Number) - require.NoError(t, gp3.Close()) -} - -func TestGlobalBlockTruncateAfterNoop(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - blocks := makeGlobalBlocks(rng, 3) - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - for i, b := range blocks { - require.NoError(t, gp.PersistBlock(fb+types.GlobalBlockNumber(i), b)) - } - // TruncateAfter at or past the last block is a no-op. - require.NoError(t, gp.TruncateAfter(fb+11)) - require.Equal(t, fb+3, gp.Next()) - require.NoError(t, gp.Close()) -} - -func TestGlobalBlockTruncateAfterBeforeFirst(t *testing.T) { - dir := t.TempDir() - rng := utils.TestRng() - committee, _ := types.GenCommittee(rng, 3) - fb := committee.FirstBlock() - blocks := makeGlobalBlocks(rng, 5) - - gp, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - for i, b := range blocks { - require.NoError(t, gp.PersistBlock(fb+types.GlobalBlockNumber(i), b)) - } - // Truncate first 2 blocks, leaving [fb+2, fb+5). - require.NoError(t, gp.TruncateBefore(fb+2)) - require.Equal(t, fb+5, gp.Next()) - - // TruncateAfter at or before the first remaining block removes everything. - require.NoError(t, gp.TruncateAfter(fb+2)) - require.Equal(t, fb+2, gp.Next()) // cursor stays at firstGlobal - require.NoError(t, gp.Close()) - - // Reload — should be empty. - gp2, err := NewGlobalBlockPersister(utils.Some(dir), committee) - require.NoError(t, err) - require.Equal(t, 0, len(gp2.ConsumeLoaded())) - require.NoError(t, gp2.Close()) -} diff --git a/sei-tendermint/internal/autobahn/consensus/persist/testonly.go b/sei-tendermint/internal/autobahn/consensus/persist/testonly.go deleted file mode 100644 index 31f852897d..0000000000 --- a/sei-tendermint/internal/autobahn/consensus/persist/testonly.go +++ /dev/null @@ -1,8 +0,0 @@ -package persist - -// SetLoadedForTest overrides loaded block data. Test-only. -func (gp *GlobalBlockPersister) SetLoadedForTest(loaded []LoadedGlobalBlock) { - for s := range gp.state.Lock() { - s.loaded = loaded - } -} diff --git a/sei-tendermint/internal/autobahn/consensus/state_test.go b/sei-tendermint/internal/autobahn/consensus/state_test.go index 43cdb6ef61..8d2df5ede9 100644 --- a/sei-tendermint/internal/autobahn/consensus/state_test.go +++ b/sei-tendermint/internal/autobahn/consensus/state_test.go @@ -18,10 +18,7 @@ import ( // keys[0] is used as the node's signing key. func newTestState(rng utils.Rng) (*State, []types.SecretKey) { committee, keys := types.GenCommittee(rng, 3) - dataState := utils.OrPanic1(data.NewState( - &data.Config{Committee: committee}, - utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee)), - )) + dataState := newTestDataState(committee) s := utils.OrPanic1(NewState(&Config{ Key: keys[0], ViewTimeout: func(types.View) time.Duration { return time.Hour }, @@ -268,12 +265,7 @@ func TestVoteTimeoutPrepareQC_PersistedRestart(t *testing.T) { PersistentStateDir: utils.Some(dir), } } - makeDataState := func() *data.State { - return utils.OrPanic1(data.NewState( - &data.Config{Committee: committee}, - utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee)), - )) - } + makeDataState := func() *data.State { return newTestDataState(committee) } view0 := types.View{Index: 0, Number: 0} pqc0 := makePrepareQC(keys, types.GenProposalAt(rng, view0)) diff --git a/sei-tendermint/internal/autobahn/data/state.go b/sei-tendermint/internal/autobahn/data/state.go index c4521c919b..dd3ae8d8ce 100644 --- a/sei-tendermint/internal/autobahn/data/state.go +++ b/sei-tendermint/internal/autobahn/data/state.go @@ -7,7 +7,6 @@ import ( "time" "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/metrics" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope" @@ -40,101 +39,10 @@ type StateAPI interface { var _ StateAPI = (*State)(nil) -// DataWAL groups the WALs used by State for crash recovery. -// Both fields are always non-nil; when stateDir is None they are no-op persisters. -// This is temporary and will go away once we switch to a proper storage solution. -type DataWAL struct { - Blocks *persist.GlobalBlockPersister - CommitQCs *persist.FullCommitQCPersister -} - -// Close shuts down both WALs. -func (dw *DataWAL) Close() error { - return errors.Join(dw.Blocks.Close(), dw.CommitQCs.Close()) -} - -// TruncateBefore removes entries fully before n from both WALs in parallel. -// A crash between the two calls just leaves stale entries in one WAL, -// which are harmless on reload. -func (dw *DataWAL) TruncateBefore(n types.GlobalBlockNumber) error { - return scope.Parallel(func(ps scope.ParallelScope) error { - ps.Spawn(func() error { - if err := dw.Blocks.TruncateBefore(n); err != nil { - return fmt.Errorf("truncate global block WAL: %w", err) - } - return nil - }) - ps.Spawn(func() error { - if err := dw.CommitQCs.TruncateBefore(n); err != nil { - return fmt.Errorf("truncate full commitqc WAL: %w", err) - } - return nil - }) - return nil - }) -} - -// reconcile fixes cursor inconsistencies between the two WALs that can -// result from crashes during parallel persistence or pruning. -// -// The possible WAL states on startup and how they are handled: -// -// Case Blocks QCs Scenario Action -// 1 empty empty Fresh start no-op -// 2 [a,b] empty QCs lost (corruption) error (statesync needed) -// 3 empty [X,Y) Blocks lost (crash) Prefix: fast-forward blocks.next to X -// 4 [a,b] [X,Y) Normal (a=X, bX) Prefix: truncate QCs to a -// 6 [a,b] [X,Y) Prune crash: QCs ahead (a=Y) Tail: truncate blocks to Y -// 8 [a,b] [X,Y) QCs ahead (normal, b fb { - // Blocks exist but QCs WAL is empty — data is corrupted. - return fmt.Errorf("corrupted WAL: blocks exist but QCs WAL is empty; statesync required to recover") - } - if err := dw.Blocks.TruncateAfter(qcNext); err != nil { - return fmt.Errorf("truncate blocks tail: %w", err) - } - // Fix prefix: align both WALs to the later start. - blocksFirst := dw.Blocks.LoadedFirst() - qcsFirst := dw.CommitQCs.LoadedFirst() - reconciled := max(blocksFirst, qcsFirst) - if reconciled > fb { - if err := dw.TruncateBefore(reconciled); err != nil { - return err - } - } - return nil -} - -// NewDataWAL constructs both global-block and global-commitqc WALs. -// When stateDir is None, the returned persisters are no-ops. -func NewDataWAL(stateDir utils.Option[string], committee *types.Committee) (*DataWAL, error) { - blocks, err := persist.NewGlobalBlockPersister(stateDir, committee) - if err != nil { - return nil, fmt.Errorf("global block WAL: %w", err) - } - commitQCs, err := persist.NewFullCommitQCPersister(stateDir, committee) - if err != nil { - _ = blocks.Close() - return nil, fmt.Errorf("full commitqc WAL: %w", err) - } - dw := &DataWAL{ - Blocks: blocks, - CommitQCs: commitQCs, - } - // Reconcile cursor inconsistency: a crash between the two parallel - // TruncateBefore calls can leave one WAL truncated while the other - // still has stale entries. Advance both to the max starting point. - if err := dw.reconcile(committee); err != nil { - _ = dw.Close() - return nil, fmt.Errorf("reconcile WALs: %w", err) - } - return dw, nil +// blockEntry is a (number, block) pair collected in runPersist batches. +type blockEntry struct { + n types.GlobalBlockNumber + blk *types.Block } type appProposalWithTimestamp struct { @@ -148,15 +56,12 @@ type inner struct { // appProposal[n] contains appProposal block >=n. appProposals map[types.GlobalBlockNumber]appProposalWithTimestamp // [first,nextAppProposal) - // blockHashes is a hash → height index mirroring blocks. Maintained - // in lockstep with blocks via insertBlock / pruneFirst, so it covers - // exactly the same retain window without a separate prune cursor or - // startup warmup. Powers BlockByHash. + // blockHashes is a hash → height index for GlobalBlockByHash. Maintained + // in lockstep with blocks via insertBlock / pruneFirst, covering the same + // retain window. // - // TODO(autobahn): remove once a writer is wired into block execution - // that populates sei-db/ledger_db/block.BlockDB. BlockDB has a built-in - // hash index that survives process restart and lives outside Autobahn's - // RetainHeight pruning, making this in-memory index obsolete. + // TODO: replace with blockDB.ReadBlockByHash once BlockDB exposes the + // primary GlobalBlockNumber alongside the block on a secondary-key lookup. blockHashes map[types.BlockHeaderHash]types.GlobalBlockNumber // first <= nextAppProposal <= nextBlockToPersist <= nextBlock <= nextQC @@ -278,66 +183,152 @@ type State struct { cfg *Config metrics *metrics.Metrics inner utils.Watch[*inner] - dataWAL *DataWAL + blockDB utils.Option[types.BlockDB] + // Captured in NewState (before any goroutines start) and consumed once at + // the top of Run to seed runPersist. Capturing here avoids a race where a + // concurrent PushQC (called via an inbound connection after the transport + // starts but before Run's scope begins) could advance inner.nextQC/nextBlock + // past unpersisted data, causing runPersist to skip writing those entries. + // + // TODO: expose persisted watermarks from BlockDB directly once runPersist no longer needs to cache all blocks/QCs in-memory. + blockDBPersistedQC types.GlobalBlockNumber + blockDBPersistedBlock types.GlobalBlockNumber } -// NewState constructs a new data State. -// dataWAL persists blocks and QCs to WALs for crash recovery and provides -// preloaded data from the previous run. Use NewDataWAL to construct it. -func NewState(cfg *Config, dataWAL *DataWAL) (*State, error) { - inner := newInner(cfg.Committee) - // Fast-forward cursors to where data starts. Use blocks as golden: - // per-block pruning may split a QC range, so blocks determine where - // useful data starts. QCs before that are kept for verification but - // don't set inner.first. - blocksFirst := dataWAL.Blocks.LoadedFirst() - qcFirst := dataWAL.CommitQCs.LoadedFirst() - dataFirst := max(blocksFirst, qcFirst) - if dataFirst > cfg.Committee.FirstBlock() { - inner.skipTo(dataFirst) - } - // Restore QCs. insertQC handles partially pruned QCs (range starts - // before inner.first) by skipping the pruned prefix. - for _, qc := range dataWAL.CommitQCs.ConsumeLoaded() { - if err := inner.insertQC(cfg.Committee, qc); err != nil { - return nil, fmt.Errorf("load QC from WAL: %w", err) - } - } - // Restore blocks. Verify contiguity as defense in depth. - expectedBlock := inner.first - for _, lb := range dataWAL.Blocks.ConsumeLoaded() { - if lb.Number < inner.first || lb.Number >= inner.nextQC { - continue // outside QC range (stale from reconcile) - } - if lb.Number != expectedBlock { - return nil, fmt.Errorf("block gap in WAL: expected %d, got %d", expectedBlock, lb.Number) - } - expectedBlock = lb.Number + 1 - if err := lb.Block.Verify(cfg.Committee); err != nil { - return nil, fmt.Errorf("load block %d from WAL: %w", lb.Number, err) - } - if err := inner.insertBlock(cfg.Committee, lb.Number, lb.Block); err != nil { - return nil, fmt.Errorf("load block %d from WAL: %w", lb.Number, err) - } - } - // Advance nextBlock through contiguous blocks. Don't use - // updateNextBlock: stale timestamps would skew metrics. - for ; inner.blocks[inner.nextBlock] != nil; inner.nextBlock++ { - } - // Data loaded from WALs was already persisted in the previous run. - inner.nextBlockToPersist = inner.nextBlock - // WAL cursor consistency was resolved by DataWAL.reconcile at construction. - // Verify the blocks persister cursor is not behind inner.nextBlock. - if dataWAL.Blocks.Next() < inner.nextBlock { - return nil, fmt.Errorf("blocks WAL cursor %d behind inner.nextBlock %d after reconciliation", - dataWAL.Blocks.Next(), inner.nextBlock) - } - return &State{ +// NewState constructs a data State, replaying persisted state from blockDB if +// set. Passing None disables persistence: Run advances nextBlockToPersist +// without any disk I/O. The caller owns blockDB and is responsible for closing +// it after Run returns; State never closes it. +// TODO(gprusak): add support for starting execution from non-zero commit QC. +func NewState(cfg *Config, blockDB utils.Option[types.BlockDB]) (*State, error) { + s := &State{ cfg: cfg, metrics: metrics.Get(), - inner: utils.NewWatch(inner), - dataWAL: dataWAL, - }, nil + inner: utils.NewWatch(newInner(cfg.Committee)), + } + if db, ok := blockDB.Get(); ok { + if err := s.loadFromBlockDB(db); err != nil { + return nil, fmt.Errorf("loadFromBlockDB: %w", err) + } + s.blockDB = blockDB + } + for inner := range s.inner.Lock() { + s.blockDBPersistedQC = inner.nextQC + s.blockDBPersistedBlock = inner.nextBlock + } + return s, nil +} + +// loadFromBlockDB replays QCs and blocks from blockDB into s.inner. +// Called from NewState before any goroutines are spawned; the lock is acquired +// only to satisfy the Watch API. +func (s *State) loadFromBlockDB(blockDB types.BlockDB) error { + for in := range s.inner.Lock() { + // Restore QCs from BlockDB. On the first QC, skipTo its GlobalRange.First + // to advance past any pruned prefix (a straddling QC starts before the + // prune boundary). Subsequent QCs must be consecutive — insertQC errors + // on any gap. + var err error + err = func() error { + it, err := blockDB.QCs(false) + if err != nil { + return fmt.Errorf("open QC iterator: %w", err) + } + defer it.Close() + for { + ok, err := it.Next() + if err != nil || !ok { + return err + } + qc, err := it.QC() + if err != nil { + return err + } + if in.first == in.nextQC { + gr := qc.QC().GlobalRange(s.cfg.Committee) + if gr.First < in.nextQC { + return fmt.Errorf("QC in BlockDB predates committee genesis %d: got %d", in.nextQC, gr.First) + } + in.skipTo(gr.First) + } + if err := in.insertQC(s.cfg.Committee, qc); err != nil { + return fmt.Errorf("load QC from BlockDB: %w", err) + } + } + }() + if err != nil { + return err + } + + // Restore blocks from BlockDB. The first retained block drives inner.first — + // a straddling QC may start before the prune boundary, so QC data alone + // cannot determine where blocks begin. insertBlock's n >= nextQC guard + // enforces the no-block-without-QC invariant. + err = func() error { + it2, err := blockDB.Blocks(false) + if err != nil { + return fmt.Errorf("open block iterator: %w", err) + } + defer it2.Close() + firstBlock := true + var nextExpect types.GlobalBlockNumber + for { + ok, err := it2.Next() + if err != nil || !ok { + return err + } + n := it2.Number() + if n >= in.nextQC { + return fmt.Errorf("block %d in BlockDB has no QC coverage (nextQC=%d)", n, in.nextQC) + } + if firstBlock { + firstBlock = false + if n < in.first { + return fmt.Errorf("block %d in BlockDB predates first QC start %d", n, in.first) + } + // Drop QC entries below the first retained block (straddling-QC + // case: QC started before the prune boundary so the QC pass + // populated in.qcs for [qcFirst, n); those entries are unreachable + // via pruneFirst which starts at first=n). + for k := in.first; k < n; k++ { + delete(in.qcs, k) + } + in.first = n + in.nextBlock = n + in.nextAppProposal = n + nextExpect = n + } + // updateNextBlock only advances nextBlock through contiguous present + // entries, so runPersist always writes [persistedBlock, nextBlock) + // fully populated. A gap here means BlockDB corruption. + if n != nextExpect { + return fmt.Errorf("block gap in BlockDB: expected %d, got %d", nextExpect, n) + } + nextExpect++ + blk, err := it2.Block() + if err != nil { + return err + } + if err := blk.Verify(s.cfg.Committee); err != nil { + return fmt.Errorf("verify block %d from BlockDB: %w", n, err) + } + if err := in.insertBlock(s.cfg.Committee, n, blk); err != nil { + return fmt.Errorf("insert block %d from BlockDB: %w", n, err) + } + } + }() + if err != nil { + return err + } + + // Advance nextBlock through contiguous loaded blocks. Don't use + // updateNextBlock: stale timestamps would skew metrics. + for ; in.blocks[in.nextBlock] != nil; in.nextBlock++ { + } + // Data loaded from BlockDB was already durably persisted. + in.nextBlockToPersist = in.nextBlock + } + return nil } // Committee returns the committee. @@ -450,20 +441,7 @@ func (s *State) NextBlock() types.GlobalBlockNumber { // GlobalBlockByHash returns the finalized GlobalBlock whose stored header // hashes to the given value, or None if no such block is currently in the -// retained range. The lookup-and-construct happens under a single lock so -// the returned block matches the looked-up hash atomically — pruning can't -// change which height a hash maps to between the index check and the block -// construction. Tracks the same retain window as Block / GlobalBlock since -// the hash index is maintained in lockstep by insertBlock / pruneFirst. -// -// Returns an error in the signature for forward-compat with the eventual -// switch to sei-db/ledger_db/block.BlockDB.GetBlockByHash. Today's -// in-memory implementation never errors. -// -// TODO(autobahn): when BlockDB is wired, take a ctx parameter and narrow -// the error contract — db-internal errors should surface by shutting down -// the persistence background task (matching how persistence handles errors -// today), so the query path's error stays bounded to context.Canceled. +// retained range. Non-blocking. func (s *State) GlobalBlockByHash(hash types.BlockHeaderHash) (utils.Option[*types.GlobalBlock], error) { for inner := range s.inner.Lock() { n, ok := inner.blockHashes[hash] @@ -645,41 +623,40 @@ func (s *State) PruneBefore(retainFrom types.GlobalBlockNumber) error { if firstToKeep <= inner.first { return nil } - // Keep at least one entry so WALs are never empty on restart. + // Keep at least one entry so BlockDB is never empty on restart. for inner.first+1 < firstToKeep { inner.pruneFirst(pruningTime, s.metrics) } ctrl.Updated() truncateTo = utils.Some(inner.first) } - // Truncate WALs outside the lock to avoid holding it during disk I/O. - if n, ok := truncateTo.Get(); ok { - return s.dataWAL.TruncateBefore(n) + // Prune BlockDB outside the lock to avoid holding it during disk I/O. + // PruneBefore advances an in-memory watermark; GC is asynchronous and not + // persisted. On restart before GC reclaims entries, NewState may see + // below-watermark blocks/QCs and set inner.first lower than the pre-crash + // watermark. This is safe (resurrected data is QC-covered committed data) + // and self-heals on the next runPruning cycle. Durable prune bounds come + // from the BlockDB retention TTL, not from PruneBefore itself. + if db, ok := s.blockDB.Get(); ok { + if n, ok := truncateTo.Get(); ok { + return db.PruneBefore(n) + } } return nil } -// runPersist is a background goroutine that persists blocks and QCs to WALs. +// runPersist is a background goroutine that persists blocks and QCs to BlockDB. // It waits for in-memory data to advance past the persistence cursors, then -// writes QCs and blocks in parallel. QCs are persisted up to nextQC (eagerly), -// blocks up to nextBlock. nextBlockToPersist advances to min(persistedQC, persistedBlock) +// writes QCs (first, per the BlockDB contract) and blocks, then flushes once +// per batch. nextBlockToPersist advances to min(persistedQC, persistedBlock) // to unblock PushAppHash only when both are durable. // Errors propagate vertically (kill the component). -func (s *State) runPersist(ctx context.Context) error { - // Initialize from nextBlockToPersist, not nextQC/nextBlock. PushQC may - // have been called before Run() (race between p2p startup and Run), - // advancing nextQC/nextBlock beyond what's been persisted. Starting - // from nextBlockToPersist ensures we don't skip unpersisted data. - var persistedQC, persistedBlock types.GlobalBlockNumber - for inner := range s.inner.Lock() { - persistedQC = inner.nextBlockToPersist - persistedBlock = inner.nextBlockToPersist - } +func (s *State) runPersist(ctx context.Context, persistedQC, persistedBlock types.GlobalBlockNumber) error { for { // Wait for unpersisted data and snapshot what needs writing. type batch struct { qcs []*types.FullCommitQC - blocks []persist.LoadedGlobalBlock + blocks []blockEntry qcEnd types.GlobalBlockNumber blockEnd types.GlobalBlockNumber } @@ -704,33 +681,27 @@ func (s *State) runPersist(ctx context.Context) error { } // Collect blocks for [persistedBlock, nextBlock). for n := persistedBlock; n < inner.nextBlock; n++ { - b.blocks = append(b.blocks, persist.LoadedGlobalBlock{ - Number: n, - Block: inner.blocks[n], - }) + b.blocks = append(b.blocks, blockEntry{n: n, blk: inner.blocks[n]}) } } - // Persist QCs and blocks in parallel. - if err := scope.Parallel(func(ps scope.ParallelScope) error { - ps.Spawn(func() error { - for _, qc := range b.qcs { - if err := s.dataWAL.CommitQCs.PersistQC(qc); err != nil { - return fmt.Errorf("persist full commitqc: %w", err) - } + if db, ok := s.blockDB.Get(); ok { + // Write QCs first (BlockDB contract: QC must precede covered blocks). + for _, qc := range b.qcs { + gr := qc.QC().GlobalRange(s.cfg.Committee) + if err := db.WriteQC(gr.First, gr.Next, qc); err != nil { + return fmt.Errorf("write QC [%d,%d): %w", gr.First, gr.Next, err) } - return nil - }) - ps.Spawn(func() error { - for _, lb := range b.blocks { - if err := s.dataWAL.Blocks.PersistBlock(lb.Number, lb.Block); err != nil { - return fmt.Errorf("persist global block %d: %w", lb.Number, err) - } + } + for _, lb := range b.blocks { + if err := db.WriteBlock(lb.n, lb.blk); err != nil { + return fmt.Errorf("write block %d: %w", lb.n, err) } - return nil - }) - return nil - }); err != nil { - return err + } + // Flush once per batch before advancing nextBlockToPersist, so that + // PushAppHash only unblocks after data is crash-durable. + if err := db.Flush(); err != nil { + return fmt.Errorf("flush BlockDB: %w", err) + } } persistedQC = b.qcEnd persistedBlock = b.blockEnd @@ -770,10 +741,12 @@ func (s *State) runPruning(ctx context.Context, after time.Duration) error { } pruningTime = inner.appProposals[inner.first].timestamp.Add(after) } - // Truncate WALs outside the lock to avoid holding it during disk I/O. - if n, ok := truncateTo.Get(); ok { - if err := s.dataWAL.TruncateBefore(n); err != nil { - return err + // Prune BlockDB outside the lock to avoid holding it during disk I/O. + if db, ok := s.blockDB.Get(); ok { + if n, ok := truncateTo.Get(); ok { + if err := db.PruneBefore(n); err != nil { + return fmt.Errorf("prune BlockDB before %d: %w", n, err) + } } } // Wait until the next pruning time. @@ -783,12 +756,11 @@ func (s *State) runPruning(ctx context.Context, after time.Duration) error { } } -// Run runs the background tasks of the data State. -// TODO(gprusak): add support for starting execution from non-zero commit QC. +// Run starts the background goroutines (persistence and pruning). func (s *State) Run(ctx context.Context) error { return scope.Run(ctx, func(ctx context.Context, scope scope.Scope) error { scope.SpawnNamed("runPersist", func() error { - return s.runPersist(ctx) + return s.runPersist(ctx, s.blockDBPersistedQC, s.blockDBPersistedBlock) }) if pruneAfter, ok := s.cfg.PruneAfter.Get(); ok { scope.SpawnNamed("runPruning", func() error { diff --git a/sei-tendermint/internal/autobahn/data/state_test.go b/sei-tendermint/internal/autobahn/data/state_test.go index 7f5c57b477..ad99bc3b99 100644 --- a/sei-tendermint/internal/autobahn/data/state_test.go +++ b/sei-tendermint/internal/autobahn/data/state_test.go @@ -5,14 +5,12 @@ import ( "errors" "fmt" "maps" - "os" - "path/filepath" "testing" "testing/synctest" "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/libs/utils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/scope" @@ -47,14 +45,52 @@ func snapshot(s *State) Snapshot { panic("unreachable") } +// newTestBlockDB opens (or creates) a LittDB-backed BlockDB at dir. +// Retention is set to 1ns so ForceGC reclaims pruned data immediately in tests. +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) + } + cfg.Retention = time.Nanosecond + db, err := littblock.NewBlockDB(cfg) + if err != nil { + t.Fatalf("littblock.NewBlockDB: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + return db +} + +// newTestState constructs a State, replays db, and returns it ready to Run. +func newTestState(t testing.TB, cfg *Config, db types.BlockDB) *State { + t.Helper() + s, err := NewState(cfg, utils.Some(db)) + require.NoError(t, err) + return s +} + +// writeToBlockDB writes QC+block pairs sequentially to db and flushes once. +// qcs[i] and blockss[i] must correspond; QCs must be in ascending order. +func writeToBlockDB(t *testing.T, db types.BlockDB, committee *types.Committee, qcs []*types.FullCommitQC, blockss [][]*types.Block) { + t.Helper() + for i, qc := range qcs { + gr := qc.QC().GlobalRange(committee) + require.NoError(t, db.WriteQC(gr.First, gr.Next, qc)) + for j, n := 0, gr.First; n < gr.Next; n++ { + require.NoError(t, db.WriteBlock(n, blockss[i][j])) + j++ + } + } + require.NoError(t, db.Flush()) +} + func TestState(t *testing.T) { ctx := t.Context() rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) if err := scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { - state := utils.OrPanic1(NewState(&Config{ - Committee: committee, - }, utils.OrPanic1(NewDataWAL(utils.None[string](), committee)))) + state := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) @@ -125,9 +161,7 @@ func TestPushConflictingBadCommitQC(t *testing.T) { ctx := t.Context() rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) - state := utils.OrPanic1(NewState(&Config{ - Committee: committee, - }, utils.OrPanic1(NewDataWAL(utils.None[string](), committee)))) + state := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) // Push a valid QC to advance inner.nextQC. qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) @@ -235,9 +269,7 @@ func TestPushQCIgnoresBlocksMatchingUnverifiedHeaders(t *testing.T) { ctx := t.Context() rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) - state := utils.OrPanic1(NewState(&Config{ - Committee: committee, - }, utils.OrPanic1(NewDataWAL(utils.None[string](), committee)))) + state := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) // Push qc1 with NO blocks — only the QC is stored. qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) @@ -281,9 +313,7 @@ func TestExecution(t *testing.T) { rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) if err := scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { - state := utils.OrPanic1(NewState(&Config{ - Committee: committee, - }, utils.OrPanic1(NewDataWAL(utils.None[string](), committee)))) + state := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) @@ -325,9 +355,7 @@ func TestPushBlockAcceptsBlockWithQC(t *testing.T) { rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) - state := utils.OrPanic1(NewState(&Config{ - Committee: committee, - }, utils.OrPanic1(NewDataWAL(utils.None[string](), committee)))) + state := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) // Push QC without blocks. qc, blocks := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) @@ -341,26 +369,12 @@ func TestPushBlockAcceptsBlockWithQC(t *testing.T) { require.Equal(t, blocks[0], got) } -// TestGlobalBlockByHash isolates the hash-keyed lookup from the -// consensus-driven harness. We push a single QC + block via the same code -// path the network would (insertBlock writes to inner.blockHashes), then: -// -// - the block's own header hash resolves to Some(*GlobalBlock) with the -// expected height/header/payload — the index points at the right -// block, atomically with the block construction -// - a zero hash and a random hash both resolve to None — distinct -// unknown-hash inputs all read as "not found", no panics -// - err is nil throughout — today's in-memory implementation has no -// failure mode; the error return on GlobalBlockByHash is reserved -// for the future BlockDB-backed path func TestGlobalBlockByHash(t *testing.T) { ctx := t.Context() rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) - state := utils.OrPanic1(NewState(&Config{ - Committee: committee, - }, utils.OrPanic1(NewDataWAL(utils.None[string](), committee)))) + state := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) qc, blocks := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) require.NoError(t, state.PushQC(ctx, qc, blocks)) @@ -393,680 +407,514 @@ func TestGlobalBlockByHash(t *testing.T) { require.False(t, ok, "GlobalBlockByHash(random) returned Some") } -// ── Reconcile tests (grouped by case number) ────────────────────────── +// ── Recovery tests ──────────────────────────────────────────────────── -// TestStateRecoveryBlocksOnly simulates a crash after blocks are written -// TestReconcileCase1Empty verifies that reconcile is a no-op on a fresh -// WAL directory with no data. -func TestReconcileCase1Empty(t *testing.T) { - t.Log("Reconcile case 1: Fresh start (empty/empty)") +// TestRecoveryEmpty verifies that NewState is a no-op on a fresh BlockDB. +func TestRecoveryEmpty(t *testing.T) { rng := utils.TestRng() committee, _ := types.GenCommittee(rng, 3) dir := t.TempDir() fb := committee.FirstBlock() - dw := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state := utils.OrPanic1(NewState(&Config{Committee: committee}, dw)) - + db := newTestBlockDB(t, dir) + state := newTestState(t, &Config{Committee: committee}, db) for inner := range state.inner.Lock() { require.Equal(t, fb, inner.first) require.Equal(t, fb, inner.nextQC) require.Equal(t, fb, inner.nextBlock) } - require.NoError(t, dw.Close()) -} - -// TestReconcileCase2Corrupted verifies that when blocks are persisted but -// QCs WAL is deleted (corruption), NewDataWAL returns a corruption error. -func TestReconcileCase2Corrupted(t *testing.T) { - t.Log("Reconcile case 2: QCs lost (corruption), returns error") - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - dir := t.TempDir() - - // Persist blocks and QCs normally. - qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) - gr1 := qc1.QC().GlobalRange(committee) - - dw1 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - require.NoError(t, dw1.CommitQCs.PersistQC(qc1)) - for i, n := 0, gr1.First; n < gr1.Next; n++ { - require.NoError(t, dw1.Blocks.PersistBlock(n, blocks1[i])) - i++ - } - require.NoError(t, dw1.Close()) - - // Simulate corruption: delete QCs WAL. - require.NoError(t, os.RemoveAll(filepath.Join(dir, "fullcommitqcs"))) - - // Reopen should fail — blocks exist but QCs are gone. - _, err := NewDataWAL(utils.Some(dir), committee) - require.Error(t, err) - require.Contains(t, err.Error(), "corrupted") } -// TestStateRecoveryQCsOnly simulates a crash after QCs are written to the -// WAL but before blocks are written (or after blocks WAL is truncated but -// QCs WAL is not). On recovery, QCs are loaded but no blocks exist. -// The cursor sync in NewState must advance the blocks persister so that -// subsequent PersistBlock calls don't fail with "out of sequence". -func TestReconcileCase3BlocksLost(t *testing.T) { - t.Log("Reconcile case 3: Blocks lost (crash), QCs survive") +// TestNewStateNilInMemoryMode verifies that NewState(cfg, None) followed by Run +// works end-to-end: QCs and blocks are accessible in memory with no BlockDB backing. +func TestNewStateNilInMemoryMode(t *testing.T) { ctx := t.Context() rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) - dir := t.TempDir() - - // First run: populate both WALs. qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) - gr1 := qc1.QC().GlobalRange(committee) - - dw1 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state1 := utils.OrPanic1(NewState(&Config{Committee: committee}, dw1)) - require.NoError(t, state1.PushQC(ctx, qc1, blocks1)) - require.NoError(t, dw1.CommitQCs.PersistQC(qc1)) - for i, n := 0, gr1.First; n < gr1.Next; n++ { - require.NoError(t, dw1.Blocks.PersistBlock(n, blocks1[i])) - i++ - } - require.NoError(t, dw1.Close()) - - // Simulate crash: delete blocks WAL directory, keep QCs WAL. - require.NoError(t, os.RemoveAll(filepath.Join(dir, "globalblocks"))) - // Second run: only QCs WAL survives. - dw2 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state2 := utils.OrPanic1(NewState(&Config{Committee: committee}, dw2)) - - // QCs loaded, blocks empty. The state needs blocks re-pushed. - // Without the cursor sync fix, PushBlock here would fail with - // "out of sequence" because the blocks persister cursor is 0 - // but inner.nextBlock was advanced by QC data. - for i, n := 0, gr1.First; n < gr1.Next; n++ { - require.NoError(t, state2.PushBlock(ctx, n, blocks1[i])) - i++ - } - - for n := gr1.First; n < gr1.Next; n++ { - got, err := state2.TryBlock(n) - require.NoError(t, err) - require.NotNil(t, got) - } + state, err := NewState(&Config{Committee: committee}, utils.None[types.BlockDB]()) + require.NoError(t, err) - // State should accept the next QC normally. - qc2, blocks2 := TestCommitQC(rng, committee, keys, utils.Some(qc1.QC())) - require.NoError(t, state2.PushQC(ctx, qc2, blocks2)) - require.Equal(t, qc2.QC().GlobalRange(committee).Next, state2.NextBlock()) - require.NoError(t, dw2.Close()) + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + s.SpawnBgNamed("state", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) + if err := state.PushQC(ctx, qc1, blocks1); err != nil { + return err + } + // Verify data is accessible (no panic, no error). + gr := qc1.QC().GlobalRange(committee) + for n := gr.First; n < gr.Next; n++ { + if _, err := state.Block(ctx, n); err != nil { + return err + } + } + return nil + })) } -func TestReconcileCase4Normal(t *testing.T) { - t.Log("Reconcile case 4: Normal (a=X, bX). -// Blocks WAL was truncated further than QCs during a crash between -// the two parallel TruncateBefore calls. -func TestReconcileCase5BlocksAhead(t *testing.T) { - t.Log("Reconcile case 5: Prune crash, blocks ahead (a>X)") - rng := utils.TestRng() - committee, keys := types.GenCommittee(rng, 3) - dir := t.TempDir() - - qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) - qc2, blocks2 := TestCommitQC(rng, committee, keys, utils.Some(qc1.QC())) - gr1 := qc1.QC().GlobalRange(committee) - gr2 := qc2.QC().GlobalRange(committee) + state := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) + require.NoError(t, state.PushQC(ctx, qc1, blocks1)) + require.NoError(t, state.PushQC(ctx, qc2, blocks2)) + require.NoError(t, state.PushQC(ctx, qc3, blocks3)) - // Persist both QCs and all blocks. - dw := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - require.NoError(t, dw.CommitQCs.PersistQC(qc1)) - require.NoError(t, dw.CommitQCs.PersistQC(qc2)) - allBlocks := append(blocks1, blocks2...) - for i, n := 0, gr1.First; n < gr2.Next; n++ { - require.NoError(t, dw.Blocks.PersistBlock(n, allBlocks[i])) - i++ + // Execute all blocks so they are eligible for pruning. + runCtx, cancel := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- state.Run(runCtx) }() + for n := gr1.First; n < gr3.Next; n++ { + require.NoError(t, state.PushAppHash(ctx, n, types.GenAppHash(rng))) } - // Simulate crash: blocks truncated to gr2.First but QCs not. - require.NoError(t, dw.Blocks.TruncateBefore(gr2.First)) - require.NoError(t, dw.Close()) + cancel() + <-done - // Reopen: blocks start at gr2.First, QCs start at gr1.First. - // Reconcile should truncate QCs to match blocks. - dw2 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state := utils.OrPanic1(NewState(&Config{Committee: committee}, dw2)) + // Prune qc1 entirely (keep from qc2 onward). + require.NoError(t, state.PruneBefore(gr2.First)) + // PruneBefore keeps one sentinel block at firstToKeep-1; capture it. + var survivingBlock types.GlobalBlockNumber for inner := range state.inner.Lock() { - require.Equal(t, gr2.First, inner.first) + survivingBlock = inner.first } - // Blocks in qc2's range should be available. - for n := gr2.First; n < gr2.Next; n++ { + for n := gr1.First; n < survivingBlock; n++ { + _, err := state.TryBlock(n) + require.ErrorIs(t, err, ErrPruned) + } + // survivingBlock itself and everything from gr2.First onward must be readable. + for n := survivingBlock; n < gr3.Next; n++ { got, err := state.TryBlock(n) require.NoError(t, err) require.NotNil(t, got) } - require.NoError(t, dw2.Close()) } -// TestStateRecoverySkipsStaleBlocks verifies that blocks loaded from the WAL -// that fall before the first QC range are not inserted into the state map. -// This can happen when the QCs WAL is pruned but the blocks WAL still has -// older entries (e.g., a crash between the two TruncateBefore calls). -func TestReconcileCase6QCsAhead(t *testing.T) { - t.Log("Reconcile case 6: Prune crash, QCs ahead (a= nextQC are ignored. This can happen when blocks are -// persisted in parallel with QCs and we crash before QCs catch up. -func TestReconcileCase7BlocksPastQCs(t *testing.T) { - t.Log("Reconcile case 7: Persist crash, blocks past QCs (b>=Y)") +// TestRecoveryBlocksBehind verifies recovery when QCs cover more range than +// blocks (e.g. a crash during block writes). Blocks up to the crash point are +// available; the rest are re-fetched via PushBlock. +func TestRecoveryBlocksBehind(t *testing.T) { + ctx := t.Context() rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) dir := t.TempDir() - // Build 2 sequential QCs. qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) qc2, blocks2 := TestCommitQC(rng, committee, keys, utils.Some(qc1.QC())) gr1 := qc1.QC().GlobalRange(committee) gr2 := qc2.QC().GlobalRange(committee) - // Persist only qc1 to QCs WAL but persist ALL blocks (qc1 + qc2) to blocks WAL. - // This simulates blocks being persisted ahead of QCs. - dw1 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - require.NoError(t, dw1.CommitQCs.PersistQC(qc1)) - allBlocks := append(blocks1, blocks2...) - for i, n := 0, gr1.First; n < gr2.Next; n++ { - require.NoError(t, dw1.Blocks.PersistBlock(n, allBlocks[i])) + // Write both QCs but only qc1's blocks (simulate crash before qc2 blocks). + db1 := newTestBlockDB(t, dir) + require.NoError(t, db1.WriteQC(gr1.First, gr1.Next, qc1)) + require.NoError(t, db1.WriteQC(gr2.First, gr2.Next, qc2)) + for i, n := 0, gr1.First; n < gr1.Next; n++ { + require.NoError(t, db1.WriteBlock(n, blocks1[i])) i++ } - require.NoError(t, dw1.Close()) + require.NoError(t, db1.Flush()) + require.NoError(t, db1.Close()) - // On recovery, only blocks within qc1's range should be loaded. - // Blocks in qc2's range have no QC and should be ignored. - dw2 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state2 := utils.OrPanic1(NewState(&Config{Committee: committee}, dw2)) + // Recovery: both QCs loaded, but only qc1's blocks. + db2 := newTestBlockDB(t, dir) + state2 := newTestState(t, &Config{Committee: committee}, db2) - // Blocks in qc1's range should be available. for n := gr1.First; n < gr1.Next; n++ { got, err := state2.TryBlock(n) require.NoError(t, err) require.NotNil(t, got) } - - // Blocks in qc2's range should NOT be in the state (no QC for them). for n := gr2.First; n < gr2.Next; n++ { _, err := state2.TryBlock(n) require.ErrorIs(t, err, ErrNotFound) } - // State should still accept qc2 normally. - require.NoError(t, state2.PushQC(t.Context(), qc2, blocks2)) + // Re-push qc2's blocks to fill the gap. + for i, n := 0, gr2.First; n < gr2.Next; n++ { + require.NoError(t, state2.PushBlock(ctx, n, blocks2[i])) + i++ + } require.Equal(t, gr2.Next, state2.NextBlock()) - require.NoError(t, dw2.Close()) } -// TestReconcileTruncatesBlocksTail verifies that blocks persisted without -// corresponding QCs are removed during WAL reconciliation. This prevents -// stale blocks from blocking new (different) blocks at the same positions. -func TestReconcileCase7BlocksTail(t *testing.T) { - t.Log("Reconcile case 7: Persist crash, tail truncation with re-push") - ctx := t.Context() +// TestRecoveryPartialQCPrefix verifies that when the QC spans a wider range +// than the available blocks (the block prefix was pruned), recovery sets first +// from the blocks iterator (not the QC iterator). +func TestRecoveryPartialQCPrefix(t *testing.T) { rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) dir := t.TempDir() - // Build 2 sequential QCs. qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) - qc2, blocks2 := TestCommitQC(rng, committee, keys, utils.Some(qc1.QC())) gr1 := qc1.QC().GlobalRange(committee) - gr2 := qc2.QC().GlobalRange(committee) + if gr1.Next-gr1.First < 3 { + t.Skip("need at least 3 blocks in QC range to test split") + } - // Persist qc1 to both WALs, but only blocks (not QC) for qc2. - // This simulates a crash during parallel persistence in runPersist. - dw1 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - require.NoError(t, dw1.CommitQCs.PersistQC(qc1)) - allBlocks := append(blocks1, blocks2...) - for i, n := 0, gr1.First; n < gr2.Next; n++ { - require.NoError(t, dw1.Blocks.PersistBlock(n, allBlocks[i])) + // Write the QC for the full range, but write blocks only from mid onwards. + // This is a valid write sequence: QC first, then blocks in ascending order + // starting from mid. It simulates a DB where the block prefix [gr1.First, mid) + // was pruned and physically removed. + mid := gr1.First + (gr1.Next-gr1.First)/2 + db1 := newTestBlockDB(t, dir) + require.NoError(t, db1.WriteQC(gr1.First, gr1.Next, qc1)) + for i, n := 0, gr1.First; n < gr1.Next; n++ { + if n >= mid { + require.NoError(t, db1.WriteBlock(n, blocks1[i])) + } i++ } - require.NoError(t, dw1.Close()) - - // Reopen: reconcile should truncate the blocks tail (qc2's blocks). - dw2 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) + require.NoError(t, db1.Flush()) + require.NoError(t, db1.Close()) - // Blocks persister cursor should now match QCs range. - require.Equal(t, dw2.CommitQCs.Next(), dw2.Blocks.Next()) + // Recovery should use blocks as golden: first == mid, not gr1.First. + db2 := newTestBlockDB(t, dir) + state2 := newTestState(t, &Config{Committee: committee}, db2) - state := utils.OrPanic1(NewState(&Config{Committee: committee}, dw2)) - - // qc1's blocks should be available. - for n := gr1.First; n < gr1.Next; n++ { - got, err := state.TryBlock(n) + for inner := range state2.inner.Lock() { + require.Equal(t, mid, inner.first, + "first should be where blocks start, not where QC starts") + require.Equal(t, gr1.Next, inner.nextQC, + "QC should still cover the full range") + // QC entries below first must be absent — they are unreachable via + // pruneFirst and should have been trimmed during recovery. + for n := gr1.First; n < mid; n++ { + _, ok := inner.qcs[n] + require.False(t, ok, "orphaned QC entry at %d should have been trimmed", n) + } + } + for n := gr1.First; n < mid; n++ { + _, err := state2.TryBlock(n) + require.ErrorIs(t, err, ErrPruned) + } + for n := mid; n < gr1.Next; n++ { + got, err := state2.TryBlock(n) require.NoError(t, err) require.NotNil(t, got) } - - // Now push qc2 with its blocks — should succeed because stale blocks - // were removed and the persister cursor was reset. - require.NoError(t, state.PushQC(ctx, qc2, blocks2)) - require.Equal(t, gr2.Next, state.NextBlock()) - require.NoError(t, dw2.Close()) } -// TestStateRecoveryBlocksBehindQCs verifies recovery when QCs cover a wider -// range than blocks (e.g. crash during block persistence). Blocks up to -// blocksEnd are available; the rest must be re-fetched via PushBlock. -func TestReconcileCase8BlocksBehind(t *testing.T) { - t.Log("Reconcile case 8: QCs ahead normal (b= inner.first +// set by the QC pass, so the "block predates first QC start" guard never fires. +func TestRecoveryAfterPruneNoGC(t *testing.T) { rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) dir := t.TempDir() - // Build 2 sequential QCs. qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) qc2, blocks2 := TestCommitQC(rng, committee, keys, utils.Some(qc1.QC())) gr1 := qc1.QC().GlobalRange(committee) gr2 := qc2.QC().GlobalRange(committee) - // Persist both QCs but only qc1's blocks to WALs. - dw1 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - require.NoError(t, dw1.CommitQCs.PersistQC(qc1)) - require.NoError(t, dw1.CommitQCs.PersistQC(qc2)) - for i, n := 0, gr1.First; n < gr1.Next; n++ { - require.NoError(t, dw1.Blocks.PersistBlock(n, blocks1[i])) - i++ - } - require.NoError(t, dw1.Close()) + // Write both QCs and all their blocks to the DB. + cfg1, err := littblock.DefaultConfig(dir) + require.NoError(t, err) + cfg1.Retention = time.Nanosecond + db1, err := littblock.NewBlockDB(cfg1) + require.NoError(t, err) + writeToBlockDB(t, db1, committee, []*types.FullCommitQC{qc1, qc2}, [][]*types.Block{blocks1, blocks2}) - // On recovery: both QCs loaded, but only qc1's blocks. - dw2 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state2 := utils.OrPanic1(NewState(&Config{Committee: committee}, dw2)) + // Prune qc1's range. GC is NOT called — pruned entries remain on disk. + require.NoError(t, db1.PruneBefore(gr2.First)) + require.NoError(t, db1.Close()) - // qc1's blocks should be available. - for n := gr1.First; n < gr1.Next; n++ { - got, err := state2.TryBlock(n) - require.NoError(t, err) - require.NotNil(t, got) - } + // Reopen the same dir without ForceGC — pruned entries may still be present. + cfg2, err := littblock.DefaultConfig(dir) + require.NoError(t, err) + cfg2.Retention = time.Nanosecond + db2, err := littblock.NewBlockDB(cfg2) + require.NoError(t, err) + t.Cleanup(func() { _ = db2.Close() }) - // qc2's blocks are missing — not yet available. - for n := gr2.First; n < gr2.Next; n++ { - _, err := state2.TryBlock(n) - require.ErrorIs(t, err, ErrNotFound) - } + // NewState must succeed — below-watermark blocks never outlive their QCs + // because blocks and QCs share the same GC filter in littblock. Without GC, + // all entries are still present and recovery treats the DB as unpruned. + // This is the PruneBefore-without-GC path: the watermark advanced but + // physical reclamation has not happened yet, so the DB looks like a + // fresh DB containing all data from qc1 and qc2. + state := newTestState(t, &Config{Committee: committee}, db2) - // Re-push qc2's blocks to fill the gap. - for i, n := 0, gr2.First; n < gr2.Next; n++ { - require.NoError(t, state2.PushBlock(ctx, n, blocks2[i])) - i++ + // Without GC all data is still present; qc1 and qc2 blocks are accessible. + for n := gr1.First; n < gr2.Next; n++ { + _, err := state.TryBlock(n) + require.NoError(t, err) } - require.Equal(t, gr2.Next, state2.NextBlock()) - require.NoError(t, dw2.Close()) } -// TestRecoveryWithPartialQCPrefix verifies that after per-block pruning -// splits a QC range, recovery sets first from blocks (not QCs), and the -// node can serve surviving blocks without re-fetching pruned ones. -func TestReconcilePartialQCPrefix(t *testing.T) { - // Reconcile: partial QC prefix from per-block pruning - t.Log("Reconcile: partial QC prefix from per-block pruning") +// TestRecoveryQCsNoBlocks verifies that NewState succeeds when the DB contains +// QCs but no blocks (crash between QC flush and block writes). The state +// cursor sits at the QC start with nextBlock == first and no block data. +func TestRecoveryQCsNoBlocks(t *testing.T) { rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) dir := t.TempDir() - // Build one QC with enough blocks to split. - qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) + qc1, _ := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) gr1 := qc1.QC().GlobalRange(committee) - if gr1.Next-gr1.First < 3 { - t.Skip("need at least 3 blocks in QC range to test split") - } - // Persist QC and all blocks. - dw := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - require.NoError(t, dw.CommitQCs.PersistQC(qc1)) - for i, n := 0, gr1.First; n < gr1.Next; n++ { - require.NoError(t, dw.Blocks.PersistBlock(n, blocks1[i])) - i++ - } - // Simulate per-block pruning: truncate blocks prefix, keep QC intact. - // This creates the split: QC covers [gr1.First, gr1.Next) but blocks - // start at mid. - mid := gr1.First + (gr1.Next-gr1.First)/2 - require.NoError(t, dw.Blocks.TruncateBefore(mid)) - require.NoError(t, dw.Close()) + db1 := newTestBlockDB(t, dir) + require.NoError(t, db1.WriteQC(gr1.First, gr1.Next, qc1)) + require.NoError(t, db1.Flush()) + require.NoError(t, db1.Close()) - // Recovery should use blocks as golden, not try to re-fetch pruned prefix. - dw2 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state2 := utils.OrPanic1(NewState(&Config{Committee: committee}, dw2)) + db2 := newTestBlockDB(t, dir) + state2 := newTestState(t, &Config{Committee: committee}, db2) - // first should be at mid (where blocks start), not gr1.First (where QC starts). for inner := range state2.inner.Lock() { - require.Equal(t, mid, inner.first, - "first should be where blocks start, not where QC starts") - require.Equal(t, gr1.Next, inner.nextQC, - "QC should still cover the full range") + require.Equal(t, gr1.First, inner.first) + require.Equal(t, gr1.Next, inner.nextQC) + require.Equal(t, gr1.First, inner.nextBlock) } - - // Blocks before mid should be pruned, not ErrNotFound. - for n := gr1.First; n < mid; n++ { + for n := gr1.First; n < gr1.Next; n++ { _, err := state2.TryBlock(n) - require.ErrorIs(t, err, ErrPruned) - } - - // Blocks at mid and above should be available. - for n := mid; n < gr1.Next; n++ { - got, err := state2.TryBlock(n) - require.NoError(t, err) - require.NotNil(t, got) + require.ErrorIs(t, err, ErrNotFound) } - require.NoError(t, dw2.Close()) } -// TestStateRejectsBlockGapInWAL verifies that NewState returns an error -// if loaded blocks have a gap (defense in depth for future storage backends). -func TestReconcileBlockGap(t *testing.T) { - // Reconcile: block gap in WAL detected (defense in depth) - t.Log("Reconcile: block gap in WAL detected (defense in depth)") +// TestRecoveryBlockGap verifies that NewState returns an error when blocks in +// BlockDB are not contiguous. WriteBlock only enforces strictly-ascending and +// QC coverage, not continuity, so a gap can arise from corruption. +func TestRecoveryBlockGap(t *testing.T) { rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) dir := t.TempDir() qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) gr1 := qc1.QC().GlobalRange(committee) - if gr1.Next-gr1.First < 3 { - t.Skip("need at least 3 blocks in QC range to test gap") - } - // Persist QC to real WAL so it loads on restart. - dw := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - require.NoError(t, dw.CommitQCs.PersistQC(qc1)) - require.NoError(t, dw.Close()) - - // Reopen and inject blocks with a gap via test helper. - dw2 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - dw2.Blocks.SetLoadedForTest([]persist.LoadedGlobalBlock{ - {Number: gr1.First, Block: blocks1[0]}, - // skip gr1.First+1 - {Number: gr1.First + 2, Block: blocks1[2]}, - }) + // TestCommitQC generates 10 global blocks, so the range is always wide + // enough to skip one block in the middle. + mid := gr1.First + (gr1.Next-gr1.First)/2 - _, err := NewState(&Config{Committee: committee}, dw2) + db1 := newTestBlockDB(t, dir) + require.NoError(t, db1.WriteQC(gr1.First, gr1.Next, qc1)) + for i, n := 0, gr1.First; n < gr1.Next; n++ { + if n != mid { + require.NoError(t, db1.WriteBlock(n, blocks1[i])) + } + i++ + } + require.NoError(t, db1.Flush()) + require.NoError(t, db1.Close()) + + db2 := newTestBlockDB(t, dir) + _, err := NewState(&Config{Committee: committee}, utils.Some[types.BlockDB](db2)) require.Error(t, err) - require.Contains(t, err.Error(), "block gap") - require.NoError(t, dw2.Close()) + require.Contains(t, err.Error(), "block gap in BlockDB") } -// ── Non-reconcile tests ─────────────────────────────────────────────── +// ── Non-recovery tests ──────────────────────────────────────────────── // TestPruningKeepsLastQCRange verifies that pruning never removes the last -// QC range, ensuring WALs are never empty and inner.first is recoverable -// on restart. +// retained block, and that a restart from a BlockDB with only that block +// recovers correctly. func TestPruningKeepsLastQCRange(t *testing.T) { ctx := t.Context() rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) - dir := t.TempDir() - dw := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state := utils.OrPanic1(NewState(&Config{Committee: committee}, dw)) - - // Push one QC with blocks and execute all of them. qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) gr1 := qc1.QC().GlobalRange(committee) - require.NoError(t, state.PushQC(ctx, qc1, blocks1)) - // Persist and execute. + // In-memory: push QC, execute all blocks, then prune everything. + state1 := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) + require.NoError(t, state1.PushQC(ctx, qc1, blocks1)) + runCtx, cancel := context.WithCancel(ctx) done := make(chan error, 1) - go func() { done <- state.Run(runCtx) }() + go func() { done <- state1.Run(runCtx) }() for n := gr1.First; n < gr1.Next; n++ { - require.NoError(t, state.PushAppHash(ctx, n, types.GenAppHash(rng))) + require.NoError(t, state1.PushAppHash(ctx, n, types.GenAppHash(rng))) } cancel() <-done - // Try to prune everything — at least one block should survive. - state.PruneBefore(gr1.Next) - for inner := range state.inner.Lock() { - require.Less(t, inner.first, gr1.Next, - "pruning should keep at least one block") + require.NoError(t, state1.PruneBefore(gr1.Next)) + var survivingBlock types.GlobalBlockNumber + for inner := range state1.inner.Lock() { + survivingBlock = inner.first + require.Less(t, survivingBlock, gr1.Next, "pruning should keep at least one block") + } + for n := gr1.First; n < survivingBlock; n++ { + _, err := state1.TryBlock(n) + require.ErrorIs(t, err, ErrPruned) } - // Restart from WALs — inner.first should be recoverable. - require.NoError(t, dw.Close()) - dw2 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state2 := utils.OrPanic1(NewState(&Config{Committee: committee}, dw2)) + // Restart: build a fresh BlockDB containing only the surviving block (as GC + // would leave it) and verify NewState correctly sets first = survivingBlock. + dir := t.TempDir() + db := newTestBlockDB(t, dir) + require.NoError(t, db.WriteQC(gr1.First, gr1.Next, qc1)) + require.NoError(t, db.WriteBlock(survivingBlock, blocks1[survivingBlock-gr1.First])) + require.NoError(t, db.Flush()) + require.NoError(t, db.Close()) + + db2 := newTestBlockDB(t, dir) + state2 := newTestState(t, &Config{Committee: committee}, db2) for inner := range state2.inner.Lock() { - require.GreaterOrEqual(t, inner.first, gr1.First, - "after restart, first should be recoverable from WAL") + require.Equal(t, survivingBlock, inner.first, + "after restart, first should match the surviving block") } - require.NoError(t, dw2.Close()) } -// TestPruningWithPartialQCRange verifies that per-block pruning can split -// a QC range, and recovery handles it correctly by using blocks as the -// golden source for inner.first. +// TestPruningWithPartialQCRange verifies per-block pruning across QC ranges: +// pruning advances first in-memory, and a restart from a post-GC BlockDB +// (with only the last surviving block) recovers correctly. func TestPruningWithPartialQCRange(t *testing.T) { ctx := t.Context() rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) - dir := t.TempDir() - dw := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state := utils.OrPanic1(NewState(&Config{Committee: committee}, dw)) - - // Push two QCs so we have two distinct QC ranges. qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) qc2, blocks2 := TestCommitQC(rng, committee, keys, utils.Some(qc1.QC())) gr1 := qc1.QC().GlobalRange(committee) gr2 := qc2.QC().GlobalRange(committee) - require.NoError(t, state.PushQC(ctx, qc1, blocks1)) - require.NoError(t, state.PushQC(ctx, qc2, blocks2)) - // Run to persist, then execute all blocks. + // In-memory: push 2 QCs, execute, then prune. + state1 := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) + require.NoError(t, state1.PushQC(ctx, qc1, blocks1)) + require.NoError(t, state1.PushQC(ctx, qc2, blocks2)) + runCtx, cancel := context.WithCancel(ctx) done := make(chan error, 1) - go func() { done <- state.Run(runCtx) }() + go func() { done <- state1.Run(runCtx) }() for n := gr1.First; n < gr2.Next; n++ { - require.NoError(t, state.PushAppHash(ctx, n, types.GenAppHash(rng))) + require.NoError(t, state1.PushAppHash(ctx, n, types.GenAppHash(rng))) } cancel() <-done - // Prune into the middle of qc1's range. Per-block pruning allows this. + // Per-block prune into middle of qc1's range. midQC1 := gr1.First + (gr1.Next-gr1.First)/2 if midQC1 > gr1.First+1 { - state.PruneBefore(midQC1) - for inner := range state.inner.Lock() { + require.NoError(t, state1.PruneBefore(midQC1)) + for inner := range state1.inner.Lock() { require.Greater(t, inner.first, gr1.First, "per-block pruning should advance past gr1.First") } } // Prune past qc1 entirely. - state.PruneBefore(gr2.Next) - - // Restart — recovery uses blocks as golden, skips partial QC prefix. - require.NoError(t, dw.Close()) - dw2 := utils.OrPanic1(NewDataWAL(utils.Some(dir), committee)) - state2 := utils.OrPanic1(NewState(&Config{Committee: committee}, dw2)) + require.NoError(t, state1.PruneBefore(gr2.Next)) + var survivingBlock types.GlobalBlockNumber + for inner := range state1.inner.Lock() { + survivingBlock = inner.first + require.GreaterOrEqual(t, survivingBlock, gr2.First) + require.Less(t, survivingBlock, gr2.Next, "at least one block should survive") + } + for n := gr1.First; n < survivingBlock; n++ { + _, err := state1.TryBlock(n) + require.ErrorIs(t, err, ErrPruned) + } - // first should be recoverable and blocks should be available. + // Restart: build a fresh BlockDB containing only the surviving block from qc2 + // (qc1 is pruned entirely) and verify NewState recovers correctly. + dir := t.TempDir() + db := newTestBlockDB(t, dir) + require.NoError(t, db.WriteQC(gr2.First, gr2.Next, qc2)) + require.NoError(t, db.WriteBlock(survivingBlock, blocks2[survivingBlock-gr2.First])) + require.NoError(t, db.Flush()) + require.NoError(t, db.Close()) + + db2 := newTestBlockDB(t, dir) + state2 := newTestState(t, &Config{Committee: committee}, db2) for inner := range state2.inner.Lock() { - require.GreaterOrEqual(t, inner.first, gr2.First, - "after restart, first should be at or past gr2.First") - require.Less(t, inner.first, gr2.Next, - "at least one block should survive") + require.Equal(t, survivingBlock, inner.first, + "after restart, first should match the surviving block") } - require.NoError(t, dw2.Close()) } // TestRunPruningEmptyState verifies that runPruning does not panic when @@ -1075,10 +923,10 @@ func TestRunPruningEmptyState(t *testing.T) { rng := utils.TestRng() committee, _ := types.GenCommittee(rng, 3) - state := utils.OrPanic1(NewState(&Config{ + state := newTestState(t, &Config{ Committee: committee, PruneAfter: utils.Some(time.Duration(0)), - }, utils.OrPanic1(NewDataWAL(utils.None[string](), committee)))) + }, newTestBlockDB(t, t.TempDir())) // Run briefly — runPruning should not panic on empty state. ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) @@ -1092,9 +940,7 @@ func TestPushBlockWaitsForQC(t *testing.T) { rng := utils.TestRng() committee, keys := types.GenCommittee(rng, 3) - state := utils.OrPanic1(NewState(&Config{ - Committee: committee, - }, utils.OrPanic1(NewDataWAL(utils.None[string](), committee)))) + state := newTestState(t, &Config{Committee: committee}, newTestBlockDB(t, t.TempDir())) // Push first QC covering [0, N). qc1, blocks1 := TestCommitQC(rng, committee, keys, utils.None[*types.CommitQC]()) diff --git a/sei-tendermint/internal/autobahn/producer/mempool_test.go b/sei-tendermint/internal/autobahn/producer/mempool_test.go index b0d7987797..6d8665ea00 100644 --- a/sei-tendermint/internal/autobahn/producer/mempool_test.go +++ b/sei-tendermint/internal/autobahn/producer/mempool_test.go @@ -33,6 +33,14 @@ type txSpec struct { Payload [32]byte } +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 +} + func (tx *txSpec) encode() []byte { return utils.OrPanic1(binary.Append(nil, binary.BigEndian, tx)) } @@ -229,10 +237,7 @@ func (env *testEnv) Run(ctx context.Context) error { func newTestEnv(rng utils.Rng, cfg *Config, app *proxy.Proxy) *testEnv { committee, keys := types.GenCommittee(rng, 1) - dataState := utils.OrPanic1(data.NewState( - &data.Config{Committee: committee}, - utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee)), - )) + dataState := newTestDataState(committee) consensusState := utils.OrPanic1(consensus.NewState(&consensus.Config{ Key: keys[0], ViewTimeout: func(types.View) time.Duration { return time.Hour }, diff --git a/sei-tendermint/internal/p2p/giga/avail_test.go b/sei-tendermint/internal/p2p/giga/avail_test.go index 2406af6350..5027c3a237 100644 --- a/sei-tendermint/internal/p2p/giga/avail_test.go +++ b/sei-tendermint/internal/p2p/giga/avail_test.go @@ -37,8 +37,9 @@ func TestAvailClientServer(t *testing.T) { s.SpawnBg(func() error { return env.Run(ctx) }) t.Log("Spawn a fake unconnected node0 to generate some conflicting blocks and push them to node2.") fakeNode0, err := consensus.NewState(&consensus.Config{ - Key: keys[0], - ViewTimeout: defaultViewTimeout, + Key: keys[0], + ViewTimeout: defaultViewTimeout, + PersistentStateDir: utils.Some(t.TempDir()), }, nodes[0].data) if err != nil { return fmt.Errorf("consensus.NewState(): %w", err) diff --git a/sei-tendermint/internal/p2p/giga/data_test.go b/sei-tendermint/internal/p2p/giga/data_test.go index f5249517b9..ab90cb44a6 100644 --- a/sei-tendermint/internal/p2p/giga/data_test.go +++ b/sei-tendermint/internal/p2p/giga/data_test.go @@ -24,7 +24,11 @@ type testNode struct { func defaultViewTimeout(view types.View) time.Duration { return time.Hour } func newTestNode(committee *types.Committee, cfg *consensus.Config) *testNode { - dataState := utils.OrPanic1(data.NewState(&data.Config{Committee: committee}, utils.OrPanic1(data.NewDataWAL(utils.None[string](), committee)))) + cfg.PersistentStateDir = utils.None[string]() + dataState, err := data.NewState(&data.Config{Committee: committee}, utils.None[types.BlockDB]()) + if err != nil { + panic(fmt.Sprintf("data.NewState: %v", err)) + } consensusState, err := consensus.NewState(cfg, dataState) if err != nil { panic(fmt.Sprintf("consensus.NewState(): %v", err)) diff --git a/sei-tendermint/internal/p2p/giga_router.go b/sei-tendermint/internal/p2p/giga_router.go index 8ec60a977d..1cc77d14d8 100644 --- a/sei-tendermint/internal/p2p/giga_router.go +++ b/sei-tendermint/internal/p2p/giga_router.go @@ -32,8 +32,9 @@ type GigaRouterCommonConfig struct { DialInterval time.Duration ValidatorAddrs map[atypes.PublicKey]GigaNodeAddr GenDoc *types.GenesisDoc - // PersistentStateDir is the on-disk root for the data WAL (and the - // validator's consensus persister in a sibling subdir). None ⇒ in-memory. + // PersistentStateDir is the on-disk root for durable state (BlockDB, + // hashvault, and the validator's consensus persister in sibling subdirs). + // If None, persistence is disabled and the node runs fully in-memory. PersistentStateDir utils.Option[string] // App is the ABCI proxy executeBlock drives. NewGigaValidatorRouter // also passes it to producer.NewState so the producer's internal diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go index 4ee2c58d11..f35bf5fed4 100644 --- a/sei-tendermint/internal/p2p/giga_router_common.go +++ b/sei-tendermint/internal/p2p/giga_router_common.go @@ -11,6 +11,7 @@ import ( "sync/atomic" "github.com/ethereum/go-ethereum/common" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/block/littblock" "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/hashvault" abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" @@ -48,27 +49,21 @@ type gigaRouterCommon struct { // by one or two under contention but never over-accepts. inboundFullnodeCount atomic.Int64 inboundFullnodeCap int64 - - // hashVault is the app-hash equivocation guard. runExecute owns its lifecycle: it builds the - // vault on entry (durable under PersistentStateDir, or a no-op when disabled / in-memory) and - // closes it on exit. - hashVault hashvault.HashVault } -// buildDataState validates the common config and constructs the data -// layer (committee, WAL, State) shared by both giga constructors. -// -// TODO(autobahn): once sei-db/ledger_db/block.BlockDB has a writer wired -// (see BlockByNumber's TODO), the data WAL is redundant. -func buildDataState(cfg *GigaRouterCommonConfig) (*data.State, error) { +// buildDataState validates the common config, constructs the committee, opens +// BlockDB (if PersistentStateDir is set), and returns an initialised data.State +// alongside the raw BlockDB handle. The caller owns blockDB and must close it +// after Run returns; data.State never closes it. +func buildDataState(cfg *GigaRouterCommonConfig) (*data.State, atypes.BlockDB, error) { if cfg.GenDoc.InitialHeight < 1 { - return nil, fmt.Errorf("GenDoc.InitialHeight = %v, want >=1", cfg.GenDoc.InitialHeight) + return nil, nil, fmt.Errorf("GenDoc.InitialHeight = %v, want >=1", cfg.GenDoc.InitialHeight) } if cfg.DialInterval <= 0 { - return nil, fmt.Errorf("GigaRouterCommonConfig.DialInterval = %v, want > 0", cfg.DialInterval) + return nil, nil, fmt.Errorf("GigaRouterCommonConfig.DialInterval = %v, want > 0", cfg.DialInterval) } if cfg.MaxInboundFullnodePeers < 0 || cfg.MaxInboundFullnodePeers > maxInboundFullnodePeers { - return nil, fmt.Errorf("GigaRouterCommonConfig.MaxInboundFullnodePeers = %v, want 0..%v", cfg.MaxInboundFullnodePeers, maxInboundFullnodePeers) + return nil, nil, fmt.Errorf("GigaRouterCommonConfig.MaxInboundFullnodePeers = %v, want 0..%v", cfg.MaxInboundFullnodePeers, maxInboundFullnodePeers) } committee, err := atypes.NewRoundRobinElection( slices.Collect(maps.Keys(cfg.ValidatorAddrs)), @@ -76,17 +71,31 @@ func buildDataState(cfg *GigaRouterCommonConfig) (*data.State, error) { cfg.GenDoc.GenesisTime, ) if err != nil { - return nil, fmt.Errorf("atypes.NewRoundRobinElection(): %w", err) + return nil, nil, fmt.Errorf("atypes.NewRoundRobinElection(): %w", err) } - dataWAL, err := data.NewDataWAL(cfg.PersistentStateDir, committee) - if err != nil { - return nil, fmt.Errorf("data.NewDataWAL(): %w", err) + var blockDB atypes.BlockDB + if dir, ok := cfg.PersistentStateDir.Get(); ok { + blockCfg, err := littblock.DefaultConfig(filepath.Join(dir, "blockdb")) + if err != nil { + return nil, nil, fmt.Errorf("littblock.DefaultConfig: %w", err) + } + blockDB, err = littblock.NewBlockDB(blockCfg) + if err != nil { + return nil, nil, fmt.Errorf("open BlockDB: %w", err) + } } - dataState, err := data.NewState(&data.Config{Committee: committee}, dataWAL) + blockDBOpt := utils.None[atypes.BlockDB]() + if blockDB != nil { + blockDBOpt = utils.Some(blockDB) + } + ds, err := data.NewState(&data.Config{Committee: committee}, blockDBOpt) if err != nil { - return nil, fmt.Errorf("data.NewState(): %w", err) + if blockDB != nil { + _ = blockDB.Close() + } + return nil, nil, fmt.Errorf("data.NewState: %w", err) } - return dataState, nil + return ds, blockDB, nil } func (r *gigaRouterCommon) LastCommittedBlockNumber() int64 { @@ -114,14 +123,6 @@ func (r *gigaRouterCommon) MaxGasEstimatedPerBlock() uint64 { // evmrpc does not read them on the receipt path. If gb.Header is nil // BlockID.Hash also stays empty; if gb.Payload is nil Block.Data.Txs // stays empty (see the malformed-block handling below). -// -// TODO(autobahn): switch this to read from sei-db/ledger_db/block.BlockDB -// once a writer is wired (e.g. from app.FinalizeBlocker or executeBlock). -// Today no production code calls BlockDB.WriteBlock, so Autobahn's in-memory -// data.State is the only place a full block lives — but it's pruned per -// Sei's RetainHeight and exposes only a height index (no GetBlockByHash). -// BlockDB has the right shape (height + hash indexes, async pruning) and -// is the long-term home for this read path. func (r *gigaRouterCommon) BlockByNumber(ctx context.Context, n atypes.GlobalBlockNumber) (*coretypes.ResultBlock, error) { gb, err := r.data.GlobalBlock(ctx, n) if err != nil { @@ -143,15 +144,11 @@ func (r *gigaRouterCommon) BlockByNumber(ctx context.Context, n atypes.GlobalBlo // (same translation as BlockByNumber). Matches CometBFT semantics for // unknown hashes: returns &ResultBlock{Block: nil} with no error. // -// Lookup-and-construct happens under a single data.State lock acquire, so -// the returned block matches the requested hash atomically. Hashes below -// the pruning watermark are not indexed and read as "unknown". Wrong-size -// inputs are rejected at the call site (env.BlockByHash) so this method -// can stay strongly typed on atypes.BlockHeaderHash. -// -// TODO(autobahn): replace this with a direct read from -// sei-db/ledger_db/block.BlockDB.GetBlockByHash once a writer is wired into -// block execution. The data.State-side index can also go away at that point. +// The lookup delegates to data.State.GlobalBlockByHash, which reads from an +// in-memory hash index rebuilt from BlockDB at construction time. Hashes not +// yet seen or below the prune watermark are read as "unknown". Wrong-size inputs are rejected +// at the call site (env.BlockByHash) so this method can stay strongly typed +// on atypes.BlockHeaderHash. func (r *gigaRouterCommon) BlockByHash(ctx context.Context, hash atypes.BlockHeaderHash) (*coretypes.ResultBlock, error) { opt, err := r.data.GlobalBlockByHash(hash) if err != nil { @@ -209,7 +206,7 @@ func (r *gigaRouterCommon) translateGlobalBlock(gb *atypes.GlobalBlock) *coretyp } } -func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlock) (*abci.ResponseCommit, error) { +func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlock, hashVault hashvault.HashVault) (*abci.ResponseCommit, error) { app := r.app hash := b.Header.Hash() var proposerAddress types.Address @@ -251,7 +248,7 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo // before the hash is proposed for AppQC voting via PushAppHash below). On restart the block is // re-executed and the identical hash is re-committed idempotently. A returned error is a benign // shutdown cancellation; genuine faults panic inside the call. See commitAppHashToVault. - if err := commitAppHashToVault(ctx, r.hashVault, b.GlobalNumber, resp.AppHash); err != nil { + if err := commitAppHashToVault(ctx, hashVault, b.GlobalNumber, resp.AppHash); err != nil { return nil, err } @@ -267,10 +264,8 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo // buildHashVault constructs the app-hash equivocation guard runExecute owns. By default it // returns a durable Pebble-backed vault rooted at /hashvault, alongside the -// other Autobahn on-disk state. It returns a no-op vault (no protection) in two cases: the operator -// explicitly set HashVaultDisabledUnsafe — logged loudly as unsafe — or there is no persistent -// state directory (in-memory mode, e.g. tests), where a durable vault would be pointless because -// the data WAL is itself a no-op (see data.NewDataWAL). +// other Autobahn on-disk state. It returns a no-op vault (no protection) when PersistentStateDir +// is unset or when the operator explicitly sets HashVaultDisabledUnsafe — both logged loudly. func buildHashVault(ctx context.Context, cfg *GigaRouterCommonConfig) (hashvault.HashVault, error) { if cfg.HashVaultDisabledUnsafe { logger.Error("################################################################") @@ -283,9 +278,12 @@ func buildHashVault(ctx context.Context, cfg *GigaRouterCommonConfig) (hashvault } dir, ok := cfg.PersistentStateDir.Get() if !ok { - logger.Error("HASHVAULT DISABLED: no persistent state dir (in-memory mode); using no-op vault. " + - "This node has NO app-hash equivocation protection. This is expected only for in-memory " + - "runs such as tests; a production node reaching this path is misconfigured.") + logger.Error("################################################################") + logger.Error("# HASHVAULT DISABLED (PersistentStateDir not set). #") + logger.Error("# This node has NO app-hash equivocation protection and is #") + logger.Error("# running in an UNSAFE configuration. Set persistent_state_dir #") + logger.Error("# in the node config to enable protection. #") + logger.Error("################################################################") return hashvault.NewNoopHashVault(), nil } hvCfg := hashvault.DefaultHashVaultConfig() @@ -337,7 +335,6 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { if err != nil { return fmt.Errorf("buildHashVault(): %w", err) } - r.hashVault = hashVault defer func() { if err := hashVault.Close(context.Background()); err != nil { logger.Error("failed to close hashvault", "err", err) @@ -379,7 +376,7 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { // first executed; if the committed app state has diverged from what the vault recorded (e.g. an // out-of-band rollback/restore), this halts the node instead of externalizing a conflicting // hash. A returned error is a benign shutdown cancellation; genuine faults panic inside. - if err := commitAppHashToVault(ctx, r.hashVault, last, info.LastBlockAppHash); err != nil { + if err := commitAppHashToVault(ctx, hashVault, last, info.LastBlockAppHash); err != nil { return err } // Losing a prefix of appHashes on crash is fine: AppQC is reached @@ -394,7 +391,7 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { if err != nil { return fmt.Errorf("r.data.GlobalBlock(%v): %w", n, err) } - commitResp, err := r.executeBlock(ctx, b) + commitResp, err := r.executeBlock(ctx, b, hashVault) if err != nil { return fmt.Errorf("r.executeBlock(%v): %w", n, err) } @@ -406,7 +403,7 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { return fmt.Errorf("r.data.PruneBefore(%v): %w", pruneBefore, err) } // Align the vault's retention with the data layer's prune boundary. - if err := r.hashVault.Prune(ctx, uint64(pruneBefore)); err != nil { + if err := hashVault.Prune(ctx, uint64(pruneBefore)); err != nil { // A canceled context just means we're shutting down between a successful executeBlock // and this prune; that's benign, not a prune failure, so don't alarm operators. if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { diff --git a/sei-tendermint/internal/p2p/giga_router_fullnode.go b/sei-tendermint/internal/p2p/giga_router_fullnode.go index afd0e1fa3f..1c3a7c43fa 100644 --- a/sei-tendermint/internal/p2p/giga_router_fullnode.go +++ b/sei-tendermint/internal/p2p/giga_router_fullnode.go @@ -6,6 +6,7 @@ import ( "math/rand/v2" "slices" + atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/producer" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/giga" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/rpc" @@ -15,10 +16,12 @@ import ( type gigaFullnodeRouter struct { *gigaRouterCommon + + blockDB atypes.BlockDB // nil if persistence is disabled } func NewGigaFullnodeRouter(cfg *GigaRouterCommonConfig, key NodeSecretKey) (*gigaFullnodeRouter, error) { - dataState, err := buildDataState(cfg) + dataState, blockDB, err := buildDataState(cfg) if err != nil { return nil, err } @@ -34,6 +37,7 @@ func NewGigaFullnodeRouter(cfg *GigaRouterCommonConfig, key NodeSecretKey) (*gig app: cfg.App, inboundFullnodeCap: int64(cfg.MaxInboundFullnodePeers), }, + blockDB: blockDB, }, nil } @@ -42,6 +46,9 @@ func (r *gigaFullnodeRouter) Mempool() utils.Option[*producer.State] { } func (r *gigaFullnodeRouter) Run(ctx context.Context) error { + if r.blockDB != nil { + defer func() { _ = r.blockDB.Close() }() + } return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { // Single-active subscriber: walk the committee in a stable order, // move to the next on disconnect. Avoids the N× QC duplication of diff --git a/sei-tendermint/internal/p2p/giga_router_fullnode_test.go b/sei-tendermint/internal/p2p/giga_router_fullnode_test.go index bcd191785b..ff58611d14 100644 --- a/sei-tendermint/internal/p2p/giga_router_fullnode_test.go +++ b/sei-tendermint/internal/p2p/giga_router_fullnode_test.go @@ -57,14 +57,12 @@ func TestGigaRouter_Fullnode(t *testing.T) { app := newTestApp() proxyApp := proxy.New(app) - // Fullnodes have no validator key and no Producer config. The data WAL - // reads PersistentStateDir = None (in-memory) for this construction- - // shape check; App is required for executeBlock but isn't exercised by - // this test. + // Fullnodes have no validator key and no Producer config. + // App is required for executeBlock but isn't exercised by this test. router, err := NewGigaFullnodeRouter(&GigaRouterCommonConfig{ DialInterval: time.Second, ValidatorAddrs: addrs, - PersistentStateDir: utils.None[string](), + PersistentStateDir: utils.Some(t.TempDir()), App: proxyApp, GenDoc: genDoc, }, makeKey(rng)) diff --git a/sei-tendermint/internal/p2p/giga_router_validator.go b/sei-tendermint/internal/p2p/giga_router_validator.go index 4c65cd1538..a126b38ca8 100644 --- a/sei-tendermint/internal/p2p/giga_router_validator.go +++ b/sei-tendermint/internal/p2p/giga_router_validator.go @@ -18,6 +18,7 @@ import ( type gigaValidatorRouter struct { *gigaRouterCommon + blockDB atypes.BlockDB // nil if persistence is disabled consensus *consensus.State producer *producer.State // validatorKey is the cached public form of cfg.ValidatorKey, used by @@ -26,7 +27,7 @@ type gigaValidatorRouter struct { } func NewGigaValidatorRouter(cfg *GigaValidatorConfig, key NodeSecretKey) (*gigaValidatorRouter, error) { - dataState, err := buildDataState(&cfg.GigaRouterCommonConfig) + dataState, blockDB, err := buildDataState(&cfg.GigaRouterCommonConfig) if err != nil { return nil, err } @@ -36,6 +37,9 @@ func NewGigaValidatorRouter(cfg *GigaValidatorConfig, key NodeSecretKey) (*gigaV PersistentStateDir: cfg.PersistentStateDir, }, dataState) if err != nil { + if blockDB != nil { + _ = blockDB.Close() + } return nil, fmt.Errorf("consensus.NewState(): %w", err) } producerState := producer.NewState(cfg.Producer, consensusState, cfg.App) @@ -51,6 +55,7 @@ func NewGigaValidatorRouter(cfg *GigaValidatorConfig, key NodeSecretKey) (*gigaV app: cfg.App, inboundFullnodeCap: int64(cfg.MaxInboundFullnodePeers), }, + blockDB: blockDB, consensus: consensusState, producer: producerState, validatorKey: cfg.ValidatorKey.Public(), @@ -62,6 +67,9 @@ func (r *gigaValidatorRouter) Mempool() utils.Option[*producer.State] { } func (r *gigaValidatorRouter) Run(ctx context.Context) error { + if r.blockDB != nil { + defer func() { _ = r.blockDB.Close() }() + } return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { // Validators dial every committee member in parallel — consensus // voting needs fan-out, not stickiness. Same connections also diff --git a/sei-tendermint/internal/p2p/giga_router_validator_test.go b/sei-tendermint/internal/p2p/giga_router_validator_test.go index 78a5f6cc66..cdec864e13 100644 --- a/sei-tendermint/internal/p2p/giga_router_validator_test.go +++ b/sei-tendermint/internal/p2p/giga_router_validator_test.go @@ -67,7 +67,7 @@ func TestGigaRouter_FinalizeBlocks(t *testing.T) { // Aggressive dialing rate to speed up startup. DialInterval: 100 * time.Millisecond, ValidatorAddrs: addrs, - PersistentStateDir: utils.None[string](), + PersistentStateDir: utils.Some(t.TempDir()), App: proxyApp, GenDoc: genDoc, }, @@ -224,7 +224,7 @@ func TestGigaRouter_EvmProxy(t *testing.T) { GigaRouterCommonConfig: GigaRouterCommonConfig{ DialInterval: time.Second, ValidatorAddrs: addrs, - PersistentStateDir: utils.None[string](), + PersistentStateDir: utils.Some(t.TempDir()), App: proxy.New(newTestApp()), GenDoc: genDoc, }, diff --git a/sei-tendermint/node/setup.go b/sei-tendermint/node/setup.go index 0b81dbe4db..24316dedd4 100644 --- a/sei-tendermint/node/setup.go +++ b/sei-tendermint/node/setup.go @@ -310,7 +310,9 @@ func buildGigaRouter( if err != nil { return nil, fmt.Errorf("buildValidatorGigaConfig: %w", err) } - rootifyPersistentStateDir(cfg.RootDir, &valCfg.GigaRouterCommonConfig) + if err := preparePersistentStateDir(cfg.RootDir, &valCfg.GigaRouterCommonConfig); err != nil { + return nil, err + } // The GigaRouter builds and owns the equivocation guard itself; just pass the operator's // enable/disable decision through as plain config. valCfg.HashVaultDisabledUnsafe = cfg.HashVaultDisabledUnsafe @@ -321,7 +323,9 @@ func buildGigaRouter( if err != nil { return nil, fmt.Errorf("buildFullnodeGigaConfig: %w", err) } - rootifyPersistentStateDir(cfg.RootDir, fnCfg) + if err := preparePersistentStateDir(cfg.RootDir, fnCfg); err != nil { + return nil, err + } // The GigaRouter builds and owns the equivocation guard itself; just pass the operator's // enable/disable decision through as plain config. fnCfg.HashVaultDisabledUnsafe = cfg.HashVaultDisabledUnsafe @@ -329,13 +333,21 @@ func buildGigaRouter( return p2p.NewGigaFullnodeRouter(fnCfg, p2p.NodeSecretKey(nodeKey)) } -// rootifyPersistentStateDir resolves a relative PersistentStateDir -// against the node's --home dir (mirrors config.go's rootify). Absolute -// paths pass through; None stays None. -func rootifyPersistentStateDir(rootDir string, c *p2p.GigaRouterCommonConfig) { - if dir, ok := c.PersistentStateDir.Get(); ok && !filepath.IsAbs(dir) { - c.PersistentStateDir = utils.Some(filepath.Join(rootDir, dir)) +// preparePersistentStateDir resolves a relative PersistentStateDir against +// the node's --home dir (mirrors config.go's rootify) and creates it if absent. +func preparePersistentStateDir(rootDir string, c *p2p.GigaRouterCommonConfig) error { + dir, ok := c.PersistentStateDir.Get() + if !ok { + return nil + } + if !filepath.IsAbs(dir) { + dir = filepath.Join(rootDir, dir) + c.PersistentStateDir = utils.Some(dir) + } + if err := os.MkdirAll(dir, 0700); err != nil { + return fmt.Errorf("creating persistent state dir %q: %w", dir, err) } + return nil } // resolveMaxInboundFullnodePeers: None ⇒ default, Some(0) ⇒ reject all, diff --git a/sei-tendermint/node/setup_test.go b/sei-tendermint/node/setup_test.go index 375a0098fa..8dd0dc953a 100644 --- a/sei-tendermint/node/setup_test.go +++ b/sei-tendermint/node/setup_test.go @@ -54,7 +54,7 @@ func writeAutobahnConfig(t *testing.T, fc *config.AutobahnFileConfig) string { return path } -func defaultFileConfig(validators []config.AutobahnValidator) *config.AutobahnFileConfig { +func defaultFileConfig(t testing.TB, validators []config.AutobahnValidator) *config.AutobahnFileConfig { return &config.AutobahnFileConfig{ Validators: validators, MaxTxsPerBlock: 5_000, @@ -62,7 +62,7 @@ func defaultFileConfig(validators []config.AutobahnValidator) *config.AutobahnFi AllowEmptyBlocks: false, BlockInterval: utils.Duration(400 * time.Millisecond), ViewTimeout: utils.Duration(1500 * time.Millisecond), - PersistentStateDir: utils.None[string](), + PersistentStateDir: utils.Some(t.TempDir()), DialInterval: utils.Duration(10 * time.Second), } } @@ -84,6 +84,20 @@ func makeTestGigaDeps() (*proxy.Proxy, *types.GenesisDoc) { return app, genDoc } +func TestBuildGigaConfig_NonePersistentStateDir(t *testing.T) { + v1 := makeValidator([]byte("val-seed"), []byte("node-seed"), "localhost:26660") + fc := defaultFileConfig(t, []config.AutobahnValidator{v1}) + fc.PersistentStateDir = utils.None[string]() + cfgFile := writeAutobahnConfig(t, fc) + nodeKey := makeTestNodeKey([]byte("node-seed")) + valKey := makeTestValidatorKey([]byte("val-seed")) + txMempool, genDoc := makeTestGigaDeps() + + result, err := buildValidatorGigaConfig(cfgFile, nodeKey, valKey, txMempool, genDoc) + require.NoError(t, err) + assert.False(t, result.PersistentStateDir.IsPresent()) +} + func TestBuildGigaConfig_EmptyPathErrors(t *testing.T) { nodeKey := makeTestNodeKey([]byte("test-node-key")) valKey := makeTestValidatorKey([]byte("val-seed")) @@ -122,9 +136,7 @@ func TestBuildGigaConfig_EnabledWithValidators(t *testing.T) { assert.Equal(t, 5*time.Second, result.DialInterval) assert.Equal(t, 3*time.Second, result.ViewTimeout(atypes.View{})) - persistDir, ok := result.PersistentStateDir.Get() - require.True(t, ok) - assert.Equal(t, "/tmp/autobahn-state", persistDir) + assert.Equal(t, utils.Some("/tmp/autobahn-state"), result.PersistentStateDir) // Verify the validator key is derived from the validator-key seed, not the node key. expectedValPub := makeTestValidatorKey([]byte("val1-seed")).Public() @@ -145,7 +157,7 @@ func TestBuildGigaConfig_EnabledWithValidators(t *testing.T) { func TestBuildGigaConfig_NoneMaxTxsPerSecond(t *testing.T) { v1 := makeValidator([]byte("val-seed"), []byte("node-seed"), "localhost:26660") - fc := defaultFileConfig([]config.AutobahnValidator{v1}) + fc := defaultFileConfig(t, []config.AutobahnValidator{v1}) cfgFile := writeAutobahnConfig(t, fc) nodeKey := makeTestNodeKey([]byte("node-seed")) valKey := makeTestValidatorKey([]byte("val-seed")) @@ -157,19 +169,6 @@ func TestBuildGigaConfig_NoneMaxTxsPerSecond(t *testing.T) { assert.False(t, result.Producer.MaxTxsPerSecond.IsPresent()) } -func TestBuildGigaConfig_NonePersistentStateDir(t *testing.T) { - v1 := makeValidator([]byte("val-seed"), []byte("node-seed"), "localhost:26660") - fc := defaultFileConfig([]config.AutobahnValidator{v1}) - cfgFile := writeAutobahnConfig(t, fc) - nodeKey := makeTestNodeKey([]byte("node-seed")) - valKey := makeTestValidatorKey([]byte("val-seed")) - txMempool, genDoc := makeTestGigaDeps() - - result, err := buildValidatorGigaConfig(cfgFile, nodeKey, valKey, txMempool, genDoc) - require.NoError(t, err) - assert.False(t, result.PersistentStateDir.IsPresent()) -} - func TestBuildGigaConfig_InvalidConfigFile(t *testing.T) { nodeKey := makeTestNodeKey([]byte("node-seed")) valKey := makeTestValidatorKey([]byte("val-seed")) @@ -188,7 +187,7 @@ func TestBuildGigaConfig_InvalidConfigFile(t *testing.T) { }) t.Run("empty validators", func(t *testing.T) { - fc := defaultFileConfig([]config.AutobahnValidator{}) + fc := defaultFileConfig(t, []config.AutobahnValidator{}) cfgFile := writeAutobahnConfig(t, fc) _, err := buildValidatorGigaConfig(cfgFile, nodeKey, valKey, txMempool, genDoc) assert.Error(t, err) @@ -202,7 +201,7 @@ func TestBuildGigaConfig_GenesisMaxGas(t *testing.T) { nodeKey := makeTestNodeKey([]byte("node-seed")) valKey := makeTestValidatorKey([]byte("val-seed")) v := makeValidator([]byte("val-seed"), []byte("node-seed"), "localhost:26660") - cfgFile := writeAutobahnConfig(t, defaultFileConfig([]config.AutobahnValidator{v})) + cfgFile := writeAutobahnConfig(t, defaultFileConfig(t, []config.AutobahnValidator{v})) t.Run("nil ConsensusParams", func(t *testing.T) { txMempool, genDoc := makeTestGigaDeps() @@ -229,7 +228,7 @@ func TestBuildGigaConfig_GenesisMaxGas(t *testing.T) { func TestBuildGigaConfig_DuplicateValidatorKey(t *testing.T) { v1 := makeValidator([]byte("val-seed"), []byte("node1"), "localhost:26660") v1dup := makeValidator([]byte("val-seed"), []byte("node2"), "localhost:26661") - fc := defaultFileConfig([]config.AutobahnValidator{v1, v1dup}) + fc := defaultFileConfig(t, []config.AutobahnValidator{v1, v1dup}) data, _ := json.Marshal(fc) path := filepath.Join(t.TempDir(), "autobahn.json") os.WriteFile(path, data, 0644) @@ -245,7 +244,7 @@ func TestBuildGigaConfig_DuplicateValidatorKey(t *testing.T) { func TestBuildGigaConfig_DuplicateNodeKey(t *testing.T) { v1 := makeValidator([]byte("val1"), []byte("same-node"), "localhost:26660") v2 := makeValidator([]byte("val2"), []byte("same-node"), "localhost:26661") - fc := defaultFileConfig([]config.AutobahnValidator{v1, v2}) + fc := defaultFileConfig(t, []config.AutobahnValidator{v1, v2}) data, _ := json.Marshal(fc) path := filepath.Join(t.TempDir(), "autobahn.json") os.WriteFile(path, data, 0644) @@ -260,7 +259,7 @@ func TestBuildGigaConfig_DuplicateNodeKey(t *testing.T) { func TestBuildGigaConfig_SelfNotInValidators(t *testing.T) { v1 := makeValidator([]byte("other-val"), []byte("other-node"), "localhost:26660") - cfgFile := writeAutobahnConfig(t, defaultFileConfig([]config.AutobahnValidator{v1})) + cfgFile := writeAutobahnConfig(t, defaultFileConfig(t, []config.AutobahnValidator{v1})) nodeKey := makeTestNodeKey([]byte("my-node")) valKey := makeTestValidatorKey([]byte("my-val")) txMempool, genDoc := makeTestGigaDeps() @@ -273,7 +272,7 @@ func TestBuildGigaConfig_SelfNotInValidators(t *testing.T) { func TestBuildGigaConfig_NodeKeyMismatch(t *testing.T) { // Validator entry has the right val key but wrong node key. v1 := makeValidator([]byte("my-val"), []byte("wrong-node"), "localhost:26660") - cfgFile := writeAutobahnConfig(t, defaultFileConfig([]config.AutobahnValidator{v1})) + cfgFile := writeAutobahnConfig(t, defaultFileConfig(t, []config.AutobahnValidator{v1})) nodeKey := makeTestNodeKey([]byte("my-node")) valKey := makeTestValidatorKey([]byte("my-val")) txMempool, genDoc := makeTestGigaDeps()