Merge branch 'feat/peerdas' into feat/hermes-custom-networks

deploy/local/docker-compose/vector-http-kafka.yaml

@@ -65,6 +65,7 @@ transforms:
  libp2p_trace_handle_status: .event.name == "LIBP2P_TRACE_HANDLE_STATUS"
  libp2p_trace_gossipsub_beacon_block : .event.name == "LIBP2P_TRACE_GOSSIPSUB_BEACON_BLOCK"
  libp2p_trace_gossipsub_beacon_attestation : .event.name == "LIBP2P_TRACE_GOSSIPSUB_BEACON_ATTESTATION"
+ libp2p_trace_gossipsub_beacon_data_column_sidecar : .event.name == "LIBP2P_TRACE_GOSSIPSUB_BEACON_DATA_COLUMN_SIDECAR"
 sinks:
  metrics:
  type: prometheus_exporter
@@ -856,3 +857,19 @@ sinks:
  enabled: true
  encoding:
  codec: json
+ libp2p_trace_gossipsub_beacon_data_column_sidecar_kafka:
+ type: kafka
+ buffer:
+ max_events: 500000
+ batch:
+ timeout_secs: 0.5
+ inputs:
+ - xatu_server_events_router.libp2p_trace_gossipsub_beacon_data_column_sidecar
+ bootstrap_servers: "${KAFKA_BROKERS}"
+ key_field: "event.id"
+ topic: libp2p-trace-gossipsub-beacon-data-column-sidecar
+ compression: snappy
+ healthcheck:
+ enabled: true
+ encoding:
+ codec: json

deploy/local/docker-compose/vector-kafka-clickhouse-libp2p.yaml

@@ -215,6 +215,7 @@ transforms:
  libp2p_trace_handle_status: .event.name == "LIBP2P_TRACE_HANDLE_STATUS"
  libp2p_trace_gossipsub_beacon_block: .event.name == "LIBP2P_TRACE_GOSSIPSUB_BEACON_BLOCK"
  libp2p_trace_gossipsub_beacon_attestation: .event.name == "LIBP2P_TRACE_GOSSIPSUB_BEACON_ATTESTATION"
+ libp2p_trace_gossipsub_beacon_data_column_sidecar: .event.name == "LIBP2P_TRACE_GOSSIPSUB_BEACON_DATA_COLUMN_SIDECAR"
  xatu_server_events_router_matched:
  type: log_to_metric
  inputs:
@@ -229,6 +230,7 @@ transforms:
  - xatu_server_events_router.libp2p_trace_handle_status
  - xatu_server_events_router.libp2p_trace_gossipsub_beacon_block
  - xatu_server_events_router.libp2p_trace_gossipsub_beacon_attestation
+ - xatu_server_events_router.libp2p_trace_gossipsub_beacon_data_column_sidecar
  metrics:
  - type: counter
  field: event.name
@@ -772,6 +774,89 @@ transforms:
  
  .updated_date_time = to_unix_timestamp(now())
 
