fix: disconnect peers that send invalid chunks during state sync (#2814)

## Summary

- When the ABCI application rejects a chunk sender via `RejectSenders`
during state sync, the peer was only blacklisted in the in-memory
snapshot pool (`snapshots.RejectPeer`) but remained connected at the P2P
layer. A malicious peer could keep sending poisoned `ChunkResponse`
messages, creating an infinite retry loop.
- Adds an `onPeerRejected` callback to the syncer that the reactor wires
up to call `StopPeerForError`, disconnecting the rejected peer at the
P2P layer.
- Adds `TestSyncer_applyChunks_onPeerRejected` to verify the callback is
invoked with the correct peer ID.

## Test plan

- [x] `go test -v -run TestSyncer_applyChunks_onPeerRejected
./statesync/`
- [x] `go test -v ./statesync/...` (all 37 tests pass)
- [x] `go test -race ./statesync/...`

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
(cherry picked from commit c4ce459)

Author: 2 people

statesync/reactor.go

@@ -275,6 +275,12 @@ func (r *Reactor) Sync(stateProvider StateProvider, discoveryTime time.Duration)
 	}
 	r.metrics.Syncing.Set(1)
 	r.syncer = newSyncer(r.cfg, r.Logger, r.conn, r.connQuery, stateProvider, r.tempDir)
+	r.syncer.setOnPeerRejected(func(peerID p2p.ID) {
+		peer := r.Switch.Peers().Get(peerID)
+		if peer != nil {
+			r.Switch.StopPeerForError(peer, "chunk sender rejected by ABCI application", r.String())
+		}
+	})
 	r.mtx.Unlock()
 
 	hook := func() {

statesync/syncer.go

@@ -60,8 +60,9 @@ type syncer struct {
 	chunkFetchers int32
 	retryTimeout  time.Duration
 
-	mtx    cmtsync.RWMutex
-	chunks *chunkQueue
+	mtx            cmtsync.RWMutex
+	chunks         *chunkQueue
+	onPeerRejected func(p2p.ID)
 }
 
 // newSyncer creates a new syncer.
@@ -86,6 +87,13 @@ func newSyncer(
 	}
 }
 
+// setOnPeerRejected sets a callback that is invoked when a peer is rejected by the ABCI
+// application during chunk application. This allows the caller to disconnect the peer at the
+// P2P layer, preventing them from continuing to send invalid chunks.
+func (s *syncer) setOnPeerRejected(fn func(p2p.ID)) {
+	s.onPeerRejected = fn
+}
+
 // AddChunk adds a chunk to the chunk queue, if any. It returns false if the chunk has already
 // been added to the queue, or an error if there's no sync in progress.
 func (s *syncer) AddChunk(chunk *chunk) (bool, error) {
@@ -403,8 +411,12 @@ func (s *syncer) applyChunks(chunks *chunkQueue) error {
 		// Reject any senders as requested by the app
 		for _, sender := range resp.RejectSenders {
 			if sender != "" {
-				s.snapshots.RejectPeer(p2p.ID(sender))
-				err := chunks.DiscardSender(p2p.ID(sender))
+				peerID := p2p.ID(sender)
+				s.snapshots.RejectPeer(peerID)
+				if s.onPeerRejected != nil {
+					s.onPeerRejected(peerID)
+				}
+				err := chunks.DiscardSender(peerID)
 				if err != nil {
 					return fmt.Errorf("failed to reject sender: %w", err)
 				}

statesync/syncer_test.go

@@ -636,6 +636,67 @@ func TestSyncer_applyChunks_RejectSenders(t *testing.T) {
 	}
 }
 
+func TestSyncer_applyChunks_onPeerRejected(t *testing.T) {
+	connQuery := &proxymocks.AppConnQuery{}
+	connSnapshot := &proxymocks.AppConnSnapshot{}
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+	cfg := config.DefaultStateSyncConfig()
+	syncer := newSyncer(*cfg, log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
+
+	// Track which peer IDs were reported via the callback.
+	var rejectedPeers []p2p.ID
+	syncer.setOnPeerRejected(func(peerID p2p.ID) {
+		rejectedPeers = append(rejectedPeers, peerID)
+	})
+
+	peerA := simplePeer("a")
+	peerB := simplePeer("b")
+	peerC := simplePeer("c")
+
+	s1 := &snapshot{Height: 1, Format: 1, Chunks: 3}
+	_, err := syncer.AddSnapshot(peerA, s1)
+	require.NoError(t, err)
+	_, err = syncer.AddSnapshot(peerB, s1)
+	require.NoError(t, err)
+	_, err = syncer.AddSnapshot(peerC, s1)
+	require.NoError(t, err)
+
+	chunks, err := newChunkQueue(s1, "")
+	require.NoError(t, err)
+	added, err := chunks.Add(&chunk{Height: 1, Format: 1, Index: 0, Chunk: []byte{0}, Sender: peerA.ID()})
+	require.True(t, added)
+	require.NoError(t, err)
+	added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 1, Chunk: []byte{1}, Sender: peerB.ID()})
+	require.True(t, added)
+	require.NoError(t, err)
+	added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 2, Chunk: []byte{2}, Sender: peerC.ID()})
+	require.True(t, added)
+	require.NoError(t, err)
+
+	// First two chunks are accepted, the third triggers rejection of sender "b".
+	connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
+		Index: 0, Chunk: []byte{0}, Sender: "a",
+	}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+	connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
+		Index: 1, Chunk: []byte{1}, Sender: "b",
+	}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+	connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
+		Index: 2, Chunk: []byte{2}, Sender: "c",
+	}).Once().Return(&abci.ResponseApplySnapshotChunk{
+		Result:        abci.ResponseApplySnapshotChunk_ACCEPT,
+		RejectSenders: []string{string(peerB.ID())},
+	}, nil)
+
+	err = syncer.applyChunks(chunks)
+	require.NoError(t, err)
+
+	require.Len(t, rejectedPeers, 1)
+	assert.Equal(t, p2p.ID("b"), rejectedPeers[0])
+	connSnapshot.AssertExpectations(t)
+}
+
 func TestSyncer_verifyApp(t *testing.T) {
 	boom := errors.New("boom")
 	const appVersion = 9