feat: add gossipsub message payload event for raw wire-bytes capture (#877)

* feat: add gossipsub message payload event for raw wire-bytes capture

Adds LIBP2P_TRACE_GOSSIPSUB_MESSAGE_PAYLOAD end-to-end: proto event and
payload messages, clmimicry handler (opt-in via
gossipSubMessagePayloadEnabled, off by default), event-ingester
registration, ClickHouse table libp2p_gossipsub_message_payload, and the
consumoor route.

The table's sorting key and distributed sharding key are content-derived
(message IDs are content hashes), so identical messages captured by
multiple vantage points deduplicate on merge. Payload bytes are stored
exactly as received off the wire, so capture works for any topic without
a decoder, including validation-rejected messages.

Claude-Session: https://claude.ai/code/session_01KDvdo7uVQpxKWgk2gLnWpE

* refactor: append validated payload route fields unconditionally

validate() already rejects events missing the wallclock fields or message
ID, so the sentinel-zero fallback branches were unreachable and read as if
missing data could silently insert epoch-zero timestamps.

Claude-Session: https://claude.ai/code/session_01KDvdo7uVQpxKWgk2gLnWpE

* refactor: drop observation-scoped columns from payload table

The table deduplicates by content-derived message ID across capture
clients, so per-observation columns were nondeterministic: whichever row
survived the merge won. peer_id_unique_key and reject_reason are dropped
(peer attribution and reject reasons live in libp2p_deliver_message and
libp2p_reject_message), and outcome moves into the sorting key so a
message delivered by one client and rejected by another keeps one row per
outcome instead of racing.

Claude-Session: https://claude.ai/code/session_01KDvdo7uVQpxKWgk2gLnWpE

* refactor: drop outcome column from payload table

The table stores message content keyed by content-derived message ID;
validation outcome is a fact about an observation, not the bytes, and
lives per-vantage in libp2p_deliver_message and libp2p_reject_message.
This also collapses fork-contested messages (delivered by some clients,
rejected by others) to a single row. The proto event keeps outcome and
reject_reason for stream consumers.

Claude-Session: https://claude.ai/code/session_01KDvdo7uVQpxKWgk2gLnWpE

* fix: accept empty message data in payload handler

A nil Data slice is a legitimate zero-byte message, and rejecting it
meant exactly that class of message was never archived. Also documents
that dedup is best-effort for messages received across wallclock slot or
partition boundaries.

Claude-Session: https://claude.ai/code/session_01KDvdo7uVQpxKWgk2gLnWpE

Author: samcm

deploy/migrations/clickhouse/xatu/006_libp2p_gossipsub_message_payload.down.sql

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

deploy/migrations/clickhouse/xatu/006_libp2p_gossipsub_message_payload.up.sql

@@ -0,0 +1,39 @@
+CREATE TABLE IF NOT EXISTS libp2p_gossipsub_message_payload_local ON CLUSTER '{cluster}'
+(
+    `updated_date_time` DateTime COMMENT 'Timestamp when the record was last updated' CODEC(DoubleDelta, ZSTD(1)),
+    `event_date_time` DateTime64(3) COMMENT 'Timestamp of the event' CODEC(DoubleDelta, ZSTD(1)),
+    `wallclock_slot` UInt32 COMMENT 'Slot number of the wall clock when the message was received' CODEC(DoubleDelta, ZSTD(1)),
+    `wallclock_slot_start_date_time` DateTime COMMENT 'Start date and time of the wall clock slot when the message was received' CODEC(DoubleDelta, ZSTD(1)),
+    `wallclock_epoch` UInt32 COMMENT 'Epoch number of the wall clock when the message was received' CODEC(DoubleDelta, ZSTD(1)),
+    `wallclock_epoch_start_date_time` DateTime COMMENT 'Start date and time of the wall clock epoch when the message was received' CODEC(DoubleDelta, ZSTD(1)),
+    `topic_layer` LowCardinality(String) COMMENT 'Layer of the topic',
+    `topic_fork_digest_value` LowCardinality(String) COMMENT 'Fork digest value of the topic',
+    `topic_name` LowCardinality(String) COMMENT 'Name of the topic',
+    `topic_encoding` LowCardinality(String) COMMENT 'Encoding of the topic',
+    `message_id` String COMMENT 'Gossipsub message ID, derived from the message contents' CODEC(ZSTD(1)),
+    `message_size` UInt32 COMMENT 'Size of the message payload in bytes' CODEC(ZSTD(1)),
+    `message_data` String COMMENT 'Raw gossipsub message payload as received off the wire (snappy-framed SSZ)' CODEC(ZSTD(1)),
+    `meta_client_name` LowCardinality(String) COMMENT 'Name of the client that generated the event',
+    `meta_client_version` LowCardinality(String) COMMENT 'Version of the client that generated the event',
+    `meta_client_implementation` LowCardinality(String) COMMENT 'Implementation of the client that generated the event',
+    `meta_client_os` LowCardinality(String) COMMENT 'Operating system of the client that generated the event',
+    `meta_client_ip` Nullable(IPv6) COMMENT 'IP address of the client that generated the event' CODEC(ZSTD(1)),
+    `meta_client_geo_city` LowCardinality(String) COMMENT 'City of the client that generated the event' CODEC(ZSTD(1)),
+    `meta_client_geo_country` LowCardinality(String) COMMENT 'Country of the client that generated the event' CODEC(ZSTD(1)),
+    `meta_client_geo_country_code` LowCardinality(String) COMMENT 'Country code of the client that generated the event' CODEC(ZSTD(1)),
+    `meta_client_geo_continent_code` LowCardinality(String) COMMENT 'Continent code of the client that generated the event' CODEC(ZSTD(1)),
+    `meta_client_geo_longitude` Nullable(Float64) COMMENT 'Longitude of the client that generated the event' CODEC(ZSTD(1)),
+    `meta_client_geo_latitude` Nullable(Float64) COMMENT 'Latitude of the client that generated the event' CODEC(ZSTD(1)),
+    `meta_client_geo_autonomous_system_number` Nullable(UInt32) COMMENT 'Autonomous system number of the client that generated the event' CODEC(ZSTD(1)),
+    `meta_client_geo_autonomous_system_organization` Nullable(String) COMMENT 'Autonomous system organization of the client that generated the event' CODEC(ZSTD(1)),
+    `meta_network_name` LowCardinality(String) COMMENT 'Ethereum network name'
+)
+ENGINE = ReplicatedReplacingMergeTree('/clickhouse/{installation}/{cluster}/tables/{shard}/{database}/{table}', '{replica}', updated_date_time)
+PARTITION BY (meta_network_name, toDate(wallclock_slot_start_date_time))
+ORDER BY (meta_network_name, wallclock_slot_start_date_time, topic_fork_digest_value, topic_name, message_id)
+COMMENT 'Contains raw gossipsub message payloads keyed by message ID. Message IDs are content-derived, and the sorting key deliberately excludes observation-specific columns (peer, client, receive time, validation outcome) so identical messages captured by multiple clients deduplicate on merge. Deduplication is best-effort: clients that receive the same message on opposite sides of a wallclock slot or partition boundary keep one row per side. Per-observation detail lives in the libp2p_deliver_message and libp2p_reject_message tables.';
+
+CREATE TABLE IF NOT EXISTS libp2p_gossipsub_message_payload ON CLUSTER '{cluster}'
+AS libp2p_gossipsub_message_payload_local
+ENGINE = Distributed('{cluster}', currentDatabase(), 'libp2p_gossipsub_message_payload_local', cityHash64(meta_network_name, topic_fork_digest_value, topic_name, message_id))
+COMMENT 'Contains raw gossipsub message payloads keyed by message ID. The sharding key is content-derived so duplicates from multiple capture clients land on the same shard and deduplicate.';

example-cl-mimicry.yaml

@@ -130,6 +130,7 @@ outputs:
 #   gossipSubAttestationEnabled: false
 #   gossipSubBlobSidecarEnabled: false
 #   gossipSubDataColumnSidecarEnabled: false
+#   gossipSubMessagePayloadEnabled: false
 
 tracing:
   enabled: false

pkg/clickhouse/route/libp2p/libp2p_gossipsub_message_payload.gen.go

@@ -0,0 +1,130 @@
+package libp2p
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/ethpandaops/xatu/pkg/clickhouse/route"
+	"github.com/ethpandaops/xatu/pkg/proto/xatu"
+)
+
+var libp2pGossipsubMessagePayloadEventNames = []xatu.Event_Name{
+	xatu.Event_LIBP2P_TRACE_GOSSIPSUB_MESSAGE_PAYLOAD,
+}
+
+func init() {
+	r, err := route.NewStaticRoute(
+		libp2pGossipsubMessagePayloadTableName,
+		libp2pGossipsubMessagePayloadEventNames,
+		func() route.ColumnarBatch { return newlibp2pGossipsubMessagePayloadBatch() },
+	)
+	if err != nil {
+		route.RecordError(err)
+
+		return
+	}
+
+	if err := route.Register(r); err != nil {
+		route.RecordError(err)
+	}
+}
+
+func (b *libp2pGossipsubMessagePayloadBatch) FlattenTo(
+	event *xatu.DecoratedEvent,
+) error {
+	if event == nil || event.GetEvent() == nil {
+		return nil
+	}
+
+	if event.GetLibp2PTraceGossipsubMessagePayload() == nil {
+		return fmt.Errorf("nil libp2p_trace_gossipsub_message_payload payload: %w", route.ErrInvalidEvent)
+	}
+
+	if err := b.validate(event); err != nil {
+		return err
+	}
+
+	b.appendRuntime(event)
+	b.appendMetadata(event)
+	b.appendPayload(event)
+	b.appendClientAdditionalData(event)
+	b.rows++
+
+	return nil
+}
+
+func (b *libp2pGossipsubMessagePayloadBatch) validate(event *xatu.DecoratedEvent) error {
+	additional := event.GetMeta().GetClient().GetLibp2PTraceGossipsubMessagePayload()
+	if additional == nil {
+		return fmt.Errorf("nil additional data: %w", route.ErrInvalidEvent)
+	}
+
+	if additional.GetMessageId() == nil {
+		return fmt.Errorf("nil MessageId: %w", route.ErrInvalidEvent)
+	}
+
+	if additional.GetWallclockSlot() == nil {
+		return fmt.Errorf("nil WallclockSlot: %w", route.ErrInvalidEvent)
+	}
+
+	if additional.GetWallclockEpoch() == nil {
+		return fmt.Errorf("nil WallclockEpoch: %w", route.ErrInvalidEvent)
+	}
+
+	return nil
+}
+
+func (b *libp2pGossipsubMessagePayloadBatch) appendRuntime(event *xatu.DecoratedEvent) {
+	b.UpdatedDateTime.Append(time.Now())
+
+	if ts := event.GetEvent().GetDateTime(); ts != nil {
+		b.EventDateTime.Append(ts.AsTime())
+	} else {
+		b.EventDateTime.Append(time.Time{})
+	}
+}
+
+func (b *libp2pGossipsubMessagePayloadBatch) appendPayload(event *xatu.DecoratedEvent) {
+	payload := event.GetLibp2PTraceGossipsubMessagePayload()
+
+	b.MessageData.AppendBytes(payload.GetData())
+}
+
+//nolint:gosec // G115: proto uint64 values are bounded by ClickHouse column schema
+func (b *libp2pGossipsubMessagePayloadBatch) appendClientAdditionalData(
+	event *xatu.DecoratedEvent,
+) {
+	// validate() rejects the event before any append when the additional data,
+	// wallclock fields, or message ID are missing, so required fields are
+	// appended unconditionally rather than falling back to sentinel zeros.
+	additional := event.GetMeta().GetClient().GetLibp2PTraceGossipsubMessagePayload()
+
+	wallclockSlot := additional.GetWallclockSlot()
+	b.WallclockSlot.Append(uint32(wallclockSlot.GetNumber().GetValue()))
+	b.WallclockSlotStartDateTime.Append(wallclockSlot.GetStartDateTime().AsTime())
+
+	wallclockEpoch := additional.GetWallclockEpoch()
+	b.WallclockEpoch.Append(uint32(wallclockEpoch.GetNumber().GetValue()))
+	b.WallclockEpochStartDateTime.Append(wallclockEpoch.GetStartDateTime().AsTime())
+
+	b.MessageID.Append(wrappedStringValue(additional.GetMessageId()))
+
+	if msgSize := additional.GetMessageSize(); msgSize != nil {
+		b.MessageSize.Append(msgSize.GetValue())
+	} else {
+		b.MessageSize.Append(0)
+	}
+
+	if topic := wrappedStringValue(additional.GetTopic()); topic != "" {
+		parsed := parseTopicFields(topic)
+		b.TopicLayer.Append(parsed.Layer)
+		b.TopicForkDigestValue.Append(parsed.ForkDigestValue)
+		b.TopicName.Append(parsed.Name)
+		b.TopicEncoding.Append(parsed.Encoding)
+	} else {
+		b.TopicLayer.Append("")
+		b.TopicForkDigestValue.Append("")
+		b.TopicName.Append("")
+		b.TopicEncoding.Append("")
+	}
+}