Retry fetch origin data column sidecars. (#15634)

* `convertToAddrInfo`: Add peer details on error.

* `fetchOriginColumns`: Remove unused argument.

* `fetchOriginColumns`: Fix typo

* `isSidecarIndexRequested`: Log requested indices.

* `fetchOriginColumns`: Retry.

* Add changelog.

* Fix Preston comment.

* `custodyGroupCountFromPeerENR`: Add agent on error messages.

* Update beacon-chain/sync/initial-sync/service.go

Co-authored-by: Preston Van Loon <pvanloon@offchainlabs.com>

* Fix `TestCustodyGroupCountFromPeer`.

* `s.fetchOriginColumns`: Use `maxAttempts` and `delay` as parameters to ease unit testing.

* Implement `TestFetchOriginColumns`.

* `SendDataColumnSidecarsByRangeRequest` and `SendDataColumnSidecarsByRootRequest`: Add option to downscore the peer on RPC error.

* `fetchOriginColumns`: Remove max attempts, and downscore peers on RPC fault.

---------

Co-authored-by: Preston Van Loon <pvanloon@offchainlabs.com>

beacon-chain/sync/initial-sync/service_test.go

@@ -2,6 +2,7 @@ package initialsync
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -13,8 +14,12 @@ import (
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/db/filesystem"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/db/kv"
 	dbtest "github.com/OffchainLabs/prysm/v6/beacon-chain/db/testing"
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p"
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/peers"
 	p2ptest "github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/testing"
+	testp2p "github.com/OffchainLabs/prysm/v6/beacon-chain/p2p/testing"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/startup"
+	prysmSync "github.com/OffchainLabs/prysm/v6/beacon-chain/sync"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/verification"
 	"github.com/OffchainLabs/prysm/v6/cmd/beacon-chain/flags"
 	fieldparams "github.com/OffchainLabs/prysm/v6/config/fieldparams"
@@ -22,10 +27,14 @@ import (
 	"github.com/OffchainLabs/prysm/v6/consensus-types/blocks"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/primitives"
 	eth "github.com/OffchainLabs/prysm/v6/proto/prysm/v1alpha1"
+	ethpb "github.com/OffchainLabs/prysm/v6/proto/prysm/v1alpha1"
 	"github.com/OffchainLabs/prysm/v6/testing/assert"
 	"github.com/OffchainLabs/prysm/v6/testing/require"
 	"github.com/OffchainLabs/prysm/v6/testing/util"
 	"github.com/OffchainLabs/prysm/v6/time/slots"
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/crypto"
+	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/paulbellamy/ratecounter"
 	logTest "github.com/sirupsen/logrus/hooks/test"
@@ -663,3 +672,147 @@ func TestFetchOriginSidecars(t *testing.T) {
 		}
 	})
 }
