fix: single in-flight identify per peer with event-driven fallback

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<PeerIdStr, PeerData>;
+  /** Track one in-flight identify call per peer/connection id */
+  private readonly identifyInProgress = new Map<PeerIdStr, string>();
 
   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<void> {
-    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<IdentifyResult>): 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<void> {
+    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);

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<typeof libp2p.services.identify.identify>
+    );
+
+    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<typeof libp2p.services.identify.identify>)
+      .mockImplementationOnce(
+        () => Promise.resolve({agentVersion: "Teku/v24.9.0"}) as ReturnType<typeof libp2p.services.identify.identify>
+      );
+
+    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);
+  });
 });