diff --git a/sei-tendermint/autobahn/types/committee.go b/sei-tendermint/autobahn/types/committee.go index 400f1a0022..315f4e0e1d 100644 --- a/sei-tendermint/autobahn/types/committee.go +++ b/sei-tendermint/autobahn/types/committee.go @@ -11,6 +11,8 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/holiman/uint256" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pb" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/protoutils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" ) @@ -131,6 +133,53 @@ func NewCommittee(weights map[PublicKey]uint64) (*Committee, error) { return newCommittee(nil, weights, 0) } +// Equal reports whether c and other have the same validators, join epochs, and weights. +func (c *Committee) Equal(other *Committee) bool { + return maps.Equal(c.weights, other.weights) && maps.Equal(c.lanes, other.lanes) +} + +var CommitteeConv = protoutils.Conv[*Committee, *pb.Committee]{ + Encode: func(c *Committee) *pb.Committee { + record := &pb.Committee{Members: make([]*pb.EpochMember, 0, c.Lanes().Len())} + for lane := range c.Lanes().All() { + record.Members = append(record.Members, &pb.EpochMember{ + LaneId: LaneIDConv.Encode(lane), + Weight: utils.Alloc(c.Weight(lane.Validator)), + }) + } + return record + }, + Decode: func(record *pb.Committee) (*Committee, error) { + if record == nil { + return nil, errors.New("missing") + } + weights := make(map[PublicKey]uint64, len(record.Members)) + lanes := make(map[PublicKey]LaneID, len(record.Members)) + for i, member := range record.Members { + if member == nil { + return nil, fmt.Errorf("member %d: missing", i) + } + if member.Weight == nil { + return nil, fmt.Errorf("member %d weight: missing", i) + } + if *member.Weight == 0 { + return nil, fmt.Errorf("member %d weight is 0", i) + } + lane, err := LaneIDConv.DecodeReq(member.LaneId) + if err != nil { + return nil, fmt.Errorf("member %d lane: %w", i, err) + } + validator := lane.Validator + if _, ok := weights[validator]; ok { + return nil, fmt.Errorf("duplicate public key %s", validator) + } + weights[validator] = *member.Weight + lanes[validator] = lane + } + return newCommittee(lanes, weights, 0) + }, +} + // DeriveNext builds the committee for epoch e>0 from this committee: // validators that remain keep Joined; new members get Joined = e. func (c *Committee) DeriveNext(weights map[PublicKey]uint64, e EpochIndex) (*Committee, error) { diff --git a/sei-tendermint/autobahn/types/committee_test.go b/sei-tendermint/autobahn/types/committee_test.go index d05d7bbe1d..3329850364 100644 --- a/sei-tendermint/autobahn/types/committee_test.go +++ b/sei-tendermint/autobahn/types/committee_test.go @@ -5,6 +5,7 @@ import ( "testing" "time" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pb" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" ) @@ -36,6 +37,33 @@ func TestNewCommittee_FiltersOutZeroWeightValidators(t *testing.T) { } } +func TestCommitteeEqual_IgnoresMapOrder(t *testing.T) { + rng := utils.TestRng() + a := GenPublicKey(rng) + b := GenPublicKey(rng) + c1 := utils.OrPanic1(NewCommittee(map[PublicKey]uint64{a: 3, b: 1})) + c2 := utils.OrPanic1(NewCommittee(map[PublicKey]uint64{b: 1, a: 3})) + require.True(t, c1.Equal(c2)) +} + +func TestCommitteeEqual_IgnoresZeroWeights(t *testing.T) { + rng := utils.TestRng() + a := GenPublicKey(rng) + b := GenPublicKey(rng) + c1 := utils.OrPanic1(NewCommittee(map[PublicKey]uint64{a: 3})) + c2 := utils.OrPanic1(NewCommittee(map[PublicKey]uint64{a: 3, b: 0})) + require.True(t, c1.Equal(c2)) +} + +func TestCommitteeEqual_RejectsDifferentWeights(t *testing.T) { + rng := utils.TestRng() + a := GenPublicKey(rng) + b := GenPublicKey(rng) + c1 := utils.OrPanic1(NewCommittee(map[PublicKey]uint64{a: 3, b: 1})) + c2 := utils.OrPanic1(NewCommittee(map[PublicKey]uint64{a: 3, b: 2})) + require.False(t, c1.Equal(c2)) +} + func TestNewCommittee_RejectsZeroTotalWeight(t *testing.T) { rng := utils.TestRng() @@ -291,3 +319,44 @@ func TestDeriveNext_StayLeaveRejoin(t *testing.T) { require.False(t, c3.HasLane(LaneID{Validator: d, Joined: 0})) requireLanesSorted(t, c3) } + +func TestCommitteeConv_RejectsZeroWeight(t *testing.T) { + rng := utils.TestRng() + lane := GenLaneID(rng) + _, err := CommitteeConv.Decode(&pb.Committee{Members: []*pb.EpochMember{{ + LaneId: LaneIDConv.Encode(lane), + Weight: utils.Alloc[uint64](0), + }}}) + require.Error(t, err) +} + +func TestCommitteeConv_RejectsDuplicatePublicKey(t *testing.T) { + rng := utils.TestRng() + lane := GenLaneID(rng) + member := &pb.EpochMember{ + LaneId: LaneIDConv.Encode(lane), + Weight: utils.Alloc[uint64](1), + } + _, err := CommitteeConv.Decode(&pb.Committee{Members: []*pb.EpochMember{member, member}}) + require.Error(t, err) +} + +func TestCommitteeConv_PreservesJoined(t *testing.T) { + rng := utils.TestRng() + a := GenPublicKey(rng) + b := GenPublicKey(rng) + c := GenPublicKey(rng) + d := GenPublicKey(rng) + genesis := utils.OrPanic1(NewCommittee(map[PublicKey]uint64{a: 1, c: 3})) + src := utils.OrPanic1(genesis.DeriveNext(map[PublicKey]uint64{a: 2, b: 4, c: 3}, 4)) + got, err := CommitteeConv.Decode(CommitteeConv.Encode(src)) + require.NoError(t, err) + require.Equal(t, src.Lane(a).OrPanic("a"), got.Lane(a).OrPanic("a")) + require.Equal(t, src.Lane(b).OrPanic("b"), got.Lane(b).OrPanic("b")) + require.Equal(t, src.Lane(c).OrPanic("c"), got.Lane(c).OrPanic("c")) + require.Equal(t, uint64(2), got.Weight(a)) + require.Equal(t, uint64(4), got.Weight(b)) + require.Equal(t, uint64(3), got.Weight(c)) + require.Equal(t, EpochIndex(4), got.Lane(b).OrPanic("b").Joined) + require.False(t, got.HasReplica(d)) +} diff --git a/sei-tendermint/autobahn/types/testonly.go b/sei-tendermint/autobahn/types/testonly.go index ad6b8a2454..12fe6d26ca 100644 --- a/sei-tendermint/autobahn/types/testonly.go +++ b/sei-tendermint/autobahn/types/testonly.go @@ -331,10 +331,25 @@ func GenProposalAt(rng utils.Rng, view View) *Proposal { // Proposal.Verify accepts it (empty tipcuts are forbidden). For tests that care // about signature weight or epoch binding rather than real lane/app data. func ProposalAt(ep *Epoch, view View, globalFirst GlobalBlockNumber) *Proposal { + p, _ := ProposalAtBlocks(ep, view, globalFirst, 1) + return p +} + +// ProposalAtBlocks is ProposalAt with a lane range of n blocks, n >= 1. +func ProposalAtBlocks(ep *Epoch, view View, globalFirst GlobalBlockNumber, n int) (*Proposal, []*Block) { view.EpochIndex = ep.EpochIndex() lane := ep.Committee().Lanes().At(0) - header := NewBlock(lane, 0, BlockHeaderHash{}, &Payload{}).Header() - return newProposal(view, time.Time{}, []*LaneRange{NewLaneRange(lane, 0, utils.Some(header))}, globalFirst) + blocks := make([]*Block, n) + var parent BlockHeaderHash + num := BlockNumber(0) + for i := 0; i < n; i++ { + b := NewBlock(lane, num, parent, &Payload{}) + blocks[i] = b + parent = b.Header().Hash() + num = b.Header().Next() + } + header := blocks[n-1].Header() + return newProposal(view, time.Time{}, []*LaneRange{NewLaneRange(lane, 0, utils.Some(header))}, globalFirst), blocks } // GenProposalForEpoch generates a Proposal at a specific view whose epochIndex, diff --git a/sei-tendermint/internal/autobahn/autobahn.proto b/sei-tendermint/internal/autobahn/autobahn.proto index 01b3f68385..23ab14cfdd 100644 --- a/sei-tendermint/internal/autobahn/autobahn.proto +++ b/sei-tendermint/internal/autobahn/autobahn.proto @@ -316,3 +316,27 @@ message ConsensusReq { TimeoutQC timeout_qc = 5; } } + +message Committee { + option (wireguard.sized) = true; + repeated EpochMember members = 1 [(wireguard.max_count) = 100]; +} + +// Committee for an execution-derived epoch. +message EpochRecord { + option (wireguard.sized) = true; + optional uint64 index = 1; // required + optional Committee committee = 2; // required +} + +// Persisted execution-derived epoch registry state. +message PersistedEpochRegistry { + repeated EpochRecord live = 1; + optional EpochRecord pending = 2; // optional +} + +message EpochMember { + option (wireguard.sized) = true; + optional LaneID lane_id = 1; // required + optional uint64 weight = 2; // required, can't be zero +} diff --git a/sei-tendermint/internal/autobahn/avail/inner_test.go b/sei-tendermint/internal/autobahn/avail/inner_test.go index e7653b2be0..8e41bd5409 100644 --- a/sei-tendermint/internal/autobahn/avail/inner_test.go +++ b/sei-tendermint/internal/autobahn/avail/inner_test.go @@ -235,7 +235,6 @@ func TestAddLane_ReportsNewLaneForEachMembershipPeriod(t *testing.T) { func TestRefreshConsensusSpec_WithholdsTipUntilNextViewEpochApplied(t *testing.T) { rng := utils.TestRng() registry, keys := epoch.GenRegistry(rng, 4) - registry.AdvanceIfNeeded(epoch.LastRoad(0)) ep0 := registry.MustEpoch(0) ep1 := registry.MustEpoch(1) @@ -273,9 +272,7 @@ func TestRefreshConsensusSpec_WithholdsTipUntilNextViewEpochApplied(t *testing.T func TestPrune_LeavesAppliedToEpochAdvance(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - registry.AdvanceIfNeeded(epoch.LastRoad(1)) - registry.AdvanceIfNeeded(epoch.LastRoad(2)) + registry, keys := epoch.GenRegistryThrough(rng, 4, 2) ep0 := registry.MustEpoch(0) ep1 := registry.MustEpoch(1) ep2 := registry.MustEpoch(2) diff --git a/sei-tendermint/internal/autobahn/avail/state_test.go b/sei-tendermint/internal/autobahn/avail/state_test.go index a7345e2e9b..34c4156b48 100644 --- a/sei-tendermint/internal/autobahn/avail/state_test.go +++ b/sei-tendermint/internal/autobahn/avail/state_test.go @@ -107,7 +107,7 @@ func TestPrune_AnchorEpochDropsClosedLane(t *testing.T) { return err } appHash0 := types.GenAppHash(rng) - if err := ds.PushAppHash(ctx, qc0.QC().GlobalRange().Next-1, appHash0); err != nil { + if err := ds.PushAppHash(ctx, qc0.QC().GlobalRange().Next-1, appHash0, nil); err != nil { return err } if err := ds.PushAppQC(ctx, data.TestAppQC(keys, types.NewAppProposal(qc0.QC().Proposal(), appHash0))); err != nil { @@ -129,17 +129,10 @@ func TestPrune_AnchorEpochDropsClosedLane(t *testing.T) { lane0 := state.LocalLane().OrPanic("genesis") sub := state.SubscribeLaneProposals(lane0, 0) - epLeave, err := registry.ActivateEpoch( - 0, - map[types.PublicKey]uint64{b.Public(): 1}, - time.Time{}, registry.FirstBlock(), - ) - if err != nil { + if err := registry.StageAndActivate(0, map[types.PublicKey]uint64{b.Public(): 1}); err != nil { return err } - if epLeave.EpochIndex() != 2 { - return fmt.Errorf("leave epoch = %d, want 2 (epoch 1 is genesis-seeded)", epLeave.EpochIndex()) - } + epLeave := registry.MustEpoch(2) // Data already holds an AppQC for epoch 0; NewState's prune stamped the // Anchor. Advance the seal cursor without admitting LastRoad tips — runEvict // is not running, and the rest of this test expects empty roads. @@ -219,7 +212,7 @@ func TestAnchorResetsState(t *testing.T) { return err } appHash := types.GenAppHash(rng) - if err := ds.PushAppHash(ctx, qc.QC().GlobalRange().Next-1, appHash); err != nil { + if err := ds.PushAppHash(ctx, qc.QC().GlobalRange().Next-1, appHash, nil); err != nil { return err } appQC := data.TestAppQC(keys, types.NewAppProposal(qc.QC().Proposal(), appHash)) @@ -334,7 +327,7 @@ func testState(t *testing.T, rng utils.Rng, stateDir utils.Option[string]) { appHash := types.GenAppHash(rng) appProposal := types.NewAppProposal(qc.Proposal(), appHash) appGR := appProposal.GlobalRange() - if err := ds.PushAppHash(ctx, appGR.Next-1, appHash); err != nil { + if err := ds.PushAppHash(ctx, appGR.Next-1, appHash, nil); err != nil { return fmt.Errorf("ds.PushAppHash(): %w", err) } for _, vote := range makeAppVotes(keys, appProposal) { @@ -381,7 +374,6 @@ func TestNextViewEpoch(t *testing.T) { t.Run("LastRoad tip pairs next epoch", func(t *testing.T) { rng := utils.TestRng() registry, keys := epoch.GenRegistry(rng, 4) - registry.AdvanceIfNeeded(epoch.LastRoad(0)) ep0 := registry.MustEpoch(0) ep1 := registry.MustEpoch(1) @@ -700,7 +692,7 @@ func TestHeaders_WaitsForPrevEpochLaneVote(t *testing.T) { stay.Public(): 1, leaver.Public(): 1, a.Public(): 1, b.Public(): 1, }) require.NoError(t, err) - registry, err := epoch.NewRegistry(genesis, 0, time.Time{}) + registry, err := epoch.NewRegistry(genesis, 0, time.Time{}, utils.None[string]()) require.NoError(t, err) ds := newTestDataState(&data.Config{Registry: registry}) state, err := NewState(stay, ds, utils.None[string]()) @@ -716,14 +708,10 @@ func TestHeaders_WaitsForPrevEpochLaneVote(t *testing.T) { header := types.NewBlock(lane, 0, types.BlockHeaderHash{}, &types.Payload{}).Header() leaverVote := types.Sign(leaver, types.NewLaneVote(header)) - epLeave, err := registry.ActivateEpoch( - 0, - map[types.PublicKey]uint64{stay.Public(): 1, a.Public(): 1, b.Public(): 1}, - time.Time{}, registry.FirstBlock(), - ) - if err != nil { + if err := registry.StageAndActivate(0, map[types.PublicKey]uint64{stay.Public(): 1, a.Public(): 1, b.Public(): 1}); err != nil { return err } + epLeave := registry.MustEpoch(2) keys := []types.SecretKey{stay, leaver, a, b} if err := TestDriveAdvance(ctx, state, keys, epLeave.EpochIndex()); err != nil { return err @@ -852,6 +840,15 @@ func newSealFixture(t *testing.T) *sealFixture { return &sealFixture{registry: registry, keys: keys, state: state, ep: ep, m: m} } +func addEpoch(t *testing.T, registry *epoch.Registry, end types.EpochIndex, keys []types.SecretKey) { + t.Helper() + weights := make(map[types.PublicKey]uint64, len(keys)) + for _, key := range keys { + weights[key.Public()] = 1 + } + require.NoError(t, registry.StageAndActivate(end, weights)) +} + func TestRunEpochAdvance_Leashes(t *testing.T) { type missing int const ( @@ -875,7 +872,7 @@ func TestRunEpochAdvance_Leashes(t *testing.T) { rng := utils.TestRng() f := newSealFixture(t) if tc.missing != registry { - f.registry.AdvanceIfNeeded(epoch.LastRoad(f.m)) + addEpoch(t, f.registry, f.m-1, f.keys) } prev := tipLink(f.ep, f.keys[0], epoch.LastRoad(f.m)-1) @@ -905,7 +902,7 @@ func TestRunEpochAdvance_Leashes(t *testing.T) { require.Equal(t, f.m, f.state.Epoch().Load().EpochIndex()) switch tc.missing { case registry: - f.registry.AdvanceIfNeeded(epoch.LastRoad(f.m)) + addEpoch(t, f.registry, f.m-1, f.keys) case appQC: setRoadAppQC(f.state, qcLast.Index(), data.TestAppQC(f.keys, types.NewAppProposal(qcLast.Proposal(), types.GenAppHash(rng)))) } @@ -975,14 +972,12 @@ func TestRunEpochAdvance_CatchUpDoesNotKill(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 2) + registry, keys := epoch.GenRegistryThrough(rng, 2, 3) ds := newTestDataState(&data.Config{Registry: registry}) state, err := NewState(keys[0], ds, utils.None[string]()) require.NoError(t, err) require.Equal(t, types.EpochIndex(0), state.Epoch().Load().EpochIndex()) - registry.AdvanceIfNeeded(epoch.LastRoad(1)) - registry.AdvanceIfNeeded(epoch.LastRoad(2)) qc := pruneToAnchors(state, anchorWalk(t, registry, keys)) require.Equal(t, types.EpochIndex(0), state.Epoch().Load().EpochIndex()) @@ -1004,13 +999,11 @@ func TestRunEpochAdvance_CatchUpDoesNotKill(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 2) + registry, keys := epoch.GenRegistryThrough(rng, 2, 3) ds := newTestDataState(&data.Config{Registry: registry}) state, err := NewState(keys[0], ds, utils.None[string]()) require.NoError(t, err) - registry.AdvanceIfNeeded(epoch.LastRoad(1)) - registry.AdvanceIfNeeded(epoch.LastRoad(2)) anchors := anchorWalk(t, registry, keys) ep0 := registry.MustEpoch(0) qc0 := types.BuildCommitQC(ep0, keys, utils.None[*types.CommitQC](), nil) @@ -1045,25 +1038,23 @@ func TestRunEpochAdvance_CatchUpDoesNotKill(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - registry.AdvanceIfNeeded(epoch.LastRoad(1)) - registry.AdvanceIfNeeded(epoch.LastRoad(2)) + registry, keys := epoch.GenRegistryThrough(rng, 4, 4) ds := newTestDataState(&data.Config{Registry: registry}) state, err := NewState(keys[0], ds, utils.None[string]()) require.NoError(t, err) anchors := anchorWalk(t, registry, keys) - pruneToAnchors(state, anchors[:1]) - registry.PruneBefore(2) + pruneToAnchors(state, anchors[:2]) + require.NoError(t, registry.PruneBefore(3)) var runErr error go func() { runErr = state.runEpochAdvance(ctx) }() synctest.Wait() require.Nil(t, runErr) require.Equal(t, types.EpochIndex(0), state.Epoch().Load().EpochIndex(), - "epoch 1 is pruned: park instead of dying") + "epoch 2 is pruned: park instead of dying") - qc := pruneToAnchors(state, anchors[1:]) + qc := pruneToAnchors(state, anchors[2:]) synctest.Wait() require.Nil(t, runErr) require.Equal(t, types.EpochIndex(3), state.Epoch().Load().EpochIndex()) @@ -1123,9 +1114,7 @@ func TestMarkCommitQCsPersisted_RefreshesSpecWhileEpochAdvanceParked(t *testing. func TestMarkCommitQCsPersisted_DoesNotRewindPruneTip(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 2) - registry.AdvanceIfNeeded(epoch.LastRoad(1)) - registry.AdvanceIfNeeded(epoch.LastRoad(2)) + registry, keys := epoch.GenRegistryThrough(rng, 2, 3) ds := newTestDataState(&data.Config{Registry: registry}) state, err := NewState(keys[0], ds, utils.None[string]()) require.NoError(t, err) diff --git a/sei-tendermint/internal/autobahn/avail/subscriptions_test.go b/sei-tendermint/internal/autobahn/avail/subscriptions_test.go index b36453f0f9..10ef3bc956 100644 --- a/sei-tendermint/internal/autobahn/avail/subscriptions_test.go +++ b/sei-tendermint/internal/autobahn/avail/subscriptions_test.go @@ -5,7 +5,6 @@ import ( "fmt" "testing" "testing/synctest" - "time" "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/data" @@ -51,8 +50,7 @@ func TestSubscribeLaneProposals_StayLeaveRejoin(t *testing.T) { require.NoError(t, err) require.Equal(t, want0.Msg().Block().Header().Hash(), got0.Msg().Block().Header().Hash()) - // Stay: epoch 1 is already seeded at genesis with the same committee; advance - // into it without ActivateEpoch (which must not rewrite existing epochs). + // Stay: epoch 1 is already seeded at genesis with the same committee. var want1, got1 *types.Signed[*types.LaneProposal] require.NoError(t, scope.Run(ctx, func(ctx context.Context, sc scope.Scope) error { sc.SpawnBgNamed("runEpochAdvance", func() error { @@ -80,13 +78,8 @@ func TestSubscribeLaneProposals_StayLeaveRejoin(t *testing.T) { // Leave: peer drops from committee at epoch 2 (first vacant after genesis seeds). // Anchor-epoch prune drops closed lane maps (same path as runEvict) and ends the subscribe. - epLeave, err := registry.ActivateEpoch( - 0, - map[types.PublicKey]uint64{b.Public(): 1}, - time.Time{}, registry.FirstBlock(), - ) - require.NoError(t, err) - require.Equal(t, types.EpochIndex(2), epLeave.EpochIndex()) + require.NoError(t, registry.StageAndActivate(0, map[types.PublicKey]uint64{b.Public(): 1})) + epLeave := registry.MustEpoch(2) require.NoError(t, scope.Run(ctx, func(ctx context.Context, sc scope.Scope) error { sc.SpawnBgNamed("runEpochAdvance", func() error { return utils.IgnoreCancel(state.runEpochAdvance(ctx)) @@ -126,14 +119,10 @@ func TestSubscribeLaneProposals_StayLeaveRejoin(t *testing.T) { gotLane = lane return nil }) - epJoin, err := registry.ActivateEpoch( - epLeave.EpochIndex(), - map[types.PublicKey]uint64{a.Public(): 1, b.Public(): 1}, - time.Time{}, registry.FirstBlock(), - ) - if err != nil { + if err := registry.StageAndActivate(1, map[types.PublicKey]uint64{a.Public(): 1, b.Public(): 1}); err != nil { return err } + epJoin := registry.MustEpoch(3) return TestDriveAdvance(ctx, state, keys, epJoin.EpochIndex()) })) lane1 := state.LocalLane().OrPanic("rejoin") @@ -171,10 +160,9 @@ func TestJoinerCatchup_LaneVotes(t *testing.T) { stateB := utils.OrPanic1(NewState(b, ds, utils.None[string]())) laneA := stateA.LocalLane().OrPanic("genesis") - activate := func(parent types.EpochIndex, weights map[types.PublicKey]uint64) *types.Epoch { - ep, err := registry.ActivateEpoch(parent, weights, time.Time{}, registry.FirstBlock()) - require.NoError(t, err) - return ep + register := func(end types.EpochIndex, weights map[types.PublicKey]uint64) *types.Epoch { + require.NoError(t, registry.StageAndActivate(end, weights)) + return registry.MustEpoch(end + 2) } advance := func(want types.EpochIndex) { require.NoError(t, scope.Run(ctx, func(ctx context.Context, sc scope.Scope) error { @@ -195,7 +183,7 @@ func TestJoinerCatchup_LaneVotes(t *testing.T) { onlyA := map[types.PublicKey]uint64{a.Public(): 1} block0 := produce(0) - epJoin := activate(0, both) + epJoin := register(0, both) advance(epJoin.EpochIndex()) require.Equal(t, types.EpochIndex(2), stateB.LocalLane().OrPanic("joiner").Joined) @@ -206,11 +194,11 @@ func TestJoinerCatchup_LaneVotes(t *testing.T) { require.Equal(t, block0.Msg().Block().Header().Hash(), batch[0].Msg().Header().Hash()) require.Equal(t, b.Public(), batch[0].Key()) - epLeave := activate(epJoin.EpochIndex(), onlyA) + epLeave := register(1, onlyA) advance(epLeave.EpochIndex()) block1 := produce(1) // while out; skip RecvBatch so the cursor stays behind block1 - epRejoin := activate(epLeave.EpochIndex(), both) + epRejoin := register(2, both) advance(epRejoin.EpochIndex()) require.Equal(t, types.EpochIndex(4), stateB.LocalLane().OrPanic("rejoiner").Joined) diff --git a/sei-tendermint/internal/autobahn/consensus/inner_test.go b/sei-tendermint/internal/autobahn/consensus/inner_test.go index 041ab68503..f61aa26375 100644 --- a/sei-tendermint/internal/autobahn/consensus/inner_test.go +++ b/sei-tendermint/internal/autobahn/consensus/inner_test.go @@ -149,7 +149,6 @@ func TestNewInnerEmpty(t *testing.T) { func TestNewInner_RejectsWALAheadOfSpec(t *testing.T) { rng := utils.TestRng() registry, keys := epoch.GenRegistry(rng, 4) - registry.AdvanceIfNeeded(epoch.LastRoad(0)) ep0 := registry.MustEpoch(0) ep1 := registry.MustEpoch(1) @@ -180,7 +179,6 @@ func TestNewInner_RejectsWALAheadOfSpec(t *testing.T) { func TestRestore_BoundaryCatchUpSpecCoversWAL(t *testing.T) { rng := utils.TestRng() registry, keys := epoch.GenRegistry(rng, 4) - registry.AdvanceIfNeeded(epoch.LastRoad(0)) ds := newTestDataState(registry) av, err := avail.NewState(keys[0], ds, utils.None[string]()) diff --git a/sei-tendermint/internal/autobahn/consensus/persist/commitqcs_test.go b/sei-tendermint/internal/autobahn/consensus/persist/commitqcs_test.go index bdde356f20..32029cde83 100644 --- a/sei-tendermint/internal/autobahn/consensus/persist/commitqcs_test.go +++ b/sei-tendermint/internal/autobahn/consensus/persist/commitqcs_test.go @@ -7,11 +7,19 @@ import ( "time" "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/epoch" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" ) +func genTestCommittee(rng utils.Rng, size int) (*types.Committee, []types.SecretKey) { + keys := utils.GenSliceN(rng, size, types.GenSecretKey) + weights := make(map[types.PublicKey]uint64, size) + for _, key := range keys { + weights[key.Public()] = 1 + } + return utils.OrPanic1(types.NewCommittee(weights)), keys +} + // liveCommitQCs drops QCs the prune anchor has moved past, mirroring the filter loadPersistedState // applies in the avail package. Pruning reclaims whole WAL files, so a pruned QC can still be on disk // when the persister reloads; only what the anchor considers live is asserted on here. @@ -61,8 +69,7 @@ func TestNewCommitQCPersisterEmptyDir(t *testing.T) { func TestPersistCommitQCAndLoad(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 3) @@ -88,8 +95,7 @@ func TestPersistCommitQCAndLoad(t *testing.T) { func TestCommitQCDeleteBeforeRemovesOldKeepsNew(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 5) @@ -110,8 +116,7 @@ func TestCommitQCDeleteBeforeRemovesOldKeepsNew(t *testing.T) { func TestCommitQCDeleteBeforeZero(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 3) @@ -135,8 +140,7 @@ func TestCommitQCDeleteBeforeZero(t *testing.T) { func TestCommitQCPersistDuplicateIsNoOp(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 3) @@ -153,8 +157,7 @@ func TestCommitQCPersistDuplicateIsNoOp(t *testing.T) { func TestCommitQCPersistGapRejected(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 5) @@ -175,8 +178,7 @@ func TestCommitQCPersistGapRejected(t *testing.T) { // ending at the newest QC is loaded. func TestLoadAllDropsCommitQCsBehindGap(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() // Build 3 sequential CommitQCs (indices 0, 1, 2). @@ -200,8 +202,7 @@ func TestLoadAllDropsCommitQCsBehindGap(t *testing.T) { func TestNoOpCommitQCPersister(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) qcs := makeSequentialCommitQCs(committee, keys, 11) // Fresh no-op persister: persist sequential QCs and track Next. @@ -221,8 +222,7 @@ func TestNoOpCommitQCPersister(t *testing.T) { func TestCommitQCDeleteBeforePastAll(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 12) @@ -252,8 +252,7 @@ func TestCommitQCDeleteBeforePastAll(t *testing.T) { // must re-establish the cursor so subsequent persists succeed. func TestCommitQCDeleteBeforePastAllCrashRecovery(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 12) @@ -290,8 +289,7 @@ func TestCommitQCDeleteBeforePastAllCrashRecovery(t *testing.T) { // re-establishes the cursor for subsequent writes. func TestCommitQCDeleteBeforeWithAnchorRecovers(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 5) @@ -319,8 +317,7 @@ func TestCommitQCDeleteBeforeWithAnchorRecovers(t *testing.T) { func TestCommitQCDeleteBeforeThenPersistMore(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 6) @@ -344,8 +341,7 @@ func TestCommitQCDeleteBeforeThenPersistMore(t *testing.T) { func TestCommitQCDeleteBeforeAlreadyPruned(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 5) @@ -372,8 +368,7 @@ func TestCommitQCDeleteBeforeAlreadyPruned(t *testing.T) { func TestCommitQCProgressiveDeleteBefore(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - committee := registry.MustEpoch(0).Committee() + committee, keys := genTestCommittee(rng, 4) dir := t.TempDir() qcs := makeSequentialCommitQCs(committee, keys, 8) diff --git a/sei-tendermint/internal/autobahn/data/state.go b/sei-tendermint/internal/autobahn/data/state.go index 03f4391f48..4425da4257 100644 --- a/sei-tendermint/internal/autobahn/data/state.go +++ b/sei-tendermint/internal/autobahn/data/state.go @@ -19,6 +19,10 @@ const blocksCacheSize = 4000 // next CommitQC range waiting for execution. var ErrOutOfOrder = errors.New("out of order") +// ErrAppHashDivergence is returned when a quorum-certified AppHash differs from +// the one local execution produced. +var ErrAppHashDivergence = errors.New("AppHash divergence") + // Config is the config for the data State. type Config struct { // Registry is the authoritative source of committee and stake information. @@ -48,7 +52,7 @@ type inner struct { // first is the exclusive low end of retained in-memory state: maps keep [first, next*). // Advanced by runPersist(). Durable copies below first live in BlockStore. - // + // Heights below first have a persisted AppQC that matched local execution. // first <= nextAppQC <= nextAppProposal <= nextBlock <= nextQC first types.GlobalBlockNumber nextAppQC types.GlobalBlockNumber @@ -136,6 +140,23 @@ func (i *inner) insertAppQC(appQC *types.AppQC) error { return nil } +// appQCCertifiesRoad reports whether road has an AppQC matching local execution. +// Roads below first are already certified by a persisted matching AppQC. +func (i *inner) appQCCertifiesRoad(road types.RoadIndex) bool { + if i.first < i.nextQC && i.qcs[i.first].qc.QC().Proposal().Index() > road { + return true + } + for n := i.first; n < i.nextQC; { + p := i.qcs[n].qc.QC().Proposal() + if p.Index() == road { + return n < i.nextAppQC && + bytes.Equal(i.appProposals[n].AppHash(), i.appQCs[n].Proposal().AppHash()) + } + n = p.GlobalRange().Next + } + return false +} + func (i *inner) insertAppProposal(appProposal *types.AppProposal) error { gr := appProposal.GlobalRange() if gr.Next <= i.nextAppProposal { @@ -229,12 +250,16 @@ func NewState(cfg *Config, blockStore types.BlockStore) (*State, error) { m.NextBlock.Execute.Set(utils.Clamp[int64](inner.nextAppProposal)) m.NextBlock.Certify.Set(utils.Clamp[int64](inner.nextAppQC)) m.NextBlock.Evict.Set(utils.Clamp[int64](inner.first)) - return &State{ + s := &State{ cfg: cfg, metrics: m, inner: utils.NewWatch(inner), blockStore: blockStore, - }, nil + } + if err := s.activateStagedEpoch(); err != nil { + return nil, fmt.Errorf("activateStagedEpoch: %w", err) + } + return s, nil } // loadFromBlockStore replays the persisted suffix from blockStore into s.inner. @@ -244,14 +269,6 @@ func loadFromBlockStore(cfg *Config, blockStore types.BlockStore) (*inner, error if err != nil { return nil, fmt.Errorf("blockStore.ReadSuffix(): %w", err) } - var commitSpan utils.Option[types.RoadRange] - if qcs := suffix.CommitQCs; len(qcs) > 0 { - commitSpan = utils.Some(types.RoadRange{ - First: qcs[0].Index(), - Next: qcs[len(qcs)-1].Index() + 1, - }) - } - cfg.Registry.SetupInitialEpochs(commitSpan) firstBlock := cfg.Registry.FirstBlock() status := suffix.Status.Or(types.SuffixRange{ First: firstBlock, @@ -301,8 +318,6 @@ func loadFromBlockStore(cfg *Config, blockStore types.BlockStore) (*inner, error if err := inner.insertAppProposal(appProposal); err != nil { return nil, fmt.Errorf("load AppProposal from BlockStore: %w", err) } - // Match PushAppHash: do not rely only on SetupInitialEpochs for NextCommitEpoch. - cfg.Registry.AdvanceIfNeeded(appProposal.RoadIndex()) inner.publishNextCommitEpoch(cfg.Registry) } for _, appQC := range suffix.AppQCs { @@ -321,6 +336,30 @@ func (s *State) First() types.GlobalBlockNumber { // Registry returns the epoch registry. func (s *State) Registry() *epoch.Registry { return s.cfg.Registry } +// activateStagedEpoch publishes the staged committee when LastRoad of the +// deriving epoch has a matching AppQC, or when that road is already below first. +func (s *State) activateStagedEpoch() error { + idx, ok := s.cfg.Registry.Pending().Get() + if !ok { + return nil + } + for inner := range s.inner.Lock() { + endEpoch := idx - 2 + if !inner.appQCCertifiesRoad(epoch.LastRoad(endEpoch)) { + return nil + } + break + } + if err := s.cfg.Registry.ActivateEpoch(idx); err != nil { + return fmt.Errorf("activate epoch %d: %w", idx, err) + } + for inner := range s.inner.Lock() { + inner.publishNextCommitEpoch(s.cfg.Registry) + return nil + } + panic("unreachable") +} + // insertBlocksByHash matches byHash against stored (already verified) QC // headers over gr ∩ [nextBlock, nextQC) and inserts hits. Advances nextBlock // when the contiguous prefix grows. Caller must hold inner's lock. @@ -638,15 +677,22 @@ func (s *State) globalBlockByHashFromDB(hash types.BlockHeaderHash) (utils.Optio return utils.Some(assembleGlobalBlock(bn.Number, bn.Block, qc)), nil } -// PushAppHash marks blocks up to n as executed and advances the epoch -// registry when n closes a CommitQC at an epoch boundary. -func (s *State) PushAppHash(ctx context.Context, n types.GlobalBlockNumber, hash types.AppHash) error { +// PushAppHash records the AppHash for executed blocks through n and stages +// C_{E+2} when n closes LastRoad(E). +func (s *State) PushAppHash(ctx context.Context, n types.GlobalBlockNumber, hash types.AppHash, weights map[types.PublicKey]uint64) error { + var endEpoch utils.Option[types.EpochIndex] for inner, ctrl := range s.inner.Lock() { if err := ctrl.WaitUntil(ctx, func() bool { return n < inner.nextBlock }); err != nil { return err } if n < inner.nextAppProposal { - return nil + if n >= inner.first { + p := inner.qcs[n].qc.QC().Proposal() + if n == p.GlobalRange().Next-1 { + endEpoch = epoch.ClosingEpoch(p.Index()) + } + } + break } p := inner.qcs[n].qc.QC().Proposal() if next, first := inner.nextAppProposal, p.GlobalRange().First; next < first { @@ -674,13 +720,7 @@ func (s *State) PushAppHash(ctx context.Context, n types.GlobalBlockNumber, hash inner.nextAppProposal += 1 } s.metrics.NextBlock.Execute.Set(utils.Clamp[int64](inner.nextAppProposal)) - // Seed cursor: at LastRoad(N) register N+1 so runEpochAdvance can advance - // it once seal and the prune/execution leashes are met. N+2 is not needed — - // ConsensusSpec withholds the RoadIndex after LastRoad(N+1) until this fires - // again. - s.cfg.Registry.AdvanceIfNeeded(p.Index()) - // Idle boundary: no further CommitQC will republish after registration. - inner.publishNextCommitEpoch(s.cfg.Registry) + endEpoch = epoch.ClosingEpoch(p.Index()) ctrl.Updated() // CRITICAL: We need to persist AppHash before we return and start executing the next block, // otherwise we lose the apphash on restart. @@ -689,7 +729,12 @@ func (s *State) PushAppHash(ctx context.Context, n types.GlobalBlockNumber, hash return err } } - return nil + if e, ok := endEpoch.Get(); ok { + if err := s.cfg.Registry.StageEpoch(e, weights); err != nil { + return fmt.Errorf("StageEpoch(%d): %w", e, err) + } + } + return s.activateStagedEpoch() } func (s *State) PushGasUsed(gasUsed int64) { @@ -736,9 +781,9 @@ func (s *State) PushAppQC(ctx context.Context, appQC *types.AppQC) error { } s.metrics.NextBlock.Certify.Set(utils.Clamp[int64](inner.nextAppQC)) ctrl.Updated() - return nil + break } - panic("unreachable") + return s.activateStagedEpoch() } func (s *State) AppQC(ctx context.Context, n types.GlobalBlockNumber) (*types.AppQC, error) { @@ -821,6 +866,7 @@ func (s *State) PruneBefore(retainFrom types.GlobalBlockNumber) error { // order and flushes once per batch. persisted.NextBlock advances with the // block tip to unblock PushAppHash only when data is durable. // Errors propagate vertically (kill the component). +// AppQCs are persisted only when their AppHash matches local execution. // // Cursors seed from BlockStore.Status() when non-zero so PushQC-before-Run heights // are not skipped. When a tip is zero, seed from the recovery floor, never bare @@ -867,7 +913,11 @@ func (s *State) runPersist(ctx context.Context) error { status.NextAppProposal = appProposal.GlobalRange().Next } for status.NextAppQC < inner.nextAppQC { - appQC := inner.appQCs[status.NextAppQC] + n := status.NextAppQC + appQC := inner.appQCs[n] + if got, want := inner.appProposals[n].AppHash(), appQC.Proposal().AppHash(); !bytes.Equal(got, want) { + return fmt.Errorf("%w at block %v: local AppHash = %v, quorum AppHash = %v", ErrAppHashDivergence, n, got, want) + } appQCs = append(appQCs, appQC) status.NextAppQC = appQC.Proposal().GlobalRange().Next status.First = status.NextAppQC - 1 @@ -900,16 +950,13 @@ func (s *State) runPersist(ctx context.Context) error { return fmt.Errorf("flush BlockStore: %w", err) } // Prune the inner state. + var pruneEpoch utils.Option[types.EpochIndex] for inner, ctrl := range s.inner.Lock() { inner.persisted = status t := time.Now() from := inner.first for inner.first < inner.persisted.First { - // Divergence detection n := inner.first - if got, want := inner.appProposals[n].AppHash(), inner.appQCs[n].Proposal().AppHash(); !bytes.Equal(got, want) { - return fmt.Errorf("AppHash divergence detected at block %v: local AppHash = %v, quorum Apphash = %v", n, got, want) - } b := inner.blocks[n] latency := t.Sub(b.Payload().CreatedAt()).Seconds() s.metrics.BlockLatency.Evict.Observe(latency) @@ -928,10 +975,15 @@ func (s *State) runPersist(ctx context.Context) error { if !ok { return fmt.Errorf("evict advanced first but Anchor is None") } - s.cfg.Registry.PruneBefore(a.Epoch.EpochIndex()) + pruneEpoch = utils.Some(a.Epoch.EpochIndex()) } ctrl.Updated() } + if idx, ok := pruneEpoch.Get(); ok { + if err := s.cfg.Registry.PruneBefore(idx); err != nil { + return fmt.Errorf("prune epoch registry before %d: %w", idx, err) + } + } } } diff --git a/sei-tendermint/internal/autobahn/data/state_recovery_test.go b/sei-tendermint/internal/autobahn/data/state_recovery_test.go index c23384575a..7dc4439532 100644 --- a/sei-tendermint/internal/autobahn/data/state_recovery_test.go +++ b/sei-tendermint/internal/autobahn/data/state_recovery_test.go @@ -375,7 +375,7 @@ func TestRunPersistSeedsFromRecoveryFloor(t *testing.T) { i++ } for n := gr2.First; n < gr2.Next; n++ { - if err := state.PushAppHash(ctx, n, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, n, types.GenAppHash(rng), nil); err != nil { return err } } @@ -424,44 +424,193 @@ func TestRecoveryBlockGap(t *testing.T) { require.Equal(t, mid, state.NextBlock(), "replay must resume at the first unfilled number") } -func TestNewState_SetupInitialEpochsFromCommitQCSpan(t *testing.T) { +func TestNewState_NextCommitEpochAtBoundaryTip(t *testing.T) { rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 4) - qc, blocks := TestCommitQC(rng, registry.MustEpoch(0), keys, utils.None[*types.CommitQC]()) + registry, keys := epoch.GenRegistryThrough(rng, 3, 2) + ep2 := registry.MustEpoch(2) + ep1 := registry.MustEpoch(1) - store := newMemoryBlockStore(t) + qc, blocks := commitQCAtRoad(ep1, keys, epoch.LastRoad(1), ep1.FirstBlock()) + db := newTestBlockStore(t, t.TempDir()) + writeToBlockStore(t, db, []*types.FullCommitQC{qc}, [][]*types.Block{blocks}) + writeAppDataToBlockStore(t, rng, db, keys, qc) + + state := newTestState(t, &Config{Registry: registry}, db) + require.Equal(t, ep2, state.NextCommitEpoch().Load()) +} + +// writeExecutedTip writes a one-block CommitQC and its AppProposal at road. +func writeExecutedTip( + t *testing.T, + store types.BlockStore, + ep *types.Epoch, + keys []types.SecretKey, + road types.RoadIndex, + appHash types.AppHash, +) types.GlobalBlockNumber { + t.Helper() + qc, blocks := commitQCAtRoad(ep, keys, road, ep.FirstBlock()) writeToBlockStore(t, store, []*types.FullCommitQC{qc}, [][]*types.Block{blocks}) + utils.OrPanic(store.WriteAppProposal(types.NewAppProposal(qc.QC().Proposal(), appHash))) + utils.OrPanic(store.Flush()) + return qc.QC().GlobalRange().Next - 1 +} - _, err := registry.EpochAt(epoch.FirstRoad(1)) - require.NoError(t, err, "precondition: genesis epochs 0 and 1 are registered") +func TestPushAppHash_ReplayClosingRoadStagesEpoch(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + keeper := keys[0].Public() + appHash := types.GenAppHash(rng) + store := newMemoryBlockStore(t) + n := writeExecutedTip(t, store, registry.MustEpoch(0), keys, epoch.LastRoad(0), appHash) + state := newTestState(t, &Config{Registry: registry}, store) + _, err := registry.EpochAt(epoch.FirstRoad(2)) + require.Error(t, err) + + staged := utils.Some(types.EpochIndex(2)) + require.NoError(t, state.PushAppHash(ctx, n, appHash, map[types.PublicKey]uint64{keeper: 9})) + require.Equal(t, staged, registry.Pending()) _, err = registry.EpochAt(epoch.FirstRoad(2)) - require.Error(t, err, "precondition: epoch 2 absent before NewState") + require.Error(t, err) - _, err = NewState(&Config{Registry: registry}, store) - require.NoError(t, err) + require.NoError(t, state.PushAppHash(ctx, n, appHash, map[types.PublicKey]uint64{keeper: 9})) + require.Equal(t, staged, registry.Pending()) + require.Error(t, state.PushAppHash(ctx, n, appHash, map[types.PublicKey]uint64{keys[1].Public(): 1})) + require.Equal(t, staged, registry.Pending()) +} + +func TestPushAppHash_RestartAtEpoch1KeepsLastRoadStake(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + base, keys := epoch.GenRegistry(rng, 3) + keeper := keys[0].Public() + later := keys[1].Public() + endWeights := map[types.PublicKey]uint64{keeper: 9} + laterWeights := map[types.PublicKey]uint64{later: 4} + + regDir := utils.Some(t.TempDir()) + registry := utils.OrPanic1(epoch.NewRegistry( + base.MustEpoch(0).Committee(), + base.FirstBlock(), + base.GenesisTimestamp(), + regDir, + )) + storeDir := t.TempDir() + store := newTestBlockStore(t, storeDir) + state := newTestState(t, &Config{Registry: registry}, store) + ep0 := registry.MustEpoch(0) + ep1 := registry.MustEpoch(1) - for _, idx := range []types.EpochIndex{0, 1} { - if _, err := registry.EpochAt(epoch.FirstRoad(idx)); err != nil { - t.Fatalf("EpochAt(epoch %d) after NewState: %v", idx, err) + var n1 types.GlobalBlockNumber + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) + qc0, blocks0 := commitQCAtRoad(ep0, keys, epoch.LastRoad(0), ep0.FirstBlock()) + if err := state.PushQC(ctx, qc0, blocks0); err != nil { + return err } - } - if _, err := registry.EpochAt(epoch.FirstRoad(2)); err == nil { - t.Fatal("epoch 2 should not be seeded from a single epoch-0 CommitQC") - } + n0 := qc0.QC().GlobalRange().Next - 1 + if err := state.PushAppHash(ctx, n0, types.GenAppHash(rng), endWeights); err != nil { + return err + } + qc1, blocks1 := commitQCAtRoad(ep1, keys, epoch.FirstRoad(1), qc0.QC().GlobalRange().Next) + if err := state.PushQC(ctx, qc1, blocks1); err != nil { + return err + } + n1 = qc1.QC().GlobalRange().Next - 1 + return state.PushAppHash(ctx, n1, types.GenAppHash(rng), laterWeights) + })) + require.Equal(t, utils.Some(types.EpochIndex(2)), registry.Pending()) + require.NoError(t, store.Close()) + + registry2 := utils.OrPanic1(epoch.NewRegistry( + base.MustEpoch(0).Committee(), + base.FirstBlock(), + base.GenesisTimestamp(), + regDir, + )) + store2 := newTestBlockStore(t, storeDir) + state2 := newTestState(t, &Config{Registry: registry2}, store2) + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state2.Run(ctx)) }) + return state2.PushAppHash(ctx, n1, types.GenAppHash(rng), laterWeights) + })) + require.Equal(t, utils.Some(types.EpochIndex(2)), registry2.Pending()) + require.NoError(t, registry2.ActivateEpoch(2)) + got := registry2.MustEpoch(2).Committee() + require.Equal(t, uint64(9), got.Weight(keeper)) + require.False(t, got.HasReplica(later)) } -func TestNewState_NextCommitEpochAtBoundaryTip(t *testing.T) { +func TestPushAppHash_ReplayMidRangeOfClosingRoadDoesNotStage(t *testing.T) { + ctx := t.Context() rng := utils.TestRng() registry, keys := epoch.GenRegistry(rng, 3) - ep1 := registry.MustEpoch(1) + keeper := keys[0].Public() + appHash := types.GenAppHash(rng) + store := newMemoryBlockStore(t) + ep := registry.MustEpoch(0) + qc, blocks := commitQCAtRoadBlocks(ep, keys, epoch.LastRoad(0), ep.FirstBlock(), 2) + writeToBlockStore(t, store, []*types.FullCommitQC{qc}, [][]*types.Block{blocks}) + utils.OrPanic(store.WriteAppProposal(types.NewAppProposal(qc.QC().Proposal(), appHash))) + utils.OrPanic(store.Flush()) + state := newTestState(t, &Config{Registry: registry}, store) - qc, blocks := commitQCAtRoad(ep1, keys, epoch.LastRoad(1), ep1.FirstBlock()) - db := newTestBlockStore(t, t.TempDir()) - writeToBlockStore(t, db, []*types.FullCommitQC{qc}, [][]*types.Block{blocks}) - writeAppDataToBlockStore(t, rng, db, keys, qc) + gr := qc.QC().GlobalRange() + require.Equal(t, uint64(2), gr.Len()) + mid, last := gr.First, gr.Next-1 + require.NoError(t, state.PushAppHash(ctx, mid, appHash, map[types.PublicKey]uint64{keeper: 9})) + require.Equal(t, utils.None[types.EpochIndex](), registry.Pending()) - state := newTestState(t, &Config{Registry: registry}, db) - ep2, err := registry.EpochAt(epoch.FirstRoad(2)) - require.NoError(t, err) - require.Equal(t, ep2, state.NextCommitEpoch().Load()) + require.NoError(t, state.PushAppHash(ctx, last, appHash, map[types.PublicKey]uint64{keeper: 9})) + require.Equal(t, utils.Some(types.EpochIndex(2)), registry.Pending()) +} + +func TestPushAppHash_ReplayMidEpochDoesNotStage(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + appHash := types.GenAppHash(rng) + store := newMemoryBlockStore(t) + n := writeExecutedTip(t, store, registry.MustEpoch(0), keys, epoch.FirstRoad(0), appHash) + state := newTestState(t, &Config{Registry: registry}, store) + + require.NoError(t, state.PushAppHash(ctx, n, appHash, map[types.PublicKey]uint64{keys[0].Public(): 9})) + require.Equal(t, utils.None[types.EpochIndex](), registry.Pending()) + _, err := registry.EpochAt(epoch.FirstRoad(2)) + require.Error(t, err) +} + +func TestNewState_ActivatesStagedEpochFromPersistedAppQC(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + keeper := keys[0].Public() + ep := registry.MustEpoch(0) + store := newMemoryBlockStore(t) + qc, blocks := commitQCAtRoad(ep, keys, epoch.LastRoad(0), ep.FirstBlock()) + writeToBlockStore(t, store, []*types.FullCommitQC{qc}, [][]*types.Block{blocks}) + writeAppDataToBlockStore(t, rng, store, keys, qc) + + require.NoError(t, registry.StageEpoch(0, map[types.PublicKey]uint64{keeper: 9})) + _, err := registry.EpochByIndex(2) + require.Error(t, err) + + _ = newTestState(t, &Config{Registry: registry}, store) + require.Equal(t, uint64(9), registry.MustEpoch(2).Committee().Weight(keeper)) + require.Equal(t, utils.None[types.EpochIndex](), registry.Pending()) +} + +func TestNewState_LeavesStagedEpochWithoutAppQC(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + keeper := keys[0].Public() + appHash := types.GenAppHash(rng) + store := newMemoryBlockStore(t) + _ = writeExecutedTip(t, store, registry.MustEpoch(0), keys, epoch.LastRoad(0), appHash) + require.NoError(t, registry.StageEpoch(0, map[types.PublicKey]uint64{keeper: 9})) + + _ = newTestState(t, &Config{Registry: registry}, store) + require.Equal(t, utils.Some(types.EpochIndex(2)), registry.Pending()) + _, err := registry.EpochByIndex(2) + require.Error(t, err) } diff --git a/sei-tendermint/internal/autobahn/data/state_test.go b/sei-tendermint/internal/autobahn/data/state_test.go index 851bff702d..7390fb8691 100644 --- a/sei-tendermint/internal/autobahn/data/state_test.go +++ b/sei-tendermint/internal/autobahn/data/state_test.go @@ -107,7 +107,7 @@ func pushAppHashesRunning(ctx context.Context, state *State, rng utils.Rng, firs return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { s.SpawnBgNamed("state.Run", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) for n := first; n < next; n++ { - if err := state.PushAppHash(ctx, n, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, n, types.GenAppHash(rng), nil); err != nil { return err } } @@ -138,10 +138,30 @@ func commitQCAtRoad( return types.NewFullCommitQC(types.NewCommitQC(votes), []*types.BlockHeader{block.Header()}), []*types.Block{block} } +func commitQCAtRoadBlocks( + ep *types.Epoch, + keys []types.SecretKey, + road types.RoadIndex, + globalFirst types.GlobalBlockNumber, + n int, +) (*types.FullCommitQC, []*types.Block) { + proposal, blocks := types.ProposalAtBlocks(ep, types.View{Index: road, Number: 0}, globalFirst, n) + votes := make([]*types.Signed[*types.CommitVote], 0, len(keys)) + for _, k := range keys { + votes = append(votes, types.Sign(k, types.NewCommitVote(proposal))) + } + headers := make([]*types.BlockHeader, len(blocks)) + for i, b := range blocks { + headers[i] = b.Header() + } + return types.NewFullCommitQC(types.NewCommitQC(votes), headers), blocks +} + func TestNextCommitEpoch_AdvancesAtIdleEpochBoundary(t *testing.T) { ctx := t.Context() rng := utils.TestRng() - registry, keys := epoch.GenRegistry(rng, 3) + registry, keys := epoch.GenRegistryThrough(rng, 3, 2) + ep2 := registry.MustEpoch(2) ep1 := registry.MustEpoch(1) state := newTestState(t, &Config{Registry: registry}, newTestBlockStore(t, t.TempDir())) @@ -150,19 +170,55 @@ func TestNextCommitEpoch_AdvancesAtIdleEpochBoundary(t *testing.T) { require.Equal(t, ep1, state.NextCommitEpoch().Load(), "mid-epoch next road is still in epoch 1") grMid := qcMid.QC().GlobalRange() require.NoError(t, pushAppHashesRunning(ctx, state, rng, grMid.First, grMid.Next)) - require.Equal(t, ep1, state.NextCommitEpoch().Load(), "mid-epoch AppHash must not seed epoch 2") + require.Equal(t, ep1, state.NextCommitEpoch().Load()) qcLast, blocksLast := commitQCAtRoad(ep1, keys, epoch.LastRoad(1), grMid.Next) require.NoError(t, state.PushQC(ctx, qcLast, blocksLast)) - _, err := registry.EpochAt(epoch.FirstRoad(2)) - require.Error(t, err, "epoch 2 must stay unregistered until the boundary AppProposal lands") - require.Equal(t, ep1, state.NextCommitEpoch().Load(), "unregistered next epoch: previous publish stands") + require.Equal(t, ep2, state.NextCommitEpoch().Load(), "next road is in epoch 2, already filled from end(0)") +} - grLast := qcLast.QC().GlobalRange() - require.NoError(t, pushAppHashesRunning(ctx, state, rng, grLast.First, grLast.Next)) - ep2, err := registry.EpochAt(epoch.FirstRoad(2)) - require.NoError(t, err) - require.Equal(t, ep2, state.NextCommitEpoch().Load()) +func TestNextCommitEpoch_RefreshesAfterAppQCActivation(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + keeper := keys[0].Public() + ep0 := registry.MustEpoch(0) + ep1 := registry.MustEpoch(1) + state := newTestState(t, &Config{Registry: registry}, newTestBlockStore(t, t.TempDir())) + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) + qc0, blocks0 := commitQCAtRoad(ep0, keys, epoch.LastRoad(0), ep0.FirstBlock()) + if err := state.PushQC(ctx, qc0, blocks0); err != nil { + return err + } + n0 := qc0.QC().GlobalRange().Next - 1 + if err := state.PushAppHash(ctx, n0, types.GenAppHash(rng), map[types.PublicKey]uint64{keeper: 9}); err != nil { + return err + } + if got := state.NextCommitEpoch().Load(); got != ep1 { + return fmt.Errorf("after LastRoad(0) QC: NextCommitEpoch = %v, want epoch 1", got.EpochIndex()) + } + + qc1, blocks1 := commitQCAtRoad(ep1, keys, epoch.LastRoad(1), qc0.QC().GlobalRange().Next) + if err := state.PushQC(ctx, qc1, blocks1); err != nil { + return err + } + if got := state.NextCommitEpoch().Load(); got != ep1 { + return fmt.Errorf("LastRoad(1) QC before epoch 2 is live: NextCommitEpoch = %v, want epoch 1", got.EpochIndex()) + } + + if err := pushAppQCForBlock(ctx, state, keys, n0); err != nil { + return err + } + ep2, err := registry.EpochByIndex(2) + if err != nil { + return fmt.Errorf("epoch 2 after AppQC: %w", err) + } + if got := state.NextCommitEpoch().Load(); got != ep2 { + return fmt.Errorf("after AppQC: NextCommitEpoch = %v, want epoch 2", got.EpochIndex()) + } + return nil + })) } func TestState(t *testing.T) { @@ -408,13 +464,13 @@ func TestExecution(t *testing.T) { // PushAppHash for a block beyond nextBlock should not succeed: // it waits for persistence which never happens for unfinalised blocks. shortCtx, cancel := context.WithTimeout(ctx, 10*time.Millisecond) - if err := state.PushAppHash(shortCtx, gr.Next, types.GenAppHash(rng)); err == nil { + if err := state.PushAppHash(shortCtx, gr.Next, types.GenAppHash(rng), nil); err == nil { cancel() return errors.New("PushAppHash expected to fail on non-finalized blocks") } cancel() for n := gr.First; n < gr.Next; n += 1 { - if err := state.PushAppHash(ctx, n, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, n, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("state.PushAppHash(): %w", err) } } @@ -446,36 +502,195 @@ func TestPushAppHashRejectsJumpOverCommitQCRange(t *testing.T) { } qcs = append(qcs, qc.QC()) } - if err := state.PushAppHash(ctx, qcs[0].GlobalRange().Next-1, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, qcs[0].GlobalRange().Next-1, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(qc1): %w", err) } if qcs[2].GlobalRange().Len() < 2 { panic("qcs[2].Len() is too small for this test") } - if err := state.PushAppHash(ctx, qcs[2].GlobalRange().Next-2, types.GenAppHash(rng)); !errors.Is(err, ErrOutOfOrder) { + if err := state.PushAppHash(ctx, qcs[2].GlobalRange().Next-2, types.GenAppHash(rng), nil); !errors.Is(err, ErrOutOfOrder) { return fmt.Errorf("PushAppHash(qc3 before qc2) error = %w, want %w", err, ErrOutOfOrder) } - if err := state.PushAppHash(ctx, qcs[2].GlobalRange().Next-1, types.GenAppHash(rng)); !errors.Is(err, ErrOutOfOrder) { + if err := state.PushAppHash(ctx, qcs[2].GlobalRange().Next-1, types.GenAppHash(rng), nil); !errors.Is(err, ErrOutOfOrder) { return fmt.Errorf("PushAppHash(qc3 before qc2) error = %w, want %w", err, ErrOutOfOrder) } - if err := state.PushAppHash(ctx, qcs[1].GlobalRange().Next-1, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, qcs[1].GlobalRange().Next-1, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(qc2): %w", err) } - if err := state.PushAppHash(ctx, qcs[2].GlobalRange().Next-1, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, qcs[2].GlobalRange().Next-1, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(qc3): %w", err) } // Inserting old stuff should be a noop. - if err := state.PushAppHash(ctx, qcs[1].GlobalRange().Next-1, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, qcs[1].GlobalRange().Next-1, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(qc2): %w", err) } - if err := state.PushAppHash(ctx, qcs[2].GlobalRange().Next-2, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, qcs[2].GlobalRange().Next-2, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(qc2): %w", err) } return nil })) } +func TestPushAppHash_MidEpochDoesNotRegister(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + state := newTestState(t, &Config{Registry: registry}, newTestBlockStore(t, t.TempDir())) + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) + qc, blocks := TestCommitQC(rng, registry.MustEpoch(0), keys, utils.None[*types.CommitQC]()) + if err := state.PushQC(ctx, qc, blocks); err != nil { + return err + } + if err := state.PushAppHash(ctx, qc.QC().GlobalRange().Next-1, types.GenAppHash(rng), nil); err != nil { + return err + } + if _, err := registry.EpochAt(epoch.FirstRoad(2)); err == nil { + return fmt.Errorf("epoch 2 must stay absent for road %d", qc.QC().Proposal().Index()) + } + if got, ok := registry.Pending().Get(); ok { + return fmt.Errorf("Pending() = %v, want nothing staged mid-epoch", got) + } + return nil + })) +} + +func TestCommitteeFill_GatedOnAppQC(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + keeper := keys[0].Public() + state := newTestState(t, &Config{Registry: registry}, newTestBlockStore(t, t.TempDir())) + ep := registry.MustEpoch(0) + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) + qc, blocks := commitQCAtRoad(ep, keys, epoch.LastRoad(0), ep.FirstBlock()) + if err := state.PushQC(ctx, qc, blocks); err != nil { + return err + } + n := qc.QC().GlobalRange().Next - 1 + if err := state.PushAppHash(ctx, n, types.GenAppHash(rng), map[types.PublicKey]uint64{keeper: 9}); err != nil { + return err + } + if _, err := registry.EpochAt(epoch.FirstRoad(2)); err == nil { + return errors.New("epoch 2 must stay staged until an AppQC finalizes end(0)") + } + if want := utils.Some(types.EpochIndex(2)); registry.Pending() != want { + return fmt.Errorf("Pending() = %v, want %v", registry.Pending(), want) + } + + if err := pushAppQCForBlock(ctx, state, keys, n); err != nil { + return err + } + filled, err := registry.EpochAt(epoch.FirstRoad(2)) + if err != nil { + return fmt.Errorf("epoch 2 after AppQC: %w", err) + } + if filled.EpochIndex() != 2 { + return fmt.Errorf("EpochIndex = %d, want 2", filled.EpochIndex()) + } + if got := filled.Committee().Weight(keeper); got != 9 { + return fmt.Errorf("Weight = %d, want 9", got) + } + if filled.Committee().Lanes().Len() != 1 { + return fmt.Errorf("lanes = %d, want 1", filled.Committee().Lanes().Len()) + } + return nil + })) +} + +func TestCommitteeFill_AppQCDivergenceLeavesStaged(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + keeper := keys[0].Public() + state := newTestState(t, &Config{Registry: registry}, newTestBlockStore(t, t.TempDir())) + ep := registry.MustEpoch(0) + qc, blocks := commitQCAtRoad(ep, keys, epoch.LastRoad(0), ep.FirstBlock()) + n := qc.QC().GlobalRange().Next - 1 + + // Stop Run before the divergent AppQC: persisting one kills the node. + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) + if err := state.PushQC(ctx, qc, blocks); err != nil { + return err + } + return state.PushAppHash(ctx, n, types.GenAppHash(rng), map[types.PublicKey]uint64{keeper: 9}) + })) + + divergent := types.NewAppProposal(qc.QC().Proposal(), types.GenAppHash(rng)) + require.NoError(t, state.PushAppQC(ctx, TestAppQC(keys, divergent))) + _, err := registry.EpochAt(epoch.FirstRoad(2)) + require.Error(t, err, "epoch 2 must stay staged after divergence") + require.Equal(t, utils.Some(types.EpochIndex(2)), registry.Pending()) +} + +func TestCommitteeFill_LaterMatchingAppQCDoesNotSkipDivergentLastRoad(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + keeper := keys[0].Public() + state := newTestState(t, &Config{Registry: registry}, newTestBlockStore(t, t.TempDir())) + ep0 := registry.MustEpoch(0) + ep1 := registry.MustEpoch(1) + qc0, blocks0 := commitQCAtRoad(ep0, keys, epoch.LastRoad(0), ep0.FirstBlock()) + gr0 := qc0.QC().GlobalRange() + qc1, blocks1 := commitQCAtRoad(ep1, keys, epoch.FirstRoad(1), gr0.Next) + + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) + if err := state.PushQC(ctx, qc0, blocks0); err != nil { + return err + } + if err := state.PushAppHash(ctx, gr0.Next-1, types.GenAppHash(rng), map[types.PublicKey]uint64{keeper: 9}); err != nil { + return err + } + if err := state.PushQC(ctx, qc1, blocks1); err != nil { + return err + } + return state.PushAppHash(ctx, qc1.QC().GlobalRange().Next-1, types.GenAppHash(rng), nil) + })) + + divergent := types.NewAppProposal(qc0.QC().Proposal(), types.GenAppHash(rng)) + require.NoError(t, state.PushAppQC(ctx, TestAppQC(keys, divergent))) + vote1, err := state.AppVote(ctx, qc1.QC().GlobalRange().First) + require.NoError(t, err) + require.NoError(t, state.PushAppQC(ctx, TestAppQC(keys, vote1.Proposal()))) + + _, err = registry.EpochAt(epoch.FirstRoad(2)) + require.Error(t, err, "epoch 2 must stay staged when LastRoad(0) diverged") + require.Equal(t, utils.Some(types.EpochIndex(2)), registry.Pending()) +} + +func TestRunPersist_HaltsBeforePersistingDivergentAppQC(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + store := newTestBlockStore(t, t.TempDir()) + state := newTestState(t, &Config{Registry: registry}, store) + ep := registry.MustEpoch(0) + qc, blocks := commitQCAtRoad(ep, keys, epoch.FirstRoad(0), ep.FirstBlock()) + n := qc.QC().GlobalRange().Next - 1 + + require.NoError(t, scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + s.SpawnBgNamed("state.Run()", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) + if err := state.PushQC(ctx, qc, blocks); err != nil { + return err + } + return state.PushAppHash(ctx, n, types.GenAppHash(rng), nil) + })) + before := store.Status().OrPanic("status after AppHash") + + divergent := types.NewAppProposal(qc.QC().Proposal(), types.GenAppHash(rng)) + require.NoError(t, state.PushAppQC(ctx, TestAppQC(keys, divergent))) + + require.ErrorIs(t, state.runPersist(ctx), ErrAppHashDivergence) + after := store.Status().OrPanic("status after divergence") + require.Equal(t, before.NextAppQC, after.NextAppQC, "divergent AppQC must not be persisted") + require.Equal(t, before.First, after.First, "eviction floor must not advance past it") +} + func TestPushBlockAcceptsBlockWithQC(t *testing.T) { ctx := t.Context() rng := utils.TestRng() @@ -556,7 +771,7 @@ func TestPushQCBeforeRunPersistsToBlockStore(t *testing.T) { s.SpawnBgNamed("state.Run", func() error { return utils.IgnoreCancel(state.Run(ctx)) }) // PushAppHash waits on persisted.NextBlock, so success implies Flush. for n := gr1.First; n < gr1.Next; n++ { - if err := state.PushAppHash(ctx, n, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, n, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(%d): %w", n, err) } } @@ -599,7 +814,7 @@ func TestEvictionWaitsForAppQC(t *testing.T) { return fmt.Errorf("PushQC(qc1): %w", err) } for n := gr1.First; n < gr1.Next; n++ { - if err := state.PushAppHash(ctx, n, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, n, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(%d): %w", n, err) } } @@ -633,7 +848,7 @@ func TestEvictionWaitsForAppQC(t *testing.T) { return fmt.Errorf("PushQC(qc2): %w", err) } for n := gr2.First; n < gr2.Next; n++ { - if err := state.PushAppHash(ctx, n, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, n, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(%d): %w", n, err) } } @@ -714,7 +929,7 @@ func TestPushAppHashBelowAnchorSucceeds(t *testing.T) { if err := state.PushQC(ctx, qc, blocks); err != nil { return fmt.Errorf("PushQC: %w", err) } - if err := state.PushAppHash(ctx, gr.Next-1, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, gr.Next-1, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(tip): %w", err) } if err := pushAppQCForBlock(ctx, state, keys, gr.First); err != nil { @@ -731,7 +946,7 @@ func TestPushAppHashBelowAnchorSucceeds(t *testing.T) { return fmt.Errorf("state.Anchor.Wait(): %w", err) } // Pushing apphash for height below the anchor should NOT expolode. - if err := state.PushAppHash(ctx, registry.FirstBlock(), types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, registry.FirstBlock(), types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash below anchor: %w", err) } return nil @@ -759,7 +974,7 @@ func TestNextToExecuteAfterAppEviction(t *testing.T) { return fmt.Errorf("PushQC(qc1): %w", err) } for n := gr1.First; n < gr1.Next; n++ { - if err := state.PushAppHash(ctx, n, types.GenAppHash(rng)); err != nil { + if err := state.PushAppHash(ctx, n, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(%d): %w", n, err) } } @@ -843,7 +1058,7 @@ func TestPushAppQCPersistsAndRecovers(t *testing.T) { return fmt.Errorf("PushQC(qc1): %w", err) } for n := gr1.First; n < gr1.Next; n++ { - if err := state1.PushAppHash(ctx, n, types.GenAppHash(rng)); err != nil { + if err := state1.PushAppHash(ctx, n, types.GenAppHash(rng), nil); err != nil { return fmt.Errorf("PushAppHash(%d): %w", n, err) } } diff --git a/sei-tendermint/internal/autobahn/epoch/registry.go b/sei-tendermint/internal/autobahn/epoch/registry.go index efc2a3e744..b4caeda415 100644 --- a/sei-tendermint/internal/autobahn/epoch/registry.go +++ b/sei-tendermint/internal/autobahn/epoch/registry.go @@ -3,9 +3,12 @@ package epoch import ( "context" "fmt" + "maps" "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/pb" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" ) @@ -28,59 +31,137 @@ func LastRoad(idx types.EpochIndex) types.RoadIndex { return FirstRoad(idx+1) - 1 } +// ClosingEpoch returns the epoch that road ends, if road is that epoch's last +// road. Otherwise None. +func ClosingEpoch(road types.RoadIndex) utils.Option[types.EpochIndex] { + idx := IndexForRoad(road) + if road != LastRoad(idx) { + return utils.None[types.EpochIndex]() + } + return utils.Some(idx) +} + type registryState struct { m map[types.EpochIndex]*types.Epoch - // live is the supported non-zero epoch indices [First, Next). Epoch 0 is - // always kept for genesis metadata and is not represented here. + // pending is the staged committee for live.Next. + pending utils.Option[*types.Committee] + // live is the supported execution-derived epoch indices [First, Next). + // Epochs 0 and 1 are genesis and are not represented here. live types.EpochRange } -// dropped reports whether idx is below live.First. Epoch 0 is never dropped. +// dropped reports whether idx is below live.First. Epochs 0 and 1 are never dropped. func (s *registryState) dropped(idx types.EpochIndex) bool { - return idx != 0 && idx < s.live.First + return idx >= 2 && idx < s.live.First +} + +// activate makes committee live for idx. It returns an error if idx is not live.Next. +func (s *registryState) activate(idx types.EpochIndex, committee *types.Committee) error { + if idx != s.live.Next { + return fmt.Errorf("epoch %d cannot be activated, want %d", idx, s.live.Next) + } + roads := types.RoadRange{First: FirstRoad(idx), Next: FirstRoad(idx + 1)} + s.m[idx] = types.NewEpoch(idx, roads, s.m[0].FirstTimestamp(), committee, s.m[0].FirstBlock()) + s.live.Next = idx + 1 + return nil } -// Registry stores activated epochs and placeholders. +func (s *registryState) clone() *registryState { + return ®istryState{ + m: maps.Clone(s.m), + pending: s.pending, + live: s.live, + } +} + +func (s *registryState) snapshot() *pb.PersistedEpochRegistry { + snapshot := &pb.PersistedEpochRegistry{ + Live: make([]*pb.EpochRecord, 0, utils.Clamp[int](s.live.Next-s.live.First)), + } + for idx := s.live.First; idx < s.live.Next; idx++ { + snapshot.Live = append(snapshot.Live, encodeEpochRecord(idx, s.m[idx].Committee())) + } + if pending, ok := s.pending.Get(); ok { + snapshot.Pending = encodeEpochRecord(s.live.Next, pending) + } + return snapshot +} + +func (s *registryState) restore(snapshot *pb.PersistedEpochRegistry) error { + // TODO: a missing pending is restaged on restart from GetValidators() at app + // tip, not stake(LastRoad(E)). Use the LastRoad validator set once it is + // persisted in app state. + if snapshot == nil { + return fmt.Errorf("missing") + } + for pos, record := range snapshot.Live { + idx, committee, err := decodeEpochRecord(record) + if err != nil { + return fmt.Errorf("live record %d: %w", pos, err) + } + if pos == 0 { + s.live.First = idx + s.live.Next = idx + } + if err := s.activate(idx, committee); err != nil { + return fmt.Errorf("live record %d: %w", pos, err) + } + if err := s.checkDerivedFromPrev(idx, committee); err != nil { + return fmt.Errorf("live record %d: %w", pos, err) + } + } + if snapshot.Pending != nil { + idx, committee, err := decodeEpochRecord(snapshot.Pending) + if err != nil { + return fmt.Errorf("pending: %w", err) + } + if idx != s.live.Next { + return fmt.Errorf("pending epoch %d, want %d", idx, s.live.Next) + } + if err := s.checkDerivedFromPrev(idx, committee); err != nil { + return fmt.Errorf("pending: %w", err) + } + s.pending = utils.Some(committee) + } + return nil +} + +// Registry stores genesis epochs 0 and 1 plus execution-derived epochs +// published by ActivateEpoch. type Registry struct { - state utils.Watch[*registryState] + state utils.Watch[*registryState] + persister persist.Persister[*pb.PersistedEpochRegistry] } -// NewRegistry creates a Registry with genesis epochs 0 and 1 (genesis committee). +// NewRegistry returns a Registry with epochs 0 and 1 using committee, firstBlock, +// and genesisTimestamp. stateDir Some opens the epoch snapshot and restores its +// live and pending committees; None keeps the registry in memory only. func NewRegistry( committee *types.Committee, firstBlock types.GlobalBlockNumber, genesisTimestamp time.Time, + stateDir utils.Option[string], ) (*Registry, error) { ep0 := types.NewEpoch(0, types.RoadRange{First: 0, Next: FirstRoad(1)}, genesisTimestamp, committee, firstBlock) ep1 := types.NewEpoch(1, types.RoadRange{First: FirstRoad(1), Next: FirstRoad(2)}, genesisTimestamp, committee, firstBlock) - return &Registry{ - state: utils.NewWatch(®istryState{ - m: map[types.EpochIndex]*types.Epoch{0: ep0, 1: ep1}, - live: types.EpochRange{First: 1, Next: 2}, - }), - }, nil -} - -// SetupInitialEpochs registers placeholders covering commitQCs and the next epoch. -// With no CommitQCs this is a no-op (epochs 0 and 1 are already present). -func (r *Registry) SetupInitialEpochs(commitQCs utils.Option[types.RoadRange]) { - span, ok := commitQCs.Get() - if !ok { - return + state := ®istryState{ + m: map[types.EpochIndex]*types.Epoch{0: ep0, 1: ep1}, + pending: utils.None[*types.Committee](), + live: types.EpochRange{First: 2, Next: 2}, } - for s, ctrl := range r.state.Lock() { - windowFirst := IndexForRoad(span.First) - windowLast := IndexForRoad(span.Next - 1) - r.ensureAround(s, span.First) - for idx := windowFirst; idx <= windowLast; idx++ { - r.ensureLocked(s, idx) + persister, loaded, err := openEpochSnapshot(stateDir) + if err != nil { + return nil, err + } + if snapshot, ok := loaded.Get(); ok { + if err := state.restore(snapshot); err != nil { + return nil, fmt.Errorf("restore epoch snapshot: %w", err) } - r.ensureAround(s, span.Next) - // TODO: replace placeholders with execution-derived committee, - // FirstTimestamp, and FirstBlock (genesis copies feed ViewSpec). - r.ensureLocked(s, windowLast+1) - ctrl.Updated() } + return &Registry{ + state: utils.NewWatch(state), + persister: persister, + }, nil } // FirstBlock returns the genesis epoch's first global block number. @@ -130,129 +211,118 @@ func (r *Registry) EpochAt(roadIndex types.RoadIndex) (*types.Epoch, error) { panic("unreachable") } -// ActivateEpoch registers the next vacant epoch after parent. parent is the -// epoch at the execution tip; the new committee is derived from it. Already- -// registered epochs are never modified. Pruned indices are skipped. -func (r *Registry) ActivateEpoch( - parent types.EpochIndex, - weights map[types.PublicKey]uint64, - firstTimestamp time.Time, - firstBlock types.GlobalBlockNumber, -) (*types.Epoch, error) { - for s, ctrl := range r.state.Lock() { - if s.dropped(parent) { - return nil, fmt.Errorf("epoch %d: %w", parent, types.ErrPruned) +// commit writes next to disk, then replaces s with next. +// s is left unchanged if Persist fails. +func (r *Registry) commit(s, next *registryState) error { + if err := r.persister.Persist(next.snapshot()); err != nil { + return fmt.Errorf("persist epoch registry: %w", err) + } + *s = *next + return nil +} + +// StageEpoch derives C_{endEpoch+2} from weights and persists it as pending. +// A no-op if that epoch is already staged or live with the same committee. +// An error if it conflicts, if the target is not live.Next, or if endEpoch+1 +// is not live. +func (r *Registry) StageEpoch(endEpoch types.EpochIndex, weights map[types.PublicKey]uint64) error { + target := endEpoch + 2 + for s := range r.state.Lock() { + if s.dropped(target) { + return fmt.Errorf("epoch %d: %w", target, types.ErrPruned) + } + ep, registered := s.m[target] + if !registered && target != s.live.Next { + return fmt.Errorf("epoch %d cannot be staged, want %d", target, s.live.Next) } - prev, ok := s.m[parent] + prev, ok := s.m[endEpoch+1] if !ok { - return nil, fmt.Errorf("epoch %d not registered", parent) + return fmt.Errorf("epoch %d not registered", endEpoch+1) } - next := parent + 1 - for { - if s.dropped(next) { - next++ - continue - } - if _, ok := s.m[next]; !ok { - break + committee, err := prev.Committee().DeriveNext(weights, target) + if err != nil { + return fmt.Errorf("DeriveNext(%d): %w", target, err) + } + if registered { + if !ep.Committee().Equal(committee) { + return fmt.Errorf("epoch %d already registered with a different committee", target) } - next++ + return nil } - committee, err := prev.Committee().DeriveNext(weights, next) - if err != nil { - return nil, err + if staged, ok := s.pending.Get(); ok { + if !staged.Equal(committee) { + return fmt.Errorf("epoch %d already staged with a different committee", target) + } + return nil } - roads := types.RoadRange{First: FirstRoad(next), Next: FirstRoad(next + 1)} - ep := types.NewEpoch(next, roads, firstTimestamp, committee, firstBlock) - s.m[next] = ep - r.extendLive(s, next) - ctrl.Updated() - return ep, nil + next := s.clone() + next.pending = utils.Some(committee) + return r.commit(s, next) } panic("unreachable") } -// makeEpoch inserts a genesis-committee placeholder at epochIdx. -// Caller must hold r.state. Epoch 0 is always present; further epochs copy from it. -func (r *Registry) makeEpoch(s *registryState, epochIdx types.EpochIndex) *types.Epoch { - ep0 := s.m[0] - firstRoad := FirstRoad(epochIdx) - epoch := types.NewEpoch( - epochIdx, - types.RoadRange{First: firstRoad, Next: FirstRoad(epochIdx + 1)}, - ep0.FirstTimestamp(), - ep0.Committee(), - ep0.FirstBlock(), - ) - s.m[epochIdx] = epoch - r.extendLive(s, epochIdx) - return epoch -} - -func (r *Registry) extendLive(s *registryState, idx types.EpochIndex) { - if idx >= s.live.Next { - s.live.Next = idx + 1 - } -} - -// ensureLocked registers a genesis-committee placeholder for idx if missing. -// Caller must hold r.state. Pruned indices are not recreated. -func (r *Registry) ensureLocked(s *registryState, idx types.EpochIndex) { - if s.dropped(idx) { - return - } - if _, ok := s.m[idx]; !ok { - r.makeEpoch(s, idx) - } -} - -// ensureAround registers the epoch containing road and its predecessor. -// Caller must hold r.state. -func (r *Registry) ensureAround(s *registryState, road types.RoadIndex) { - center := IndexForRoad(road) - if center > 0 { - r.ensureLocked(s, center-1) +// ActivateEpoch publishes the staged committee for idx. +// A no-op if idx is already live. An error if idx is not the staged epoch. +func (r *Registry) ActivateEpoch(idx types.EpochIndex) error { + for s, ctrl := range r.state.Lock() { + if _, ok := s.m[idx]; ok { + return nil + } + if s.dropped(idx) { + return fmt.Errorf("epoch %d: %w", idx, types.ErrPruned) + } + committee, ok := s.pending.Get() + if !ok { + return fmt.Errorf("epoch %d is not staged", idx) + } + next := s.clone() + if err := next.activate(idx, committee); err != nil { + return err + } + next.pending = utils.None[*types.Committee]() + if err := r.commit(s, next); err != nil { + return err + } + ctrl.Updated() + return nil } - r.ensureLocked(s, center) + panic("unreachable") } -// AdvanceIfNeeded registers epoch M+1 when roadIndex is LastRoad(M). -// M+2 is not seeded: tip may race to LastRoad(M+1) before AppQC, but -// ConsensusSpec withholds that next RoadIndex until M+1's AppQC boundary fires -// AdvanceIfNeeded again. -func (r *Registry) AdvanceIfNeeded(roadIndex types.RoadIndex) { - tipEpoch := IndexForRoad(roadIndex) - if roadIndex != LastRoad(tipEpoch) { - return - } - for s, ctrl := range r.state.Lock() { - r.ensureLocked(s, tipEpoch+1) - ctrl.Updated() +// Pending returns the staged epoch index, or None. +func (r *Registry) Pending() utils.Option[types.EpochIndex] { + for s := range r.state.Lock() { + if _, ok := s.pending.Get(); !ok { + return utils.None[types.EpochIndex]() + } + return utils.Some(s.live.Next) } + panic("unreachable") } -// PruneBefore drops supported epochs in [live.First, keep). Epoch 0 is kept -// for genesis metadata. keep is exclusive and only moves live.First forward. -func (r *Registry) PruneBefore(keep types.EpochIndex) { +// PruneBefore drops epochs in [live.First, keep). Epochs 0 and 1 and the +// latest live epoch are kept. keep is exclusive and only moves live.First +// forward. Staged committees are not dropped. +func (r *Registry) PruneBefore(keep types.EpochIndex) error { for s, ctrl := range r.state.Lock() { + if s.live.First == s.live.Next { + return nil + } + keep = min(keep, s.live.Next-1) if keep <= s.live.First { - return + return nil } - for idx := s.live.First; idx < keep; idx++ { - delete(s.m, idx) + next := s.clone() + for idx := next.live.First; idx < keep; idx++ { + delete(next.m, idx) } - s.live.First = keep - if s.live.First > s.live.Next { - s.live.Next = s.live.First + next.live.First = keep + if err := r.commit(s, next); err != nil { + return err } ctrl.Updated() - } -} - -// Live returns the supported non-zero epoch window [First, Next). -func (r *Registry) Live() types.EpochRange { - for s := range r.state.Lock() { - return s.live + return nil } panic("unreachable") } diff --git a/sei-tendermint/internal/autobahn/epoch/registry_test.go b/sei-tendermint/internal/autobahn/epoch/registry_test.go index 2fe87a8fa3..78c131186d 100644 --- a/sei-tendermint/internal/autobahn/epoch/registry_test.go +++ b/sei-tendermint/internal/autobahn/epoch/registry_test.go @@ -7,10 +7,15 @@ import ( "time" "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pb" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" ) +type errPersister struct{ err error } + +func (e errPersister) Persist(*pb.PersistedEpochRegistry) error { return e.err } + func makeRegistry(t *testing.T) (*Registry, *types.Committee) { t.Helper() rng := utils.TestRng() @@ -19,12 +24,23 @@ func makeRegistry(t *testing.T) (*Registry, *types.Committee) { types.GenSecretKey(rng).Public(): 1, types.GenSecretKey(rng).Public(): 1, })) - r := utils.OrPanic1(NewRegistry(committee, 0, time.Time{})) + r := utils.OrPanic1(NewRegistry(committee, 0, time.Time{}, utils.None[string]())) return r, committee } -func midRoad(idx types.EpochIndex) types.RoadIndex { - return FirstRoad(idx) + EpochLength/2 +func TestClosingEpoch(t *testing.T) { + got, ok := ClosingEpoch(LastRoad(0)).Get() + require.True(t, ok) + require.Equal(t, types.EpochIndex(0), got) + + got, ok = ClosingEpoch(LastRoad(1)).Get() + require.True(t, ok) + require.Equal(t, types.EpochIndex(1), got) + + _, ok = ClosingEpoch(FirstRoad(1)).Get() + require.False(t, ok) + _, ok = ClosingEpoch(LastRoad(0) - 1).Get() + require.False(t, ok) } func TestRegistry_EpochByIndex_UnknownReturnsNotFound(t *testing.T) { @@ -64,195 +80,294 @@ func TestNewRegistry_Genesis(t *testing.T) { if _, err := r.EpochAt(FirstRoad(2)); err == nil { t.Fatal("EpochAt(FirstRoad(2)) expected error for unregistered epoch, got nil") } + ep, err := r.WaitForEpoch(t.Context(), 0) + require.NoError(t, err) + require.Equal(t, types.EpochIndex(0), ep.EpochIndex()) + ep, err = r.WaitForEpoch(t.Context(), 1) + require.NoError(t, err) + require.Equal(t, types.EpochIndex(1), ep.EpochIndex()) } -func TestEpochAt_FoundAfterAdvanceIfNeeded(t *testing.T) { - r, _ := makeRegistry(t) - if _, err := r.EpochAt(FirstRoad(1)); err != nil { - t.Fatalf("epoch 1 must be present from NewRegistry: %v", err) - } - r.AdvanceIfNeeded(0) - if _, err := r.EpochAt(FirstRoad(2)); err == nil { - t.Fatal("AdvanceIfNeeded(0) must not seed epoch 2") - } - r.AdvanceIfNeeded(LastRoad(0)) - ep, err := r.EpochAt(FirstRoad(1)) - if err != nil { - t.Fatalf("EpochAt(FirstRoad(1)) after last road of epoch 0: %v", err) - } - if ep.EpochIndex() != 1 { - t.Fatalf("EpochAt(FirstRoad(1)).EpochIndex() = %d, want 1", ep.EpochIndex()) - } - if _, err := r.EpochAt(FirstRoad(2)); err == nil { - t.Fatal("AdvanceIfNeeded must not seed epoch 2") - } +func addFromEnd(t *testing.T, r *Registry, end types.EpochIndex, weights map[types.PublicKey]uint64) *types.Epoch { + t.Helper() + require.NoError(t, r.StageAndActivate(end, weights)) + return r.MustEpoch(end + 2) } -func TestSetupInitialEpochs(t *testing.T) { - for _, tc := range []struct { - name string - span utils.Option[types.RoadRange] - want []types.EpochIndex - absent types.EpochIndex - }{ - { - name: "empty None is no-op", - span: utils.None[types.RoadRange](), - want: []types.EpochIndex{0, 1}, - absent: 2, - }, - { - name: "mid CommitQC seeds placeholder next", - span: utils.Some(types.RoadRange{First: midRoad(5), Next: midRoad(5) + 1}), - want: []types.EpochIndex{4, 5, 6}, - absent: 7, - }, - { - name: "commit span from first", - span: utils.Some(types.RoadRange{ - First: midRoad(2), - Next: midRoad(5) + 1, - }), - want: []types.EpochIndex{1, 2, 3, 4, 5, 6}, - absent: 7, - }, - } { - t.Run(tc.name, func(t *testing.T) { - r, _ := makeRegistry(t) - r.SetupInitialEpochs(tc.span) - for _, idx := range tc.want { - if _, err := r.EpochAt(FirstRoad(idx)); err != nil { - t.Fatalf("EpochAt(epoch %d): %v", idx, err) - } - } - if _, err := r.EpochAt(FirstRoad(tc.absent)); err == nil { - t.Fatalf("EpochAt(epoch %d) should not be present", tc.absent) - } - }) - } +func TestStageEpoch_FromEndStake(t *testing.T) { + rng := utils.TestRng() + a := types.GenSecretKey(rng) + b := types.GenSecretKey(rng) + committee := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{ + a.Public(): 1, b.Public(): 1, + })) + r := utils.OrPanic1(NewRegistry(committee, 0, time.Time{}, utils.None[string]())) + genesis := r.MustEpoch(1).Committee() + _, err := r.EpochByIndex(2) + require.Error(t, err) + + ep := addFromEnd(t, r, 0, map[types.PublicKey]uint64{b.Public(): 7}) + require.Equal(t, types.EpochIndex(2), ep.EpochIndex()) + require.Equal(t, FirstRoad(2), ep.RoadRange().First) + require.Equal(t, FirstRoad(3), ep.RoadRange().Next) + require.Equal(t, genesis, r.MustEpoch(1).Committee()) + require.False(t, ep.Committee().HasReplica(a.Public())) + require.Equal(t, uint64(7), ep.Committee().Weight(b.Public())) + require.Equal(t, ep, r.MustEpoch(2)) + + require.NoError(t, r.StageAndActivate(0, map[types.PublicKey]uint64{b.Public(): 7, a.Public(): 0})) + require.Equal(t, ep, r.MustEpoch(2)) } -func TestActivateEpoch_SkipsExistingSeeds(t *testing.T) { +func TestStageEpoch_RequiresNext(t *testing.T) { + rng := utils.TestRng() + pk := types.GenSecretKey(rng).Public() + committee := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{ + pk: 1, types.GenSecretKey(rng).Public(): 1, + })) + r := utils.OrPanic1(NewRegistry(committee, 0, time.Time{}, utils.None[string]())) + require.Error(t, r.StageEpoch(5, map[types.PublicKey]uint64{pk: 1})) +} + +func TestStageEpoch_RefusesWhileUnconfirmed(t *testing.T) { r, committee := makeRegistry(t) - r.SetupInitialEpochs(utils.None[types.RoadRange]()) - seeded := r.MustEpoch(1) - seededCommittee := seeded.Committee() + pk := committee.Lanes().At(0).Validator + weights := map[types.PublicKey]uint64{pk: 1} + require.NoError(t, r.StageEpoch(0, weights)) + require.Error(t, r.StageEpoch(1, weights)) + require.Equal(t, utils.Some(types.EpochIndex(2)), r.Pending()) + + require.NoError(t, r.StageEpoch(0, weights)) + require.Equal(t, utils.Some(types.EpochIndex(2)), r.Pending()) + + require.NoError(t, r.ActivateEpoch(2)) + require.NoError(t, r.StageEpoch(1, weights)) + require.Equal(t, utils.Some(types.EpochIndex(3)), r.Pending()) +} + +func TestStageEpoch_KeepsRegisteredEpoch(t *testing.T) { + r, committee := makeRegistry(t) pk := committee.Lanes().At(0).Validator - ep, err := r.ActivateEpoch( - 0, - map[types.PublicKey]uint64{pk: 1}, - time.Time{}, - r.FirstBlock(), - ) - require.NoError(t, err) - require.Equal(t, types.EpochIndex(2), ep.EpochIndex()) - require.Equal(t, FirstRoad(2), ep.RoadRange().First) - require.Equal(t, FirstRoad(3), ep.RoadRange().Next) - got := r.MustEpoch(1) - require.Equal(t, seededCommittee, got.Committee()) - _, ok := ep.Committee().Lane(pk).Get() - require.True(t, ok) - require.Equal(t, 1, ep.Committee().Lanes().Len()) + first := addFromEnd(t, r, 0, map[types.PublicKey]uint64{pk: 1}) + + other := committee.Lanes().At(1).Validator + require.Error(t, r.StageAndActivate(0, map[types.PublicKey]uint64{other: 5})) + require.Equal(t, first, r.MustEpoch(2)) } -func TestActivateEpoch_RejoinJoinedFromLatestNotPlaceholder(t *testing.T) { +func TestStageEpoch_RejoinTakesNewJoined(t *testing.T) { rng := utils.TestRng() a := types.GenSecretKey(rng) b := types.GenSecretKey(rng) committee := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{ a.Public(): 1, b.Public(): 1, })) - r := utils.OrPanic1(NewRegistry(committee, 0, time.Time{})) - r.SetupInitialEpochs(utils.None[types.RoadRange]()) - - epLeave, err := r.ActivateEpoch( - 0, - map[types.PublicKey]uint64{b.Public(): 1}, - time.Time{}, r.FirstBlock(), - ) - require.NoError(t, err) + r := utils.OrPanic1(NewRegistry(committee, 0, time.Time{}, utils.None[string]())) + + epLeave := addFromEnd(t, r, 0, map[types.PublicKey]uint64{b.Public(): 1}) require.Equal(t, types.EpochIndex(2), epLeave.EpochIndex()) require.False(t, epLeave.Committee().HasReplica(a.Public())) - // Seed a genesis-committee placeholder ahead of the activated epoch. Deriving from that - // slot would treat A as still present and keep Joined=0. - r.AdvanceIfNeeded(LastRoad(2)) - seeded := r.MustEpoch(3) - require.True(t, seeded.Committee().HasReplica(a.Public())) - - epJoin, err := r.ActivateEpoch( - epLeave.EpochIndex(), - map[types.PublicKey]uint64{a.Public(): 1, b.Public(): 1}, - time.Time{}, r.FirstBlock(), - ) - require.NoError(t, err) - require.Equal(t, types.EpochIndex(4), epJoin.EpochIndex()) + epJoin := addFromEnd(t, r, 1, map[types.PublicKey]uint64{a.Public(): 1, b.Public(): 1}) + require.Equal(t, types.EpochIndex(3), epJoin.EpochIndex()) lane := epJoin.Committee().Lane(a.Public()).OrPanic("rejoin") - require.Equal(t, types.EpochIndex(4), lane.Joined) + require.Equal(t, types.EpochIndex(3), lane.Joined) } -func TestWaitForEpoch_FastPathAndWait(t *testing.T) { +func TestStageEpoch_PendingIsInvisible(t *testing.T) { synctest.Test(t, func(t *testing.T) { - r, _ := makeRegistry(t) - ep, err := r.WaitForEpoch(t.Context(), 0) - require.NoError(t, err) - require.Equal(t, types.EpochIndex(0), ep.EpochIndex()) - - ep, err = r.WaitForEpoch(t.Context(), 1) - require.NoError(t, err) - require.Equal(t, types.EpochIndex(1), ep.EpochIndex()) + r, committee := makeRegistry(t) + pk := committee.Lanes().At(0).Validator + require.NoError(t, r.StageEpoch(0, map[types.PublicKey]uint64{pk: 1})) + require.Equal(t, utils.Some(types.EpochIndex(2)), r.Pending()) + _, err := r.EpochByIndex(2) + require.Error(t, err) _, err = r.EpochAt(FirstRoad(2)) require.Error(t, err) var got *types.Epoch - var waitErr error - go func() { - got, waitErr = r.WaitForEpoch(t.Context(), 2) - }() + go func() { got, _ = r.WaitForEpoch(t.Context(), 2) }() synctest.Wait() - require.Nil(t, got, "WaitForEpoch returned before AdvanceIfNeeded") + require.Nil(t, got, "WaitForEpoch returned for a staged committee") - r.AdvanceIfNeeded(LastRoad(1)) + require.NoError(t, r.ActivateEpoch(2)) synctest.Wait() - require.NoError(t, waitErr) require.Equal(t, types.EpochIndex(2), got.EpochIndex()) }) } -func TestPruneBefore_DropsIntermediateKeepsGenesis(t *testing.T) { - r, _ := makeRegistry(t) - r.AdvanceIfNeeded(LastRoad(0)) - r.AdvanceIfNeeded(LastRoad(1)) - _ = r.MustEpoch(1) - _ = r.MustEpoch(2) +func TestActivateEpoch_IdempotentAndRefusesUnstaged(t *testing.T) { + r, committee := makeRegistry(t) + pk := committee.Lanes().At(0).Validator + require.Error(t, r.ActivateEpoch(2)) - r.PruneBefore(2) - require.Equal(t, types.EpochRange{First: 2, Next: 3}, r.Live()) - _ = r.MustEpoch(0) - _, err := r.EpochByIndex(1) + require.NoError(t, r.StageEpoch(0, map[types.PublicKey]uint64{pk: 1})) + require.NoError(t, r.ActivateEpoch(2)) + ep := r.MustEpoch(2) + require.NoError(t, r.ActivateEpoch(2)) + require.Equal(t, ep, r.MustEpoch(2)) +} + +func TestPruneBefore_KeepsLatestAndPending(t *testing.T) { + r, committee := makeRegistry(t) + pk := committee.Lanes().At(0).Validator + weights := map[types.PublicKey]uint64{pk: 1} + for endEpoch := types.EpochIndex(0); endEpoch < 4; endEpoch++ { + require.NoError(t, r.StageAndActivate(endEpoch, weights)) + } + require.NoError(t, r.StageEpoch(4, weights)) + + require.NoError(t, r.PruneBefore(10)) // clamped to the latest live epoch + _ = r.MustEpoch(5) + require.Equal(t, utils.Some(types.EpochIndex(6)), r.Pending()) + require.NoError(t, r.ActivateEpoch(6)) +} + +func TestPruneBefore_DropsDerivedKeepsGenesis(t *testing.T) { + r, committee := makeRegistry(t) + pk := committee.Lanes().At(0).Validator + weights := map[types.PublicKey]uint64{pk: 1} + _ = addFromEnd(t, r, 0, weights) + _ = addFromEnd(t, r, 1, weights) + _ = addFromEnd(t, r, 2, weights) + + require.NoError(t, r.PruneBefore(4)) + _, err := r.EpochByIndex(2) require.ErrorIs(t, err, types.ErrPruned) - _ = r.MustEpoch(2) + _, err = r.EpochByIndex(3) + require.ErrorIs(t, err, types.ErrPruned) + _ = r.MustEpoch(0) + _ = r.MustEpoch(1) + _ = r.MustEpoch(4) require.Equal(t, types.GlobalBlockNumber(0), r.FirstBlock()) - _, err = r.EpochAt(FirstRoad(1)) + _, err = r.EpochAt(FirstRoad(2)) require.ErrorIs(t, err, types.ErrPruned) - _, err = r.WaitForEpoch(t.Context(), 1) + _, err = r.WaitForEpoch(t.Context(), 2) require.ErrorIs(t, err, types.ErrPruned) - r.PruneBefore(1) // no rewind - _ = r.MustEpoch(2) + require.NoError(t, r.PruneBefore(1)) // no rewind + _ = r.MustEpoch(4) - pk := r.MustEpoch(0).Committee().Lanes().At(0).Validator - ep, err := r.ActivateEpoch( - 0, - map[types.PublicKey]uint64{pk: 1}, - time.Time{}, - r.FirstBlock(), - ) - require.NoError(t, err) - require.Equal(t, types.EpochIndex(3), ep.EpochIndex()) - _, err = r.EpochByIndex(1) + ep := addFromEnd(t, r, 3, weights) + require.Equal(t, types.EpochIndex(5), ep.EpochIndex()) + require.NoError(t, r.PruneBefore(5)) + _, err = r.EpochByIndex(4) + require.ErrorIs(t, err, types.ErrPruned) + _ = r.MustEpoch(5) + + require.NoError(t, r.PruneBefore(10)) // latest live epoch is retained + _ = r.MustEpoch(5) + _ = r.MustEpoch(0) + _ = r.MustEpoch(1) +} + +func TestNewRegistry_RestoresPrunedSnapshot(t *testing.T) { + rng := utils.TestRng() + a := types.GenSecretKey(rng).Public() + b := types.GenSecretKey(rng).Public() + genesis := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{a: 1, b: 1})) + dir := utils.Some(t.TempDir()) + + r1 := utils.OrPanic1(NewRegistry(genesis, 7, time.Unix(10, 0), dir)) + require.NoError(t, r1.StageAndActivate(0, map[types.PublicKey]uint64{b: 2})) + require.NoError(t, r1.StageAndActivate(1, map[types.PublicKey]uint64{a: 3, b: 2})) + require.NoError(t, r1.StageAndActivate(2, map[types.PublicKey]uint64{a: 4, b: 2})) + require.NoError(t, r1.PruneBefore(3)) + + r2 := utils.OrPanic1(NewRegistry(genesis, 7, time.Unix(10, 0), dir)) + + _ = r2.MustEpoch(1) + _, err := r2.EpochByIndex(2) require.ErrorIs(t, err, types.ErrPruned) + ep3 := r2.MustEpoch(3) + ep4 := r2.MustEpoch(4) + require.Equal(t, utils.None[types.EpochIndex](), r2.Pending()) + require.Equal(t, uint64(3), ep3.Committee().Weight(a)) + require.Equal(t, types.EpochIndex(3), ep3.Committee().Lane(a).OrPanic("a in epoch 3").Joined) + require.Equal(t, types.EpochIndex(3), ep4.Committee().Lane(a).OrPanic("a in epoch 4").Joined) +} + +func TestNewRegistry_RestoresSnapshotPending(t *testing.T) { + rng := utils.TestRng() + a := types.GenSecretKey(rng).Public() + genesis := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{a: 1})) + dir := utils.Some(t.TempDir()) + + r1 := utils.OrPanic1(NewRegistry(genesis, 7, time.Unix(10, 0), dir)) + require.NoError(t, r1.StageEpoch(0, map[types.PublicKey]uint64{a: 2})) + + r2 := utils.OrPanic1(NewRegistry(genesis, 7, time.Unix(10, 0), dir)) + _ = r2.MustEpoch(1) + require.Equal(t, utils.Some(types.EpochIndex(2)), r2.Pending()) + _, err := r2.EpochByIndex(2) + require.Error(t, err) + require.False(t, errors.Is(err, types.ErrPruned)) + + require.NoError(t, r2.ActivateEpoch(2)) + require.Equal(t, uint64(2), r2.MustEpoch(2).Committee().Weight(a)) + require.Equal(t, utils.None[types.EpochIndex](), r2.Pending()) + + r3 := utils.OrPanic1(NewRegistry(genesis, 7, time.Unix(10, 0), dir)) + require.Equal(t, uint64(2), r3.MustEpoch(2).Committee().Weight(a)) + require.Equal(t, utils.None[types.EpochIndex](), r3.Pending()) +} + +func TestNewRegistry_RestoresLatestAndPendingAfterPrune(t *testing.T) { + rng := utils.TestRng() + a := types.GenSecretKey(rng).Public() + genesis := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{a: 1})) + dir := utils.Some(t.TempDir()) + weights := map[types.PublicKey]uint64{a: 2} + + r1 := utils.OrPanic1(NewRegistry(genesis, 7, time.Unix(10, 0), dir)) + for endEpoch := types.EpochIndex(0); endEpoch < 4; endEpoch++ { + require.NoError(t, r1.StageAndActivate(endEpoch, weights)) + } + require.NoError(t, r1.StageEpoch(4, weights)) + require.NoError(t, r1.PruneBefore(10)) + + r2 := utils.OrPanic1(NewRegistry(genesis, 7, time.Unix(10, 0), dir)) + _ = r2.MustEpoch(5) + require.Equal(t, utils.Some(types.EpochIndex(6)), r2.Pending()) + require.NoError(t, r2.ActivateEpoch(6)) + require.Equal(t, uint64(2), r2.MustEpoch(6).Committee().Weight(a)) +} + +func TestCommit_PersistFailureLeavesMemoryUnchanged(t *testing.T) { + r, committee := makeRegistry(t) + pk := committee.Lanes().At(0).Validator + weights := map[types.PublicKey]uint64{pk: 1} + ok := r.persister + disk := errors.New("disk") + + r.persister = errPersister{err: disk} + require.ErrorIs(t, r.StageEpoch(0, weights), disk) + require.Equal(t, utils.None[types.EpochIndex](), r.Pending()) + + r.persister = ok + require.NoError(t, r.StageEpoch(0, weights)) + require.Equal(t, utils.Some(types.EpochIndex(2)), r.Pending()) + + r.persister = errPersister{err: disk} + require.ErrorIs(t, r.ActivateEpoch(2), disk) + require.Equal(t, utils.Some(types.EpochIndex(2)), r.Pending()) + _, err := r.EpochByIndex(2) + require.Error(t, err) + + r.persister = ok + require.NoError(t, r.ActivateEpoch(2)) + _ = r.MustEpoch(2) + require.NoError(t, r.StageAndActivate(1, weights)) + require.NoError(t, r.StageAndActivate(2, weights)) + + r.persister = errPersister{err: disk} + require.ErrorIs(t, r.PruneBefore(10), disk) + _ = r.MustEpoch(2) + + r.persister = ok + require.NoError(t, r.PruneBefore(10)) + _ = r.MustEpoch(4) } diff --git a/sei-tendermint/internal/autobahn/epoch/snapshot.go b/sei-tendermint/internal/autobahn/epoch/snapshot.go new file mode 100644 index 0000000000..0e081fe883 --- /dev/null +++ b/sei-tendermint/internal/autobahn/epoch/snapshot.go @@ -0,0 +1,87 @@ +package epoch + +import ( + "errors" + "fmt" + "os" + "path/filepath" + + "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/pb" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" +) + +const ( + epochSnapshotDir = "epochs" + epochSnapshotPrefix = "registry" +) + +func openEpochSnapshot( + stateDir utils.Option[string], +) (persist.Persister[*pb.PersistedEpochRegistry], utils.Option[*pb.PersistedEpochRegistry], error) { + root, ok := stateDir.Get() + if !ok { + return persist.NewPersister[*pb.PersistedEpochRegistry](utils.None[string](), epochSnapshotPrefix) + } + dir := filepath.Join(root, epochSnapshotDir) + if err := os.MkdirAll(dir, 0700); err != nil { + return nil, utils.None[*pb.PersistedEpochRegistry](), fmt.Errorf("create epoch snapshot dir %s: %w", dir, err) + } + return persist.NewPersister[*pb.PersistedEpochRegistry](utils.Some(dir), epochSnapshotPrefix) +} + +func encodeEpochRecord(idx types.EpochIndex, committee *types.Committee) *pb.EpochRecord { + return &pb.EpochRecord{ + Index: utils.Alloc(uint64(idx)), + Committee: types.CommitteeConv.Encode(committee), + } +} + +func decodeEpochRecord(record *pb.EpochRecord) (types.EpochIndex, *types.Committee, error) { + if record == nil { + return 0, nil, errors.New("missing") + } + if record.Index == nil { + return 0, nil, errors.New("index: missing") + } + idx := types.EpochIndex(*record.Index) + if idx < 2 { + return 0, nil, fmt.Errorf("genesis epoch %d", idx) + } + committee, err := types.CommitteeConv.DecodeReq(record.Committee) + if err != nil { + return 0, nil, err + } + for lane := range committee.Lanes().All() { + if lane.Joined > idx { + return 0, nil, fmt.Errorf("member joined epoch %d after committee epoch %d", lane.Joined, idx) + } + } + return idx, committee, nil +} + +// checkDerivedFromPrev returns an error if committee is not DeriveNext of +// epoch idx-1. A predecessor dropped by PruneBefore cannot be checked and is +// accepted; a predecessor missing for any other reason is an error. +func (s *registryState) checkDerivedFromPrev(idx types.EpochIndex, committee *types.Committee) error { + prev, ok := s.m[idx-1] + if !ok { + if !s.dropped(idx - 1) { + return fmt.Errorf("missing predecessor epoch %d", idx-1) + } + return nil + } + weights := make(map[types.PublicKey]uint64, committee.Lanes().Len()) + for lane := range committee.Lanes().All() { + weights[lane.Validator] = committee.Weight(lane.Validator) + } + want, err := prev.Committee().DeriveNext(weights, idx) + if err != nil { + return err + } + if !want.Equal(committee) { + return fmt.Errorf("does not follow epoch %d", idx-1) + } + return nil +} diff --git a/sei-tendermint/internal/autobahn/epoch/snapshot_test.go b/sei-tendermint/internal/autobahn/epoch/snapshot_test.go new file mode 100644 index 0000000000..a574f1af91 --- /dev/null +++ b/sei-tendermint/internal/autobahn/epoch/snapshot_test.go @@ -0,0 +1,112 @@ +package epoch + +import ( + "testing" + "time" + + "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pb" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" +) + +func TestRestoreSnapshotValidatesShape(t *testing.T) { + rng := utils.TestRng() + a := types.GenSecretKey(rng).Public() + b := types.GenSecretKey(rng).Public() + genesis := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{a: 1})) + c2 := utils.OrPanic1(genesis.DeriveNext(map[types.PublicKey]uint64{a: 1, b: 1}, 2)) + c3 := utils.OrPanic1(c2.DeriveNext(map[types.PublicKey]uint64{a: 1, b: 1}, 3)) + c4 := utils.OrPanic1(c3.DeriveNext(map[types.PublicKey]uint64{a: 1, b: 1}, 4)) + restore := func(snapshot *pb.PersistedEpochRegistry) error { + r := utils.OrPanic1(NewRegistry(genesis, 0, time.Time{}, utils.None[string]())) + for state := range r.state.Lock() { + return state.restore(snapshot) + } + panic("unreachable") + } + + require.Error(t, restore(&pb.PersistedEpochRegistry{ + Live: []*pb.EpochRecord{encodeEpochRecord(2, c2), encodeEpochRecord(4, c4)}, + })) + require.Error(t, restore(&pb.PersistedEpochRegistry{ + Live: []*pb.EpochRecord{encodeEpochRecord(2, c2)}, + Pending: encodeEpochRecord(4, c4), + })) +} + +func TestRestoreSnapshotRejectsBrokenDerivation(t *testing.T) { + rng := utils.TestRng() + a := types.GenSecretKey(rng).Public() + b := types.GenSecretKey(rng).Public() + genesis := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{a: 1})) + other := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{a: 1, b: 1})) + restore := func(snapshot *pb.PersistedEpochRegistry) error { + r := utils.OrPanic1(NewRegistry(genesis, 0, time.Time{}, utils.None[string]())) + for state := range r.state.Lock() { + return state.restore(snapshot) + } + panic("unreachable") + } + + require.Error(t, restore(&pb.PersistedEpochRegistry{ + Live: []*pb.EpochRecord{encodeEpochRecord(2, other)}, + })) + require.Error(t, restore(&pb.PersistedEpochRegistry{ + Pending: encodeEpochRecord(2, other), + })) +} + +func TestRestoreSnapshotAcceptsDerivedChain(t *testing.T) { + rng := utils.TestRng() + a := types.GenSecretKey(rng).Public() + b := types.GenSecretKey(rng).Public() + genesis := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{a: 1})) + c2 := utils.OrPanic1(genesis.DeriveNext(map[types.PublicKey]uint64{a: 1, b: 1}, 2)) + c3 := utils.OrPanic1(c2.DeriveNext(map[types.PublicKey]uint64{a: 1, b: 1}, 3)) + restore := func(snapshot *pb.PersistedEpochRegistry) error { + r := utils.OrPanic1(NewRegistry(genesis, 0, time.Time{}, utils.None[string]())) + for state := range r.state.Lock() { + return state.restore(snapshot) + } + panic("unreachable") + } + + require.NoError(t, restore(&pb.PersistedEpochRegistry{ + Live: []*pb.EpochRecord{encodeEpochRecord(2, c2)}, + })) + require.NoError(t, restore(&pb.PersistedEpochRegistry{ + Live: []*pb.EpochRecord{encodeEpochRecord(2, c2)}, + Pending: encodeEpochRecord(3, c3), + })) + require.NoError(t, restore(&pb.PersistedEpochRegistry{ + Pending: encodeEpochRecord(2, c2), + })) +} + +func TestRestoreSnapshotAllowsUnchainedFirstLive(t *testing.T) { + rng := utils.TestRng() + a := types.GenSecretKey(rng).Public() + b := types.GenSecretKey(rng).Public() + genesis := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{a: 1})) + other := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{b: 1})) + r := utils.OrPanic1(NewRegistry(genesis, 0, time.Time{}, utils.None[string]())) + for state := range r.state.Lock() { + require.NoError(t, state.restore(&pb.PersistedEpochRegistry{ + Live: []*pb.EpochRecord{encodeEpochRecord(3, other)}, + })) + return + } + panic("unreachable") +} + +func TestDecodeEpochRecordRejectsFutureJoin(t *testing.T) { + rng := utils.TestRng() + a := types.GenSecretKey(rng).Public() + b := types.GenSecretKey(rng).Public() + genesis := utils.OrPanic1(types.NewCommittee(map[types.PublicKey]uint64{a: 1})) + committee := utils.OrPanic1(genesis.DeriveNext(map[types.PublicKey]uint64{a: 1, b: 1}, 3)) + + _, _, err := decodeEpochRecord(encodeEpochRecord(2, committee)) + require.Error(t, err) +} diff --git a/sei-tendermint/internal/autobahn/epoch/testonly.go b/sei-tendermint/internal/autobahn/epoch/testonly.go index cbb672288e..1c16e24c31 100644 --- a/sei-tendermint/internal/autobahn/epoch/testonly.go +++ b/sei-tendermint/internal/autobahn/epoch/testonly.go @@ -11,15 +11,45 @@ import ( // Returns the generated secret keys as well. // Intended for use in tests only. func GenRegistry(rng utils.Rng, size int) (*Registry, []types.SecretKey) { + committee, sks, _ := genCommittee(rng, size) + firstBlock := types.GenGlobalBlockNumber(rng) % 1000000 + registry := utils.OrPanic1(NewRegistry(committee, firstBlock, time.Now(), utils.None[string]())) + return registry, sks +} + +// GenRegistryThrough returns a Registry with genesis epochs 0 and 1 plus +// live execution-derived committees through last. +func GenRegistryThrough( + rng utils.Rng, + size int, + last types.EpochIndex, +) (*Registry, []types.SecretKey) { + committee, sks, weights := genCommittee(rng, size) + firstBlock := types.GenGlobalBlockNumber(rng) % 1000000 + registry := utils.OrPanic1(NewRegistry(committee, firstBlock, time.Now(), utils.None[string]())) + for target := types.EpochIndex(2); target <= last; target++ { + utils.OrPanic(registry.StageEpoch(target-2, weights)) + utils.OrPanic(registry.ActivateEpoch(target)) + } + return registry, sks +} + +func genCommittee(rng utils.Rng, size int) (*types.Committee, []types.SecretKey, map[types.PublicKey]uint64) { sks := utils.GenSliceN(rng, size, types.GenSecretKey) weights := map[types.PublicKey]uint64{} for _, sk := range sks { weights[sk.Public()] = 1000 + uint64(rng.Intn(1000)) //nolint:gosec } committee := utils.OrPanic1(types.NewCommittee(weights)) - firstBlock := types.GenGlobalBlockNumber(rng) % 1000000 - registry := utils.OrPanic1(NewRegistry(committee, firstBlock, time.Now())) - return registry, sks + return committee, sks, weights +} + +// StageAndActivate derives C_{endEpoch+2} from weights and publishes it. +func (r *Registry) StageAndActivate(endEpoch types.EpochIndex, weights map[types.PublicKey]uint64) error { + if err := r.StageEpoch(endEpoch, weights); err != nil { + return err + } + return r.ActivateEpoch(endEpoch + 2) } // MustEpoch returns the registered epoch at i. Panics if it is missing. diff --git a/sei-tendermint/internal/autobahn/pb/autobahn.pb.go b/sei-tendermint/internal/autobahn/pb/autobahn.pb.go index 523ce103d8..c9b26b3f5b 100644 --- a/sei-tendermint/internal/autobahn/pb/autobahn.pb.go +++ b/sei-tendermint/internal/autobahn/pb/autobahn.pb.go @@ -2237,6 +2237,208 @@ func (*ConsensusReq_TimeoutVote) isConsensusReq_T() {} func (*ConsensusReq_TimeoutQc) isConsensusReq_T() {} +type Committee struct { + state protoimpl.MessageState `protogen:"open.v1"` + Members []*EpochMember `protobuf:"bytes,1,rep,name=members,proto3" json:"members,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Committee) Reset() { + *x = Committee{} + mi := &file_autobahn_autobahn_proto_msgTypes[35] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Committee) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Committee) ProtoMessage() {} + +func (x *Committee) ProtoReflect() protoreflect.Message { + mi := &file_autobahn_autobahn_proto_msgTypes[35] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Committee.ProtoReflect.Descriptor instead. +func (*Committee) Descriptor() ([]byte, []int) { + return file_autobahn_autobahn_proto_rawDescGZIP(), []int{35} +} + +func (x *Committee) GetMembers() []*EpochMember { + if x != nil { + return x.Members + } + return nil +} + +// Committee for an execution-derived epoch. +type EpochRecord struct { + state protoimpl.MessageState `protogen:"open.v1"` + Index *uint64 `protobuf:"varint,1,opt,name=index,proto3,oneof" json:"index,omitempty"` // required + Committee *Committee `protobuf:"bytes,2,opt,name=committee,proto3,oneof" json:"committee,omitempty"` // required + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *EpochRecord) Reset() { + *x = EpochRecord{} + mi := &file_autobahn_autobahn_proto_msgTypes[36] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *EpochRecord) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*EpochRecord) ProtoMessage() {} + +func (x *EpochRecord) ProtoReflect() protoreflect.Message { + mi := &file_autobahn_autobahn_proto_msgTypes[36] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use EpochRecord.ProtoReflect.Descriptor instead. +func (*EpochRecord) Descriptor() ([]byte, []int) { + return file_autobahn_autobahn_proto_rawDescGZIP(), []int{36} +} + +func (x *EpochRecord) GetIndex() uint64 { + if x != nil && x.Index != nil { + return *x.Index + } + return 0 +} + +func (x *EpochRecord) GetCommittee() *Committee { + if x != nil { + return x.Committee + } + return nil +} + +// Persisted execution-derived epoch registry state. +type PersistedEpochRegistry struct { + state protoimpl.MessageState `protogen:"open.v1"` + Live []*EpochRecord `protobuf:"bytes,1,rep,name=live,proto3" json:"live,omitempty"` + Pending *EpochRecord `protobuf:"bytes,2,opt,name=pending,proto3,oneof" json:"pending,omitempty"` // optional + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *PersistedEpochRegistry) Reset() { + *x = PersistedEpochRegistry{} + mi := &file_autobahn_autobahn_proto_msgTypes[37] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *PersistedEpochRegistry) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PersistedEpochRegistry) ProtoMessage() {} + +func (x *PersistedEpochRegistry) ProtoReflect() protoreflect.Message { + mi := &file_autobahn_autobahn_proto_msgTypes[37] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PersistedEpochRegistry.ProtoReflect.Descriptor instead. +func (*PersistedEpochRegistry) Descriptor() ([]byte, []int) { + return file_autobahn_autobahn_proto_rawDescGZIP(), []int{37} +} + +func (x *PersistedEpochRegistry) GetLive() []*EpochRecord { + if x != nil { + return x.Live + } + return nil +} + +func (x *PersistedEpochRegistry) GetPending() *EpochRecord { + if x != nil { + return x.Pending + } + return nil +} + +type EpochMember struct { + state protoimpl.MessageState `protogen:"open.v1"` + LaneId *LaneID `protobuf:"bytes,1,opt,name=lane_id,json=laneId,proto3,oneof" json:"lane_id,omitempty"` // required + Weight *uint64 `protobuf:"varint,2,opt,name=weight,proto3,oneof" json:"weight,omitempty"` // required, can't be zero + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *EpochMember) Reset() { + *x = EpochMember{} + mi := &file_autobahn_autobahn_proto_msgTypes[38] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *EpochMember) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*EpochMember) ProtoMessage() {} + +func (x *EpochMember) ProtoReflect() protoreflect.Message { + mi := &file_autobahn_autobahn_proto_msgTypes[38] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use EpochMember.ProtoReflect.Descriptor instead. +func (*EpochMember) Descriptor() ([]byte, []int) { + return file_autobahn_autobahn_proto_rawDescGZIP(), []int{38} +} + +func (x *EpochMember) GetLaneId() *LaneID { + if x != nil { + return x.LaneId + } + return nil +} + +func (x *EpochMember) GetWeight() uint64 { + if x != nil && x.Weight != nil { + return *x.Weight + } + return 0 +} + var File_autobahn_autobahn_proto protoreflect.FileDescriptor const file_autobahn_autobahn_proto_rawDesc = "" + @@ -2460,7 +2662,26 @@ const file_autobahn_autobahn_proto_rawDesc = "" + "\ftimeout_vote\x18\x04 \x01(\v2\x19.autobahn.FullTimeoutVoteH\x00R\vtimeoutVote\x124\n" + "\n" + "timeout_qc\x18\x05 \x01(\v2\x13.autobahn.TimeoutQCH\x00R\ttimeoutQc:\x06\xe8\x88\xe2\xab\f\x01B\x03\n" + - "\x01tJ\x04\b\x02\x10\x03J\x04\b\x03\x10\x04R\fprepare_voteR\vcommit_voteBGZEgithub.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pbb\x06proto3" + "\x01tJ\x04\b\x02\x10\x03J\x04\b\x03\x10\x04R\fprepare_voteR\vcommit_vote\"L\n" + + "\tCommittee\x127\n" + + "\amembers\x18\x01 \x03(\v2\x15.autobahn.EpochMemberB\x06Ј\xe2\xab\fdR\amembers:\x06\xe8\x88\xe2\xab\f\x01\"\x80\x01\n" + + "\vEpochRecord\x12\x19\n" + + "\x05index\x18\x01 \x01(\x04H\x00R\x05index\x88\x01\x01\x126\n" + + "\tcommittee\x18\x02 \x01(\v2\x13.autobahn.CommitteeH\x01R\tcommittee\x88\x01\x01:\x06\xe8\x88\xe2\xab\f\x01B\b\n" + + "\x06_indexB\f\n" + + "\n" + + "_committee\"\x85\x01\n" + + "\x16PersistedEpochRegistry\x12)\n" + + "\x04live\x18\x01 \x03(\v2\x15.autobahn.EpochRecordR\x04live\x124\n" + + "\apending\x18\x02 \x01(\v2\x15.autobahn.EpochRecordH\x00R\apending\x88\x01\x01B\n" + + "\n" + + "\b_pending\"y\n" + + "\vEpochMember\x12.\n" + + "\alane_id\x18\x01 \x01(\v2\x10.autobahn.LaneIDH\x00R\x06laneId\x88\x01\x01\x12\x1b\n" + + "\x06weight\x18\x02 \x01(\x04H\x01R\x06weight\x88\x01\x01:\x06\xe8\x88\xe2\xab\f\x01B\n" + + "\n" + + "\b_lane_idB\t\n" + + "\a_weightBGZEgithub.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/pbb\x06proto3" var ( file_autobahn_autobahn_proto_rawDescOnce sync.Once @@ -2474,7 +2695,7 @@ func file_autobahn_autobahn_proto_rawDescGZIP() []byte { return file_autobahn_autobahn_proto_rawDescData } -var file_autobahn_autobahn_proto_msgTypes = make([]protoimpl.MessageInfo, 36) +var file_autobahn_autobahn_proto_msgTypes = make([]protoimpl.MessageInfo, 40) var file_autobahn_autobahn_proto_goTypes = []any{ (*Timestamp)(nil), // 0: autobahn.Timestamp (*Duration)(nil), // 1: autobahn.Duration @@ -2511,10 +2732,14 @@ var file_autobahn_autobahn_proto_goTypes = []any{ (*SignedBlockHeader)(nil), // 32: autobahn.SignedBlockHeader (*SignedAppProposal)(nil), // 33: autobahn.SignedAppProposal (*ConsensusReq)(nil), // 34: autobahn.ConsensusReq - nil, // 35: autobahn.TransactionHeader.PropertiesEntry + (*Committee)(nil), // 35: autobahn.Committee + (*EpochRecord)(nil), // 36: autobahn.EpochRecord + (*PersistedEpochRegistry)(nil), // 37: autobahn.PersistedEpochRegistry + (*EpochMember)(nil), // 38: autobahn.EpochMember + nil, // 39: autobahn.TransactionHeader.PropertiesEntry } var file_autobahn_autobahn_proto_depIdxs = []int32{ - 35, // 0: autobahn.TransactionHeader.properties:type_name -> autobahn.TransactionHeader.PropertiesEntry + 39, // 0: autobahn.TransactionHeader.properties:type_name -> autobahn.TransactionHeader.PropertiesEntry 2, // 1: autobahn.TransactionHeader.timestamps:type_name -> autobahn.TransactionTimestamps 3, // 2: autobahn.Transaction.header:type_name -> autobahn.TransactionHeader 6, // 3: autobahn.LaneID.validator:type_name -> autobahn.PublicKey @@ -2578,11 +2803,16 @@ var file_autobahn_autobahn_proto_depIdxs = []int32{ 28, // 61: autobahn.ConsensusReq.commit_vote_v2:type_name -> autobahn.SignedProposal 22, // 62: autobahn.ConsensusReq.timeout_vote:type_name -> autobahn.FullTimeoutVote 21, // 63: autobahn.ConsensusReq.timeout_qc:type_name -> autobahn.TimeoutQC - 64, // [64:64] is the sub-list for method output_type - 64, // [64:64] is the sub-list for method input_type - 64, // [64:64] is the sub-list for extension type_name - 64, // [64:64] is the sub-list for extension extendee - 0, // [0:64] is the sub-list for field type_name + 38, // 64: autobahn.Committee.members:type_name -> autobahn.EpochMember + 35, // 65: autobahn.EpochRecord.committee:type_name -> autobahn.Committee + 36, // 66: autobahn.PersistedEpochRegistry.live:type_name -> autobahn.EpochRecord + 36, // 67: autobahn.PersistedEpochRegistry.pending:type_name -> autobahn.EpochRecord + 7, // 68: autobahn.EpochMember.lane_id:type_name -> autobahn.LaneID + 69, // [69:69] is the sub-list for method output_type + 69, // [69:69] is the sub-list for method input_type + 69, // [69:69] is the sub-list for extension type_name + 69, // [69:69] is the sub-list for extension extendee + 0, // [0:69] is the sub-list for field type_name } func init() { file_autobahn_autobahn_proto_init() } @@ -2624,13 +2854,16 @@ func file_autobahn_autobahn_proto_init() { (*ConsensusReq_TimeoutVote)(nil), (*ConsensusReq_TimeoutQc)(nil), } + file_autobahn_autobahn_proto_msgTypes[36].OneofWrappers = []any{} + file_autobahn_autobahn_proto_msgTypes[37].OneofWrappers = []any{} + file_autobahn_autobahn_proto_msgTypes[38].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_autobahn_autobahn_proto_rawDesc), len(file_autobahn_autobahn_proto_rawDesc)), NumEnums: 0, - NumMessages: 36, + NumMessages: 40, NumExtensions: 0, NumServices: 0, }, diff --git a/sei-tendermint/internal/autobahn/pb/autobahn.wireguard.go b/sei-tendermint/internal/autobahn/pb/autobahn.wireguard.go index 38fea05046..f64f973969 100644 --- a/sei-tendermint/internal/autobahn/pb/autobahn.wireguard.go +++ b/sei-tendermint/internal/autobahn/pb/autobahn.wireguard.go @@ -119,6 +119,18 @@ func (*ConsensusReq) MaxSize() int { return 1111608 } +func (*Committee) MaxSize() int { + return 6200 +} + +func (*EpochRecord) MaxSize() int { + return 6214 +} + +func (*EpochMember) MaxSize() int { + return 60 +} + func init() { // Register the wireguard.Schema generated for autobahn.Timestamp. runtime.MustRegister[*Timestamp](runtime.Schema{ @@ -360,4 +372,27 @@ func init() { 5: {MaxCount: 1, Nested: utils.Some(reflect.TypeFor[*TimeoutQC]())}, }) + // Register the wireguard.Schema generated for autobahn.Committee. + runtime.MustRegister[*Committee](runtime.Schema{ + 1: {MaxCount: 100, Nested: utils.Some(reflect.TypeFor[*EpochMember]())}, + }) + + // Register the wireguard.Schema generated for autobahn.EpochRecord. + runtime.MustRegister[*EpochRecord](runtime.Schema{ + 1: {MaxCount: 1}, + 2: {MaxCount: 1, Nested: utils.Some(reflect.TypeFor[*Committee]())}, + }) + + // Register the wireguard.Schema generated for autobahn.PersistedEpochRegistry. + runtime.MustRegister[*PersistedEpochRegistry](runtime.Schema{ + 1: {Nested: utils.Some(reflect.TypeFor[*EpochRecord]())}, + 2: {MaxCount: 1, Nested: utils.Some(reflect.TypeFor[*EpochRecord]())}, + }) + + // Register the wireguard.Schema generated for autobahn.EpochMember. + runtime.MustRegister[*EpochMember](runtime.Schema{ + 1: {MaxCount: 1, Nested: utils.Some(reflect.TypeFor[*LaneID]())}, + 2: {MaxCount: 1}, + }) + } diff --git a/sei-tendermint/internal/autobahn/producer/mempool_test.go b/sei-tendermint/internal/autobahn/producer/mempool_test.go index cde96b205e..b21074c66b 100644 --- a/sei-tendermint/internal/autobahn/producer/mempool_test.go +++ b/sei-tendermint/internal/autobahn/producer/mempool_test.go @@ -225,7 +225,7 @@ func (env *testEnv) Run(ctx context.Context) error { if err != nil { return fmt.Errorf("app.FinalizeBlock(): %w", err) } - if err := env.data.PushAppHash(ctx, i, resp.AppHash); err != nil { + if err := env.data.PushAppHash(ctx, i, resp.AppHash, nil); err != nil { return err } } @@ -576,14 +576,10 @@ func TestProducer_LeaveCancelsAndRejoinStartsNewLane(t *testing.T) { return err } - epLeave, err := registry.ActivateEpoch( - 0, - map[types.PublicKey]uint64{b.Public(): 1}, - time.Time{}, registry.FirstBlock(), - ) - if err != nil { + if err := registry.StageAndActivate(0, map[types.PublicKey]uint64{b.Public(): 1}); err != nil { return err } + epLeave := registry.MustEpoch(2) if err := avail.TestDriveAdvance(ctx, availState, keys, epLeave.EpochIndex()); err != nil { return err } @@ -605,14 +601,10 @@ func TestProducer_LeaveCancelsAndRejoinStartsNewLane(t *testing.T) { return fmt.Errorf("TryInsertTx after leave: got %v, want ErrNotProducing", err) } - epJoin, err := registry.ActivateEpoch( - epLeave.EpochIndex(), - map[types.PublicKey]uint64{a.Public(): 1, b.Public(): 1}, - time.Time{}, registry.FirstBlock(), - ) - if err != nil { + if err := registry.StageAndActivate(1, map[types.PublicKey]uint64{a.Public(): 1, b.Public(): 1}); err != nil { return err } + epJoin := registry.MustEpoch(3) if err := avail.TestDriveAdvance(ctx, availState, keys, epJoin.EpochIndex()); err != nil { return err } @@ -651,12 +643,8 @@ func TestInsertTx_WaitUnblocksOnLeave(t *testing.T) { // Let InsertTx reach getMempool Wait (mempool still None — producer not running). time.Sleep(20 * time.Millisecond) - epLeave, err := registry.ActivateEpoch( - 0, - map[types.PublicKey]uint64{b.Public(): 1}, - time.Time{}, registry.FirstBlock(), - ) - require.NoError(t, err) + require.NoError(t, registry.StageAndActivate(0, map[types.PublicKey]uint64{b.Public(): 1})) + epLeave := registry.MustEpoch(2) require.NoError(t, scope.Run(ctx, func(ctx context.Context, sc scope.Scope) error { sc.SpawnBgNamed("avail", func() error { return utils.IgnoreCancel(availState.Run(ctx)) diff --git a/sei-tendermint/internal/p2p/giga/avail_test.go b/sei-tendermint/internal/p2p/giga/avail_test.go index fe05f51ffd..d30800ca21 100644 --- a/sei-tendermint/internal/p2p/giga/avail_test.go +++ b/sei-tendermint/internal/p2p/giga/avail_test.go @@ -112,7 +112,7 @@ func TestAvailClientServer(t *testing.T) { if err := utils.TestDiff(want, got); err != nil { return err } - if err := node.data.PushAppHash(ctx, n, h); err != nil { + if err := node.data.PushAppHash(ctx, n, h, nil); err != nil { return fmt.Errorf("node.data.PushAppHash(): %w", err) } } diff --git a/sei-tendermint/internal/p2p/giga/consensus_test.go b/sei-tendermint/internal/p2p/giga/consensus_test.go index 284932c6c6..3d6f468036 100644 --- a/sei-tendermint/internal/p2p/giga/consensus_test.go +++ b/sei-tendermint/internal/p2p/giga/consensus_test.go @@ -59,7 +59,7 @@ func TestConsensusClientServer(t *testing.T) { wantAppProposal = utils.Some(types.NewAppProposal(qc.QC().Proposal(), types.GenAppHash(rng))) } p := wantAppProposal.OrPanic("missing app proposal") - if err := n.data.PushAppHash(ctx, idx, p.AppHash()); err != nil { + if err := n.data.PushAppHash(ctx, idx, p.AppHash(), nil); err != nil { return fmt.Errorf("ds.PushAppProposal(): %w", err) } } diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go index e80837d5ba..bc50d1b714 100644 --- a/sei-tendermint/internal/p2p/giga_router_common.go +++ b/sei-tendermint/internal/p2p/giga_router_common.go @@ -78,7 +78,7 @@ func BuildDataState(cfg *GigaRouterCommonConfig, blockStore atypes.BlockStore) ( if err != nil { return nil, fmt.Errorf("genesis committee: %w", err) } - registry, err := epoch.NewRegistry(genesisCommittee, firstBlock, cfg.GenDoc.GenesisTime) + registry, err := epoch.NewRegistry(genesisCommittee, firstBlock, cfg.GenDoc.GenesisTime, cfg.PersistentStateDir) if err != nil { return nil, fmt.Errorf("epoch.NewRegistry(): %w", err) } @@ -165,14 +165,7 @@ func (r *gigaRouterCommon) BlockByHash(ctx context.Context, hash atypes.BlockHea // without a nil-check on the same type. The "no such block" case is // rejected at the BlockByHash call site before delegating here. // -// LastCommit is non-nil with empty Signatures, mirroring executeBlock's -// FinalizeBlock call which passes an empty abci.CommitInfo. Under Autobahn -// the committee is fixed by genesis (no validator-set updates), so the -// application is not in control of jailing — surfacing N "absent sig" -// entries here would make trace replay's BeginBlock bump missed-block -// counters and diverge from production. ToReqBeginBlock skips the per- -// validator loop when Signatures is empty, so empty Votes flow into -// distribution/slashing on both paths. +// LastCommit is non-nil with empty Signatures, matching executeBlock's empty CommitInfo. func (r *gigaRouterCommon) translateGlobalBlock(gb *atypes.GlobalBlock) *coretypes.ResultBlock { srcTxs := gb.Payload.Txs() tmTxs := make(types.Txs, len(srcTxs)) @@ -191,7 +184,12 @@ func (r *gigaRouterCommon) translateGlobalBlock(gb *atypes.GlobalBlock) *coretyp Height: utils.Clamp[int64](gb.GlobalNumber), Time: gb.Timestamp, }, - Data: types.Data{Txs: tmTxs}, + Data: types.Data{Txs: tmTxs}, + // Autobahn does not feed per-validator votes into the app. Filling N + // absent signatures would make trace replay's BeginBlock bump + // missed-block counters and diverge from production. ToReqBeginBlock + // skips the per-validator loop when Signatures is empty, so empty + // Votes flow into distribution/slashing on both paths. LastCommit: &types.Commit{}, }, } @@ -247,7 +245,11 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo if err != nil { return nil, fmt.Errorf("app.Commit(): %w", err) } - if err := r.data.PushAppHash(ctx, b.GlobalNumber, resp.AppHash); err != nil { + weights, err := committeeWeights(app.GetValidators()) + if err != nil { + return nil, err + } + if err := r.data.PushAppHash(ctx, b.GlobalNumber, resp.AppHash, weights); err != nil { return nil, fmt.Errorf("r.data.PushAppHash(%v): %w", b.GlobalNumber, err) } r.data.PushGasUsed(finalizeBlockGasUsed(resp)) @@ -433,7 +435,11 @@ func (r *gigaRouterCommon) runExecute(ctx context.Context) error { } // Losing a prefix of appHashes on crash is fine: AppQC is reached // once everyone votes on apphashes of a suffix of finalized blocks. - if err := r.data.PushAppHash(ctx, last, info.LastBlockAppHash); err != nil { + weights, err := committeeWeights(app.GetValidators()) + if err != nil { + return err + } + if err := r.data.PushAppHash(ctx, last, info.LastBlockAppHash, weights); err != nil { return fmt.Errorf("r.data.PushAppHash(): %w", err) } } @@ -571,3 +577,30 @@ func (r *gigaRouterCommon) evmProxy(validator atypes.PublicKey) utils.Option[*et } return utils.None[*ethrpc.Client]() } + +// committeeWeights maps the bonded validator set after Commit to voting power. +func committeeWeights(vals []abci.ValidatorUpdate) (map[atypes.PublicKey]uint64, error) { + weights := make(map[atypes.PublicKey]uint64, len(vals)) + for _, v := range vals { + if v.Power <= 0 { + continue + } + pk, err := crypto.PubKeyFromProto(v.PubKey) + if err != nil { + return nil, fmt.Errorf("PubKeyFromProto: %w", err) + } + apk, err := atypes.PublicKeyFromBytes(pk.Bytes()) + if err != nil { + return nil, fmt.Errorf("PublicKeyFromBytes: %w", err) + } + if _, dup := weights[apk]; dup { + return nil, fmt.Errorf("duplicate public key %s", apk) + } + power, ok := utils.SafeCast[uint64](v.Power) + if !ok { + return nil, fmt.Errorf("validator power %d does not fit uint64", v.Power) + } + weights[apk] = power + } + return weights, nil +} diff --git a/sei-tendermint/internal/p2p/giga_router_common_test.go b/sei-tendermint/internal/p2p/giga_router_common_test.go index 62954bf9e9..a3b4a6983d 100644 --- a/sei-tendermint/internal/p2p/giga_router_common_test.go +++ b/sei-tendermint/internal/p2p/giga_router_common_test.go @@ -11,6 +11,8 @@ import ( "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" + "github.com/sei-protocol/sei-chain/sei-tendermint/crypto" + "github.com/sei-protocol/sei-chain/sei-tendermint/crypto/ed25519" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/blockstore" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/data" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/epoch" @@ -118,7 +120,7 @@ func TestBuildDataStateStartsRecoveryAtAppTip(t *testing.T) { committee, err := atypes.NewCommittee(map[atypes.PublicKey]uint64{key.Public(): 1}) require.NoError(t, err) - registry, err := epoch.NewRegistry(committee, atypes.GlobalBlockNumber(genDoc.InitialHeight), genDoc.GenesisTime) + registry, err := epoch.NewRegistry(committee, atypes.GlobalBlockNumber(genDoc.InitialHeight), genDoc.GenesisTime, utils.None[string]()) require.NoError(t, err) qc, blocks := data.TestCommitQC(rng, registry.MustEpoch(0), keys, utils.None[*atypes.CommitQC]()) gr := qc.QC().GlobalRange() @@ -145,3 +147,37 @@ func TestBuildDataStateStartsRecoveryAtAppTip(t *testing.T) { require.NoError(t, err) require.Equal(t, blocks[gr.Len()/2].Header().Hash(), got.Header().Hash()) } + +func TestCommitteeWeights(t *testing.T) { + rng := utils.TestRng() + sk := ed25519.TestSecretKey(utils.GenBytes(rng, 32)) + wantPK := utils.OrPanic1(atypes.PublicKeyFromBytes(sk.Public().Bytes())) + got, err := committeeWeights([]abci.ValidatorUpdate{{ + PubKey: crypto.PubKeyToProto(sk.Public()), + Power: 42, + }}) + require.NoError(t, err) + require.Equal(t, map[atypes.PublicKey]uint64{wantPK: 42}, got) +} + +func TestCommitteeWeights_SkipsZeroPower(t *testing.T) { + rng := utils.TestRng() + sk := ed25519.TestSecretKey(utils.GenBytes(rng, 32)) + got, err := committeeWeights([]abci.ValidatorUpdate{{ + PubKey: crypto.PubKeyToProto(sk.Public()), + Power: 0, + }}) + require.NoError(t, err) + require.Empty(t, got) +} + +func TestCommitteeWeights_DuplicateKey(t *testing.T) { + rng := utils.TestRng() + sk := ed25519.TestSecretKey(utils.GenBytes(rng, 32)) + pk := crypto.PubKeyToProto(sk.Public()) + _, err := committeeWeights([]abci.ValidatorUpdate{ + {PubKey: pk, Power: 1}, + {PubKey: pk, Power: 2}, + }) + require.Error(t, err) +}