Passing signal to metadata

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

src/actors/runtime.test.ts

@@ -34,7 +34,8 @@ class Hello {
 class Counter {
   private count: number;
   private watchTarget = new WatchTarget<number>();
-  public metadata?: { extraSum: number };
+  public metadata?: { extraSum: number; signal?: AbortSignal };
+  private listenerAttached = Promise.withResolvers<void>();
 
   constructor(protected state: ActorState) {
     this.count = 0;
@@ -43,6 +44,17 @@ class Counter {
     });
   }
 
+  waitForAbortSignalAttach() {
+    return this.listenerAttached.promise;
+  }
+  shouldFailWhenSignalIsAborted() {
+    const p = Promise.withResolvers<string>();
+    this.metadata?.signal?.addEventListener("abort", () => {
+      p.resolve("HELLO ITS RESOLVED!");
+    }, { once: true });
+    this.listenerAttached.resolve();
+    return p.promise;
+  }
   callSayHello(): Promise<string> {
     const hello = this.state.stub(Hello).id(this.state.id);
     return hello.sayHello();
@@ -105,16 +117,37 @@ const runServer = (
   };
 };
 
-Deno.test("counter tests", async () => {
+Deno.test("actor tests", async () => {
   const rt = new StdActorRuntime();
   let reqCount = 0;
-  await using _server = runServer(rt, () => {
+
+  let nextRequest = Promise.withResolvers<() => void>();
+
+  let abortRequestFn = nextRequest.promise;
+  await using _server = runServer(rt, (rq) => {
     reqCount++;
+    nextRequest.resolve(() => {
+      rq.signal.dispatchEvent(new Event("abort"));
+    });
+    nextRequest = Promise.withResolvers<() => void>();
+    abortRequestFn = nextRequest.promise;
   });
   const actorId = "1234";
   const counterStub = actors.stub(Counter, { maxWsChunkSize: 64 });
 
   const counterActor = counterStub.id(actorId);
+
+  const waitForAttach = counterActor.waitForAbortSignalAttach().then((p) => p);
+  await nextRequest.promise;
+  const shouldReturnHelloOnlyWhenAborted = counterActor
+    .shouldFailWhenSignalIsAborted().then(
+      (p) => p,
+    );
+  const abort = await abortRequestFn;
+  await waitForAttach;
+  abort();
+  assertEquals(await shouldReturnHelloOnlyWhenAborted, "HELLO ITS RESOLVED!");
+
   using rpcActor = counterStub.id("12345").rpc();
 
   const helloUploadActor = actors.stub(Hello).id("123456");

src/actors/silo.ts

@@ -5,6 +5,7 @@ import { ActorState } from "./state.ts";
 import type { ActorStorage } from "./storage.ts";
 import {
   type ActorInvoker,
+  type BaseMetadata,
   create,
   createHttpInvoker,
   type EnrichMetadataFn,
@@ -101,6 +102,9 @@ export class ActorSilo<TEnv extends object = object> {
       )
       : metadata;
 
+    if (metadata && typeof metadata === "object") {
+      (metadata as BaseMetadata).signal = req?.signal;
+    }
     if (isWellKnownRPCMethod(String(method))) {
       const chan = rpc(this.invoker, metadata);
       return chan;

src/actors/stubutil.ts

@@ -220,6 +220,10 @@ export type EnrichMetadataFn<
   metadata: TMetadata,
   req: Request,
 ) => EnrichedMetadata;
+
+export interface BaseMetadata {
+  signal?: AbortSignal;
+}
 /**
  * Represents an actor proxy.
  */
@@ -233,11 +237,15 @@ export type ActorProxy<Actor> =
   & (Actor extends { metadata?: infer TMetadata } ? Actor extends {
       enrichMetadata: EnrichMetadataFn<infer TPartialMetadata, any>;
     } ? {
-        withMetadata(metadata: TPartialMetadata): ActorProxy<Actor>;
+        withMetadata(
+          metadata: Omit<TPartialMetadata, keyof BaseMetadata>,
+        ): ActorProxy<Actor>;
         rpc(): ActorProxy<Actor> & Disposable;
       }
     : {
-      withMetadata(metadata: TMetadata): ActorProxy<Actor>;
+      withMetadata(
+        metadata: Omit<TMetadata, keyof BaseMetadata>,
+      ): ActorProxy<Actor>;
       rpc(): ActorProxy<Actor> & Disposable;
     }
     : { rpc(): ActorProxy<Actor> & Disposable });