increase crawl timeout

**What type of PR is this?**
Feature + Bugfix

**What does this PR do? Why is it needed?**
Starting at Fusaka, the beacon node can pull blobs with the
`engine_getBlobsV2` API from the execution layer.
This reduces by a lot the burden on the beacon node. However, the beacon
node should be able to work 100% correctly without this execution layer
help.

This PR introduces the `--disable-get-blobs-v2` flag to simulate a 0%
success rate of this engine API.

This PR also fixes:
- https://github.com/OffchainLabs/prysm/issues/16171

Please read commit by commit with commit messages.

**How to test it:**
For the `--disable-get-blobs-v2` part:

Run the beacon node with the `--disable-get-blobs-v2` flag in DEBUG
mode.
For every block with commitments, the following log should be displayed:
```
[2025-12-19 15:36:25.49] DEBUG sync: No data column sidecars constructed from the execution client ...
```

And the following log should **never** be displayed:
```
[2026-01-05 10:19:00.55] DEBUG sync: Constructed data column sidecars from the execution client count=...
```

For the #16171 part:
- No ERROR log showed in the linked issue should never be displayed.

**Acknowledgements**
- [x] I have read
[CONTRIBUTING.md](https://github.com/prysmaticlabs/prysm/blob/develop/CONTRIBUTING.md).
- [x] I have included a uniquely named [changelog fragment
file](https://github.com/prysmaticlabs/prysm/blob/develop/CONTRIBUTING.md#maintaining-changelogmd).
- [x] I have added a description with sufficient context for reviewers
to understand this PR.
- [x] I have tested that my changes work as expected and I added a
testing plan to the PR description (if applicable).

.github/workflows/check-logs.yml

@@ -0,0 +1,23 @@
+name: Check log.go files
+on: [ pull_request ]
+
+jobs:
+  check-logs:
+    runs-on: ubuntu-4
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Set up Go 1.25.1
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.25.1'
+
+      - name: Install ripgrep
+        run: sudo apt-get install -y ripgrep
+
+      - name: Check log.go files
+        run: ./hack/check-logs.sh

.gitignore

@@ -38,9 +38,13 @@ metaData
 
 # execution API authentication
 jwt.hex
+execution/
 
 # manual testing
 tmp
 
 # spectest coverage reports
 report.txt
+
+# execution client data
+execution/

CHANGELOG.md

@@ -4,6 +4,47 @@ All notable changes to this project will be documented in this file.
 
 The format is based on Keep a Changelog, and this project adheres to Semantic Versioning.
 
+## [v7.1.1](https://github.com/prysmaticlabs/prysm/compare/v7.1.0...v7.1.1) - 2025-12-18
+
+Release highlights:
+
+- Fixed potential deadlock scenario in data column batch verification
+- Improved processing and metrics for cells and proofs
+
+We are aware of [an issue](https://github.com/OffchainLabs/prysm/issues/16160) where Prysm struggles to sync from an out of sync state. We will have another release before the end of the year to address this issue.
+
+Our postmortem document from the December 4th mainnet issue has been published on our [documentation site](https://prysm.offchainlabs.com/docs/misc/mainnet-postmortems/)
+
+### Added
+
+- Track the dependent root of the latest finalized checkpoint in forkchoice. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16103)
+- Proposal design document to implement graffiti. Currently it is empty by default and the idea is to have it of the form GE168dPR63af. [[PR]](https://github.com/prysmaticlabs/prysm/pull/15983)
+- Add support for detecting and logging per address reachability via libp2p AutoNAT v2. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16100)
+- Static analyzer that ensures each `httputil.HandleError` call is followed by a `return` statement. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16134)
+- Prometheus histogram `cells_and_proofs_from_structured_computation_milliseconds` to track computation time for cells and proofs from structured blobs. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16115)
+- Prometheus histogram `get_blobs_v2_latency_milliseconds` to track RPC latency for `getBlobsV2` calls to the execution layer. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16115)
+
+### Changed
+
+- Optimise migratetocold by not doing brute force for loop. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16101)
+- e2e sync committee evaluator now skips the first slot after startup, we already skip the fork epoch for checks here, this skip only applies on startup, due to altair always from 0 and validators need to warm up. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16145)
+- Run `ComputeCellsAndProofsFromFlat` in parallel to improve performance when computing cells and proofs. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16115)
+- Run `ComputeCellsAndProofsFromStructured` in parallel to improve performance when computing cells and proofs. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16115)
+
+### Removed
+
+- Unnecessary copy is removed from Eth1DataHasEnoughSupport. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16118)
+
+### Fixed
+
+- Incorrect constructor return type [#16084](https://github.com/OffchainLabs/prysm/pull/16084). [[PR]](https://github.com/prysmaticlabs/prysm/pull/16084)
+- Fixed possible race when validating two attestations at the same time. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16105)
+- Fix missing return after version header check in SubmitAttesterSlashingsV2. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16126)
+- Fix deadlock in data column gossip KZG batch verification when a caller times out preventing result delivery. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16141)
+- Fixed replay state issue in rest api caused by attester and sync committee duties endpoints. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16136)
+- Do not error when committee has been computed correctly but updating the cache failed. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16142)
+- Prevent blocked sends to the KZG batch verifier when the caller context is already canceled, avoiding useless queueing and potential hangs. [[PR]](https://github.com/prysmaticlabs/prysm/pull/16144)
+
 ## [v7.1.0](https://github.com/prysmaticlabs/prysm/compare/v7.0.0...v7.1.0) - 2025-12-10
 
 This release includes several key features/fixes. If you are running v7.0.0 then you should update to v7.0.1 or later and remove the flag `--disable-last-epoch-targets`. 

api/apiutil/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "common.go",
         "header.go",
+        "log.go",
     ],
     importpath = "github.com/OffchainLabs/prysm/v7/api/apiutil",
     visibility = ["//visibility:public"],

api/apiutil/header.go

@@ -5,8 +5,6 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-
-	log "github.com/sirupsen/logrus"
 )
 
 type mediaRange struct {

api/apiutil/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package apiutil
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "api/apiutil")

api/client/beacon/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package beacon
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "beacon")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "api/client/beacon")

api/client/builder/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "bid.go",
         "client.go",
         "errors.go",
+        "log.go",
         "types.go",
     ],
     importpath = "github.com/OffchainLabs/prysm/v7/api/client/builder",
@@ -63,6 +64,5 @@ go_test(
         "@com_github_ethereum_go_ethereum//common/hexutil:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
-        "@com_github_sirupsen_logrus//:go_default_library",
     ],
 )

api/client/builder/client.go

@@ -25,7 +25,7 @@ import (
 	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
 	"github.com/OffchainLabs/prysm/v7/runtime/version"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
@@ -70,7 +70,7 @@ type requestLogger struct{}
 func (*requestLogger) observe(r *http.Request) (e error) {
 	b := bytes.NewBuffer(nil)
 	if r.Body == nil {
-		log.WithFields(log.Fields{
+		log.WithFields(logrus.Fields{
 			"bodyBase64": "(nil value)",
 			"url":        r.URL.String(),
 		}).Info("Builder http request")
@@ -87,7 +87,7 @@ func (*requestLogger) observe(r *http.Request) (e error) {
 		return err
 	}
 	r.Body = io.NopCloser(b)
-	log.WithFields(log.Fields{
+	log.WithFields(logrus.Fields{
 		"bodyBase64": string(body),
 		"url":        r.URL.String(),
 	}).Info("Builder http request")

api/client/builder/client_test.go

@@ -23,7 +23,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/testing/assert"
 	"github.com/OffchainLabs/prysm/v7/testing/require"
 	"github.com/OffchainLabs/prysm/v7/testing/util"
-	log "github.com/sirupsen/logrus"
 )
 
 type roundtrip func(*http.Request) (*http.Response, error)

api/client/builder/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package builder
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "api/client/builder")

api/client/event/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "event_stream.go",
+        "log.go",
         "utils.go",
     ],
     importpath = "github.com/OffchainLabs/prysm/v7/api/client/event",
@@ -23,8 +24,5 @@ go_test(
         "utils_test.go",
     ],
     embed = [":go_default_library"],
-    deps = [
-        "//testing/require:go_default_library",
-        "@com_github_sirupsen_logrus//:go_default_library",
-    ],
+    deps = ["//testing/require:go_default_library"],
 )

api/client/event/event_stream.go

@@ -10,7 +10,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/api"
 	"github.com/OffchainLabs/prysm/v7/api/client"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 )
 
 const (

api/client/event/event_stream_test.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/OffchainLabs/prysm/v7/testing/require"
-	log "github.com/sirupsen/logrus"
 )
 
 func TestNewEventStream(t *testing.T) {

api/client/event/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package event
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "api/client/event")

api/grpc/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "grpcutils.go",
+        "log.go",
         "parameters.go",
     ],
     importpath = "github.com/OffchainLabs/prysm/v7/api/grpc",

api/grpc/grpcutils.go

@@ -32,7 +32,7 @@ func LogRequests(
 	)
 	start := time.Now()
 	err := invoker(ctx, method, req, reply, cc, opts...)
-	logrus.WithField("backend", header["x-backend"]).
+	log.WithField("backend", header["x-backend"]).
 		WithField("method", method).WithField("duration", time.Since(start)).
 		Debug("gRPC request finished.")
 	return err
@@ -58,7 +58,7 @@ func LogStream(
 		grpc.Header(&header),
 	)
 	strm, err := streamer(ctx, sd, conn, method, opts...)
-	logrus.WithField("backend", header["x-backend"]).
+	log.WithField("backend", header["x-backend"]).
 		WithField("method", method).
 		Debug("gRPC stream started.")
 	return strm, err
@@ -71,7 +71,7 @@ func AppendHeaders(parent context.Context, headers []string) context.Context {
 		if h != "" {
 			keyValue := strings.Split(h, "=")
 			if len(keyValue) < 2 {
-				logrus.Warnf("Incorrect gRPC header flag format. Skipping %v", keyValue[0])
+				log.Warnf("Incorrect gRPC header flag format. Skipping %v", keyValue[0])
 				continue
 			}
 			parent = metadata.AppendToOutgoingContext(parent, keyValue[0], strings.Join(keyValue[1:], "=")) // nolint:fatcontext

api/grpc/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package grpc
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "api/grpc")

api/server/httprest/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package httprest
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "httprest")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "api/server/httprest")

api/server/middleware/BUILD.bazel

@@ -3,6 +3,7 @@ load("@prysm//tools/go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "log.go",
         "middleware.go",
         "util.go",
     ],
@@ -27,6 +28,5 @@ go_test(
         "//api:go_default_library",
         "//testing/assert:go_default_library",
         "//testing/require:go_default_library",
-        "@com_github_sirupsen_logrus//:go_default_library",
     ],
 )

api/server/middleware/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package middleware
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "api/server/middleware")

api/server/middleware/middleware.go

@@ -9,7 +9,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/api"
 	"github.com/OffchainLabs/prysm/v7/api/apiutil"
 	"github.com/rs/cors"
-	log "github.com/sirupsen/logrus"
 )
 
 type Middleware func(http.Handler) http.Handler

api/server/middleware/middleware_test.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/OffchainLabs/prysm/v7/api"
 	"github.com/OffchainLabs/prysm/v7/testing/require"
-	log "github.com/sirupsen/logrus"
 )
 
 // frozenHeaderRecorder allows asserting that response headers were not modified

async/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "debounce.go",
         "every.go",
+        "log.go",
         "multilock.go",
         "scatter.go",
     ],

async/every.go

@@ -6,8 +6,6 @@ import (
 	"reflect"
 	"runtime"
 	"time"
-
-	log "github.com/sirupsen/logrus"
 )
 
 // RunEvery runs the provided command periodically.

async/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package async
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "async")

beacon-chain/blockchain/BUILD.bazel

