diff --git a/packages/protocol-identify/src/consts.ts b/packages/protocol-identify/src/consts.ts index 0dde24239d..7b9014f772 100644 --- a/packages/protocol-identify/src/consts.ts +++ b/packages/protocol-identify/src/consts.ts @@ -11,6 +11,13 @@ export const MULTICODEC_IDENTIFY_PUSH_PROTOCOL_VERSION = '1.0.0' // https://github.com/libp2p/go-libp2p/blob/8d2e54e1637041d5cf4fac1e531287560bd1f4ac/p2p/protocol/identify/id.go#L52 export const MAX_IDENTIFY_MESSAGE_SIZE = 1024 * 8 +// Maximum number of LP-framed messages we will read from a single identify or +// identify-push stream before treating remaining data as truncated. The +// identify spec is silent on this; the value matches go-libp2p's `maxMessages` +// by convention. See the proposed update at +// https://github.com/libp2p/specs/pull/709 +export const MAX_IDENTIFY_MESSAGES = 10 + // https://github.com/libp2p/go-libp2p/blob/0385ec924bad172f74a74db09939e97c079b1420/p2p/protocol/identify/id.go#L47C7-L47C25 export const MAX_PUSH_CONCURRENCY = 32 diff --git a/packages/protocol-identify/src/identify-push.ts b/packages/protocol-identify/src/identify-push.ts index b490c13a47..10b6c039c0 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, consumeIdentifyMessage, defaultValues } from './utils.ts' +import { AbstractIdentify, consumeIdentifyMessage, defaultValues, mergeIdentifyMessages, readIdentifyMessages } from './utils.ts' import type { IdentifyPush as IdentifyPushInterface, IdentifyPushComponents, IdentifyPushInit } from './index.ts' import type { Stream, Startable, Connection } from '@libp2p/interface' import type { ConnectionManager } from '@libp2p/interface-internal' @@ -144,14 +144,9 @@ export class IdentifyPush extends AbstractIdentify implements Startable, Identif signal: AbortSignal.timeout(this.timeout) } - const pb = pbStream(stream, { - maxDataLength: this.maxMessageSize - }).pb(IdentifyMessage) + const messages = await readIdentifyMessages(stream, this.maxMessageSize, options, log) - const message = await pb.read(options) - await stream.close(options) - - await consumeIdentifyMessage(this.components.peerStore, this.components.events, log, connection, message) + await consumeIdentifyMessage(this.components.peerStore, this.components.events, log, connection, mergeIdentifyMessages(messages)) log.trace('handled push from %p', connection.remotePeer) } diff --git a/packages/protocol-identify/src/identify.ts b/packages/protocol-identify/src/identify.ts index 269e12744d..88eb920323 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, consumeIdentifyMessage, defaultValues, getCleanMultiaddr } from './utils.ts' +import { AbstractIdentify, consumeIdentifyMessage, defaultValues, getCleanMultiaddr, mergeIdentifyMessages, readIdentifyMessages } from './utils.ts' import type { Identify as IdentifyInterface, IdentifyComponents, IdentifyInit } from './index.ts' import type { IdentifyResult, AbortOptions, Connection, Stream, Startable, Logger, NewStreamOptions } from '@libp2p/interface' @@ -60,14 +60,9 @@ export class Identify extends AbstractIdentify implements Startable, IdentifyInt }) log = stream.log.newScope('identify') - const pb = pbStream(stream, { - maxDataLength: this.maxMessageSize - }).pb(IdentifyMessage) + const messages = await readIdentifyMessages(stream, this.maxMessageSize, options, log) - const message = await pb.read(options) - await pb.unwrap().unwrap().close(options) - - return message + return mergeIdentifyMessages(messages) } catch (err: any) { log?.error('identify failed - %e', err) stream?.abort(err) diff --git a/packages/protocol-identify/src/utils.ts b/packages/protocol-identify/src/utils.ts index 6b993de87e..748f084ef1 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 { Identify as IdentifyMessage } from './pb/message.ts' -import type { Libp2pEvents, IdentifyResult, SignedPeerRecord, Logger, Connection, Peer, PeerData, PeerStore, Startable, Stream } from '@libp2p/interface' +import type { AbortOptions, Libp2pEvents, IdentifyResult, SignedPeerRecord, Logger, Connection, Peer, PeerData, PeerStore, Startable, Stream } from '@libp2p/interface' import type { Multiaddr } from '@multiformats/multiaddr' import type { TypedEventTarget } from 'main-event' @@ -171,6 +172,82 @@ export async function consumeIdentifyMessage (peerStore: PeerStore, events: Type return result } +/** + * Merge multiple received Identify messages into one. Repeated `listenAddrs` + * are concatenated as-is — peerstore handles deduplication downstream. + * `protocols` are deduplicated via Set since they're string identifiers. + */ +export function mergeIdentifyMessages (messages: IdentifyMessage[]): IdentifyMessage { + const merged: IdentifyMessage = { ...messages[0] } + + for (const msg of messages.slice(1)) { + if (msg.protocolVersion != null) { + merged.protocolVersion = msg.protocolVersion + } + if (msg.agentVersion != null) { + merged.agentVersion = msg.agentVersion + } + if (msg.publicKey != null) { + merged.publicKey = msg.publicKey + } + if (msg.observedAddr != null) { + merged.observedAddr = msg.observedAddr + } + if (msg.signedPeerRecord != null) { + merged.signedPeerRecord = msg.signedPeerRecord + } + merged.listenAddrs = [...merged.listenAddrs, ...msg.listenAddrs] + merged.protocols = [...new Set([...merged.protocols, ...msg.protocols])] + } + + return merged +} + +/** + * Read up to MAX_IDENTIFY_MESSAGES LP-framed Identify messages from the + * stream, then close. Used by both identify and identify-push receive paths. + * + * Any error after at least one successful read is treated as "stop reading" + * (we keep what we got — peerstore handles deduplication/merge downstream). + * Close errors are swallowed and the stream is aborted instead — preserves + * the read result while ensuring transport cleanup. + * + * Multi-message identify is per the proposed spec update at + * https://github.com/libp2p/specs/pull/709. + */ +export async function readIdentifyMessages (stream: Stream, maxMessageSize: number, options: AbortOptions, log: Logger): Promise { + 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 diff --git a/packages/protocol-identify/test/index.spec.ts b/packages/protocol-identify/test/index.spec.ts index d170604834..895263185a 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 }) @@ -164,6 +166,151 @@ describe('identify', () => { expect(outgoingStream).to.have.property('status', 'aborted') }) + it('should succeed if the remote closes the stream after sending the identify response', async () => { + identify = new Identify(components) + + await start(identify) + + const remotePeer = peerIdFromPrivateKey(await generateKeyPair('Ed25519')) + const message: IdentifyMessage = { + listenAddrs: [ + multiaddr('/ip4/123.123.123.123/tcp/123').bytes + ], + protocols: [ + '/foo/bar/1.0' + ], + publicKey: publicKeyToProtobuf(remotePeer.publicKey) + } + + const [outgoingStream] = await streamPair() + const connection = stubInterface({ + 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 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 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 @@ -204,6 +351,7 @@ describe('identify', () => { agentVersion: 'secret-agent', protocolVersion: '9000' }))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) @@ -257,6 +405,7 @@ describe('identify', () => { publicKey: publicKeyToProtobuf(remotePeer.publicKey), signedPeerRecord: oldPeerRecord.marshal() }))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) @@ -308,6 +457,7 @@ describe('identify', () => { const [outgoingStream, incomingStream] = await streamPair() incomingStream.send(lp.encode.single(IdentifyMessage.encode(message))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) @@ -356,6 +506,7 @@ describe('identify', () => { const [outgoingStream, incomingStream] = await streamPair() incomingStream.send(lp.encode.single(IdentifyMessage.encode(message))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) @@ -430,6 +581,7 @@ describe('identify', () => { const [outgoingStream, incomingStream] = await streamPair() incomingStream.send(lp.encode.single(IdentifyMessage.encode(message))) + void incomingStream.close() const connection = stubInterface({ remotePeer }) diff --git a/packages/protocol-identify/test/push.spec.ts b/packages/protocol-identify/test/push.spec.ts index 920598a03c..36274d8e00 100644 --- a/packages/protocol-identify/test/push.spec.ts +++ b/packages/protocol-identify/test/push.spec.ts @@ -2,6 +2,7 @@ import { generateKeyPair, publicKeyToProtobuf } from '@libp2p/crypto/keys' 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' @@ -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,105 @@ 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 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 diff --git a/packages/protocol-identify/test/utils.spec.ts b/packages/protocol-identify/test/utils.spec.ts new file mode 100644 index 0000000000..919131cd6b --- /dev/null +++ b/packages/protocol-identify/test/utils.spec.ts @@ -0,0 +1,182 @@ +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, readIdentifyMessages } from '../src/utils.js' + +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') + }) +}) + +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() + }) +})