feat: don't dial peers that failed before, make dialer use dial queue (#2478)

weboko · web-flow · commit 35acdf8fa5a8 · 2025-07-15T23:13:58.000+02:00
* make dialer use dial queue

* skip undiable peers

* cover new cases in tests

* expose dialer config at connection manager

* update tests with new config

* add more tests
diff --git a/.cspell.json b/.cspell.json
@@ -61,6 +61,7 @@
     "ineed",
     "IPAM",
     "ipfs",
+    "cooldown",
     "iwant",
     "jdev",
     "jswaku",
diff --git a/packages/core/src/lib/connection_manager/connection_limiter.spec.ts b/packages/core/src/lib/connection_manager/connection_limiter.spec.ts
@@ -48,7 +48,10 @@ describe("ConnectionLimiter", () => {
   const defaultOptions = {
     maxBootstrapPeers: 2,
     pingKeepAlive: 300,
-    relayKeepAlive: 300
+    relayKeepAlive: 300,
+    maxDialingPeers: 3,
+    failedDialCooldown: 60,
+    dialCooldown: 10
   };
 
   beforeEach(() => {
@@ -761,7 +764,10 @@ describe("ConnectionLimiter", () => {
       const customOptions = {
         maxBootstrapPeers: 1,
         pingKeepAlive: 300,
-        relayKeepAlive: 300
+        relayKeepAlive: 300,
+        maxDialingPeers: 3,
+        failedDialCooldown: 60,
+        dialCooldown: 10
       };
 
       connectionLimiter = new ConnectionLimiter({
@@ -834,7 +840,10 @@ describe("ConnectionLimiter", () => {
       const customOptions = {
         maxBootstrapPeers: 1,
         pingKeepAlive: 300,
-        relayKeepAlive: 300
+        relayKeepAlive: 300,
+        maxDialingPeers: 3,
+        failedDialCooldown: 60,
+        dialCooldown: 10
       };
 
       connectionLimiter = new ConnectionLimiter({
@@ -882,7 +891,10 @@ describe("ConnectionLimiter", () => {
       const customOptions = {
         maxBootstrapPeers: 1,
         pingKeepAlive: 300,
-        relayKeepAlive: 300
+        relayKeepAlive: 300,
+        maxDialingPeers: 3,
+        failedDialCooldown: 60,
+        dialCooldown: 10
       };
 
       connectionLimiter = new ConnectionLimiter({
@@ -928,7 +940,10 @@ describe("ConnectionLimiter", () => {
       const customOptions = {
         maxBootstrapPeers: 10,
         pingKeepAlive: 300,
-        relayKeepAlive: 300
+        relayKeepAlive: 300,
+        maxDialingPeers: 3,
+        failedDialCooldown: 60,
+        dialCooldown: 10
       };
 
       connectionLimiter = new ConnectionLimiter({
diff --git a/packages/core/src/lib/connection_manager/connection_manager.ts b/packages/core/src/lib/connection_manager/connection_manager.ts
@@ -24,6 +24,9 @@ const log = new Logger("connection-manager");
 const DEFAULT_MAX_BOOTSTRAP_PEERS_ALLOWED = 1;
 const DEFAULT_PING_KEEP_ALIVE_SEC = 5 * 60;
 const DEFAULT_RELAY_KEEP_ALIVE_SEC = 5 * 60;
+const DEFAULT_MAX_DIALING_PEERS = 3;
+const DEFAULT_FAILED_DIAL_COOLDOWN_SEC = 60;
+const DEFAULT_DIAL_COOLDOWN_SEC = 10;
 
 type ConnectionManagerConstructorOptions = {
   libp2p: Libp2p;
@@ -55,6 +58,9 @@ export class ConnectionManager implements IConnectionManager {
       maxBootstrapPeers: DEFAULT_MAX_BOOTSTRAP_PEERS_ALLOWED,
       pingKeepAlive: DEFAULT_PING_KEEP_ALIVE_SEC,
       relayKeepAlive: DEFAULT_RELAY_KEEP_ALIVE_SEC,
+      maxDialingPeers: DEFAULT_MAX_DIALING_PEERS,
+      failedDialCooldown: DEFAULT_FAILED_DIAL_COOLDOWN_SEC,
+      dialCooldown: DEFAULT_DIAL_COOLDOWN_SEC,
       ...options.config
     };
 
@@ -74,7 +80,8 @@ export class ConnectionManager implements IConnectionManager {
 
     this.dialer = new Dialer({
       libp2p: options.libp2p,
-      shardReader: this.shardReader
+      shardReader: this.shardReader,
+      options: this.options
     });
 
     this.discoveryDialer = new DiscoveryDialer({
@@ -97,13 +104,15 @@ export class ConnectionManager implements IConnectionManager {
   }
 
   public start(): void {
+    this.dialer.start();
     this.networkMonitor.start();
     this.discoveryDialer.start();
     this.keepAliveManager.start();
     this.connectionLimiter.start();
   }
 
   public stop(): void {
+    this.dialer.stop();
     this.networkMonitor.stop();
     this.discoveryDialer.stop();
     this.keepAliveManager.stop();
diff --git a/packages/core/src/lib/connection_manager/dialer.spec.ts b/packages/core/src/lib/connection_manager/dialer.spec.ts
@@ -1,5 +1,5 @@
 import { PeerId } from "@libp2p/interface";
-import { Libp2p } from "@waku/interfaces";
+import { ConnectionManagerOptions, Libp2p } from "@waku/interfaces";
 import { expect } from "chai";
 import sinon from "sinon";
 
@@ -13,6 +13,7 @@ describe("Dialer", () => {
   let mockPeerId: PeerId;
   let mockPeerId2: PeerId;
   let clock: sinon.SinonFakeTimers;
+  let mockOptions: ConnectionManagerOptions;
 
   const createMockPeerId = (id: string): PeerId =>
     ({
@@ -31,10 +32,21 @@ describe("Dialer", () => {
       isPeerOnNetwork: sinon.stub().resolves(true)
     } as unknown as sinon.SinonStubbedInstance<ShardReader>;
 
+    mockOptions = {
+      maxBootstrapPeers: 1,
+      pingKeepAlive: 300,
+      relayKeepAlive: 300,
+      maxDialingPeers: 3,
+      failedDialCooldown: 60,
+      dialCooldown: 10
+    };
+
     mockPeerId = createMockPeerId("12D3KooWTest1");
     mockPeerId2 = createMockPeerId("12D3KooWTest2");
 
-    clock = sinon.useFakeTimers();
+    clock = sinon.useFakeTimers({
+      now: 1000000000000
+    });
   });
 
   afterEach(() => {
@@ -49,7 +61,8 @@ describe("Dialer", () => {
     it("should create dialer with libp2p and shardReader", () => {
       dialer = new Dialer({
         libp2p,
-        shardReader: mockShardReader
+        shardReader: mockShardReader,
+        options: mockOptions
       });
 
       expect(dialer).to.be.instanceOf(Dialer);
@@ -60,7 +73,8 @@ describe("Dialer", () => {
     beforeEach(() => {
       dialer = new Dialer({
         libp2p,
-        shardReader: mockShardReader
+        shardReader: mockShardReader,
+        options: mockOptions
       });
     });
 
@@ -98,7 +112,8 @@ describe("Dialer", () => {
     beforeEach(() => {
       dialer = new Dialer({
         libp2p,
-        shardReader: mockShardReader
+        shardReader: mockShardReader,
+        options: mockOptions
       });
       dialer.start();
     });
@@ -135,7 +150,8 @@ describe("Dialer", () => {
     beforeEach(() => {
       dialer = new Dialer({
         libp2p,
-        shardReader: mockShardReader
+        shardReader: mockShardReader,
+        options: mockOptions
       });
       dialer.start();
     });
@@ -152,14 +168,28 @@ describe("Dialer", () => {
 
     it("should add peer to queue when queue is not empty", async () => {
       const dialStub = libp2p.dial as sinon.SinonStub;
-      dialStub.resolves();
 
-      void dialer.dial(mockPeerId);
+      let resolveFirstDial: () => void;
+      const firstDialPromise = new Promise<void>((resolve) => {
+        resolveFirstDial = resolve;
+      });
+      dialStub.onFirstCall().returns(firstDialPromise);
+      dialStub.onSecondCall().resolves();
+
+      const firstDialCall = dialer.dial(mockPeerId);
 
-      dialStub.resetHistory();
       await dialer.dial(mockPeerId2);
 
-      expect(dialStub.called).to.be.true;
+      expect(dialStub.calledOnce).to.be.true;
+      expect(dialStub.calledWith(mockPeerId)).to.be.true;
+
+      resolveFirstDial!();
+      await firstDialCall;
+
+      clock.tick(500);
+      await Promise.resolve();
+
+      expect(dialStub.calledTwice).to.be.true;
       expect(dialStub.calledWith(mockPeerId2)).to.be.true;
     });
 
@@ -186,7 +216,54 @@ describe("Dialer", () => {
       clock.tick(5000);
       await dialer.dial(mockPeerId);
 
-      expect(dialStub.called).to.be.true;
+      expect(dialStub.called).to.be.false;
+    });
+
+    it("should skip peer when failed to dial recently", async () => {
+      const dialStub = libp2p.dial as sinon.SinonStub;
+      dialStub.rejects(new Error("Dial failed"));
+
+      await dialer.dial(mockPeerId);
+      expect(dialStub.calledOnce).to.be.true;
+
+      dialStub.resetHistory();
+      dialStub.resolves();
+
+      clock.tick(30000);
+
+      await dialer.dial(mockPeerId);
+      expect(dialStub.called).to.be.false;
+    });
+
+    it("should populate queue if has active dial", async () => {
+      const dialStub = libp2p.dial as sinon.SinonStub;
+      const mockPeerId3 = createMockPeerId("12D3KooWTest3");
+
+      let resolveFirstDial: () => void;
+      const firstDialPromise = new Promise<void>((resolve) => {
+        resolveFirstDial = resolve;
+      });
+      dialStub.onFirstCall().returns(firstDialPromise);
+      dialStub.onSecondCall().resolves();
+      dialStub.onThirdCall().resolves();
+
+      const firstDialCall = dialer.dial(mockPeerId);
+
+      await dialer.dial(mockPeerId2);
+      await dialer.dial(mockPeerId3);
+
+      expect(dialStub.calledOnce).to.be.true;
+      expect(dialStub.calledWith(mockPeerId)).to.be.true;
+
+      resolveFirstDial!();
+      await firstDialCall;
+
+      clock.tick(500);
+      await Promise.resolve();
+
+      expect(dialStub.callCount).to.equal(3);
+      expect(dialStub.calledWith(mockPeerId2)).to.be.true;
+      expect(dialStub.calledWith(mockPeerId3)).to.be.true;
     });
 
     it("should allow redial after cooldown period", async () => {
@@ -252,13 +329,49 @@ describe("Dialer", () => {
       expect(dialStub.calledOnce).to.be.true;
       expect(dialStub.calledWith(mockPeerId)).to.be.true;
     });
+
+    it("should allow redial after failed dial cooldown expires", async () => {
+      const dialStub = libp2p.dial as sinon.SinonStub;
+      dialStub.onFirstCall().rejects(new Error("Dial failed"));
+      dialStub.onSecondCall().resolves();
+      await dialer.dial(mockPeerId);
+      expect(dialStub.calledOnce).to.be.true;
+      clock.tick(60001);
+      await dialer.dial(mockPeerId);
+      expect(dialStub.calledTwice).to.be.true;
+    });
+
+    it("should handle queue overflow by adding peers to queue", async () => {
+      const dialStub = libp2p.dial as sinon.SinonStub;
+      const peers = [];
+      for (let i = 0; i < 100; i++) {
+        peers.push(createMockPeerId(`12D3KooWTest${i}`));
+      }
+      let resolveFirstDial: () => void;
+      const firstDialPromise = new Promise<void>((resolve) => {
+        resolveFirstDial = resolve;
+      });
+      dialStub.onFirstCall().returns(firstDialPromise);
+      dialStub.resolves();
+      const firstDialCall = dialer.dial(peers[0]);
+      for (let i = 1; i < 100; i++) {
+        await dialer.dial(peers[i]);
+      }
+      expect(dialStub.calledOnce).to.be.true;
+      resolveFirstDial!();
+      await firstDialCall;
+      clock.tick(500);
+      await Promise.resolve();
+      expect(dialStub.callCount).to.be.greaterThan(1);
+    });
   });
 
   describe("queue processing", () => {
     beforeEach(() => {
       dialer = new Dialer({
         libp2p,
-        shardReader: mockShardReader
+        shardReader: mockShardReader,
+        options: mockOptions
       });
       dialer.start();
     });
@@ -334,7 +447,8 @@ describe("Dialer", () => {
     beforeEach(() => {
       dialer = new Dialer({
         libp2p,
-        shardReader: mockShardReader
+        shardReader: mockShardReader,
+        options: mockOptions
       });
       dialer.start();
     });
@@ -368,7 +482,8 @@ describe("Dialer", () => {
     it("should handle complete dial lifecycle", async () => {
       dialer = new Dialer({
         libp2p,
-        shardReader: mockShardReader
+        shardReader: mockShardReader,
+        options: mockOptions
       });
       dialer.start();
 
@@ -386,7 +501,8 @@ describe("Dialer", () => {
     it("should handle multiple peers with different shard configurations", async () => {
       dialer = new Dialer({
         libp2p,
-        shardReader: mockShardReader
+        shardReader: mockShardReader,
+        options: mockOptions
       });
       dialer.start();
 
diff --git a/packages/core/src/lib/connection_manager/dialer.ts b/packages/core/src/lib/connection_manager/dialer.ts
diff --git a/packages/interfaces/src/connection_manager.ts b/packages/interfaces/src/connection_manager.ts