+
+func TestFetchOriginColumns(t *testing.T) {
+	// Load the trusted setup.
+	err := kzg.Start()
+	require.NoError(t, err)
+
+	// Setup test environment
+	params.SetupTestConfigCleanup(t)
+	cfg := params.BeaconConfig().Copy()
+	cfg.FuluForkEpoch = 0
+	params.OverrideBeaconConfig(cfg)
+
+	const (
+		delay     = 0
+		blobCount = 1
+	)
+
+	t.Run("block has no commitments", func(t *testing.T) {
+		service := new(Service)
+
+		// Create a block with no blob commitments
+		block := util.NewBeaconBlockFulu()
+		signedBlock, err := blocks.NewSignedBeaconBlock(block)
+		require.NoError(t, err)
+		roBlock, err := blocks.NewROBlock(signedBlock)
+		require.NoError(t, err)
+
+		err = service.fetchOriginColumns(roBlock, delay)
+		require.NoError(t, err)
+	})
+
+	t.Run("FetchDataColumnSidecars succeeds immediately", func(t *testing.T) {
+		storage := filesystem.NewEphemeralDataColumnStorage(t)
+		p2p := p2ptest.NewTestP2P(t)
+
+		service := &Service{
+			cfg: &Config{
+				P2P:               p2p,
+				DataColumnStorage: storage,
+			},
+		}
+
+		// Create a block with blob commitments and sidecars
+		roBlock, _, verifiedSidecars := util.GenerateTestFuluBlockWithSidecars(t, blobCount)
+
+		// Store all sidecars in advance so FetchDataColumnSidecars succeeds immediately
+		err := storage.Save(verifiedSidecars)
+		require.NoError(t, err)
+
+		err = service.fetchOriginColumns(roBlock, delay)
+		require.NoError(t, err)
+	})
+
+	t.Run("first attempt to FetchDataColumnSidecars fails but second attempt succeeds", func(t *testing.T) {
+		numberOfCustodyGroups := params.BeaconConfig().NumberOfCustodyGroups
+		storage := filesystem.NewEphemeralDataColumnStorage(t)
+
+		// Custody columns with this private key and 4-cgc: 31, 81, 97, 105
+		privateKeyBytes := [32]byte{1}
+		privateKey, err := crypto.UnmarshalSecp256k1PrivateKey(privateKeyBytes[:])
+		require.NoError(t, err)
+
+		protocol := fmt.Sprintf("%s/ssz_snappy", p2p.RPCDataColumnSidecarsByRangeTopicV1)
+
+		p2p, other := testp2p.NewTestP2P(t), testp2p.NewTestP2P(t, libp2p.Identity(privateKey))
+		p2p.Peers().SetConnectionState(other.PeerID(), peers.Connected)
+		p2p.Connect(other)
+
+		p2p.Peers().SetChainState(other.PeerID(), &ethpb.StatusV2{
+			HeadSlot: 5,
+		})
+
+		other.ENR().Set(peerdas.Cgc(numberOfCustodyGroups))
+		p2p.Peers().UpdateENR(other.ENR(), other.PeerID())
+
+		expectedRequest := &ethpb.DataColumnSidecarsByRangeRequest{
+			StartSlot: 0,
+			Count:     1,
+			Columns:   []uint64{1, 17, 19, 42, 75, 87, 102, 117},
+		}
+
+		clock := startup.NewClock(time.Now(), [fieldparams.RootLength]byte{})
+
+		gs := startup.NewClockSynchronizer()
+		err = gs.SetClock(startup.NewClock(time.Unix(4113849600, 0), [fieldparams.RootLength]byte{}))
+		require.NoError(t, err)
+
+		waiter := verification.NewInitializerWaiter(gs, nil, nil)
+		initializer, err := waiter.WaitForInitializer(t.Context())
+		require.NoError(t, err)
+
+		newDataColumnsVerifier := newDataColumnsVerifierFromInitializer(initializer)
+
+		// Create a block with blob commitments and sidecars
+		roBlock, _, verifiedRoSidecars := util.GenerateTestFuluBlockWithSidecars(t, blobCount)
+
+		ctxMap, err := prysmSync.ContextByteVersionsForValRoot(params.BeaconConfig().GenesisValidatorsRoot)
+		require.NoError(t, err)
+
+		service := &Service{
+			ctx:                    t.Context(),
+			clock:                  clock,
+			newDataColumnsVerifier: newDataColumnsVerifier,
+			cfg: &Config{
+				P2P:               p2p,
+				DataColumnStorage: storage,
+			},
+			ctxMap: ctxMap,
+		}
+
+		// Do not respond any sidecar on the first attempt, and respond everything requested on the second one.
+		firstAttempt := true
+		other.SetStreamHandler(protocol, func(stream network.Stream) {
+			actualRequest := new(ethpb.DataColumnSidecarsByRangeRequest)
+			err := other.Encoding().DecodeWithMaxLength(stream, actualRequest)
+			assert.NoError(t, err)
+			assert.DeepEqual(t, expectedRequest, actualRequest)
+
+			if firstAttempt {
+				firstAttempt = false
+				err = stream.CloseWrite()
+				assert.NoError(t, err)
+				return
+			}
+
+			for _, column := range actualRequest.Columns {
+				err = prysmSync.WriteDataColumnSidecarChunk(stream, clock, other.Encoding(), verifiedRoSidecars[column].DataColumnSidecar)
+				assert.NoError(t, err)
+			}
+
+			err = stream.CloseWrite()
+			assert.NoError(t, err)
+		})
+
+		err = service.fetchOriginColumns(roBlock, delay)
+		require.NoError(t, err)
+
+		// Check all corresponding sidecars are saved in the store.
+		summary := storage.Summary(roBlock.Root())
+		for _, index := range expectedRequest.Columns {
+			require.Equal(t, true, summary.HasIndex(index))
+		}
+	})
+}