fix(js/ts/python): settle cancelled delays and finalizers once

src/fable-library-py/fable_library/async_.py

@@ -4,7 +4,7 @@
 from asyncio import Future, ensure_future
 from collections.abc import Awaitable, Callable, Iterable
 from concurrent.futures import ThreadPoolExecutor
-from threading import Timer
+from threading import RLock, Timer
 from typing import (
     Any,
 )
@@ -80,19 +80,70 @@ def is_cancellation_requested(token: CancellationToken) -> bool:
     return token.is_cancelled
 
 
-def sleep(milliseconds_duetime: int | TimeSpan) -> Async[None]:
-    def cont(ctx: IAsyncContext[None]):
-        def cancel():
-            ctx.on_cancel(OperationCanceledError())
+def call_on_owner_loop(loop: asyncio.AbstractEventLoop, action: Callable[[], None]) -> None:
+    # decision: foreign-thread cancellation settles on the owning loop while it is running
+    if loop.is_closed() or not loop.is_running():
+        action()
+        return
+    try:
+        current_loop = asyncio.get_running_loop()
+    except RuntimeError:
+        current_loop = None
+    if current_loop is loop:
+        action()
+        return
+    try:
+        loop.call_soon_threadsafe(action)
+    except RuntimeError:
+        if not loop.is_closed():
+            raise
+        action()
 
-        token_id = ctx.cancel_token.add_listener(cancel)
 
-        def timeout():
-            ctx.cancel_token.remove_listener(token_id)
-            ctx.on_success(None)
+def sleep(milliseconds_duetime: int | TimeSpan) -> Async[None]:
+    def cont(ctx: IAsyncContext[None]) -> None:
+        loop = asyncio.get_running_loop()
+        gate = RLock()
+        completed = False
+        timer: asyncio.TimerHandle | None = None
+        token_id: int | None = None
+
+        def settle(cancelled: bool) -> None:
+            nonlocal completed
+            with gate:
+                if completed:
+                    return
+                cancelled = cancelled or ctx.cancel_token.is_cancelled
+                completed = True
+                if timer is not None:
+                    timer.cancel()
+                listener_id = token_id
+            # Dispose outside the gate: another thread may already be running cancel().
+            if listener_id is not None:
+                ctx.cancel_token.remove_listener(listener_id)
+            if cancelled:
+                ctx.on_cancel(OperationCanceledError())
+            else:
+                ctx.on_success(None)
 
-        due_time = to_milliseconds(milliseconds_duetime) / 1000.0
-        ctx.trampoline.run_later(timeout, due_time)
+        def cancel() -> None:
+            call_on_owner_loop(loop, lambda: settle(True))
+
+        # invariant: installation and settlement cannot interleave before both resources are owned
+        try:
+            with gate:
+                token_id = ctx.cancel_token.add_listener(cancel)
+                if not completed:
+                    due_time = to_milliseconds(milliseconds_duetime) / 1000.0
+                    timer = ctx.trampoline.run_later(lambda: settle(False), due_time)
+                    if completed:
+                        timer.cancel()
+        except Exception:
+            with gate:
+                completed = True
+            if token_id is not None:
+                ctx.cancel_token.remove_listener(token_id)
+            raise
 
     return protected_cont(cont)
 

src/fable-library-py/fable_library/async_builder.py

@@ -245,9 +245,9 @@ def run_later(
         self,
         action: Callable[[], None],
         due_time: float = 0.0,
-    ):
+    ) -> asyncio.TimerHandle:
         loop = asyncio.get_running_loop()
-        loop.call_later(due_time, action)
+        return loop.call_later(due_time, action)
 
     def run(self, action: Callable[[], None]):
         loop = asyncio.get_running_loop()
@@ -345,15 +345,27 @@ def ReturnFrom[T](self, computation: Async[T]) -> Async[T]:
     def TryFinally[T](self, computation: Async[T], compensation: Callable[[], None]) -> Async[T]:
         def cont(ctx: IAsyncContext[T]) -> None:
             def on_success(x: T) -> None:
