From 643a31939915faca9f0c63b11ece0ab6ab940c01 Mon Sep 17 00:00:00 2001 From: Grzegorz Prusak Date: Fri, 31 Jul 2026 14:29:06 +0200 Subject: [PATCH 1/3] WIP --- sei-tendermint/autobahn/types/epoch_duo.go | 55 +---- sei-tendermint/internal/autobahn/avail/app.go | 111 ++++------- .../internal/autobahn/avail/block_votes.go | 3 +- .../internal/autobahn/avail/commit.go | 61 +++--- .../autobahn/avail/epoch_transition.go | 188 +++--------------- .../internal/autobahn/avail/inner.go | 42 ++-- .../internal/autobahn/avail/lane.go | 42 ++-- .../internal/autobahn/avail/persistence.go | 29 +-- .../internal/autobahn/avail/state.go | 4 +- .../internal/autobahn/epoch/registry.go | 80 ++++---- 10 files changed, 199 insertions(+), 416 deletions(-) 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/internal/autobahn/avail/app.go b/sei-tendermint/internal/autobahn/avail/app.go index 65dc14178b..630a337354 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,33 +11,29 @@ 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 err := ctrl.Wait(ctx); err != nil { - return nil, nil, err - } + if err := ctrl.WaitUntil(ctx, func() bool { return types.NextOpt(inner.app.anchor) > idx }); err!= nil { + return nil,err } + return inner.app.anchor.OrPanic("missing anchor"), nil } panic("unreachable") } @@ -45,38 +42,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 +71,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 +80,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 +108,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..87d1cb2ff8 100644 --- a/sei-tendermint/internal/autobahn/avail/commit.go +++ b/sei-tendermint/internal/autobahn/avail/commit.go @@ -17,6 +17,23 @@ func (s *State) waitForCommitQC(ctx context.Context, idx types.RoadIndex) error return err } +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. func (s *State) CommitQC(ctx context.Context, idx types.RoadIndex) (*types.CommitQC, error) { if err := s.waitForCommitQC(ctx, idx); err != nil { @@ -40,55 +57,41 @@ func (s *State) CommitQC(ctx context.Context, idx types.RoadIndex) (*types.Commi // // Admit-then-verify is intentional backpressure for ahead-of-window QCs. 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) + qc, epoch, err := s.commitQCAndEpoch(ctx, n) if err != nil { return nil, err } - ep, err := s.epochDuo.Load().EpochForRoad(qc.Proposal().Index()) - 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/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..7d6062ebc5 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 @@ -64,10 +64,10 @@ func newInner(registry *epoch.Registry, commitTip types.RoadIndex, loaded utils. } 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](), @@ -161,14 +161,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 +173,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 +201,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 +215,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..588428e3fb 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 { @@ -249,7 +252,7 @@ func (s *State) collectPersistBatch(ctx context.Context, persistedAnchorNext typ // 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 types.NextOpt(inner.app.anchor) != persistedAnchorNext { return true } for _, ls := range inner.lanes.byID { @@ -271,23 +274,13 @@ func (s *State) collectPersistBatch(ctx context.Context, persistedAnchorNext typ for n := commitQCNext; 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 types.NextOpt(inner.app.anchor) != persistedAnchorNext { + 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..21ebf7154a 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 } diff --git a/sei-tendermint/internal/autobahn/epoch/registry.go b/sei-tendermint/internal/autobahn/epoch/registry.go index 42f0aaf475..805de7b378 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,21 +193,15 @@ 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. @@ -225,15 +209,27 @@ func (r *Registry) DuoAt(roadIndex types.RoadIndex) (types.EpochDuo, error) { // 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 + i := IndexForRoad(roadIndex) + 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) 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") } + From b5c7bf320f60d2f6255007a958d0e73de4b304d1 Mon Sep 17 00:00:00 2001 From: Grzegorz Prusak Date: Fri, 31 Jul 2026 14:38:03 +0200 Subject: [PATCH 2/3] deadcode --- sei-tendermint/internal/autobahn/avail/app.go | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/sei-tendermint/internal/autobahn/avail/app.go b/sei-tendermint/internal/autobahn/avail/app.go index 630a337354..3edc517115 100644 --- a/sei-tendermint/internal/autobahn/avail/app.go +++ b/sei-tendermint/internal/autobahn/avail/app.go @@ -25,19 +25,6 @@ func (s *State) LastAppQC() utils.Option[*types.AppQC] { 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) WaitForAnchor(ctx context.Context, idx types.RoadIndex) (*PruneAnchor, error) { - for inner, ctrl := range s.inner.Lock() { - if err := ctrl.WaitUntil(ctx, func() bool { return types.NextOpt(inner.app.anchor) > idx }); err!= nil { - return nil,err - } - return inner.app.anchor.OrPanic("missing anchor"), nil - } - panic("unreachable") -} - // PushAppVote pushes an AppVote to the state. // 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). From 80a11fa5a4bc0ae70a6d8efe6627eef1a0f393c7 Mon Sep 17 00:00:00 2001 From: Grzegorz Prusak Date: Fri, 31 Jul 2026 14:56:48 +0200 Subject: [PATCH 3/3] fix --- sei-tendermint/internal/autobahn/avail/inner.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sei-tendermint/internal/autobahn/avail/inner.go b/sei-tendermint/internal/autobahn/avail/inner.go index 7d6062ebc5..06e0e2e04e 100644 --- a/sei-tendermint/internal/autobahn/avail/inner.go +++ b/sei-tendermint/internal/autobahn/avail/inner.go @@ -174,7 +174,7 @@ func verifyCommitQCInDuo(duo types.EpochDuo, qc *types.CommitQC) error { // already satisfied. Adds Current lanes; does not delete old lanes // (TODO(lane-expiry)). Touches epoch + lane votes (reweight). func (i *inner) advanceEpoch(epoch *types.Epoch) bool { - if i.epoch.Load().EpochIndex() < epoch.EpochIndex() { + if i.epoch.Load().EpochIndex() >= epoch.EpochIndex() { return false } c := epoch.Committee()