+ del(.event)
+ del(.meta)
+ del(.data)
+ del(.path)
+ libp2p_trace_gossipsub_beacon_data_column_sidecar_formatted:
+ type: remap
+ inputs:
+ - xatu_server_events_router.libp2p_trace_gossipsub_beacon_data_column_sidecar
+ source: |-
+ event_date_time, err = parse_timestamp(.event.date_time, format: "%+");
+ if err == null {
+ .event_date_time = to_unix_timestamp(event_date_time, unit: "milliseconds")
+ } else {
+ .error = err
+ .error_description = "failed to parse event date time"
+ log(., level: "error", rate_limit_secs: 60)
+ }
+
+ peer_id_key, err = .meta.client.additional_data.metadata.peer_id + .meta.ethereum.network.name
+ if err != null {
+ .error = err
+ .error_description = "failed to generate peer id unique key"
+ log(., level: "error", rate_limit_secs: 60)
+ }
+ .peer_id_unique_key = seahash(peer_id_key)
+ .unique_key = seahash(.event.id)
+
+ .propagation_slot_start_diff = .meta.client.additional_data.propagation.slot_start_diff
+
+ .state_root = .data.signed_block_header.message.state_root
+ .column_index = .data.column_index
+ .kzg_commitments = .data.kzg_commitments
+ .kzg_proof = .data.kzg_proof
+ .kzg_commitments_inclusion_proof = .data.kzg_commitments_inclusion_proof
+
+ .slot = .data.data.slot
+ slot_start_date_time, err = parse_timestamp(.meta.client.additional_data.slot.start_date_time, format: "%+");
+ if err == null {
+ .slot_start_date_time = to_unix_timestamp(slot_start_date_time)
+ } else {
+ .error = err
+ .error_description = "failed to parse slot start date time"
+ log(., level: "error", rate_limit_secs: 60)
+ }
+ .epoch = .meta.client.additional_data.epoch.number
+ epoch_start_date_time, err = parse_timestamp(.meta.client.additional_data.epoch.start_date_time, format: "%+");
+ if err == null {
+ .epoch_start_date_time = to_unix_timestamp(epoch_start_date_time)
+ } else {
+ .error = err
+ .error_description = "failed to parse epoch start date time"
+ log(., level: "error", rate_limit_secs: 60)
+ }
+
+ .wallclock_slot = .meta.client.additional_data.wallclock_slot.number
+ wallclock_slot_start_date_time, err = parse_timestamp(.meta.client.additional_data.wallclock_slot.start_date_time, format: "%+");
+ if err == null {
+ .wallclock_slot_start_date_time = to_unix_timestamp(wallclock_slot_start_date_time)
+ } else {
+ .error = err
+ .error_description = "failed to parse wallclock slot start date time"
+ log(., level: "error", rate_limit_secs: 60)
+ }
+ .wallclock_epoch = .meta.client.additional_data.epoch.number
+ wallclock_epoch_start_date_time, err = parse_timestamp(.meta.client.additional_data.wallclock_epoch.start_date_time, format: "%+");
+ if err == null {
+ .wallclock_epoch_start_date_time = to_unix_timestamp(wallclock_epoch_start_date_time)
+ } else {
+ .error = err
+ .error_description = "failed to parse wallclock epoch start date time"
+ log(., level: "error", rate_limit_secs: 60)
+ }
+
+ .topic_layer = topicParts[1]
+ .topic_fork_digest_value = topicParts[2]
+ .topic_name = topicParts[3]
+ .topic_encoding = topicParts[4]
+
+ .message_size = .meta.client.additional_data.message_size
+ .message_id = .meta.client.additional_data.message_id
+ 
+ .updated_date_time = to_unix_timestamp(now())
+
  del(.event)
  del(.meta)
  del(.data)

deploy/migrations/clickhouse/037_libp2p_gossipsub_beacon_data_column_sidecar.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS libp2p_gossipsub_beacon_data_column_sidecar ON CLUSTER '{cluster}';
+DROP TABLE IF EXISTS libp2p_gossipsub_beacon_data_column_sidecar_local ON CLUSTER '{cluster}';

deploy/migrations/clickhouse/037_libp2p_gossipsub_beacon_data_column_sidecar.up.sql

