fix(identify): handle multiple envelopes from go

packages/protocol-identify/src/consts.ts

@@ -11,6 +11,13 @@ export const MULTICODEC_IDENTIFY_PUSH_PROTOCOL_VERSION = '1.0.0'
 // https://github.com/libp2p/go-libp2p/blob/8d2e54e1637041d5cf4fac1e531287560bd1f4ac/p2p/protocol/identify/id.go#L52
 export const MAX_IDENTIFY_MESSAGE_SIZE = 1024 * 8
 
+// Maximum number of LP-framed messages we will read from a single identify or
+// identify-push stream before treating remaining data as truncated. The
+// identify spec is silent on this; the value matches go-libp2p's `maxMessages`
+// by convention. See the proposed update at
+// https://github.com/libp2p/specs/pull/709
+export const MAX_IDENTIFY_MESSAGES = 10
+
 // https://github.com/libp2p/go-libp2p/blob/0385ec924bad172f74a74db09939e97c079b1420/p2p/protocol/identify/id.go#L47C7-L47C25
 export const MAX_PUSH_CONCURRENCY = 32
 

packages/protocol-identify/src/identify-push.ts

@@ -13,7 +13,7 @@ import {
   PUSH_DEBOUNCE_MS
 } from './consts.ts'
 import { Identify as IdentifyMessage } from './pb/message.ts'
-import { AbstractIdentify, consumeIdentifyMessage, defaultValues } from './utils.ts'
+import { AbstractIdentify, consumeIdentifyMessage, defaultValues, mergeIdentifyMessages, readIdentifyMessages } from './utils.ts'
 import type { IdentifyPush as IdentifyPushInterface, IdentifyPushComponents, IdentifyPushInit } from './index.ts'
 import type { Stream, Startable, Connection } from '@libp2p/interface'
 import type { ConnectionManager } from '@libp2p/interface-internal'
@@ -144,14 +144,9 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif
       signal: AbortSignal.timeout(this.timeout)
     }
 
-    const pb = pbStream(stream, {
-      maxDataLength: this.maxMessageSize
-    }).pb(IdentifyMessage)
+    const messages = await readIdentifyMessages(stream, this.maxMessageSize, options, log)
 
-    const message = await pb.read(options)
-    await stream.close(options)
-
-    await consumeIdentifyMessage(this.components.peerStore, this.components.events, log, connection, message)
+    await consumeIdentifyMessage(this.components.peerStore, this.components.events, log, connection, mergeIdentifyMessages(messages))
 
     log.trace('handled push from %p', connection.remotePeer)
   }

packages/protocol-identify/src/identify.ts

@@ -11,7 +11,7 @@ import {
   MULTICODEC_IDENTIFY_PROTOCOL_VERSION
 } from './consts.ts'
 import { Identify as IdentifyMessage } from './pb/message.ts'
-import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr } from './utils.ts'
+import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, mergeIdentifyMessages, readIdentifyMessages } from './utils.ts'
 import type { Identify as IdentifyInterface, IdentifyComponents, IdentifyInit } from './index.ts'
 import type { IdentifyResult, AbortOptions, Connection, Stream, Startable, Logger, NewStreamOptions } from '@libp2p/interface'
 
@@ -60,14 +60,9 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt
       })
       log = stream.log.newScope('identify')
 
-      const pb = pbStream(stream, {
-        maxDataLength: this.maxMessageSize
-      }).pb(IdentifyMessage)
+      const messages = await readIdentifyMessages(stream, this.maxMessageSize, options, log)
 
