cl, sentinel: quiet data column sidecar misses (#21686)

## Summary

Fixes the data column sidecar miss path so expected misses no longer
produce noisy download logs, while keeping the wire behavior compatible
with peers that signal an empty multi-chunk response by closing without
a response code.

Closes #21670

## Details

- Return structured req/resp `resource unavailable` responses for
by-root misses and valid pre-Fulu by-root requests, while validating
malformed pre-Fulu by-root requests first so they return `invalid
request`.
- Restore by-range empty-success responses to a zero-byte close,
including zero-count, all-pre-Fulu, all-future, and missing-sidecar
ranges.
- In the httpreqresp client bridge, synthesize success code 0 with an
empty body when a negotiated multi-chunk protocol returns `io.EOF`
before any response-code byte. Single-chunk EOFs, partial reads, and
stream errors still surface as HTTP 400.
- Keep handler `responseErr` propagation for real storage/write errors.
Across the Sentinel HTTP/gRPC boundary those transport failures can
still be flattened by gRPC, while structured peer response codes are
surfaced as typed `PeerResponseError` values.
- Bound peer error-message decoding to a 10-byte varint prefix and a
256-byte decoded message.
- Downgrade expected data column sidecar misses to trace while keeping
other download errors at debug.

## Validation

- `go test ./cl/sentinel/httpreqresp ./cl/sentinel/service
./cl/sentinel/handlers ./cl/das -count=1`
- `make lint`

Author: domiwei

cl/das/peer_das.go

@@ -19,6 +19,7 @@ import (
 	"github.com/erigontech/erigon/cl/phase1/core/state/lru"
 	gossipmgr "github.com/erigontech/erigon/cl/phase1/network/gossip"
 	"github.com/erigontech/erigon/cl/rpc"
+	"github.com/erigontech/erigon/cl/sentinel/httpreqresp"
 	"github.com/erigontech/erigon/cl/utils/eth_clock"
 	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/common/crypto/kzg"
@@ -822,6 +823,10 @@ mainloop:
 			}
 		case result := <-resultChan:
 			if result.err != nil {
+				if isExpectedColumnDownloadMiss(result.err) {
+					log.Trace("column sidecars unavailable from peer", "pid", result.pid, "err", result.err)
+					continue
+				}
 				log.Debug("failed to download columns from peer", "pid", result.pid, "err", result.err)
 				//d.rpc.BanPeer(result.pid)
 				continue
@@ -941,6 +946,17 @@ mainloop:
 	return nil
 }
 
+func isExpectedColumnDownloadMiss(err error) bool {
+	if err == nil {
+		return false
+	}
+	var peerErr *httpreqresp.PeerResponseError
+	if errors.As(err, &peerErr) {
+		return peerErr.Code == httpreqresp.ResponseCodeResourceUnavailable
+	}
+	return false
+}
+
 type downloadTableEntry struct {
 	blockRoot common.Hash
 	slot      uint64

cl/das/peer_das_test.go

@@ -0,0 +1,46 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package das
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/erigontech/erigon/cl/sentinel/httpreqresp"
+)
+
+func TestIsExpectedColumnDownloadMiss(t *testing.T) {
+	require.False(t, isExpectedColumnDownloadMiss(nil))
+	require.True(t, isExpectedColumnDownloadMiss(&httpreqresp.PeerResponseError{
+		Code: httpreqresp.ResponseCodeResourceUnavailable,
+	}))
+	require.True(t, isExpectedColumnDownloadMiss(fmt.Errorf("column miss: %w", &httpreqresp.PeerResponseError{
+		Code: httpreqresp.ResponseCodeResourceUnavailable,
+	})))
+	require.False(t, isExpectedColumnDownloadMiss(&httpreqresp.PeerResponseError{
+		Code:    httpreqresp.ResponseCodeServerError,
+		Message: "broken",
+	}))
+	require.False(t, isExpectedColumnDownloadMiss(&httpreqresp.HTTPError{
+		StatusCode: 400,
+		Body:       "Read Code: EOF",
+	}))
+	require.False(t, isExpectedColumnDownloadMiss(errors.New("peer error code: 2 (server error). Error message: broken")))
+}

cl/sentinel/handlers/data_cloumn_sidecar.go

@@ -13,40 +13,61 @@ import (
 	"github.com/libp2p/go-libp2p/core/network"
 )
 
+var errInvalidDataColumnIndex = errors.New("invalid column index")
+
+func writeDataColumnSidecarsEmptySuccess(s network.Stream) error {
+	return nil
+}
+
 func (c *ConsensusHandlers) dataColumnSidecarsByRangeHandler(s network.Stream) error {
 	curEpoch := c.ethClock.GetCurrentEpoch()
-	if curEpoch < c.beaconConfig.FuluForkEpoch {
-		return nil
-	}
 
 	// Use current epoch's version for decoding (supports Fulu and GLOAS)
 	version := c.beaconConfig.GetCurrentStateVersion(curEpoch)
 	req := &cltypes.ColumnSidecarsByRangeRequest{}
 	if err := ssz_snappy.DecodeAndReadNoForkDigest(s, req, version); err != nil {
-		return err
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
+	}
+	if curEpoch < c.beaconConfig.FuluForkEpoch {
+		return writeDataColumnSidecarsEmptySuccess(s)
 	}
 
 	// check params.
 	var (
-		endSlot   = req.StartSlot + req.Count
-		startSlot = max(req.StartSlot, c.beaconConfig.FuluForkEpoch*c.beaconConfig.SlotsPerEpoch)
+		fuluStartSlot = c.beaconConfig.FuluForkEpoch * c.beaconConfig.SlotsPerEpoch
+		endSlot       = req.StartSlot + req.Count
 	)
-	if endSlot-startSlot > c.beaconConfig.MinEpochsForDataColumnSidecarsRequests*c.beaconConfig.SlotsPerEpoch {
-		return errors.New("request range is too large")
+	if endSlot < req.StartSlot {
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
+	}
+	if req.Columns.Length() > int(c.beaconConfig.NumberOfColumns) {
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
 	}
-	solid.RangeErr(req.Columns, func(index int, columnIndex uint64, length int) error {
+	if err := solid.RangeErr(req.Columns, func(index int, columnIndex uint64, length int) error {
 		if columnIndex >= c.beaconConfig.NumberOfColumns {
-			return errors.New("invalid column index")
+			return errInvalidDataColumnIndex
 		}
 		return nil
-	})
+	}); err != nil {
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
+	}
+	if req.Count == 0 || req.Columns.Length() == 0 || endSlot <= fuluStartSlot {
+		return writeDataColumnSidecarsEmptySuccess(s)
+	}
+	startSlot := max(req.StartSlot, fuluStartSlot)
+	if endSlot-startSlot > c.beaconConfig.MinEpochsForDataColumnSidecarsRequests*c.beaconConfig.SlotsPerEpoch {
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
+	}
 
 	// Consume additional rate-limit tokens: slots × columns per slot, capped at config max.
-	if cost := min(int(req.Count)*req.Columns.Length(), int(c.beaconConfig.MaxRequestDataColumnSidecars)) - 1; !c.consumeRateLimit(s, cost) {
+	if cost := dataColumnSidecarsRequestCost(endSlot-startSlot, uint64(req.Columns.Length()), c.beaconConfig.MaxRequestDataColumnSidecars); !c.consumeRateLimit(s, cost) {
 		return nil
 	}
 
 	curSlot := c.ethClock.GetCurrentSlot()
+	if startSlot > curSlot {
+		return writeDataColumnSidecarsEmptySuccess(s)
+	}
 
 	tx, err := c.indiciesDB.BeginRo(c.ctx)
 	if err != nil {
@@ -55,6 +76,7 @@ func (c *ConsensusHandlers) dataColumnSidecarsByRangeHandler(s network.Stream) e
 	defer tx.Rollback()
 
 	count := 0
+	var responseErr error
 	for slot := startSlot; slot < endSlot; slot++ {
 		if slot > curSlot {
 			// slot is in the future
@@ -84,6 +106,7 @@ func (c *ConsensusHandlers) dataColumnSidecarsByRangeHandler(s network.Stream) e
 			exists, err := c.dataColumnStorage.ColumnSidecarExists(c.ctx, slot, blockRoot, int64(columnIndex))
 			if err != nil {
 				log.Debug("failed to check if data column sidecar exists", "error", err)
+				responseErr = err
 				return false
 			}
 			if !exists {
@@ -94,56 +117,83 @@ func (c *ConsensusHandlers) dataColumnSidecarsByRangeHandler(s network.Stream) e
 			forkDigest, err := c.ethClock.ComputeForkDigest(slot / c.beaconConfig.SlotsPerEpoch)
 			if err != nil {
 				log.Debug("failed to compute fork digest", "error", err)
+				responseErr = err
 				return false
 			}
 			if _, err := s.Write([]byte{SuccessfulResponsePrefix}); err != nil {
 				log.Debug("failed to write success byte", "error", err)
+				responseErr = err
 				return false
 			}
 
 			if _, err := s.Write(forkDigest[:]); err != nil {
 				log.Debug("failed to write fork digest", "error", err)
+				responseErr = err
 				return false
 			}
 
 			if err := c.dataColumnStorage.WriteStream(s, slot, blockRoot, columnIndex); err != nil {
 				log.Debug("failed to write stream data column sidecar", "error", err)
+				responseErr = err
 				return false
 			}
 			count++
 			return true
 		})
+		if responseErr != nil {
+			break
+		}
 		if count >= int(c.beaconConfig.MaxRequestDataColumnSidecars) {
 			// max number of sidecars reached
 			break
 		}
 	}
-
+	if responseErr != nil {
+		return responseErr
+	}
+	if count == 0 {
+		return writeDataColumnSidecarsEmptySuccess(s)
+	}
 	return nil
 }
 
 func (c *ConsensusHandlers) dataColumnSidecarsByRootHandler(s network.Stream) error {
 	curEpoch := c.ethClock.GetCurrentEpoch()
-	if curEpoch < c.beaconConfig.FuluForkEpoch {
-		return nil
-	}
 
 	// Use current epoch's version for decoding (supports Fulu and GLOAS)
 	version := c.beaconConfig.GetCurrentStateVersion(curEpoch)
 	req := solid.NewDynamicListSSZ[*cltypes.DataColumnsByRootIdentifier](int(c.beaconConfig.MaxRequestBlocksDeneb))
 	if err := ssz_snappy.DecodeAndReadNoForkDigest(s, req, version); err != nil {
-		return err
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
 	}
 	if req.Len() > int(c.beaconConfig.MaxRequestBlocksDeneb) {
-		return errors.New("request is too large")
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
 	}
 
 	// Consume additional rate-limit tokens: sum of column counts across all roots.
 	totalColumns := 0
 	for i := 0; i < req.Len(); i++ {
-		totalColumns += req.Get(i).Columns.Length()
+		columns := req.Get(i).Columns
+		if columns.Length() > int(c.beaconConfig.NumberOfColumns) {
+			return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
+		}
+		if err := solid.RangeErr(columns, func(index int, columnIndex uint64, length int) error {
+			if columnIndex >= c.beaconConfig.NumberOfColumns {
+				return errInvalidDataColumnIndex
+			}
+			return nil
+		}); err != nil {
+			return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
+		}
+		totalColumns += columns.Length()
+	}
+	if curEpoch < c.beaconConfig.FuluForkEpoch {
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, ResourceUnavailablePrefix)
 	}
-	if cost := min(totalColumns, int(c.beaconConfig.MaxRequestDataColumnSidecars)) - 1; !c.consumeRateLimit(s, cost) {
+	if totalColumns == 0 {
+		return writeDataColumnSidecarsEmptySuccess(s)
+	}
+	if cost := dataColumnSidecarsRequestCost(1, uint64(totalColumns), c.beaconConfig.MaxRequestDataColumnSidecars); !c.consumeRateLimit(s, cost) {
 		return nil
 	}
 
@@ -157,6 +207,7 @@ func (c *ConsensusHandlers) dataColumnSidecarsByRootHandler(s network.Stream) er
 	defer tx.Rollback()
 
 	count := 0
+	var responseErr error
 	for i := 0; i < req.Len(); i++ {
 		id := req.Get(i)
 		blockRoot := id.BlockRoot
@@ -188,14 +239,11 @@ func (c *ConsensusHandlers) dataColumnSidecarsByRootHandler(s network.Stream) er
 				// max number of sidecars reached
 				return false
 			}
-			if columnIndex >= c.beaconConfig.NumberOfColumns {
-				// skip invalid column index
-				return true
-			}
 
 			exists, err := c.dataColumnStorage.ColumnSidecarExists(c.ctx, *slot, blockRoot, int64(columnIndex))
 			if err != nil {
 				log.Debug("failed to check if data column sidecar exists", "error", err)
+				responseErr = err
 				return false
 			}
 			if !exists {
@@ -206,25 +254,52 @@ func (c *ConsensusHandlers) dataColumnSidecarsByRootHandler(s network.Stream) er
 			forkDigest, err := c.ethClock.ComputeForkDigest(*slot / c.beaconConfig.SlotsPerEpoch)
 			if err != nil {
 				log.Debug("failed to compute fork digest", "error", err)
+				responseErr = err
 				return false
 			}
 			if _, err := s.Write([]byte{SuccessfulResponsePrefix}); err != nil {
 				log.Debug("failed to write success byte", "error", err)
+				responseErr = err
 				return false
 			}
 
 			if _, err := s.Write(forkDigest[:]); err != nil {
 				log.Debug("failed to write fork digest", "error", err)
+				responseErr = err
 				return false
 			}
 
 			if err := c.dataColumnStorage.WriteStream(s, *slot, blockRoot, columnIndex); err != nil {
 				log.Debug("failed to write stream data column sidecar", "error", err)
+				responseErr = err
 				return false
 			}
 			count++
 			return true
 		})
+		if responseErr != nil {
+			break
+		}
+	}
+	if responseErr != nil {
+		return responseErr
+	}
+	if count == 0 {
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, ResourceUnavailablePrefix)
 	}
 	return nil
 }
+
+func dataColumnSidecarsRequestCost(slots, columns, maxSidecars uint64) int {
+	if slots == 0 || columns == 0 || maxSidecars == 0 {
+		return 0
+	}
+	if slots > maxSidecars/columns {
+		return int(maxSidecars) - 1
+	}
+	sidecars := slots * columns
+	if sidecars > maxSidecars {
+		sidecars = maxSidecars
+	}
+	return int(sidecars) - 1
+}