@@ -0,0 +1,90 @@
+CREATE TABLE libp2p_gossipsub_beacon_data_column_sidecar_local on cluster '{cluster}' (
+ unique_key Int64,
+ updated_date_time DateTime CODEC(DoubleDelta, ZSTD(1)),
+ event_date_time DateTime64(3) Codec(DoubleDelta, ZSTD(1)),
+ slot UInt32 Codec(DoubleDelta, ZSTD(1)),
+ slot_start_date_time DateTime Codec(DoubleDelta, ZSTD(1)),
+ epoch UInt32 Codec(DoubleDelta, ZSTD(1)),
+ epoch_start_date_time DateTime Codec(DoubleDelta, ZSTD(1)),
+ column_index UInt64 Codec(ZSTD(1)),
+ kzg_commitments Array(String) Codec(ZSTD(1)),
+ kzg_proof String Codec(ZSTD(1)),
+ kzg_commitments_inclusion_proof String Codec(ZSTD(1)),
+ state_root FixedString(66) Codec(ZSTD(1)),
+ wallclock_slot UInt32 Codec(DoubleDelta, ZSTD(1)),
+ wallclock_slot_start_date_time DateTime Codec(DoubleDelta, ZSTD(1)),
+ wallclock_epoch UInt32 Codec(DoubleDelta, ZSTD(1)),
+ wallclock_epoch_start_date_time DateTime Codec(DoubleDelta, ZSTD(1)),
+ propagation_slot_start_diff UInt32 Codec(ZSTD(1)),
+ peer_id_unique_key Int64,
+ message_id String CODEC(ZSTD(1)),
+ message_size UInt32 Codec(ZSTD(1)),
+ topic_layer LowCardinality(String),
+ topic_fork_digest_value LowCardinality(String),
+ topic_name LowCardinality(String),
+ topic_encoding LowCardinality(String),
+ meta_client_name LowCardinality(String),
+ meta_client_id String Codec(ZSTD(1)),
+ meta_client_version LowCardinality(String),
+ meta_client_implementation LowCardinality(String),
+ meta_client_os LowCardinality(String),
+ meta_client_ip Nullable(IPv6) Codec(ZSTD(1)),
+ meta_client_geo_city LowCardinality(String) Codec(ZSTD(1)),
+ meta_client_geo_country LowCardinality(String) Codec(ZSTD(1)),
+ meta_client_geo_country_code LowCardinality(String) Codec(ZSTD(1)),
+ meta_client_geo_continent_code LowCardinality(String) Codec(ZSTD(1)),
+ meta_client_geo_longitude Nullable(Float64) Codec(ZSTD(1)),
+ meta_client_geo_latitude Nullable(Float64) Codec(ZSTD(1)),
+ meta_client_geo_autonomous_system_number Nullable(UInt32) Codec(ZSTD(1)),
+ meta_client_geo_autonomous_system_organization Nullable(String) Codec(ZSTD(1)),
+ meta_network_id Int32 Codec(DoubleDelta, ZSTD(1)),
+ meta_network_name LowCardinality(String)
+) Engine = ReplicatedReplacingMergeTree('/clickhouse/{installation}/{cluster}/tables/{shard}/{database}/{table}', '{replica}', updated_date_time)
+PARTITION BY toStartOfMonth(slot_start_date_time)
+ORDER BY (slot_start_date_time, unique_key, meta_network_name, meta_client_name);
+
+ALTER TABLE libp2p_gossipsub_beacon_data_column_sidecar_local ON CLUSTER '{cluster}'
+MODIFY COMMENT 'Table for libp2p gossipsub beacon data column sidecar data.',
+COMMENT COLUMN unique_key 'Unique identifier for each record',
+COMMENT COLUMN updated_date_time 'Timestamp when the record was last updated',
+COMMENT COLUMN event_date_time 'Timestamp of the event with millisecond precision',
+COMMENT COLUMN slot 'Slot number associated with the event',
+COMMENT COLUMN slot_start_date_time 'Start date and time of the slot',
+COMMENT COLUMN epoch 'The epoch number in the data column sidecar',
+COMMENT COLUMN epoch_start_date_time 'The wall clock time when the epoch started',
+COMMENT COLUMN column_index 'Index of the column in the data',
+COMMENT COLUMN kzg_commitments 'Commitments for the column',
+COMMENT COLUMN kzg_proof 'Proof for the column',
+COMMENT COLUMN kzg_commitments_inclusion_proof 'Inclusion proof for the column',
+COMMENT COLUMN state_root 'State root for signed block header',
+COMMENT COLUMN wallclock_slot 'Slot number of the wall clock when the event was received',
+COMMENT COLUMN wallclock_slot_start_date_time 'Start date and time of the wall clock slot when the event was received',
+COMMENT COLUMN wallclock_epoch 'Epoch number of the wall clock when the event was received',
+COMMENT COLUMN wallclock_epoch_start_date_time 'Start date and time of the wall clock epoch when the event was received',
+COMMENT COLUMN propagation_slot_start_diff 'Difference in slot start time for propagation',
+COMMENT COLUMN peer_id_unique_key 'Unique key associated with the identifier of the peer',
+COMMENT COLUMN message_id 'Identifier of the message',
+COMMENT COLUMN message_size 'Size of the message in bytes',
+COMMENT COLUMN topic_layer 'Layer of the topic in the gossipsub protocol',
+COMMENT COLUMN topic_fork_digest_value 'Fork digest value of the topic',
+COMMENT COLUMN topic_name 'Name of the topic',
+COMMENT COLUMN topic_encoding 'Encoding used for the topic',
+COMMENT COLUMN meta_client_name 'Name of the client that generated the event',
+COMMENT COLUMN meta_client_id 'Unique Session ID of the client that generated the event. This changes every time the client is restarted.',
+COMMENT COLUMN meta_client_version 'Version of the client that generated the event',
+COMMENT COLUMN meta_client_implementation 'Implementation of the client that generated the event',
+COMMENT COLUMN meta_client_os 'Operating system of the client that generated the event',
+COMMENT COLUMN meta_client_ip 'IP address of the client that generated the event',
+COMMENT COLUMN meta_client_geo_city 'City of the client that generated the event',
+COMMENT COLUMN meta_client_geo_country 'Country of the client that generated the event',
+COMMENT COLUMN meta_client_geo_country_code 'Country code of the client that generated the event',
+COMMENT COLUMN meta_client_geo_continent_code 'Continent code of the client that generated the event',
+COMMENT COLUMN meta_client_geo_longitude 'Longitude of the client that generated the event',
+COMMENT COLUMN meta_client_geo_latitude 'Latitude of the client that generated the event',
+COMMENT COLUMN meta_client_geo_autonomous_system_number 'Autonomous system number of the client that generated the event',
+COMMENT COLUMN meta_client_geo_autonomous_system_organization 'Autonomous system organization of the client that generated the event',
+COMMENT COLUMN meta_network_id 'Network ID associated with the client',
+COMMENT COLUMN meta_network_name 'Name of the network associated with the client';
+
+CREATE TABLE libp2p_gossipsub_beacon_data_column_sidecar on cluster '{cluster}' AS libp2p_gossipsub_beacon_data_column_sidecar_local
+ENGINE = Distributed('{cluster}', default, libp2p_gossipsub_beacon_data_column_sidecar_local, unique_key);