@@ -14,6 +14,7 @@ go_library(
         "head_sync_committee_info.go",
         "init_sync_process_block.go",
         "log.go",
+        "log_helpers.go",
         "merge_ascii_art.go",
         "metrics.go",
         "options.go",

beacon-chain/blockchain/log.go

@@ -1,164 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package blockchain
 
-import (
-	"encoding/hex"
-	"fmt"
-	"time"
+import "github.com/sirupsen/logrus"
 
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/blocks"
-	"github.com/OffchainLabs/prysm/v7/config/params"
-	consensus_types "github.com/OffchainLabs/prysm/v7/consensus-types"
-	"github.com/OffchainLabs/prysm/v7/consensus-types/interfaces"
-	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
-	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
-	"github.com/OffchainLabs/prysm/v7/runtime/version"
-	prysmTime "github.com/OffchainLabs/prysm/v7/time"
-	"github.com/OffchainLabs/prysm/v7/time/slots"
-	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
-)
-
-var log = logrus.WithField("prefix", "blockchain")
-
-// logs state transition related data every slot.
-func logStateTransitionData(b interfaces.ReadOnlyBeaconBlock) error {
-	log := log.WithField("slot", b.Slot())
-	if len(b.Body().Attestations()) > 0 {
-		log = log.WithField("attestations", len(b.Body().Attestations()))
-	}
-	if len(b.Body().AttesterSlashings()) > 0 {
-		log = log.WithField("attesterSlashings", len(b.Body().AttesterSlashings()))
-	}
-	if len(b.Body().ProposerSlashings()) > 0 {
-		log = log.WithField("proposerSlashings", len(b.Body().ProposerSlashings()))
-	}
-	if len(b.Body().VoluntaryExits()) > 0 {
-		log = log.WithField("voluntaryExits", len(b.Body().VoluntaryExits()))
-	}
-	if b.Version() >= version.Altair {
-		agg, err := b.Body().SyncAggregate()
-		if err != nil {
-			return err
-		}
-		log = log.WithField("syncBitsCount", agg.SyncCommitteeBits.Count())
-	}
-	if b.Version() >= version.Bellatrix {
-		p, err := b.Body().Execution()
-		if err != nil {
-			return err
-		}
-		log = log.WithField("payloadHash", fmt.Sprintf("%#x", bytesutil.Trunc(p.BlockHash())))
-		txs, err := p.Transactions()
-		switch {
-		case errors.Is(err, consensus_types.ErrUnsupportedField):
-		case err != nil:
-			return err
-		default:
-			log = log.WithField("txCount", len(txs))
-			txsPerSlotCount.Set(float64(len(txs)))
-		}
-	}
-	if b.Version() >= version.Deneb {
-		kzgs, err := b.Body().BlobKzgCommitments()
-		if err != nil {
-			log.WithError(err).Error("Failed to get blob KZG commitments")
-		} else if len(kzgs) > 0 {
-			log = log.WithField("kzgCommitmentCount", len(kzgs))
-		}
-	}
-	if b.Version() >= version.Electra {
-		eReqs, err := b.Body().ExecutionRequests()
-		if err != nil {
-			log.WithError(err).Error("Failed to get execution requests")
-		} else {
-			if len(eReqs.Deposits) > 0 {
-				log = log.WithField("depositRequestCount", len(eReqs.Deposits))
-			}
-			if len(eReqs.Consolidations) > 0 {
-				log = log.WithField("consolidationRequestCount", len(eReqs.Consolidations))
-			}
-			if len(eReqs.Withdrawals) > 0 {
-				log = log.WithField("withdrawalRequestCount", len(eReqs.Withdrawals))
-			}
-		}
-	}
-	log.Info("Finished applying state transition")
-	return nil
-}
-
-func logBlockSyncStatus(block interfaces.ReadOnlyBeaconBlock, blockRoot [32]byte, justified, finalized *ethpb.Checkpoint, receivedTime time.Time, genesis time.Time, daWaitedTime time.Duration) error {
-	startTime, err := slots.StartTime(genesis, block.Slot())
-	if err != nil {
-		return err
-	}
-	level := log.Logger.GetLevel()
-	if level >= logrus.DebugLevel {
-		parentRoot := block.ParentRoot()
-		lf := logrus.Fields{
-			"slot":                       block.Slot(),
-			"slotInEpoch":                block.Slot() % params.BeaconConfig().SlotsPerEpoch,
-			"block":                      fmt.Sprintf("0x%s...", hex.EncodeToString(blockRoot[:])[:8]),
-			"epoch":                      slots.ToEpoch(block.Slot()),
-			"justifiedEpoch":             justified.Epoch,
-			"justifiedRoot":              fmt.Sprintf("0x%s...", hex.EncodeToString(justified.Root)[:8]),
-			"finalizedEpoch":             finalized.Epoch,
-			"finalizedRoot":              fmt.Sprintf("0x%s...", hex.EncodeToString(finalized.Root)[:8]),
-			"parentRoot":                 fmt.Sprintf("0x%s...", hex.EncodeToString(parentRoot[:])[:8]),
-			"version":                    version.String(block.Version()),
-			"sinceSlotStartTime":         prysmTime.Now().Sub(startTime),
-			"chainServiceProcessedTime":  prysmTime.Now().Sub(receivedTime) - daWaitedTime,
-			"dataAvailabilityWaitedTime": daWaitedTime,
-		}
-		log.WithFields(lf).Debug("Synced new block")
-	} else {
-		log.WithFields(logrus.Fields{
-			"slot":           block.Slot(),
-			"block":          fmt.Sprintf("0x%s...", hex.EncodeToString(blockRoot[:])[:8]),
-			"finalizedEpoch": finalized.Epoch,
-			"finalizedRoot":  fmt.Sprintf("0x%s...", hex.EncodeToString(finalized.Root)[:8]),
-			"epoch":          slots.ToEpoch(block.Slot()),
-		}).Info("Synced new block")
-	}
-	return nil
-}
-
-// logs payload related data every slot.
-func logPayload(block interfaces.ReadOnlyBeaconBlock) error {
-	isExecutionBlk, err := blocks.IsExecutionBlock(block.Body())
-	if err != nil {
-		return errors.Wrap(err, "could not determine if block is execution block")
-	}
-	if !isExecutionBlk {
-		return nil
-	}
-	payload, err := block.Body().Execution()
-	if err != nil {
-		return err
-	}
-	if payload.GasLimit() == 0 {
-		return errors.New("gas limit should not be 0")
-	}
-	gasUtilized := float64(payload.GasUsed()) / float64(payload.GasLimit())
-	fields := logrus.Fields{
-		"blockHash":   fmt.Sprintf("%#x", bytesutil.Trunc(payload.BlockHash())),
-		"parentHash":  fmt.Sprintf("%#x", bytesutil.Trunc(payload.ParentHash())),
-		"blockNumber": payload.BlockNumber(),
-		"gasUtilized": fmt.Sprintf("%.2f", gasUtilized),
-	}
-	if block.Version() >= version.Capella {
-		withdrawals, err := payload.Withdrawals()
-		if err != nil {
-			return errors.Wrap(err, "could not get withdrawals")
-		}
-		fields["withdrawals"] = len(withdrawals)
-		changes, err := block.Body().BLSToExecutionChanges()
-		if err != nil {
-			return errors.Wrap(err, "could not get BLSToExecutionChanges")
-		}
-		if len(changes) > 0 {
-			fields["blsToExecutionChanges"] = len(changes)
-		}
-	}
-	log.WithFields(fields).Debug("Synced new payload")
-	return nil
-}
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/blockchain")

beacon-chain/blockchain/log_helpers.go

@@ -0,0 +1,162 @@
+package blockchain
+
+import (
+	"encoding/hex"
+	"fmt"
+	"time"
+
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/blocks"
+	"github.com/OffchainLabs/prysm/v7/config/params"
+	consensus_types "github.com/OffchainLabs/prysm/v7/consensus-types"
+	"github.com/OffchainLabs/prysm/v7/consensus-types/interfaces"
+	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
+	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
+	"github.com/OffchainLabs/prysm/v7/runtime/version"
+	prysmTime "github.com/OffchainLabs/prysm/v7/time"
+	"github.com/OffchainLabs/prysm/v7/time/slots"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+// logs state transition related data every slot.
+func logStateTransitionData(b interfaces.ReadOnlyBeaconBlock) error {
+	log := log.WithField("slot", b.Slot())
+	if len(b.Body().Attestations()) > 0 {
+		log = log.WithField("attestations", len(b.Body().Attestations()))
+	}
+	if len(b.Body().AttesterSlashings()) > 0 {
+		log = log.WithField("attesterSlashings", len(b.Body().AttesterSlashings()))
+	}
+	if len(b.Body().ProposerSlashings()) > 0 {
+		log = log.WithField("proposerSlashings", len(b.Body().ProposerSlashings()))
+	}
+	if len(b.Body().VoluntaryExits()) > 0 {
+		log = log.WithField("voluntaryExits", len(b.Body().VoluntaryExits()))
+	}
+	if b.Version() >= version.Altair {
+		agg, err := b.Body().SyncAggregate()
+		if err != nil {
+			return err
+		}
+		log = log.WithField("syncBitsCount", agg.SyncCommitteeBits.Count())
+	}
+	if b.Version() >= version.Bellatrix {
+		p, err := b.Body().Execution()
+		if err != nil {
+			return err
+		}
+		log = log.WithField("payloadHash", fmt.Sprintf("%#x", bytesutil.Trunc(p.BlockHash())))
+		txs, err := p.Transactions()
+		switch {
+		case errors.Is(err, consensus_types.ErrUnsupportedField):
+		case err != nil:
+			return err
+		default:
+			log = log.WithField("txCount", len(txs))
+			txsPerSlotCount.Set(float64(len(txs)))
+		}
+	}
+	if b.Version() >= version.Deneb {
+		kzgs, err := b.Body().BlobKzgCommitments()
+		if err != nil {
+			log.WithError(err).Error("Failed to get blob KZG commitments")
+		} else if len(kzgs) > 0 {
+			log = log.WithField("kzgCommitmentCount", len(kzgs))
+		}
+	}
+	if b.Version() >= version.Electra {
+		eReqs, err := b.Body().ExecutionRequests()
+		if err != nil {
+			log.WithError(err).Error("Failed to get execution requests")
+		} else {
+			if len(eReqs.Deposits) > 0 {
+				log = log.WithField("depositRequestCount", len(eReqs.Deposits))
+			}
+			if len(eReqs.Consolidations) > 0 {
+				log = log.WithField("consolidationRequestCount", len(eReqs.Consolidations))
+			}
+			if len(eReqs.Withdrawals) > 0 {
+				log = log.WithField("withdrawalRequestCount", len(eReqs.Withdrawals))
+			}
+		}
+	}
+	log.Info("Finished applying state transition")
+	return nil
+}
+
+func logBlockSyncStatus(block interfaces.ReadOnlyBeaconBlock, blockRoot [32]byte, justified, finalized *ethpb.Checkpoint, receivedTime time.Time, genesis time.Time, daWaitedTime time.Duration) error {
+	startTime, err := slots.StartTime(genesis, block.Slot())
+	if err != nil {
+		return err
+	}
+	level := log.Logger.GetLevel()
+	if level >= logrus.DebugLevel {
+		parentRoot := block.ParentRoot()
+		lf := logrus.Fields{
+			"slot":                       block.Slot(),
+			"slotInEpoch":                block.Slot() % params.BeaconConfig().SlotsPerEpoch,
+			"block":                      fmt.Sprintf("0x%s...", hex.EncodeToString(blockRoot[:])[:8]),
+			"epoch":                      slots.ToEpoch(block.Slot()),
+			"justifiedEpoch":             justified.Epoch,
+			"justifiedRoot":              fmt.Sprintf("0x%s...", hex.EncodeToString(justified.Root)[:8]),
+			"finalizedEpoch":             finalized.Epoch,
+			"finalizedRoot":              fmt.Sprintf("0x%s...", hex.EncodeToString(finalized.Root)[:8]),
+			"parentRoot":                 fmt.Sprintf("0x%s...", hex.EncodeToString(parentRoot[:])[:8]),
+			"version":                    version.String(block.Version()),
+			"sinceSlotStartTime":         prysmTime.Now().Sub(startTime),
+			"chainServiceProcessedTime":  prysmTime.Now().Sub(receivedTime) - daWaitedTime,
+			"dataAvailabilityWaitedTime": daWaitedTime,
+		}
+		log.WithFields(lf).Debug("Synced new block")
+	} else {
+		log.WithFields(logrus.Fields{
+			"slot":           block.Slot(),
+			"block":          fmt.Sprintf("0x%s...", hex.EncodeToString(blockRoot[:])[:8]),
+			"finalizedEpoch": finalized.Epoch,
+			"finalizedRoot":  fmt.Sprintf("0x%s...", hex.EncodeToString(finalized.Root)[:8]),
+			"epoch":          slots.ToEpoch(block.Slot()),
+		}).Info("Synced new block")
+	}
+	return nil
+}
+
+// logs payload related data every slot.
+func logPayload(block interfaces.ReadOnlyBeaconBlock) error {
+	isExecutionBlk, err := blocks.IsExecutionBlock(block.Body())
+	if err != nil {
+		return errors.Wrap(err, "could not determine if block is execution block")
+	}
+	if !isExecutionBlk {
+		return nil
+	}
+	payload, err := block.Body().Execution()
+	if err != nil {
+		return err
+	}
+	if payload.GasLimit() == 0 {
+		return errors.New("gas limit should not be 0")
+	}
+	gasUtilized := float64(payload.GasUsed()) / float64(payload.GasLimit())
+	fields := logrus.Fields{
+		"blockHash":   fmt.Sprintf("%#x", bytesutil.Trunc(payload.BlockHash())),
+		"parentHash":  fmt.Sprintf("%#x", bytesutil.Trunc(payload.ParentHash())),
+		"blockNumber": payload.BlockNumber(),
+		"gasUtilized": fmt.Sprintf("%.2f", gasUtilized),
+	}
+	if block.Version() >= version.Capella {
+		withdrawals, err := payload.Withdrawals()
+		if err != nil {
+			return errors.Wrap(err, "could not get withdrawals")
+		}
+		fields["withdrawals"] = len(withdrawals)
+		changes, err := block.Body().BLSToExecutionChanges()
+		if err != nil {
+			return errors.Wrap(err, "could not get BLSToExecutionChanges")
+		}
+		if len(changes) > 0 {
+			fields["blsToExecutionChanges"] = len(changes)
+		}
+	}
+	log.WithFields(fields).Debug("Synced new payload")
+	return nil
+}

beacon-chain/blockchain/log_test.go

@@ -34,7 +34,7 @@ func Test_logStateTransitionData(t *testing.T) {
 				require.NoError(t, err)
 				return wb
 			},
-			want: "\"Finished applying state transition\" prefix=blockchain slot=0",
+			want: "\"Finished applying state transition\" package=beacon-chain/blockchain slot=0",
 		},
 		{name: "has attestation",
 			b: func() interfaces.ReadOnlyBeaconBlock {
@@ -42,7 +42,7 @@ func Test_logStateTransitionData(t *testing.T) {
 				require.NoError(t, err)
 				return wb
 			},
-			want: "\"Finished applying state transition\" attestations=1 prefix=blockchain slot=0",
+			want: "\"Finished applying state transition\" attestations=1 package=beacon-chain/blockchain slot=0",
 		},
 		{name: "has deposit",
 			b: func() interfaces.ReadOnlyBeaconBlock {
@@ -53,7 +53,7 @@ func Test_logStateTransitionData(t *testing.T) {
 				require.NoError(t, err)
 				return wb
 			},
-			want: "\"Finished applying state transition\" attestations=1 prefix=blockchain slot=0",
+			want: "\"Finished applying state transition\" attestations=1 package=beacon-chain/blockchain slot=0",
 		},
 		{name: "has attester slashing",
 			b: func() interfaces.ReadOnlyBeaconBlock {
@@ -62,7 +62,7 @@ func Test_logStateTransitionData(t *testing.T) {
 				require.NoError(t, err)
 				return wb
 			},
-			want: "\"Finished applying state transition\" attesterSlashings=1 prefix=blockchain slot=0",
+			want: "\"Finished applying state transition\" attesterSlashings=1 package=beacon-chain/blockchain slot=0",
 		},
 		{name: "has proposer slashing",
 			b: func() interfaces.ReadOnlyBeaconBlock {
@@ -71,7 +71,7 @@ func Test_logStateTransitionData(t *testing.T) {
 				require.NoError(t, err)
 				return wb
 			},
-			want: "\"Finished applying state transition\" prefix=blockchain proposerSlashings=1 slot=0",
+			want: "\"Finished applying state transition\" package=beacon-chain/blockchain proposerSlashings=1 slot=0",
 		},
 		{name: "has exit",
 			b: func() interfaces.ReadOnlyBeaconBlock {
@@ -80,7 +80,7 @@ func Test_logStateTransitionData(t *testing.T) {
 				require.NoError(t, err)
 				return wb
 			},
-			want: "\"Finished applying state transition\" prefix=blockchain slot=0 voluntaryExits=1",
+			want: "\"Finished applying state transition\" package=beacon-chain/blockchain slot=0 voluntaryExits=1",
 		},
 		{name: "has everything",
 			b: func() interfaces.ReadOnlyBeaconBlock {
@@ -93,11 +93,11 @@ func Test_logStateTransitionData(t *testing.T) {
 				require.NoError(t, err)
 				return wb
 			},
-			want: "\"Finished applying state transition\" attestations=1 attesterSlashings=1 prefix=blockchain proposerSlashings=1 slot=0 voluntaryExits=1",
+			want: "\"Finished applying state transition\" attestations=1 attesterSlashings=1 package=beacon-chain/blockchain proposerSlashings=1 slot=0 voluntaryExits=1",
 		},
 		{name: "has payload",
 			b:    func() interfaces.ReadOnlyBeaconBlock { return wrappedPayloadBlk },
-			want: "\"Finished applying state transition\" payloadHash=0x010203 prefix=blockchain slot=0 syncBitsCount=0 txCount=2",
+			want: "\"Finished applying state transition\" package=beacon-chain/blockchain payloadHash=0x010203 slot=0 syncBitsCount=0 txCount=2",
 		},
 	}
 	for _, tt := range tests {

beacon-chain/blockchain/process_block.go

@@ -106,7 +106,12 @@ func (s *Service) postBlockProcess(cfg *postBlockProcessConfig) error {
 
 	// Pre-Fulu the caches are updated when computing the payload attributes
 	if cfg.postState.Version() >= version.Fulu {
-		go s.updateCachesPostBlockProcessing(cfg)
+		go func() {
+			ctx, cancel := context.WithTimeout(s.ctx, slotDeadline)
+			defer cancel()
+			cfg.ctx = ctx
+			s.updateCachesPostBlockProcessing(cfg)
+		}()
 	}
 	return nil
 }
@@ -929,6 +934,8 @@ func (s *Service) lateBlockTasks(ctx context.Context) {
 		// After Fulu, we can update the caches asynchronously after sending FCU to the engine
 		defer func() {
 			go func() {
+				ctx, cancel := context.WithTimeout(s.ctx, slotDeadline)
+				defer cancel()
 				lastState.CopyAllTries()
 				if err := transition.UpdateNextSlotCache(ctx, lastRoot, lastState); err != nil {
 					log.WithError(err).Debug("Could not update next slot state cache")

beacon-chain/blockchain/testing/BUILD.bazel

@@ -3,7 +3,10 @@ load("@prysm//tools/go:def.bzl", "go_library")
 go_library(
     name = "go_default_library",
     testonly = True,
-    srcs = ["mock.go"],
+    srcs = [
+        "log.go",
+        "mock.go",
+    ],
     importpath = "github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/testing",
     visibility = [
         "//beacon-chain:__subpackages__",

beacon-chain/blockchain/testing/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package testing
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/blockchain/testing")

beacon-chain/blockchain/testing/mock.go

@@ -30,7 +30,6 @@ import (
 	enginev1 "github.com/OffchainLabs/prysm/v7/proto/engine/v1"
 	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
 	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
 )
 
 var ErrNilState = errors.New("nil state")
@@ -267,7 +266,7 @@ func (s *ChainService) ReceiveBlockInitialSync(ctx context.Context, block interf
 		if err := s.DB.SaveBlock(ctx, block); err != nil {
 			return err
 		}
-		logrus.Infof("Saved block with root: %#x at slot %d", signingRoot, block.Block().Slot())
+		log.Infof("Saved block with root: %#x at slot %d", signingRoot, block.Block().Slot())
 	}
 	s.Root = signingRoot[:]
 	s.Block = block
@@ -296,7 +295,7 @@ func (s *ChainService) ReceiveBlockBatch(ctx context.Context, blks []blocks.ROBl
 			if err := s.DB.SaveBlock(ctx, b); err != nil {
 				return err
 			}
-			logrus.Infof("Saved block with root: %#x at slot %d", signingRoot, b.Block().Slot())
+			log.Infof("Saved block with root: %#x at slot %d", signingRoot, b.Block().Slot())
 		}
 		s.Root = signingRoot[:]
 		s.Block = b
@@ -328,7 +327,7 @@ func (s *ChainService) ReceiveBlock(ctx context.Context, block interfaces.ReadOn
 		if err := s.DB.SaveBlock(ctx, block); err != nil {
 			return err
 		}
-		logrus.Infof("Saved block with root: %#x at slot %d", signingRoot, block.Block().Slot())
+		log.Infof("Saved block with root: %#x at slot %d", signingRoot, block.Block().Slot())
 	}
 	s.Root = signingRoot[:]
 	s.Block = block
@@ -585,11 +584,11 @@ func (s *ChainService) UpdateHead(ctx context.Context, slot primitives.Slot) {
 	ojc := &ethpb.Checkpoint{}
 	st, root, err := prepareForkchoiceState(ctx, slot, bytesutil.ToBytes32(s.Root), [32]byte{}, [32]byte{}, ojc, ojc)
 	if err != nil {
-		logrus.WithError(err).Error("Could not update head")
+		log.WithError(err).Error("Could not update head")
 	}
 	err = s.ForkChoiceStore.InsertNode(ctx, st, root)
 	if err != nil {
-		logrus.WithError(err).Error("Could not insert node to forkchoice")
+		log.WithError(err).Error("Could not insert node to forkchoice")
 	}
 }
 

beacon-chain/builder/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package builder
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "builder")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/builder")

beacon-chain/cache/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "doc.go",
         "error.go",
         "interfaces.go",
+        "log.go",
         "payload_id.go",
         "proposer_indices.go",
         "proposer_indices_disabled.go",  # keep

beacon-chain/cache/attestation.go

@@ -9,7 +9,6 @@ import (
 	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
 	"github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1/attestation"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 )
 
 type attGroup struct {

beacon-chain/cache/committee.go

@@ -17,7 +17,6 @@ import (
 	lru "github.com/hashicorp/golang-lru"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	log "github.com/sirupsen/logrus"
 )
 
 const (

beacon-chain/cache/depositsnapshot/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
         "deposit_pruner.go",
         "deposit_tree.go",
         "deposit_tree_snapshot.go",
+        "log.go",
         "merkle_tree.go",
     ],
     importpath = "github.com/OffchainLabs/prysm/v7/beacon-chain/cache/depositsnapshot",

beacon-chain/cache/depositsnapshot/deposit_inserter.go

@@ -20,7 +20,6 @@ var (
 		Name: "beacondb_all_deposits_eip4881",
 		Help: "The number of total deposits in memory",
 	})
-	log = logrus.WithField("prefix", "cache")
 )
 
 // InsertDeposit into the database. If deposit or block number are nil

beacon-chain/cache/depositsnapshot/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package depositsnapshot
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/cache/depositsnapshot")

beacon-chain/cache/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package cache
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/cache")

beacon-chain/cache/sync_committee.go

@@ -11,7 +11,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	log "github.com/sirupsen/logrus"
 	"k8s.io/client-go/tools/cache"
 )
 

beacon-chain/cache/tracked_validators.go

@@ -9,7 +9,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/sirupsen/logrus"
 )
 
 const (
@@ -67,7 +66,7 @@ func (t *TrackedValidatorsCache) Validator(index primitives.ValidatorIndex) (Tra
 
 	val, ok := item.(TrackedValidator)
 	if !ok {
-		logrus.Errorf("Failed to cast tracked validator from cache, got unexpected item type %T", item)
+		log.Errorf("Failed to cast tracked validator from cache, got unexpected item type %T", item)
 		return TrackedValidator{}, false
 	}
 
@@ -113,7 +112,7 @@ func (t *TrackedValidatorsCache) Indices() map[primitives.ValidatorIndex]bool {
 	for cacheKey := range items {
 		index, err := fromCacheKey(cacheKey)
 		if err != nil {
-			logrus.WithError(err).Error("Failed to get validator index from cache key")
+			log.WithError(err).Error("Failed to get validator index from cache key")
 			continue
 		}
 

beacon-chain/core/altair/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
         "deposit.go",
         "epoch_precompute.go",
         "epoch_spec.go",
+        "log.go",
         "reward.go",
         "sync_committee.go",
         "transition.go",

beacon-chain/core/altair/epoch_spec.go

@@ -7,7 +7,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/time"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	"github.com/OffchainLabs/prysm/v7/config/params"
-	log "github.com/sirupsen/logrus"
 )
 
 // ProcessSyncCommitteeUpdates  processes sync client committee updates for the beacon state.

beacon-chain/core/altair/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package altair
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/core/altair")

beacon-chain/core/blocks/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package blocks
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "blocks")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/core/blocks")

beacon-chain/core/electra/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "deposits.go",
         "effective_balance_updates.go",
         "error.go",
+        "log.go",
         "registry_updates.go",
         "transition.go",
         "transition_no_verify_sig.go",

beacon-chain/core/electra/consolidations.go

@@ -17,7 +17,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/ethereum/go-ethereum/common/math"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 )
 
 // ProcessPendingConsolidations implements the spec definition below. This method makes mutating

beacon-chain/core/electra/deposits.go

@@ -18,7 +18,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/runtime/version"
 	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 )
 
 // ProcessDeposits is one of the operations performed on each processed

beacon-chain/core/electra/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package electra
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/core/electra")

beacon-chain/core/electra/withdrawals.go

@@ -17,7 +17,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 )
 
 // ProcessWithdrawalRequests processes the validator withdrawals from the provided execution payload

beacon-chain/core/helpers/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
         "block.go",
         "genesis.go",
         "legacy.go",
+        "log.go",
         "metrics.go",
         "randao.go",
         "ranges.go",

beacon-chain/core/helpers/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package helpers
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/core/helpers")

beacon-chain/core/helpers/sync_committee.go

@@ -14,7 +14,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
 	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 )
 
 var (

beacon-chain/core/helpers/validators.go

@@ -21,7 +21,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	log "github.com/sirupsen/logrus"
 )
 
 var (

beacon-chain/core/transition/interop/log.go

@@ -1,10 +1,9 @@
-// Package interop contains useful utilities for persisting
-// ssz-encoded states and blocks to disk during each state
-// transition for development purposes.
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package interop
 
-import (
-	"github.com/sirupsen/logrus"
-)
+import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "interop")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/core/transition/interop")

beacon-chain/core/transition/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package transition
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "state")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/core/transition")

beacon-chain/das/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package das
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "das")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/das")

beacon-chain/db/filesystem/BUILD.bazel

@@ -38,6 +38,7 @@ go_library(
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@com_github_spf13_afero//:go_default_library",
+        "@org_golang_x_sync//errgroup:go_default_library",
     ],
 )
 

beacon-chain/db/filesystem/data_column.go

@@ -25,6 +25,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/pkg/errors"
 	"github.com/spf13/afero"
+	"golang.org/x/sync/errgroup"
 )
 
 const (
@@ -185,73 +186,162 @@ func (dcs *DataColumnStorage) WarmCache() {
 
 	highestStoredEpoch := primitives.Epoch(0)
 
-	// Walk the data column filesystem to warm up the cache.
-	if err := afero.Walk(dcs.fs, ".", func(path string, info os.FileInfo, fileErr error) (err error) {
-		if fileErr != nil {
-			return fileErr
-		}
-
-		// If not a leaf, skip.
-		if info.IsDir() {
-			return nil
-		}
-
-		// Extract metadata from the file path.
-		fileMetadata, err := extractFileMetadata(path)
-		if err != nil {
-			log.WithError(err).Error("Error encountered while extracting file metadata")
-			return nil
-		}
-
-		// Open the data column filesystem file.
-		f, err := dcs.fs.Open(path)
-		if err != nil {
-			log.WithError(err).Error("Error encountered while opening data column filesystem file")
-			return nil
-		}
-
-		// Close the file.
-		defer func() {
-			// Overwrite the existing error only if it is nil, since the close error is less important.
-			closeErr := f.Close()
-			if closeErr != nil && err == nil {
-				err = closeErr
-			}
-		}()
-
-		// Read the metadata of the file.
-		metadata, err := dcs.metadata(f)
-		if err != nil {
-			log.WithError(err).Error("Error encountered while reading metadata from data column filesystem file")
-			return nil
-		}
-
-		// Check the indices.
-		indices := metadata.indices.all()
-		if len(indices) == 0 {
-			return nil
-		}
-
-		// Build the ident.
-		dataColumnsIdent := DataColumnsIdent{Root: fileMetadata.blockRoot, Epoch: fileMetadata.epoch, Indices: indices}
-
-		// Update the highest stored epoch.
-		highestStoredEpoch = max(highestStoredEpoch, fileMetadata.epoch)
-
-		// Set the ident in the cache.
-		if err := dcs.cache.set(dataColumnsIdent); err != nil {
-			log.WithError(err).Error("Error encountered while ensuring data column filesystem cache")
-		}
-
-		return nil
-	}); err != nil {
-		log.WithError(err).Error("Error encountered while walking data column filesystem.")
+	// List all period directories
+	periodFileInfos, err := afero.ReadDir(dcs.fs, ".")
+	if err != nil {
+		log.WithError(err).Error("Error reading top directory during warm cache")
+		return
 	}
 
-	// Prune the cache and the filesystem.
+	// Iterate through periods
+	for _, periodFileInfo := range periodFileInfos {
+		if !periodFileInfo.IsDir() {
+			continue
+		}
+
+		periodPath := periodFileInfo.Name()
+
+		// List all epoch directories in this period
+		epochFileInfos, err := afero.ReadDir(dcs.fs, periodPath)
+		if err != nil {
+			log.WithError(err).WithField("period", periodPath).Error("Error reading period directory during warm cache")
+			continue
+		}
+
+		// Iterate through epochs
+		for _, epochFileInfo := range epochFileInfos {
+			if !epochFileInfo.IsDir() {
+				continue
+			}
+
+			epochPath := path.Join(periodPath, epochFileInfo.Name())
+
+			// List all .sszs files in this epoch
+			files, err := listEpochFiles(dcs.fs, epochPath)
+			if err != nil {
+				log.WithError(err).WithField("epoch", epochPath).Error("Error listing epoch files during warm cache")
+				continue
+			}
+
+			if len(files) == 0 {
+				continue
+			}
+
+			// Process all files in this epoch in parallel
+			epochHighest, err := dcs.processEpochFiles(files)
+			if err != nil {
+				log.WithError(err).WithField("epoch", epochPath).Error("Error processing epoch files during warm cache")
+			}
+
+			highestStoredEpoch = max(highestStoredEpoch, epochHighest)
+		}
+	}
+
+	// Prune the cache and the filesystem
 	dcs.prune()
 
-	log.WithField("elapsed", time.Since(start)).Info("Data column filesystem cache warm-up complete")
+	totalElapsed := time.Since(start)
+
+	// Log summary
+	log.WithField("elapsed", totalElapsed).Info("Data column filesystem cache warm-up complete")
+}
+
+// listEpochFiles lists all .sszs files in an epoch directory.
+func listEpochFiles(fs afero.Fs, epochPath string) ([]string, error) {
+	fileInfos, err := afero.ReadDir(fs, epochPath)
+	if err != nil {
+		return nil, errors.Wrap(err, "read epoch directory")
+	}
+
+	files := make([]string, 0, len(fileInfos))
+	for _, fileInfo := range fileInfos {
+		if fileInfo.IsDir() {
+			continue
+		}
+
+		fileName := fileInfo.Name()
+		if strings.HasSuffix(fileName, "."+dataColumnsFileExtension) {
+			files = append(files, path.Join(epochPath, fileName))
+		}
+	}
+
+	return files, nil
+}
+
+// processEpochFiles processes all .sszs files in an epoch directory in parallel.
+func (dcs *DataColumnStorage) processEpochFiles(files []string) (primitives.Epoch, error) {
+	var (
+		eg errgroup.Group
+		mu sync.Mutex
+	)
+
+	highestEpoch := primitives.Epoch(0)
+	for _, filePath := range files {
+		eg.Go(func() error {
+			epoch, err := dcs.processFile(filePath)
+			if err != nil {
+				log.WithError(err).WithField("file", filePath).Error("Error processing file during warm cache")
+				return nil
+			}
+
+			mu.Lock()
+			defer mu.Unlock()
+			highestEpoch = max(highestEpoch, epoch)
+
+			return nil
+		})
+	}
+
+	if err := eg.Wait(); err != nil {
+		return highestEpoch, err
+	}
+
+	return highestEpoch, nil
+}
+
+// processFile processes a single .sszs file.
+func (dcs *DataColumnStorage) processFile(filePath string) (primitives.Epoch, error) {
+	// Extract metadata from the file path
+	fileMetadata, err := extractFileMetadata(filePath)
+	if err != nil {
+		return 0, errors.Wrap(err, "extract file metadata")
+	}
+
+	// Open the file (each goroutine gets its own FD)
+	f, err := dcs.fs.Open(filePath)
+	if err != nil {
+		return 0, errors.Wrap(err, "open file")
+	}
+	defer func() {
+		if closeErr := f.Close(); closeErr != nil {
+			log.WithError(closeErr).WithField("file", filePath).Error("Error closing file during warm cache")
+		}
+	}()
+
+	// Read metadata
+	metadata, err := dcs.metadata(f)
+	if err != nil {
+		return 0, errors.Wrap(err, "read metadata")
+	}
+
+	// Extract indices
+	indices := metadata.indices.all()
+	if len(indices) == 0 {
+		return fileMetadata.epoch, nil // No indices, skip
+	}
+
+	// Build ident and set in cache (thread-safe)
+	dataColumnsIdent := DataColumnsIdent{
+		Root:    fileMetadata.blockRoot,
+		Epoch:   fileMetadata.epoch,
+		Indices: indices,
+	}
+
+	if err := dcs.cache.set(dataColumnsIdent); err != nil {
+		return 0, errors.Wrap(err, "cache set")
+	}
+
+	return fileMetadata.epoch, nil
 }
 
 // Summary returns the DataColumnStorageSummary.

beacon-chain/db/filesystem/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package filesystem
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "filesystem")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/db/filesystem")

beacon-chain/db/kv/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package kv
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "db")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/db/kv")

beacon-chain/db/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package db
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "db")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/db")

beacon-chain/db/pruner/BUILD.bazel

@@ -2,7 +2,10 @@ load("@prysm//tools/go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["pruner.go"],
+    srcs = [
+        "log.go",
+        "pruner.go",
+    ],
     importpath = "github.com/OffchainLabs/prysm/v7/beacon-chain/db/pruner",
     visibility = [
         "//beacon-chain:__subpackages__",

beacon-chain/db/pruner/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package pruner
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/db/pruner")

beacon-chain/db/pruner/pruner.go

@@ -13,8 +13,6 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-var log = logrus.WithField("prefix", "db-pruner")
-
 const (
 	// defaultPrunableBatchSize is the number of slots that can be pruned at once.
 	defaultPrunableBatchSize = 32

beacon-chain/db/slasherkv/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package slasherkv
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "slasherdb")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/db/slasherkv")

beacon-chain/execution/BUILD.bazel

@@ -40,6 +40,7 @@ go_library(
         "//beacon-chain/state/state-native:go_default_library",
         "//beacon-chain/state/stategen:go_default_library",
         "//beacon-chain/verification:go_default_library",
+        "//cmd/beacon-chain/flags:go_default_library",
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",

beacon-chain/execution/engine_client.go

@@ -11,6 +11,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/execution/types"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/verification"
+	"github.com/OffchainLabs/prysm/v7/cmd/beacon-chain/flags"
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
@@ -538,6 +539,10 @@ func (s *Service) GetBlobsV2(ctx context.Context, versionedHashes []common.Hash)
 		return nil, errors.New(fmt.Sprintf("%s is not supported", GetBlobsV2))
 	}
 
+	if flags.Get().DisableGetBlobsV2 {
+		return []*pb.BlobAndProofV2{}, nil
+	}
+
 	result := make([]*pb.BlobAndProofV2, len(versionedHashes))
 	err := s.rpcClient.CallContext(ctx, &result, GetBlobsV2, versionedHashes)
 

beacon-chain/execution/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package execution
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "execution")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/execution")

beacon-chain/forkchoice/doubly-linked-tree/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "errors.go",
         "forkchoice.go",
         "last_root.go",
+        "log.go",
         "metrics.go",
         "node.go",
         "on_tick.go",

beacon-chain/forkchoice/doubly-linked-tree/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package doublylinkedtree
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/forkchoice/doubly-linked-tree")

beacon-chain/forkchoice/doubly-linked-tree/metrics.go

@@ -3,12 +3,9 @@ package doublylinkedtree
 import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/sirupsen/logrus"
 )
 
 var (
-	log = logrus.WithField("prefix", "forkchoice-doublylinkedtree")
-
 	headSlotNumber = promauto.NewGauge(
 		prometheus.GaugeOpts{
 			Name: "doublylinkedtree_head_slot",

beacon-chain/light-client/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package light_client
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "light-client")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/light-client")

beacon-chain/monitor/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "doc.go",
+        "log.go",
         "metrics.go",
         "process_attestation.go",
         "process_block.go",

beacon-chain/monitor/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package monitor
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/monitor")

beacon-chain/monitor/metrics.go

@@ -3,11 +3,9 @@ package monitor
 import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/sirupsen/logrus"
 )
 
 var (
-	log = logrus.WithField("prefix", "monitor")
 	// TODO: The Prometheus gauge vectors and counters in this package deprecate the
 	// corresponding gauge vectors and counters in the validator client.
 

beacon-chain/monitor/process_attestation_test.go

@@ -54,8 +54,8 @@ func TestProcessIncludedAttestationTwoTracked(t *testing.T) {
 		AggregationBits: bitfield.Bitlist{0b11, 0b1},
 	}
 	s.processIncludedAttestation(t.Context(), state, att)
-	wanted1 := "\"Attestation included\" balanceChange=0 correctHead=true correctSource=true correctTarget=true head=0x68656c6c6f2d inclusionSlot=2 newBalance=32000000000 prefix=monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=2"
-	wanted2 := "\"Attestation included\" balanceChange=100000000 correctHead=true correctSource=true correctTarget=true head=0x68656c6c6f2d inclusionSlot=2 newBalance=32000000000 prefix=monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=12"
+	wanted1 := "\"Attestation included\" balanceChange=0 correctHead=true correctSource=true correctTarget=true head=0x68656c6c6f2d inclusionSlot=2 newBalance=32000000000 package=beacon-chain/monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=2"
+	wanted2 := "\"Attestation included\" balanceChange=100000000 correctHead=true correctSource=true correctTarget=true head=0x68656c6c6f2d inclusionSlot=2 newBalance=32000000000 package=beacon-chain/monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=12"
 	require.LogsContain(t, hook, wanted1)
 	require.LogsContain(t, hook, wanted2)
 }
@@ -123,8 +123,8 @@ func TestProcessUnaggregatedAttestationStateCached(t *testing.T) {
 	}
 	require.NoError(t, s.config.StateGen.SaveState(ctx, root, state))
 	s.processUnaggregatedAttestation(t.Context(), att)
-	wanted1 := "\"Processed unaggregated attestation\" head=0x68656c6c6f2d prefix=monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=2"
-	wanted2 := "\"Processed unaggregated attestation\" head=0x68656c6c6f2d prefix=monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=12"
+	wanted1 := "\"Processed unaggregated attestation\" head=0x68656c6c6f2d package=beacon-chain/monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=2"
+	wanted2 := "\"Processed unaggregated attestation\" head=0x68656c6c6f2d package=beacon-chain/monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=12"
 	require.LogsContain(t, hook, wanted1)
 	require.LogsContain(t, hook, wanted2)
 }
@@ -161,7 +161,7 @@ func TestProcessAggregatedAttestationStateNotCached(t *testing.T) {
 		},
 	}
 	s.processAggregatedAttestation(ctx, att)
-	require.LogsContain(t, hook, "\"Processed attestation aggregation\" aggregatorIndex=2 beaconBlockRoot=0x000000000000 prefix=monitor slot=1 sourceRoot=0x68656c6c6f2d targetRoot=0x68656c6c6f2d")
+	require.LogsContain(t, hook, "\"Processed attestation aggregation\" aggregatorIndex=2 beaconBlockRoot=0x000000000000 package=beacon-chain/monitor slot=1 sourceRoot=0x68656c6c6f2d targetRoot=0x68656c6c6f2d")
 	require.LogsContain(t, hook, "Skipping aggregated attestation due to state not found in cache")
 	logrus.SetLevel(logrus.InfoLevel)
 }
@@ -199,9 +199,9 @@ func TestProcessAggregatedAttestationStateCached(t *testing.T) {
 
 	require.NoError(t, s.config.StateGen.SaveState(ctx, root, state))
 	s.processAggregatedAttestation(ctx, att)
-	require.LogsContain(t, hook, "\"Processed attestation aggregation\" aggregatorIndex=2 beaconBlockRoot=0x68656c6c6f2d prefix=monitor slot=1 sourceRoot=0x68656c6c6f2d targetRoot=0x68656c6c6f2d")
-	require.LogsContain(t, hook, "\"Processed aggregated attestation\" head=0x68656c6c6f2d prefix=monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=2")
-	require.LogsDoNotContain(t, hook, "\"Processed aggregated attestation\" head=0x68656c6c6f2d prefix=monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=12")
+	require.LogsContain(t, hook, "\"Processed attestation aggregation\" aggregatorIndex=2 beaconBlockRoot=0x68656c6c6f2d package=beacon-chain/monitor slot=1 sourceRoot=0x68656c6c6f2d targetRoot=0x68656c6c6f2d")
+	require.LogsContain(t, hook, "\"Processed aggregated attestation\" head=0x68656c6c6f2d package=beacon-chain/monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=2")
+	require.LogsDoNotContain(t, hook, "\"Processed aggregated attestation\" head=0x68656c6c6f2d package=beacon-chain/monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=12")
 }
 
 func TestProcessAttestations(t *testing.T) {
@@ -239,8 +239,8 @@ func TestProcessAttestations(t *testing.T) {
 	wrappedBlock, err := blocks.NewBeaconBlock(block)
 	require.NoError(t, err)
 	s.processAttestations(ctx, state, wrappedBlock)
-	wanted1 := "\"Attestation included\" balanceChange=0 correctHead=true correctSource=true correctTarget=true head=0x68656c6c6f2d inclusionSlot=2 newBalance=32000000000 prefix=monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=2"
-	wanted2 := "\"Attestation included\" balanceChange=100000000 correctHead=true correctSource=true correctTarget=true head=0x68656c6c6f2d inclusionSlot=2 newBalance=32000000000 prefix=monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=12"
+	wanted1 := "\"Attestation included\" balanceChange=0 correctHead=true correctSource=true correctTarget=true head=0x68656c6c6f2d inclusionSlot=2 newBalance=32000000000 package=beacon-chain/monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=2"
+	wanted2 := "\"Attestation included\" balanceChange=100000000 correctHead=true correctSource=true correctTarget=true head=0x68656c6c6f2d inclusionSlot=2 newBalance=32000000000 package=beacon-chain/monitor slot=1 source=0x68656c6c6f2d target=0x68656c6c6f2d validatorIndex=12"
 	require.LogsContain(t, hook, wanted1)
 	require.LogsContain(t, hook, wanted2)
 

beacon-chain/monitor/process_block_test.go

@@ -43,7 +43,7 @@ func TestProcessSlashings(t *testing.T) {
 					},
 				},
 			},
-			wantedErr: "\"Proposer slashing was included\" bodyRoot1= bodyRoot2= prefix=monitor proposerIndex=2",
+			wantedErr: "\"Proposer slashing was included\" bodyRoot1= bodyRoot2= package=beacon-chain/monitor proposerIndex=2",
 		},
 		{
 			name: "Proposer slashing an untracked index",
@@ -89,7 +89,7 @@ func TestProcessSlashings(t *testing.T) {
 				},
 			},
 			wantedErr: "\"Attester slashing was included\" attestationSlot1=0 attestationSlot2=0 attesterIndex=1 " +