-      const message = await pb.read(options)
-      await pb.unwrap().unwrap().close(options)
-
-      return message
+      return mergeIdentifyMessages(messages)
     } catch (err: any) {
       log?.error('identify failed - %e', err)
       stream?.abort(err)

packages/protocol-identify/src/utils.ts

@@ -2,12 +2,13 @@ import { publicKeyFromProtobuf } from '@libp2p/crypto/keys'
 import { InvalidMessageError } from '@libp2p/interface'
 import { peerIdFromCID, peerIdFromPublicKey } from '@libp2p/peer-id'
 import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record'
+import { pbStream } from '@libp2p/utils'
 import { multiaddr } from '@multiformats/multiaddr'
 import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
-import { IDENTIFY_PROTOCOL_VERSION, MAX_IDENTIFY_MESSAGE_SIZE, MAX_PUSH_CONCURRENCY } from './consts.ts'
+import { IDENTIFY_PROTOCOL_VERSION, MAX_IDENTIFY_MESSAGE_SIZE, MAX_IDENTIFY_MESSAGES, MAX_PUSH_CONCURRENCY } from './consts.ts'
+import { Identify as IdentifyMessage } from './pb/message.ts'
 import type { IdentifyComponents, IdentifyInit } from './index.ts'
-import type { Identify as IdentifyMessage } from './pb/message.ts'
-import type { Libp2pEvents, IdentifyResult, SignedPeerRecord, Logger, Connection, Peer, PeerData, PeerStore, Startable, Stream } from '@libp2p/interface'
+import type { AbortOptions, Libp2pEvents, IdentifyResult, SignedPeerRecord, Logger, Connection, Peer, PeerData, PeerStore, Startable, Stream } from '@libp2p/interface'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { TypedEventTarget } from 'main-event'
 
@@ -171,6 +172,82 @@ export async function consumeIdentifyMessage (peerStore: PeerStore, events: Type
   return result
 }
 
+/**
+ * Merge multiple received Identify messages into one. Repeated `listenAddrs`
+ * are concatenated as-is — peerstore handles deduplication downstream.
+ * `protocols` are deduplicated via Set since they're string identifiers.
+ */
+export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMessage {
+  const merged: IdentifyMessage = { ...messages[0] }
+
+  for (const msg of messages.slice(1)) {
+    if (msg.protocolVersion != null) {
+      merged.protocolVersion = msg.protocolVersion
+    }
+    if (msg.agentVersion != null) {
+      merged.agentVersion = msg.agentVersion
+    }
+    if (msg.publicKey != null) {
+      merged.publicKey = msg.publicKey
+    }
+    if (msg.observedAddr != null) {
+      merged.observedAddr = msg.observedAddr
+    }
+    if (msg.signedPeerRecord != null) {
+      merged.signedPeerRecord = msg.signedPeerRecord
+    }
+    merged.listenAddrs = [...merged.listenAddrs, ...msg.listenAddrs]
+    merged.protocols = [...new Set([...merged.protocols, ...msg.protocols])]
+  }
+
+  return merged
+}
+
+/**
+ * Read up to MAX_IDENTIFY_MESSAGES LP-framed Identify messages from the
+ * stream, then close. Used by both identify and identify-push receive paths.
+ *
+ * Any error after at least one successful read is treated as "stop reading"
+ * (we keep what we got — peerstore handles deduplication/merge downstream).
+ * Close errors are swallowed and the stream is aborted instead — preserves
+ * the read result while ensuring transport cleanup.
+ *
+ * Multi-message identify is per the proposed spec update at
+ * https://github.com/libp2p/specs/pull/709.
+ */
+export async function readIdentifyMessages (stream: Stream, maxMessageSize: number, options: AbortOptions, log: Logger): Promise<IdentifyMessage[]> {
+  const pb = pbStream(stream, {
+    maxDataLength: maxMessageSize
+  }).pb(IdentifyMessage)
+
+  const messages: IdentifyMessage[] = []
+
+  for (let i = 0; i < MAX_IDENTIFY_MESSAGES; i++) {
+    try {
+      messages.push(await pb.read(options))
+    } catch (err: any) {
+      if (messages.length === 0) {
+        throw err
+      }
+      log.trace('stopped reading identify - %e', err)
+      break
+    }
+  }
+
+  if (messages.length >= MAX_IDENTIFY_MESSAGES) {
+    log('reached MAX_IDENTIFY_MESSAGES, returning truncated identify')
+  }
+
+  try {
+    await stream.close(options)
+  } catch (err: any) {
+    log.trace('error closing identify stream after read - %e', err)
+    stream.abort(err)
+  }
+
+  return messages
+}
+
 export interface AbstractIdentifyInit extends IdentifyInit {
   protocol: string
   log: Logger

packages/protocol-identify/test/index.spec.ts

@@ -82,6 +82,7 @@ describe('identify', () => {
 
     const [outgoingStream, incomingStream] = await streamPair()
     incomingStream.send(lp.encode.single(IdentifyMessage.encode(message)))
+    void incomingStream.close()
     const connection = stubInterface<Connection>({
       remotePeer
     })
@@ -109,6 +110,7 @@ describe('identify', () => {
       protocols: [],
       publicKey: publicKeyToProtobuf(otherPeer.publicKey)
     })))
+    void incomingStream.close()
     const connection = stubInterface<Connection>({
       remotePeer
     })
@@ -164,6 +166,151 @@ describe('identify', () => {
     expect(outgoingStream).to.have.property('status', 'aborted')
   })
 
+  it('should succeed if the remote closes the stream after sending the identify response', async () => {
+    identify = new Identify(components)
+
+    await start(identify)
+
+    const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519'))
+    const message: IdentifyMessage = {
+      listenAddrs: [
+        multiaddr('/ip4/123.123.123.123/tcp/123').bytes
+      ],
+      protocols: [
+        '/foo/bar/1.0'
+      ],
+      publicKey: publicKeyToProtobuf(remotePeer.publicKey)
+    }
+
+    const [outgoingStream] = await streamPair()
+    const connection = stubInterface<Connection>({
+      remotePeer
+    })
+    connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream)
+
+    const identifyPromise = identify.identify(connection)
+
+    // Wait for identify to register its stream listener
+    await new Promise<void>(resolve => setTimeout(resolve, 0))
+
+    // send the identify message with a trailing byte then close immediately.
+    const encoded = lp.encode.single(IdentifyMessage.encode(message)).subarray()
+    const combined = new Uint8Array(encoded.byteLength + 1)
+    combined.set(encoded)
+    outgoingStream.push(combined) // appends to readBuffer, schedules setTimeout(dispatchReadBuffer, 0)
+    outgoingStream.remoteWriteStatus = 'closed' // set before dispatchReadBuffer fires
+
+    const response = await identifyPromise
+
+    expect(response.peerId.toString()).to.equal(remotePeer.toString())
+    expect(response.protocols).to.deep.equal(message.protocols)
+    expect(response.listenAddrs.map(ma => ma.toString())).to.deep.equal(['/ip4/123.123.123.123/tcp/123'])
+  })
+
+  it('should merge multiple identify messages from the remote', async () => {
+    identify = new Identify(components)
+
+    await start(identify)
+
+    const remotePrivateKey = await generateKeyPair('Ed25519')
+    const remotePeer = peerIdFromPrivateKey(remotePrivateKey)
+
+    const signedPeerRecord = await RecordEnvelope.seal(new PeerRecord({
+      peerId: remotePeer,
+      multiaddrs: [
+        multiaddr('/ip4/127.0.0.1/tcp/5678')
+      ]
+    }), remotePrivateKey)
+    const peerRecordEnvelope = signedPeerRecord.marshal()
+
+    // simulate go-libp2p splitting a large identify response into two messages:
+    // first message has everything except signedPeerRecord, second has only signedPeerRecord
+    const firstMessage: IdentifyMessage = {
+      listenAddrs: [
+        multiaddr('/ip4/123.123.123.123/tcp/123').bytes
+      ],
+      protocols: [
+        '/foo/bar/1.0'
+      ],
+      publicKey: publicKeyToProtobuf(remotePeer.publicKey)
+    }
+    const secondMessage: IdentifyMessage = {
+      listenAddrs: [],
+      protocols: [],
+      signedPeerRecord: peerRecordEnvelope
+    }
+
+    const [outgoingStream] = await streamPair()
+    const connection = stubInterface<Connection>({
+      remotePeer
+    })
+    connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream)
+
+    const identifyPromise = identify.identify(connection)
+
+    // Wait for identify to register its stream listener
+    await new Promise<void>(resolve => setTimeout(resolve, 0))
+
+    const encoded1 = lp.encode.single(IdentifyMessage.encode(firstMessage)).subarray()
+    const encoded2 = lp.encode.single(IdentifyMessage.encode(secondMessage)).subarray()
+    const combined = new Uint8Array(encoded1.byteLength + encoded2.byteLength)
+    combined.set(encoded1)
+    combined.set(encoded2, encoded1.byteLength)
+    outgoingStream.push(combined) // appends to readBuffer, schedules setTimeout(dispatchReadBuffer, 0)
+    outgoingStream.remoteWriteStatus = 'closed' // set before dispatchReadBuffer fires
+
+    await identifyPromise
+
+    // should have stored the signedPeerRecord from the second message
+    expect(components.peerStore.patch.callCount).to.equal(1)
+    expect(components.peerStore.patch.getCall(0).args[1])
+      .to.have.property('peerRecordEnvelope').that.equalBytes(peerRecordEnvelope)
+  })
+
+  it('should propagate UnexpectedEOFError to the caller when remote closes without sending', async () => {
+    identify = new Identify(components)
+    await start(identify)
+
+    const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519'))
+    const [outgoingStream, incomingStream] = await streamPair()
+    const connection = stubInterface<Connection>({ remotePeer })
+    connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream)
+    void incomingStream.close()
+
+    await expect(identify.identify(connection))
+      .to.eventually.be.rejected()
+      .with.property('name', 'UnexpectedEOFError')
+  })
+
+  it('should still resolve with the message when close() throws after a successful read', async () => {
+    identify = new Identify(components)
+    await start(identify)
+
+    const remotePrivateKey = await generateKeyPair('Ed25519')
+    const remotePeer = peerIdFromPrivateKey(remotePrivateKey)
+    const message: IdentifyMessage = {
+      listenAddrs: [],
+      protocols: ['/foo/1.0'],
+      publicKey: publicKeyToProtobuf(remotePeer.publicKey)
+    }
+
+    const [outgoingStream, incomingStream] = await streamPair()
+    const connection = stubInterface<Connection>({ remotePeer })
+    connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream)
+
+    const originalClose = outgoingStream.close.bind(outgoingStream)
+    outgoingStream.close = async (...args: any[]) => {
+      await originalClose(...args)
+      throw Object.assign(new Error('simulated close failure'), { name: 'StreamStateError' })
+    }
+
+    incomingStream.send(lp.encode.single(IdentifyMessage.encode(message)))
+    void incomingStream.close()
+
+    const result = await identify.identify(connection)
+    expect(result.protocols).to.include('/foo/1.0')
+  })
+
   it('should limit incoming identify message sizes', async () => {
     const maxMessageSize = 100
 
@@ -204,6 +351,7 @@ describe('identify', () => {
       agentVersion: 'secret-agent',
       protocolVersion: '9000'
     })))
