From ff8b76c96727ac3b85279d1ce740aa76d4693406 Mon Sep 17 00:00:00 2001 From: Wen Date: Wed, 16 Sep 2026 13:41:47 +0200 Subject: [PATCH 1/3] Retain the last Autobahn lane block across prune so production does not stall on a zero parent hash. --- .../internal/autobahn/avail/inner.go | 78 +++++++-- .../internal/autobahn/avail/inner_test.go | 155 +++++++++++++++++- .../internal/autobahn/avail/state.go | 33 ++-- .../internal/autobahn/avail/state_test.go | 103 +++++++++++- 4 files changed, 338 insertions(+), 31 deletions(-) diff --git a/sei-tendermint/internal/autobahn/avail/inner.go b/sei-tendermint/internal/autobahn/avail/inner.go index 7e819e3ff2..b7915992dd 100644 --- a/sei-tendermint/internal/autobahn/avail/inner.go +++ b/sei-tendermint/internal/autobahn/avail/inner.go @@ -10,6 +10,54 @@ import ( "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" ) +// blockQueue is a per-lane block queue. +type blockQueue struct { + queue[types.BlockNumber, *types.Signed[*types.LaneProposal]] + // last is the last proposal this node pushed, or None if prune jumped past + // every block it held. + last utils.Option[*types.Signed[*types.LaneProposal]] +} + +func newBlockQueue() *blockQueue { + return &blockQueue{queue: *newQueue[types.BlockNumber, *types.Signed[*types.LaneProposal]]()} +} + +func (q *blockQueue) pushBack(p *types.Signed[*types.LaneProposal]) { + q.queue.pushBack(p) + q.last = utils.Some(p) +} + +func (q *blockQueue) prune(newFirst types.BlockNumber) { + if newFirst <= q.first { + return + } + if newFirst > q.next { + // The Anchor certified past every block this lane holds, so the block it + // names as the new tip was never seen locally. + q.last = utils.None[*types.Signed[*types.LaneProposal]]() + } + q.queue.prune(newFirst) +} + +// unpersistedLast returns the last block once it has left the active range and +// block persistence has not reached it. +func (q *blockQueue) unpersistedLast(nextToPersist types.BlockNumber) utils.Option[*types.Signed[*types.LaneProposal]] { + if p, ok := q.last.Get(); ok { + if n := p.Msg().Block().Header().BlockNumber(); n < q.first && nextToPersist <= n { + return utils.Some(p) + } + } + return utils.None[*types.Signed[*types.LaneProposal]]() +} + +// retentionFloor returns the lowest block number the WAL must still hold. +func (q *blockQueue) retentionFloor() types.BlockNumber { + if p, ok := q.last.Get(); ok { + return min(q.first, p.Msg().Block().Header().BlockNumber()) + } + return q.first +} + // inner holds roads and per-LaneID block/vote maps. type inner struct { persistedCommitQC utils.AtomicSend[utils.Option[*types.CommitQC]] // latest persisted CommitQC @@ -25,7 +73,7 @@ type inner struct { // When it lags applied, epochForVote falls back to this committee for // departing-lane voters. anchorEpoch utils.Option[*types.Epoch] - blocks map[types.LaneID]*queue[types.BlockNumber, *types.Signed[*types.LaneProposal]] + blocks map[types.LaneID]*blockQueue votes map[types.LaneID]*queue[types.BlockNumber, *blockVotes] // nextBlockToPersist tracks per-lane how far block persistence has progressed. // RecvBatch only yields blocks below this cursor for voting. @@ -61,7 +109,7 @@ func newInner(ep *types.Epoch, first types.RoadIndex) *inner { persistedCommitQC: utils.NewAtomicSend(utils.None[*types.CommitQC]()), consensusSpec: utils.NewAtomicSend(types.ConsensusSpec{CommitQC: utils.None[*types.CommitQC](), Epoch: ep}), roads: roads, - blocks: map[types.LaneID]*queue[types.BlockNumber, *types.Signed[*types.LaneProposal]]{}, + blocks: map[types.LaneID]*blockQueue{}, votes: map[types.LaneID]*queue[types.BlockNumber, *blockVotes]{}, nextBlockToPersist: map[types.LaneID]types.BlockNumber{}, } @@ -85,18 +133,20 @@ func (i *inner) restoreBlocks(blocks map[types.LaneID][]persist.LoadedBlock) err return fmt.Errorf("lane %s: loaded %d blocks exceeds capacity %d", lane, len(bs), BlocksPerLane) } if b.Number < q.next { + // Certified. Restore last from the proposal at first-1 when present. + if b.Number+1 == q.first { + q.last = utils.Some(b.Proposal) + } continue } if b.Number != q.next { return fmt.Errorf("lane %s: non-contiguous persisted blocks: expected %d, got %d", lane, q.next, b.Number) } - // We check the parent hash only for the blocks above the anchor, because: - // * node can cast LaneVote for the block of the lane without checking the parent hash, - // in case the previous block was already (executed and) pruned from memory. - // * current WAL implementation is lazily pruning on disk, so old executed blocks might be loaded on startup. - if q.Len() > 0 { + // In-range blocks must parent to last when last is known (queue tip, or + // first-1 restored above). Certified WAL records are not re-checked. + if prev, ok := q.last.Get(); ok { ph := b.Proposal.Msg().Block().Header().ParentHash() - if q.q[q.next-1].Msg().Block().Header().Hash() != ph { + if prev.Msg().Block().Header().Hash() != ph { return fmt.Errorf("lane %s: parent hash mismatch at block %d", lane, b.Number) } } @@ -192,7 +242,7 @@ func (i *inner) addLane(lane types.LaneID) bool { if _, ok := i.blocks[lane]; ok { return false } - i.blocks[lane] = newQueue[types.BlockNumber, *types.Signed[*types.LaneProposal]]() + i.blocks[lane] = newBlockQueue() i.votes[lane] = newQueue[types.BlockNumber, *blockVotes]() i.nextBlockToPersist[lane] = 0 return true @@ -246,10 +296,16 @@ func (i *inner) prune(anchor data.Anchor) int { for lane, vq := range i.votes { lr := anchor.CommitQC.LaneRange(lane) bq := i.blocks[lane] + // TODO: when prune jumps past what this node holds, seed the parent from + // lr.LastHash() at Next()-1 (non-empty range only). Empty ranges still + // carry a zero LastHash, so a later QC cannot replace a conflicting + // local predecessor or recover a tip the WAL does not have. vq.prune(lr.Next()) bq.prune(lr.Next()) - if i.nextBlockToPersist[lane] < lr.Next() { - i.nextBlockToPersist[lane] = lr.Next() + // A lagging cursor stops at retentionFloor so an unflushed last can still + // be written. The cursor is never rewound: already past last means it is on disk. + if floor := bq.retentionFloor(); i.nextBlockToPersist[lane] < floor { + i.nextBlockToPersist[lane] = floor } } if i.roads.Len() == 0 { diff --git a/sei-tendermint/internal/autobahn/avail/inner_test.go b/sei-tendermint/internal/autobahn/avail/inner_test.go index a55c3d4b37..ada25c624e 100644 --- a/sei-tendermint/internal/autobahn/avail/inner_test.go +++ b/sei-tendermint/internal/autobahn/avail/inner_test.go @@ -34,6 +34,139 @@ func contiguousBlocks(key types.SecretKey, lane types.LaneID, n int, rng utils.R return bs } +func TestBlockQueueRetainsLast(t *testing.T) { + rng := utils.TestRng() + key := types.GenSecretKey(rng) + lane := types.LaneID{Validator: key.Public(), Joined: 0} + blocks := contiguousBlocks(key, lane, 3, rng) + q := newBlockQueue() + for _, b := range blocks { + q.pushBack(b.Proposal) + } + + q.prune(2) + require.Equal(t, types.BlockNumber(2), q.first) + require.Equal(t, types.BlockNumber(3), q.next) + require.Equal(t, utils.Some(blocks[2].Proposal), q.last) + // Block 2 is still active and carries the chain, so nothing below first is needed. + require.Equal(t, types.BlockNumber(2), q.retentionFloor()) + + q.prune(3) + require.Equal(t, types.BlockNumber(3), q.first) + require.Equal(t, types.BlockNumber(3), q.next) + require.Equal(t, utils.Some(blocks[2].Proposal), q.last) + require.Equal(t, types.BlockNumber(2), q.retentionFloor()) + + q.prune(5) + require.Equal(t, types.BlockNumber(5), q.first) + require.Equal(t, types.BlockNumber(5), q.next) + require.Equal(t, utils.None[*types.Signed[*types.LaneProposal]](), q.last) + require.Equal(t, types.BlockNumber(5), q.retentionFloor()) +} + +func TestInnerPruneViaQCRetainsLast(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + ep := registry.MustEpoch(0) + lane := ep.Committee().Lane(keys[0].Public()).OrPanic("lane") + blocks := contiguousBlocks(keys[0], lane, 3, rng) + i := newInner(ep, 0) + for _, b := range blocks { + i.blocks[lane].pushBack(b.Proposal) + } + + last := blocks[len(blocks)-1].Proposal.Msg().Block().Header() + qc := types.BuildCommitQC(ep, keys, utils.None[*types.CommitQC](), map[types.LaneID]*types.LaneQC{ + lane: types.NewLaneQC(makeLaneVotes(keys, last)), + }) + lr := qc.LaneRange(lane) + require.Equal(t, types.BlockNumber(3), lr.Next()) + require.Equal(t, last.Hash(), lr.LastHash()) + + i.prune(data.Anchor{ + CommitQC: qc, + AppQC: data.TestAppQC(keys, types.NewAppProposal(qc.Proposal(), types.AppHash{})), + Epoch: ep, + }) + q := i.blocks[lane] + require.Equal(t, types.BlockNumber(3), q.first) + require.Equal(t, types.BlockNumber(3), q.next) + require.Equal(t, utils.Some(blocks[2].Proposal), q.last) + require.Equal(t, types.BlockNumber(2), q.retentionFloor()) + require.Equal(t, types.BlockNumber(2), i.nextBlockToPersist[lane]) + p, ok := q.unpersistedLast(i.nextBlockToPersist[lane]).Get() + require.True(t, ok) + require.Equal(t, blocks[2].Proposal, p) +} + +func TestBlockQueueLastSurvivesRestart(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + key := keys[0] + lane := registry.MustEpoch(0).Committee().Lane(key.Public()).OrPanic("lane") + block := contiguousBlocks(key, lane, 1, rng)[0] + dir := t.TempDir() + + persister, _, err := persist.NewBlockPersister(utils.Some(dir)) + require.NoError(t, err) + require.NoError(t, persister.PruneAndPersist( + lane, + 0, + []*types.Signed[*types.LaneProposal]{block.Proposal}, + )) + q := newBlockQueue() + q.pushBack(block.Proposal) + q.prune(1) + require.NoError(t, persister.PruneAndPersist(lane, q.retentionFloor(), nil)) + require.NoError(t, persister.Close()) + + persister, loaded, err := persist.NewBlockPersister(utils.Some(dir)) + require.NoError(t, err) + require.NoError(t, persister.Close()) + i := newInner(registry.MustEpoch(0), 0) + i.blocks[lane].prune(1) + require.NoError(t, i.restoreBlocks(loaded)) + last, ok := i.blocks[lane].last.Get() + require.True(t, ok) + require.Equal(t, block.Proposal.Msg().Block().Header().Hash(), last.Msg().Block().Header().Hash()) +} + +func TestBlockQueueUnflushedLastSurvivesRestart(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + key := keys[0] + lane := registry.MustEpoch(0).Committee().Lane(key.Public()).OrPanic("lane") + block := contiguousBlocks(key, lane, 1, rng)[0] + dir := t.TempDir() + + persister, _, err := persist.NewBlockPersister(utils.Some(dir)) + require.NoError(t, err) + q := newBlockQueue() + q.pushBack(block.Proposal) + q.prune(1) + + p, ok := q.unpersistedLast(0).Get() + require.True(t, ok) + require.Equal(t, block.Proposal, p) + require.NoError(t, persister.PruneAndPersist( + lane, + q.retentionFloor(), + []*types.Signed[*types.LaneProposal]{p}, + )) + require.False(t, q.unpersistedLast(1).IsPresent()) + require.NoError(t, persister.Close()) + + persister, loaded, err := persist.NewBlockPersister(utils.Some(dir)) + require.NoError(t, err) + require.NoError(t, persister.Close()) + i := newInner(registry.MustEpoch(0), 0) + i.blocks[lane].prune(1) + require.NoError(t, i.restoreBlocks(loaded)) + last, ok := i.blocks[lane].last.Get() + require.True(t, ok) + require.Equal(t, block.Proposal.Msg().Block().Header().Hash(), last.Msg().Block().Header().Hash()) +} + func TestRestoreInner_Empty(t *testing.T) { rng := utils.TestRng() registry, _ := epoch.GenRegistry(rng, 4) @@ -90,6 +223,24 @@ func TestRestoreInner_LoadedBlocks(t *testing.T) { require.Equal(t, types.BlockNumber(0), q.next) }) + t.Run("last below anchor", func(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 4) + lane := registry.MustEpoch(0).Committee().Lane(keys[0].Public()).OrPanic("keys[0]") + blocks := contiguousBlocks(keys[0], lane, 2, rng) + i := newInner(registry.MustEpoch(0), 0) + i.blocks[lane].prune(2) + + err := i.restoreBlocks(map[types.LaneID][]persist.LoadedBlock{lane: blocks}) + require.NoError(t, err) + q := i.blocks[lane] + require.Equal(t, types.BlockNumber(2), q.first) + require.Equal(t, types.BlockNumber(2), q.next) + require.Equal(t, utils.Some(blocks[1].Proposal), q.last) + require.Equal(t, types.BlockNumber(1), q.retentionFloor()) + require.Equal(t, types.BlockNumber(2), i.nextBlockToPersist[lane]) + }) + t.Run("foreign loaded lane does not touch committee queues", func(t *testing.T) { rng := utils.TestRng() registry, _ := epoch.GenRegistry(rng, 4) @@ -218,7 +369,7 @@ func TestAddLane_ReportsNewLaneForEachMembershipPeriod(t *testing.T) { a := types.GenSecretKey(rng) i := &inner{ - blocks: map[types.LaneID]*queue[types.BlockNumber, *types.Signed[*types.LaneProposal]]{}, + blocks: map[types.LaneID]*blockQueue{}, votes: map[types.LaneID]*queue[types.BlockNumber, *blockVotes]{}, nextBlockToPersist: map[types.LaneID]types.BlockNumber{}, } @@ -249,7 +400,7 @@ func TestRefreshConsensusSpec_WithholdsTipUntilNextViewEpochApplied(t *testing.T persistedCommitQC: utils.NewAtomicSend(utils.None[*types.CommitQC]()), consensusSpec: utils.NewAtomicSend(types.ConsensusSpec{CommitQC: utils.None[*types.CommitQC](), Epoch: ep0}), roads: newQueue[types.RoadIndex, *road](), - blocks: map[types.LaneID]*queue[types.BlockNumber, *types.Signed[*types.LaneProposal]]{}, + blocks: map[types.LaneID]*blockQueue{}, votes: map[types.LaneID]*queue[types.BlockNumber, *blockVotes]{}, nextBlockToPersist: map[types.LaneID]types.BlockNumber{}, } diff --git a/sei-tendermint/internal/autobahn/avail/state.go b/sei-tendermint/internal/autobahn/avail/state.go index 102c3c611d..5b9e1346cc 100644 --- a/sei-tendermint/internal/autobahn/avail/state.go +++ b/sei-tendermint/internal/autobahn/avail/state.go @@ -458,7 +458,6 @@ func (s *State) PushBlock(ctx context.Context, p *types.Signed[*types.LanePropos if !ok { return nil } - // not needed any more if q.next != n { return nil } @@ -468,11 +467,8 @@ func (s *State) PushBlock(ctx context.Context, p *types.Signed[*types.LanePropos // chain than we already have). We log it to aid debugging stalled // lanes but do not return an error — the caller should not tear // down the peer connection over an equivocating producer. - // NOTE: after pruning (q.first >= q.next), we cannot verify the parent - // hash because the previous block is gone. This is safe because - // headers() never follows the first block's parentHash in a LaneRange. - if q.first < q.next { - prevHash := q.q[q.next-1].Msg().Block().Header().Hash() + if prev, ok := q.last.Get(); ok { + prevHash := prev.Msg().Block().Header().Hash() if h.ParentHash() != prevHash { logger.Error("parent hash mismatch (producer equivocation)", "lane", lane, @@ -662,9 +658,9 @@ func (s *State) ProduceLocalBlock(lane types.LaneID, n types.BlockNumber, payloa if q.next != n { return nil, fmt.Errorf("unexpected block number: got %v, want %v", n, q.next) } - var parent types.BlockHeaderHash - if q.first < q.next { - parent = q.q[q.next-1].Msg().Block().Header().Hash() + parent := types.BlockHeaderHash{} + if prev, ok := q.last.Get(); ok { + parent = prev.Msg().Block().Header().Hash() } result = types.Sign(s.key, types.NewLaneProposal(types.NewBlock(lane, q.next, parent, payload))) q.pushBack(result) @@ -853,15 +849,18 @@ type persistBatch struct { commitQCs commitQCsBatch } -// setNextBlockToPersist sets the per-lane block persistence cursor to next. -// Called once per lane after that lane's batch has been flushed so that -// RecvBatch (and therefore voting) can unblock. Safe for concurrent -// callers (acquires s.inner lock internally). +// setNextBlockToPersist advances the per-lane persistence cursor to next when +// next is ahead of the current cursor. RecvBatch yields headers strictly below +// this cursor, so a successful flush unblocks voting. Safe for concurrent callers. func (s *State) setNextBlockToPersist(lane types.LaneID, next types.BlockNumber) { for inner, ctrl := range s.inner.Lock() { if _, ok := inner.blocks[lane]; !ok { return } + if inner.nextBlockToPersist[lane] >= next { + // prune may have already advanced the cursor while this batch was on disk. + return + } inner.nextBlockToPersist[lane] = next ctrl.Updated() } @@ -907,8 +906,12 @@ func (s *State) collectPersistBatch(ctx context.Context) (*persistBatch, error) b.commitQCs.tail = append(b.commitQCs.tail, inner.roads.q[n].commitQC) } for lane, q := range inner.blocks { - bb := blocksBatch{first: q.first} - for n := max(inner.nextBlockToPersist[lane], q.first); n < q.next; n++ { + cursor := inner.nextBlockToPersist[lane] + bb := blocksBatch{first: q.retentionFloor()} + if p, ok := q.unpersistedLast(cursor).Get(); ok { + bb.tail = append(bb.tail, p) + } + for n := max(cursor, q.first); n < q.next; n++ { bb.tail = append(bb.tail, q.q[n]) } b.blocks[lane] = bb diff --git a/sei-tendermint/internal/autobahn/avail/state_test.go b/sei-tendermint/internal/autobahn/avail/state_test.go index 34c4156b48..00c1400add 100644 --- a/sei-tendermint/internal/autobahn/avail/state_test.go +++ b/sei-tendermint/internal/autobahn/avail/state_test.go @@ -26,9 +26,9 @@ func pushPeerLaneBlock(state *State, key types.SecretKey, payload *types.Payload return nil, ErrLaneClosed } n := q.next - var parent types.BlockHeaderHash - if q.first < q.next { - parent = q.q[q.next-1].Msg().Block().Header().Hash() + parent := types.BlockHeaderHash{} + if prev, ok := q.last.Get(); ok { + parent = prev.Msg().Block().Header().Hash() } b = types.Sign(key, types.NewLaneProposal(types.NewBlock(lane, n, parent, payload))) q.pushBack(b) @@ -63,6 +63,21 @@ func makeLaneVotes(keys []types.SecretKey, h *types.BlockHeader) []*types.Signed return votes } +func pruneToHeader(state *State, keys []types.SecretKey, h *types.BlockHeader) { + ep := state.SubscribeConsensusSpec().Load().Epoch + qc := types.BuildCommitQC(ep, keys, utils.None[*types.CommitQC](), map[types.LaneID]*types.LaneQC{ + h.Lane(): types.NewLaneQC(makeLaneVotes(keys, h)), + }) + for inner, ctrl := range state.inner.Lock() { + inner.prune(data.Anchor{ + CommitQC: qc, + AppQC: data.TestAppQC(keys, types.NewAppProposal(qc.Proposal(), types.AppHash{})), + Epoch: ep, + }) + ctrl.Updated() + } +} + func qcPayloadHashes(qc *types.FullCommitQC) byLane[types.PayloadHash] { x := byLane[types.PayloadHash]{} for _, h := range qc.Headers() { @@ -560,6 +575,88 @@ func TestPushBlockRejectsBadParentHash(t *testing.T) { require.Equal(t, types.BlockNumber(1), state.NextBlock(lane)) } +func TestPushBlockRejectsBadRetainedParentHash(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + state := utils.OrPanic1(NewState( + keys[0], + newTestDataState(&data.Config{Registry: registry}), + utils.None[string](), + )) + lane := registry.MustEpoch(0).Committee().Lane(keys[0].Public()).OrPanic("lane") + + first, err := state.ProduceLocalBlock(lane, state.NextBlock(lane), types.GenPayload(rng)) + require.NoError(t, err) + pruneToHeader(state, keys, first.Msg().Block().Header()) + + block := types.NewBlock(lane, 1, types.GenBlockHeaderHash(rng), types.GenPayload(rng)) + require.NoError(t, state.PushBlock(ctx, types.Sign(keys[0], types.NewLaneProposal(block)))) + require.Equal(t, types.BlockNumber(1), state.NextBlock(lane)) +} + +func TestProduceLocalBlockUsesRetainedLast(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + state := utils.OrPanic1(NewState( + keys[0], + newTestDataState(&data.Config{Registry: registry}), + utils.None[string](), + )) + lane := registry.MustEpoch(0).Committee().Lane(keys[0].Public()).OrPanic("lane") + + first, err := state.ProduceLocalBlock(lane, state.NextBlock(lane), types.GenPayload(rng)) + require.NoError(t, err) + pruneToHeader(state, keys, first.Msg().Block().Header()) + + second, err := state.ProduceLocalBlock(lane, state.NextBlock(lane), types.GenPayload(rng)) + require.NoError(t, err) + require.Equal(t, first.Msg().Block().Header().Hash(), second.Msg().Block().Header().ParentHash()) +} + +func TestCollectPersistBatchWritesUnflushedLast(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + state := utils.OrPanic1(NewState( + keys[0], + newTestDataState(&data.Config{Registry: registry}), + utils.None[string](), + )) + lane := registry.MustEpoch(0).Committee().Lane(keys[0].Public()).OrPanic("lane") + + first, err := state.ProduceLocalBlock(lane, state.NextBlock(lane), types.GenPayload(rng)) + require.NoError(t, err) + pruneToHeader(state, keys, first.Msg().Block().Header()) + + batch, err := state.collectPersistBatch(ctx) + require.NoError(t, err) + bb := batch.blocks[lane] + require.Equal(t, types.BlockNumber(0), bb.first) + require.Equal(t, []*types.Signed[*types.LaneProposal]{first}, bb.tail) +} + +func TestSetNextBlockToPersistDoesNotRewind(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + state := utils.OrPanic1(NewState( + keys[0], + newTestDataState(&data.Config{Registry: registry}), + utils.None[string](), + )) + lane := registry.MustEpoch(0).Committee().Lane(keys[0].Public()).OrPanic("lane") + + state.setNextBlockToPersist(lane, 5) + state.setNextBlockToPersist(lane, 3) + for inner := range state.inner.Lock() { + require.Equal(t, types.BlockNumber(5), inner.nextBlockToPersist[lane]) + } + state.setNextBlockToPersist(lane, 6) + for inner := range state.inner.Lock() { + require.Equal(t, types.BlockNumber(6), inner.nextBlockToPersist[lane]) + } +} + func TestPushBlockRejectsWrongSigner(t *testing.T) { ctx := t.Context() rng := utils.TestRng() From a54a6efaa7420679163067272ac59ea1c358d36a Mon Sep 17 00:00:00 2001 From: Wen Date: Wed, 16 Sep 2026 15:03:05 +0200 Subject: [PATCH 2/3] Do not parent-check PushBlock against a last block that prune has already cut. --- .../internal/autobahn/avail/inner.go | 8 ++-- .../internal/autobahn/avail/inner_test.go | 47 +++++++++++++++++++ .../internal/autobahn/avail/state.go | 6 ++- .../internal/autobahn/avail/state_test.go | 25 +++++++++- 4 files changed, 78 insertions(+), 8 deletions(-) diff --git a/sei-tendermint/internal/autobahn/avail/inner.go b/sei-tendermint/internal/autobahn/avail/inner.go index b7915992dd..b380b2eb09 100644 --- a/sei-tendermint/internal/autobahn/avail/inner.go +++ b/sei-tendermint/internal/autobahn/avail/inner.go @@ -142,11 +142,11 @@ func (i *inner) restoreBlocks(blocks map[types.LaneID][]persist.LoadedBlock) err if b.Number != q.next { return fmt.Errorf("lane %s: non-contiguous persisted blocks: expected %d, got %d", lane, q.next, b.Number) } - // In-range blocks must parent to last when last is known (queue tip, or - // first-1 restored above). Certified WAL records are not re-checked. - if prev, ok := q.last.Get(); ok { + // Parent is checked only inside [first, next). last restored from + // first-1 is for local production, not this check. + if q.first < q.next { ph := b.Proposal.Msg().Block().Header().ParentHash() - if prev.Msg().Block().Header().Hash() != ph { + if q.q[q.next-1].Msg().Block().Header().Hash() != ph { return fmt.Errorf("lane %s: parent hash mismatch at block %d", lane, b.Number) } } diff --git a/sei-tendermint/internal/autobahn/avail/inner_test.go b/sei-tendermint/internal/autobahn/avail/inner_test.go index ada25c624e..a742ca5022 100644 --- a/sei-tendermint/internal/autobahn/avail/inner_test.go +++ b/sei-tendermint/internal/autobahn/avail/inner_test.go @@ -99,6 +99,33 @@ func TestInnerPruneViaQCRetainsLast(t *testing.T) { require.Equal(t, blocks[2].Proposal, p) } +func TestInnerPruneKeepsLastAheadOfQC(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + ep := registry.MustEpoch(0) + lane := ep.Committee().Lane(keys[0].Public()).OrPanic("lane") + blocks := contiguousBlocks(keys[0], lane, 3, rng) + i := newInner(ep, 0) + for _, b := range blocks { + i.blocks[lane].pushBack(b.Proposal) + } + + qc := types.BuildCommitQC(ep, keys, utils.None[*types.CommitQC](), map[types.LaneID]*types.LaneQC{ + lane: types.NewLaneQC(makeLaneVotes(keys, blocks[0].Proposal.Msg().Block().Header())), + }) + require.Equal(t, types.BlockNumber(1), qc.LaneRange(lane).Next()) + + i.prune(data.Anchor{ + CommitQC: qc, + AppQC: data.TestAppQC(keys, types.NewAppProposal(qc.Proposal(), types.AppHash{})), + Epoch: ep, + }) + q := i.blocks[lane] + require.Equal(t, types.BlockNumber(1), q.first) + require.Equal(t, types.BlockNumber(3), q.next) + require.Equal(t, utils.Some(blocks[2].Proposal), q.last) +} + func TestBlockQueueLastSurvivesRestart(t *testing.T) { rng := utils.TestRng() registry, keys := epoch.GenRegistry(rng, 3) @@ -241,6 +268,26 @@ func TestRestoreInner_LoadedBlocks(t *testing.T) { require.Equal(t, types.BlockNumber(2), i.nextBlockToPersist[lane]) }) + t.Run("leftover below first is not parent-checked", func(t *testing.T) { + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 4) + lane := registry.MustEpoch(0).Committee().Lane(keys[0].Public()).OrPanic("keys[0]") + old := testSignedBlock(keys[0], lane, 0, types.BlockHeaderHash{}, rng) + live := testSignedBlock(keys[0], lane, 1, types.GenBlockHeaderHash(rng), rng) + i := newInner(registry.MustEpoch(0), 0) + i.blocks[lane].prune(1) + + err := i.restoreBlocks(map[types.LaneID][]persist.LoadedBlock{lane: { + {Number: 0, Proposal: old}, + {Number: 1, Proposal: live}, + }}) + require.NoError(t, err) + q := i.blocks[lane] + require.Equal(t, types.BlockNumber(1), q.first) + require.Equal(t, types.BlockNumber(2), q.next) + require.Equal(t, utils.Some(live), q.last) + }) + t.Run("foreign loaded lane does not touch committee queues", func(t *testing.T) { rng := utils.TestRng() registry, _ := epoch.GenRegistry(rng, 4) diff --git a/sei-tendermint/internal/autobahn/avail/state.go b/sei-tendermint/internal/autobahn/avail/state.go index 5b9e1346cc..9641d3bbf9 100644 --- a/sei-tendermint/internal/autobahn/avail/state.go +++ b/sei-tendermint/internal/autobahn/avail/state.go @@ -467,8 +467,10 @@ func (s *State) PushBlock(ctx context.Context, p *types.Signed[*types.LanePropos // chain than we already have). We log it to aid debugging stalled // lanes but do not return an error — the caller should not tear // down the peer connection over an equivocating producer. - if prev, ok := q.last.Get(); ok { - prevHash := prev.Msg().Block().Header().Hash() + // Parent is checked only while the predecessor is still in [first, next). + // last retained below first is for local production, not this check. + if q.first < q.next { + prevHash := q.q[q.next-1].Msg().Block().Header().Hash() if h.ParentHash() != prevHash { logger.Error("parent hash mismatch (producer equivocation)", "lane", lane, diff --git a/sei-tendermint/internal/autobahn/avail/state_test.go b/sei-tendermint/internal/autobahn/avail/state_test.go index 00c1400add..8394719271 100644 --- a/sei-tendermint/internal/autobahn/avail/state_test.go +++ b/sei-tendermint/internal/autobahn/avail/state_test.go @@ -575,7 +575,7 @@ func TestPushBlockRejectsBadParentHash(t *testing.T) { require.Equal(t, types.BlockNumber(1), state.NextBlock(lane)) } -func TestPushBlockRejectsBadRetainedParentHash(t *testing.T) { +func TestPushBlockDoesNotCheckParentAfterPrune(t *testing.T) { ctx := t.Context() rng := utils.TestRng() registry, keys := epoch.GenRegistry(rng, 3) @@ -592,7 +592,28 @@ func TestPushBlockRejectsBadRetainedParentHash(t *testing.T) { block := types.NewBlock(lane, 1, types.GenBlockHeaderHash(rng), types.GenPayload(rng)) require.NoError(t, state.PushBlock(ctx, types.Sign(keys[0], types.NewLaneProposal(block)))) - require.Equal(t, types.BlockNumber(1), state.NextBlock(lane)) + require.Equal(t, types.BlockNumber(2), state.NextBlock(lane)) +} + +func TestPushBlockRecoversWhenCertifiedLastDiffers(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + registry, keys := epoch.GenRegistry(rng, 3) + state := utils.OrPanic1(NewState( + keys[0], + newTestDataState(&data.Config{Registry: registry}), + utils.None[string](), + )) + lane := registry.MustEpoch(0).Committee().Lane(keys[0].Public()).OrPanic("lane") + + _, err := state.ProduceLocalBlock(lane, state.NextBlock(lane), types.GenPayload(rng)) + require.NoError(t, err) + certified := types.NewBlock(lane, 0, types.BlockHeaderHash{}, types.GenPayload(rng)) + pruneToHeader(state, keys, certified.Header()) + + next := types.NewBlock(lane, 1, certified.Header().Hash(), types.GenPayload(rng)) + require.NoError(t, state.PushBlock(ctx, types.Sign(keys[0], types.NewLaneProposal(next)))) + require.Equal(t, types.BlockNumber(2), state.NextBlock(lane)) } func TestProduceLocalBlockUsesRetainedLast(t *testing.T) { From f8a3863fe8f81846461fe557d0d71d018ad57b44 Mon Sep 17 00:00:00 2001 From: Wen Date: Wed, 16 Sep 2026 15:36:19 +0200 Subject: [PATCH 3/3] Document that last sits at first-1 or above, and keep the jump-prune TODO on the clear. --- sei-tendermint/internal/autobahn/avail/inner.go | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/sei-tendermint/internal/autobahn/avail/inner.go b/sei-tendermint/internal/autobahn/avail/inner.go index b380b2eb09..7e1e89e61f 100644 --- a/sei-tendermint/internal/autobahn/avail/inner.go +++ b/sei-tendermint/internal/autobahn/avail/inner.go @@ -13,8 +13,7 @@ import ( // blockQueue is a per-lane block queue. type blockQueue struct { queue[types.BlockNumber, *types.Signed[*types.LaneProposal]] - // last is the last proposal this node pushed, or None if prune jumped past - // every block it held. + // last is None, or this node's last pushed proposal at height >= first-1. last utils.Option[*types.Signed[*types.LaneProposal]] } @@ -27,13 +26,15 @@ func (q *blockQueue) pushBack(p *types.Signed[*types.LaneProposal]) { q.last = utils.Some(p) } +// prune drops [first, newFirst). last is kept when newFirst <= next and +// cleared when newFirst > next. func (q *blockQueue) prune(newFirst types.BlockNumber) { if newFirst <= q.first { return } if newFirst > q.next { - // The Anchor certified past every block this lane holds, so the block it - // names as the new tip was never seen locally. + // TODO: seed last from a non-empty LaneRange LastHash at Next()-1. + // Empty ranges carry a zero LastHash, so they cannot replace a local last. q.last = utils.None[*types.Signed[*types.LaneProposal]]() } q.queue.prune(newFirst) @@ -296,10 +297,6 @@ func (i *inner) prune(anchor data.Anchor) int { for lane, vq := range i.votes { lr := anchor.CommitQC.LaneRange(lane) bq := i.blocks[lane] - // TODO: when prune jumps past what this node holds, seed the parent from - // lr.LastHash() at Next()-1 (non-empty range only). Empty ranges still - // carry a zero LastHash, so a later QC cannot replace a conflicting - // local predecessor or recover a tip the WAL does not have. vq.prune(lr.Next()) bq.prune(lr.Next()) // A lagging cursor stops at retentionFloor so an unflushed last can still