-				"beaconBlockRoot1=0x000000000000 beaconBlockRoot2=0x000000000000 blockInclusionSlot=0 prefix=monitor sourceEpoch1=1 sourceEpoch2=0 targetEpoch1=0 targetEpoch2=0",
+				"beaconBlockRoot1=0x000000000000 beaconBlockRoot2=0x000000000000 blockInclusionSlot=0 package=beacon-chain/monitor sourceEpoch1=1 sourceEpoch2=0 targetEpoch1=0 targetEpoch2=0",
 		},
 		{
 			name: "Attester slashing untracked index",
@@ -149,7 +149,7 @@ func TestProcessProposedBlock(t *testing.T) {
 				StateRoot:     bytesutil.PadTo([]byte("state-world"), 32),
 				Body:          &ethpb.BeaconBlockBody{},
 			},
-			wantedErr: "\"Proposed beacon block was included\" balanceChange=100000000 blockRoot=0x68656c6c6f2d newBalance=32000000000 parentRoot=0x68656c6c6f2d prefix=monitor proposerIndex=12 slot=6 version=0",
+			wantedErr: "\"Proposed beacon block was included\" balanceChange=100000000 blockRoot=0x68656c6c6f2d newBalance=32000000000 package=beacon-chain/monitor parentRoot=0x68656c6c6f2d proposerIndex=12 slot=6 version=0",
 		},
 		{
 			name: "Block proposed by untracked validator",
@@ -224,10 +224,10 @@ func TestProcessBlock_AllEventsTrackedVals(t *testing.T) {
 	root, err := b.GetBlock().HashTreeRoot()
 	require.NoError(t, err)
 	require.NoError(t, s.config.StateGen.SaveState(ctx, root, genesis))
-	wanted1 := fmt.Sprintf("\"Proposed beacon block was included\" balanceChange=100000000 blockRoot=%#x newBalance=32000000000 parentRoot=0xf732eaeb7fae prefix=monitor proposerIndex=15 slot=1 version=1", bytesutil.Trunc(root[:]))
-	wanted2 := fmt.Sprintf("\"Proposer slashing was included\" bodyRoot1=0x000100000000 bodyRoot2=0x000200000000 prefix=monitor proposerIndex=%d slashingSlot=0 slot=1", idx)
-	wanted3 := "\"Sync committee contribution included\" balanceChange=0 contribCount=3 expectedContribCount=3 newBalance=32000000000 prefix=monitor validatorIndex=1"
-	wanted4 := "\"Sync committee contribution included\" balanceChange=0 contribCount=1 expectedContribCount=1 newBalance=32000000000 prefix=monitor validatorIndex=2"
+	wanted1 := fmt.Sprintf("\"Proposed beacon block was included\" balanceChange=100000000 blockRoot=%#x newBalance=32000000000 package=beacon-chain/monitor parentRoot=0xf732eaeb7fae proposerIndex=15 slot=1 version=1", bytesutil.Trunc(root[:]))
+	wanted2 := fmt.Sprintf("\"Proposer slashing was included\" bodyRoot1=0x000100000000 bodyRoot2=0x000200000000 package=beacon-chain/monitor proposerIndex=%d slashingSlot=0 slot=1", idx)
+	wanted3 := "\"Sync committee contribution included\" balanceChange=0 contribCount=3 expectedContribCount=3 newBalance=32000000000 package=beacon-chain/monitor validatorIndex=1"
+	wanted4 := "\"Sync committee contribution included\" balanceChange=0 contribCount=1 expectedContribCount=1 newBalance=32000000000 package=beacon-chain/monitor validatorIndex=2"
 	wrapped, err := blocks.NewSignedBeaconBlock(b)
 	require.NoError(t, err)
 	s.processBlock(ctx, wrapped)
@@ -279,7 +279,7 @@ func TestLogAggregatedPerformance(t *testing.T) {
 	s.logAggregatedPerformance()
 	wanted := "\"Aggregated performance since launch\" attestationInclusion=\"80.00%\"" +
 		" averageInclusionDistance=1.2 balanceChangePct=\"0.95%\" correctlyVotedHeadPct=\"66.67%\" " +
-		"correctlyVotedSourcePct=\"91.67%\" correctlyVotedTargetPct=\"100.00%\" prefix=monitor startBalance=31700000000 " +
+		"correctlyVotedSourcePct=\"91.67%\" correctlyVotedTargetPct=\"100.00%\" package=beacon-chain/monitor startBalance=31700000000 " +
 		"startEpoch=0 totalAggregations=0 totalProposedBlocks=1 totalRequested=15 totalSyncContributions=0 " +
 		"validatorIndex=1"
 	require.LogsContain(t, hook, wanted)

beacon-chain/monitor/process_exit_test.go

@@ -43,7 +43,7 @@ func TestProcessExitsFromBlockTrackedIndices(t *testing.T) {
 	wb, err := blocks.NewBeaconBlock(block)
 	require.NoError(t, err)
 	s.processExitsFromBlock(wb)
-	require.LogsContain(t, hook, "\"Voluntary exit was included\" prefix=monitor slot=0 validatorIndex=2")
+	require.LogsContain(t, hook, "\"Voluntary exit was included\" package=beacon-chain/monitor slot=0 validatorIndex=2")
 }
 
 func TestProcessExitsFromBlockUntrackedIndices(t *testing.T) {
@@ -99,7 +99,7 @@ func TestProcessExitP2PTrackedIndices(t *testing.T) {
 		Signature: make([]byte, 96),
 	}
 	s.processExit(exit)
-	require.LogsContain(t, hook, "\"Voluntary exit was processed\" prefix=monitor validatorIndex=1")
+	require.LogsContain(t, hook, "\"Voluntary exit was processed\" package=beacon-chain/monitor validatorIndex=1")
 }
 
 func TestProcessExitP2PUntrackedIndices(t *testing.T) {

beacon-chain/monitor/process_sync_committee_test.go

@@ -22,7 +22,7 @@ func TestProcessSyncCommitteeContribution(t *testing.T) {
 	}
 
 	s.processSyncCommitteeContribution(contrib)
-	require.LogsContain(t, hook, "\"Sync committee aggregation processed\" prefix=monitor validatorIndex=1")
+	require.LogsContain(t, hook, "\"Sync committee aggregation processed\" package=beacon-chain/monitor validatorIndex=1")
 	require.LogsDoNotContain(t, hook, "validatorIndex=2")
 }
 
@@ -53,7 +53,7 @@ func TestProcessSyncAggregate(t *testing.T) {
 	require.NoError(t, err)
 
 	s.processSyncAggregate(beaconState, wrappedBlock)
-	require.LogsContain(t, hook, "\"Sync committee contribution included\" balanceChange=0 contribCount=1 expectedContribCount=4 newBalance=32000000000 prefix=monitor validatorIndex=1")
-	require.LogsContain(t, hook, "\"Sync committee contribution included\" balanceChange=100000000 contribCount=2 expectedContribCount=2 newBalance=32000000000 prefix=monitor validatorIndex=12")
+	require.LogsContain(t, hook, "\"Sync committee contribution included\" balanceChange=0 contribCount=1 expectedContribCount=4 newBalance=32000000000 package=beacon-chain/monitor validatorIndex=1")
+	require.LogsContain(t, hook, "\"Sync committee contribution included\" balanceChange=100000000 contribCount=2 expectedContribCount=2 newBalance=32000000000 package=beacon-chain/monitor validatorIndex=12")
 	require.LogsDoNotContain(t, hook, "validatorIndex=2")
 }

beacon-chain/monitor/service_test.go

@@ -148,7 +148,7 @@ func TestStart(t *testing.T) {
 	// wait for Logrus
 	time.Sleep(1000 * time.Millisecond)
 	require.LogsContain(t, hook, "Synced to head epoch, starting reporting performance")
-	require.LogsContain(t, hook, "\"Starting service\" prefix=monitor validatorIndices=\"[1 2 12 15]\"")
+	require.LogsContain(t, hook, "\"Starting service\" package=beacon-chain/monitor validatorIndices=\"[1 2 12 15]\"")
 	s.Lock()
 	require.Equal(t, s.isLogging, true, "monitor is not running")
 	s.Unlock()
@@ -235,7 +235,7 @@ func TestMonitorRoutine(t *testing.T) {
 
 	// Wait for Logrus
 	time.Sleep(1000 * time.Millisecond)
-	wanted1 := fmt.Sprintf("\"Proposed beacon block was included\" balanceChange=100000000 blockRoot=%#x newBalance=32000000000 parentRoot=0xf732eaeb7fae prefix=monitor proposerIndex=15 slot=1 version=1", bytesutil.Trunc(root[:]))
+	wanted1 := fmt.Sprintf("\"Proposed beacon block was included\" balanceChange=100000000 blockRoot=%#x newBalance=32000000000 package=beacon-chain/monitor parentRoot=0xf732eaeb7fae proposerIndex=15 slot=1 version=1", bytesutil.Trunc(root[:]))
 	require.LogsContain(t, hook, wanted1)
 
 }

beacon-chain/node/log.go

@@ -1,5 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package node
 
 import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "node")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/node")

beacon-chain/node/registration/log.go

@@ -1 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package registration
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/node/registration")

beacon-chain/node/registration/p2p.go

@@ -7,7 +7,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/cmd"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 	"github.com/urfave/cli/v2"
 	"gopkg.in/yaml.v2"
 )

beacon-chain/operations/attestations/kv/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "aggregated.go",
         "block.go",
         "kv.go",
+        "log.go",
         "seen_bits.go",
         "unaggregated.go",
     ],

beacon-chain/operations/attestations/kv/aggregated.go

@@ -13,7 +13,6 @@ import (
 	attaggregation "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1/attestation/aggregation/attestations"
 	"github.com/OffchainLabs/prysm/v7/runtime/version"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 )
 
 // AggregateUnaggregatedAttestations aggregates the unaggregated attestations and saves the

beacon-chain/operations/attestations/kv/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package kv
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/operations/attestations/kv")

beacon-chain/operations/attestations/kv/unaggregated.go

@@ -9,7 +9,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1/attestation"
 	"github.com/OffchainLabs/prysm/v7/runtime/version"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 )
 
 // SaveUnaggregatedAttestation saves an unaggregated attestation in cache.

