diff --git a/packages/beacon-node/src/network/peers/peerManager.ts b/packages/beacon-node/src/network/peers/peerManager.ts index 81d52f7897bb..3fc132d0371f 100644 --- a/packages/beacon-node/src/network/peers/peerManager.ts +++ b/packages/beacon-node/src/network/peers/peerManager.ts @@ -1,4 +1,4 @@ -import {Connection, PeerId, PrivateKey} from "@libp2p/interface"; +import {Connection, type IdentifyResult, PeerId, PrivateKey} from "@libp2p/interface"; import {BitArray} from "@chainsafe/ssz"; import {BeaconConfig} from "@lodestar/config"; import {LoggerNode} from "@lodestar/logger/node"; @@ -162,6 +162,8 @@ export class PeerManager { // A single map of connected peers with all necessary data to handle PINGs, STATUS, and metrics private connectedPeers: Map; + /** Track one in-flight identify call per peer/connection id */ + private readonly identifyInProgress = new Map(); private opts: PeerManagerOpts; private intervals: NodeJS.Timeout[] = []; @@ -193,6 +195,7 @@ export class PeerManager { this.libp2p.services.components.events.addEventListener(Libp2pEvent.connectionOpen, this.onLibp2pPeerConnect); this.libp2p.services.components.events.addEventListener(Libp2pEvent.connectionClose, this.onLibp2pPeerDisconnect); + this.libp2p.services.components.events.addEventListener("peer:identify", this.onPeerIdentify); this.networkEventBus.on(NetworkEvent.reqRespRequest, this.onRequest); this.lastStatus = this.statusCache.get(); @@ -236,6 +239,7 @@ export class PeerManager { Libp2pEvent.connectionClose, this.onLibp2pPeerDisconnect ); + this.libp2p.services.components.events.removeEventListener("peer:identify", this.onPeerIdentify); this.networkEventBus.off(NetworkEvent.reqRespRequest, this.onRequest); for (const interval of this.intervals) clearInterval(interval); } @@ -486,7 +490,25 @@ export class PeerManager { // peers that close identify right after connection open or turn out to be // irrelevant. if (peerData?.agentVersion === null) { - void this.identifyPeer(peer.toString(), prettyPrintPeerId(peer), getConnection(this.libp2p, peer.toString())); + const peerIdStr = peer.toString(); + const connection = getConnection(this.libp2p, peerIdStr); + if (!connection || connection.status !== "open") { + this.logger.debug("Peer has no open connection for identify", {peerId: prettyPrintPeerId(peer)}); + return; + } + + const identifyKey = connection.id; + if (this.identifyInProgress.get(peerIdStr) === identifyKey) { + return; + } + + this.identifyInProgress.set(peerIdStr, identifyKey); + void this.identifyPeer(peerIdStr, prettyPrintPeerId(peer), connection, identifyKey).finally(() => { + // Clear only if this identify attempt is still the active one for this peer + if (this.identifyInProgress.get(peerIdStr) === identifyKey) { + this.identifyInProgress.delete(peerIdStr); + } + }); } } } @@ -845,6 +867,7 @@ export class PeerManager { // remove the ping and status timer for the peer this.connectedPeers.delete(peerIdStr); + this.identifyInProgress.delete(peerIdStr); this.logger.verbose(logMessage, logContext); this.networkEventBus.emit(NetworkEvent.peerDisconnected, {peer: peerIdStr}); @@ -862,14 +885,47 @@ export class PeerManager { } } - private async identifyPeer(peerIdStr: string, peerIdPretty: string, connection?: Connection): Promise { - if (!connection || connection.status !== "open") { + /** + * Consume successful identify results from libp2p events. + * This captures agentVersion from identify-push or successful inbound/outbound identify, + * even if our explicit identify request failed earlier. + */ + private onPeerIdentify = (evt: CustomEvent): void => { + const {peerId, agentVersion} = evt.detail; + if (!agentVersion) return; + + const peerIdStr = peerId.toString(); + const peerData = this.connectedPeers.get(peerIdStr); + if (!peerData) return; + + peerData.agentVersion = agentVersion; + peerData.agentClient = getKnownClientFromAgentVersion(agentVersion); + this.identifyInProgress.delete(peerIdStr); + }; + + private async identifyPeer( + peerIdStr: string, + peerIdPretty: string, + connection: Connection, + identifyKey: string + ): Promise { + if (this.identifyInProgress.get(peerIdStr) !== identifyKey) { + return; + } + + if (connection.status !== "open") { this.logger.debug("Peer has no open connection for identify", {peerId: peerIdPretty}); return; } try { const result = await this.libp2p.services.identify.identify(connection); + + // A newer identify attempt may have superseded this one (e.g. reconnect). + if (this.identifyInProgress.get(peerIdStr) !== identifyKey) { + return; + } + const agentVersion = result.agentVersion; if (agentVersion) { const connectedPeerData = this.connectedPeers.get(peerIdStr); diff --git a/packages/beacon-node/test/e2e/network/peers/peerManager.test.ts b/packages/beacon-node/test/e2e/network/peers/peerManager.test.ts index deecd6e6ee25..2cde954688f6 100644 --- a/packages/beacon-node/test/e2e/network/peers/peerManager.test.ts +++ b/packages/beacon-node/test/e2e/network/peers/peerManager.test.ts @@ -342,4 +342,158 @@ describe("network / peers / PeerManager", () => { expect(peerData?.agentVersion).toBe("Nimbus/v25.0.0"); expect(peerData?.agentClient).toBe(ClientKind.Nimbus); }); + + it("Should deduplicate in-flight identify requests for the same connection", async () => { + const {libp2p, peerManager, statusCache, networkEventBus} = await mockModules(); + + let resolveIdentify!: (value: {agentVersion: string}) => void; + const identifyPromise = new Promise<{agentVersion: string}>((resolve) => { + resolveIdentify = resolve; + }); + + vi.spyOn(libp2p.services.identify, "identify").mockImplementation( + () => identifyPromise as ReturnType + ); + + const connection = { + id: "connection-1", + direction: "inbound", + status: "open", + remotePeer: peerId1, + close: async () => {}, + abort: () => {}, + } as unknown as Connection; + + getConnectionsMap(libp2p).set(peerId1.toString(), {key: peerId1, value: [connection]}); + await peerManager["onLibp2pPeerConnect"](new CustomEvent("evt", {detail: connection})); + + const remoteStatus = statusCache.get(); + networkEventBus.emit(NetworkEvent.reqRespRequest, { + request: {method: ReqRespMethod.Status, body: remoteStatus}, + peer: peerId1, + peerClient: "Unknown", + }); + networkEventBus.emit(NetworkEvent.reqRespRequest, { + request: {method: ReqRespMethod.Status, body: remoteStatus}, + peer: peerId1, + peerClient: "Unknown", + }); + + await sleep(0); + expect(libp2p.services.identify.identify).toHaveBeenCalledTimes(1); + + resolveIdentify({agentVersion: "Prysm/v6.0.0"}); + await sleep(0); + + const peerData = peerManager["connectedPeers"].get(peerId1.toString()); + expect(peerData?.agentVersion).toBe("Prysm/v6.0.0"); + expect(peerData?.agentClient).toBe(ClientKind.Prysm); + }); + + it("Should allow a new identify attempt after reconnect and ignore stale previous result", async () => { + const {libp2p, peerManager, statusCache, networkEventBus} = await mockModules(); + + let resolveFirstIdentify!: (value: {agentVersion: string}) => void; + const firstIdentifyPromise = new Promise<{agentVersion: string}>((resolve) => { + resolveFirstIdentify = resolve; + }); + + vi.spyOn(libp2p.services.identify, "identify") + .mockImplementationOnce(() => firstIdentifyPromise as ReturnType) + .mockImplementationOnce( + () => Promise.resolve({agentVersion: "Teku/v24.9.0"}) as ReturnType + ); + + const connection1 = { + id: "connection-1", + direction: "inbound", + status: "open", + remotePeer: peerId1, + close: async () => {}, + abort: () => {}, + } as unknown as Connection; + + getConnectionsMap(libp2p).set(peerId1.toString(), {key: peerId1, value: [connection1]}); + await peerManager["onLibp2pPeerConnect"](new CustomEvent("evt", {detail: connection1})); + + const remoteStatus = statusCache.get(); + networkEventBus.emit(NetworkEvent.reqRespRequest, { + request: {method: ReqRespMethod.Status, body: remoteStatus}, + peer: peerId1, + peerClient: "Unknown", + }); + await sleep(0); + + const closedConnection1 = {...connection1, status: "closed"} as Connection; + getConnectionsMap(libp2p).set(peerId1.toString(), {key: peerId1, value: [closedConnection1]}); + await peerManager["onLibp2pPeerDisconnect"](new CustomEvent("evt", {detail: closedConnection1})); + + const connection2 = { + id: "connection-2", + direction: "inbound", + status: "open", + remotePeer: peerId1, + close: async () => {}, + abort: () => {}, + } as unknown as Connection; + getConnectionsMap(libp2p).set(peerId1.toString(), {key: peerId1, value: [connection2]}); + await peerManager["onLibp2pPeerConnect"](new CustomEvent("evt", {detail: connection2})); + + networkEventBus.emit(NetworkEvent.reqRespRequest, { + request: {method: ReqRespMethod.Status, body: remoteStatus}, + peer: peerId1, + peerClient: "Unknown", + }); + await sleep(0); + + expect(libp2p.services.identify.identify).toHaveBeenCalledTimes(2); + + // Resolve old identify last; it must not overwrite new connection's identify result. + resolveFirstIdentify({agentVersion: "Lighthouse/v6.0.1"}); + await sleep(0); + + const peerData = peerManager["connectedPeers"].get(peerId1.toString()); + expect(peerData?.agentVersion).toBe("Teku/v24.9.0"); + expect(peerData?.agentClient).toBe(ClientKind.Teku); + }); + + it("Should update agentVersion via peer:identify event even if explicit identify fails", async () => { + const {libp2p, peerManager, statusCache, networkEventBus} = await mockModules(); + + vi.spyOn(libp2p.services.identify, "identify").mockRejectedValue(new Error("Unexpected EOF")); + + const connection = { + id: "connection-1", + direction: "inbound", + status: "open", + remotePeer: peerId1, + close: async () => {}, + abort: () => {}, + } as unknown as Connection; + + getConnectionsMap(libp2p).set(peerId1.toString(), {key: peerId1, value: [connection]}); + await peerManager["onLibp2pPeerConnect"](new CustomEvent("evt", {detail: connection})); + + const remoteStatus = statusCache.get(); + networkEventBus.emit(NetworkEvent.reqRespRequest, { + request: {method: ReqRespMethod.Status, body: remoteStatus}, + peer: peerId1, + peerClient: "Unknown", + }); + await sleep(0); + + libp2p.services.components.events.dispatchEvent( + new CustomEvent("peer:identify", { + detail: { + peerId: peerId1, + agentVersion: "Lighthouse/v6.0.1", + }, + }) + ); + await sleep(0); + + const peerData = peerManager["connectedPeers"].get(peerId1.toString()); + expect(peerData?.agentVersion).toBe("Lighthouse/v6.0.1"); + expect(peerData?.agentClient).toBe(ClientKind.Lighthouse); + }); });