+    void incomingStream.close()
     const connection = stubInterface<Connection>({
       remotePeer
     })
@@ -257,6 +405,7 @@ describe('identify', () => {
       publicKey: publicKeyToProtobuf(remotePeer.publicKey),
       signedPeerRecord: oldPeerRecord.marshal()
     })))
+    void incomingStream.close()
     const connection = stubInterface<Connection>({
       remotePeer
     })
@@ -308,6 +457,7 @@ describe('identify', () => {
 
     const [outgoingStream, incomingStream] = await streamPair()
     incomingStream.send(lp.encode.single(IdentifyMessage.encode(message)))
+    void incomingStream.close()
     const connection = stubInterface<Connection>({
       remotePeer
     })
@@ -356,6 +506,7 @@ describe('identify', () => {
 
     const [outgoingStream, incomingStream] = await streamPair()
     incomingStream.send(lp.encode.single(IdentifyMessage.encode(message)))
+    void incomingStream.close()
     const connection = stubInterface<Connection>({
       remotePeer
     })
@@ -430,6 +581,7 @@ describe('identify', () => {
 
     const [outgoingStream, incomingStream] = await streamPair()
     incomingStream.send(lp.encode.single(IdentifyMessage.encode(message)))
+    void incomingStream.close()
     const connection = stubInterface<Connection>({
       remotePeer
     })