beacon-chain/operations/attestations/log.go

@@ -1,7 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package attestations
 
-import (
-	"github.com/sirupsen/logrus"
-)
+import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "pool/attestations")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/operations/attestations")

beacon-chain/operations/blstoexec/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "doc.go",
+        "log.go",
         "pool.go",
     ],
     importpath = "github.com/OffchainLabs/prysm/v7/beacon-chain/operations/blstoexec",

beacon-chain/operations/blstoexec/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package blstoexec
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/operations/blstoexec")

beacon-chain/operations/slashings/log.go

@@ -1,7 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
 package slashings
 
-import (
-	"github.com/sirupsen/logrus"
-)
+import "github.com/sirupsen/logrus"
 
-var log = logrus.WithField("prefix", "pool/slashings")
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/operations/slashings")

beacon-chain/operations/voluntaryexits/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "doc.go",
+        "log.go",
         "pool.go",
     ],
     importpath = "github.com/OffchainLabs/prysm/v7/beacon-chain/operations/voluntaryexits",

beacon-chain/operations/voluntaryexits/log.go

@@ -0,0 +1,9 @@
+// Code generated by hack/gen-logs.sh; DO NOT EDIT.
+// This file is created and regenerated automatically. Anything added here might get removed.
+package voluntaryexits
+
+import "github.com/sirupsen/logrus"
+
+// The prefix for logs from this package will be the text after the last slash in the package path.
+// If you wish to change this, you should add your desired name in the runtime/logging/logrus-prefixed-formatter/prefix-replacement.go file.
+var log = logrus.WithField("package", "beacon-chain/operations/voluntaryexits")

