diff --git a/sei-tendermint/autobahn/types/epoch_duo.go b/sei-tendermint/autobahn/types/epoch_duo.go index 95b07ae467..39f5acad4c 100644 --- a/sei-tendermint/autobahn/types/epoch_duo.go +++ b/sei-tendermint/autobahn/types/epoch_duo.go @@ -18,67 +18,32 @@ type EpochDuo struct { // NewEpochDuo builds a Prev|Current window. Panics unless // prev≠None ⇔ current.EpochIndex()>0, and when Prev is present it must be // contiguous with Current. -func NewEpochDuo(current *Epoch, prev utils.Option[*Epoch]) EpochDuo { +func NewEpochDuo(current *Epoch, prev utils.Option[*Epoch]) (EpochDuo,error) { cur := current.EpochIndex() p, hasPrev := prev.Get() if hasPrev != (cur > 0) { - panic(fmt.Sprintf("NewEpochDuo: Prev present=%v but Current epoch %d (want Prev iff Current>0)", - hasPrev, cur)) + return EpochDuo{},fmt.Errorf("NewEpochDuo: Prev present=%v but Current epoch %d (want Prev iff Current>0)", + hasPrev, cur) } if hasPrev { if p.EpochIndex()+1 != cur { - panic(fmt.Sprintf("NewEpochDuo: Prev epoch %d not contiguous with Current %d", - p.EpochIndex(), cur)) + return EpochDuo{},fmt.Errorf("NewEpochDuo: Prev epoch %d not contiguous with Current %d", + p.EpochIndex(), cur) } if got, want := p.RoadRange().Next, current.RoadRange().First; got != want { - panic(fmt.Sprintf("NewEpochDuo: Prev roads end at %d, Current starts at %d", got, want)) + panic(fmt.Errorf("NewEpochDuo: Prev roads end at %d, Current starts at %d", got, want)) } } - return EpochDuo{Prev: prev, Current: current} -} - -var ErrRoadBeforeWindow = errors.New("road before epoch duo window") -var ErrRoadAfterWindow = errors.New("road after epoch duo window") - -// RoadStatus classifies a road relative to a Prev|Current window for admit waits. -type RoadStatus int - -const ( - RoadReady RoadStatus = iota // in the admitted window - RoadStale // behind the window (soft-drop / ErrPruned) - RoadFuture // ahead of the window (backpressure wait) -) - -// RoadStatusCurrent classifies roadIdx against Current only (CommitQC tip). -func (w EpochDuo) RoadStatusCurrent(roadIdx RoadIndex) RoadStatus { - if w.Current.RoadRange().Has(roadIdx) { - return RoadReady - } - if roadIdx < w.Current.RoadRange().First { - return RoadStale - } - return RoadFuture -} - -// RoadStatusDuo classifies roadIdx against Prev|Current (AppVote/AppQC). -func (w EpochDuo) RoadStatusDuo(roadIdx RoadIndex) RoadStatus { - _, err := w.EpochForRoad(roadIdx) - if err == nil { - return RoadReady - } - if errors.Is(err, ErrRoadBeforeWindow) { - return RoadStale - } - return RoadFuture + return EpochDuo{Prev: prev, Current: current},nil } // EpochForRoad returns the epoch containing roadIdx. // Window is [Prev.First or Current.First, Current.Next). Outside → // ErrRoadBeforeWindow / ErrRoadAfterWindow. Under contiguous Prev|Current there // is no gap, so a miss after the after-window check is always before-window. -func (w EpochDuo) EpochForRoad(roadIdx RoadIndex) (*Epoch, error) { +func (w EpochDuo) ByRoad(roadIdx RoadIndex) (*Epoch, error) { if roadIdx >= w.Current.RoadRange().Next { - return nil, fmt.Errorf("road %d after window %v: %w", roadIdx, w, ErrRoadAfterWindow) + return nil, errors.New("road belongs to future epoch") } if w.Current.RoadRange().Has(roadIdx) { return w.Current, nil @@ -86,7 +51,7 @@ func (w EpochDuo) EpochForRoad(roadIdx RoadIndex) (*Epoch, error) { if prev, ok := w.Prev.Get(); ok && prev.RoadRange().Has(roadIdx) { return prev, nil } - return nil, fmt.Errorf("road %d before window %v: %w", roadIdx, w, ErrRoadBeforeWindow) + return nil, ErrPruned } func (w EpochDuo) String() string { diff --git a/sei-tendermint/autobahn/types/proposal.go b/sei-tendermint/autobahn/types/proposal.go index a45e14bc84..44b2f7fa18 100644 --- a/sei-tendermint/autobahn/types/proposal.go +++ b/sei-tendermint/autobahn/types/proposal.go @@ -125,28 +125,33 @@ func (v View) Next() View { return v } +type ConsensusSpec struct { + Epochs EpochDuo + CommitQC utils.Option[*CommitQC] + // Genesis floors used only when CommitQC is None (chain start). + GenesisFirstBlock GlobalBlockNumber + GenesisTimestamp time.Time +} + +// Epoch is the proposing/voting epoch (Epochs.Current). +func (cs *ConsensusSpec) Epoch() *Epoch { return cs.Epochs.Current } + +func (cs *ConsensusSpec) Index() RoadIndex { return NextIndexOpt(cs.CommitQC) } + // ViewSpec is the local context for starting a view: justification QCs plus a // Prev|Current EpochDuo. Attached AppQC may be Current or Current-1 (Prev lag). type ViewSpec struct { // WARNING: currently we have implicit assumption that // TimeoutQC.View().Index == CommitQC.Index.Next(), // I.e. that TimeoutQC comes from the expected consensus instance. - CommitQC utils.Option[*CommitQC] + *ConsensusSpec TimeoutQC utils.Option[*TimeoutQC] - Epochs EpochDuo - // Genesis floors used only when CommitQC is None (chain start). - // Copied from Registry; ignored when CommitQC is present. - GenesisFirstBlock GlobalBlockNumber - GenesisTimestamp time.Time } -// Epoch is the proposing/voting epoch (Epochs.Current). -func (vs *ViewSpec) Epoch() *Epoch { return vs.Epochs.Current } - // NextGlobalBlock returns the first global block number expected in the next proposal. // CommitQC is None only at chain start, in which case it returns GenesisFirstBlock. // For all other views, including the first view of a non-genesis epoch, CommitQC is present and it returns CommitQC.GlobalRange().Next. -func (vs *ViewSpec) NextGlobalBlock() GlobalBlockNumber { +func (vs ViewSpec) NextGlobalBlock() GlobalBlockNumber { if cQC, ok := vs.CommitQC.Get(); ok { return cQC.GlobalRange().Next } @@ -154,16 +159,18 @@ func (vs *ViewSpec) NextGlobalBlock() GlobalBlockNumber { } // View is the view justified by vs. -func (vs *ViewSpec) View() View { - idx := NextIndexOpt(vs.CommitQC) - if view := NextViewOpt(vs.TimeoutQC); view.Index == idx { - view.EpochIndex = vs.Epoch().EpochIndex() - return view +func (vs ViewSpec) View() View { + if qc,ok := vs.TimeoutQC.Get(); ok { + return qc.View() + } + return View{ + EpochIndex: vs.Epoch().EpochIndex(), + Index: vs.Index(), + Number: 0, } - return View{Index: idx, Number: 0, EpochIndex: vs.Epoch().EpochIndex()} } -func (vs *ViewSpec) NextTimestamp() time.Time { +func (vs ViewSpec) NextTimestamp() time.Time { if cQC, ok := vs.CommitQC.Get(); ok { return cQC.Proposal().NextTimestamp() } diff --git a/sei-tendermint/autobahn/types/testonly.go b/sei-tendermint/autobahn/types/testonly.go index 36c1df8681..ce2276747e 100644 --- a/sei-tendermint/autobahn/types/testonly.go +++ b/sei-tendermint/autobahn/types/testonly.go @@ -16,7 +16,7 @@ import ( // iff Current is epoch 0; otherwise a synthetic contiguous Prev is created. func EpochDuoForTest(current *Epoch) EpochDuo { if current.EpochIndex() == 0 { - return NewEpochDuo(current, utils.None[*Epoch]()) + return utils.OrPanic1(NewEpochDuo(current, utils.None[*Epoch]())) } first := current.RoadRange().First if first == 0 { @@ -28,7 +28,7 @@ func EpochDuoForTest(current *Epoch) EpochDuo { RoadRange{First: 0, Next: first}, current.Committee(), ) - return NewEpochDuo(current, utils.Some(prev)) + return utils.OrPanic1(NewEpochDuo(current, utils.Some(prev))) } // BuildCommitQC builds a valid CommitQC from explicit lane QCs and an optional app QC. @@ -46,7 +46,7 @@ func BuildCommitQC( laneQCs map[LaneID]*LaneQC, appQC utils.Option[*AppQC], ) *CommitQC { - vs := ViewSpec{CommitQC: prev, Epochs: EpochDuoForTest(epoch)} + vs := ViewSpec{ConsensusSpec:&ConsensusSpec{CommitQC: prev, Epochs: EpochDuoForTest(epoch)}} if len(laneQCs) == 0 { laneQCs = oneBlockLaneQCMap(vs, keys) } diff --git a/sei-tendermint/internal/autobahn/avail/app.go b/sei-tendermint/internal/autobahn/avail/app.go index 65dc14178b..68c424d1e1 100644 --- a/sei-tendermint/internal/autobahn/avail/app.go +++ b/sei-tendermint/internal/autobahn/avail/app.go @@ -2,6 +2,7 @@ package avail import ( "context" + "errors" "fmt" "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" @@ -10,31 +11,33 @@ import ( // appProgress owns the in-memory AppQC tip and AppVote accumulators. type appProgress struct { - latestAppQC utils.Option[*types.AppQC] - votes *queue[types.GlobalBlockNumber, appVotes] + anchor utils.Option[*PruneAnchor] + votes *queue[types.GlobalBlockNumber, appVotes] } // LastAppQC returns the latest observed AppQC. func (s *State) LastAppQC() utils.Option[*types.AppQC] { for inner := range s.inner.Lock() { - return inner.app.latestAppQC + if anchor,ok := inner.app.anchor.Get(); ok { + return utils.Some(anchor.AppQC) + } } - panic("unreachable") + return utils.None[*types.AppQC]() } // WaitForAppQC waits until there is an AppQC for the given index or higher. // Returns this AppQC and the corresponding CommitQC. // Together they provide enough information to prune the availability state. -func (s *State) WaitForAppQC(ctx context.Context, idx types.RoadIndex) (*types.AppQC, *types.CommitQC, error) { +func (s *State) waitForAnchor(ctx context.Context, idx types.RoadIndex) (*PruneAnchor, error) { for inner, ctrl := range s.inner.Lock() { for { - if appQC, ok := inner.app.latestAppQC.Get(); ok { - if x := appQC.Proposal().RoadIndex(); x >= idx && inner.commits.qcs.next > x { - return appQC, inner.commits.qcs.q[x], nil + if anchor, ok := inner.app.anchor.Get(); ok { + if x := anchor.AppQC.Proposal().RoadIndex(); x >= idx { + return anchor, nil } } if err := ctrl.Wait(ctx); err != nil { - return nil, nil, err + return nil, err } } } @@ -45,38 +48,25 @@ func (s *State) WaitForAppQC(ctx context.Context, idx types.RoadIndex) (*types.A // Same admit-then-verify as PushAppQC: far-future roads park until the duo // and CommitQC tip catch up (one stream goroutine; does not block others). func (s *State) PushAppVote(ctx context.Context, v *types.Signed[*types.AppVote]) error { - idx := v.Msg().Proposal().RoadIndex() - // A vote may arrive before its CommitQC advances the tip. - if err := s.waitForCommitQC(ctx, idx); err != nil { - return err - } - // Too-early roads (ahead of Prev|Current) backpressure; too-late are dropped. - admitted, err := s.waitForEpochDuoOrDropStale(ctx, "AppVote", idx) + qc,epoch,err := s.commitQCAndEpoch(ctx, v.Msg().Proposal().RoadIndex()) if err != nil { + if errors.Is(err,types.ErrPruned) { + return nil + } return err } - duo, ok := admitted.Get() - if !ok { - return nil + if err := v.Msg().Proposal().Verify(qc); err != nil { + return fmt.Errorf("invalid vote: %w", err) } - ep := utils.OrPanic1(duo.EpochForRoad(idx)) - if got, want := v.Msg().Proposal().EpochIndex(), ep.EpochIndex(); got != want { - return fmt.Errorf("appVote epoch_index %d, want %d", got, want) - } - committee := ep.Committee() + committee := epoch.Committee() if err := v.VerifySig(committee); err != nil { return fmt.Errorf("v.VerifySig(): %w", err) } for inner, ctrl := range s.inner.Lock() { // Early exit if not useful (we collect <=1 AppQC per road index). - if idx < types.NextOpt(inner.app.latestAppQC) { + if qc.Index() < types.NextOpt(inner.app.anchor) { return nil } - // Verify the vote against the CommitQC. - qc := inner.commits.qcs.q[idx] - if err := v.Msg().Proposal().Verify(qc); err != nil { - return fmt.Errorf("invalid vote: %w", err) - } // Push the vote. n := v.Msg().Proposal().GlobalNumber() q := inner.app.votes @@ -87,10 +77,8 @@ func (s *State) PushAppVote(ctx context.Context, v *types.Signed[*types.AppVote] if !ok { return nil } - updated, err := inner.pushPruneAnchor(&PruneAnchor{AppQC: appQC, CommitQC: qc}) - if err != nil { - return err - } + // Anchor is always valid here. + updated := utils.OrPanic1(inner.pushPruneAnchor(&PruneAnchor{AppQC: appQC, CommitQC: qc, Epoch: epoch})) if updated { ctrl.Updated() } @@ -98,19 +86,25 @@ func (s *State) PushAppVote(ctx context.Context, v *types.Signed[*types.AppVote] return nil } -// PushAppQC requires a justifying CommitQC. Epoch slide is async in -// runAdvanceEpoch (same as PushCommitQC). Prune before insert so latestAppQC is -// visible before the advance task observes the new tip. -// -// Same admit-then-verify as PushCommitQC. +// PushAppQC requires a justifying CommitQC. +// Prunes state up to AppQC.Proposal().Index(). func (s *State) PushAppQC(ctx context.Context, appQC *types.AppQC, commitQC *types.CommitQC) error { + // If epoch is from the future then we are unable to process AppQC. + // If epoch is pruned, then it is from the past and there is no point participating in the consensus. + epoch, err := s.data.Registry().WaitForEpoch(ctx, appQC.Proposal().EpochIndex()) + if err != nil { + if errors.Is(err,types.ErrPruned) { + return nil + } + return err + } // Check whether it is needed before verifying. for inner := range s.inner.Lock() { - if types.NextOpt(inner.app.latestAppQC) > appQC.Proposal().RoadIndex() { + if types.NextOpt(inner.app.anchor) >= appQC.Next() { return nil } } - // Pair consistency only; ahead-of-window still waits in waitForEpochDuo. + // Verify appQC <-> commitQC consistency. if appQC.Proposal().RoadIndex() != commitQC.Proposal().Index() { return fmt.Errorf("mismatched QCs: appQC index %v, commitQC index %v", appQC.Proposal().RoadIndex(), commitQC.Proposal().Index()) } @@ -120,38 +114,17 @@ func (s *State) PushAppQC(ctx context.Context, appQC *types.AppQC, commitQC *typ if !commitQC.GlobalRange().Has(appQC.Proposal().GlobalNumber()) { return fmt.Errorf("appQC GlobalNumber not in commitQC range") } - idx := commitQC.Proposal().Index() - admitted, err := s.waitForEpochDuoOrDropStale(ctx, "AppQC", idx) - if err != nil { - return err - } - duo, ok := admitted.Get() - if !ok { - return nil - } - ep := utils.OrPanic1(duo.EpochForRoad(idx)) - if err := appQC.Verify(ep); err != nil { + if err := appQC.Verify(epoch); err != nil { return fmt.Errorf("appQC.Verify(): %w", err) } - if err := commitQC.Verify(ep); err != nil { + if err := commitQC.Verify(epoch); err != nil { return fmt.Errorf("commitQC.Verify(): %w", err) } - // Seal CommitQC paired with this AppQC: incoming AppQC satisfies the prune - // leash; still wait on the execution leash when idx closes Current. - if duo.Current.RoadRange().Next-1 == idx && ep.EpochIndex() == duo.Current.EpochIndex() { - if err := s.waitSealLeashes(ctx, duo.Current, idx, utils.Some(appQC.Proposal().EpochIndex())); err != nil { - return err - } - } + anchor := &PruneAnchor{AppQC: appQC, CommitQC: commitQC, Epoch: epoch} for inner, ctrl := range s.inner.Lock() { - updated, err := inner.pushPruneAnchor(&PruneAnchor{AppQC: appQC, CommitQC: commitQC}) - if err != nil { - return err - } - if !updated { - return nil + if utils.OrPanic1(inner.pushPruneAnchor(anchor)) { + ctrl.Updated() } - ctrl.Updated() } return nil } diff --git a/sei-tendermint/internal/autobahn/avail/block_votes.go b/sei-tendermint/internal/autobahn/avail/block_votes.go index d971be183d..cfe8315248 100644 --- a/sei-tendermint/internal/autobahn/avail/block_votes.go +++ b/sei-tendermint/internal/autobahn/avail/block_votes.go @@ -71,8 +71,7 @@ func (bv *blockVotes) pushVote(ep *types.Epoch, vote *types.Signed[*types.LaneVo // reweight recomputes already-stored votes under new Current after advanceEpoch. // Zero-weight signers are removed from byKey. Callers wake waiters via // ctrl.Updated() after advanceEpoch (not via a return flag). -func (bv *blockVotes) reweight(newEpoch *types.Epoch) { - c := newEpoch.Committee() +func (bv *blockVotes) reweight(c *types.Committee) { clear(bv.byHash) quorum := c.LaneQuorum() for k, vote := range bv.byKey { diff --git a/sei-tendermint/internal/autobahn/avail/commit.go b/sei-tendermint/internal/autobahn/avail/commit.go index d04df157a7..fd824c3cf2 100644 --- a/sei-tendermint/internal/autobahn/avail/commit.go +++ b/sei-tendermint/internal/autobahn/avail/commit.go @@ -6,15 +6,35 @@ import ( "fmt" "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" ) -// waitForCommitQC waits until the durable CommitQC tip has advanced past idx. +// waitForCommitQC waits until commitqc queue reaches idx. +// CommitQC at idx is NOT guaranteed to be persisted yet. func (s *State) waitForCommitQC(ctx context.Context, idx types.RoadIndex) error { - _, err := s.LastCommitQC().Wait(ctx, func(qc utils.Option[*types.CommitQC]) bool { - return types.NextIndexOpt(qc) > idx - }) - return err + for inner,ctrl := range s.inner.Lock() { + if err := ctrl.WaitUntil(ctx, func() bool { return idx < inner.commits.qcs.next }); err!=nil { + return err + } + } + return nil +} + +// Fetches CommitQC and a matching epoch. They are NOT guaranteed to be persisted. +func (s *State) commitQCAndEpoch(ctx context.Context, idx types.RoadIndex) (*types.CommitQC, *types.Epoch, error) { + if err := s.waitForCommitQC(ctx, idx); err != nil { + return nil, nil, err + } + for inner := range s.inner.Lock() { + if idx < inner.commits.qcs.first { + return nil, nil, types.ErrPruned + } + qc := inner.commits.qcs.q[idx] + if epoch := inner.epoch.Load(); epoch.EpochIndex()==qc.Proposal().EpochIndex() { + return qc,epoch,nil + } + return qc,inner.app.anchor.OrPanic("missing anchor").Epoch,nil + } + panic("unreachable") } // CommitQC returns the CommitQC for the given index. @@ -31,64 +51,46 @@ func (s *State) CommitQC(ctx context.Context, idx types.RoadIndex) (*types.Commi panic("unreachable") } -// PushCommitQC admits qc for Current only (too early waits; stale drops). -// Epoch slide is async in runAdvanceEpoch (tip may sit at Current.Next while -// Current still N; N+1 CommitQCs park on waitForEpoch until the duo advances). -// -// Seal (last road of Current): prune + execution leashes before admit -// (interlocking doc CommitQC admission). -// -// Admit-then-verify is intentional backpressure for ahead-of-window QCs. +// PushCommitQC pushes qc to the commit queue. +// Blocks until all previous CommitQCs are available and State enters this qc's epoch. +// Silently drops qc if not needed. +// NOT guaranteed to be persisted yet. func (s *State) PushCommitQC(ctx context.Context, qc *types.CommitQC) error { - idx := qc.Proposal().Index() - if idx > 0 { - if err := s.waitForCommitQC(ctx, idx-1); err != nil { + // Await previous CommitQC. + if i := qc.Proposal().Index(); i>0 { + if err:=s.waitForCommitQC(ctx,i-1); err!=nil { return err } } - admitted, err := s.waitForEpochOrDropStale(ctx, "CommitQC", idx) + // Await Epoch. + epoch, err := s.Epoch(ctx, qc.Proposal().EpochIndex()) if err != nil { + if errors.Is(err,types.ErrPruned); err!=nil { + return nil + } return err } - duo, ok := admitted.Get() - if !ok { - return nil - } - ep := duo.Current - if err := qc.Verify(ep); err != nil { + // Verify qc. + if err := qc.Verify(epoch); err != nil { return fmt.Errorf("qc.Verify(): %w", err) } - if err := s.waitSealLeashes(ctx, ep, idx, utils.None[types.EpochIndex]()); err != nil { - return err - } - + // Push. for inner, ctrl := range s.inner.Lock() { - if !inner.commits.push(qc) { - return nil + if inner.commits.push(qc) { + ctrl.Updated() } - // persistedCommitQC advances only after durable persist (or no-op persister). - ctrl.Updated() - return nil } return nil } // fullCommitQC returns the FullCommitQC for road n. -// ErrRoadBeforeWindow → ErrPruned (export may jump ahead). ErrRoadAfterWindow hard-fails. func (s *State) fullCommitQC(ctx context.Context, n types.RoadIndex) (*types.FullCommitQC, error) { - qc, err := s.CommitQC(ctx, n) - if err != nil { - return nil, err - } - ep, err := s.epochDuo.Load().EpochForRoad(qc.Proposal().Index()) + qc, epoch, err := s.commitQCAndEpoch(ctx, n) if err != nil { - if errors.Is(err, types.ErrRoadBeforeWindow) { - return nil, types.ErrPruned - } return nil, err } var commitHeaders []*types.BlockHeader - for lane := range ep.Committee().Lanes().All() { + for lane := range epoch.Committee().Lanes().All() { headers, err := s.headers(ctx, qc.LaneRange(lane)) if err != nil { return nil, err diff --git a/sei-tendermint/internal/autobahn/avail/commit_progress.go b/sei-tendermint/internal/autobahn/avail/commit_progress.go index b1ea6b832b..66ba90db63 100644 --- a/sei-tendermint/internal/autobahn/avail/commit_progress.go +++ b/sei-tendermint/internal/autobahn/avail/commit_progress.go @@ -11,7 +11,7 @@ import ( // write persistedCommitQC. type commitProgress struct { qcs *queue[types.RoadIndex, *types.CommitQC] - persistedCommitQC utils.AtomicSend[utils.Option[*types.CommitQC]] + consensusSpec utils.AtomicSend[*types.ConsensusSpec] } // push inserts qc at qcs.next. Returns false if idx is not the tip @@ -25,8 +25,3 @@ func (c *commitProgress) push(qc *types.CommitQC) bool { metrics.ObserveCommitQC(qc) return true } - -// markPersisted publishes the latest durably persisted CommitQC. -func (c *commitProgress) markPersisted(qc *types.CommitQC) { - c.persistedCommitQC.Store(utils.Some(qc)) -} diff --git a/sei-tendermint/internal/autobahn/avail/epoch_transition.go b/sei-tendermint/internal/autobahn/avail/epoch_transition.go index 9c66517089..79dc58446a 100644 --- a/sei-tendermint/internal/autobahn/avail/epoch_transition.go +++ b/sei-tendermint/internal/autobahn/avail/epoch_transition.go @@ -3,144 +3,17 @@ package avail import ( "context" "errors" - "fmt" - "log/slog" "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" - "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" ) -// epochProgress is the active Prev|Current admission window. -type epochProgress = utils.AtomicSend[types.EpochDuo] - -func logStaleRoad(what string, roadIdx types.RoadIndex, duo types.EpochDuo) { - // Debug: Info is too chatty at epoch boundaries (many peers a road behind). - logger.Debug("dropping stale "+what+": road behind window", - slog.Uint64("road", uint64(roadIdx)), "duo", duo.String()) -} - -// waitUntilRoad waits until status is not RoadFuture; RoadStale → ErrPruned -// with the deciding duo still returned for logging. -func (s *State) waitUntilRoad( - ctx context.Context, - roadIdx types.RoadIndex, - status func(types.EpochDuo) types.RoadStatus, -) (types.EpochDuo, error) { - duo, err := s.epochDuo.Wait(ctx, func(duo types.EpochDuo) bool { - return status(duo) != types.RoadFuture - }) - if err != nil { - return types.EpochDuo{}, err - } - switch status(duo) { - case types.RoadReady: - return duo, nil - case types.RoadStale: - return duo, types.ErrPruned - default: - // Wait predicate forbids Future; hitting it is an internal bug. - panic(fmt.Sprintf("waitUntilRoad: unexpected RoadFuture for road %d after Wait", roadIdx)) - } -} - -// waitForEpoch waits until roadIdx is in Current (CommitQC tip). -func (s *State) waitForEpoch(ctx context.Context, roadIdx types.RoadIndex) (types.EpochDuo, error) { - return s.waitUntilRoad(ctx, roadIdx, func(d types.EpochDuo) types.RoadStatus { - return d.RoadStatusCurrent(roadIdx) - }) -} - -// waitForEpochDuo waits until roadIdx is in Prev|Current (AppVote/AppQC). -func (s *State) waitForEpochDuo(ctx context.Context, roadIdx types.RoadIndex) (types.EpochDuo, error) { - return s.waitUntilRoad(ctx, roadIdx, func(d types.EpochDuo) types.RoadStatus { - return d.RoadStatusDuo(roadIdx) - }) -} - -// waitForEpochOrDropStale is PushCommitQC admit: wait for Current, soft-drop if stale. -func (s *State) waitForEpochOrDropStale( - ctx context.Context, what string, roadIdx types.RoadIndex, -) (utils.Option[types.EpochDuo], error) { - return s.waitRoadOrDropStale(ctx, what, roadIdx, s.waitForEpoch) -} - -// waitForEpochDuoOrDropStale is PushAppVote/PushAppQC admit: wait for Prev|Current, soft-drop if stale. -func (s *State) waitForEpochDuoOrDropStale( - ctx context.Context, what string, roadIdx types.RoadIndex, -) (utils.Option[types.EpochDuo], error) { - return s.waitRoadOrDropStale(ctx, what, roadIdx, s.waitForEpochDuo) -} - -func (s *State) waitRoadOrDropStale( - ctx context.Context, - what string, - roadIdx types.RoadIndex, - wait func(context.Context, types.RoadIndex) (types.EpochDuo, error), -) (utils.Option[types.EpochDuo], error) { - duo, err := wait(ctx, roadIdx) - if err != nil { - if errors.Is(err, types.ErrPruned) { - logStaleRoad(what, roadIdx, duo) - return utils.None[types.EpochDuo](), nil - } - return utils.None[types.EpochDuo](), err - } - return utils.Some(duo), nil -} - -// waitForAppQC blocks until latest AppQC is from epochIdx or later. -// -// Seal prune leash (interlocking doc): Availability admits the last CommitQC of -// epoch N only after an AppQC for N exists. Also used by runAdvanceEpoch as a -// no-op once admit already waited. Epoch 0 is not special-cased: leaving 0 -// still needs an AppQC anchor for restart (Current>0 requires one). -func (s *State) waitForAppQC(ctx context.Context, epochIdx types.EpochIndex) error { - for inner, ctrl := range s.inner.Lock() { - ready := func() bool { - appQC, ok := inner.app.latestAppQC.Get() - if !ok { - return false - } - return appQC.Proposal().EpochIndex() >= epochIdx - } - if ready() { - return nil - } - attrs := []any{slog.Uint64("want_epoch", uint64(epochIdx))} - if appQC, ok := inner.app.latestAppQC.Get(); ok { - attrs = append(attrs, - slog.Uint64("latest_app_qc_road", uint64(appQC.Proposal().RoadIndex())), - slog.Uint64("latest_app_qc_epoch", uint64(appQC.Proposal().EpochIndex())), - ) - } - logger.Warn("waiting for AppQC before sealing epoch", attrs...) - return ctrl.WaitUntil(ctx, ready) - } - panic("unreachable") -} - -// waitSealLeashes enforces the interlocking-doc seal conditions before admitting -// the last CommitQC of ep: AppQC for ep (unless incomingAppEpoch already -// satisfies) and registry WaitForDuo for the next road (execution leash). -func (s *State) waitSealLeashes( - ctx context.Context, - ep *types.Epoch, - idx types.RoadIndex, - incomingAppEpoch utils.Option[types.EpochIndex], -) error { - last := ep.RoadRange().Next - 1 - if idx != last { - return nil - } - if e, ok := incomingAppEpoch.Get(); !ok || e < ep.EpochIndex() { - if err := s.waitForAppQC(ctx, ep.EpochIndex()); err != nil { - return err - } - } - if _, err := s.data.Registry().WaitForDuo(ctx, last+1); err != nil { - return fmt.Errorf("WaitForDuo(%d): %w", last+1, err) - } - return nil +// waitForEpoch wait for epoch to advance to roadIdx. +// Returned EpochDuo may be past roadIdx. +func (s *State) Epoch(ctx context.Context, i types.EpochIndex) (*types.Epoch, error) { + epoch,err := s.epoch.Wait(ctx, func(epoch *types.Epoch) bool { return i <= epoch.EpochIndex() }) + if err!=nil { return nil, err } + if epoch.EpochIndex()!=i { return nil,types.ErrPruned } + return epoch,nil } // runAdvanceEpoch is the sole post-construction writer of epochDuo. When @@ -149,37 +22,30 @@ func (s *State) waitSealLeashes( // leashes (no-op if already met), then advances. N+1 CommitQCs park on // waitForEpoch until the duo slides. func (s *State) runAdvanceEpoch(ctx context.Context) error { - for { - duo := s.epochDuo.Load() - epochIdx := duo.Current.EpochIndex() - last := duo.Current.RoadRange().Next - 1 - + return s.epoch.Iter(ctx, func(ctx context.Context, epoch *types.Epoch) error { for inner, ctrl := range s.inner.Lock() { - if err := ctrl.WaitUntil(ctx, func() bool { - return inner.commits.qcs.next > last - }); err != nil { - return err + return ctrl.WaitUntil(ctx,func() bool { + // All commits of the current epoch are required. + if inner.commits.qcs.next < epoch.RoadRange().Next { + return false + } + anchor,ok := inner.app.anchor.Get() + // Anchor in the current epoch is required. + return ok && anchor.Epoch.EpochIndex() >= epoch.EpochIndex() + }) + } + epoch,err:=s.data.Registry().WaitForEpoch(ctx,epoch.EpochIndex()+1) + if err!=nil { + if errors.Is(err,types.ErrPruned); err!=nil { + return nil } - } - - if err := s.waitForAppQC(ctx, epochIdx); err != nil { return err } - nextDuo, err := s.data.Registry().WaitForDuo(ctx, last+1) - if err != nil { - return err - } - - for inner, ctrl := range s.inner.Lock() { - live := inner.epoch.Load() - if live.Current.EpochIndex() != epochIdx { - break - } - if inner.commits.qcs.next <= last { - break + for inner,ctrl := range s.inner.Lock() { + if inner.advanceEpoch(epoch) { + ctrl.Updated() } - inner.advanceEpoch(nextDuo) - ctrl.Updated() } - } + return nil + }) } diff --git a/sei-tendermint/internal/autobahn/avail/inner.go b/sei-tendermint/internal/autobahn/avail/inner.go index 2562b4897e..86761e599f 100644 --- a/sei-tendermint/internal/autobahn/avail/inner.go +++ b/sei-tendermint/internal/autobahn/avail/inner.go @@ -18,7 +18,7 @@ import ( // MaybePruneAndPersistLane, but the new member must also appear in // inner.lanes before the next persist cycle. type inner struct { - epoch epochProgress + epoch utils.AtomicSend[*types.Epoch] app appProgress commits commitProgress lanes laneCollection @@ -51,9 +51,9 @@ func (ls *loadedAvailState) nextCommitQC() types.RoadIndex { } func newInner(registry *epoch.Registry, commitTip types.RoadIndex, loaded utils.Option[*loadedAvailState]) (*inner, error) { - startEpochDuo, err := registry.DuoAt(commitTip) - if err != nil { - return nil, fmt.Errorf("DuoAt(%d): %w", commitTip, err) + startEpochDuo, ok := registry.DuoAt(commitTip) + if !ok { + return nil, fmt.Errorf("DuoAt(%d): epoch missing", commitTip) } lanes := map[types.LaneID]*laneState{} // TODO(lane-id): also seed Prev lanes before pruning so restart applies the @@ -63,15 +63,20 @@ func newInner(registry *epoch.Registry, commitTip types.RoadIndex, loaded utils. lanes[lane] = newLaneState() } + genesisSpec,ok := registry.ConsensusSpec(utils.None[*types.CommitQC]()) + if !ok { + return nil, fmt.Errorf("registry.ConsensusSpec(): not found") + } + i := &inner{ - epoch: utils.NewAtomicSend(startEpochDuo), + epoch: utils.NewAtomicSend(startEpochDuo.Current), app: appProgress{ - latestAppQC: utils.None[*types.AppQC](), - votes: newQueue[types.GlobalBlockNumber, appVotes](), + anchor: utils.None[*PruneAnchor](), + votes: newQueue[types.GlobalBlockNumber, appVotes](), }, commits: commitProgress{ qcs: newQueue[types.RoadIndex, *types.CommitQC](), - persistedCommitQC: utils.NewAtomicSend(utils.None[*types.CommitQC]()), + consensusSpec: utils.NewAtomicSend(genesisSpec), }, lanes: laneCollection{byID: lanes}, } @@ -124,7 +129,9 @@ func newInner(registry *epoch.Registry, commitTip types.RoadIndex, loaded utils. i.commits.qcs.pushBack(lqc.QC) } if i.commits.qcs.next > i.commits.qcs.first { - i.commits.markPersisted(i.commits.qcs.q[i.commits.qcs.next-1]) + spec,ok := registry.ConsensusSpec(utils.Some(i.commits.qcs.q[i.commits.qcs.next-1])) + if !ok { return nil, fmt.Errorf("registry.ConsensusSpec(): not found") } + i.commits.consensusSpec.Store(spec) } // Restore blocks; create queues for any WAL lane (including outside Current). @@ -161,14 +168,11 @@ func newInner(registry *epoch.Registry, commitTip types.RoadIndex, loaded utils. // verifyCommitQCInDuo verifies qc against startEpochDuo (Prev|Current at restore). func verifyCommitQCInDuo(duo types.EpochDuo, qc *types.CommitQC) error { - ep, err := duo.EpochForRoad(qc.Proposal().Index()) + ep, err := duo.ByRoad(qc.Proposal().Index()) if err != nil { return fmt.Errorf("epoch lookup: %w", err) } - if err := qc.Verify(ep); err != nil { - return fmt.Errorf("verify: %w", err) - } - return nil + return qc.Verify(ep) } // advanceEpoch installs nextDuo at a boundary. Sole post-construction writer of @@ -176,17 +180,21 @@ func verifyCommitQCInDuo(duo types.EpochDuo, qc *types.CommitQC) error { // after Current and that seal leashes (waitForAppQC, registry WaitForDuo) are // already satisfied. Adds Current lanes; does not delete old lanes // (TODO(lane-expiry)). Touches epoch + lane votes (reweight). -func (i *inner) advanceEpoch(nextDuo types.EpochDuo) { - current := nextDuo.Current - for lane := range current.Committee().Lanes().All() { +func (i *inner) advanceEpoch(epoch *types.Epoch) bool { + if i.epoch.Load().EpochIndex() >= epoch.EpochIndex() { + return false + } + c := epoch.Committee() + for lane := range c.Lanes().All() { i.lanes.getOrInsert(lane) } for _, ls := range i.lanes.byID { for n := ls.votes.first; n < ls.votes.next; n++ { - ls.votes.q[n].reweight(current) + ls.votes.q[n].reweight(c) } } - i.epoch.Store(nextDuo) + i.epoch.Store(epoch) + return true } // pushPruneAnchor advances queue boundaries for an AppQC and its matching @@ -200,10 +208,10 @@ func (i *inner) pushPruneAnchor(anchor *PruneAnchor) (bool, error) { if idx != commitQC.Proposal().Index() { return false, fmt.Errorf("mismatched QCs: appQC index %v, commitQC index %v", idx, commitQC.Proposal().Index()) } - if idx < types.NextOpt(i.app.latestAppQC) { + if idx < types.NextOpt(i.app.anchor) { return false, nil } - i.app.latestAppQC = utils.Some(appQC) + i.app.anchor = utils.Some(anchor) metrics.ObserveAppQC(appQC) i.commits.qcs.prune(idx) i.commits.push(commitQC) @@ -214,5 +222,8 @@ func (i *inner) pushPruneAnchor(anchor *PruneAnchor) (bool, error) { ls.blocks.prune(lr.First()) ls.durable.floorNext(lr.First()) } + if anchor.Epoch.EpochIndex() > i.epoch.Load().EpochIndex() { + i.advanceEpoch(anchor.Epoch) + } return true, nil } diff --git a/sei-tendermint/internal/autobahn/avail/lane.go b/sei-tendermint/internal/autobahn/avail/lane.go index 8dbc654c65..59dba7cd0b 100644 --- a/sei-tendermint/internal/autobahn/avail/lane.go +++ b/sei-tendermint/internal/autobahn/avail/lane.go @@ -50,8 +50,7 @@ func (s *State) PushBlock(ctx context.Context, p *types.Signed[*types.LanePropos // Snapshot Current once for off-lock verify. Unlike PushVote (which parks // until Current accepts the signer), we do not wait for future committees — // lane proposals are not reweighted across epoch advances. - duo := s.epochDuo.Load() - c := duo.Current.Committee() + c := s.epoch.Load().Committee() if err := p.Msg().Verify(c); err != nil { return fmt.Errorf("block.Verify(): %w", err) } @@ -109,23 +108,16 @@ func (s *State) PushBlock(ctx context.Context, p *types.Signed[*types.LanePropos func (s *State) PushVote(ctx context.Context, vote *types.Signed[*types.LaneVote]) error { h := vote.Msg().Header() // Future-epoch voters park (one stream goroutine) until Current includes them. - var committee *types.Committee - var verifiedEpoch types.EpochIndex - for inner, ctrl := range s.inner.Lock() { - if err := ctrl.WaitUntil(ctx, func() bool { - c := inner.epoch.Load().Current.Committee() - return c.Weight(vote.Key()) > 0 && c.HasLane(h.Lane()) - }); err != nil { - return err - } - duo := inner.epoch.Load() - committee = duo.Current.Committee() - verifiedEpoch = duo.Current.EpochIndex() - } - if err := vote.Msg().Verify(committee); err != nil { + epoch,err := s.epoch.Wait(ctx, func(epoch *types.Epoch) bool { + // TODO: this is not a reliable criterion: fix once we have proper line lifecycle management + c := s.epoch.Load().Committee() + return c.Weight(vote.Key()) > 0 && c.HasLane(h.Lane()) + }) + if err!=nil { return err } + if err := vote.Msg().Verify(epoch.Committee()); err != nil { return fmt.Errorf("vote.Verify(): %w", err) } - if err := vote.VerifySig(committee); err != nil { + if err := vote.VerifySig(epoch.Committee()); err != nil { return fmt.Errorf("vote.Verify(): %w", err) } for inner, ctrl := range s.inner.Lock() { @@ -139,11 +131,9 @@ func (s *State) PushVote(ctx context.Context, vote *types.Signed[*types.LaneVote }); err != nil { return err } - // WaitUntil may release the lock; re-check membership under live Current. - live := inner.epoch.Load() - if live.Current.EpochIndex() != verifiedEpoch && - (live.Current.Committee().Weight(vote.Key()) == 0 || - !live.Current.Committee().HasLane(h.Lane())) { + // Check if the lane is still live + epoch := inner.epoch.Load() + if (epoch.Committee().Weight(vote.Key()) == 0 || !epoch.Committee().HasLane(h.Lane())) { return nil } if h.BlockNumber() < q.first { @@ -152,7 +142,7 @@ func (s *State) PushVote(ctx context.Context, vote *types.Signed[*types.LaneVote for q.next <= h.BlockNumber() { q.pushBack(newBlockVotes()) } - if q.q[h.BlockNumber()].pushVote(live.Current, vote).IsPresent() { + if q.q[h.BlockNumber()].pushVote(epoch, vote).IsPresent() { ctrl.Updated() } } @@ -223,8 +213,8 @@ func (s *State) WaitForLaneQCs( for inner, ctrl := range s.inner.Lock() { laneQCs := map[types.LaneID]*types.LaneQC{} for { - ep := inner.epoch.Load().Current - for lane := range ep.Committee().Lanes().All() { + epoch := inner.epoch.Load() + for lane := range epoch.Committee().Lanes().All() { first := types.LaneRangeOpt(prev, lane).Next() for i := range types.BlockNumber(types.MaxLaneRangeInProposal) { if qc, ok := inner.lanes.laneQC(lane, first+i).Get(); ok { @@ -235,7 +225,7 @@ func (s *State) WaitForLaneQCs( } } if len(laneQCs) > 0 { - return laneQCs, ep, nil + return laneQCs, epoch, nil } if err := ctrl.Wait(ctx); err != nil { return nil, nil, err diff --git a/sei-tendermint/internal/autobahn/avail/persistence.go b/sei-tendermint/internal/autobahn/avail/persistence.go index f3d49aee7a..3ad48e842f 100644 --- a/sei-tendermint/internal/autobahn/avail/persistence.go +++ b/sei-tendermint/internal/autobahn/avail/persistence.go @@ -30,8 +30,11 @@ const innerFile = "avail_inner" type PruneAnchor struct { AppQC *types.AppQC CommitQC *types.CommitQC + Epoch *types.Epoch } +func (a *PruneAnchor) Next() types.RoadIndex { return a.AppQC.Next() } + // PruneAnchorConv converts between PruneAnchor and its protobuf representation. var PruneAnchorConv = protoutils.Conv[*PruneAnchor, *pb.PersistedAvailPruneAnchor]{ Encode: func(a *PruneAnchor) *pb.PersistedAvailPruneAnchor { @@ -129,9 +132,10 @@ func loadPersistedState(dir utils.Option[string]) (utils.Option[*loadedAvailStat // TODO: use a single WAL for anchor and CommitQCs to make // this atomic rather than relying on write order. func (s *State) runPersist(ctx context.Context, pers persisters) error { - var persistedAnchorNext types.RoadIndex + // TODO(gprusak): persistedRange should be initialized from the persister itself. + var persistedRange types.RoadRange for { - batch, err := s.collectPersistBatch(ctx, persistedAnchorNext) + batch, err := s.collectPersistBatch(ctx, persistedRange) if err != nil { return err } @@ -145,7 +149,7 @@ func (s *State) runPersist(ctx context.Context, pers persisters) error { return fmt.Errorf("persist prune anchor: %w", err) } s.advancePersistedBlockStart(anchor.CommitQC) - persistedAnchorNext = anchor.CommitQC.Proposal().Index() + 1 + persistedRange.First = anchor.CommitQC.Proposal().Index() + 1 anchorQC = utils.Some(anchor.CommitQC) } @@ -162,11 +166,22 @@ func (s *State) runPersist(ctx context.Context, pers persisters) error { // 2. Persist commit-QCs and per-lane blocks in parallel. // Callees handle empty inputs gracefully (no-op when nothing to write/truncate). - if err := scope.Parallel(func(ps scope.ParallelScope) error { - ps.Spawn(func() error { - return pers.commitQCs.MaybePruneAndPersist(anchorQC, batch.commitQCs, utils.Some(func(qc *types.CommitQC) { - s.markCommitQCsPersisted(qc) - })) + if err := scope.Run(ctx, func(ctx context.Context, scope scope.Scope) error { + scope.Spawn(func() error { + if err:=pers.commitQCs.PruneAndPersist(anchorQC, batch.commitQCs); err!=nil { return err } + if n := len(batch.commitQCs); n>0 { + qc := batch.commitQCs[n-1] + persistedRange.Next = qc.Index()+1 + // Bump the consensus spec so that validator can start participating in + // the next consensus instance. + // We bump it once per batch, since all previous instances have already finished. + spec,err:=s.data.Registry().WaitForConsensusSpec(ctx, utils.Some(qc)) + if err!=nil { return fmt.Errorf("WaitForConsensusSpec(): %w",err) } + for inner := range s.inner.Lock() { + inner.commits.consensusSpec.Store(spec) + } + } + return nil }) // Collect lanes: any lane with blocks in this batch, plus all lanes // in the anchor epoch (for WAL pruning). @@ -182,8 +197,8 @@ func (s *State) runPersist(ctx context.Context, pers persisters) error { } for lane := range batchLanes { proposals := blocksByLane[lane] - ps.Spawn(func() error { - return pers.blocks.MaybePruneAndPersistLane(lane, anchorQC, proposals, utils.Some(markBlock)) + scope.Spawn(func() error { + return pers.blocks.PruneAndPersistLane(lane, anchorQC, proposals, utils.Some(markBlock)) }) } return nil @@ -230,26 +245,28 @@ func (s *State) markBlockPersisted(lane types.LaneID, next types.BlockNumber) { // markCommitQCsPersisted publishes the latest persisted CommitQC, // gating consensus from advancing until the QC is durable. -func (s *State) markCommitQCsPersisted(qc *types.CommitQC) { - for inner, ctrl := range s.inner.Lock() { - inner.commits.markPersisted(qc) - ctrl.Updated() +func (s *State) setConsensusSpec(ctx context.Context, qc *types.CommitQC) error { + spec,err := s.data.Registry().WaitForConsensusSpec(ctx,utils.Some(qc)) + if err!=nil { + if err==types.ErrPruned { return nil } } + for inner := range s.inner.Lock() { + if inner.commits.consensusSpec.Load().Index() <= spec.Index() { + inner.commits.consensusSpec.Store(spec) + } + } + return nil } // collectPersistBatch waits for new blocks or commitQCs and collects them under lock. -func (s *State) collectPersistBatch(ctx context.Context, persistedAnchorNext types.RoadIndex) (persistBatch, error) { +// persistedRange represents (anchor,commits.next) range. +// TODO(gprusak): this is inconsistent that SoT for persisted commit range is an input arg, +// while lane persistence status is internal to State. +func (s *State) collectPersistBatch(ctx context.Context, persistedRange types.RoadRange) (persistBatch, error) { var b persistBatch for inner, ctrl := range s.inner.Lock() { - // Derive the CommitQC persist cursor from persistedCommitQC. This is - // safe because persistedCommitQC is only advanced by markCommitQCsPersisted - // (after disk write) and on startup (from disk). Applying a prune anchor - // does not update persistedCommitQC, so this always reflects persistence - // state. The max clamp with commits.qcs.first handles an anchor - // fast-forwarding the queue past the cursor. - commitQCNext := types.NextIndexOpt(inner.commits.persistedCommitQC.Load()) if err := ctrl.WaitUntil(ctx, func() bool { - if types.NextOpt(inner.app.latestAppQC) != persistedAnchorNext { + if persistedRange.First < types.NextOpt(inner.app.anchor) || persistedRange.Next < inner.commits.qcs.next { return true } for _, ls := range inner.lanes.byID { @@ -257,7 +274,7 @@ func (s *State) collectPersistBatch(ctx context.Context, persistedAnchorNext typ return true } } - return commitQCNext < inner.commits.qcs.next + return false }); err != nil { return b, err } @@ -267,27 +284,16 @@ func (s *State) collectPersistBatch(ctx context.Context, persistedAnchorNext typ b.blocks = append(b.blocks, ls.blocks.q[n]) } } - commitQCNext = max(commitQCNext, inner.commits.qcs.first) - for n := commitQCNext; n < inner.commits.qcs.next; n++ { + for n := max(persistedRange.Next, inner.commits.qcs.first); n < inner.commits.qcs.next; n++ { b.commitQCs = append(b.commitQCs, inner.commits.qcs.q[n]) } - if types.NextOpt(inner.app.latestAppQC) != persistedAnchorNext { - if appQC, ok := inner.app.latestAppQC.Get(); ok { - idx := appQC.Proposal().RoadIndex() - if qc, ok := inner.commits.qcs.q[idx]; ok { - b.pruneAnchor = utils.Some(&PruneAnchor{ - AppQC: appQC, - CommitQC: qc, - }) - // Capture under the same lock as the anchor so an epoch slide - // cannot move its committee out of the live duo before I/O. - ep, err := inner.epoch.Load().EpochForRoad(qc.Proposal().Index()) - if err != nil { - return b, fmt.Errorf("EpochForRoad(%d): %w", qc.Proposal().Index(), err) - } - for lane := range ep.Committee().Lanes().All() { - b.pruneLanes = append(b.pruneLanes, lane) - } + if persistedRange.First < types.NextOpt(inner.app.anchor) { + if anchor, ok := inner.app.anchor.Get(); ok { + b.pruneAnchor = utils.Some(anchor) + // Capture under the same lock as the anchor so an epoch slide + // cannot move its committee out of the live duo before I/O. + for lane := range anchor.Epoch.Committee().Lanes().All() { + b.pruneLanes = append(b.pruneLanes, lane) } } } diff --git a/sei-tendermint/internal/autobahn/avail/state.go b/sei-tendermint/internal/autobahn/avail/state.go index ffdcbd04ba..551fd86424 100644 --- a/sei-tendermint/internal/autobahn/avail/state.go +++ b/sei-tendermint/internal/autobahn/avail/state.go @@ -28,7 +28,7 @@ type State struct { key types.SecretKey data *data.State inner utils.Watch[*inner] - epochDuo utils.AtomicRecv[types.EpochDuo] // Load-only view of inner.epoch + epoch utils.AtomicRecv[*types.Epoch] // Load-only view of inner.epoch // persisters groups all disk persistence components. // Always initialized: real when stateDir is set, no-op otherwise. @@ -65,7 +65,7 @@ func NewState(key types.SecretKey, data *data.State, stateDir utils.Option[strin key: key, data: data, inner: utils.NewWatch(inner), - epochDuo: inner.epoch.Subscribe(), + epoch: inner.epoch.Subscribe(), persisters: pers, }, nil } @@ -86,15 +86,13 @@ func (s *State) NextCommitQC() types.RoadIndex { } // Data returns the data state. -func (s *State) Data() *data.State { - return s.data -} +func (s *State) Data() *data.State { return s.data } // LastCommitQC returns receiver of the LastCommitQC. // The tip is the latest durably persisted CommitQC, not merely the admitted tip. -func (s *State) LastCommitQC() utils.AtomicRecv[utils.Option[*types.CommitQC]] { +func (s *State) ConsensusSpec() utils.AtomicRecv[*types.ConsensusSpec] { for inner := range s.inner.Lock() { - return inner.commits.persistedCommitQC.Subscribe() + return inner.commits.consensusSpec.Subscribe() } panic("unreachable") } @@ -108,12 +106,8 @@ func (s *State) LastCommitQC() utils.AtomicRecv[utils.Option[*types.CommitQC]] { // not spawn goroutines. func (s *State) Run(ctx context.Context) error { return scope.Run(ctx, func(ctx context.Context, scope scope.Scope) error { - scope.SpawnNamed("persist", func() error { - return s.runPersist(ctx, s.persisters) - }) - scope.SpawnNamed("advanceEpoch", func() error { - return s.runAdvanceEpoch(ctx) - }) + scope.SpawnNamed("persist", func() error { return s.runPersist(ctx, s.persisters) }) + scope.SpawnNamed("advanceEpoch", func() error { return s.runAdvanceEpoch(ctx) }) // Task inserting FullCommitQCs and local blocks to data state. // ErrPruned jumps n forward (AppQC/window prune during catch-up): skipped // roads need not be exported locally — peers can PushQC into data. diff --git a/sei-tendermint/internal/autobahn/avail/testonly.go b/sei-tendermint/internal/autobahn/avail/testonly.go index 1af8e8da1e..cf5e0c9c6b 100644 --- a/sei-tendermint/internal/autobahn/avail/testonly.go +++ b/sei-tendermint/internal/autobahn/avail/testonly.go @@ -77,13 +77,13 @@ func RunTestNetwork(ctx context.Context, states []*State) error { s.Spawn(func() error { next := types.RoadIndex(0) for { - appQC, commitQC, err := from.WaitForAppQC(ctx, next) + anchor, err := from.waitForAnchor(ctx, next) if err != nil { return err } - next = appQC.Next() + next = anchor.AppQC.Next() for _, to := range states { - if err := to.PushAppQC(ctx, appQC, commitQC); err != nil { + if err := to.PushAppQC(ctx, anchor.AppQC, anchor.CommitQC); err != nil { return err } } diff --git a/sei-tendermint/internal/autobahn/consensus/inner.go b/sei-tendermint/internal/autobahn/consensus/inner.go index b0a0e84a24..0b47a2726c 100644 --- a/sei-tendermint/internal/autobahn/consensus/inner.go +++ b/sei-tendermint/internal/autobahn/consensus/inner.go @@ -81,7 +81,6 @@ import ( "fmt" "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/internal/autobahn/pb" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" "github.com/sei-protocol/seilog" @@ -96,88 +95,54 @@ const innerFile = "inner" // pushCommitQC on State), and epoch transitions are explicit. // Genesis floors for ViewSpec come from State.registry (see State.viewSpec). type inner struct { + spec *types.ConsensusSpec persistedInner - epochs types.EpochDuo } -// View returns the current view, embedding the epoch's index. -// Genesis floors are unused here (View only needs CommitQC/TimeoutQC/Epochs). -func (i inner) View() types.View { - vs := types.ViewSpec{CommitQC: i.CommitQC, TimeoutQC: i.TimeoutQC, Epochs: i.epochs} - return vs.View() +// viewSpec builds a ViewSpec for i, taking genesis floors from the registry +// (used only when CommitQC is None). +func (i inner) ViewSpec() types.ViewSpec { + return types.ViewSpec{ + ConsensusSpec: i.spec, + TimeoutQC: i.TimeoutQC, + } } +// View returns the current view. +func (i inner) View() types.View { return i.ViewSpec().View() } + // newInner creates the inner state from persisted data loaded by NewPersister. // data is None on fresh start (persistence disabled or no prior state). // Returns error if persisted state is corrupt (see persistedInner.validate) or // required epochs are missing from the registry. -func newInner(data utils.Option[*pb.PersistedInner], registry *epoch.Registry) (inner, error) { - var persisted persistedInner +func newInner(data utils.Option[*pb.PersistedInner], spec *types.ConsensusSpec) (inner, error) { + persisted := &persistedInner{} if p, ok := data.Get(); ok { - decoded, err := innerProtoConv.Decode(p) - if err != nil { + var err error + if persisted, err = innerProtoConv.Decode(p); err != nil { return inner{}, fmt.Errorf("corrupt persisted state: %w", err) } - persisted = *decoded + logger.Info("restored consensus state", "state", innerProtoConv.Encode(persisted)) } - - // View duo = tipcut; CommitQC may be prior epoch. Seeding is data's; - // missing epoch hard-fails. Tip order: NewState requires avail ≥ consensus; - // avail/consensus may lag data and catch up in Run. - nextViewRoad := types.NextIndexOpt(persisted.CommitQC) - duo, err := registry.DuoAt(nextViewRoad) - if err != nil { - return inner{}, fmt.Errorf("DuoAt(%d): %w", nextViewRoad, err) + if specIdx := types.NextIndexOpt(spec.CommitQC); specIdx persisted.Index { + persisted = &persistedInner{Index:specIdx} } - commitEpoch := duo.Current - if cqc, ok := persisted.CommitQC.Get(); ok { - commitEpoch, err = registry.EpochAt(cqc.Proposal().Index()) - if err != nil { - return inner{}, fmt.Errorf("EpochAt(%d): %w", cqc.Proposal().Index(), err) - } - } - if err := persisted.validate(commitEpoch, duo); err != nil { + if err := persisted.Verify(spec); err != nil { return inner{}, err } - - logger.Info("restored consensus state", "state", innerProtoConv.Encode(&persisted)) - - return inner{persistedInner: persisted, epochs: duo}, nil + return inner{spec: spec, persistedInner: *persisted}, nil } -func (s *State) pushCommitQC(qc *types.CommitQC) error { - if qc.Proposal().Index() < s.innerRecv.Load().View().Index { - return nil - } - // Re-verify. Epoch must be seeded; missing → hard error (no WaitForDuo). - ep, err := s.registry.EpochAt(qc.Proposal().Index()) - if err != nil { - return fmt.Errorf("EpochAt(%d): %w", qc.Proposal().Index(), err) - } - if err := qc.Verify(ep); err != nil { - return fmt.Errorf("qc.Verify(): %w", err) - } +func (s *State) pushSpec(spec *types.ConsensusSpec) { for iSend := range s.inner.Lock() { i := iSend.Load() - if qc.Proposal().Index() < i.View().Index { - return nil + if newIndex := types.NextIndexOpt(spec.CommitQC); newIndex > i.Index { + iSend.Store(inner{spec: spec, persistedInner: persistedInner{Index: newIndex}}) } - nextRoad := qc.Proposal().Index() + 1 - nextDuo := i.epochs - if !i.epochs.Current.RoadRange().Has(nextRoad) { - // Tipcut past Current: DuoAt must already be seeded. - duo, err := s.registry.DuoAt(nextRoad) - if err != nil { - logger.Error("tipcut duo not in registry after avail CommitQC tip", - "road", nextRoad) - return fmt.Errorf("DuoAt(%d): %w", nextRoad, err) - } - nextDuo = duo - } - iSend.Store(inner{persistedInner: persistedInner{CommitQC: utils.Some(qc)}, epochs: nextDuo}) } - return nil } func (s *State) waitForView(ctx context.Context, view types.View) (types.ViewSpec, error) { @@ -193,7 +158,7 @@ func (s *State) pushTimeoutQC(ctx context.Context, qc *types.TimeoutQC) error { return nil } // Verify checks the invariant: TimeoutQC.View().Index == CommitQC.Index + 1 - if err := qc.Verify(i.epochs.Current, i.CommitQC); err != nil { + if err := qc.Verify(i.spec.Epoch(), i.spec.CommitQC); err != nil { return fmt.Errorf("qc.Verify(): %w", err) } for isend := range s.inner.Lock() { @@ -202,7 +167,7 @@ func (s *State) pushTimeoutQC(ctx context.Context, qc *types.TimeoutQC) error { return nil } // TimeoutQC advances view number; clear votes and prepareQC. Epochs unchanged. - isend.Store(inner{persistedInner: persistedInner{CommitQC: i.CommitQC, TimeoutQC: utils.Some(qc)}, epochs: i.epochs}) + isend.Store(inner{spec: i.spec, persistedInner: persistedInner{Index: i.Index, TimeoutQC: utils.Some(qc)}}) } return nil } diff --git a/sei-tendermint/internal/autobahn/consensus/persist/blocks.go b/sei-tendermint/internal/autobahn/consensus/persist/blocks.go index 585eb1d74b..ecc1648af1 100644 --- a/sei-tendermint/internal/autobahn/consensus/persist/blocks.go +++ b/sei-tendermint/internal/autobahn/consensus/persist/blocks.go @@ -294,7 +294,7 @@ func (bp *BlockPersister) getOrCreateLane(lane types.LaneID) (*laneWAL, error) { // // The per-lane lock is held for the entire truncate-then-append sequence, // so concurrent calls on the same lane serialize correctly. -func (bp *BlockPersister) MaybePruneAndPersistLane( +func (bp *BlockPersister) PruneAndPersistLane( lane types.LaneID, anchor utils.Option[*types.CommitQC], proposals []*types.Signed[*types.LaneProposal], diff --git a/sei-tendermint/internal/autobahn/consensus/persist/commitqcs.go b/sei-tendermint/internal/autobahn/consensus/persist/commitqcs.go index d0291410b7..553cc26c59 100644 --- a/sei-tendermint/internal/autobahn/consensus/persist/commitqcs.go +++ b/sei-tendermint/internal/autobahn/consensus/persist/commitqcs.go @@ -136,10 +136,9 @@ func (cp *CommitQCPersister) LoadNext() types.RoadIndex { // need not coordinate ordering. // afterEach, when present, is called after each successful append. It is // invoked while the lock is held, so it must not re-enter the persister. -func (cp *CommitQCPersister) MaybePruneAndPersist( +func (cp *CommitQCPersister) PruneAndPersist( anchor utils.Option[*types.CommitQC], commitQCs []*types.CommitQC, - afterEach utils.Option[func(*types.CommitQC)], ) error { for s := range cp.state.Lock() { if qc, ok := anchor.Get(); ok { @@ -147,14 +146,10 @@ func (cp *CommitQCPersister) MaybePruneAndPersist( return err } } - fn, hasFn := afterEach.Get() for _, c := range commitQCs { if err := s.persistCommitQC(c); err != nil { return err } - if hasFn { - fn(c) - } } return nil } diff --git a/sei-tendermint/internal/autobahn/consensus/persisted_inner.go b/sei-tendermint/internal/autobahn/consensus/persisted_inner.go index fc30274a8d..541eb8296d 100644 --- a/sei-tendermint/internal/autobahn/consensus/persisted_inner.go +++ b/sei-tendermint/internal/autobahn/consensus/persisted_inner.go @@ -60,7 +60,7 @@ import ( // the persisted viewSpec. TODO: consider rebroadcasting CommitQC on restart // to help peers sync faster after cluster-wide outages. type persistedInner struct { - CommitQC utils.Option[*types.CommitQC] + Index types.RoadIndex PrepareQC utils.Option[*types.PrepareQC] TimeoutQC utils.Option[*types.TimeoutQC] @@ -74,55 +74,41 @@ type persistedInner struct { // // commitEp verifies the persisted CommitQC. viewDuo is DuoAt(tipcut): Current // stamps/verifies the open view; at a boundary commitEp and Current differ. -func (p *persistedInner) validate(commitEp *types.Epoch, viewDuo types.EpochDuo) error { - viewEp := viewDuo.Current - if cqc, ok := p.CommitQC.Get(); ok { - if err := cqc.Verify(commitEp); err != nil { - return fmt.Errorf("corrupt persisted state: CommitQC failed verification: %w", err) - } +func (p *persistedInner) Verify(spec *types.ConsensusSpec) error { + if got,want := p.Index,types.NextIndexOpt(spec.CommitQC); got!=want { + return fmt.Errorf("p.Index = %v, want %v",got,want) } - // TimeoutQC index must equal NextIndexOpt(CommitQC) (i.e., CommitQC.Index+1, or 0 if missing). // Since we persist the entire inner state atomically, a mismatched index is always corrupt. if tqc, ok := p.TimeoutQC.Get(); ok { - tqcIndex := tqc.View().Index - expectedIndex := types.NextIndexOpt(p.CommitQC) - if tqcIndex != expectedIndex { - return fmt.Errorf("corrupt persisted state: TimeoutQC has index %d but expected %d", tqcIndex, expectedIndex) - } - if err := tqc.Verify(viewEp, p.CommitQC); err != nil { - return fmt.Errorf("corrupt persisted state: TimeoutQC failed verification: %w", err) + if err := tqc.Verify(spec.Epochs.Current, spec.CommitQC); err != nil { + return fmt.Errorf("p.TimeoutQC.Verify(): %w", err) } } - vs := types.ViewSpec{ - CommitQC: p.CommitQC, - TimeoutQC: p.TimeoutQC, - Epochs: viewDuo, - GenesisFirstBlock: 0, // validate does not use NextGlobalBlock; floors unused here - } + vs := types.ViewSpec{ConsensusSpec: spec, TimeoutQC: p.TimeoutQC} currentView := vs.View() - committee := viewEp.Committee() + committee := vs.Epoch().Committee() // checkViewAndSig validates that a persisted field has the current view and a valid signature. // Since inner is persisted atomically, any view mismatch indicates corrupt state. checkViewAndSig := func(name string, view types.View, verifyErr error) error { if view != currentView { - return fmt.Errorf("corrupt persisted state: %s has view %v but current view is %v", name, view, currentView) + return fmt.Errorf("%s has view %v but current view is %v", name, view, currentView) } if verifyErr != nil { - return fmt.Errorf("corrupt persisted state: %s failed verification: %w", name, verifyErr) + return fmt.Errorf("%s failed verification: %w", name, verifyErr) } return nil } // PrepareQC is required when CommitVote is present (CommitVote requires PrepareQC justification). if pqc, ok := p.PrepareQC.Get(); ok { - if err := checkViewAndSig("PrepareQC", pqc.Proposal().View(), pqc.Verify(viewEp)); err != nil { + if err := checkViewAndSig("PrepareQC", pqc.Proposal().View(), pqc.Verify(vs.Epoch())); err != nil { return err } } else if p.CommitVote.IsPresent() { - return fmt.Errorf("corrupt persisted state: CommitVote present without PrepareQC") + return fmt.Errorf("CommitVote present without PrepareQC") } if v, ok := p.CommitVote.Get(); ok { if err := checkViewAndSig("CommitVote", v.Msg().Proposal().View(), v.VerifySig(committee)); err != nil { @@ -135,7 +121,7 @@ func (p *persistedInner) validate(commitEp *types.Epoch, viewDuo types.EpochDuo) } } if v, ok := p.TimeoutVote.Get(); ok { - if err := checkViewAndSig("TimeoutVote", v.View(), v.Verify(viewEp)); err != nil { + if err := checkViewAndSig("TimeoutVote", v.View(), v.Verify(vs.Epoch())); err != nil { return err } } diff --git a/sei-tendermint/internal/autobahn/consensus/state.go b/sei-tendermint/internal/autobahn/consensus/state.go index d207d7cd75..d419b6759f 100644 --- a/sei-tendermint/internal/autobahn/consensus/state.go +++ b/sei-tendermint/internal/autobahn/consensus/state.go @@ -104,16 +104,14 @@ func newState( pers utils.Option[persist.Persister[*pb.PersistedInner]], persistedData utils.Option[*pb.PersistedInner], ) (*State, error) { - initialInner, err := newInner(persistedData, data.Registry()) - if err != nil { - return nil, fmt.Errorf("newInner: %w", err) - } - availState, err := avail.NewState(cfg.Key, data, cfg.PersistentStateDir) if err != nil { return nil, fmt.Errorf("avail.NewState: %w", err) } - + initialInner, err := newInner(persistedData, availState.ConsensusSpec().Load()) + if err != nil { + return nil, fmt.Errorf("newInner: %w", err) + } innerSend := utils.Alloc(utils.NewAtomicSend(initialInner)) registry := data.Registry() s := &State{ @@ -129,13 +127,7 @@ func newState( prepareVotes: utils.NewMutex(newPrepareVotes()), commitVotes: utils.NewMutex(newCommitVotes()), - myView: utils.NewAtomicSend(types.ViewSpec{ - CommitQC: initialInner.CommitQC, - TimeoutQC: initialInner.TimeoutQC, - Epochs: initialInner.epochs, - GenesisFirstBlock: registry.FirstBlock(), - GenesisTimestamp: registry.FirstTimestamp(), - }), + myView: utils.NewAtomicSend(initialInner.ViewSpec()), myProposal: utils.NewAtomicSend(utils.None[*types.FullProposal]()), myPrepareVote: utils.NewAtomicSend(utils.None[*types.ConsensusReqPrepareVote]()), myCommitVote: utils.NewAtomicSend(utils.None[*types.ConsensusReqCommitVote]()), @@ -199,65 +191,48 @@ func (s *State) PushTimeoutQC(ctx context.Context, qc *types.TimeoutQC) error { return s.pushTimeoutQC(ctx, qc) } -// TODO: scope prepareVotes, commitVotes, and timeoutVotes to a single epoch -// so stale votes from a previous epoch are automatically dropped on transition. - // PushPrepareVote processes an unverified Prepare vote message. -func (s *State) PushPrepareVote(vote *types.Signed[*types.PrepareVote]) error { - // Contract: accept only Current-epoch votes (innerRecv). Others drop without - // error (avoid wrong-committee verify / peer teardown). No redelivery — - // around an epoch boundary peers may disagree for a window; recovery is via - // view timeout (typically one timeout round per transition). - i := s.innerRecv.Load() - if voteEp := vote.Msg().Proposal().View().EpochIndex; voteEp != i.epochs.Current.EpochIndex() { - logger.Debug("dropping prepare vote for non-current epoch", - "vote_epoch", uint64(voteEp), "current_epoch", uint64(i.epochs.Current.EpochIndex())) - return nil - } - committee := i.epochs.Current.Committee() - if err := vote.VerifySig(committee); err != nil { +func (s *State) PushPrepareVote(ctx context.Context, vote *types.Signed[*types.PrepareVote]) error { + epochIdx := vote.Msg().Proposal().View().EpochIndex + vs,err := s.myView.Wait(ctx, func(vs types.ViewSpec) bool { return vs.Epoch().EpochIndex() >= epochIdx }) + if err!=nil { return err } + if vs.View().EpochIndex != epochIdx { return nil } + if err := vote.VerifySig(vs.Epoch().Committee()); err != nil { return fmt.Errorf("vote.VerifySig(): %w", err) } for pv := range s.prepareVotes.Lock() { - pv.pushVote(committee, vote) + // TODO: prune old epoch votes at some point. + pv.pushVote(vs.Epoch().Committee(), vote) } return nil } // PushCommitVote processes an unverified CommitVote message. -func (s *State) PushCommitVote(vote *types.Signed[*types.CommitVote]) error { - // Same Current-epoch contract as PushPrepareVote. - i := s.innerRecv.Load() - if voteEp := vote.Msg().Proposal().View().EpochIndex; voteEp != i.epochs.Current.EpochIndex() { - logger.Debug("dropping commit vote for non-current epoch", - "vote_epoch", uint64(voteEp), "current_epoch", uint64(i.epochs.Current.EpochIndex())) - return nil - } - committee := i.epochs.Current.Committee() - if err := vote.VerifySig(committee); err != nil { +func (s *State) PushCommitVote(ctx context.Context, vote *types.Signed[*types.CommitVote]) error { + epochIdx := vote.Msg().Proposal().View().EpochIndex + vs,err := s.myView.Wait(ctx, func(vs types.ViewSpec) bool { return vs.Epoch().EpochIndex() >= epochIdx }) + if err!=nil { return err } + if vs.View().EpochIndex != epochIdx { return nil } + if err := vote.VerifySig(vs.Epoch().Committee()); err != nil { return fmt.Errorf("vote.VerifySig(): %w", err) } for cv := range s.commitVotes.Lock() { - cv.pushVote(committee, vote) + cv.pushVote(vs.Epoch().Committee(), vote) } return nil } // PushTimeoutVote processes an unverified FullTimeoutVote message. -func (s *State) PushTimeoutVote(vote *types.FullTimeoutVote) error { - // Same Current-epoch contract as PushPrepareVote. - i := s.innerRecv.Load() - if voteEp := vote.View().EpochIndex; voteEp != i.epochs.Current.EpochIndex() { - logger.Debug("dropping timeout vote for non-current epoch", - "vote_epoch", uint64(voteEp), "current_epoch", uint64(i.epochs.Current.EpochIndex())) - return nil - } - ep := i.epochs.Current - if err := vote.Verify(ep); err != nil { - return fmt.Errorf("vote.Verify(): %w", err) +func (s *State) PushTimeoutVote(ctx context.Context, vote *types.FullTimeoutVote) error { + epochIdx := vote.View().EpochIndex + vs,err := s.myView.Wait(ctx, func(vs types.ViewSpec) bool { return vs.Epoch().EpochIndex() >= epochIdx }) + if err!=nil { return err } + if vs.View().EpochIndex != epochIdx { return nil } + if err := vote.Verify(vs.Epoch()); err != nil { + return fmt.Errorf("vote.VerifySig(): %w", err) } for tv := range s.timeoutVotes.Lock() { - tv.pushVote(ep.Committee(), vote) + tv.pushVote(vs.Epoch().Committee(), vote) } return nil } @@ -320,18 +295,6 @@ func updateOutput[T types.ConsensusReq](w *utils.AtomicSend[utils.Option[T]], v } } -// viewSpec builds a ViewSpec for i, taking genesis floors from the registry -// (used only when CommitQC is None). -func (s *State) viewSpec(i inner) types.ViewSpec { - return types.ViewSpec{ - CommitQC: i.CommitQC, - TimeoutQC: i.TimeoutQC, - Epochs: i.epochs, - GenesisFirstBlock: s.registry.FirstBlock(), - GenesisTimestamp: s.registry.FirstTimestamp(), - } -} - // Updates the outputs based on the inner state. // Persists state to disk before broadcasting votes to ensure votes are durable // before dissemination (prevents double-voting on crash). @@ -339,7 +302,7 @@ func (s *State) viewSpec(i inner) types.ViewSpec { // timers, neither of which constitutes a vote. func (s *State) runOutputs(ctx context.Context) error { return s.innerRecv.Iter(ctx, func(ctx context.Context, i inner) error { - vs := s.viewSpec(i) + vs := i.ViewSpec() old := s.myView.Load() if old.View().Less(vs.View()) { s.myView.Store(vs) @@ -383,14 +346,12 @@ func (s *State) Run(ctx context.Context) error { return nil }) }) - scope.SpawnNamed("pushCommitQC", func() error { + scope.SpawnNamed("pushSpec", func() error { // Pull the CommitQC tip back from avail after it has been logged and // verified at admit. Tip watch may coalesce; pushCommitQC re-verifies // against the QC's epoch and aligns the duo without replaying roads. - return s.avail.LastCommitQC().Iter(ctx, func(ctx context.Context, last utils.Option[*types.CommitQC]) error { - if qc, ok := last.Get(); ok { - return s.pushCommitQC(qc) - } + return s.avail.ConsensusSpec().Iter(ctx, func(ctx context.Context, spec *types.ConsensusSpec) error { + s.pushSpec(spec) return nil }) }) diff --git a/sei-tendermint/internal/autobahn/consensus/testonly.go b/sei-tendermint/internal/autobahn/consensus/testonly.go index 98cfa683d0..1a1abd66d6 100644 --- a/sei-tendermint/internal/autobahn/consensus/testonly.go +++ b/sei-tendermint/internal/autobahn/consensus/testonly.go @@ -43,7 +43,7 @@ func RunTestNetwork(ctx context.Context, states []*State) error { return from.SubscribePrepareVote().Iter(ctx, func(_ context.Context, msg utils.Option[*types.ConsensusReqPrepareVote]) error { if vote, ok := msg.Get(); ok { for _, to := range states { - if err := to.PushPrepareVote(vote.Signed); err != nil { + if err := to.PushPrepareVote(ctx, vote.Signed); err != nil { return err } } @@ -58,7 +58,7 @@ func RunTestNetwork(ctx context.Context, states []*State) error { return nil } for _, to := range states { - if err := to.PushCommitVote(vote.Signed); err != nil { + if err := to.PushCommitVote(ctx, vote.Signed); err != nil { return err } } @@ -69,7 +69,7 @@ func RunTestNetwork(ctx context.Context, states []*State) error { return from.SubscribeTimeoutVote().Iter(ctx, func(_ context.Context, msg utils.Option[*types.FullTimeoutVote]) error { if vote, ok := msg.Get(); ok { for _, to := range states { - if err := to.PushTimeoutVote(vote); err != nil { + if err := to.PushTimeoutVote(ctx, vote); err != nil { return err } } diff --git a/sei-tendermint/internal/autobahn/data/state.go b/sei-tendermint/internal/autobahn/data/state.go index 68b0d3ed29..c97de94fb1 100644 --- a/sei-tendermint/internal/autobahn/data/state.go +++ b/sei-tendermint/internal/autobahn/data/state.go @@ -220,9 +220,9 @@ func NewState(cfg *Config, blockDB types.BlockDB) (*State, error) { } initRoad = lastQC.QC().Proposal().Index() + 1 } - initDuo, err := cfg.Registry.DuoAt(initRoad) - if err != nil { - return nil, fmt.Errorf("init epochDuo: %w", err) + initDuo, ok := cfg.Registry.DuoAt(initRoad) + if !ok { + return nil, fmt.Errorf("missing epoch duo") } in.epochDuo = utils.NewAtomicSend(initDuo) s.epochDuo = in.epochDuo.Subscribe() @@ -342,9 +342,9 @@ func (s *State) loadFromBlockDB(blockDB types.BlockDB) error { // gr.First == n it still fires for the covering QC when the scan opened // inside that QC's range. insertQC clips it to [nextQC, gr.Next). lastQC = qc - ep, err := s.cfg.Registry.EpochAt(qc.QC().Proposal().Index()) - if err != nil { - return fmt.Errorf("load QC from BlockDB: epoch lookup: %w", err) + ep, ok := s.cfg.Registry.EpochAt(qc.QC().Proposal().Index()) + if !ok { + return fmt.Errorf("load QC from BlockDB: missing epoch") } if err := in.insertQC(qc, ep); err != nil { return fmt.Errorf("load QC from BlockDB: %w", err) @@ -361,9 +361,9 @@ func (s *State) loadFromBlockDB(blockDB types.BlockDB) error { } blk := blkOpt.OrPanic(fmt.Sprintf("block %d absent at a HasBlock position", n)) storedQC := in.qcs[n] - e, err := s.cfg.Registry.EpochAt(storedQC.QC().Proposal().Index()) - if err != nil { - return fmt.Errorf("load block %d from BlockDB: epoch lookup: %w", n, err) + e, ok := s.cfg.Registry.EpochAt(storedQC.QC().Proposal().Index()) + if !ok { + return fmt.Errorf("load block %d from BlockDB: epoch lookup: missing epoch", n) } if err := blk.Verify(e.Committee()); err != nil { return fmt.Errorf("verify block %d from BlockDB: %w", n, err) @@ -449,65 +449,47 @@ func (s *State) insertBlocksByHash(inner *inner, gr types.GlobalRange, byHash ma // horizon; do not soft-admit them via Registry. func (s *State) PushQC(ctx context.Context, qc *types.FullCommitQC, blocks []*types.Block) error { gr := qc.QC().GlobalRange() - needQC, err := func() (bool, error) { + needQC, duo, err := func() (bool, types.EpochDuo, error) { for inner, ctrl := range s.inner.Lock() { if err := ctrl.WaitUntil(ctx, func() bool { return gr.First <= inner.nextQC && gr.First < inner.nextAppProposal+blocksCacheSize }); err != nil { - return false, err + return false, types.EpochDuo{}, err } - return inner.nextQC == gr.First, nil + return inner.nextQC == gr.First, inner.epochDuo.Load(), nil } panic("unreachable") }() if err != nil { return err } - idx := qc.QC().Proposal().Index() - duo := s.epochDuo.Load() - ep, err := duo.EpochForRoad(idx) - if err != nil { - if !needQC && errors.Is(err, types.ErrRoadBeforeWindow) { - return nil - } - return err + ei := qc.QC().Proposal().EpochIndex() + if duo.Current.EpochIndex()!=ei { + if duo, err = s.cfg.Registry.WaitForDuo(ctx, ei); err!=nil { return err } } // Verify data. if needQC { - if err := qc.Verify(ep); err != nil { + if err := qc.Verify(duo.Current); err != nil { return fmt.Errorf("qc.Verify(): %w", err) } } // Blocks share the QC's epoch (unlike PushBlock, which uses the stored QC). byHash := map[types.BlockHeaderHash]*types.Block{} - committee := ep.Committee() + committee := duo.Current.Committee() for _, b := range blocks { byHash[b.Header().Hash()] = b if err := b.Verify(committee); err != nil { return fmt.Errorf("b.Verify(): %w", err) } - } - // Closing Current: WaitForDuo(tipcut) before mutating nextQC. - nextDuo := utils.None[types.EpochDuo]() - if needQC && duo.Current.RoadRange().IsLastRoad(idx) { - nt, err := s.cfg.Registry.WaitForDuo(ctx, idx+1) - if err != nil { - return err - } - nextDuo = utils.Some(nt) - } + } for inner, ctrl := range s.inner.Lock() { if needQC { - // Only the first inserter may advance epochDuo. - applied := inner.nextQC == gr.First for inner.nextQC < gr.Next { inner.qcs[inner.nextQC] = qc inner.nextQC += 1 } - if applied { - if nd, ok := nextDuo.Get(); ok { - inner.epochDuo.Store(nd) - } + if inner.epochDuo.Load().Current.EpochIndex() < duo.Current.EpochIndex() { + inner.epochDuo.Store(duo) } ctrl.Updated() } @@ -560,7 +542,7 @@ func (s *State) PushBlock(ctx context.Context, n types.GlobalBlockNumber, block } // n in [nextBlock, nextQC): QC is contiguous in that range. var err error - ep, err = s.epochDuo.Load().EpochForRoad(inner.qcs[n].QC().Proposal().Index()) + ep, err = s.epochDuo.Load().ByRoad(inner.qcs[n].QC().Proposal().Index()) if err != nil { return fmt.Errorf("epoch not in window: %w", err) } diff --git a/sei-tendermint/internal/autobahn/epoch/registry.go b/sei-tendermint/internal/autobahn/epoch/registry.go index 42f0aaf475..756e22fc08 100644 --- a/sei-tendermint/internal/autobahn/epoch/registry.go +++ b/sei-tendermint/internal/autobahn/epoch/registry.go @@ -55,10 +55,7 @@ type registryState = map[types.EpochIndex]*types.Epoch // // TODO(autobahn): replace genesis placeholders with epoch info on blocks. type Registry struct { - state utils.RWMutex[registryState] - // epochGen bumps on every new registration; WaitForDuo waits on it so - // filling a gap still wakes waiters. - epochGen utils.AtomicSend[uint64] + state utils.Watch[registryState] // Genesis floors from GenDoc (InitialHeight / GenesisTime). genesisFirstBlock types.GlobalBlockNumber genesisTimestamp time.Time @@ -74,8 +71,7 @@ func NewRegistry( ) (*Registry, error) { ep := types.NewEpoch(0, types.RoadRange{First: 0, Next: FirstRoad(1)}, committee) return &Registry{ - state: utils.NewRWMutex(registryState{0: ep}), - epochGen: utils.NewAtomicSend(uint64(0)), + state: utils.NewWatch(registryState{0: ep}), genesisFirstBlock: firstBlock, genesisTimestamp: genesisTimestamp, genesisCommittee: committee, @@ -101,12 +97,13 @@ func (r *Registry) SetupInitialDuo(commitQCs utils.Option[types.RoadRange]) erro // Avail WAL and BlockDB prune independently. Avail may restart at the // retained span's first epoch and still need its Prev. r.EnsureDuoAt(span.First) - for s := range r.state.Lock() { + for s,ctrl := range r.state.Lock() { for idx := windowFirst; idx <= windowLast; idx++ { if _, ok := s[idx]; ok { continue } r.makeEpoch(s, idx) + ctrl.Updated() } } r.EnsureDuoAt(span.Next) @@ -137,15 +134,13 @@ func (r *Registry) FirstTimestamp() time.Time { // EpochAt returns the epoch containing roadIndex. // Error if that epoch is not registered. -func (r *Registry) EpochAt(roadIndex types.RoadIndex) (*types.Epoch, error) { - epochIdx := IndexForRoad(roadIndex) - for s := range r.state.RLock() { - if ep, ok := s[epochIdx]; ok { - return ep, nil +func (r *Registry) EpochAt(roadIndex types.RoadIndex) (*types.Epoch, bool) { + for s := range r.state.Lock() { + if ep, ok := s[IndexForRoad(roadIndex)]; ok { + return ep, true } - return nil, fmt.Errorf("epoch %d (road %d) not registered", epochIdx, roadIndex) } - panic("unreachable") + return nil, false } // makeEpoch inserts a genesis-committee placeholder at epochIdx. @@ -157,20 +152,15 @@ func (r *Registry) makeEpoch(s registryState, epochIdx types.EpochIndex) *types. firstRoad := FirstRoad(epochIdx) epoch := types.NewEpoch(epochIdx, types.RoadRange{First: firstRoad, Next: FirstRoad(epochIdx + 1)}, r.genesisCommittee) s[epochIdx] = epoch - r.epochGen.Store(r.epochGen.Load() + 1) return epoch } // EnsureEpoch registers a genesis-committee placeholder for idx if missing. func (r *Registry) EnsureEpoch(idx types.EpochIndex) { - for s := range r.state.RLock() { - if _, ok := s[idx]; ok { - return - } - } - for s := range r.state.Lock() { + for s,ctrl := range r.state.Lock() { if _, ok := s[idx]; !ok { r.makeEpoch(s, idx) + ctrl.Updated() } } } @@ -203,37 +193,64 @@ func (r *Registry) AdvanceIfNeeded(roadIndex types.RoadIndex) { // DuoAt returns the EpochDuo centered on the epoch containing roadIndex. // Current must already be registered. Prev absent only for epoch 0; missing // Prev for center > 0 is a hard error (no soft-degrade to Current-only). -func (r *Registry) DuoAt(roadIndex types.RoadIndex) (types.EpochDuo, error) { - centerIdx := IndexForRoad(roadIndex) - current, err := r.EpochAt(FirstRoad(centerIdx)) - if err != nil { - return types.EpochDuo{}, fmt.Errorf("epoch %d (road %d) not in registry", centerIdx, roadIndex) - } +func (r *Registry) DuoAt(roadIndex types.RoadIndex) (types.EpochDuo, bool) { + current, ok := r.EpochAt(roadIndex) + if !ok { return types.EpochDuo{},false } prev := utils.None[*types.Epoch]() - if centerIdx > 0 { - p, err := r.EpochAt(FirstRoad(centerIdx - 1)) - if err != nil { - return types.EpochDuo{}, fmt.Errorf("epoch %d prev (road %d) not in registry", centerIdx-1, roadIndex) - } + if current.EpochIndex() > 0 { + p,_ := r.EpochAt(current.RoadRange().First-1) prev = utils.Some(p) } - return types.NewEpochDuo(current, prev), nil + return utils.OrPanic1(types.NewEpochDuo(current, prev)), true } // WaitForDuo blocks until DuoAt(roadIndex) succeeds. // Waits on epochGen (any registration), so filling Prev after Current is // already present still unblocks. Must not hold the avail/data inner lock // (execution may seed via AdvanceIfNeeded). -func (r *Registry) WaitForDuo(ctx context.Context, roadIndex types.RoadIndex) (types.EpochDuo, error) { - sub := r.epochGen.Subscribe() - for { - // Capture gen before DuoAt so a registration between check and Wait still wakes. - seen := sub.Load() - if duo, err := r.DuoAt(roadIndex); err == nil { - return duo, nil - } - if _, err := sub.Wait(ctx, func(gen uint64) bool { return gen > seen }); err != nil { - return types.EpochDuo{}, err +func (r *Registry) WaitForDuo(ctx context.Context, i types.EpochIndex) (types.EpochDuo, error) { + current,err := r.WaitForEpoch(ctx,i) + if err!=nil { return types.EpochDuo{},nil } + prev := utils.None[*types.Epoch]() + if i>0 { + p,err := r.WaitForEpoch(ctx,i-1) + if err!=nil { return types.EpochDuo{},nil } + prev = utils.Some(p) + } + return types.NewEpochDuo(current,prev) +} + +func (r *Registry) ConsensusSpec(prev utils.Option[*types.CommitQC]) (*types.ConsensusSpec, bool) { + duo,ok := r.DuoAt(types.NextIndexOpt(prev)) + if !ok { return nil,false } + return &types.ConsensusSpec { + Epochs: duo, + CommitQC: prev, + GenesisFirstBlock: r.genesisFirstBlock, + GenesisTimestamp: r.genesisTimestamp, + },true +} + +func (r *Registry) WaitForConsensusSpec(ctx context.Context, prev utils.Option[*types.CommitQC]) (*types.ConsensusSpec, error) { + duo,err := r.WaitForDuo(ctx,IndexForRoad(types.NextIndexOpt(prev))) + if err!=nil { return nil,err } + return &types.ConsensusSpec { + Epochs: duo, + CommitQC: prev, + GenesisFirstBlock: r.genesisFirstBlock, + GenesisTimestamp: r.genesisTimestamp, + },nil +} + +func (r *Registry) WaitForEpoch(ctx context.Context, i types.EpochIndex) (*types.Epoch, error) { + for inner,ctrl := range r.state.Lock() { + for { + if current, ok := inner[i]; ok { + return current,nil + } + if err:= ctrl.Wait(ctx); err!=nil { return nil,err } } } + panic("unreachable") } + diff --git a/sei-tendermint/internal/autobahn/epoch/testonly.go b/sei-tendermint/internal/autobahn/epoch/testonly.go index 4288dca6ed..7da94b33db 100644 --- a/sei-tendermint/internal/autobahn/epoch/testonly.go +++ b/sei-tendermint/internal/autobahn/epoch/testonly.go @@ -12,7 +12,7 @@ import ( // EpochAtTip (or EpochAt) so View.EpochIndex matches the road's epoch when // GenRegistry starts away from genesis. func (r *Registry) LatestEpoch() *types.Epoch { - for s := range r.state.RLock() { + for s := range r.state.Lock() { var best types.EpochIndex var ep *types.Epoch for idx, e := range s { @@ -32,7 +32,9 @@ func (r *Registry) LatestEpoch() *types.Epoch { // EpochAtTip is the epoch for the next CommitQC after prev (road 0 if none). // Intended for test CommitQC chains when GenRegistry's start epoch may be ≫ 0. func (r *Registry) EpochAtTip(prev utils.Option[*types.CommitQC]) *types.Epoch { - return utils.OrPanic1(r.EpochAt(types.NextIndexOpt(prev))) + epoch,ok := r.EpochAt(types.NextIndexOpt(prev)) + if !ok { panic("r.EpochAt() missing") } + return epoch } // GenRegistry generates a random Registry of the given committee size, starting diff --git a/sei-tendermint/internal/p2p/giga/consensus.go b/sei-tendermint/internal/p2p/giga/consensus.go index 6d5d2bace7..b700e5a9ca 100644 --- a/sei-tendermint/internal/p2p/giga/consensus.go +++ b/sei-tendermint/internal/p2p/giga/consensus.go @@ -114,15 +114,15 @@ func (x *Service) serverConsensus(ctx context.Context, server rpc.Server[API]) e } switch req := req.(type) { case *types.ConsensusReqPrepareVote: - if err := x.validatorState().PushPrepareVote(req.Signed); err != nil { + if err := x.validatorState().PushPrepareVote(ctx, req.Signed); err != nil { return fmt.Errorf("x.validatorState().PushPrepareVote(): %w", err) } case *types.ConsensusReqCommitVote: - if err := x.validatorState().PushCommitVote(req.Signed); err != nil { + if err := x.validatorState().PushCommitVote(ctx, req.Signed); err != nil { return fmt.Errorf("x.validatorState().PushCommitVote(): %w", err) } case *types.FullTimeoutVote: - if err := x.validatorState().PushTimeoutVote(req); err != nil { + if err := x.validatorState().PushTimeoutVote(ctx, req); err != nil { return fmt.Errorf("x.validatorState().PushTimeoutVote(): %w", err) } case *types.FullProposal: