fix(consensus): end a catch-up pass that cannot send

beginCatchupAttempt charged a budget slot before loadMeta, the part-set
header check and loadPart, none of which ended the pass when they failed.
The peer supplies both the missing bit-array a pass is budgeted from and
the part-set header those checks run against, and ValidateBasic accepts
any well-formed pair, so it could open a maximum-size pass under a header
matching no stored block.

That bought two things. A pass that can never send charged a block-store
read and an error log to every gossip tick for its whole budget - roughly
160s at the production 100ms cadence - and since the retry deadline was
armed when the pass opened, it had long expired by the time the budget
ran out, so the next pass began immediately and the interval never
throttled anything. The same inflated budget also survived the peer
replacing its state, at the same height, with the stored block's real
header and a single missing part: the budget is deliberately not
re-derived once a pass is open, so the remainder landed as duplicate
sends of that one part.

Move the send into sendCatchupBlockPart and abandon the pass, arming a
fresh retry deadline, whenever it reports that no part reached the peer.
A pass now costs at most one failed attempt per interval, and only a pass
that got as far as sending keeps its budget.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RoXmbFrVf1BqVW1BHZv6Va

Author: lklimek

internal/consensus/gossiper.go

@@ -4,6 +4,7 @@ package consensus
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -191,28 +192,41 @@ func (g *msgGossiper) GossipBlockPartsForCatchup(
 	if !g.beginCatchupAttempt(prs) {
 		return
 	}
-	index, ok := prs.ProposalBlockParts.Not().PickRandom()
-	if !ok {
-		return
+	if err := g.sendCatchupBlockPart(ctx, prs); err != nil {
+		// The peer controls both the budget a pass opens with and the inputs that
+		// fail here, so a pass that cannot produce a send is abandoned rather than
+		// charged one slot per tick.
+		g.endCatchupPass()
 	}
+}
+
+// sendCatchupBlockPart sends the peer one part of its incomplete block, drawn at
+// random from the parts it reports missing. A non-nil error reports only that no
+// part reached the peer; the caller is not expected to inspect it.
+func (g *msgGossiper) sendCatchupBlockPart(ctx context.Context, prs *cstypes.PeerRoundState) error {
 	logger := g.logger.With([]any{
 		"height", prs.Height,
 		"round", prs.Round,
 	})
+	index, ok := prs.ProposalBlockParts.Not().PickRandom()
+	if !ok {
+		// The peer replaced its bit-array since the pass drew a budget from it.
+		return errors.New("peer reports no missing block part")
+	}
 	meta, err := g.blockStore.loadMeta(prs.Height)
 	if err != nil {
 		logger.Error("couldn't find a block meta", "error", err)
-		return
+		return err
 	}
 	err = g.ensurePeerPartSetHeader(meta.BlockID.PartSetHeader, prs.ProposalBlockPartSetHeader)
 	if err != nil {
 		logger.Error("block and peer part-set headers do not match", "error", err)
-		return
+		return err
 	}
 	part, err := g.blockStore.loadPart(prs.Height, index)
 	if err != nil {
 		logger.Error("couldn't find a block part", "part_index", index, "error", err)
-		return
+		return err
 	}
 	// Catch-up gossip: do NOT optimistically record the part as delivered.
 	//
@@ -227,13 +241,16 @@ func (g *msgGossiper) GossipBlockPartsForCatchup(
 	err = g.syncProposalBlockPart(ctx, part, prs.Height, meta.Round, false)
 	if err != nil {
 		logger.Error("failed to sync proposal block part to the peer", "error", err)
+		return err
 	}
+	return nil
 }
 
 // beginCatchupAttempt draws a send from the current catch-up pass, reporting
 // false once the pass is spent and until its retry interval elapses. A pass is
-// worth one send per part the peer reported missing when it opened, so a peer
-// that dropped those parts is served again every interval while it stays behind.
+// worth at most one send per part the peer reported missing when it opened, so a
+// peer that dropped those parts is served again every interval while it stays
+// behind. Any attempt that fails ends the pass early, via endCatchupPass.
 func (g *msgGossiper) beginCatchupAttempt(prs *cstypes.PeerRoundState) bool {
 	if prs.ProposalBlockParts == nil {
 		return false
@@ -267,6 +284,15 @@ func (g *msgGossiper) beginCatchupAttempt(prs *cstypes.PeerRoundState) bool {
 	return true
 }
 
+// endCatchupPass abandons the rest of the current pass and starts its retry
+// interval now. Without it a pass that cannot send costs a block-store read and
+// an error log on every gossip tick until its budget runs out, by which point
+// the deadline armed when it opened has long expired.
+func (g *msgGossiper) endCatchupPass() {
+	g.catchupRemaining = 0
+	g.catchupRetryAt = g.clock.Now().Add(catchupResendInterval)
+}
+
 // GossipCommit sends a commit message to the peer
 func (g *msgGossiper) GossipCommit(ctx context.Context, rs cstypes.RoundState, prs *cstypes.PeerRoundState) {
 	if prs.HasCommit {

internal/consensus/gossiper_test.go

@@ -18,6 +18,7 @@ import (
 	p2pmocks "github.com/dashpay/tenderdash/internal/p2p/mocks"
 	"github.com/dashpay/tenderdash/internal/state/mocks"
 	"github.com/dashpay/tenderdash/internal/test/factory"
+	"github.com/dashpay/tenderdash/libs/bits"
 	"github.com/dashpay/tenderdash/libs/log"
 	tmrand "github.com/dashpay/tenderdash/libs/rand"
 	tmcons "github.com/dashpay/tenderdash/proto/tendermint/consensus"
@@ -682,6 +683,111 @@ func (suite *GossiperSuiteTest) TestGossipBlockPartsForCatchupBudgetsMissingPart
 	suite.Require().Equal(2, sends, "the missing part is retried once the interval elapses")
 }
 
+// TestGossipBlockPartsForCatchupMismatchedHeaderEndsPass covers a pass that can
+// never produce a send. The peer supplies both the part-set header and the size
+// of the missing bit-array, and only their encoding is validated, so a pass
+// opened on a maximum-size array with a header matching no stored block would
+// otherwise charge a block-store read and an error log to every gossip tick for
+// the whole budget - and, its deadline having been armed when the pass opened,
+// reopen immediately afterwards.
+func (suite *GossiperSuiteTest) TestGossipBlockPartsForCatchupMismatchedHeaderEndsPass() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stored := types.NewPartSetFromData(tmrand.Bytes(100), 100)
+	blockMeta := types.BlockMeta{BlockID: types.BlockID{PartSetHeader: stored.Header()}}
+
+	// Largest pass the peer can ask for: every bit of a maximum-length array
+	// reported missing, under a header no stored block can match.
+	bogus := types.NewPartSetFromData(tmrand.Bytes(100), 100)
+	suite.Require().False(bogus.Header().Equals(stored.Header()))
+	suite.ps.PRS = cstypes.PeerRoundState{
+		Height:                     999,
+		Round:                      0,
+		ProposalBlockParts:         bits.NewBitArray(int(types.MaxBlockPartsCount)),
+		ProposalBlockPartSetHeader: bogus.Header(),
+	}
+
+	// The data channel has no expectations: a mismatched header must send nothing.
+	metaReads := 0
+	suite.blockStore.On("LoadBlockMeta", int64(999)).
+		Run(func(_ mock.Arguments) { metaReads++ }).
+		Return(&blockMeta)
+
+	for range 50 {
+		suite.gossiper.GossipBlockPartsForCatchup(ctx, cstypes.RoundState{}, suite.ps.GetRoundState())
+	}
+	suite.Require().Equal(1, metaReads,
+		"a pass that cannot send must cost one attempt per interval, not one per tick")
+
+	suite.clock.Advance(catchupResendInterval)
+	for range 50 {
+		suite.gossiper.GossipBlockPartsForCatchup(ctx, cstypes.RoundState{}, suite.ps.GetRoundState())
+	}
+	suite.Require().Equal(2, metaReads, "each elapsed interval allows exactly one further attempt")
+}
+
+// TestGossipBlockPartsForCatchupMalformedHeaderDoesNotFundSends pins the budget
+// to a pass that got as far as sending. A pass is deliberately not re-derived
+// from the peer's bit-array once open, so a budget opened against an unusable
+// header would otherwise stay spendable when the peer replaces its state, at the
+// same height, with the stored block's real header and a single missing part:
+// the whole inflated budget then lands as duplicate sends of that one part.
+func (suite *GossiperSuiteTest) TestGossipBlockPartsForCatchupMalformedHeaderDoesNotFundSends() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	partSet := types.NewPartSetFromData(tmrand.Bytes(100), 100)
+	suite.Require().Equal(uint32(1), partSet.Total())
+	blockMeta := types.BlockMeta{BlockID: types.BlockID{PartSetHeader: partSet.Header()}}
+
+	bogus := types.NewPartSetFromData(tmrand.Bytes(100), 100)
+	suite.Require().False(bogus.Header().Equals(partSet.Header()))
+
+	// Open the largest pass available, under a header that cannot match.
+	suite.ps.PRS = cstypes.PeerRoundState{
+		Height:                     999,
+		Round:                      0,
+		ProposalBlockParts:         bits.NewBitArray(int(types.MaxBlockPartsCount)),
+		ProposalBlockPartSetHeader: bogus.Header(),
+	}
+
+	suite.blockStore.On("LoadBlockMeta", int64(999)).Return(&blockMeta)
+	suite.blockStore.On("LoadBlockPart", int64(999), 0).Return(partSet.GetPart(0))
+
+	sends := 0
+	suite.dataCh.On("Send", mock.Anything, mock.Anything).
+		Run(func(args mock.Arguments) {
+			env, ok := args.Get(1).(p2p.Envelope)
+			if !ok {
+				return
+			}
+			if _, ok := env.Message.(*tmcons.BlockPart); ok {
+				sends++
+			}
+		}).
+		Return(nil)
+
+	suite.gossiper.GossipBlockPartsForCatchup(ctx, cstypes.RoundState{}, suite.ps.GetRoundState())
+
+	// Same height, now with the real header and one part outstanding.
+	suite.ps.PRS.ProposalBlockPartSetHeader = partSet.Header()
+	suite.ps.PRS.ProposalBlockParts = partSet.BitArray().Not()
+
+	for range 50 {
+		suite.gossiper.GossipBlockPartsForCatchup(ctx, cstypes.RoundState{}, suite.ps.GetRoundState())
+	}
+	suite.Require().Equal(0, sends,
+		"a budget opened against an unusable header must not fund sends after the peer swaps it out")
+
+	// The next interval opens a fresh pass, budgeted on what the peer now reports.
+	suite.clock.Advance(catchupResendInterval)
+	for range 50 {
+		suite.gossiper.GossipBlockPartsForCatchup(ctx, cstypes.RoundState{}, suite.ps.GetRoundState())
+	}
+	suite.Require().Equal(1, sends, "the fresh pass is worth the single part the peer now reports missing")
+}
+
 // TestGossipBlockPartsForCatchupPeerHasEveryPart covers the peer reporting a
 // complete part set: there is nothing to send, and no pass may be opened.
 func (suite *GossiperSuiteTest) TestGossipBlockPartsForCatchupPeerHasEveryPart() {