beacon-chain/p2p/BUILD.bazel

@@ -13,12 +13,15 @@ go_library(
         "doc.go",
         "fork.go",
         "fork_watcher.go",
+        "gossip_peer_controller.go",
+        "gossip_peer_crawler.go",
         "gossip_scoring_params.go",
         "gossip_topic_mappings.go",
         "handshake.go",
         "info.go",
         "interfaces.go",
         "log.go",
+        "log_helpers.go",
         "message_id.go",
         "monitoring.go",
         "options.go",
@@ -51,6 +54,7 @@ go_library(
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/kv:go_default_library",
         "//beacon-chain/p2p/encoder:go_default_library",
+        "//beacon-chain/p2p/gossipcrawler:go_default_library",
         "//beacon-chain/p2p/peers:go_default_library",
         "//beacon-chain/p2p/peers/peerdata:go_default_library",
         "//beacon-chain/p2p/peers/scorers:go_default_library",
@@ -115,6 +119,7 @@ go_library(
         "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@org_golang_google_protobuf//proto:go_default_library",
+        "@org_golang_x_sync//semaphore:go_default_library",
     ],
 )
 
@@ -128,6 +133,8 @@ go_test(
         "dial_relay_node_test.go",
         "discovery_test.go",
         "fork_test.go",
+        "gossip_peer_controller_test.go",
+        "gossip_peer_crawler_test.go",
         "gossip_scoring_params_test.go",
         "gossip_topic_mappings_test.go",
         "message_id_test.go",
@@ -154,9 +161,11 @@ go_test(
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/peerdas:go_default_library",
         "//beacon-chain/core/signing:go_default_library",
+        "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/iface:go_default_library",
         "//beacon-chain/db/testing:go_default_library",
         "//beacon-chain/p2p/encoder:go_default_library",
+        "//beacon-chain/p2p/gossipcrawler:go_default_library",
         "//beacon-chain/p2p/peers:go_default_library",
         "//beacon-chain/p2p/peers/peerdata:go_default_library",
         "//beacon-chain/p2p/peers/scorers:go_default_library",
@@ -203,8 +212,10 @@ go_test(
         "@com_github_libp2p_go_libp2p_pubsub//pb:go_default_library",
         "@com_github_multiformats_go_multiaddr//:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus/testutil:go_default_library",
         "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
         "@com_github_sirupsen_logrus//hooks/test:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
         "@org_golang_google_protobuf//proto:go_default_library",
     ],
 )

beacon-chain/p2p/broadcaster.go

@@ -22,6 +22,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/monitoring/tracing/trace"
 	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
 	"github.com/OffchainLabs/prysm/v7/time/slots"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/pkg/errors"
 	ssz "github.com/prysmaticlabs/fastssz"
 	"github.com/sirupsen/logrus"
@@ -60,7 +61,10 @@ func (s *Service) Broadcast(ctx context.Context, msg proto.Message) error {
 	if !ok {
 		return errors.Errorf("message of %T does not support marshaller interface", msg)
 	}
-	return s.broadcastObject(ctx, castMsg, fmt.Sprintf(topic, forkDigest))
+
+	fullTopic := fmt.Sprintf(topic, forkDigest) + s.Encoding().ProtocolSuffix()
+
+	return s.broadcastObject(ctx, castMsg, fullTopic)
 }
 
 // BroadcastAttestation broadcasts an attestation to the p2p network, the message is assumed to be
@@ -106,6 +110,7 @@ func (s *Service) BroadcastSyncCommitteeMessage(ctx context.Context, subnet uint
 }
 
 func (s *Service) internalBroadcastAttestation(ctx context.Context, subnet uint64, att ethpb.Att, forkDigest [fieldparams.VersionLength]byte) {
+	topic := AttestationSubnetTopic(forkDigest, subnet)
 	_, span := trace.StartSpan(ctx, "p2p.internalBroadcastAttestation")
 	defer span.End()
 	ctx = trace.NewContext(context.Background(), span) // clear parent context / deadline.
@@ -116,7 +121,7 @@ func (s *Service) internalBroadcastAttestation(ctx context.Context, subnet uint6
 
 	// Ensure we have peers with this subnet.
 	s.subnetLocker(subnet).RLock()
-	hasPeer := s.hasPeerWithSubnet(attestationToTopic(subnet, forkDigest))
+	hasPeer := s.hasPeerWithTopic(topic)
 	s.subnetLocker(subnet).RUnlock()
 
 	span.SetAttributes(
@@ -131,7 +136,7 @@ func (s *Service) internalBroadcastAttestation(ctx context.Context, subnet uint6
 			s.subnetLocker(subnet).Lock()
 			defer s.subnetLocker(subnet).Unlock()
 
-			if err := s.FindAndDialPeersWithSubnets(ctx, AttestationSubnetTopicFormat, forkDigest, minimumPeersPerSubnetForBroadcast, map[uint64]bool{subnet: true}); err != nil {
+			if err := s.gossipDialer.DialPeersForTopicBlocking(ctx, topic, minimumPeersPerSubnetForBroadcast); err != nil {
 				return errors.Wrap(err, "find peers with subnets")
 			}
 
@@ -153,13 +158,14 @@ func (s *Service) internalBroadcastAttestation(ctx context.Context, subnet uint6
 		return
 	}
 
-	if err := s.broadcastObject(ctx, att, attestationToTopic(subnet, forkDigest)); err != nil {
+	if err := s.broadcastObject(ctx, att, topic); err != nil {
 		log.WithError(err).Error("Failed to broadcast attestation")
 		tracing.AnnotateError(span, err)
 	}
 }
 
 func (s *Service) broadcastSyncCommittee(ctx context.Context, subnet uint64, sMsg *ethpb.SyncCommitteeMessage, forkDigest [fieldparams.VersionLength]byte) {
+	topic := SyncCommitteeSubnetTopic(forkDigest, subnet)
 	_, span := trace.StartSpan(ctx, "p2p.broadcastSyncCommittee")
 	defer span.End()
 	ctx = trace.NewContext(context.Background(), span) // clear parent context / deadline.
@@ -173,7 +179,7 @@ func (s *Service) broadcastSyncCommittee(ctx context.Context, subnet uint64, sMs
 	// to ensure that we can reuse the same subnet locker.
 	wrappedSubIdx := subnet + syncLockerVal
 	s.subnetLocker(wrappedSubIdx).RLock()
-	hasPeer := s.hasPeerWithSubnet(syncCommitteeToTopic(subnet, forkDigest))
+	hasPeer := s.hasPeerWithTopic(topic)
 	s.subnetLocker(wrappedSubIdx).RUnlock()
 
 	span.SetAttributes(
@@ -187,7 +193,7 @@ func (s *Service) broadcastSyncCommittee(ctx context.Context, subnet uint64, sMs
 		if err := func() error {
 			s.subnetLocker(wrappedSubIdx).Lock()
 			defer s.subnetLocker(wrappedSubIdx).Unlock()
-			if err := s.FindAndDialPeersWithSubnets(ctx, SyncCommitteeSubnetTopicFormat, forkDigest, minimumPeersPerSubnetForBroadcast, map[uint64]bool{subnet: true}); err != nil {
+			if err := s.gossipDialer.DialPeersForTopicBlocking(ctx, topic, minimumPeersPerSubnetForBroadcast); err != nil {
 				return errors.Wrap(err, "find peers with subnets")
 			}
 
@@ -205,7 +211,7 @@ func (s *Service) broadcastSyncCommittee(ctx context.Context, subnet uint64, sMs
 		return
 	}
 
-	if err := s.broadcastObject(ctx, sMsg, syncCommitteeToTopic(subnet, forkDigest)); err != nil {
+	if err := s.broadcastObject(ctx, sMsg, topic); err != nil {
 		log.WithError(err).Error("Failed to broadcast sync committee message")
 		tracing.AnnotateError(span, err)
 	}
@@ -233,6 +239,7 @@ func (s *Service) BroadcastBlob(ctx context.Context, subnet uint64, blob *ethpb.
 }
 
 func (s *Service) internalBroadcastBlob(ctx context.Context, subnet uint64, blobSidecar *ethpb.BlobSidecar, forkDigest [fieldparams.VersionLength]byte) {
+	topic := BlobSubnetTopic(forkDigest, subnet)
 	_, span := trace.StartSpan(ctx, "p2p.internalBroadcastBlob")
 	defer span.End()
 	ctx = trace.NewContext(context.Background(), span) // clear parent context / deadline.
@@ -243,7 +250,7 @@ func (s *Service) internalBroadcastBlob(ctx context.Context, subnet uint64, blob
 
 	wrappedSubIdx := subnet + blobSubnetLockerVal
 	s.subnetLocker(wrappedSubIdx).RLock()
-	hasPeer := s.hasPeerWithSubnet(blobSubnetToTopic(subnet, forkDigest))
+	hasPeer := s.hasPeerWithTopic(topic)
 	s.subnetLocker(wrappedSubIdx).RUnlock()
 
 	if !hasPeer {
@@ -252,7 +259,7 @@ func (s *Service) internalBroadcastBlob(ctx context.Context, subnet uint64, blob
 			s.subnetLocker(wrappedSubIdx).Lock()
 			defer s.subnetLocker(wrappedSubIdx).Unlock()
 
-			if err := s.FindAndDialPeersWithSubnets(ctx, BlobSubnetTopicFormat, forkDigest, minimumPeersPerSubnetForBroadcast, map[uint64]bool{subnet: true}); err != nil {
+			if err := s.gossipDialer.DialPeersForTopicBlocking(ctx, topic, minimumPeersPerSubnetForBroadcast); err != nil {
 				return errors.Wrap(err, "find peers with subnets")
 			}
 
@@ -264,7 +271,7 @@ func (s *Service) internalBroadcastBlob(ctx context.Context, subnet uint64, blob
 		}
 	}
 
-	if err := s.broadcastObject(ctx, blobSidecar, blobSubnetToTopic(subnet, forkDigest)); err != nil {
+	if err := s.broadcastObject(ctx, blobSidecar, topic); err != nil {
 		log.WithError(err).Error("Failed to broadcast blob sidecar")
 		tracing.AnnotateError(span, err)
 	}
@@ -293,7 +300,7 @@ func (s *Service) BroadcastLightClientOptimisticUpdate(ctx context.Context, upda
 	}
 
 	digest := params.ForkDigest(slots.ToEpoch(update.AttestedHeader().Beacon().Slot))
-	if err := s.broadcastObject(ctx, update, lcOptimisticToTopic(digest)); err != nil {
+	if err := s.broadcastObject(ctx, update, LcOptimisticToTopic(digest)); err != nil {
 		log.WithError(err).Debug("Failed to broadcast light client optimistic update")
 		err := errors.Wrap(err, "could not publish message")
 		tracing.AnnotateError(span, err)
@@ -327,7 +334,7 @@ func (s *Service) BroadcastLightClientFinalityUpdate(ctx context.Context, update
 	}
 
 	forkDigest := params.ForkDigest(slots.ToEpoch(update.AttestedHeader().Beacon().Slot))
-	if err := s.broadcastObject(ctx, update, lcFinalityToTopic(forkDigest)); err != nil {
+	if err := s.broadcastObject(ctx, update, LcFinalityToTopic(forkDigest)); err != nil {
 		log.WithError(err).Debug("Failed to broadcast light client finality update")
 		err := errors.Wrap(err, "could not publish message")
 		tracing.AnnotateError(span, err)
@@ -357,58 +364,67 @@ func (s *Service) BroadcastDataColumnSidecars(ctx context.Context, sidecars []bl
 	return nil
 }
 
-// broadcastDataColumnSidecars broadcasts multiple data column sidecars to the p2p network, after ensuring
-// there is at least one peer in each needed subnet. If not, it will attempt to find one before broadcasting.
-// It returns when all broadcasts are complete, or the context is cancelled (whichever comes first).
+// broadcastDataColumnSidecars broadcasts multiple data column sidecars to the p2p network.
+// For sidecars with available peers, it uses batch publishing.
+// For sidecars without peers, it finds peers first and then publishes individually.
+// Both paths run in parallel. It returns when all broadcasts are complete, or the context is cancelled.
 func (s *Service) broadcastDataColumnSidecars(ctx context.Context, forkDigest [fieldparams.VersionLength]byte, sidecars []blocks.VerifiedRODataColumn) {
 	type rootAndIndex struct {
 		root  [fieldparams.RootLength]byte
 		index uint64
 	}
 
-	var (
-		wg      sync.WaitGroup
-		timings sync.Map
-	)
-
+	var timings sync.Map
 	logLevel := logrus.GetLevel()
-
 	slotPerRoot := make(map[[fieldparams.RootLength]byte]primitives.Slot, 1)
+
+	topicFunc := func(sidecar blocks.VerifiedRODataColumn) (topic string, wrappedSubIdx uint64, subnet uint64) {
+		subnet = peerdas.ComputeSubnetForDataColumnSidecar(sidecar.Index)
+		topic = DataColumnSubnetTopic(forkDigest, subnet)
+		wrappedSubIdx = subnet + dataColumnSubnetVal
+		return
+	}
+
+	sidecarsWithPeers := make([]blocks.VerifiedRODataColumn, 0, len(sidecars))
+	var sidecarsWithoutPeers []blocks.VerifiedRODataColumn
+
+	// Categorize sidecars by peer availability.
 	for _, sidecar := range sidecars {
 		slotPerRoot[sidecar.BlockRoot()] = sidecar.Slot()
 
-		wg.Go(func() {
-			// Add tracing to the function.
-			ctx, span := trace.StartSpan(s.ctx, "p2p.broadcastDataColumnSidecars")
+		topic, wrappedSubIdx, _ := topicFunc(sidecar)
+		// Check if we have a peer for this subnet (use RLock for read-only check).
+		mu := s.subnetLocker(wrappedSubIdx)
+		mu.RLock()
+		hasPeer := s.hasPeerWithTopic(topic)
+		mu.RUnlock()
+
+		if hasPeer {
+			sidecarsWithPeers = append(sidecarsWithPeers, sidecar)
+			continue
+		}
+
+		sidecarsWithoutPeers = append(sidecarsWithoutPeers, sidecar)
+	}
+
+	var batchWg, individualWg sync.WaitGroup
+
+	// Batch publish sidecars that already have peers
+	var messageBatch pubsub.MessageBatch
+	for _, sidecar := range sidecarsWithPeers {
+		batchWg.Go(func() {
+			_, span := trace.StartSpan(ctx, "p2p.broadcastDataColumnSidecars")
+			ctx := trace.NewContext(s.ctx, span)
 			defer span.End()
 
-			// Compute the subnet for this data column sidecar.
-			subnet := peerdas.ComputeSubnetForDataColumnSidecar(sidecar.Index)
+			topic, _, _ := topicFunc(sidecar)
 
-			// Build the topic corresponding to subnet column subnet and this fork digest.
-			topic := dataColumnSubnetToTopic(subnet, forkDigest)
-
-			// Compute the wrapped subnet index.
-			wrappedSubIdx := subnet + dataColumnSubnetVal
-
-			// Find peers if needed.
-			if err := s.findPeersIfNeeded(ctx, wrappedSubIdx, DataColumnSubnetTopicFormat, forkDigest, subnet); err != nil {
+			if err := s.batchObject(ctx, &messageBatch, sidecar, topic); err != nil {
 				tracing.AnnotateError(span, err)
-				log.WithError(err).Error("Cannot find peers if needed")
+				log.WithError(err).Error("Cannot batch data column sidecar")
 				return
 			}
 
-			// Broadcast the data column sidecar to the network.
-			if err := s.broadcastObject(ctx, sidecar, topic); err != nil {
-				tracing.AnnotateError(span, err)
-				log.WithError(err).Error("Cannot broadcast data column sidecar")
-				return
-			}
-
-			// Increase the number of successful broadcasts.
-			dataColumnSidecarBroadcasts.Inc()
-
-			// Record the timing for log purposes.
 			if logLevel >= logrus.DebugLevel {
 				root := sidecar.BlockRoot()
 				timings.Store(rootAndIndex{root: root, index: sidecar.Index}, time.Now())
@@ -416,8 +432,50 @@ func (s *Service) broadcastDataColumnSidecars(ctx context.Context, forkDigest [f
 		})
 	}
 
-	// Wait for all broadcasts to finish.
-	wg.Wait()
+	// For sidecars without peers, find peers and publish individually (no batching).
+	for _, sidecar := range sidecarsWithoutPeers {
+		individualWg.Go(func() {
+			_, span := trace.StartSpan(ctx, "p2p.broadcastDataColumnSidecars")
+			ctx := trace.NewContext(s.ctx, span)
+			defer span.End()
+
+			topic, wrappedSubIdx, _ := topicFunc(sidecar)
+
+			// Find peers for this sidecar's subnet.
+			if err := s.findPeersIfNeeded(ctx, wrappedSubIdx, topic); err != nil {
+				tracing.AnnotateError(span, err)
+				log.WithError(err).Error("Cannot find peers if needed")
+				return
+			}
+
+			// Publish individually (not batched) since we just found peers.
+			if err := s.broadcastObject(ctx, sidecar, topic); err != nil {
+				tracing.AnnotateError(span, err)
+				log.WithError(err).Error("Cannot broadcast data column sidecar")
+				return
+			}
+
+			dataColumnSidecarBroadcasts.Inc()
+
+			if logLevel >= logrus.DebugLevel {
+				root := sidecar.BlockRoot()
+				timings.Store(rootAndIndex{root: root, index: sidecar.Index}, time.Now())
+			}
+		})
+	}
+
+	// Wait for batch to be populated, then publish.
+	batchWg.Wait()
+	if len(sidecarsWithPeers) > 0 {
+		if err := s.pubsub.PublishBatch(&messageBatch); err != nil {
+			log.WithError(err).Error("Cannot publish batch for data column sidecars")
+		} else {
+			dataColumnSidecarBroadcasts.Add(float64(len(sidecarsWithPeers)))
+		}
+	}
+
+	// Wait for all individual publishes to complete.
+	individualWg.Wait()
 
 	// The rest of this function is only for debug logging purposes.
 	if logLevel < logrus.DebugLevel {
@@ -487,45 +545,51 @@ func (s *Service) broadcastDataColumnSidecars(ctx context.Context, forkDigest [f
 func (s *Service) findPeersIfNeeded(
 	ctx context.Context,
 	wrappedSubIdx uint64,
-	topicFormat string,
-	forkDigest [fieldparams.VersionLength]byte,
-	subnet uint64,
+	topic string,
 ) error {
 	// Sending a data column sidecar to only one peer is not ideal,
 	// but it ensures at least one peer receives it.
 	s.subnetLocker(wrappedSubIdx).Lock()
 	defer s.subnetLocker(wrappedSubIdx).Unlock()
 
-	// No peers found, attempt to find peers with this subnet.
-	if err := s.FindAndDialPeersWithSubnets(ctx, topicFormat, forkDigest, minimumPeersPerSubnetForBroadcast, map[uint64]bool{subnet: true}); err != nil {
+	if err := s.gossipDialer.DialPeersForTopicBlocking(ctx, topic, minimumPeersPerSubnetForBroadcast); err != nil {
 		return errors.Wrap(err, "find peers with subnet")
 	}
-
 	return nil
 }
 
-// method to broadcast messages to other peers in our gossip mesh.
-func (s *Service) broadcastObject(ctx context.Context, obj ssz.Marshaler, topic string) error {
+// encodeGossipMessage encodes an object for gossip transmission.
+// It returns the encoded bytes and the full topic with protocol suffix.
+func (s *Service) encodeGossipMessage(obj ssz.Marshaler) ([]byte, error) {
+	buf := new(bytes.Buffer)
+	if _, err := s.Encoding().EncodeGossip(buf, obj); err != nil {
+		return nil, fmt.Errorf("could not encode message: %w", err)
+	}
+	return buf.Bytes(), nil
+}
+
+// broadcastObject broadcasts a message to other peers in our gossip mesh.
+func (s *Service) broadcastObject(ctx context.Context, obj ssz.Marshaler, fullTopic string) error {
 	ctx, span := trace.StartSpan(ctx, "p2p.broadcastObject")
 	defer span.End()
 
-	span.SetAttributes(trace.StringAttribute("topic", topic))
+	span.SetAttributes(trace.StringAttribute("topic", fullTopic))
 
-	buf := new(bytes.Buffer)
-	if _, err := s.Encoding().EncodeGossip(buf, obj); err != nil {
-		err := errors.Wrap(err, "could not encode message")
+	data, err := s.encodeGossipMessage(obj)
+	if err != nil {
 		tracing.AnnotateError(span, err)
 		return err
 	}
 
 	if span.IsRecording() {
-		id := hash.FastSum64(buf.Bytes())
-		messageLen := int64(buf.Len())
+		id := hash.FastSum64(data)
+		messageLen := int64(len(data))
 		// lint:ignore uintcast -- It's safe to do this for tracing.
 		iid := int64(id)
 		span = trace.AddMessageSendEvent(span, iid, messageLen /*uncompressed*/, messageLen /*compressed*/)
 	}
-	if err := s.PublishToTopic(ctx, topic+s.Encoding().ProtocolSuffix(), buf.Bytes()); err != nil {
+
+	if err := s.PublishToTopic(ctx, fullTopic, data); err != nil {
 		err := errors.Wrap(err, "could not publish message")
 		tracing.AnnotateError(span, err)
 		return err
@@ -533,26 +597,32 @@ func (s *Service) broadcastObject(ctx context.Context, obj ssz.Marshaler, topic
 	return nil
 }
 
-func attestationToTopic(subnet uint64, forkDigest [fieldparams.VersionLength]byte) string {
-	return fmt.Sprintf(AttestationSubnetTopicFormat, forkDigest, subnet)
-}
+// batchObject adds an object to a message batch for a future broadcast.
+// The caller MUST publish the batch after all messages have been added.
+func (s *Service) batchObject(ctx context.Context, batch *pubsub.MessageBatch, obj ssz.Marshaler, fullTopic string) error {
+	ctx, span := trace.StartSpan(ctx, "p2p.batchObject")
+	defer span.End()
 
-func syncCommitteeToTopic(subnet uint64, forkDigest [fieldparams.VersionLength]byte) string {
-	return fmt.Sprintf(SyncCommitteeSubnetTopicFormat, forkDigest, subnet)
-}
+	span.SetAttributes(trace.StringAttribute("topic", fullTopic))
 
-func blobSubnetToTopic(subnet uint64, forkDigest [fieldparams.VersionLength]byte) string {
-	return fmt.Sprintf(BlobSubnetTopicFormat, forkDigest, subnet)
-}
+	data, err := s.encodeGossipMessage(obj)
+	if err != nil {
+		tracing.AnnotateError(span, err)
+		return err
+	}
 
-func lcOptimisticToTopic(forkDigest [4]byte) string {
-	return fmt.Sprintf(LightClientOptimisticUpdateTopicFormat, forkDigest)
-}
+	if span.IsRecording() {
+		id := hash.FastSum64(data)
+		messageLen := int64(len(data))
+		// lint:ignore uintcast -- It's safe to do this for tracing.
+		iid := int64(id)
+		span = trace.AddMessageSendEvent(span, iid, messageLen /*uncompressed*/, messageLen /*compressed*/)
+	}
 
-func lcFinalityToTopic(forkDigest [4]byte) string {
-	return fmt.Sprintf(LightClientFinalityUpdateTopicFormat, forkDigest)
-}
-
-func dataColumnSubnetToTopic(subnet uint64, forkDigest [fieldparams.VersionLength]byte) string {
-	return fmt.Sprintf(DataColumnSubnetTopicFormat, forkDigest, subnet)
+	if err := s.addToBatch(ctx, batch, fullTopic, data); err != nil {
+		err := errors.Wrap(err, "could not publish message")
+		tracing.AnnotateError(span, err)
+		return err
+	}
+	return nil
 }

beacon-chain/p2p/broadcaster_test.go

@@ -30,8 +30,11 @@ import (
 	"github.com/OffchainLabs/prysm/v7/testing/require"
 	"github.com/OffchainLabs/prysm/v7/testing/util"
 	"github.com/OffchainLabs/prysm/v7/time/slots"
+	"github.com/ethereum/go-ethereum/p2p/enode"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/protocol"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -109,6 +112,7 @@ func TestService_Attestation_Subnet(t *testing.T) {
 	if gtm := GossipTypeMapping[reflect.TypeFor[*ethpb.Attestation]()]; gtm != AttestationSubnetTopicFormat {
 		t.Errorf("Constant is out of date. Wanted %s, got %s", AttestationSubnetTopicFormat, gtm)
 	}
+	s := Service{}
 
 	tests := []struct {
 		att   *ethpb.Attestation
@@ -121,7 +125,7 @@ func TestService_Attestation_Subnet(t *testing.T) {
 					Slot:           2,
 				},
 			},
-			topic: "/eth2/00000000/beacon_attestation_2",
+			topic: "/eth2/00000000/beacon_attestation_2" + s.Encoding().ProtocolSuffix(),
 		},
 		{
 			att: &ethpb.Attestation{
@@ -130,7 +134,7 @@ func TestService_Attestation_Subnet(t *testing.T) {
 					Slot:           10,
 				},
 			},
-			topic: "/eth2/00000000/beacon_attestation_21",
+			topic: "/eth2/00000000/beacon_attestation_21" + s.Encoding().ProtocolSuffix(),
 		},
 		{
 			att: &ethpb.Attestation{
@@ -139,12 +143,12 @@ func TestService_Attestation_Subnet(t *testing.T) {
 					Slot:           529,
 				},
 			},
-			topic: "/eth2/00000000/beacon_attestation_8",
+			topic: "/eth2/00000000/beacon_attestation_8" + s.Encoding().ProtocolSuffix(),
 		},
 	}
 	for _, tt := range tests {
 		subnet := helpers.ComputeSubnetFromCommitteeAndSlot(100, tt.att.Data.CommitteeIndex, tt.att.Data.Slot)
-		assert.Equal(t, tt.topic, attestationToTopic(subnet, [4]byte{} /* fork digest */), "Wrong topic")
+		assert.Equal(t, tt.topic, AttestationSubnetTopic([4]byte{}, subnet), "Wrong topic")
 	}
 }
 
@@ -173,14 +177,12 @@ func TestService_BroadcastAttestation(t *testing.T) {
 	msg := util.HydrateAttestation(&ethpb.Attestation{AggregationBits: bitfield.NewBitlist(7)})
 	subnet := uint64(5)
 
-	topic := AttestationSubnetTopicFormat
-	GossipTypeMapping[reflect.TypeFor[*ethpb.Attestation]()] = topic
+	GossipTypeMapping[reflect.TypeFor[*ethpb.Attestation]()] = AttestationSubnetTopicFormat
 	digest, err := p.currentForkDigest()
 	require.NoError(t, err)
-	topic = fmt.Sprintf(topic, digest, subnet)
+	topic := AttestationSubnetTopic(digest, subnet)
 
 	// External peer subscribes to the topic.
-	topic += p.Encoding().ProtocolSuffix()
 	sub, err := p2.SubscribeToTopic(topic)
 	require.NoError(t, err)
 
@@ -226,6 +228,7 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 	// Setup bootnode.
 	cfg := &Config{PingInterval: testPingInterval, DB: db}
 	cfg.UDPPort = uint(port)
+	cfg.TCPPort = uint(port)
 	_, pkey := createAddrAndPrivKey(t)
 	ipAddr := net.ParseIP("127.0.0.1")
 	genesisTime := time.Now()
@@ -251,8 +254,9 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 
 	var listeners []*listenerWrapper
 	var hosts []host.Host
+	var configs []*Config
 	// setup other nodes.
-	cfg = &Config{
+	baseCfg := &Config{
 		Discv5BootStrapAddrs: []string{bootNode.String()},
 		MaxPeers:             2,
 		PingInterval:         testPingInterval,
@@ -261,11 +265,21 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 	// Setup 2 different hosts
 	for i := uint(1); i <= 2; i++ {
 		h, pkey, ipAddr := createHost(t, port+i)
-		cfg.UDPPort = uint(port + i)
-		cfg.TCPPort = uint(port + i)
+
+		// Create a new config for each service to avoid shared mutations
+		cfg := &Config{
+			Discv5BootStrapAddrs: baseCfg.Discv5BootStrapAddrs,
+			MaxPeers:             baseCfg.MaxPeers,
+			PingInterval:         baseCfg.PingInterval,
+			DB:                   baseCfg.DB,
+			UDPPort:              uint(port + i),
+			TCPPort:              uint(port + i),
+		}
+
 		if len(listeners) > 0 {
 			cfg.Discv5BootStrapAddrs = append(cfg.Discv5BootStrapAddrs, listeners[len(listeners)-1].Self().String())
 		}
+
 		s := &Service{
 			cfg:                   cfg,
 			genesisTime:           genesisTime,
@@ -278,18 +292,22 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 		close(s.custodyInfoSet)
 
 		listener, err := s.startDiscoveryV5(ipAddr, pkey)
-		// Set for 2nd peer
+		assert.NoError(t, err, "Could not start discovery for node")
+
+		// Set listener for the service
+		s.dv5Listener = listener
+		s.metaData = wrapper.WrappedMetadataV0(new(ethpb.MetaDataV0))
+
+		// Set subnet for 2nd peer
 		if i == 2 {
-			s.dv5Listener = listener
-			s.metaData = wrapper.WrappedMetadataV0(new(ethpb.MetaDataV0))
 			bitV := bitfield.NewBitvector64()
 			bitV.SetBitAt(subnet, true)
 			err := s.updateSubnetRecordWithMetadata(bitV)
 			require.NoError(t, err)
 		}
-		assert.NoError(t, err, "Could not start discovery for node")
 		listeners = append(listeners, listener)
 		hosts = append(hosts, h)
+		configs = append(configs, cfg)
 	}
 	defer func() {
 		// Close down all peers.
@@ -324,7 +342,7 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 		pubsub:                ps1,
 		dv5Listener:           listeners[0],
 		joinedTopics:          map[string]*pubsub.Topic{},
-		cfg:                   cfg,
+		cfg:                   configs[0],
 		genesisTime:           time.Now(),
 		genesisValidatorsRoot: bytesutil.PadTo([]byte{'A'}, 32),
 		subnetsLock:           make(map[uint64]*sync.RWMutex),
@@ -340,7 +358,7 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 		pubsub:                ps2,
 		dv5Listener:           listeners[1],
 		joinedTopics:          map[string]*pubsub.Topic{},
-		cfg:                   cfg,
+		cfg:                   configs[1],
 		genesisTime:           time.Now(),
 		genesisValidatorsRoot: bytesutil.PadTo([]byte{'A'}, 32),
 		subnetsLock:           make(map[uint64]*sync.RWMutex),
@@ -353,14 +371,12 @@ func TestService_BroadcastAttestationWithDiscoveryAttempts(t *testing.T) {
 	go p2.listenForNewNodes()
 
 	msg := util.HydrateAttestation(&ethpb.Attestation{AggregationBits: bitfield.NewBitlist(7)})
-	topic := AttestationSubnetTopicFormat
-	GossipTypeMapping[reflect.TypeFor[*ethpb.Attestation]()] = topic
+	GossipTypeMapping[reflect.TypeFor[*ethpb.Attestation]()] = AttestationSubnetTopicFormat
 	digest, err := p.currentForkDigest()
 	require.NoError(t, err)
-	topic = fmt.Sprintf(topic, digest, subnet)
+	topic := AttestationSubnetTopic(digest, subnet)
 
 	// External peer subscribes to the topic.
-	topic += p.Encoding().ProtocolSuffix()
 	// We don't use our internal subscribe method
 	// due to using floodsub over here.
 	tpHandle, err := p2.JoinTopic(topic)
@@ -431,14 +447,12 @@ func TestService_BroadcastSyncCommittee(t *testing.T) {
 	msg := util.HydrateSyncCommittee(&ethpb.SyncCommitteeMessage{})
 	subnet := uint64(5)
 
-	topic := SyncCommitteeSubnetTopicFormat
-	GossipTypeMapping[reflect.TypeFor[*ethpb.SyncCommitteeMessage]()] = topic
+	GossipTypeMapping[reflect.TypeFor[*ethpb.SyncCommitteeMessage]()] = SyncCommitteeSubnetTopicFormat
 	digest, err := p.currentForkDigest()
 	require.NoError(t, err)
-	topic = fmt.Sprintf(topic, digest, subnet)
+	topic := SyncCommitteeSubnetTopic(digest, subnet)
 
 	// External peer subscribes to the topic.
-	topic += p.Encoding().ProtocolSuffix()
 	sub, err := p2.SubscribeToTopic(topic)
 	require.NoError(t, err)
 
@@ -508,14 +522,12 @@ func TestService_BroadcastBlob(t *testing.T) {
 	}
 	subnet := uint64(0)
 
-	topic := BlobSubnetTopicFormat
-	GossipTypeMapping[reflect.TypeFor[*ethpb.BlobSidecar]()] = topic
+	GossipTypeMapping[reflect.TypeFor[*ethpb.BlobSidecar]()] = BlobSubnetTopicFormat
 	digest, err := p.currentForkDigest()
 	require.NoError(t, err)
-	topic = fmt.Sprintf(topic, digest, subnet)
+	topic := BlobSubnetTopic(digest, subnet)
 
 	// External peer subscribes to the topic.
-	topic += p.Encoding().ProtocolSuffix()
 	sub, err := p2.SubscribeToTopic(topic)
 	require.NoError(t, err)
 
@@ -575,10 +587,9 @@ func TestService_BroadcastLightClientOptimisticUpdate(t *testing.T) {
 	require.NoError(t, err)
 
 	GossipTypeMapping[reflect.TypeOf(msg)] = LightClientOptimisticUpdateTopicFormat
-	topic := fmt.Sprintf(LightClientOptimisticUpdateTopicFormat, params.ForkDigest(slots.ToEpoch(msg.AttestedHeader().Beacon().Slot)))
+	topic := LcOptimisticToTopic(params.ForkDigest(slots.ToEpoch(msg.AttestedHeader().Beacon().Slot)))
 
 	// External peer subscribes to the topic.
-	topic += p.Encoding().ProtocolSuffix()
 	sub, err := p2.SubscribeToTopic(topic)
 	require.NoError(t, err)
 
@@ -651,10 +662,9 @@ func TestService_BroadcastLightClientFinalityUpdate(t *testing.T) {
 	require.NoError(t, err)
 
 	GossipTypeMapping[reflect.TypeOf(msg)] = LightClientFinalityUpdateTopicFormat
-	topic := fmt.Sprintf(LightClientFinalityUpdateTopicFormat, params.ForkDigest(slots.ToEpoch(msg.AttestedHeader().Beacon().Slot)))
+	topic := LcFinalityToTopic(params.ForkDigest(slots.ToEpoch(msg.AttestedHeader().Beacon().Slot)))
 
 	// External peer subscribes to the topic.
-	topic += p.Encoding().ProtocolSuffix()
 	sub, err := p2.SubscribeToTopic(topic)
 	require.NoError(t, err)
 
@@ -702,7 +712,6 @@ func TestService_BroadcastDataColumn(t *testing.T) {
 	const (
 		port        = 2000
 		columnIndex = 12
-		topicFormat = DataColumnSubnetTopicFormat
 	)
 
 	ctx := t.Context()
@@ -760,7 +769,17 @@ func TestService_BroadcastDataColumn(t *testing.T) {
 	require.NoError(t, err)
 
 	subnet := peerdas.ComputeSubnetForDataColumnSidecar(columnIndex)
-	topic := fmt.Sprintf(topicFormat, digest, subnet) + service.Encoding().ProtocolSuffix()
+	topic := DataColumnSubnetTopic(digest, subnet)
+
+	crawler, err := NewGossipPeerCrawler(t.Context(), service, listener, 1*time.Second, 1*time.Second, 10,
+		func(n *enode.Node) bool { return true },
+		service.Peers().Scorers().Score)
+	require.NoError(t, err)
+	err = crawler.Start(func(ctx context.Context, node *enode.Node) ([]string, error) {
+		return []string{topic}, nil
+	})
+	require.NoError(t, err)
+	service.gossipDialer = NewGossipPeerDialer(t.Context(), crawler, service.PubSub().ListPeers, service.DialPeers)
 
 	_, verifiedRoSidecars := util.CreateTestVerifiedRoDataColumnSidecars(t, []util.DataColumnParam{{Index: columnIndex}})
 	verifiedRoSidecar := verifiedRoSidecars[0]
@@ -787,3 +806,190 @@ func TestService_BroadcastDataColumn(t *testing.T) {
 	require.NoError(t, service.Encoding().DecodeGossip(msg.Data, &result))
 	require.DeepEqual(t, &result, verifiedRoSidecar)
 }
+
+type topicInvoked struct {
+	topic string
+	pid   peer.ID
+}
+
+// rpcOrderTracer is a RawTracer implementation that captures the order of SendRPC calls.
+// It records the topics of messages sent via pubsub to verify round-robin ordering.
+type rpcOrderTracer struct {
+	mu      sync.Mutex
+	invoked []*topicInvoked
+	byTopic map[string][]peer.ID
+}
+
+func (t *rpcOrderTracer) SendRPC(rpc *pubsub.RPC, pid peer.ID) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	for _, msg := range rpc.GetPublish() {
+		invoked := &topicInvoked{topic: msg.GetTopic(), pid: pid}
+		t.invoked = append(t.invoked, invoked)
+		t.byTopic[invoked.topic] = append(t.byTopic[invoked.topic], invoked.pid)
+	}
+}
+
+func newRpcOrderTracer() *rpcOrderTracer {
+	return &rpcOrderTracer{byTopic: make(map[string][]peer.ID)}
+}
+
+func (t *rpcOrderTracer) getTopics() []string {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	result := make([]string, len(t.invoked))
+	for i := range t.invoked {
+		result[i] = t.invoked[i].topic
+	}
+	return result
+}
+
+// No-op implementations for other RawTracer methods.
+func (*rpcOrderTracer) AddPeer(peer.ID, protocol.ID)          {}
+func (*rpcOrderTracer) RemovePeer(peer.ID)                    {}
+func (*rpcOrderTracer) Join(string)                           {}
+func (*rpcOrderTracer) Leave(string)                          {}
+func (*rpcOrderTracer) Graft(peer.ID, string)                 {}
+func (*rpcOrderTracer) Prune(peer.ID, string)                 {}
+func (*rpcOrderTracer) ValidateMessage(*pubsub.Message)       {}
+func (*rpcOrderTracer) DeliverMessage(*pubsub.Message)        {}
+func (*rpcOrderTracer) RejectMessage(*pubsub.Message, string) {}
+func (*rpcOrderTracer) DuplicateMessage(*pubsub.Message)      {}
+func (*rpcOrderTracer) ThrottlePeer(peer.ID)                  {}
+func (*rpcOrderTracer) RecvRPC(*pubsub.RPC)                   {}
+func (*rpcOrderTracer) DropRPC(*pubsub.RPC, peer.ID)          {}
+func (*rpcOrderTracer) UndeliverableMessage(*pubsub.Message)  {}
+
+// TestService_BroadcastDataColumnRoundRobin verifies that when broadcasting multiple
+// data column sidecars, messages are interleaved in round-robin order by column index
+// rather than sending all copies of one column before the next.
+//
+// Without batch publishing: A,A,A,A,B,B,B,B (all peers for column A, then all for column B)
+// With batch publishing:    A,B,A,B,A,B,A,B (interleaved by message ID)
+func TestService_BroadcastDataColumnRoundRobin(t *testing.T) {
+	const (
+		port        = 2100
+		topicFormat = DataColumnSubnetTopicFormat
+	)
+
+	ctx := t.Context()
+
+	// Load the KZG trust setup.
+	err := kzg.Start()
+	require.NoError(t, err)
+
+	gFlags := new(flags.GlobalFlags)
+	gFlags.MinimumPeersPerSubnet = 1
+	flags.Init(gFlags)
+	defer flags.Init(new(flags.GlobalFlags))
+
+	// Create a tracer to capture the order of SendRPC calls.
+	tracer := newRpcOrderTracer()
+
+	// Create the publisher node with the tracer injected.
+	p1 := p2ptest.NewTestP2PWithPubsubOptions(t, []pubsub.Option{pubsub.WithRawTracer(tracer)})
+
+	// Create subscriber peers.
+	expectedPeers := []*p2ptest.TestP2P{
+		p2ptest.NewTestP2P(t),
+		p2ptest.NewTestP2P(t),
+	}
+
+	// Connect peers.
+	for _, p := range expectedPeers {
+		p1.Connect(p)
+	}
+	require.NotEqual(t, 0, len(p1.BHost.Network().Peers()), "No peers")
+
+	// Create a host for discovery.
+	_, pkey, ipAddr := createHost(t, port)
+
+	// Create a shared DB for the service.
+	db := testDB.SetupDB(t)
+
+	// Create and close the custody info channel immediately since custodyInfo is already set.
+	custodyInfoSet := make(chan struct{})
+	close(custodyInfoSet)
+
+	service := &Service{
+		ctx:                   ctx,
+		host:                  p1.BHost,
+		pubsub:                p1.PubSub(),
+		joinedTopics:          map[string]*pubsub.Topic{},
+		cfg:                   &Config{DB: db},
+		genesisTime:           time.Now(),
+		genesisValidatorsRoot: bytesutil.PadTo([]byte{'A'}, 32),
+		subnetsLock:           make(map[uint64]*sync.RWMutex),
+		subnetsLockLock:       sync.Mutex{},
+		peers:                 peers.NewStatus(ctx, &peers.StatusConfig{ScorerParams: &scorers.Config{}}),
+		custodyInfo:           &custodyInfo{},
+		custodyInfoSet:        custodyInfoSet,
+	}
+
+	// Create a listener for discovery.
+	listener, err := service.startDiscoveryV5(ipAddr, pkey)
+	require.NoError(t, err)
+	service.dv5Listener = listener
+
+	digest, err := service.currentForkDigest()
+	require.NoError(t, err)
+
+	// Create multiple data column sidecars with different column indices.
+	// Use indices that map to different subnets: 0, 32, 64 (assuming 128 columns and 64 subnets).
+	columnIndices := []uint64{0, 32, 64}
+	params := make([]util.DataColumnParam, len(columnIndices))
+	for i, idx := range columnIndices {
+		params[i] = util.DataColumnParam{Index: idx}
+	}
+	_, verifiedRoSidecars := util.CreateTestVerifiedRoDataColumnSidecars(t, params)
+
+	expectedTopics := make(map[string]bool)
+	// Subscribe peers to the relevant topics.
+	for _, idx := range columnIndices {
+		subnet := peerdas.ComputeSubnetForDataColumnSidecar(idx)
+		topic := fmt.Sprintf(topicFormat, digest, subnet) + service.Encoding().ProtocolSuffix()
+		for _, p := range expectedPeers {
+			_, err = p.SubscribeToTopic(topic)
+			require.NoError(t, err)
+		}
+		expectedTopics[topic] = true
+	}
+	// libp2p needs some time to establish mesh connections.
+	time.Sleep(100 * time.Millisecond)
+
+	// Broadcast all sidecars.
+	err = service.BroadcastDataColumnSidecars(ctx, verifiedRoSidecars)
+	require.NoError(t, err)
+	// Give some time for messages to be sent.
+	time.Sleep(100 * time.Millisecond)
+
+	topics := tracer.getTopics()
+	if len(topics) == 0 {
+		t.Fatal("Expected at least one message for each topic to be sent to each peer")
+	}
+
+	unseen := make(map[string]bool)
+	for k := range expectedTopics {
+		unseen[k] = true
+	}
+	// Verify round-robin invariant: before all message IDs are seen, no message ID may be repeated.
+	// In round-robin order, we should see each topic once before any topic repeats.
+	for _, topic := range topics {
+		if !expectedTopics[topic] {
+			continue
+		}
+		if !unseen[topic] {
+			t.Errorf("Topic %s repeated before all topics were seen once. This violates round-robin ordering.", topic)
+		}
+		delete(unseen, topic)
+		if len(unseen) == 0 {
+			break // all have been seen
+		}
+	}
+	require.Equal(t, 0, len(unseen))
+
+	// Verify that we actually saw all expected topics.
+	for topic := range expectedTopics {
+		require.Equal(t, len(expectedPeers), len(tracer.byTopic[topic]))
+	}
+}