fix(metrics): avoid consuming protocol frames before app readers (libp2p#3394)

* fix(metrics): avoid consuming protocol frames before app readers

---------

Co-authored-by: lodekeeper <lodekeeper@users.noreply.github.com>
Co-authored-by: dozyio <37986489+dozyio@users.noreply.github.com>

Author: lodekeeper

packages/metrics-opentelemetry/src/index.ts

@@ -152,9 +152,21 @@ class OpenTelemetryMetrics implements Metrics {
    * in and out
    */
   _track (stream: MessageStream, name: string): void {
-    stream.addEventListener('message', (evt) => {
-      this._incrementValue(`${name} received`, evt.data.byteLength)
-    })
+    // AbstractMessageStream dispatches buffered inbound data only when there is
+    // at least one `message` listener. Adding a metrics listener too early can
+    // consume the read buffer before protocol handlers attach their own readers
+    // (e.g. identify), causing lost first frames.
+    //
+    // Instead, wrap dispatchEvent to observe already-dispatched message events
+    // without increasing listenerCount('message').
+    const dispatchEvent = stream.dispatchEvent.bind(stream)
+    stream.dispatchEvent = (evt: Event): boolean => {
+      if (evt.type === 'message') {
+        this._incrementValue(`${name} received`, (evt as MessageEvent<{ byteLength: number }>).data.byteLength)
+      }
+
+      return dispatchEvent(evt)
+    }
 
     const send = stream.send.bind(stream)
     stream.send = (buf) => {

packages/metrics-prometheus/src/index.ts

@@ -236,9 +236,23 @@ class PrometheusMetrics implements Metrics {
    * in and out
    */
   _track (stream: MessageStream, name: string): void {
-    stream.addEventListener('message', (evt) => {
-      this._incrementValue(`${name} received`, evt.data.byteLength)
-    })
+    // Do not attach a `message` listener here.
+    //
+    // AbstractMessageStream dispatches buffered inbound data only when there is
+    // at least one `message` listener. Adding a metrics listener too early can
+    // cause the read buffer to be consumed before protocol handlers attach their
+    // own readers (e.g. identify), leading to lost first frames.
+    //
+    // Instead, wrap dispatchEvent to observe already-dispatched message events
+    // without increasing listenerCount('message').
+    const dispatchEvent = stream.dispatchEvent.bind(stream)
+    stream.dispatchEvent = (evt: Event): boolean => {
+      if (evt.type === 'message') {
+        this._incrementValue(`${name} received`, (evt as MessageEvent<{ byteLength: number }>).data.byteLength)
+      }
+
+      return dispatchEvent(evt)
+    }
 
     const send = stream.send.bind(stream)
     stream.send = (buf) => {

packages/metrics-prometheus/test/streams.spec.ts

@@ -40,10 +40,21 @@ describe('streams', () => {
     // track incoming stream
     metrics.trackMultiaddrConnection(outbound)
 
+    // attach reader before data arrives so the stream dispatches message events
+    const iterator = outbound[Symbol.asyncIterator]()
+
     // send data to the remote over the tracked stream
     const data = Uint8Array.from([0, 1, 2, 3, 4])
     inbound.send(data)
 
+    const first = await Promise.race([
+      iterator.next(),
+      new Promise<never>((_resolve, reject) => setTimeout(() => reject(new Error('timed out waiting for first frame')), 200))
+    ])
+
+    expect(first.done).to.equal(false)
+    expect(first.value?.byteLength).to.equal(data.length)
+
     await Promise.all([
       pEvent(inbound, 'close'),
       outbound.close(),
@@ -78,6 +89,43 @@ describe('streams', () => {
     expect(scrapedMetrics).to.include(`libp2p_data_transfer_bytes_total{protocol="${outbound.protocol} sent"} ${data.length}`)
   })
 
+  it('should not consume early protocol data before app listeners are attached', async () => {
+    const [outbound, inbound] = await streamPair()
+
+    const metrics = prometheusMetrics()({
+      logger: defaultLogger()
+    })
+
+    // Track stream before any app-level reader/listener is attached
+    metrics.trackProtocolStream(outbound)
+
+    // Remote sends data first
+    const data = Uint8Array.from([9, 8, 7, 6, 5])
+    inbound.send(data)
+
+    // Allow data to arrive and be buffered
+    await new Promise((resolve) => setTimeout(resolve, 25))
+
+    // App starts reading later - should still receive first frame
+    const iterator = outbound[Symbol.asyncIterator]()
+    const first = await Promise.race([
+      iterator.next(),
+      new Promise<never>((_resolve, reject) => setTimeout(() => reject(new Error('timed out waiting for first frame')), 200))
+    ])
+
+    expect(first.done).to.equal(false)
+    expect(first.value?.byteLength).to.equal(data.length)
+
+    await Promise.all([
+      pEvent(inbound, 'close'),
+      outbound.close(),
+      inbound.close()
+    ])
+
+    const scrapedMetrics = await client.register.metrics()
+    expect(scrapedMetrics).to.include(`libp2p_data_transfer_bytes_total{protocol="${inbound.protocol} received"} ${data.length}`)
+  })
+
   it('should track bytes received over outbound streams', async () => {
     const [outbound, inbound] = await streamPair()
 
@@ -88,10 +136,21 @@ describe('streams', () => {
     // track incoming stream
     metrics.trackProtocolStream(outbound)
 
+    // attach reader before data arrives so the stream dispatches message events
+    const iterator = outbound[Symbol.asyncIterator]()
+
     // send data from remote to local
     const data = Uint8Array.from([0, 1, 2, 3, 4])
     inbound.send(data)
 
+    const first = await Promise.race([
+      iterator.next(),
+      new Promise<never>((_resolve, reject) => setTimeout(() => reject(new Error('timed out waiting for first frame')), 200))
+    ])
+
+    expect(first.done).to.equal(false)
+    expect(first.value?.byteLength).to.equal(data.length)
+
     await Promise.all([
       pEvent(inbound, 'close'),
       outbound.close(),

packages/metrics-simple/src/index.ts

@@ -416,9 +416,21 @@ class SimpleMetrics implements Metrics, Startable {
    * in and out
    */
   _track (stream: MessageStream, name: string): void {
-    stream.addEventListener('message', (evt) => {
-      this._incrementValue(`${name} received`, evt.data.byteLength)
-    })
+    // AbstractMessageStream dispatches buffered inbound data only when there is
+    // at least one `message` listener. Adding a metrics listener too early can
+    // consume the read buffer before protocol handlers attach their own readers
+    // (e.g. identify), causing lost first frames.
+    //
+    // Instead, wrap dispatchEvent to observe already-dispatched message events
+    // without increasing listenerCount('message').
+    const dispatchEvent = stream.dispatchEvent.bind(stream)
+    stream.dispatchEvent = (evt: Event): boolean => {
+      if (evt.type === 'message') {
+        this._incrementValue(`${name} received`, (evt as MessageEvent<{ byteLength: number }>).data.byteLength)
+      }
+
+      return dispatchEvent(evt)
+    }
 
     const send = stream.send.bind(stream)
     stream.send = (buf) => {