feat: lighten retry logic for LightPush

packages/interfaces/src/sender.ts

@@ -1,10 +1,23 @@
 import type { IEncoder, IMessage } from "./message.js";
-import { ProtocolUseOptions, SDKProtocolResult } from "./protocols.js";
+import { SDKProtocolResult } from "./protocols.js";
+
+export type ISenderOptions = {
+  /**
+   * Enables retry of a message that was failed to be sent.
+   * @default false
+   */
+  autoRetry?: boolean;
+  /**
+   * Sets number of attempts if `autoRetry` is enabled.
+   * @default 3
+   */
+  maxAttempts?: number;
+};
 
 export interface ISender {
   send: (
     encoder: IEncoder,
     message: IMessage,
-    sendOptions?: ProtocolUseOptions
+    sendOptions?: ISenderOptions
   ) => Promise<SDKProtocolResult>;
 }

packages/sdk/src/protocols/light_push/light_push.spec.ts

@@ -0,0 +1,10 @@
+// TODO: add them after decoupling `BaseProtocolSDK` from LightPush
+describe("LightPush SDK", () => {
+  it("should fail to send if pubsub topics are misconfigured");
+
+  it("should fail to send if no connected peers found");
+
+  it("should send to number of used peers");
+
+  it("should retry on failure if specified");
+});

packages/sdk/src/protocols/light_push/light_push.ts

@@ -6,28 +6,33 @@ import {
   LightPushCore
 } from "@waku/core";
 import {
+  type CoreProtocolResult,
   Failure,
   type IEncoder,
   ILightPush,
   type IMessage,
+  type ISenderOptions,
   type Libp2p,
   type ProtocolCreateOptions,
   ProtocolError,
-  ProtocolUseOptions,
   SDKProtocolResult
 } from "@waku/interfaces";
 import { ensurePubsubTopicIsConfigured, Logger } from "@waku/utils";
 
-import { ReliabilityMonitorManager } from "../../reliability_monitor/index.js";
-import { SenderReliabilityMonitor } from "../../reliability_monitor/sender.js";
 import { BaseProtocolSDK } from "../base_protocol.js";
 
 const log = new Logger("sdk:light-push");
 
-class LightPush extends BaseProtocolSDK implements ILightPush {
-  public readonly protocol: LightPushCore;
+const DEFAULT_MAX_ATTEMPTS = 3;
+const DEFAULT_SEND_OPTIONS: ISenderOptions = {
+  autoRetry: false,
+  maxAttempts: DEFAULT_MAX_ATTEMPTS
+};
+
+type RetryCallback = (peer: Peer) => Promise<CoreProtocolResult>;
 
-  private readonly reliabilityMonitor: SenderReliabilityMonitor;
+export class LightPush extends BaseProtocolSDK implements ILightPush {
+  public readonly protocol: LightPushCore;
 
   public constructor(
     connectionManager: ConnectionManager,
@@ -42,17 +47,13 @@ class LightPush extends BaseProtocolSDK implements ILightPush {
       }
     );
 
-    this.reliabilityMonitor = ReliabilityMonitorManager.createSenderMonitor(
-      this.renewPeer.bind(this)
-    );
-
     this.protocol = this.core as LightPushCore;
   }
 
   public async send(
     encoder: IEncoder,
     message: IMessage,
-    _options?: ProtocolUseOptions
+    options: ISenderOptions = DEFAULT_SEND_OPTIONS
   ): Promise<SDKProtocolResult> {
     const successes: PeerId[] = [];
     const failures: Failure[] = [];
@@ -105,14 +106,10 @@ class LightPush extends BaseProtocolSDK implements ILightPush {
       if (failure) {
         failures.push(failure);
 
-        const connectedPeer = this.connectedPeers.find((connectedPeer) =>
-          connectedPeer.id.equals(failure.peerId)
-        );
-
-        if (connectedPeer) {
-          void this.reliabilityMonitor.attemptRetriesOrRenew(
-            connectedPeer.id,
-            () => this.protocol.send(encoder, message, connectedPeer)
+        if (options?.autoRetry) {
+          void this.attemptRetries(
+            (peer: Peer) => this.protocol.send(encoder, message, peer),
+            options.maxAttempts
           );
         }
       }
@@ -129,6 +126,32 @@ class LightPush extends BaseProtocolSDK implements ILightPush {
     };
   }
 
+  private async attemptRetries(
+    fn: RetryCallback,
+    maxAttempts?: number
+  ): Promise<void> {
+    maxAttempts = maxAttempts || DEFAULT_MAX_ATTEMPTS;
+    const connectedPeers = await this.getConnectedPeers();
+
+    if (connectedPeers.length === 0) {
+      log.warn("Cannot retry with no connected peers.");
+      return;
+    }
+
+    for (let i = 0; i < maxAttempts; i++) {
+      const peer = connectedPeers[i % connectedPeers.length]; // always present as we checked for the length already
+      const response = await fn(peer);
+
+      if (response.success) {
+        return;
+      }
+
+      log.info(
+        `Attempted retry for peer:${peer.id} failed with:${response?.failure?.error}`
+      );
+    }
+  }
+
   private async getConnectedPeers(): Promise<Peer[]> {
     const peerIDs = this.libp2p.getPeers();
 

packages/sdk/src/reliability_monitor/index.ts

@@ -7,14 +7,12 @@ import {
 } from "@waku/interfaces";
 
 import { ReceiverReliabilityMonitor } from "./receiver.js";
-import { SenderReliabilityMonitor } from "./sender.js";
 
 export class ReliabilityMonitorManager {
   private static receiverMonitors: Map<
     PubsubTopic,
     ReceiverReliabilityMonitor
   > = new Map();
-  private static senderMonitor: SenderReliabilityMonitor | undefined;
 
   public static createReceiverMonitor(
     pubsubTopic: PubsubTopic,
@@ -44,30 +42,17 @@ export class ReliabilityMonitorManager {
     return monitor;
   }
 
-  public static createSenderMonitor(
-    renewPeer: (peerId: PeerId) => Promise<Peer | undefined>
-  ): SenderReliabilityMonitor {
-    if (!ReliabilityMonitorManager.senderMonitor) {
-      ReliabilityMonitorManager.senderMonitor = new SenderReliabilityMonitor(
-        renewPeer
-      );
-    }
-    return ReliabilityMonitorManager.senderMonitor;
-  }
-
   private constructor() {}
 
   public static stop(pubsubTopic: PubsubTopic): void {
     this.receiverMonitors.delete(pubsubTopic);
-    this.senderMonitor = undefined;
   }
 
   public static stopAll(): void {
     for (const [pubsubTopic, monitor] of this.receiverMonitors) {
       monitor.setMaxMissedMessagesThreshold(undefined);
       monitor.setMaxPingFailures(undefined);
       this.receiverMonitors.delete(pubsubTopic);
-      this.senderMonitor = undefined;
     }
   }
 }