feat: track DataTransform metrics (#8639)

**Motivation**

- we usually have to uncompress more messages than needed so it's good
to track it in DataTransform

**Description**

- track `compress` and `uncompress` times by topic type
- in this instance, this node only subscribe to 8 column subnets but we
usually have to uncompress 9 or up to 10/11 DataColumnSidecars per slot,
they will likely be duplicated in the end. See also
https://github.com/ChainSafe/js-libp2p-gossipsub/pull/536

part of #8629
<img width="1046" height="485" alt="Screenshot 2025-11-28 at 17 02 42"
src="https://github.com/user-attachments/assets/df190b29-6681-48de-a04e-cd79ab82858d"
/>

Co-authored-by: Tuyen Nguyen <twoeths@users.noreply.github.com>

packages/beacon-node/src/network/gossip/encoding.ts

@@ -8,6 +8,7 @@ import {DataTransform} from "@chainsafe/libp2p-gossipsub/types";
 import {ForkName} from "@lodestar/params";
 import {intToBytes} from "@lodestar/utils";
 import {MESSAGE_DOMAIN_VALID_SNAPPY} from "./constants.js";
+import {Eth2GossipsubMetrics} from "./metrics.js";
 import {GossipTopicCache, getGossipSSZType} from "./topic.js";
 
 // Load WASM
@@ -70,7 +71,8 @@ export function msgIdFn(gossipTopicCache: GossipTopicCache, msg: Message): Uint8
 export class DataTransformSnappy implements DataTransform {
   constructor(
     private readonly gossipTopicCache: GossipTopicCache,
-    private readonly maxSizePerMessage: number
+    private readonly maxSizePerMessage: number,
+    private readonly metrics: Eth2GossipsubMetrics | null
   ) {}
 
   /**
@@ -87,6 +89,7 @@ export class DataTransformSnappy implements DataTransform {
     const uncompressedDataLength = uncompressedData.length;
     const topic = this.gossipTopicCache.getTopic(topicStr);
     const sszType = getGossipSSZType(topic);
+    this.metrics?.dataTransform.inbound.inc({type: topic.type});
 
     if (uncompressedDataLength < sszType.minSize) {
       throw Error(`ssz_snappy decoded data length ${uncompressedDataLength} < ${sszType.minSize}`);
@@ -102,7 +105,9 @@ export class DataTransformSnappy implements DataTransform {
    * Takes the data to be published (a topic and associated data) transforms the data. The
    * transformed data will then be used to create a `RawGossipsubMessage` to be sent to peers.
    */
-  outboundTransform(_topicStr: string, data: Uint8Array): Uint8Array {
+  outboundTransform(topicStr: string, data: Uint8Array): Uint8Array {
+    const topic = this.gossipTopicCache.getTopic(topicStr);
+    this.metrics?.dataTransform.outbound.inc({type: topic.type});
     if (data.length > this.maxSizePerMessage) {
       throw Error(`ssz_snappy encoded data length ${data.length} > ${this.maxSizePerMessage}`);
     }

packages/beacon-node/src/network/gossip/gossipsub.ts

@@ -89,6 +89,13 @@ export class Eth2Gossipsub extends GossipSub {
     const gossipTopicCache = new GossipTopicCache(config);
 
     const scoreParams = computeGossipPeerScoreParams({config, eth2Context: modules.eth2Context});
+    let metrics: Eth2GossipsubMetrics | null = null;
+    if (metricsRegister) {
+      metrics = createEth2GossipsubMetrics(metricsRegister);
+      metrics.gossipMesh.peersByType.addCollect(() =>
+        this.onScrapeLodestarMetrics(metrics as Eth2GossipsubMetrics, networkConfig)
+      );
+    }
 
     // Gossipsub parameters defined here:
     // https://github.com/ethereum/consensus-specs/blob/v1.1.10/specs/phase0/p2p-interface.md#the-gossip-domain-gossipsub
@@ -116,7 +123,7 @@ export class Eth2Gossipsub extends GossipSub {
       fastMsgIdFn: fastMsgIdFn,
       msgIdFn: msgIdFn.bind(msgIdFn, gossipTopicCache),
       msgIdToStrFn: msgIdToStrFn,
-      dataTransform: new DataTransformSnappy(gossipTopicCache, config.MAX_PAYLOAD_SIZE),
+      dataTransform: new DataTransformSnappy(gossipTopicCache, config.MAX_PAYLOAD_SIZE, metrics),
       metricsRegister: metricsRegister as MetricsRegister | null,
       metricsTopicStrToLabel: metricsRegister
         ? getMetricsTopicStrToLabel(networkConfig, {disableLightClientServer: opts.disableLightClientServer ?? false})
@@ -141,11 +148,6 @@ export class Eth2Gossipsub extends GossipSub {
     this.events = events;
     this.gossipTopicCache = gossipTopicCache;
 
-    if (metricsRegister) {
-      const metrics = createEth2GossipsubMetrics(metricsRegister);
-      metrics.gossipMesh.peersByType.addCollect(() => this.onScrapeLodestarMetrics(metrics, networkConfig));
-    }
-
     this.addEventListener("gossipsub:message", this.onGossipsubMessage.bind(this));
     this.events.on(NetworkEvent.gossipMessageValidationResult, this.onValidationResult.bind(this));
 

packages/beacon-node/src/network/gossip/metrics.ts

@@ -67,5 +67,17 @@ export function createEth2GossipsubMetrics(register: RegistryMetricCreator) {
         labelNames: ["subnet", "boundary"],
       }),
     },
+    dataTransform: {
+      inbound: register.counter<{type: GossipType}>({
+        name: "lodestar_gossip_data_transform_inbound_total",
+        help: "Total number of inbound data transforms by gossip type",
+        labelNames: ["type"],
+      }),
+      outbound: register.counter<{type: GossipType}>({
+        name: "lodestar_gossip_data_transform_outbound_total",
+        help: "Total number of outbound data transforms by gossip type",
+        labelNames: ["type"],
+      }),
+    },
   };
 }