From f2cd51212f1d15bf87befb2e8f3c3bfc4f89b760 Mon Sep 17 00:00:00 2001 From: William Morriss Date: Fri, 27 Mar 2026 04:53:49 -0500 Subject: [PATCH 01/21] fix close crash --- packages/protocol-identify/src/identify.ts | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index 89936de0ae..46786812d1 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/packages/protocol-identify/src/identify.ts @@ -1,5 +1,5 @@ import { publicKeyFromProtobuf, publicKeyToProtobuf } from '@libp2p/crypto/keys' -import { InvalidMessageError, serviceCapabilities } from '@libp2p/interface' +import { InvalidMessageError, StreamStateError, serviceCapabilities } from '@libp2p/interface' import { peerIdFromCID } from '@libp2p/peer-id' import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record' import { isGlobalUnicast, isPrivate, pbStream } from '@libp2p/utils' @@ -65,7 +65,14 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt }).pb(IdentifyMessage) const message = await pb.read(options) - await pb.unwrap().unwrap().close(options) + try { + await pb.unwrap().unwrap().close(options) + } catch (err) { + // Remote may have already closed the stream, triggering a StreamStateError + if (!(err instanceof StreamStateError)) { + throw err + } + } return message } catch (err: any) { From 66f69848ffc51013c89a4f66e421ed8b60627d14 Mon Sep 17 00:00:00 2001 From: William Morriss Date: Thu, 2 Apr 2026 03:21:14 -0500 Subject: [PATCH 02/21] test: close after sending identify response Co-Authored-By: Claude Sonnet 4.6 --- packages/protocol-identify/test/index.spec.ts | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/packages/protocol-identify/test/index.spec.ts b/packages/protocol-identify/test/index.spec.ts index d170604834..bd25c11678 100644 --- a/packages/protocol-identify/test/index.spec.ts +++ b/packages/protocol-identify/test/index.spec.ts @@ -164,6 +164,47 @@ 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({ + 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(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 limit incoming identify message sizes', async () => { const maxMessageSize = 100 From 160f796bf4d4179275917ce3450302e6fe87d9e4 Mon Sep 17 00:00:00 2001 From: William Morriss Date: Mon, 6 Apr 2026 14:38:43 -0500 Subject: [PATCH 03/21] fix: read all identify messages until stream closes Co-Authored-By: Claude Sonnet 4.6 --- packages/protocol-identify/src/identify.ts | 59 +++++++++++++--- packages/protocol-identify/test/index.spec.ts | 67 +++++++++++++++++++ 2 files changed, 117 insertions(+), 9 deletions(-) diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index 46786812d1..b91c331ba7 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/packages/protocol-identify/src/identify.ts @@ -1,8 +1,8 @@ import { publicKeyFromProtobuf, publicKeyToProtobuf } from '@libp2p/crypto/keys' -import { InvalidMessageError, StreamStateError, serviceCapabilities } from '@libp2p/interface' +import { InvalidMessageError, serviceCapabilities } from '@libp2p/interface' import { peerIdFromCID } from '@libp2p/peer-id' import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record' -import { isGlobalUnicast, isPrivate, pbStream } from '@libp2p/utils' +import { UnexpectedEOFError, isGlobalUnicast, isPrivate, pbStream } from '@libp2p/utils' import { CODE_IP6, CODE_IP6ZONE, CODE_P2P } from '@multiformats/multiaddr' import { IP_OR_DOMAIN, TCP } from '@multiformats/multiaddr-matcher' import { setMaxListeners } from 'main-event' @@ -64,14 +64,55 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt maxDataLength: this.maxMessageSize }).pb(IdentifyMessage) - const message = await pb.read(options) - try { - await pb.unwrap().unwrap().close(options) - } catch (err) { - // Remote may have already closed the stream, triggering a StreamStateError - if (!(err instanceof StreamStateError)) { - throw err + // Read all messages until the stream closes - go-libp2p may send multiple + // messages for large identify responses (e.g. splitting off SignedPeerRecord) + const MAX_IDENTIFY_MESSAGES = 10 + const messages: IdentifyMessage[] = [] + + for (let i = 0; i < MAX_IDENTIFY_MESSAGES; i++) { + try { + messages.push(await pb.read(options)) + } catch (err) { + if (messages.length > 0 && err instanceof UnexpectedEOFError) { + break } + + throw err + } + } + + if (messages.length === 0) { + throw new InvalidMessageError('No identify message received') + } + + await pb.unwrap().unwrap().close(options) + + // Merge all messages into one - later messages supply missing fields + const message = messages[0] + + for (const msg of messages.slice(1)) { + if (msg.protocolVersion != null) { + message.protocolVersion = msg.protocolVersion + } + + if (msg.agentVersion != null) { + message.agentVersion = msg.agentVersion + } + + if (msg.publicKey != null) { + message.publicKey = msg.publicKey + } + + if (msg.observedAddr != null) { + message.observedAddr = msg.observedAddr + } + + if (msg.signedPeerRecord != null) { + message.signedPeerRecord = msg.signedPeerRecord + } + + message.listenAddrs = [...message.listenAddrs, ...msg.listenAddrs] + message.protocols = [...new Set([...message.protocols, ...msg.protocols])] } return message diff --git a/packages/protocol-identify/test/index.spec.ts b/packages/protocol-identify/test/index.spec.ts index bd25c11678..9eb7e4efe1 100644 --- a/packages/protocol-identify/test/index.spec.ts +++ b/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({ remotePeer }) @@ -109,6 +110,7 @@ describe('identify', () => { protocols: [], publicKey: publicKeyToProtobuf(otherPeer.publicKey) }))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) @@ -205,6 +207,66 @@ describe('identify', () => { 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({ + 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(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 limit incoming identify message sizes', async () => { const maxMessageSize = 100 @@ -245,6 +307,7 @@ describe('identify', () => { agentVersion: 'secret-agent', protocolVersion: '9000' }))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) @@ -298,6 +361,7 @@ describe('identify', () => { publicKey: publicKeyToProtobuf(remotePeer.publicKey), signedPeerRecord: oldPeerRecord.marshal() }))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) @@ -349,6 +413,7 @@ describe('identify', () => { const [outgoingStream, incomingStream] = await streamPair() incomingStream.send(lp.encode.single(IdentifyMessage.encode(message))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) @@ -397,6 +462,7 @@ describe('identify', () => { const [outgoingStream, incomingStream] = await streamPair() incomingStream.send(lp.encode.single(IdentifyMessage.encode(message))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) @@ -471,6 +537,7 @@ describe('identify', () => { const [outgoingStream, incomingStream] = await streamPair() incomingStream.send(lp.encode.single(IdentifyMessage.encode(message))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) From dce014e65887ea7ebd43aa4d9f0775a17b2cee73 Mon Sep 17 00:00:00 2001 From: William Morriss Date: Mon, 6 Apr 2026 14:43:28 -0500 Subject: [PATCH 04/21] doc: more-concise comments --- packages/protocol-identify/src/identify.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index b91c331ba7..8a7da9c004 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/packages/protocol-identify/src/identify.ts @@ -64,8 +64,8 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt maxDataLength: this.maxMessageSize }).pb(IdentifyMessage) - // Read all messages until the stream closes - go-libp2p may send multiple - // messages for large identify responses (e.g. splitting off SignedPeerRecord) + // Large responses can be subdivided. + // Read all messages until the stream closes. const MAX_IDENTIFY_MESSAGES = 10 const messages: IdentifyMessage[] = [] @@ -87,7 +87,7 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt await pb.unwrap().unwrap().close(options) - // Merge all messages into one - later messages supply missing fields + // Merge any subsequent identify messages into the first response const message = messages[0] for (const msg of messages.slice(1)) { From aa154ebcb86bb44445cc22e6bf840cedcdc19d42 Mon Sep 17 00:00:00 2001 From: William Morriss Date: Wed, 15 Apr 2026 13:59:09 -0500 Subject: [PATCH 05/21] feat(identify): support multi-message identify per spec PR #709 Assisted-by: Claude:claude-sonnet-4-6 --- packages/protocol-identify/src/consts.ts | 5 + .../protocol-identify/src/identify-push.ts | 35 ++- packages/protocol-identify/src/identify.ts | 40 +--- packages/protocol-identify/src/utils.ts | 126 ++++++++++- packages/protocol-identify/test/index.spec.ts | 101 +++++++++ packages/protocol-identify/test/push.spec.ts | 62 +++++- packages/protocol-identify/test/utils.spec.ts | 203 ++++++++++++++++++ 7 files changed, 526 insertions(+), 46 deletions(-) create mode 100644 packages/protocol-identify/test/utils.spec.ts diff --git a/packages/protocol-identify/src/consts.ts b/packages/protocol-identify/src/consts.ts index 0dde24239d..3613bb14e2 100644 --- a/packages/protocol-identify/src/consts.ts +++ b/packages/protocol-identify/src/consts.ts @@ -11,6 +11,11 @@ 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 +// Large identify messages are split into smaller messages +export const FIRST_IDENTIFY_MESSAGE_MAX_SIZE = 1024 * 2 +export const SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE = 1024 * 4 +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 diff --git a/packages/protocol-identify/src/identify-push.ts b/packages/protocol-identify/src/identify-push.ts index 114676fa67..5f62bd9d1e 100644 --- a/packages/protocol-identify/src/identify-push.ts +++ b/packages/protocol-identify/src/identify-push.ts @@ -1,6 +1,6 @@ -import { serviceCapabilities } from '@libp2p/interface' +import { InvalidMessageError, serviceCapabilities } from '@libp2p/interface' import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record' -import { debounce, pbStream } from '@libp2p/utils' +import { UnexpectedEOFError, debounce, pbStream } from '@libp2p/utils' import { CODE_P2P } from '@multiformats/multiaddr' import drain from 'it-drain' import parallel from 'it-parallel' @@ -13,7 +13,7 @@ import { PUSH_DEBOUNCE_MS } from './consts.js' import { Identify as IdentifyMessage } from './pb/message.js' -import { AbstractIdentify, consumeIdentifyMessage, defaultValues } from './utils.js' +import { AbstractIdentify, buildIdentifyMessages, consumeIdentifyMessage, defaultValues, mergeIdentifyMessages } from './utils.js' import type { IdentifyPush as IdentifyPushInterface, IdentifyPushComponents, IdentifyPushInit } from './index.js' import type { Stream, Startable, Connection } from '@libp2p/interface' import type { ConnectionManager } from '@libp2p/interface-internal' @@ -99,16 +99,18 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif maxDataLength: self.maxMessageSize }).pb(IdentifyMessage) - await pb.write({ + const msgs = buildIdentifyMessages({ listenAddrs: listenAddresses.map(ma => ma.bytes), signedPeerRecord: signedPeerRecord.marshal(), protocols: supportedProtocols, agentVersion, protocolVersion - }, { - signal }) + for (const msg of msgs) { + await pb.write(msg, { signal }) + } + await stream.close({ signal }) @@ -148,10 +150,27 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif maxDataLength: this.maxMessageSize }).pb(IdentifyMessage) - const message = await pb.read(options) + const messages: IdentifyMessage[] = [] + + for (let i = 0; i < 10; i++) { + try { + messages.push(await pb.read(options)) + } catch (err) { + if (messages.length > 0 && err instanceof UnexpectedEOFError) { + break + } + + throw err + } + } + + if (messages.length === 0) { + throw new InvalidMessageError('No identify message received') + } + 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) } diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index 4036f5c95a..61100b4426 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/packages/protocol-identify/src/identify.ts @@ -11,7 +11,7 @@ import { MULTICODEC_IDENTIFY_PROTOCOL_VERSION } from './consts.js' import { Identify as IdentifyMessage } from './pb/message.js' -import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr } from './utils.js' +import { AbstractIdentify, buildIdentifyMessages, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, mergeIdentifyMessages } from './utils.js' import type { Identify as IdentifyInterface, IdentifyComponents, IdentifyInit } from './index.js' import type { IdentifyResult, AbortOptions, Connection, Stream, Startable, Logger, NewStreamOptions } from '@libp2p/interface' @@ -87,35 +87,7 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt await pb.unwrap().unwrap().close(options) - // Merge any subsequent identify messages into the first response - const message = messages[0] - - for (const msg of messages.slice(1)) { - if (msg.protocolVersion != null) { - message.protocolVersion = msg.protocolVersion - } - - if (msg.agentVersion != null) { - message.agentVersion = msg.agentVersion - } - - if (msg.publicKey != null) { - message.publicKey = msg.publicKey - } - - if (msg.observedAddr != null) { - message.observedAddr = msg.observedAddr - } - - if (msg.signedPeerRecord != null) { - message.signedPeerRecord = msg.signedPeerRecord - } - - message.listenAddrs = [...message.listenAddrs, ...msg.listenAddrs] - message.protocols = [...new Set([...message.protocols, ...msg.protocols])] - } - - return message + return mergeIdentifyMessages(messages) } catch (err: any) { log?.error('identify failed - %e', err) stream?.abort(err) @@ -226,7 +198,7 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt const pb = pbStream(stream).pb(IdentifyMessage) log('send response') - await pb.write({ + const msgs = buildIdentifyMessages({ protocolVersion: this.host.protocolVersion, agentVersion: this.host.agentVersion, publicKey: publicKeyToProtobuf(this.components.privateKey.publicKey), @@ -234,10 +206,12 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt signedPeerRecord, observedAddr, protocols: peerData.protocols - }, { - signal }) + for (const msg of msgs) { + await pb.write(msg, { signal }) + } + log('close write') await pb.unwrap().unwrap().close({ signal diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index d10b6579b2..8e3fb27584 100644 --- a/packages/protocol-identify/src/utils.ts +++ b/packages/protocol-identify/src/utils.ts @@ -2,11 +2,12 @@ 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 { isPrivate } 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.js' +import { FIRST_IDENTIFY_MESSAGE_MAX_SIZE, IDENTIFY_PROTOCOL_VERSION, MAX_IDENTIFY_MESSAGE_SIZE, MAX_PUSH_CONCURRENCY, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE } from './consts.js' +import { Identify as IdentifyMessage } from './pb/message.js' import type { IdentifyComponents, IdentifyInit } from './index.js' -import type { Identify as IdentifyMessage } from './pb/message.js' import type { 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,127 @@ export async function consumeIdentifyMessage (peerStore: PeerStore, events: Type return result } +/** + * Merge multiple received Identify messages into one + */ +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 +} + +/** + * Split an outgoing Identify message into chunks that respect the per-message size limits: + * - first message SHOULD NOT exceed 2 KB + * - subsequent messages SHOULD NOT exceed 4 KB + * + * Addresses are sorted so that publicly-reachable ones appear in the first + * message, improving backwards-compatibility with old receivers that stop + * after the first message. + */ +export function buildIdentifyMessages (msg: IdentifyMessage): IdentifyMessage[] { + // Sort: non-private (public + circuit-relay-via-public-relay) first + const sortedAddrs = [...msg.listenAddrs].sort((a, b) => { + try { + const aPublic = isPrivate(multiaddr(a)) ? 0 : 1 + const bPublic = isPrivate(multiaddr(b)) ? 0 : 1 + return bPublic - aPublic + } catch { + return 0 + } + }) + + const full: IdentifyMessage = { ...msg, listenAddrs: sortedAddrs } + if (IdentifyMessage.encode(full).length <= FIRST_IDENTIFY_MESSAGE_MAX_SIZE) { + return [full] + } + + const messages: IdentifyMessage[] = [] + const remainingAddrs = [...sortedAddrs] + const remainingProtocols = [...msg.protocols] + + // First message carries all scalar fields + signedPeerRecord, then as many + // addresses and protocols as fit within 2 KB. + const first: IdentifyMessage = { + protocolVersion: msg.protocolVersion, + agentVersion: msg.agentVersion, + publicKey: msg.publicKey, + observedAddr: msg.observedAddr, + signedPeerRecord: msg.signedPeerRecord, + listenAddrs: [], + protocols: [] + } + + while (remainingAddrs.length > 0) { + const candidate: IdentifyMessage = { ...first, listenAddrs: [...first.listenAddrs, remainingAddrs[0]] } + if (IdentifyMessage.encode(candidate).length <= FIRST_IDENTIFY_MESSAGE_MAX_SIZE) { + first.listenAddrs.push(remainingAddrs.shift()!) + } else { + break + } + } + + while (remainingProtocols.length > 0) { + const candidate: IdentifyMessage = { ...first, protocols: [...first.protocols, remainingProtocols[0]] } + if (IdentifyMessage.encode(candidate).length <= FIRST_IDENTIFY_MESSAGE_MAX_SIZE) { + first.protocols.push(remainingProtocols.shift()!) + } else { + break + } + } + + messages.push(first) + + // Subsequent messages carry the remaining addresses and protocols in ≤4 KB chunks. + while (remainingAddrs.length > 0 || remainingProtocols.length > 0) { + const subsequent: IdentifyMessage = { listenAddrs: [], protocols: [] } + + while (remainingAddrs.length > 0) { + const candidate: IdentifyMessage = { ...subsequent, listenAddrs: [...subsequent.listenAddrs, remainingAddrs[0]] } + if (IdentifyMessage.encode(candidate).length <= SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE) { + subsequent.listenAddrs.push(remainingAddrs.shift()!) + } else { + // Single address exceeds limit; send it anyway to avoid getting stuck. + subsequent.listenAddrs.push(remainingAddrs.shift()!) + break + } + } + + while (remainingProtocols.length > 0) { + const candidate: IdentifyMessage = { ...subsequent, protocols: [...subsequent.protocols, remainingProtocols[0]] } + if (IdentifyMessage.encode(candidate).length <= SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE) { + subsequent.protocols.push(remainingProtocols.shift()!) + } else { + break + } + } + + messages.push(subsequent) + } + + return messages +} + export interface AbstractIdentifyInit extends IdentifyInit { protocol: string log: Logger diff --git a/packages/protocol-identify/test/index.spec.ts b/packages/protocol-identify/test/index.spec.ts index 9eb7e4efe1..0c43432e6a 100644 --- a/packages/protocol-identify/test/index.spec.ts +++ b/packages/protocol-identify/test/index.spec.ts @@ -519,6 +519,107 @@ describe('identify', () => { expect(result.observedAddr).to.be.undefined() }) + it('should split large identify response into multiple messages', async () => { + identify = new Identify(components) + + await start(identify) + + // Many private addresses so the response exceeds 2 KB + const manyAddrs = Array.from({ length: 300 }, (_, i) => + multiaddr(`/ip4/10.0.${Math.floor(i / 256)}.${i % 256}/tcp/1234`) + ) + components.addressManager.getAddresses.returns(manyAddrs) + + // Supply a pre-sealed small signedPeerRecord so handleProtocol does not + // seal one containing all 300 addresses (which would be > 2 KB alone). + const selfRecord = await RecordEnvelope.seal(new PeerRecord({ + peerId: components.peerId, + multiaddrs: [multiaddr('/ip4/5.5.5.5/tcp/9000')] + }), components.privateKey) + + components.peerStore.get.resolves({ + id: components.peerId, + addresses: [], + protocols: ['/foo/1.0'], + metadata: new Map(), + tags: new Map(), + peerRecordEnvelope: selfRecord.marshal() + }) + + const [outgoingStream, incomingStream] = await streamPair() + const connection = stubInterface({ + remoteAddr: multiaddr('/ip4/5.5.5.5/tcp/9000') + }) + + void identify.handleProtocol(incomingStream, connection) + + // Read all messages until the stream closes + const pb = pbStream(outgoingStream).pb(IdentifyMessage) + const messages: IdentifyMessage[] = [] + + try { + for (let i = 0; i < 10; i++) { + messages.push(await pb.read()) + } + } catch { + // stream closed after last message — expected + } + + expect(messages.length).to.be.greaterThan(1, 'expected response to be split into multiple messages') + + // Scalar fields only in first message + expect(messages[0].protocolVersion).to.be.ok() + expect(messages[0].agentVersion).to.be.ok() + + // All addresses present across all messages + const allAddrs = messages.flatMap(m => m.listenAddrs) + expect(allAddrs).to.have.lengthOf(manyAddrs.length) + }) + + it('should order public addresses before private in the identify response', async () => { + identify = new Identify(components) + + await start(identify) + + const publicAddr = multiaddr('/ip4/1.2.3.4/tcp/1234') + // Many private addresses first, then the one public address at the end + const manyAddrs = [ + ...Array.from({ length: 300 }, (_, i) => + multiaddr(`/ip4/10.0.${Math.floor(i / 256)}.${i % 256}/tcp/1234`) + ), + publicAddr + ] + components.addressManager.getAddresses.returns(manyAddrs) + + // Supply a pre-sealed small signedPeerRecord so it doesn't consume the whole first message. + const selfRecord = await RecordEnvelope.seal(new PeerRecord({ + peerId: components.peerId, + multiaddrs: [multiaddr('/ip4/5.5.5.5/tcp/9000')] + }), components.privateKey) + + components.peerStore.get.resolves({ + id: components.peerId, + addresses: [], + protocols: [], + metadata: new Map(), + tags: new Map(), + peerRecordEnvelope: selfRecord.marshal() + }) + + const [outgoingStream, incomingStream] = await streamPair() + const connection = stubInterface({ + remoteAddr: multiaddr('/ip4/5.5.5.5/tcp/9000') + }) + + void identify.handleProtocol(incomingStream, connection) + + const pb = pbStream(outgoingStream).pb(IdentifyMessage) + const firstMessage = await pb.read() + + const firstMessageAddrs = firstMessage.listenAddrs.map(a => multiaddr(a).toString()) + expect(firstMessageAddrs).to.include('/ip4/1.2.3.4/tcp/1234', 'public address not in first message') + }) + it('should ignore observed non global unicast IPv6 addresses', async () => { identify = new Identify(components) diff --git a/packages/protocol-identify/test/push.spec.ts b/packages/protocol-identify/test/push.spec.ts index 920598a03c..5381aeed34 100644 --- a/packages/protocol-identify/test/push.spec.ts +++ b/packages/protocol-identify/test/push.spec.ts @@ -1,4 +1,5 @@ import { generateKeyPair, publicKeyToProtobuf } from '@libp2p/crypto/keys' +import { PeerRecord, RecordEnvelope } from '@libp2p/peer-record' import { start, stop } from '@libp2p/interface' import { defaultLogger } from '@libp2p/logger' import { peerIdFromPrivateKey } from '@libp2p/peer-id' @@ -126,8 +127,8 @@ describe('identify (push)', () => { const updatedProtocol = '/special-new-protocol/1.0.0' const updatedAddress = multiaddr('/ip4/127.0.0.1/tcp/48322') - const pb = pbStream(outgoingStream) - void pb.write({ + const pb = pbStream(outgoingStream).pb(IdentifyMessage) + await pb.write({ publicKey: publicKeyToProtobuf(remotePeer.publicKey), protocols: [ updatedProtocol @@ -135,7 +136,8 @@ describe('identify (push)', () => { listenAddrs: [ updatedAddress.bytes ] - }, IdentifyMessage) + }) + await outgoingStream.close() components.peerStore.patch.reset() @@ -150,6 +152,60 @@ describe('identify (push)', () => { expect(update.addresses?.map(({ multiaddr }) => multiaddr.toString())).deep.equals([updatedAddress.toString()]) }) + it('should handle multiple push messages and merge them', async () => { + identify = new IdentifyPush(components) + + await start(identify) + + const remotePrivateKey = await generateKeyPair('Ed25519') + const remotePeer = peerIdFromPrivateKey(remotePrivateKey) + const [outgoingStream, incomingStream] = await streamPair() + const connection = stubInterface({ + remotePeer + }) + + const addr1 = multiaddr('/ip4/127.0.0.1/tcp/1234') + const addr2 = multiaddr('/ip4/127.0.0.1/tcp/5678') + const protocol1 = '/protocol-a/1.0.0' + const protocol2 = '/protocol-b/1.0.0' + const sharedProtocol = '/shared/1.0.0' + + // Simulate a sender that splits the push across two messages + const signedPeerRecord = await RecordEnvelope.seal(new PeerRecord({ + peerId: remotePeer, + multiaddrs: [addr1] + }), remotePrivateKey) + + const pb = pbStream(outgoingStream).pb(IdentifyMessage) + await pb.write({ + publicKey: publicKeyToProtobuf(remotePeer.publicKey), + listenAddrs: [addr1.bytes], + protocols: [protocol1, sharedProtocol] + }) + await pb.write({ + listenAddrs: [addr2.bytes], + protocols: [protocol2, sharedProtocol], + signedPeerRecord: signedPeerRecord.marshal() + }) + await outgoingStream.close() + + components.peerStore.patch.reset() + + await identify.handleProtocol(incomingStream, connection) + + expect(components.peerStore.patch.callCount).to.equal(1) + + const update = components.peerStore.patch.getCall(0).args[1] + + // Addresses from both messages should be present + const addrs = update.addresses?.map(({ multiaddr: ma }: { multiaddr: { toString(): string } }) => ma.toString()) ?? [] + // signedPeerRecord was included so addresses come from that + expect(addrs).to.include(addr1.toString()) + + // signedPeerRecord from second message should be stored + expect(update.peerRecordEnvelope).to.deep.equal(signedPeerRecord.marshal()) + }) + it('should time out during push identify', async () => { identify = new IdentifyPush(components, { timeout: 10 diff --git a/packages/protocol-identify/test/utils.spec.ts b/packages/protocol-identify/test/utils.spec.ts new file mode 100644 index 0000000000..b29c5b0c92 --- /dev/null +++ b/packages/protocol-identify/test/utils.spec.ts @@ -0,0 +1,203 @@ +import { multiaddr } from '@multiformats/multiaddr' +import { expect } from 'aegir/chai' +import { FIRST_IDENTIFY_MESSAGE_MAX_SIZE, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE } from '../src/consts.js' +import { Identify as IdentifyMessage } from '../src/pb/message.js' +import { buildIdentifyMessages, mergeIdentifyMessages } from '../src/utils.js' + +// Enough private addresses to push a message well past the 2 KB threshold. +function manyPrivateAddrs (count: number): Uint8Array[] { + return Array.from({ length: count }, (_, i) => + multiaddr(`/ip4/10.0.${Math.floor(i / 256)}.${i % 256}/tcp/1234`).bytes + ) +} + +describe('buildIdentifyMessages', () => { + it('returns a single message when content fits within 2 KB', () => { + const msg: IdentifyMessage = { + protocolVersion: '1.0.0', + agentVersion: 'test/1.0', + listenAddrs: [multiaddr('/ip4/1.2.3.4/tcp/1234').bytes], + protocols: ['/foo/1.0'] + } + + const messages = buildIdentifyMessages(msg) + + expect(messages).to.have.lengthOf(1) + expect(messages[0].listenAddrs).to.have.lengthOf(1) + expect(messages[0].protocols).to.deep.equal(['/foo/1.0']) + expect(messages[0].protocolVersion).to.equal('1.0.0') + expect(messages[0].agentVersion).to.equal('test/1.0') + }) + + it('splits into multiple messages when content exceeds 2 KB', () => { + const listenAddrs = manyPrivateAddrs(300) + + const msg: IdentifyMessage = { + protocolVersion: '1.0.0', + listenAddrs, + protocols: [] + } + + const messages = buildIdentifyMessages(msg) + + expect(messages.length).to.be.greaterThan(1) + + // First message must fit within 2 KB + expect(IdentifyMessage.encode(messages[0]).length) + .to.be.lessThanOrEqual(FIRST_IDENTIFY_MESSAGE_MAX_SIZE) + + // Subsequent messages must fit within 4 KB + for (const subsequent of messages.slice(1)) { + expect(IdentifyMessage.encode(subsequent).length) + .to.be.lessThanOrEqual(SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE) + } + + // All addresses must be present across all messages + const allAddrs = messages.flatMap(m => m.listenAddrs) + expect(allAddrs).to.have.lengthOf(listenAddrs.length) + }) + + it('places public addresses before private addresses', () => { + const publicAddr = multiaddr('/ip4/1.2.3.4/tcp/1234').bytes + // Put the public address at the end of a large list of private ones + const listenAddrs = [...manyPrivateAddrs(300), publicAddr] + + const msg: IdentifyMessage = { + listenAddrs, + protocols: [] + } + + const messages = buildIdentifyMessages(msg) + + expect(messages.length).to.be.greaterThan(1, 'expected message to be split') + + // Public address must appear in the first message + const firstMessageAddrs = messages[0].listenAddrs.map(a => multiaddr(a).toString()) + expect(firstMessageAddrs).to.include('/ip4/1.2.3.4/tcp/1234', 'public address not in first message') + }) + + it('puts scalar fields only in the first message', () => { + const msg: IdentifyMessage = { + protocolVersion: '1.0.0', + agentVersion: 'test/1.0', + observedAddr: multiaddr('/ip4/5.6.7.8/tcp/9000').bytes, + listenAddrs: manyPrivateAddrs(300), + protocols: [] + } + + const messages = buildIdentifyMessages(msg) + + expect(messages.length).to.be.greaterThan(1, 'expected message to be split') + + expect(messages[0].protocolVersion).to.equal('1.0.0') + expect(messages[0].agentVersion).to.equal('test/1.0') + expect(messages[0].observedAddr).to.deep.equal(multiaddr('/ip4/5.6.7.8/tcp/9000').bytes) + + for (const subsequent of messages.slice(1)) { + expect(subsequent.protocolVersion).to.be.undefined() + expect(subsequent.agentVersion).to.be.undefined() + expect(subsequent.observedAddr).to.be.undefined() + } + }) + + it('puts signedPeerRecord in the first message', () => { + const signedPeerRecord = new Uint8Array(100).fill(1) + + const msg: IdentifyMessage = { + signedPeerRecord, + listenAddrs: manyPrivateAddrs(300), + protocols: [] + } + + const messages = buildIdentifyMessages(msg) + + expect(messages.length).to.be.greaterThan(1, 'expected message to be split') + expect(messages[0].signedPeerRecord).to.deep.equal(signedPeerRecord) + + for (const subsequent of messages.slice(1)) { + expect(subsequent.signedPeerRecord).to.be.undefined() + } + }) +}) + +describe('mergeIdentifyMessages', () => { + it('returns a single message unchanged', () => { + const msg: IdentifyMessage = { + protocolVersion: '1.0.0', + listenAddrs: [multiaddr('/ip4/1.2.3.4/tcp/1234').bytes], + protocols: ['/foo/1.0'] + } + + const merged = mergeIdentifyMessages([msg]) + + expect(merged.protocolVersion).to.equal('1.0.0') + expect(merged.listenAddrs).to.have.lengthOf(1) + expect(merged.protocols).to.deep.equal(['/foo/1.0']) + }) + + it('later scalar fields override earlier ones', () => { + const first: IdentifyMessage = { + listenAddrs: [], + protocols: [], + protocolVersion: 'old-proto', + agentVersion: 'old-agent' + } + const second: IdentifyMessage = { + listenAddrs: [], + protocols: [], + protocolVersion: 'new-proto', + agentVersion: 'new-agent' + } + + const merged = mergeIdentifyMessages([first, second]) + + expect(merged.protocolVersion).to.equal('new-proto') + expect(merged.agentVersion).to.equal('new-agent') + }) + + it('appends listenAddrs from subsequent messages', () => { + const addr1 = multiaddr('/ip4/1.2.3.4/tcp/1234').bytes + const addr2 = multiaddr('/ip4/5.6.7.8/tcp/5678').bytes + const first: IdentifyMessage = { listenAddrs: [addr1], protocols: [] } + const second: IdentifyMessage = { listenAddrs: [addr2], protocols: [] } + + const merged = mergeIdentifyMessages([first, second]) + + expect(merged.listenAddrs).to.have.lengthOf(2) + }) + + it('deduplicates protocols across messages', () => { + const first: IdentifyMessage = { listenAddrs: [], protocols: ['/foo/1.0', '/bar/1.0'] } + const second: IdentifyMessage = { listenAddrs: [], protocols: ['/bar/1.0', '/baz/1.0'] } + + const merged = mergeIdentifyMessages([first, second]) + + expect(merged.protocols).to.deep.equal(['/foo/1.0', '/bar/1.0', '/baz/1.0']) + }) + + it('later signedPeerRecord overrides earlier', () => { + const record1 = new Uint8Array(10).fill(1) + const record2 = new Uint8Array(10).fill(2) + const first: IdentifyMessage = { listenAddrs: [], protocols: [], signedPeerRecord: record1 } + const second: IdentifyMessage = { listenAddrs: [], protocols: [], signedPeerRecord: record2 } + + const merged = mergeIdentifyMessages([first, second]) + + expect(merged.signedPeerRecord).to.deep.equal(record2) + }) + + it('missing scalar fields in later messages do not clear earlier values', () => { + const first: IdentifyMessage = { + listenAddrs: [], + protocols: [], + protocolVersion: '1.0.0', + agentVersion: 'agent/1.0' + } + const second: IdentifyMessage = { listenAddrs: [], protocols: [] } + + const merged = mergeIdentifyMessages([first, second]) + + expect(merged.protocolVersion).to.equal('1.0.0') + expect(merged.agentVersion).to.equal('agent/1.0') + }) +}) From f67bef88d4f719240fd6bbba2bbc0e7133b6d840 Mon Sep 17 00:00:00 2001 From: William Morriss Date: Wed, 15 Apr 2026 15:08:17 -0500 Subject: [PATCH 06/21] fix(identify): defer oversized signedPeerRecord to ensure first message has addresses Assisted-by: Claude:claude-sonnet-4-6 --- packages/protocol-identify/src/utils.ts | 91 ++++++++++++------- packages/protocol-identify/test/utils.spec.ts | 71 +++++++++++++++ 2 files changed, 128 insertions(+), 34 deletions(-) diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index 8e3fb27584..006f6db860 100644 --- a/packages/protocol-identify/src/utils.ts +++ b/packages/protocol-identify/src/utils.ts @@ -201,6 +201,32 @@ export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMes return merged } +/** + * Greedily pack items from `remaining` into `current` as long as the encoded + * message produced by `buildCandidate` stays within `maxSize`. + * + * When `guaranteeFirst` is true the size check is skipped for the very first + * item so that the caller always makes progress (used when `current` must end + * up non-empty regardless of size). + */ +function packItems ( + current: T[], + remaining: T[], + maxSize: number, + buildCandidate: (items: T[]) => IdentifyMessage, + guaranteeFirst = false +): void { + while (remaining.length > 0) { + if (!guaranteeFirst || current.length > 0) { + const candidate = buildCandidate([...current, remaining[0]]) + if (IdentifyMessage.encode(candidate).length > maxSize) { + break + } + } + current.push(remaining.shift()!) + } +} + /** * Split an outgoing Identify message into chunks that respect the per-message size limits: * - first message SHOULD NOT exceed 2 KB @@ -243,49 +269,46 @@ export function buildIdentifyMessages (msg: IdentifyMessage): IdentifyMessage[] protocols: [] } - while (remainingAddrs.length > 0) { - const candidate: IdentifyMessage = { ...first, listenAddrs: [...first.listenAddrs, remainingAddrs[0]] } - if (IdentifyMessage.encode(candidate).length <= FIRST_IDENTIFY_MESSAGE_MAX_SIZE) { - first.listenAddrs.push(remainingAddrs.shift()!) - } else { - break - } - } + packItems(first.listenAddrs, remainingAddrs, FIRST_IDENTIFY_MESSAGE_MAX_SIZE, + items => ({ ...first, listenAddrs: items })) - while (remainingProtocols.length > 0) { - const candidate: IdentifyMessage = { ...first, protocols: [...first.protocols, remainingProtocols[0]] } - if (IdentifyMessage.encode(candidate).length <= FIRST_IDENTIFY_MESSAGE_MAX_SIZE) { - first.protocols.push(remainingProtocols.shift()!) - } else { - break - } + // If signedPeerRecord is so large that no address fits alongside it, defer it + // to its own standalone message so the first message can carry addresses instead. + // The deferred record may exceed 4 KB but cannot be subdivided. + let deferredSignedPeerRecord: Uint8Array | undefined + if (first.listenAddrs.length === 0 && remainingAddrs.length > 0 && first.signedPeerRecord != null) { + deferredSignedPeerRecord = first.signedPeerRecord + first.signedPeerRecord = undefined + + // Re-pack without signedPeerRecord, guaranteeing at least one address. + packItems(first.listenAddrs, remainingAddrs, FIRST_IDENTIFY_MESSAGE_MAX_SIZE, + items => ({ ...first, listenAddrs: items }), true) } + packItems(first.protocols, remainingProtocols, FIRST_IDENTIFY_MESSAGE_MAX_SIZE, + items => ({ ...first, protocols: items })) + messages.push(first) + if (deferredSignedPeerRecord != null) { + const spr: IdentifyMessage = { listenAddrs: [], protocols: [], signedPeerRecord: deferredSignedPeerRecord } + + packItems(spr.listenAddrs, remainingAddrs, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE, + items => ({ ...spr, listenAddrs: items })) + packItems(spr.protocols, remainingProtocols, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE, + items => ({ ...spr, protocols: items })) + + messages.push(spr) + } + // Subsequent messages carry the remaining addresses and protocols in ≤4 KB chunks. while (remainingAddrs.length > 0 || remainingProtocols.length > 0) { const subsequent: IdentifyMessage = { listenAddrs: [], protocols: [] } - while (remainingAddrs.length > 0) { - const candidate: IdentifyMessage = { ...subsequent, listenAddrs: [...subsequent.listenAddrs, remainingAddrs[0]] } - if (IdentifyMessage.encode(candidate).length <= SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE) { - subsequent.listenAddrs.push(remainingAddrs.shift()!) - } else { - // Single address exceeds limit; send it anyway to avoid getting stuck. - subsequent.listenAddrs.push(remainingAddrs.shift()!) - break - } - } - - while (remainingProtocols.length > 0) { - const candidate: IdentifyMessage = { ...subsequent, protocols: [...subsequent.protocols, remainingProtocols[0]] } - if (IdentifyMessage.encode(candidate).length <= SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE) { - subsequent.protocols.push(remainingProtocols.shift()!) - } else { - break - } - } + packItems(subsequent.listenAddrs, remainingAddrs, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE, + items => ({ ...subsequent, listenAddrs: items }), true) + packItems(subsequent.protocols, remainingProtocols, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE, + items => ({ ...subsequent, protocols: items })) messages.push(subsequent) } diff --git a/packages/protocol-identify/test/utils.spec.ts b/packages/protocol-identify/test/utils.spec.ts index b29c5b0c92..d1cd1489a1 100644 --- a/packages/protocol-identify/test/utils.spec.ts +++ b/packages/protocol-identify/test/utils.spec.ts @@ -57,6 +57,77 @@ describe('buildIdentifyMessages', () => { expect(allAddrs).to.have.lengthOf(listenAddrs.length) }) + it('spans three or more messages when content exceeds 6 KB', () => { + // First message holds ≤2 KB, each subsequent holds ≤4 KB. + // 700 addresses × ~10 bytes each ≈ 7 KB — requires at least three messages. + const listenAddrs = manyPrivateAddrs(700) + + const msg: IdentifyMessage = { + protocolVersion: '1.0.0', + agentVersion: 'test/1.0', + listenAddrs, + protocols: ['/foo/1.0', '/bar/1.0'] + } + + const messages = buildIdentifyMessages(msg) + + expect(messages.length).to.be.greaterThan(2, 'expected at least three messages') + + expect(IdentifyMessage.encode(messages[0]).length) + .to.be.lessThanOrEqual(FIRST_IDENTIFY_MESSAGE_MAX_SIZE) + + for (const subsequent of messages.slice(1)) { + expect(IdentifyMessage.encode(subsequent).length) + .to.be.lessThanOrEqual(SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE) + } + + // Every address must appear in exactly one message + const allAddrs = messages.flatMap(m => m.listenAddrs) + expect(allAddrs).to.have.lengthOf(listenAddrs.length) + + // Scalar fields only in first message + expect(messages[0].protocolVersion).to.equal('1.0.0') + for (const subsequent of messages.slice(1)) { + expect(subsequent.protocolVersion).to.be.undefined() + } + }) + + it('returns a single message when signedPeerRecord and addresses together fit within 2 KB', () => { + const msg: IdentifyMessage = { + protocolVersion: '1.0.0', + signedPeerRecord: new Uint8Array(100).fill(1), + listenAddrs: [multiaddr('/ip4/1.2.3.4/tcp/1234').bytes], + protocols: ['/foo/1.0'] + } + + const messages = buildIdentifyMessages(msg) + + expect(messages).to.have.lengthOf(1) + expect(messages[0].signedPeerRecord).to.deep.equal(msg.signedPeerRecord) + expect(messages[0].listenAddrs).to.have.lengthOf(1) + }) + + it('includes at least one address in the first message even when signedPeerRecord is large', () => { + // A signedPeerRecord large enough that scalar fields alone exceed 2 KB, + // which would prevent any addresses from fitting without special handling. + const largeSignedPeerRecord = new Uint8Array(2048).fill(1) + + const msg: IdentifyMessage = { + protocolVersion: '1.0.0', + agentVersion: 'test/1.0', + signedPeerRecord: largeSignedPeerRecord, + listenAddrs: [multiaddr('/ip4/1.2.3.4/tcp/1234').bytes, ...manyPrivateAddrs(10)], + protocols: ['/foo/1.0'] + } + + const messages = buildIdentifyMessages(msg) + + expect(messages[0].listenAddrs.length).to.be.greaterThan(0, 'first message must contain at least one address') + + const allAddrs = messages.flatMap(m => m.listenAddrs) + expect(allAddrs).to.have.lengthOf(msg.listenAddrs.length) + }) + it('places public addresses before private addresses', () => { const publicAddr = multiaddr('/ip4/1.2.3.4/tcp/1234').bytes // Put the public address at the end of a large list of private ones From 45fa22f329274c4bbb2cc7393ad67dfe681813d9 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sat, 9 May 2026 19:26:57 +0700 Subject: [PATCH 07/21] refactor(identify): drop multi-message send Reverts buildIdentifyMessages, packItems, and the deferred-SPR path so identify and identify-push send a single message. Receive-side multi-message support is retained. Multi-message send is deferred to a follow-up so older js-libp2p receivers without the close-error fix in this PR don't regress on identify when this ships. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/src/consts.ts | 8 +- .../protocol-identify/src/identify-push.ts | 10 +- packages/protocol-identify/src/identify.ts | 10 +- packages/protocol-identify/src/utils.ts | 109 +--------- packages/protocol-identify/test/index.spec.ts | 57 ------ packages/protocol-identify/test/utils.spec.ts | 190 +----------------- 6 files changed, 15 insertions(+), 369 deletions(-) diff --git a/packages/protocol-identify/src/consts.ts b/packages/protocol-identify/src/consts.ts index 3613bb14e2..7b9014f772 100644 --- a/packages/protocol-identify/src/consts.ts +++ b/packages/protocol-identify/src/consts.ts @@ -11,9 +11,11 @@ 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 -// Large identify messages are split into smaller messages -export const FIRST_IDENTIFY_MESSAGE_MAX_SIZE = 1024 * 2 -export const SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE = 1024 * 4 +// 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 diff --git a/packages/protocol-identify/src/identify-push.ts b/packages/protocol-identify/src/identify-push.ts index e750dc3a16..5d35e716a3 100644 --- a/packages/protocol-identify/src/identify-push.ts +++ b/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, buildIdentifyMessages, consumeIdentifyMessage, defaultValues, mergeIdentifyMessages } from './utils.ts' +import { AbstractIdentify, consumeIdentifyMessage, defaultValues, mergeIdentifyMessages } 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' @@ -99,18 +99,16 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif maxDataLength: self.maxMessageSize }).pb(IdentifyMessage) - const msgs = buildIdentifyMessages({ + await pb.write({ listenAddrs: listenAddresses.map(ma => ma.bytes), signedPeerRecord: signedPeerRecord.marshal(), protocols: supportedProtocols, agentVersion, protocolVersion + }, { + signal }) - for (const msg of msgs) { - await pb.write(msg, { signal }) - } - await stream.close({ signal }) diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index b7b9deb793..33d4ef68ad 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/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, buildIdentifyMessages, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, mergeIdentifyMessages } from './utils.ts' +import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, mergeIdentifyMessages } 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' @@ -198,7 +198,7 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt const pb = pbStream(stream).pb(IdentifyMessage) log('send response') - const msgs = buildIdentifyMessages({ + await pb.write({ protocolVersion: this.host.protocolVersion, agentVersion: this.host.agentVersion, publicKey: publicKeyToProtobuf(this.components.privateKey.publicKey), @@ -206,12 +206,10 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt signedPeerRecord, observedAddr, protocols: peerData.protocols + }, { + signal }) - for (const msg of msgs) { - await pb.write(msg, { signal }) - } - log('close write') await pb.unwrap().unwrap().close({ signal diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index b665321c37..c754edd326 100644 --- a/packages/protocol-identify/src/utils.ts +++ b/packages/protocol-identify/src/utils.ts @@ -2,10 +2,9 @@ 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 { defaultMultiaddrSorter } from '@libp2p/utils' import { multiaddr } from '@multiformats/multiaddr' import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string' -import { FIRST_IDENTIFY_MESSAGE_MAX_SIZE, IDENTIFY_PROTOCOL_VERSION, MAX_IDENTIFY_MESSAGE_SIZE, MAX_PUSH_CONCURRENCY, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE } from './consts.ts' +import { IDENTIFY_PROTOCOL_VERSION, MAX_IDENTIFY_MESSAGE_SIZE, MAX_PUSH_CONCURRENCY } from './consts.ts' import { Identify as IdentifyMessage } from './pb/message.ts' import type { IdentifyComponents, IdentifyInit } from './index.ts' import type { Libp2pEvents, IdentifyResult, SignedPeerRecord, Logger, Connection, Peer, PeerData, PeerStore, Startable, Stream } from '@libp2p/interface' @@ -201,112 +200,6 @@ export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMes return merged } -/** - * Greedily pack items from `remaining` into `current` as long as the encoded - * message produced by `buildCandidate` stays within `maxSize`. - * - * When `guaranteeFirst` is true the size check is skipped for the very first - * item so that the caller always makes progress (used when `current` must end - * up non-empty regardless of size). - */ -function packItems ( - current: T[], - remaining: T[], - maxSize: number, - buildCandidate: (items: T[]) => IdentifyMessage, - guaranteeFirst = false -): void { - while (remaining.length > 0) { - if (!guaranteeFirst || current.length > 0) { - const candidate = buildCandidate([...current, remaining[0]]) - if (IdentifyMessage.encode(candidate).length > maxSize) { - break - } - } - current.push(remaining.shift()!) - } -} - -/** - * Split an outgoing Identify message into chunks that respect the per-message size limits: - * - first message SHOULD NOT exceed 2 KB - * - subsequent messages SHOULD NOT exceed 4 KB - * - * Addresses are sorted so that publicly-reachable ones appear in the first - * message, improving backwards-compatibility with old receivers that stop - * after the first message. - */ -export function buildIdentifyMessages (msg: IdentifyMessage): IdentifyMessage[] { - const sortedAddrs = defaultMultiaddrSorter(msg.listenAddrs.map(a => multiaddr(a))).map((ma: Multiaddr) => ma.bytes) - - const full: IdentifyMessage = { ...msg, listenAddrs: sortedAddrs } - if (IdentifyMessage.encode(full).length <= FIRST_IDENTIFY_MESSAGE_MAX_SIZE) { - return [full] - } - - const messages: IdentifyMessage[] = [] - const remainingAddrs = [...sortedAddrs] - const remainingProtocols = [...msg.protocols] - - // First message carries all scalar fields + signedPeerRecord, then as many - // addresses and protocols as fit within 2 KB. - const first: IdentifyMessage = { - protocolVersion: msg.protocolVersion, - agentVersion: msg.agentVersion, - publicKey: msg.publicKey, - observedAddr: msg.observedAddr, - signedPeerRecord: msg.signedPeerRecord, - listenAddrs: [], - protocols: [] - } - - packItems(first.listenAddrs, remainingAddrs, FIRST_IDENTIFY_MESSAGE_MAX_SIZE, - items => ({ ...first, listenAddrs: items })) - - // If signedPeerRecord is so large that no address fits alongside it, defer it - // to its own standalone message so the first message can carry addresses instead. - // The deferred record may exceed 4 KB but cannot be subdivided. - let deferredSignedPeerRecord: Uint8Array | undefined - if (first.listenAddrs.length === 0 && remainingAddrs.length > 0 && first.signedPeerRecord != null) { - deferredSignedPeerRecord = first.signedPeerRecord - first.signedPeerRecord = undefined - - // Re-pack without signedPeerRecord, guaranteeing at least one address. - packItems(first.listenAddrs, remainingAddrs, FIRST_IDENTIFY_MESSAGE_MAX_SIZE, - items => ({ ...first, listenAddrs: items }), true) - } - - packItems(first.protocols, remainingProtocols, FIRST_IDENTIFY_MESSAGE_MAX_SIZE, - items => ({ ...first, protocols: items })) - - messages.push(first) - - if (deferredSignedPeerRecord != null) { - const spr: IdentifyMessage = { listenAddrs: [], protocols: [], signedPeerRecord: deferredSignedPeerRecord } - - packItems(spr.listenAddrs, remainingAddrs, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE, - items => ({ ...spr, listenAddrs: items })) - packItems(spr.protocols, remainingProtocols, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE, - items => ({ ...spr, protocols: items })) - - messages.push(spr) - } - - // Subsequent messages carry the remaining addresses and protocols in ≤4 KB chunks. - while (remainingAddrs.length > 0 || remainingProtocols.length > 0) { - const subsequent: IdentifyMessage = { listenAddrs: [], protocols: [] } - - packItems(subsequent.listenAddrs, remainingAddrs, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE, - items => ({ ...subsequent, listenAddrs: items }), true) - packItems(subsequent.protocols, remainingProtocols, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE, - items => ({ ...subsequent, protocols: items })) - - messages.push(subsequent) - } - - return messages -} - export interface AbstractIdentifyInit extends IdentifyInit { protocol: string log: Logger diff --git a/packages/protocol-identify/test/index.spec.ts b/packages/protocol-identify/test/index.spec.ts index 0c43432e6a..d3514e2b21 100644 --- a/packages/protocol-identify/test/index.spec.ts +++ b/packages/protocol-identify/test/index.spec.ts @@ -519,63 +519,6 @@ describe('identify', () => { expect(result.observedAddr).to.be.undefined() }) - it('should split large identify response into multiple messages', async () => { - identify = new Identify(components) - - await start(identify) - - // Many private addresses so the response exceeds 2 KB - const manyAddrs = Array.from({ length: 300 }, (_, i) => - multiaddr(`/ip4/10.0.${Math.floor(i / 256)}.${i % 256}/tcp/1234`) - ) - components.addressManager.getAddresses.returns(manyAddrs) - - // Supply a pre-sealed small signedPeerRecord so handleProtocol does not - // seal one containing all 300 addresses (which would be > 2 KB alone). - const selfRecord = await RecordEnvelope.seal(new PeerRecord({ - peerId: components.peerId, - multiaddrs: [multiaddr('/ip4/5.5.5.5/tcp/9000')] - }), components.privateKey) - - components.peerStore.get.resolves({ - id: components.peerId, - addresses: [], - protocols: ['/foo/1.0'], - metadata: new Map(), - tags: new Map(), - peerRecordEnvelope: selfRecord.marshal() - }) - - const [outgoingStream, incomingStream] = await streamPair() - const connection = stubInterface({ - remoteAddr: multiaddr('/ip4/5.5.5.5/tcp/9000') - }) - - void identify.handleProtocol(incomingStream, connection) - - // Read all messages until the stream closes - const pb = pbStream(outgoingStream).pb(IdentifyMessage) - const messages: IdentifyMessage[] = [] - - try { - for (let i = 0; i < 10; i++) { - messages.push(await pb.read()) - } - } catch { - // stream closed after last message — expected - } - - expect(messages.length).to.be.greaterThan(1, 'expected response to be split into multiple messages') - - // Scalar fields only in first message - expect(messages[0].protocolVersion).to.be.ok() - expect(messages[0].agentVersion).to.be.ok() - - // All addresses present across all messages - const allAddrs = messages.flatMap(m => m.listenAddrs) - expect(allAddrs).to.have.lengthOf(manyAddrs.length) - }) - it('should order public addresses before private in the identify response', async () => { identify = new Identify(components) diff --git a/packages/protocol-identify/test/utils.spec.ts b/packages/protocol-identify/test/utils.spec.ts index d1cd1489a1..a724e5c931 100644 --- a/packages/protocol-identify/test/utils.spec.ts +++ b/packages/protocol-identify/test/utils.spec.ts @@ -1,195 +1,7 @@ import { multiaddr } from '@multiformats/multiaddr' import { expect } from 'aegir/chai' -import { FIRST_IDENTIFY_MESSAGE_MAX_SIZE, SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE } from '../src/consts.js' import { Identify as IdentifyMessage } from '../src/pb/message.js' -import { buildIdentifyMessages, mergeIdentifyMessages } from '../src/utils.js' - -// Enough private addresses to push a message well past the 2 KB threshold. -function manyPrivateAddrs (count: number): Uint8Array[] { - return Array.from({ length: count }, (_, i) => - multiaddr(`/ip4/10.0.${Math.floor(i / 256)}.${i % 256}/tcp/1234`).bytes - ) -} - -describe('buildIdentifyMessages', () => { - it('returns a single message when content fits within 2 KB', () => { - const msg: IdentifyMessage = { - protocolVersion: '1.0.0', - agentVersion: 'test/1.0', - listenAddrs: [multiaddr('/ip4/1.2.3.4/tcp/1234').bytes], - protocols: ['/foo/1.0'] - } - - const messages = buildIdentifyMessages(msg) - - expect(messages).to.have.lengthOf(1) - expect(messages[0].listenAddrs).to.have.lengthOf(1) - expect(messages[0].protocols).to.deep.equal(['/foo/1.0']) - expect(messages[0].protocolVersion).to.equal('1.0.0') - expect(messages[0].agentVersion).to.equal('test/1.0') - }) - - it('splits into multiple messages when content exceeds 2 KB', () => { - const listenAddrs = manyPrivateAddrs(300) - - const msg: IdentifyMessage = { - protocolVersion: '1.0.0', - listenAddrs, - protocols: [] - } - - const messages = buildIdentifyMessages(msg) - - expect(messages.length).to.be.greaterThan(1) - - // First message must fit within 2 KB - expect(IdentifyMessage.encode(messages[0]).length) - .to.be.lessThanOrEqual(FIRST_IDENTIFY_MESSAGE_MAX_SIZE) - - // Subsequent messages must fit within 4 KB - for (const subsequent of messages.slice(1)) { - expect(IdentifyMessage.encode(subsequent).length) - .to.be.lessThanOrEqual(SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE) - } - - // All addresses must be present across all messages - const allAddrs = messages.flatMap(m => m.listenAddrs) - expect(allAddrs).to.have.lengthOf(listenAddrs.length) - }) - - it('spans three or more messages when content exceeds 6 KB', () => { - // First message holds ≤2 KB, each subsequent holds ≤4 KB. - // 700 addresses × ~10 bytes each ≈ 7 KB — requires at least three messages. - const listenAddrs = manyPrivateAddrs(700) - - const msg: IdentifyMessage = { - protocolVersion: '1.0.0', - agentVersion: 'test/1.0', - listenAddrs, - protocols: ['/foo/1.0', '/bar/1.0'] - } - - const messages = buildIdentifyMessages(msg) - - expect(messages.length).to.be.greaterThan(2, 'expected at least three messages') - - expect(IdentifyMessage.encode(messages[0]).length) - .to.be.lessThanOrEqual(FIRST_IDENTIFY_MESSAGE_MAX_SIZE) - - for (const subsequent of messages.slice(1)) { - expect(IdentifyMessage.encode(subsequent).length) - .to.be.lessThanOrEqual(SUBSEQUENT_IDENTIFY_MESSAGE_MAX_SIZE) - } - - // Every address must appear in exactly one message - const allAddrs = messages.flatMap(m => m.listenAddrs) - expect(allAddrs).to.have.lengthOf(listenAddrs.length) - - // Scalar fields only in first message - expect(messages[0].protocolVersion).to.equal('1.0.0') - for (const subsequent of messages.slice(1)) { - expect(subsequent.protocolVersion).to.be.undefined() - } - }) - - it('returns a single message when signedPeerRecord and addresses together fit within 2 KB', () => { - const msg: IdentifyMessage = { - protocolVersion: '1.0.0', - signedPeerRecord: new Uint8Array(100).fill(1), - listenAddrs: [multiaddr('/ip4/1.2.3.4/tcp/1234').bytes], - protocols: ['/foo/1.0'] - } - - const messages = buildIdentifyMessages(msg) - - expect(messages).to.have.lengthOf(1) - expect(messages[0].signedPeerRecord).to.deep.equal(msg.signedPeerRecord) - expect(messages[0].listenAddrs).to.have.lengthOf(1) - }) - - it('includes at least one address in the first message even when signedPeerRecord is large', () => { - // A signedPeerRecord large enough that scalar fields alone exceed 2 KB, - // which would prevent any addresses from fitting without special handling. - const largeSignedPeerRecord = new Uint8Array(2048).fill(1) - - const msg: IdentifyMessage = { - protocolVersion: '1.0.0', - agentVersion: 'test/1.0', - signedPeerRecord: largeSignedPeerRecord, - listenAddrs: [multiaddr('/ip4/1.2.3.4/tcp/1234').bytes, ...manyPrivateAddrs(10)], - protocols: ['/foo/1.0'] - } - - const messages = buildIdentifyMessages(msg) - - expect(messages[0].listenAddrs.length).to.be.greaterThan(0, 'first message must contain at least one address') - - const allAddrs = messages.flatMap(m => m.listenAddrs) - expect(allAddrs).to.have.lengthOf(msg.listenAddrs.length) - }) - - it('places public addresses before private addresses', () => { - const publicAddr = multiaddr('/ip4/1.2.3.4/tcp/1234').bytes - // Put the public address at the end of a large list of private ones - const listenAddrs = [...manyPrivateAddrs(300), publicAddr] - - const msg: IdentifyMessage = { - listenAddrs, - protocols: [] - } - - const messages = buildIdentifyMessages(msg) - - expect(messages.length).to.be.greaterThan(1, 'expected message to be split') - - // Public address must appear in the first message - const firstMessageAddrs = messages[0].listenAddrs.map(a => multiaddr(a).toString()) - expect(firstMessageAddrs).to.include('/ip4/1.2.3.4/tcp/1234', 'public address not in first message') - }) - - it('puts scalar fields only in the first message', () => { - const msg: IdentifyMessage = { - protocolVersion: '1.0.0', - agentVersion: 'test/1.0', - observedAddr: multiaddr('/ip4/5.6.7.8/tcp/9000').bytes, - listenAddrs: manyPrivateAddrs(300), - protocols: [] - } - - const messages = buildIdentifyMessages(msg) - - expect(messages.length).to.be.greaterThan(1, 'expected message to be split') - - expect(messages[0].protocolVersion).to.equal('1.0.0') - expect(messages[0].agentVersion).to.equal('test/1.0') - expect(messages[0].observedAddr).to.deep.equal(multiaddr('/ip4/5.6.7.8/tcp/9000').bytes) - - for (const subsequent of messages.slice(1)) { - expect(subsequent.protocolVersion).to.be.undefined() - expect(subsequent.agentVersion).to.be.undefined() - expect(subsequent.observedAddr).to.be.undefined() - } - }) - - it('puts signedPeerRecord in the first message', () => { - const signedPeerRecord = new Uint8Array(100).fill(1) - - const msg: IdentifyMessage = { - signedPeerRecord, - listenAddrs: manyPrivateAddrs(300), - protocols: [] - } - - const messages = buildIdentifyMessages(msg) - - expect(messages.length).to.be.greaterThan(1, 'expected message to be split') - expect(messages[0].signedPeerRecord).to.deep.equal(signedPeerRecord) - - for (const subsequent of messages.slice(1)) { - expect(subsequent.signedPeerRecord).to.be.undefined() - } - }) -}) +import { mergeIdentifyMessages } from '../src/utils.js' describe('mergeIdentifyMessages', () => { it('returns a single message unchanged', () => { From b63b851d5d19797abd7904f46b4e9b60016c08fe Mon Sep 17 00:00:00 2001 From: tabcat Date: Sat, 9 May 2026 19:38:33 +0700 Subject: [PATCH 08/21] fix(identify): symmetric dedup for listenAddrs in mergeIdentifyMessages Aligns listenAddrs merge with the existing protocols dedup. Duplicates across multiple identify messages would otherwise inflate the peerStore entry. Invisible in practice today (no senders duplicate) but worth locking in to prevent future drift. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/src/utils.ts | 24 +++++++++++++++++-- packages/protocol-identify/test/utils.spec.ts | 21 ++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index c754edd326..cd15f87d5b 100644 --- a/packages/protocol-identify/src/utils.ts +++ b/packages/protocol-identify/src/utils.ts @@ -4,6 +4,7 @@ import { peerIdFromCID, peerIdFromPublicKey } from '@libp2p/peer-id' import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record' import { multiaddr } from '@multiformats/multiaddr' import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string' +import { toString as uint8ArrayToString } from 'uint8arrays/to-string' import { IDENTIFY_PROTOCOL_VERSION, MAX_IDENTIFY_MESSAGE_SIZE, MAX_PUSH_CONCURRENCY } from './consts.ts' import { Identify as IdentifyMessage } from './pb/message.ts' import type { IdentifyComponents, IdentifyInit } from './index.ts' @@ -175,7 +176,11 @@ export async function consumeIdentifyMessage (peerStore: PeerStore, events: Type * Merge multiple received Identify messages into one */ export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMessage { - const merged: IdentifyMessage = { ...messages[0] } + const merged: IdentifyMessage = { + ...messages[0], + listenAddrs: dedupBytes(messages[0].listenAddrs), + protocols: [...new Set(messages[0].protocols)] + } for (const msg of messages.slice(1)) { if (msg.protocolVersion != null) { @@ -193,13 +198,28 @@ export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMes if (msg.signedPeerRecord != null) { merged.signedPeerRecord = msg.signedPeerRecord } - merged.listenAddrs = [...merged.listenAddrs, ...msg.listenAddrs] + merged.listenAddrs = dedupBytes([...merged.listenAddrs, ...msg.listenAddrs]) merged.protocols = [...new Set([...merged.protocols, ...msg.protocols])] } return merged } +/** + * Deduplicate a list of byte arrays by their content. Set cannot be used + * directly because Uint8Array equality is reference-based. + */ +function dedupBytes (bytes: Uint8Array[]): Uint8Array[] { + const seen = new Map() + for (const buf of bytes) { + const key = uint8ArrayToString(buf, 'base64') + if (!seen.has(key)) { + seen.set(key, buf) + } + } + return [...seen.values()] +} + export interface AbstractIdentifyInit extends IdentifyInit { protocol: string log: Logger diff --git a/packages/protocol-identify/test/utils.spec.ts b/packages/protocol-identify/test/utils.spec.ts index a724e5c931..0638c5559f 100644 --- a/packages/protocol-identify/test/utils.spec.ts +++ b/packages/protocol-identify/test/utils.spec.ts @@ -1,3 +1,4 @@ +import { Buffer } from 'node:buffer' import { multiaddr } from '@multiformats/multiaddr' import { expect } from 'aegir/chai' import { Identify as IdentifyMessage } from '../src/pb/message.js' @@ -69,6 +70,26 @@ describe('mergeIdentifyMessages', () => { expect(merged.signedPeerRecord).to.deep.equal(record2) }) + it('deduplicates listenAddrs across messages', () => { + const addr1 = multiaddr('/ip4/1.2.3.4/tcp/4001').bytes + const addr2 = multiaddr('/ip4/5.6.7.8/tcp/4001').bytes + + const merged = mergeIdentifyMessages([ + { + listenAddrs: [addr1, addr2], + protocols: [] + }, + { + listenAddrs: [addr1, addr1, addr2], + protocols: [] + } + ]) + + expect(merged.listenAddrs).to.have.lengthOf(2) + const hex = merged.listenAddrs.map(b => Buffer.from(b).toString('hex')) + expect(new Set(hex).size).to.equal(2) + }) + it('missing scalar fields in later messages do not clear earlier values', () => { const first: IdentifyMessage = { listenAddrs: [], From f277ea207aac146bcc119cfd5110d182911537cd Mon Sep 17 00:00:00 2001 From: tabcat Date: Sat, 9 May 2026 20:45:04 +0700 Subject: [PATCH 09/21] feat(identify): add isEofLike helper for receive-side EOF discrimination Centralises the check used by both _identify and identify-push handlers to decide whether a read error after at least one successful message means "remote finished" vs "real failure". Treats UnexpectedEOFError and "remote write side no longer writable" as equivalent EOF signals so transport-level resets after success aren't dropped as failures. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/src/utils.ts | 21 ++++++++++++++++ packages/protocol-identify/test/utils.spec.ts | 24 ++++++++++++++++++- 2 files changed, 44 insertions(+), 1 deletion(-) diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index cd15f87d5b..173e8d8a91 100644 --- a/packages/protocol-identify/src/utils.ts +++ b/packages/protocol-identify/src/utils.ts @@ -2,6 +2,7 @@ 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 { UnexpectedEOFError } from '@libp2p/utils' import { multiaddr } from '@multiformats/multiaddr' import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string' import { toString as uint8ArrayToString } from 'uint8arrays/to-string' @@ -220,6 +221,26 @@ function dedupBytes (bytes: Uint8Array[]): Uint8Array[] { return [...seen.values()] } +/** + * Returns true if the error from a `pb.read()` call should be treated as + * "remote finished sending" — either a clean EOF (UnexpectedEOFError) or any + * other error encountered when the remote write side is no longer writable. + * + * Used to distinguish "peer finished" from "real error mid-stream". When the + * caller already has at least one successful message, an EOF-like signal + * means the multi-message read loop should break and merge what it has; + * otherwise the error propagates. + */ +export function isEofLike (err: unknown, stream: Stream): boolean { + if (err instanceof UnexpectedEOFError) { + return true + } + if (stream.remoteWriteStatus !== 'writable') { + return true + } + return false +} + export interface AbstractIdentifyInit extends IdentifyInit { protocol: string log: Logger diff --git a/packages/protocol-identify/test/utils.spec.ts b/packages/protocol-identify/test/utils.spec.ts index 0638c5559f..2aada71125 100644 --- a/packages/protocol-identify/test/utils.spec.ts +++ b/packages/protocol-identify/test/utils.spec.ts @@ -1,8 +1,10 @@ import { Buffer } from 'node:buffer' +import { UnexpectedEOFError } from '@libp2p/utils' import { multiaddr } from '@multiformats/multiaddr' import { expect } from 'aegir/chai' import { Identify as IdentifyMessage } from '../src/pb/message.js' -import { mergeIdentifyMessages } from '../src/utils.js' +import { isEofLike, mergeIdentifyMessages } from '../src/utils.js' +import type { Stream } from '@libp2p/interface' describe('mergeIdentifyMessages', () => { it('returns a single message unchanged', () => { @@ -105,3 +107,23 @@ describe('mergeIdentifyMessages', () => { expect(merged.agentVersion).to.equal('agent/1.0') }) }) + +describe('isEofLike', () => { + it('returns true for UnexpectedEOFError', () => { + const err = new UnexpectedEOFError('eof') + const stream = { remoteWriteStatus: 'writable' } as any as Stream + expect(isEofLike(err, stream)).to.be.true() + }) + + it('returns true when remote write side is no longer writable', () => { + const err = new Error('reset') + const stream = { remoteWriteStatus: 'closed' } as any as Stream + expect(isEofLike(err, stream)).to.be.true() + }) + + it('returns false for non-eof errors when stream is still writable', () => { + const err = new Error('parse error') + const stream = { remoteWriteStatus: 'writable' } as any as Stream + expect(isEofLike(err, stream)).to.be.false() + }) +}) From b997840aace3d1a3551a52d131390aa82c7a84f9 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sat, 9 May 2026 21:36:36 +0700 Subject: [PATCH 10/21] fix(identify): harden _identify receive loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit C2: wrap post-read close() in try/catch with trace log. The original bug in this area was that StreamStateError from byteStream.unwrap() pushing buffered trailing bytes back to a closed-read stream propagated through the outer catch and aborted the stream — discarding the successfully-read message. Swallowing the close error preserves the read result, which is what the caller actually wanted. I1: import MAX_IDENTIFY_MESSAGES from consts instead of shadowing it locally with a literal. I2: log when the cap is reached without seeing EOF, so silent truncation surfaces in operator logs. I4: replace UnexpectedEOFError-only break with isEofLike, which also treats "remote write side no longer writable" as EOF after at least one successful message — catches stream-reset-after-success as a success path. Co-Authored-By: William Morriss Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/src/identify.ts | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index 33d4ef68ad..7a9c4cc428 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/packages/protocol-identify/src/identify.ts @@ -2,16 +2,17 @@ import { publicKeyFromProtobuf, publicKeyToProtobuf } from '@libp2p/crypto/keys' import { InvalidMessageError, serviceCapabilities } from '@libp2p/interface' import { peerIdFromCID } from '@libp2p/peer-id' import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record' -import { UnexpectedEOFError, isGlobalUnicast, isPrivate, pbStream } from '@libp2p/utils' +import { isGlobalUnicast, isPrivate, pbStream } from '@libp2p/utils' import { CODE_IP6, CODE_IP6ZONE, CODE_P2P } from '@multiformats/multiaddr' import { IP_OR_DOMAIN, TCP } from '@multiformats/multiaddr-matcher' import { setMaxListeners } from 'main-event' import { + MAX_IDENTIFY_MESSAGES, MULTICODEC_IDENTIFY_PROTOCOL_NAME, MULTICODEC_IDENTIFY_PROTOCOL_VERSION } from './consts.ts' import { Identify as IdentifyMessage } from './pb/message.ts' -import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, mergeIdentifyMessages } from './utils.ts' +import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, isEofLike, mergeIdentifyMessages } 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' @@ -64,16 +65,15 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt maxDataLength: this.maxMessageSize }).pb(IdentifyMessage) - // Large responses can be subdivided. - // Read all messages until the stream closes. - const MAX_IDENTIFY_MESSAGES = 10 + // Large responses can be subdivided per spec PR libp2p/specs#709. + // Read up to MAX_IDENTIFY_MESSAGES until the stream closes. const messages: IdentifyMessage[] = [] for (let i = 0; i < MAX_IDENTIFY_MESSAGES; i++) { try { messages.push(await pb.read(options)) } catch (err) { - if (messages.length > 0 && err instanceof UnexpectedEOFError) { + if (messages.length > 0 && isEofLike(err, stream)) { break } @@ -85,7 +85,15 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt throw new InvalidMessageError('No identify message received') } - await pb.unwrap().unwrap().close(options) + if (messages.length >= MAX_IDENTIFY_MESSAGES) { + log?.('reached MAX_IDENTIFY_MESSAGES (%d) without EOF, returning truncated identify', MAX_IDENTIFY_MESSAGES) + } + + try { + await pb.unwrap().unwrap().close(options) + } catch (err) { + log?.trace('error closing identify stream after read - %e', err) + } return mergeIdentifyMessages(messages) } catch (err: any) { From 6409344341990f560a4906acbdf9c8d2bc59ea29 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sat, 9 May 2026 23:31:13 +0700 Subject: [PATCH 11/21] fix(identify-push): harden receive loop with same C2/I1/I2/I4 fixes Mirrors the changes to identify._identify so the push receive handler also: imports MAX_IDENTIFY_MESSAGES from consts, uses isEofLike for the break condition, logs at the cap-reached boundary, and swallows close errors with a trace log. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../protocol-identify/src/identify-push.ts | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/packages/protocol-identify/src/identify-push.ts b/packages/protocol-identify/src/identify-push.ts index 5d35e716a3..10a5d5b33f 100644 --- a/packages/protocol-identify/src/identify-push.ts +++ b/packages/protocol-identify/src/identify-push.ts @@ -1,6 +1,6 @@ import { InvalidMessageError, serviceCapabilities } from '@libp2p/interface' import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record' -import { UnexpectedEOFError, debounce, pbStream } from '@libp2p/utils' +import { debounce, pbStream } from '@libp2p/utils' import { CODE_P2P } from '@multiformats/multiaddr' import drain from 'it-drain' import parallel from 'it-parallel' @@ -8,12 +8,13 @@ import { setMaxListeners } from 'main-event' import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string' import { toString as uint8ArrayToString } from 'uint8arrays/to-string' import { + MAX_IDENTIFY_MESSAGES, MULTICODEC_IDENTIFY_PUSH_PROTOCOL_NAME, MULTICODEC_IDENTIFY_PUSH_PROTOCOL_VERSION, PUSH_DEBOUNCE_MS } from './consts.ts' import { Identify as IdentifyMessage } from './pb/message.ts' -import { AbstractIdentify, consumeIdentifyMessage, defaultValues, mergeIdentifyMessages } from './utils.ts' +import { AbstractIdentify, consumeIdentifyMessage, defaultValues, isEofLike, mergeIdentifyMessages } 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' @@ -150,11 +151,11 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif const messages: IdentifyMessage[] = [] - for (let i = 0; i < 10; i++) { + for (let i = 0; i < MAX_IDENTIFY_MESSAGES; i++) { try { messages.push(await pb.read(options)) } catch (err) { - if (messages.length > 0 && err instanceof UnexpectedEOFError) { + if (messages.length > 0 && isEofLike(err, stream)) { break } @@ -166,7 +167,15 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif throw new InvalidMessageError('No identify message received') } - await stream.close(options) + if (messages.length >= MAX_IDENTIFY_MESSAGES) { + log('reached MAX_IDENTIFY_MESSAGES (%d) without EOF, returning truncated identify push', MAX_IDENTIFY_MESSAGES) + } + + try { + await stream.close(options) + } catch (err) { + log.trace('error closing identify-push stream after read - %e', err) + } await consumeIdentifyMessage(this.components.peerStore, this.components.events, log, connection, mergeIdentifyMessages(messages)) From 2313cfc6c04f0f0befe00ee8875c672337cd2864 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sat, 9 May 2026 23:56:47 +0700 Subject: [PATCH 12/21] test(identify): receive-side regression tests; drop obsolete sort test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds T1 (empty stream rejection), T2-cap (cap boundary), T3 (non-EOF mid-stream), T-reset (reset-after-success returns merged result), and T-close-swallow (C2 regression: close error after read does not discard the message). Equivalent push-receive tests added. Removes "should order public addresses before private in the identify response" — the assertion no longer guards real behavior since the in-protocol address sorter was removed in the multi-send revert. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/test/index.spec.ts | 216 ++++++++++++++---- packages/protocol-identify/test/push.spec.ts | 64 ++++++ 2 files changed, 236 insertions(+), 44 deletions(-) diff --git a/packages/protocol-identify/test/index.spec.ts b/packages/protocol-identify/test/index.spec.ts index d3514e2b21..c926f86cb0 100644 --- a/packages/protocol-identify/test/index.spec.ts +++ b/packages/protocol-identify/test/index.spec.ts @@ -267,6 +267,178 @@ describe('identify', () => { .to.have.property('peerRecordEnvelope').that.equalBytes(peerRecordEnvelope) }) + it('should reject when remote closes the stream without sending any message', async () => { + identify = new Identify(components) + await start(identify) + + const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519')) + const [outgoingStream] = await streamPair() + const connection = stubInterface({ remotePeer }) + connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) + + // Do not push any bytes - just close the remote write side. + outgoingStream.remoteWriteStatus = 'closed' + + // The receive loop only treats an EOF as "remote finished" once at least + // one message has been read; with zero messages, the EOF propagates as + // UnexpectedEOFError so the caller sees a hard failure for an empty stream. + await expect(identify.identify(connection)) + .to.eventually.be.rejected() + .with.property('name', 'UnexpectedEOFError') + }) + + it('should return merged identify when peer sends MAX_IDENTIFY_MESSAGES + 1 messages', async () => { + identify = new Identify(components) + await start(identify) + + const remotePrivateKey = await generateKeyPair('Ed25519') + const remotePeer = peerIdFromPrivateKey(remotePrivateKey) + + // First message carries publicKey so identify validates. + // Subsequent 10 messages each carry one extra protocol. + // Total 11 messages - reader should truncate to first 10. + const messages: IdentifyMessage[] = [{ + listenAddrs: [], + protocols: ['/test/0/1.0.0'], + publicKey: publicKeyToProtobuf(remotePeer.publicKey) + }] + for (let i = 1; i < 11; i++) { + messages.push({ listenAddrs: [], protocols: [`/test/${i}/1.0.0`] }) + } + + const [outgoingStream] = await streamPair() + const connection = stubInterface({ remotePeer }) + connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) + + const identifyPromise = identify.identify(connection) + await new Promise(resolve => setTimeout(resolve, 0)) + + const encoded = messages.map(m => lp.encode.single(IdentifyMessage.encode(m)).subarray()) + const totalLen = encoded.reduce((n, e) => n + e.byteLength, 0) + const combined = new Uint8Array(totalLen) + let off = 0 + for (const e of encoded) { combined.set(e, off); off += e.byteLength } + + outgoingStream.push(combined) + outgoingStream.remoteWriteStatus = 'closed' + + const result = await identifyPromise + // Reader merged at most MAX_IDENTIFY_MESSAGES (10) - the 11th protocol is truncated. + expect(result.protocols).to.have.lengthOf.at.most(10) + expect(result.protocols).to.not.include('/test/10/1.0.0') + }) + + it('should reject when peer sends invalid bytes mid-stream', async () => { + identify = new Identify(components) + await start(identify) + + const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519')) + const firstMessage: IdentifyMessage = { + listenAddrs: [], + protocols: ['/foo/bar/1.0'], + publicKey: publicKeyToProtobuf(remotePeer.publicKey) + } + + const [outgoingStream] = await streamPair() + const connection = stubInterface({ remotePeer }) + connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) + + const identifyPromise = identify.identify(connection, { + signal: AbortSignal.timeout(500) + }) + await new Promise(resolve => setTimeout(resolve, 0)) + + const encoded1 = lp.encode.single(IdentifyMessage.encode(firstMessage)).subarray() + // Append a varint length prefix that promises far more bytes than provided. + // The pb.read() in the second iteration will see the prefix, try to read, + // hit EOF early without remoteWriteStatus changing, and throw. + // With remoteWriteStatus still 'writable', isEofLike returns false and the + // error propagates. + const garbage = new Uint8Array([0xff, 0xff, 0xff, 0xff, 0x07]) + const combined = new Uint8Array(encoded1.byteLength + garbage.byteLength) + combined.set(encoded1) + combined.set(garbage, encoded1.byteLength) + + outgoingStream.push(combined) + // Note: do NOT set remoteWriteStatus to 'closed' - keep it 'writable'. + + await expect(identifyPromise).to.eventually.be.rejected() + }) + + it('should return the message when remote resets the stream after a successful write', async () => { + identify = new Identify(components) + await start(identify) + + const remotePrivateKey = await generateKeyPair('Ed25519') + const remotePeer = peerIdFromPrivateKey(remotePrivateKey) + + 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({ remotePeer }) + connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) + + const identifyPromise = identify.identify(connection) + await new Promise(resolve => setTimeout(resolve, 0)) + + const encoded = lp.encode.single(IdentifyMessage.encode(message)).subarray() + outgoingStream.push(encoded) + // Set status to a non-'writable' value to simulate stream-reset-after-success. + // This documents the I4 widening: any non-'writable' status post-success + // is treated as EOF. + outgoingStream.remoteWriteStatus = 'closed' + + const result = await identifyPromise + expect(result.protocols).to.include('/foo/bar/1.0') + }) + + it('should succeed even if close() throws after a successful read (C2 regression)', async () => { + identify = new Identify(components) + await start(identify) + + const remotePrivateKey = await generateKeyPair('Ed25519') + const remotePeer = peerIdFromPrivateKey(remotePrivateKey) + + 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({ remotePeer }) + connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) + + // Force close() to throw with a StreamStateError-shaped error after a + // successful read. This simulates the original bug: byteStream.unwrap() + // throwing because the read buffer holds trailing bytes that can't be + // pushed back to a closed-read stream. + const originalClose = outgoingStream.close.bind(outgoingStream) + outgoingStream.close = async (...args: any[]) => { + // Let the underlying close run, then throw afterwards. + await originalClose(...args) + throw Object.assign(new Error('simulated close-after-read failure'), { name: 'StreamStateError' }) + } + + const identifyPromise = identify.identify(connection) + await new Promise(resolve => setTimeout(resolve, 0)) + + const encoded = lp.encode.single(IdentifyMessage.encode(message)).subarray() + outgoingStream.push(encoded) + outgoingStream.remoteWriteStatus = 'closed' + + const result = await identifyPromise + expect(result.protocols).to.include('/foo/bar/1.0') + }) + it('should limit incoming identify message sizes', async () => { const maxMessageSize = 100 @@ -519,50 +691,6 @@ describe('identify', () => { expect(result.observedAddr).to.be.undefined() }) - it('should order public addresses before private in the identify response', async () => { - identify = new Identify(components) - - await start(identify) - - const publicAddr = multiaddr('/ip4/1.2.3.4/tcp/1234') - // Many private addresses first, then the one public address at the end - const manyAddrs = [ - ...Array.from({ length: 300 }, (_, i) => - multiaddr(`/ip4/10.0.${Math.floor(i / 256)}.${i % 256}/tcp/1234`) - ), - publicAddr - ] - components.addressManager.getAddresses.returns(manyAddrs) - - // Supply a pre-sealed small signedPeerRecord so it doesn't consume the whole first message. - const selfRecord = await RecordEnvelope.seal(new PeerRecord({ - peerId: components.peerId, - multiaddrs: [multiaddr('/ip4/5.5.5.5/tcp/9000')] - }), components.privateKey) - - components.peerStore.get.resolves({ - id: components.peerId, - addresses: [], - protocols: [], - metadata: new Map(), - tags: new Map(), - peerRecordEnvelope: selfRecord.marshal() - }) - - const [outgoingStream, incomingStream] = await streamPair() - const connection = stubInterface({ - remoteAddr: multiaddr('/ip4/5.5.5.5/tcp/9000') - }) - - void identify.handleProtocol(incomingStream, connection) - - const pb = pbStream(outgoingStream).pb(IdentifyMessage) - const firstMessage = await pb.read() - - const firstMessageAddrs = firstMessage.listenAddrs.map(a => multiaddr(a).toString()) - expect(firstMessageAddrs).to.include('/ip4/1.2.3.4/tcp/1234', 'public address not in first message') - }) - it('should ignore observed non global unicast IPv6 addresses', async () => { identify = new Identify(components) diff --git a/packages/protocol-identify/test/push.spec.ts b/packages/protocol-identify/test/push.spec.ts index 5381aeed34..ee06c7d607 100644 --- a/packages/protocol-identify/test/push.spec.ts +++ b/packages/protocol-identify/test/push.spec.ts @@ -206,6 +206,70 @@ describe('identify (push)', () => { expect(update.peerRecordEnvelope).to.deep.equal(signedPeerRecord.marshal()) }) + it('should reject incoming push when remote closes the stream without sending any message', async () => { + identify = new IdentifyPush(components) + await start(identify) + + const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519')) + const [outgoingStream, incomingStream] = await streamPair() + const connection = stubInterface({ remotePeer }) + + // Close the writer side without sending any bytes. The push handler reads + // from the incoming stream; with no messages received the receive loop's + // EOF propagates as UnexpectedEOFError (the InvalidMessageError fallback + // is only reachable after a successful loop completion with zero messages). + outgoingStream.remoteWriteStatus = 'closed' + incomingStream.remoteWriteStatus = 'closed' + void outgoingStream.close() + + components.peerStore.patch.reset() + + await expect(identify.handleProtocol(incomingStream, connection)) + .to.eventually.be.rejected() + .with.property('name', 'UnexpectedEOFError') + + expect(components.peerStore.patch.callCount).to.equal(0) + }) + + it('should succeed even if close() throws after a successful read (C2 regression)', async () => { + identify = new IdentifyPush(components) + await start(identify) + + const remotePrivateKey = await generateKeyPair('Ed25519') + const remotePeer = peerIdFromPrivateKey(remotePrivateKey) + const [outgoingStream, incomingStream] = await streamPair() + const connection = stubInterface({ remotePeer }) + + const updatedProtocol = '/special-new-protocol/1.0.0' + const updatedAddress = multiaddr('/ip4/127.0.0.1/tcp/48322') + + const pb = pbStream(outgoingStream).pb(IdentifyMessage) + await pb.write({ + publicKey: publicKeyToProtobuf(remotePeer.publicKey), + protocols: [updatedProtocol], + listenAddrs: [updatedAddress.bytes] + }) + await outgoingStream.close() + + // Force close() on the incoming (server-side) stream to throw after the + // message has been read. The push handler awaits stream.close(options) and + // wraps it in try/catch; the read result must still be consumed. + const originalClose = incomingStream.close.bind(incomingStream) + incomingStream.close = async (...args: any[]) => { + await originalClose(...args) + throw Object.assign(new Error('simulated close-after-read failure'), { name: 'StreamStateError' }) + } + + components.peerStore.patch.reset() + + await identify.handleProtocol(incomingStream, connection) + + // patch should still happen — close error must be swallowed. + expect(components.peerStore.patch.callCount).to.equal(1) + const update = components.peerStore.patch.getCall(0).args[1] + expect(update.protocols).to.include(updatedProtocol) + }) + it('should time out during push identify', async () => { identify = new IdentifyPush(components, { timeout: 10 From 252311090f89d98bfe8c82427ba0773c2f1d4850 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sun, 10 May 2026 00:10:21 +0700 Subject: [PATCH 13/21] chore(identify): fix pre-existing import order in push.spec.ts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move @libp2p/peer-record after @libp2p/peer-id to satisfy the existing import/order eslint rule. Pre-existing — was originally introduced in aa154ebcb but not flagged until lint ran on this branch. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/test/push.spec.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/protocol-identify/test/push.spec.ts b/packages/protocol-identify/test/push.spec.ts index ee06c7d607..ad66b8515f 100644 --- a/packages/protocol-identify/test/push.spec.ts +++ b/packages/protocol-identify/test/push.spec.ts @@ -1,8 +1,8 @@ import { generateKeyPair, publicKeyToProtobuf } from '@libp2p/crypto/keys' -import { PeerRecord, RecordEnvelope } from '@libp2p/peer-record' import { start, stop } from '@libp2p/interface' import { defaultLogger } from '@libp2p/logger' import { peerIdFromPrivateKey } from '@libp2p/peer-id' +import { PeerRecord, RecordEnvelope } from '@libp2p/peer-record' import { streamPair, pbStream } from '@libp2p/utils' import { multiaddr } from '@multiformats/multiaddr' import { expect } from 'aegir/chai' From 66258c24a10e46b816bed240db1972c6bac2a596 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sun, 10 May 2026 00:21:48 +0700 Subject: [PATCH 14/21] refactor(identify): drop unreachable empty-message branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The post-loop `if (messages.length === 0) throw new InvalidMessageError(...)` in both _identify and the push handleProtocol was unreachable: the read-loop catch only breaks on EOF when messages.length > 0, otherwise it re-throws the underlying UnexpectedEOFError. The empty-stream case is therefore handled by the propagating EOF, never by the post-loop check. Removes the dead block in both files. identify-push no longer needs the InvalidMessageError import. Behavior is unchanged — empty streams continue to reject with UnexpectedEOFError, matching the existing test assertions. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/src/identify-push.ts | 6 +----- packages/protocol-identify/src/identify.ts | 4 ---- 2 files changed, 1 insertion(+), 9 deletions(-) diff --git a/packages/protocol-identify/src/identify-push.ts b/packages/protocol-identify/src/identify-push.ts index 10a5d5b33f..fbaddcd54c 100644 --- a/packages/protocol-identify/src/identify-push.ts +++ b/packages/protocol-identify/src/identify-push.ts @@ -1,4 +1,4 @@ -import { InvalidMessageError, serviceCapabilities } from '@libp2p/interface' +import { serviceCapabilities } from '@libp2p/interface' import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record' import { debounce, pbStream } from '@libp2p/utils' import { CODE_P2P } from '@multiformats/multiaddr' @@ -163,10 +163,6 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif } } - if (messages.length === 0) { - throw new InvalidMessageError('No identify message received') - } - if (messages.length >= MAX_IDENTIFY_MESSAGES) { log('reached MAX_IDENTIFY_MESSAGES (%d) without EOF, returning truncated identify push', MAX_IDENTIFY_MESSAGES) } diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index 7a9c4cc428..591775bd1f 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/packages/protocol-identify/src/identify.ts @@ -81,10 +81,6 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt } } - if (messages.length === 0) { - throw new InvalidMessageError('No identify message received') - } - if (messages.length >= MAX_IDENTIFY_MESSAGES) { log?.('reached MAX_IDENTIFY_MESSAGES (%d) without EOF, returning truncated identify', MAX_IDENTIFY_MESSAGES) } From 14b6519ec4eb595d4be0611eacec0fe9d21fb256 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sun, 10 May 2026 02:32:02 +0700 Subject: [PATCH 15/21] revert(identify): drop dedupBytes for listenAddrs PeerStore handles dedup of incoming listenAddrs downstream, so deduping in the merge step was redundant. Restores the prior plain-concat behavior for listenAddrs while keeping the Set-based dedup on protocols (string ids). Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/src/utils.ts | 28 ++++--------------- packages/protocol-identify/test/utils.spec.ts | 21 -------------- 2 files changed, 5 insertions(+), 44 deletions(-) diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index 173e8d8a91..cecd38a3d9 100644 --- a/packages/protocol-identify/src/utils.ts +++ b/packages/protocol-identify/src/utils.ts @@ -5,7 +5,6 @@ import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record' import { UnexpectedEOFError } from '@libp2p/utils' import { multiaddr } from '@multiformats/multiaddr' import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string' -import { toString as uint8ArrayToString } from 'uint8arrays/to-string' import { IDENTIFY_PROTOCOL_VERSION, MAX_IDENTIFY_MESSAGE_SIZE, MAX_PUSH_CONCURRENCY } from './consts.ts' import { Identify as IdentifyMessage } from './pb/message.ts' import type { IdentifyComponents, IdentifyInit } from './index.ts' @@ -174,14 +173,12 @@ export async function consumeIdentifyMessage (peerStore: PeerStore, events: Type } /** - * Merge multiple received Identify messages into one + * Merge multiple received Identify messages into one. Repeated `listenAddrs` + * are concatenated as-is — peerstore handles dedup downstream. `protocols` are + * deduplicated via Set since they're string identifiers. */ export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMessage { - const merged: IdentifyMessage = { - ...messages[0], - listenAddrs: dedupBytes(messages[0].listenAddrs), - protocols: [...new Set(messages[0].protocols)] - } + const merged: IdentifyMessage = { ...messages[0] } for (const msg of messages.slice(1)) { if (msg.protocolVersion != null) { @@ -199,28 +196,13 @@ export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMes if (msg.signedPeerRecord != null) { merged.signedPeerRecord = msg.signedPeerRecord } - merged.listenAddrs = dedupBytes([...merged.listenAddrs, ...msg.listenAddrs]) + merged.listenAddrs = [...merged.listenAddrs, ...msg.listenAddrs] merged.protocols = [...new Set([...merged.protocols, ...msg.protocols])] } return merged } -/** - * Deduplicate a list of byte arrays by their content. Set cannot be used - * directly because Uint8Array equality is reference-based. - */ -function dedupBytes (bytes: Uint8Array[]): Uint8Array[] { - const seen = new Map() - for (const buf of bytes) { - const key = uint8ArrayToString(buf, 'base64') - if (!seen.has(key)) { - seen.set(key, buf) - } - } - return [...seen.values()] -} - /** * Returns true if the error from a `pb.read()` call should be treated as * "remote finished sending" — either a clean EOF (UnexpectedEOFError) or any diff --git a/packages/protocol-identify/test/utils.spec.ts b/packages/protocol-identify/test/utils.spec.ts index 2aada71125..03d19be461 100644 --- a/packages/protocol-identify/test/utils.spec.ts +++ b/packages/protocol-identify/test/utils.spec.ts @@ -1,4 +1,3 @@ -import { Buffer } from 'node:buffer' import { UnexpectedEOFError } from '@libp2p/utils' import { multiaddr } from '@multiformats/multiaddr' import { expect } from 'aegir/chai' @@ -72,26 +71,6 @@ describe('mergeIdentifyMessages', () => { expect(merged.signedPeerRecord).to.deep.equal(record2) }) - it('deduplicates listenAddrs across messages', () => { - const addr1 = multiaddr('/ip4/1.2.3.4/tcp/4001').bytes - const addr2 = multiaddr('/ip4/5.6.7.8/tcp/4001').bytes - - const merged = mergeIdentifyMessages([ - { - listenAddrs: [addr1, addr2], - protocols: [] - }, - { - listenAddrs: [addr1, addr1, addr2], - protocols: [] - } - ]) - - expect(merged.listenAddrs).to.have.lengthOf(2) - const hex = merged.listenAddrs.map(b => Buffer.from(b).toString('hex')) - expect(new Set(hex).size).to.equal(2) - }) - it('missing scalar fields in later messages do not clear earlier values', () => { const first: IdentifyMessage = { listenAddrs: [], From 40b523f7dab5cb8e337daecf207f9c9e7e7e1789 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sun, 10 May 2026 03:13:50 +0700 Subject: [PATCH 16/21] refactor(identify): inline EOF check, switch to stream.close, abort on close error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drops the isEofLike helper (and its standalone unit tests) — the check is inlined at both call sites, using err.name === 'UnexpectedEOFError' (the codebase's bundle-safe error-discrimination idiom) plus the stream-state fallback for transport-level resets. In identify._identify, the close call moves from pb.unwrap().unwrap().close to stream.close — same effect, simpler chain, matches identify-push. If close throws, abort the stream so transport-level cleanup runs even though the read result is preserved. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../protocol-identify/src/identify-push.ts | 12 ++++++---- packages/protocol-identify/src/identify.ts | 17 ++++++------- packages/protocol-identify/src/utils.ts | 21 ---------------- packages/protocol-identify/test/utils.spec.ts | 24 +------------------ 4 files changed, 17 insertions(+), 57 deletions(-) diff --git a/packages/protocol-identify/src/identify-push.ts b/packages/protocol-identify/src/identify-push.ts index fbaddcd54c..f96fd645bf 100644 --- a/packages/protocol-identify/src/identify-push.ts +++ b/packages/protocol-identify/src/identify-push.ts @@ -14,7 +14,7 @@ import { PUSH_DEBOUNCE_MS } from './consts.ts' import { Identify as IdentifyMessage } from './pb/message.ts' -import { AbstractIdentify, consumeIdentifyMessage, defaultValues, isEofLike, mergeIdentifyMessages } from './utils.ts' +import { AbstractIdentify, consumeIdentifyMessage, defaultValues, mergeIdentifyMessages } 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' @@ -154,8 +154,9 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif for (let i = 0; i < MAX_IDENTIFY_MESSAGES; i++) { try { messages.push(await pb.read(options)) - } catch (err) { - if (messages.length > 0 && isEofLike(err, stream)) { + } catch (err: any) { + // remote finished or stream torn down — keep what we have + if (messages.length > 0 && (err?.name === 'UnexpectedEOFError' || stream.remoteWriteStatus !== 'writable')) { break } @@ -164,13 +165,14 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif } if (messages.length >= MAX_IDENTIFY_MESSAGES) { - log('reached MAX_IDENTIFY_MESSAGES (%d) without EOF, returning truncated identify push', MAX_IDENTIFY_MESSAGES) + log('reached MAX_IDENTIFY_MESSAGES, returning truncated identify push') } try { await stream.close(options) - } catch (err) { + } catch (err: any) { log.trace('error closing identify-push stream after read - %e', err) + stream.abort(err) } await consumeIdentifyMessage(this.components.peerStore, this.components.events, log, connection, mergeIdentifyMessages(messages)) diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index 591775bd1f..6977dbc91a 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/packages/protocol-identify/src/identify.ts @@ -12,7 +12,7 @@ import { MULTICODEC_IDENTIFY_PROTOCOL_VERSION } from './consts.ts' import { Identify as IdentifyMessage } from './pb/message.ts' -import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, isEofLike, mergeIdentifyMessages } from './utils.ts' +import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, mergeIdentifyMessages } 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' @@ -65,15 +65,15 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt maxDataLength: this.maxMessageSize }).pb(IdentifyMessage) - // Large responses can be subdivided per spec PR libp2p/specs#709. - // Read up to MAX_IDENTIFY_MESSAGES until the stream closes. + // Read up to MAX_IDENTIFY_MESSAGES (per libp2p/specs#709). const messages: IdentifyMessage[] = [] for (let i = 0; i < MAX_IDENTIFY_MESSAGES; i++) { try { messages.push(await pb.read(options)) - } catch (err) { - if (messages.length > 0 && isEofLike(err, stream)) { + } catch (err: any) { + // remote finished or stream torn down — keep what we have + if (messages.length > 0 && (err?.name === 'UnexpectedEOFError' || stream.remoteWriteStatus !== 'writable')) { break } @@ -82,13 +82,14 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt } if (messages.length >= MAX_IDENTIFY_MESSAGES) { - log?.('reached MAX_IDENTIFY_MESSAGES (%d) without EOF, returning truncated identify', MAX_IDENTIFY_MESSAGES) + log?.('reached MAX_IDENTIFY_MESSAGES, returning truncated identify') } try { - await pb.unwrap().unwrap().close(options) - } catch (err) { + await stream.close(options) + } catch (err: any) { log?.trace('error closing identify stream after read - %e', err) + stream.abort(err) } return mergeIdentifyMessages(messages) diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index cecd38a3d9..62beed6389 100644 --- a/packages/protocol-identify/src/utils.ts +++ b/packages/protocol-identify/src/utils.ts @@ -2,7 +2,6 @@ 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 { UnexpectedEOFError } 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' @@ -203,26 +202,6 @@ export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMes return merged } -/** - * Returns true if the error from a `pb.read()` call should be treated as - * "remote finished sending" — either a clean EOF (UnexpectedEOFError) or any - * other error encountered when the remote write side is no longer writable. - * - * Used to distinguish "peer finished" from "real error mid-stream". When the - * caller already has at least one successful message, an EOF-like signal - * means the multi-message read loop should break and merge what it has; - * otherwise the error propagates. - */ -export function isEofLike (err: unknown, stream: Stream): boolean { - if (err instanceof UnexpectedEOFError) { - return true - } - if (stream.remoteWriteStatus !== 'writable') { - return true - } - return false -} - export interface AbstractIdentifyInit extends IdentifyInit { protocol: string log: Logger diff --git a/packages/protocol-identify/test/utils.spec.ts b/packages/protocol-identify/test/utils.spec.ts index 03d19be461..a724e5c931 100644 --- a/packages/protocol-identify/test/utils.spec.ts +++ b/packages/protocol-identify/test/utils.spec.ts @@ -1,9 +1,7 @@ -import { UnexpectedEOFError } from '@libp2p/utils' import { multiaddr } from '@multiformats/multiaddr' import { expect } from 'aegir/chai' import { Identify as IdentifyMessage } from '../src/pb/message.js' -import { isEofLike, mergeIdentifyMessages } from '../src/utils.js' -import type { Stream } from '@libp2p/interface' +import { mergeIdentifyMessages } from '../src/utils.js' describe('mergeIdentifyMessages', () => { it('returns a single message unchanged', () => { @@ -86,23 +84,3 @@ describe('mergeIdentifyMessages', () => { expect(merged.agentVersion).to.equal('agent/1.0') }) }) - -describe('isEofLike', () => { - it('returns true for UnexpectedEOFError', () => { - const err = new UnexpectedEOFError('eof') - const stream = { remoteWriteStatus: 'writable' } as any as Stream - expect(isEofLike(err, stream)).to.be.true() - }) - - it('returns true when remote write side is no longer writable', () => { - const err = new Error('reset') - const stream = { remoteWriteStatus: 'closed' } as any as Stream - expect(isEofLike(err, stream)).to.be.true() - }) - - it('returns false for non-eof errors when stream is still writable', () => { - const err = new Error('parse error') - const stream = { remoteWriteStatus: 'writable' } as any as Stream - expect(isEofLike(err, stream)).to.be.false() - }) -}) From b3eb27addfb3b8d0a99600d00e7e1e74e14d7d85 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sun, 10 May 2026 05:19:39 +0700 Subject: [PATCH 17/21] refactor(identify): simplify receive-loop catch to lenient MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Any error after at least one successful read keeps the messages we got and breaks the loop. The error name and stream-state checks come out — the data we've accumulated is already-validated parsed envelopes regardless of why the next read failed, and applying it is consistent with the spec's "missing fields ignored" merge semantics. T3 updates from "rejects on garbage" to "returns the message before garbage" to match the new behavior. The trailing comment referencing the removed isEofLike helper is also cleaned up. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/src/identify-push.ts | 9 ++++----- packages/protocol-identify/src/identify.ts | 9 ++++----- packages/protocol-identify/test/index.spec.ts | 13 +++++-------- 3 files changed, 13 insertions(+), 18 deletions(-) diff --git a/packages/protocol-identify/src/identify-push.ts b/packages/protocol-identify/src/identify-push.ts index f96fd645bf..a399fad66a 100644 --- a/packages/protocol-identify/src/identify-push.ts +++ b/packages/protocol-identify/src/identify-push.ts @@ -155,12 +155,11 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif try { messages.push(await pb.read(options)) } catch (err: any) { - // remote finished or stream torn down — keep what we have - if (messages.length > 0 && (err?.name === 'UnexpectedEOFError' || stream.remoteWriteStatus !== 'writable')) { - break + if (messages.length === 0) { + throw err } - - throw err + log.trace('stopped reading identify push - %e', err) + break } } diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index 6977dbc91a..d3f116409e 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/packages/protocol-identify/src/identify.ts @@ -72,12 +72,11 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt try { messages.push(await pb.read(options)) } catch (err: any) { - // remote finished or stream torn down — keep what we have - if (messages.length > 0 && (err?.name === 'UnexpectedEOFError' || stream.remoteWriteStatus !== 'writable')) { - break + if (messages.length === 0) { + throw err } - - throw err + log?.trace('stopped reading identify - %e', err) + break } } diff --git a/packages/protocol-identify/test/index.spec.ts b/packages/protocol-identify/test/index.spec.ts index c926f86cb0..acac94c259 100644 --- a/packages/protocol-identify/test/index.spec.ts +++ b/packages/protocol-identify/test/index.spec.ts @@ -328,7 +328,7 @@ describe('identify', () => { expect(result.protocols).to.not.include('/test/10/1.0.0') }) - it('should reject when peer sends invalid bytes mid-stream', async () => { + it('should return successfully-read messages when peer sends garbage bytes mid-stream', async () => { identify = new Identify(components) await start(identify) @@ -349,20 +349,17 @@ describe('identify', () => { await new Promise(resolve => setTimeout(resolve, 0)) const encoded1 = lp.encode.single(IdentifyMessage.encode(firstMessage)).subarray() - // Append a varint length prefix that promises far more bytes than provided. - // The pb.read() in the second iteration will see the prefix, try to read, - // hit EOF early without remoteWriteStatus changing, and throw. - // With remoteWriteStatus still 'writable', isEofLike returns false and the - // error propagates. + // Append a varint length prefix that promises more bytes than provided so + // pb.read on iteration 2 will block until the abort signal fires. const garbage = new Uint8Array([0xff, 0xff, 0xff, 0xff, 0x07]) const combined = new Uint8Array(encoded1.byteLength + garbage.byteLength) combined.set(encoded1) combined.set(garbage, encoded1.byteLength) outgoingStream.push(combined) - // Note: do NOT set remoteWriteStatus to 'closed' - keep it 'writable'. - await expect(identifyPromise).to.eventually.be.rejected() + const result = await identifyPromise + expect(result.protocols).to.include('/foo/bar/1.0') }) it('should return the message when remote resets the stream after a successful write', async () => { From 41ed8da6036ade0b1cb0778d649a0ea63bcb409f Mon Sep 17 00:00:00 2001 From: tabcat Date: Sun, 10 May 2026 05:25:32 +0700 Subject: [PATCH 18/21] refactor(identify): extract readIdentifyMessages helper Both identify._identify and identify-push.handleProtocol had near-identical multi-message read + close + abort logic. Extracts the common shape to utils.ts so the two call sites collapse to a single helper call. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../protocol-identify/src/identify-push.ts | 32 +----------- packages/protocol-identify/src/identify.ts | 33 +----------- packages/protocol-identify/src/utils.ts | 50 ++++++++++++++++++- 3 files changed, 52 insertions(+), 63 deletions(-) diff --git a/packages/protocol-identify/src/identify-push.ts b/packages/protocol-identify/src/identify-push.ts index a399fad66a..10b6c039c0 100644 --- a/packages/protocol-identify/src/identify-push.ts +++ b/packages/protocol-identify/src/identify-push.ts @@ -8,13 +8,12 @@ import { setMaxListeners } from 'main-event' import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string' import { toString as uint8ArrayToString } from 'uint8arrays/to-string' import { - MAX_IDENTIFY_MESSAGES, MULTICODEC_IDENTIFY_PUSH_PROTOCOL_NAME, MULTICODEC_IDENTIFY_PUSH_PROTOCOL_VERSION, PUSH_DEBOUNCE_MS } from './consts.ts' import { Identify as IdentifyMessage } from './pb/message.ts' -import { AbstractIdentify, consumeIdentifyMessage, defaultValues, mergeIdentifyMessages } 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' @@ -145,34 +144,7 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif signal: AbortSignal.timeout(this.timeout) } - const pb = pbStream(stream, { - maxDataLength: this.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 push - %e', err) - break - } - } - - if (messages.length >= MAX_IDENTIFY_MESSAGES) { - log('reached MAX_IDENTIFY_MESSAGES, returning truncated identify push') - } - - try { - await stream.close(options) - } catch (err: any) { - log.trace('error closing identify-push stream after read - %e', err) - stream.abort(err) - } + const messages = await readIdentifyMessages(stream, this.maxMessageSize, options, log) await consumeIdentifyMessage(this.components.peerStore, this.components.events, log, connection, mergeIdentifyMessages(messages)) diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index d3f116409e..88eb920323 100644 --- a/packages/protocol-identify/src/identify.ts +++ b/packages/protocol-identify/src/identify.ts @@ -7,12 +7,11 @@ import { CODE_IP6, CODE_IP6ZONE, CODE_P2P } from '@multiformats/multiaddr' import { IP_OR_DOMAIN, TCP } from '@multiformats/multiaddr-matcher' import { setMaxListeners } from 'main-event' import { - MAX_IDENTIFY_MESSAGES, MULTICODEC_IDENTIFY_PROTOCOL_NAME, MULTICODEC_IDENTIFY_PROTOCOL_VERSION } from './consts.ts' import { Identify as IdentifyMessage } from './pb/message.ts' -import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, mergeIdentifyMessages } 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' @@ -61,35 +60,7 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt }) log = stream.log.newScope('identify') - const pb = pbStream(stream, { - maxDataLength: this.maxMessageSize - }).pb(IdentifyMessage) - - // Read up to MAX_IDENTIFY_MESSAGES (per libp2p/specs#709). - 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) - } + const messages = await readIdentifyMessages(stream, this.maxMessageSize, options, log) return mergeIdentifyMessages(messages) } catch (err: any) { diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index 62beed6389..df5fa40c64 100644 --- a/packages/protocol-identify/src/utils.ts +++ b/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 { 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' @@ -202,6 +203,51 @@ export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMes 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 dedup/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 { + 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 From 176d3f1e188c9ac4d2c2645ff72eaba801c1af57 Mon Sep 17 00:00:00 2001 From: tabcat Date: Sun, 10 May 2026 05:31:12 +0700 Subject: [PATCH 19/21] test(identify): move helper-behavior tests to direct readIdentifyMessages units Adds 5 unit tests against readIdentifyMessages in utils.spec.ts: happy path, empty-stream rejection, cap boundary, lenient-catch on read error after success, and close-error preserves messages with abort. Removes the integration-level equivalents from index.spec.ts (5) and push.spec.ts (2) that were exercising the same helper paths through the full identify / push handler. Coverage of "helper is wired in correctly" stays in the existing "should merge multiple identify messages from the remote" and "should handle multiple push messages and merge them" tests. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/test/index.spec.ts | 169 ------------------ packages/protocol-identify/test/push.spec.ts | 64 ------- packages/protocol-identify/test/utils.spec.ts | 98 +++++++++- 3 files changed, 97 insertions(+), 234 deletions(-) diff --git a/packages/protocol-identify/test/index.spec.ts b/packages/protocol-identify/test/index.spec.ts index acac94c259..9eb7e4efe1 100644 --- a/packages/protocol-identify/test/index.spec.ts +++ b/packages/protocol-identify/test/index.spec.ts @@ -267,175 +267,6 @@ describe('identify', () => { .to.have.property('peerRecordEnvelope').that.equalBytes(peerRecordEnvelope) }) - it('should reject when remote closes the stream without sending any message', async () => { - identify = new Identify(components) - await start(identify) - - const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519')) - const [outgoingStream] = await streamPair() - const connection = stubInterface({ remotePeer }) - connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) - - // Do not push any bytes - just close the remote write side. - outgoingStream.remoteWriteStatus = 'closed' - - // The receive loop only treats an EOF as "remote finished" once at least - // one message has been read; with zero messages, the EOF propagates as - // UnexpectedEOFError so the caller sees a hard failure for an empty stream. - await expect(identify.identify(connection)) - .to.eventually.be.rejected() - .with.property('name', 'UnexpectedEOFError') - }) - - it('should return merged identify when peer sends MAX_IDENTIFY_MESSAGES + 1 messages', async () => { - identify = new Identify(components) - await start(identify) - - const remotePrivateKey = await generateKeyPair('Ed25519') - const remotePeer = peerIdFromPrivateKey(remotePrivateKey) - - // First message carries publicKey so identify validates. - // Subsequent 10 messages each carry one extra protocol. - // Total 11 messages - reader should truncate to first 10. - const messages: IdentifyMessage[] = [{ - listenAddrs: [], - protocols: ['/test/0/1.0.0'], - publicKey: publicKeyToProtobuf(remotePeer.publicKey) - }] - for (let i = 1; i < 11; i++) { - messages.push({ listenAddrs: [], protocols: [`/test/${i}/1.0.0`] }) - } - - const [outgoingStream] = await streamPair() - const connection = stubInterface({ remotePeer }) - connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) - - const identifyPromise = identify.identify(connection) - await new Promise(resolve => setTimeout(resolve, 0)) - - const encoded = messages.map(m => lp.encode.single(IdentifyMessage.encode(m)).subarray()) - const totalLen = encoded.reduce((n, e) => n + e.byteLength, 0) - const combined = new Uint8Array(totalLen) - let off = 0 - for (const e of encoded) { combined.set(e, off); off += e.byteLength } - - outgoingStream.push(combined) - outgoingStream.remoteWriteStatus = 'closed' - - const result = await identifyPromise - // Reader merged at most MAX_IDENTIFY_MESSAGES (10) - the 11th protocol is truncated. - expect(result.protocols).to.have.lengthOf.at.most(10) - expect(result.protocols).to.not.include('/test/10/1.0.0') - }) - - it('should return successfully-read messages when peer sends garbage bytes mid-stream', async () => { - identify = new Identify(components) - await start(identify) - - const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519')) - const firstMessage: IdentifyMessage = { - listenAddrs: [], - protocols: ['/foo/bar/1.0'], - publicKey: publicKeyToProtobuf(remotePeer.publicKey) - } - - const [outgoingStream] = await streamPair() - const connection = stubInterface({ remotePeer }) - connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) - - const identifyPromise = identify.identify(connection, { - signal: AbortSignal.timeout(500) - }) - await new Promise(resolve => setTimeout(resolve, 0)) - - const encoded1 = lp.encode.single(IdentifyMessage.encode(firstMessage)).subarray() - // Append a varint length prefix that promises more bytes than provided so - // pb.read on iteration 2 will block until the abort signal fires. - const garbage = new Uint8Array([0xff, 0xff, 0xff, 0xff, 0x07]) - const combined = new Uint8Array(encoded1.byteLength + garbage.byteLength) - combined.set(encoded1) - combined.set(garbage, encoded1.byteLength) - - outgoingStream.push(combined) - - const result = await identifyPromise - expect(result.protocols).to.include('/foo/bar/1.0') - }) - - it('should return the message when remote resets the stream after a successful write', async () => { - identify = new Identify(components) - await start(identify) - - const remotePrivateKey = await generateKeyPair('Ed25519') - const remotePeer = peerIdFromPrivateKey(remotePrivateKey) - - 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({ remotePeer }) - connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) - - const identifyPromise = identify.identify(connection) - await new Promise(resolve => setTimeout(resolve, 0)) - - const encoded = lp.encode.single(IdentifyMessage.encode(message)).subarray() - outgoingStream.push(encoded) - // Set status to a non-'writable' value to simulate stream-reset-after-success. - // This documents the I4 widening: any non-'writable' status post-success - // is treated as EOF. - outgoingStream.remoteWriteStatus = 'closed' - - const result = await identifyPromise - expect(result.protocols).to.include('/foo/bar/1.0') - }) - - it('should succeed even if close() throws after a successful read (C2 regression)', async () => { - identify = new Identify(components) - await start(identify) - - const remotePrivateKey = await generateKeyPair('Ed25519') - const remotePeer = peerIdFromPrivateKey(remotePrivateKey) - - 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({ remotePeer }) - connection.newStream.withArgs('/ipfs/id/1.0.0').resolves(outgoingStream) - - // Force close() to throw with a StreamStateError-shaped error after a - // successful read. This simulates the original bug: byteStream.unwrap() - // throwing because the read buffer holds trailing bytes that can't be - // pushed back to a closed-read stream. - const originalClose = outgoingStream.close.bind(outgoingStream) - outgoingStream.close = async (...args: any[]) => { - // Let the underlying close run, then throw afterwards. - await originalClose(...args) - throw Object.assign(new Error('simulated close-after-read failure'), { name: 'StreamStateError' }) - } - - const identifyPromise = identify.identify(connection) - await new Promise(resolve => setTimeout(resolve, 0)) - - const encoded = lp.encode.single(IdentifyMessage.encode(message)).subarray() - outgoingStream.push(encoded) - outgoingStream.remoteWriteStatus = 'closed' - - const result = await identifyPromise - expect(result.protocols).to.include('/foo/bar/1.0') - }) - it('should limit incoming identify message sizes', async () => { const maxMessageSize = 100 diff --git a/packages/protocol-identify/test/push.spec.ts b/packages/protocol-identify/test/push.spec.ts index ad66b8515f..2cffe577b8 100644 --- a/packages/protocol-identify/test/push.spec.ts +++ b/packages/protocol-identify/test/push.spec.ts @@ -206,70 +206,6 @@ describe('identify (push)', () => { expect(update.peerRecordEnvelope).to.deep.equal(signedPeerRecord.marshal()) }) - it('should reject incoming push when remote closes the stream without sending any message', async () => { - identify = new IdentifyPush(components) - await start(identify) - - const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519')) - const [outgoingStream, incomingStream] = await streamPair() - const connection = stubInterface({ remotePeer }) - - // Close the writer side without sending any bytes. The push handler reads - // from the incoming stream; with no messages received the receive loop's - // EOF propagates as UnexpectedEOFError (the InvalidMessageError fallback - // is only reachable after a successful loop completion with zero messages). - outgoingStream.remoteWriteStatus = 'closed' - incomingStream.remoteWriteStatus = 'closed' - void outgoingStream.close() - - components.peerStore.patch.reset() - - await expect(identify.handleProtocol(incomingStream, connection)) - .to.eventually.be.rejected() - .with.property('name', 'UnexpectedEOFError') - - expect(components.peerStore.patch.callCount).to.equal(0) - }) - - it('should succeed even if close() throws after a successful read (C2 regression)', async () => { - identify = new IdentifyPush(components) - await start(identify) - - const remotePrivateKey = await generateKeyPair('Ed25519') - const remotePeer = peerIdFromPrivateKey(remotePrivateKey) - const [outgoingStream, incomingStream] = await streamPair() - const connection = stubInterface({ remotePeer }) - - const updatedProtocol = '/special-new-protocol/1.0.0' - const updatedAddress = multiaddr('/ip4/127.0.0.1/tcp/48322') - - const pb = pbStream(outgoingStream).pb(IdentifyMessage) - await pb.write({ - publicKey: publicKeyToProtobuf(remotePeer.publicKey), - protocols: [updatedProtocol], - listenAddrs: [updatedAddress.bytes] - }) - await outgoingStream.close() - - // Force close() on the incoming (server-side) stream to throw after the - // message has been read. The push handler awaits stream.close(options) and - // wraps it in try/catch; the read result must still be consumed. - const originalClose = incomingStream.close.bind(incomingStream) - incomingStream.close = async (...args: any[]) => { - await originalClose(...args) - throw Object.assign(new Error('simulated close-after-read failure'), { name: 'StreamStateError' }) - } - - components.peerStore.patch.reset() - - await identify.handleProtocol(incomingStream, connection) - - // patch should still happen — close error must be swallowed. - expect(components.peerStore.patch.callCount).to.equal(1) - const update = components.peerStore.patch.getCall(0).args[1] - expect(update.protocols).to.include(updatedProtocol) - }) - it('should time out during push identify', async () => { identify = new IdentifyPush(components, { timeout: 10 diff --git a/packages/protocol-identify/test/utils.spec.ts b/packages/protocol-identify/test/utils.spec.ts index a724e5c931..919131cd6b 100644 --- a/packages/protocol-identify/test/utils.spec.ts +++ b/packages/protocol-identify/test/utils.spec.ts @@ -1,7 +1,10 @@ +import { defaultLogger } from '@libp2p/logger' +import { streamPair } from '@libp2p/utils' import { multiaddr } from '@multiformats/multiaddr' import { expect } from 'aegir/chai' +import * as lp from 'it-length-prefixed' import { Identify as IdentifyMessage } from '../src/pb/message.js' -import { mergeIdentifyMessages } from '../src/utils.js' +import { mergeIdentifyMessages, readIdentifyMessages } from '../src/utils.js' describe('mergeIdentifyMessages', () => { it('returns a single message unchanged', () => { @@ -84,3 +87,96 @@ describe('mergeIdentifyMessages', () => { expect(merged.agentVersion).to.equal('agent/1.0') }) }) + +describe('readIdentifyMessages', () => { + const log = defaultLogger().forComponent('test') + + function encodeAll (msgs: IdentifyMessage[]): Uint8Array { + const parts = msgs.map(m => lp.encode.single(IdentifyMessage.encode(m)).subarray()) + const total = parts.reduce((n, p) => n + p.byteLength, 0) + const out = new Uint8Array(total) + let off = 0 + for (const p of parts) { out.set(p, off); off += p.byteLength } + return out + } + + it('returns all messages sent before clean EOF', async () => { + const [outgoingStream, incomingStream] = await streamPair() + const msgs: IdentifyMessage[] = [ + { listenAddrs: [], protocols: ['/a/1.0'] }, + { listenAddrs: [], protocols: ['/b/1.0'] } + ] + incomingStream.send(encodeAll(msgs)) + void incomingStream.close() + + const result = await readIdentifyMessages(outgoingStream, 8192, {}, log) + expect(result).to.have.lengthOf(2) + expect(result[0].protocols).to.deep.equal(['/a/1.0']) + expect(result[1].protocols).to.deep.equal(['/b/1.0']) + }) + + it('throws when stream closes without any messages', async () => { + const [outgoingStream, incomingStream] = await streamPair() + void incomingStream.close() + + await expect(readIdentifyMessages(outgoingStream, 8192, {}, log)) + .to.eventually.be.rejected() + .with.property('name', 'UnexpectedEOFError') + }) + + it('stops at MAX_IDENTIFY_MESSAGES even if more were sent', async () => { + const [outgoingStream, incomingStream] = await streamPair() + const msgs: IdentifyMessage[] = [] + for (let i = 0; i < 11; i++) { + msgs.push({ listenAddrs: [], protocols: [`/test/${i}/1.0`] }) + } + incomingStream.send(encodeAll(msgs)) + void incomingStream.close() + + const result = await readIdentifyMessages(outgoingStream, 8192, {}, log) + expect(result).to.have.lengthOf(10) + expect(result.map(m => m.protocols[0])).to.not.include('/test/10/1.0') + }) + + it('returns partial data when a read fails after at least one success', async () => { + const [outgoingStream, incomingStream] = await streamPair() + const valid: IdentifyMessage = { listenAddrs: [], protocols: ['/foo/1.0'] } + const encodedValid = lp.encode.single(IdentifyMessage.encode(valid)).subarray() + // varint length prefix that promises far more bytes than provided — pb.read on + // iteration 2 will block until the abort signal fires. + const garbage = new Uint8Array([0xff, 0xff, 0xff, 0xff, 0x07]) + const combined = new Uint8Array(encodedValid.byteLength + garbage.byteLength) + combined.set(encodedValid) + combined.set(garbage, encodedValid.byteLength) + incomingStream.send(combined) + + const result = await readIdentifyMessages(outgoingStream, 8192, { signal: AbortSignal.timeout(500) }, log) + expect(result).to.have.lengthOf(1) + expect(result[0].protocols).to.deep.equal(['/foo/1.0']) + }) + + it('preserves messages and aborts the stream when close() throws', async () => { + const [outgoingStream, incomingStream] = await streamPair() + const msg: IdentifyMessage = { listenAddrs: [], protocols: ['/foo/1.0'] } + incomingStream.send(lp.encode.single(IdentifyMessage.encode(msg)).subarray()) + void incomingStream.close() + + let aborted = false + const originalAbort = outgoingStream.abort.bind(outgoingStream) + outgoingStream.abort = (err: Error) => { + aborted = true + originalAbort(err) + } + + const originalClose = outgoingStream.close.bind(outgoingStream) + outgoingStream.close = async (...args: any[]) => { + await originalClose(...args) + throw Object.assign(new Error('simulated close failure'), { name: 'StreamStateError' }) + } + + const result = await readIdentifyMessages(outgoingStream, 8192, {}, log) + expect(result).to.have.lengthOf(1) + expect(result[0].protocols).to.deep.equal(['/foo/1.0']) + expect(aborted).to.be.true() + }) +}) From 5f989ef3d3783a4931f60625656b0ab3c10e47bb Mon Sep 17 00:00:00 2001 From: tabcat Date: Sun, 10 May 2026 06:06:52 +0700 Subject: [PATCH 20/21] test(identify): restore minimal caller-wiring tests for empty-stream + close-throws The recent dedup move pulled the helper-behavior tests into utils.spec.ts as direct readIdentifyMessages units, but in doing so dropped end-to-end verification that those error/recovery surfaces flow through the public caller (Identify.identify / IdentifyPush.handleProtocol). A future refactor of the caller wrapper could regress without any unit failure surfacing it. Adds 4 short integration assertions (~15 lines each, much smaller than the originals): empty-stream UnexpectedEOFError propagation and close-error recovery, on both identify and identify-push code paths. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/protocol-identify/test/index.spec.ts | 44 ++++++++++++++++++ packages/protocol-identify/test/push.spec.ts | 45 +++++++++++++++++++ 2 files changed, 89 insertions(+) diff --git a/packages/protocol-identify/test/index.spec.ts b/packages/protocol-identify/test/index.spec.ts index 9eb7e4efe1..895263185a 100644 --- a/packages/protocol-identify/test/index.spec.ts +++ b/packages/protocol-identify/test/index.spec.ts @@ -267,6 +267,50 @@ describe('identify', () => { .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({ 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({ 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 diff --git a/packages/protocol-identify/test/push.spec.ts b/packages/protocol-identify/test/push.spec.ts index 2cffe577b8..36274d8e00 100644 --- a/packages/protocol-identify/test/push.spec.ts +++ b/packages/protocol-identify/test/push.spec.ts @@ -206,6 +206,51 @@ describe('identify (push)', () => { expect(update.peerRecordEnvelope).to.deep.equal(signedPeerRecord.marshal()) }) + it('should propagate UnexpectedEOFError when remote closes incoming push without sending', async () => { + identify = new IdentifyPush(components) + await start(identify) + + const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519')) + const [outgoingStream, incomingStream] = await streamPair() + const connection = stubInterface({ remotePeer }) + void outgoingStream.close() + + components.peerStore.patch.reset() + + await expect(identify.handleProtocol(incomingStream, connection)) + .to.eventually.be.rejected() + .with.property('name', 'UnexpectedEOFError') + expect(components.peerStore.patch.callCount).to.equal(0) + }) + + it('should still consume the push message when close() throws after a successful read', async () => { + identify = new IdentifyPush(components) + await start(identify) + + const remotePrivateKey = await generateKeyPair('Ed25519') + const remotePeer = peerIdFromPrivateKey(remotePrivateKey) + const [outgoingStream, incomingStream] = await streamPair() + const connection = stubInterface({ remotePeer }) + + const pb = pbStream(outgoingStream).pb(IdentifyMessage) + await pb.write({ + publicKey: publicKeyToProtobuf(remotePeer.publicKey), + protocols: ['/foo/1.0'], + listenAddrs: [] + }) + await outgoingStream.close() + + const originalClose = incomingStream.close.bind(incomingStream) + incomingStream.close = async (...args: any[]) => { + await originalClose(...args) + throw Object.assign(new Error('simulated close failure'), { name: 'StreamStateError' }) + } + + components.peerStore.patch.reset() + await identify.handleProtocol(incomingStream, connection) + expect(components.peerStore.patch.callCount).to.equal(1) + }) + it('should time out during push identify', async () => { identify = new IdentifyPush(components, { timeout: 10 From cf7e95baabfc51093dcade6be33891068bbdcc9d Mon Sep 17 00:00:00 2001 From: tabcat Date: Sun, 10 May 2026 07:01:24 +0700 Subject: [PATCH 21/21] docs(identify): reword 'dedup' to 'deduplication' for spell check --- packages/protocol-identify/src/utils.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index df5fa40c64..748f084ef1 100644 --- a/packages/protocol-identify/src/utils.ts +++ b/packages/protocol-identify/src/utils.ts @@ -174,8 +174,8 @@ export async function consumeIdentifyMessage (peerStore: PeerStore, events: Type /** * Merge multiple received Identify messages into one. Repeated `listenAddrs` - * are concatenated as-is — peerstore handles dedup downstream. `protocols` are - * deduplicated via Set since they're string identifiers. + * 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] } @@ -208,9 +208,9 @@ export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMes * 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 dedup/merge downstream). Close - * errors are swallowed and the stream is aborted instead — preserves the - * read result while ensuring transport cleanup. + * (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.