Fixes abort usage

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

src/actors/util/watch.ts

@@ -1,4 +1,5 @@
 export type { ChannelUpgrader } from "./channels/channel.ts";
+
 /**
  * Watches events and returns async iterators for the events.
  */
@@ -9,25 +10,38 @@ export class WatchTarget<T> {
   subscribe(signal?: AbortSignal): AsyncIterableIterator<T> {
     const subscriptionId = crypto.randomUUID();
     const queue: Array<(value: IteratorResult<T>) => void> = [];
+    let isCancelled = false;
 
     const pushQueue = (value: IteratorResult<T>) => {
       queue.forEach((resolve) => resolve(value));
     };
 
     const nextPromise = () =>
       new Promise<IteratorResult<T>>((resolve) => {
-        queue.push((v) => {
-          if (signal?.aborted) {
-            return resolve({ value: undefined, done: true });
-          }
-          resolve(v);
-        });
+        // If already cancelled, resolve immediately
+        if (isCancelled || signal?.aborted) {
+          return resolve({ value: undefined, done: true });
+        }
+
+        // Push a new handler to the queue to resolve when a value is emitted
+        queue.push((v) => resolve(v));
+
+        // Listen for the abort signal and resolve immediately if it happens
+        signal?.addEventListener(
+          "abort",
+          () => {
+            isCancelled = true;
+            resolve({ value: undefined, done: true });
+          },
+          { once: true }, // Only fire once on the first abort
+        );
       });
 
     const iterator: AsyncIterableIterator<T> = {
       next: () => nextPromise(),
       return: () => {
         // Clean up the subscription when the consumer is done
+        isCancelled = true;
         delete this.subscribers[subscriptionId];
         return Promise.resolve({ value: undefined, done: true });
       },
@@ -36,11 +50,19 @@ export class WatchTarget<T> {
       },
     };
 
+    // If the signal is already aborted at the time of subscription, return immediately
+    if (signal?.aborted) {
+      iterator?.return?.(); // Clean up immediately
+      return iterator;
+    }
+
     this.subscribers[subscriptionId] = (value: T) => {
       pushQueue({ value, done: false });
     };
+
+    // Handle the signal being aborted after subscription
     signal?.addEventListener("abort", () => {
-      iterator?.return?.();
+      iterator?.return?.(); // Immediately cancel the iterator
     });
 
     return iterator;