Fix disconnection allow messages to ne cleanedup

Signed-off-by: Marcos Candeia <[email protected]>

src/actors/proxyutil.ts

@@ -66,6 +66,7 @@ export class ActorAwaiter<
   DuplexChannel<any, any> {
   ch: Promise<TChannel> | null = null;
   ctrl: AbortController;
+  _disconnected: PromiseWithResolvers<void> = Promise.withResolvers();
   constructor(
     protected actorName: string,
     protected actorMethod: string,
@@ -138,6 +139,9 @@ export class ActorAwaiter<
       };
       await Promise.all([recvLoop(), sendLoop()]);
       if (ch.signal.aborted && !reliableCh.signal.aborted) {
+        const prev = this._disconnected;
+        this._disconnected = Promise.withResolvers();
+        prev.resolve();
         console.error("channel closed, retrying...");
         await new Promise((resolve) => setTimeout(resolve, 2e3)); // retrying in 2 second
         return nextConnection();
@@ -151,6 +155,10 @@ export class ActorAwaiter<
     return this.ch;
   }
 
+  get disconnected() {
+    return this._disconnected.promise;
+  }
+
   async send(value: unknown): Promise<void> {
     const ch = await this.channel;
     await ch.send(value);
@@ -312,12 +320,11 @@ export const createRPCInvoker = <
       const id = crypto.randomUUID();
       const response = Promise.withResolvers<TResponse>();
       pendingRequests.set(id, { response });
-      channel.closed.finally(() => {
-        const resolver = pendingRequests.get(id);
-        if (resolver) {
-          resolver.response.reject(new Error("Channel closed"));
-        }
-      });
+      const cleanup = () => {
+        response.reject(new Error("Channel closed"));
+      };
+      channel.closed.finally(cleanup);
+      channel.disconnected?.finally(cleanup);
 
       try {
         await channel.send({

src/actors/util/channels/channel.ts

@@ -150,6 +150,7 @@ export interface DuplexChannel<TSend, TReceive> extends Disposable {
   close: () => void | Promise<void>;
   closed: Promise<void>;
   signal: AbortSignal;
+  disconnected?: Promise<void>; // used when the channel allows reconnections
 }
 
 /**