-                compensation()
+                try:
+                    compensation()
+                except Exception as error:
+                    ctx.on_error(error)
+                    return
                 ctx.on_success(x)
 
             def on_error(x: Exception) -> None:
-                compensation()
+                try:
+                    compensation()
+                except Exception as error:
+                    ctx.on_error(error)
+                    return
                 ctx.on_error(x)
 
             def on_cancel(x: OperationCanceledError) -> None:
-                compensation()
+                # decision: ignores compensation failures during cancellation, matching .NET Async.TryFinally
+                try:
+                    compensation()
+                except Exception:
+                    pass
                 ctx.on_cancel(x)
 
             ctx_ = IAsyncContext.create(ctx.trampoline, ctx.cancel_token, on_success, on_error, on_cancel)

src/fable-library-py/fable_library/mailbox_processor.py

@@ -1,16 +1,19 @@
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Callable
 from queue import SimpleQueue
 from threading import RLock
 from typing import Any
 
-from .async_ import from_continuations, start_immediate
+from .async_ import call_on_owner_loop, from_continuations, start_immediate
 from .async_builder import (
     Async,
     CancellationToken,
     Continuations,
+    IAsyncContext,
     OperationCanceledError,
+    protected_cont,
 )
 
 
@@ -101,34 +104,44 @@ def receive(self) -> Async[Msg]:
         Returns:
             An asynchronous computation which will consume the
             first message in arrival order. No thread is blocked while