go.mod

@@ -2,7 +2,7 @@ module github.com/ethpandaops/xatu
 
 go 1.22.0
 
-replace github.com/probe-lab/hermes => github.com/ethpandaops/hermes v0.0.0-20240501013510-f8170ec5da4d
+replace github.com/probe-lab/hermes => github.com/ethpandaops/hermes v0.0.0-20240508034827-0c42cafd65a5
 
 require (
  github.com/IBM/sarama v1.43.0
@@ -31,7 +31,7 @@ require (
  github.com/pkg/errors v0.9.1
  github.com/probe-lab/hermes v0.0.0-20240327153144-a2528356b4f7
  github.com/prometheus/client_golang v1.19.0
- github.com/prysmaticlabs/prysm/v5 v5.0.2
+ github.com/prysmaticlabs/prysm/v5 v5.0.4-0.20240507051732-39060d20cdd9
  github.com/r3labs/sse/v2 v2.10.0
  github.com/redis/go-redis/v9 v9.5.1
  github.com/sirupsen/logrus v1.9.3
@@ -94,7 +94,7 @@ require (
  github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
  github.com/eapache/queue v1.1.0 // indirect
  github.com/elastic/gosigar v0.14.2 // indirect
- github.com/ethereum/c-kzg-4844 v0.4.0 // indirect
+ github.com/ethereum/c-kzg-4844 v1.0.1-0.20240422190800-13be436f5927 // indirect
  github.com/fatih/color v1.16.0 // indirect
  github.com/ferranbt/fastssz v0.1.3 // indirect
  github.com/flynn/noise v1.1.0 // indirect

go.sum

@@ -256,8 +256,8 @@ github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1m
 github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5ynNVH9qI8YYLbd1fK2po=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
-github.com/ethereum/c-kzg-4844 v0.4.0 h1:3MS1s4JtA868KpJxroZoepdV0ZKBp3u/O5HcZ7R3nlY=
-github.com/ethereum/c-kzg-4844 v0.4.0/go.mod h1:VewdlzQmpT5QSrVhbBuGoCdFJkpaJlO1aQputP83wc0=
+github.com/ethereum/c-kzg-4844 v1.0.1-0.20240422190800-13be436f5927 h1:ffWmm0RUR2+VqJsCkf94HqgEwZi2fgbm2iq+O/GdJNI=
+github.com/ethereum/c-kzg-4844 v1.0.1-0.20240422190800-13be436f5927/go.mod h1:VewdlzQmpT5QSrVhbBuGoCdFJkpaJlO1aQputP83wc0=
 github.com/ethereum/go-ethereum v1.13.14 h1:EwiY3FZP94derMCIam1iW4HFVrSgIcpsu0HwTQtm6CQ=
 github.com/ethereum/go-ethereum v1.13.14/go.mod h1:TN8ZiHrdJwSe8Cb6x+p0hs5CxhJZPbqB7hHkaUXcmIU=
 github.com/ethpandaops/beacon v0.36.0 h1:BIsb0ftrrOSwsORCNPvq7df3O5f6sVcFm7guGLpFztc=
@@ -266,8 +266,8 @@ github.com/ethpandaops/ethcore v0.0.0-20240422023000-2a5727b18756 h1:8JWjrRfP14m
 github.com/ethpandaops/ethcore v0.0.0-20240422023000-2a5727b18756/go.mod h1:ZvKqL6CKxiraefdXPHeJurV2pDD/f2HF2uklDVdrry8=
 github.com/ethpandaops/ethwallclock v0.3.0 h1:xF5fwtBf+bHFHZKBnwiPFEuelW3sMM7SD3ZNFq1lJY4=
 github.com/ethpandaops/ethwallclock v0.3.0/go.mod h1:y0Cu+mhGLlem19vnAV2x0hpFS5KZ7oOi2SWYayv9l24=
-github.com/ethpandaops/hermes v0.0.0-20240501013510-f8170ec5da4d h1:mWBuU7mhpN11c2bbcImFyLSl+TJ208N4aGhMhadflIw=
-github.com/ethpandaops/hermes v0.0.0-20240501013510-f8170ec5da4d/go.mod h1:WOtvdP1zl53vuoOX6PRZ9oAbbJUiDso50efmAjoUZzI=
+github.com/ethpandaops/hermes v0.0.0-20240508034827-0c42cafd65a5 h1:kbXv4MfY+3xJ5ldL5bxf6AVU0CqmmYFfYSSHssoOUYE=
+github.com/ethpandaops/hermes v0.0.0-20240508034827-0c42cafd65a5/go.mod h1:Z65YW+OhX67VHToofu2+wZDdMUgQ1AnvaEPgB9GzTC0=
 github.com/evanphx/json-patch v4.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
@@ -878,8 +878,8 @@ github.com/prysmaticlabs/prombbolt v0.0.0-20210126082820-9b7adba6db7c h1:9PHRCuO
 github.com/prysmaticlabs/prombbolt v0.0.0-20210126082820-9b7adba6db7c/go.mod h1:ZRws458tYHS/Zs936OQ6oCrL+Ict5O4Xpwve1UQ6C9M=
 github.com/prysmaticlabs/protoc-gen-go-cast v0.0.0-20230228205207-28762a7b9294 h1:q9wE0ZZRdTUAAeyFP/w0SwBEnCqlVy2+on6X2/e+eAU=
 github.com/prysmaticlabs/protoc-gen-go-cast v0.0.0-20230228205207-28762a7b9294/go.mod h1:ZVEbRdnMkGhp/pu35zq4SXxtvUwWK0J1MATtekZpH2Y=
-github.com/prysmaticlabs/prysm/v5 v5.0.2 h1:xcSUvrCVfOGslKYUb5Hpyz98N9I8fC2p7DMAZfiqEIA=
-github.com/prysmaticlabs/prysm/v5 v5.0.2/go.mod h1:XG4nOU925zemOimoexcrFP4oA57f+RTQbp7V/TH9UOM=
+github.com/prysmaticlabs/prysm/v5 v5.0.4-0.20240507051732-39060d20cdd9 h1:z6b+3uqU4t2lJAAYIB75Yp8sBaCeo4p1/+uHxZxXkmw=
+github.com/prysmaticlabs/prysm/v5 v5.0.4-0.20240507051732-39060d20cdd9/go.mod h1:5Pt2wu6NOUTSzLWcoEco/SraJ4kEEx5HubrHBTsmGmE=
 github.com/quic-go/qpack v0.4.0 h1:Cr9BXA1sQS2SmDUWjSofMPNKmvF6IiIfDRmgU0w1ZCo=
 github.com/quic-go/qpack v0.4.0/go.mod h1:UZVnYIfi5GRk+zI9UMaCPsmZ2xKJP7XBUvVyT1Knj9A=
 github.com/quic-go/quic-go v0.42.0 h1:uSfdap0eveIl8KXnipv9K7nlwZ5IqLlYOpJ58u5utpM=

pkg/clmimicry/config.go

@@ -131,17 +131,18 @@ func (h *NodeConfig) AsHermesConfig() *hermes.NodeConfig {
 }
 
 type EventConfig struct {
- RecvRPCEnabled bool `yaml:"recvRpcEnabled" default:"false"`
- SendRPCEnabled bool `yaml:"sendRpcEnabled" default:"false"`
- AddPeerEnabled bool `yaml:"addPeerEnabled" default:"true"`
- RemovePeerEnabled bool `yaml:"removePeerEnabled" default:"true"`
- ConnectedEnabled bool `yaml:"connectedEnabled" default:"true"`
- DisconnectedEnabled bool `yaml:"disconnectedEnabled" default:"true"`
- JoinEnabled bool `yaml:"joinEnabled" default:"true"`
- HandleMetadataEnabled bool `yaml:"handleMetadataEnabled" default:"true"`
- HandleStatusEnabled bool `yaml:"handleStatusEnabled" default:"true"`
- GossipSubBeaconBlockEnabled bool `yaml:"gossipSubBeaconBlockEnabled" default:"true"`
- GossipSubAttestationEnabled bool `yaml:"gossipSubAttestationEnabled" default:"true"`
+ RecvRPCEnabled bool `yaml:"recvRpcEnabled" default:"false"`
+ SendRPCEnabled bool `yaml:"sendRpcEnabled" default:"false"`
+ AddPeerEnabled bool `yaml:"addPeerEnabled" default:"true"`
+ RemovePeerEnabled bool `yaml:"removePeerEnabled" default:"true"`
+ ConnectedEnabled bool `yaml:"connectedEnabled" default:"true"`
+ DisconnectedEnabled bool `yaml:"disconnectedEnabled" default:"true"`
+ JoinEnabled bool `yaml:"joinEnabled" default:"true"`
+ HandleMetadataEnabled bool `yaml:"handleMetadataEnabled" default:"true"`
+ HandleStatusEnabled bool `yaml:"handleStatusEnabled" default:"true"`
+ GossipSubBeaconBlockEnabled bool `yaml:"gossipSubBeaconBlockEnabled" default:"true"`
+ GossipSubAttestationEnabled bool `yaml:"gossipSubAttestationEnabled" default:"true"`
+ GossipSubDataColumnSidecarEnabled bool `yaml:"gossipSubDataColumnSidecarEnabled" default:"true"`
 }
 
 func (e *EventConfig) Validate() error {

pkg/clmimicry/event.go

@@ -457,6 +457,18 @@ func (m *Mimicry) handleHandleMessageEvent(ctx context.Context,
  return nil
  }
 
+ if strings.Contains(topic, "data_column_sidecar") {
+ if !m.Config.Events.GossipSubDataColumnSidecarEnabled {
+ return nil
+ }
+
+ if err := m.handleGossipDataColumnSidecar(ctx, clientMeta, event, payload); err != nil {
+ return errors.Wrap(err, "failed to handle gossipsub data column sidecar")
+ }
+
+ return nil
+ }
+
  if strings.Contains(topic, "beacon_block") {
  if !m.Config.Events.GossipSubBeaconBlockEnabled {
  return nil