feat: add retries to lightpush

Author: danisharora099

packages/sdk/src/protocols/sender/lightpush/index.ts

@@ -97,7 +97,17 @@ class LightPushSDK extends BaseProtocolSDK implements ILightPushSDK {
         }
         if (failure) {
           failures.push(failure);
-          void this.reliabilityMonitor.handleSendFailure(failure);
+
+          if (failure.peerId) {
+            const peer = this.connectedPeers.find((connectedPeer) =>
+              connectedPeer.id.equals(failure.peerId)
+            );
+            if (peer) {
+              void this.reliabilityMonitor.attemptRetries(failure.peerId, () =>
+                this.protocol.send(encoder, message, peer)
+              );
+            }
+          }
         }
       } else {
         log.error("Failed unexpectedly while sending:", result.reason);

packages/sdk/src/protocols/sender/reliability_monitor.ts

@@ -1,5 +1,5 @@
 import type { Peer, PeerId } from "@libp2p/interface";
-import { Failure, PeerIdStr } from "@waku/interfaces";
+import { CoreProtocolResult, PeerIdStr } from "@waku/interfaces";
 import { Logger } from "@waku/utils";
 
 const log = new Logger("sdk:sender:reliability_monitor");
@@ -13,27 +13,45 @@ export class SenderReliabilityMonitor {
 
   public constructor(private renewPeer: (peerId: PeerId) => Promise<Peer>) {}
 
-  public handleSendFailure = async (failure: Failure): Promise<void> => {
-    if (failure.peerId) {
-      const peerIdStr = failure.peerId.toString();
-      const currentAttempts = this.attempts.get(peerIdStr) || 0;
-      this.attempts.set(peerIdStr, currentAttempts + 1);
-
-      if (currentAttempts + 1 >= this.maxAttemptsBeforeRenewal) {
-        try {
-          const newPeer = await this.renewPeer(failure.peerId);
-          log.info(
-            `Renewed peer ${failure.peerId.toString()} to ${newPeer.id.toString()}`
-          );
+  public async attemptRetries(
+    peerId: PeerId,
+    protocolSend: () => Promise<CoreProtocolResult>
+  ): Promise<void> {
+    const peerIdStr = peerId.toString();
+    const currentAttempts = this.attempts.get(peerIdStr) || 0;
+    this.attempts.set(peerIdStr, currentAttempts + 1);
 
+    if (currentAttempts + 1 < this.maxAttemptsBeforeRenewal) {
+      try {
+        const result = await protocolSend();
+        if (result.success) {
+          log.info(`Successfully sent message after retry to ${peerIdStr}`);
           this.attempts.delete(peerIdStr);
-          this.attempts.set(newPeer.id.toString(), 0);
-        } catch (error) {
+        } else {
           log.error(
-            `Failed to renew peer ${failure.peerId.toString()}: ${error}`
+            `Failed to send message after retry to ${peerIdStr}: ${result.failure}`
           );
+          await this.attemptRetries(peerId, protocolSend);
         }
+      } catch (error) {
+        log.error(
+          `Failed to send message after retry to ${peerIdStr}: ${error}`
+        );
+        await this.attemptRetries(peerId, protocolSend);
+      }
+    } else {
+      try {
+        const newPeer = await this.renewPeer(peerId);
+        log.info(
+          `Renewed peer ${peerId.toString()} to ${newPeer.id.toString()}`
+        );
+
+        this.attempts.delete(peerIdStr);
+        this.attempts.set(newPeer.id.toString(), 0);
+        await protocolSend();
+      } catch (error) {
+        log.error(`Failed to renew peer ${peerId.toString()}: ${error}`);
       }
     }
-  };
+  }
 }