-            waiting for further messages. Raises a TimeoutException if
-            the timeout is exceeded.
+            waiting for further messages. Cancelling the token of the
+            receiving computation cancels a pending receive.
         """
 
-        def callback(conts: Continuations[Any]):
-            if self.continuation:
-                raise Exception("Receive can only be called once!")
-
-            self.continuation = conts
-
+        def cont(ctx: IAsyncContext[Msg]) -> None:
+            loop = asyncio.get_running_loop()
+            token_id: int | None = None
+
+            def on_message(msg: Msg) -> None:
+                if token_id is not None:
+                    ctx.cancel_token.remove_listener(token_id)
+                ctx.on_success(msg)
+
+            continuation: Continuations[Any] = (on_message, ctx.on_error, ctx.on_cancel)
+
+            def on_cancel() -> None:
+                with self.lock:
+                    if self.continuation is not continuation:
+                        return
+                    self.continuation = None
+                call_on_owner_loop(loop, lambda: ctx.on_cancel(OperationCanceledError()))
+
+            with self.lock:
+                if self.continuation:
+                    raise Exception("Receive can only be called once!")
+                self.continuation = continuation
+
+            token_id = ctx.cancel_token.add_listener(on_cancel)
+            with self.lock:
+                if self.continuation is not continuation:
+                    return
             self.__process_events()
 
-        return from_continuations(callback)
+        return protected_cont(cont)
 
     def __process_events(self) -> None:
-        if not self.continuation:
-            return
-
-        # Cancellation of async workflows is more tricky in Python than
-        # with F# so we check the cancellation token for each process.
-        if self.token.is_cancellation_requested:
-            self.continuation, cont = None, self.continuation
-            if cont:
-                cont[2](OperationCanceledError("Mailbox was cancelled"))
-            return
-
         with self.lock:
-            if self.messages.empty():
+            if not self.continuation or self.messages.empty():
                 return
             msg = self.messages.get()
             self.continuation, cont = None, self.continuation
@@ -143,7 +156,7 @@ def start(
         cancellation_token: CancellationToken | None = None,
     ) -> MailboxProcessor[Msg]:
         mbox: MailboxProcessor[Msg] = MailboxProcessor(body, cancellation_token)
-        start_immediate(body(mbox))
+        start_immediate(body(mbox), mbox.token)
         return mbox
 
 
@@ -163,7 +176,7 @@ def post_and_async_reply[Msg, Reply](
 
 def start_instance(mbox: MailboxProcessor[Any]) -> None:
     body = mbox.body(mbox)
-    return start_immediate(body)
+    return start_immediate(body, mbox.token)
 
 
 def start[Msg](

src/fable-library-ts/Async.ts

@@ -159,15 +159,18 @@ export function sequential<T>(computations: Iterable<Async<T>>) {
 
 export function sleep(millisecondsDueTime: number) {
   return protectedCont((ctx: IAsyncContext<void>) => {
-    let tokenId: number;
-    const timeoutId = setTimeout(() => {
-      ctx.cancelToken.removeListener(tokenId);
-      ctx.onSuccess(void 0);
-    }, millisecondsDueTime);
-    tokenId = ctx.cancelToken.addListener(() => {
+    let completed = false;
+    let tokenId: number | undefined;
+    const settle = (cancelled: boolean) => {
+      if (completed) { return; }
+      completed = true;
       clearTimeout(timeoutId);
-      ctx.onCancel(new OperationCanceledException());
-    });
+      if (tokenId !== undefined) { ctx.cancelToken.removeListener(tokenId); }
+      if (cancelled) { ctx.onCancel(new OperationCanceledException()); }
+      else { ctx.onSuccess(void 0); }
+    };
+    const timeoutId = setTimeout(() => settle(false), millisecondsDueTime);
+    tokenId = ctx.cancelToken.addListener(() => settle(true));
   });
 }
 

src/fable-library-ts/AsyncBuilder.ts

@@ -203,15 +203,22 @@ export class AsyncBuilder {
     return protectedCont((ctx: IAsyncContext<T>) => {
       computation({
         onSuccess: (x: T) => {
-          compensation();
+          try { compensation(); } catch (error) {
+            ctx.onError(ensureErrorOrException(error));
+            return;
+          }
           ctx.onSuccess(x);
         },
         onError: (x: any) => {
-          compensation();
+          try { compensation(); } catch (error) {
+            ctx.onError(ensureErrorOrException(error));
+            return;
+          }
           ctx.onError(x);
         },
         onCancel: (x: any) => {
-          compensation();
+          // decision: cancellation remains terminal when compensation throws, matching .NET Async.TryFinally
+          try { compensation(); } catch { /* Cancellation takes precedence. */ }
           ctx.onCancel(x);
         },
         cancelToken: ctx.cancelToken,

src/fable-library-ts/MailboxProcessor.ts

@@ -1,7 +1,7 @@
 import { defaultCancellationToken } from "./Async.ts";
 import { fromContinuations } from "./Async.ts";
 import { startImmediate } from "./Async.ts";
-import { Async } from "./AsyncBuilder.ts";
+import { Async, IAsyncContext, OperationCanceledException, protectedCont } from "./AsyncBuilder.ts";
 import { Continuation, Continuations } from "./AsyncBuilder.ts";
 import { CancellationToken } from "./AsyncBuilder.ts";
 import { Exception } from "./Util.ts";
@@ -80,12 +80,25 @@ export function startInstance<Msg>($this: MailboxProcessor<Msg>) {
 }
 
 export function receive<Msg>($this: MailboxProcessor<Msg>) {
-  return fromContinuations((conts: Continuations<Msg>) => {
+  return protectedCont((ctx: IAsyncContext<Msg>) => {
     if ($this.continuation) {
       throw new Exception("Receive can only be called once!");
     }
-    $this.continuation = conts[0];
-    __processEvents($this);
+    let tokenId: number | undefined;
+    const continuation = (message: Msg) => {
+      if (tokenId !== undefined) { ctx.cancelToken.removeListener(tokenId); }
+      ctx.onSuccess(message);
+    };
+    $this.continuation = continuation;
+    tokenId = ctx.cancelToken.addListener(() => {
+      if ($this.continuation === continuation) {
+        delete $this.continuation;
+        ctx.onCancel(new OperationCanceledException());
+      }
+    });
+    if ($this.continuation === continuation) {
+      __processEvents($this);
+    }
   });
 }