fix(js/ts/python): synchronize cancellation registrations

src/Fable.Build/Test/Python.fs

@@ -23,6 +23,10 @@ let handle (args: string list) =
 
     Directory.clean buildDir
 
+    // Native runtime tests use controlled timers and real threads alongside generated F# regressions.
+    for testFile in Directory.GetFiles(sourceDir, "test_*.py") do
+        File.Copy(testFile, Path.Combine(buildDir, Path.GetFileName(testFile)), overwrite = true)
+
     Command.Run("uv", "sync")
 
     // Install local fable-library as editable package for testing

src/Fable.Transforms/Python/Replacements.fs

@@ -2851,7 +2851,10 @@ let exceptions (com: ICompiler) (ctx: Context) r t (i: CallInfo) (thisArg: Expr
             Helper.ConstructorCall(e, t, args, ?loc = r) |> Some
     | "get_Message", Some e -> Helper.GlobalCall("str", t, [ thisArg.Value ], ?loc = r) |> Some
     | "get_StackTrace", Some e -> getFieldWith r t e "stack" |> Some
-    | "get_InnerException", Some e -> getFieldWith r t e "inner_exception" |> Some
+    | "get_InnerException", Some e ->
+        // Caught exceptions are typed as Python's built-in Exception, which lacks this attribute.
+        Helper.GlobalCall("getattr", t, [ e; makeStrConst "inner_exception" ], ?loc = r)
+        |> Some
     | _ -> None
 
 let unchecked (com: ICompiler) (ctx: Context) r t (i: CallInfo) (_: Expr option) (args: Expr list) =

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

@@ -58,7 +58,7 @@ def cont(ctx: IAsyncContext[T]):
 def create_cancellation_token(arg: int | bool | None = None) -> CancellationToken:
     cancelled = arg if isinstance(arg, bool) else False
     token = CancellationToken(cancelled)
-    if isinstance(arg, int):
+    if isinstance(arg, int) and not isinstance(arg, bool):
         timer = Timer(arg / 1000.0, token.cancel)
         timer.start()
 
@@ -112,6 +112,8 @@ def cancel() -> None:
             ctx.on_cancel(OperationCanceledError())
 
         token_id[0] = ctx.cancel_token.add_listener(cancel)
+        if ctx.cancel_token.is_cancelled:
+            return
         event.AddHandler(handler)
 
     return protected_cont(cont)

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

@@ -4,16 +4,15 @@
 from abc import abstractmethod
 from collections.abc import Callable
 from dataclasses import dataclass, field
-from threading import Lock, RLock
+from threading import Condition, Lock, RLock, get_ident
 from typing import (
     Any,
     Literal,
-    Protocol,
     overload,
 )
 
 from .protocols import IEnumerable_1
-from .util import Disposable, IDisposable
+from .util import IDisposable
 
 
 class OperationCanceledError(Exception):
@@ -28,58 +27,105 @@ def __init__(self, msg: str | None = None) -> None:
 ]
 
 
-class _Listener(Protocol):
-    def __call__(self, __state: Any | None = None) -> None: ...
+_NO_STATE = object()
+
+
+class CancellationCallbackError(ExceptionGroup[Exception]):
+    @property
+    def inner_exception(self) -> Exception:
+        # invariant: F# InnerException exposes the first failure without discarding the remaining errors
+        return self.exceptions[0]
 
 
 class CancellationToken:
-    __slots__ = "cancelled", "idx", "listeners", "lock"
+    __slots__ = "_condition", "_running", "cancelled", "idx", "listeners", "lock"
 
-    def __init__(self, cancelled: bool = False):
+    def __init__(self, cancelled: bool = False) -> None:
         self.cancelled = cancelled
         self.listeners: dict[int, Callable[[], None]] = {}
         self.idx = 0
         self.lock = RLock()
+        self._condition = Condition(self.lock)
+        self._running: dict[int, int] = {}
 
     @property
-    def is_cancelled(self):
-        return self.cancelled
+    def is_cancelled(self) -> bool:
+        with self.lock:
+            return self.cancelled
 
     is_cancellation_requested = is_cancelled
 
     def cancel(self) -> None:
-        cancel = False
         with self.lock:
-            if not self.cancelled:
-                cancel = True
-                self.cancelled = True
+            if self.cancelled:
+                return
+            self.cancelled = True
+            # decision: snapshots IDs newest first to match .NET while allowing disposal of pending callbacks
+            pending = list(reversed(self.listeners))
+
+        errors: list[Exception] = []
+        for listener_id in pending:
+            try:
+                self._invoke_listener(listener_id)
+            except Exception as error:
+                errors.append(error)
+        if errors:
+            raise CancellationCallbackError("Cancellation callbacks failed", errors)
 
-        if cancel:
-            for listener in self.listeners.values():
-                listener()
+    def _invoke_listener(self, listener_id: int) -> None:
+        with self.lock:
+            listener = self.listeners.pop(listener_id, None)
+            if listener is None:
+                return
+            self._running[listener_id] = get_ident()
+        # invariant: user callbacks execute outside the registry lock
+        try:
+            listener()
+        finally:
+            with self._condition:
+                del self._running[listener_id]
+                self._condition.notify_all()
 
     def add_listener(self, f: Callable[[], None]) -> int:
         with self.lock:
-            id = self.idx
-            self.idx = self.idx + 1
-            self.listeners[id] = f
-
-        return id
+            listener_id = self.idx
+            self.idx += 1
+            if not self.cancelled:
+                self.listeners[listener_id] = f
+                return listener_id
+        # .NET invokes late registrations synchronously and does not retain them.
+        f()
+        return listener_id
 
     def remove_listener(self, id: int) -> None:
-        with self.lock:
-            del self.listeners[id]
+        with self._condition:
+            self.listeners.pop(id, None)
+            # invariant: disposal waits for a running callback except when that callback disposes itself
+            while id in self._running and self._running[id] != get_ident():
+                self._condition.wait()
+
+    @overload
+    def register(self, f: Callable[[], None]) -> IDisposable: ...
+
+    @overload
+    def register[S](self, f: Callable[[S], None], state: S) -> IDisposable: ...
+
+    def register(self, f: Callable[..., None], state: object = _NO_STATE) -> IDisposable:
+        listener_id = self.add_listener(f if state is _NO_STATE else lambda: f(state))
+        return _CancellationRegistration(self, listener_id)
+
+    def Dispose(self) -> None:
+        # Source disposal is currently a no-op, as in the compiler replacement and JS runtime.
+        pass
 
-    def register(self, f: _Listener, state: Any | None = None) -> None:
-        if state:
-            id = self.add_listener(lambda: f(state))
-        else:
-            id = self.add_listener(f)
 
-        def dispose():
-            self.remove_listener(id)
+class _CancellationRegistration:
+    def __init__(self, token: CancellationToken, listener_id: int) -> None:
+        self._token = token
+        self._listener_id = listener_id
 
-        Disposable.create(dispose)
+    def Dispose(self) -> None:
+        self._token.remove_listener(self._listener_id)
 
 
 class IAsyncContext[T]:
@@ -217,6 +263,7 @@ def protected_cont[T](f: Async[T]) -> Async[T]:
     def _protected_cont(ctx: IAsyncContext[T]):
         if ctx.cancel_token and ctx.cancel_token.is_cancelled:
             ctx.on_cancel(OperationCanceledError())
+            return
 
         def fn():
             try:

src/fable-library-ts/Async.ts

@@ -104,6 +104,7 @@ export function awaitEvent<Del extends Function, T>(event: IEvent$2<Del, T>, can
       if (cancelAction != null) { cancelAction(); }
       ctx.onCancel(new OperationCanceledException());
     });
+    if (ctx.cancelToken.isCancelled) { return; }
     event.AddHandler(handler);
   });
 }

src/fable-library-ts/AsyncBuilder.ts

@@ -12,6 +12,16 @@ export type Continuations<T> = [
   Continuation<OperationCanceledException>
 ];
 
+// AggregateError requires ES2021; the runtime also supports ES2020 hosts.
+export class CancellationCallbackError extends Exception {
+  public readonly errors: unknown[];
+  constructor(errors: unknown[]) {
+    // invariant: F# InnerException exposes the first failure without discarding the remaining errors
+    super("Cancellation callbacks failed", errors[0] as Exception);
+    this.errors = errors;
+  }
+}
+
 export class CancellationToken implements IDisposable {
   private _id: number;
   private _cancelled: boolean;
@@ -27,14 +37,23 @@ export class CancellationToken implements IDisposable {
   public cancel() {
     if (!this._cancelled) {
       this._cancelled = true;
-      for (const [, listener] of this._listeners) {
-        listener();
+      // decision: visit newest registrations first, matching .NET; disposal can skip pending callbacks
+      const pending = Array.from(this._listeners.keys()).reverse();
+      const errors: unknown[] = [];
+      for (const id of pending) {
+        const listener = this._listeners.get(id);
+        this._listeners.delete(id);
+        if (listener) {
+          try { listener(); } catch (error) { errors.push(error); }
+        }
       }
+      if (errors.length > 0) { throw new CancellationCallbackError(errors); }
     }
   }
   public addListener(f: () => void) {
     const id = this._id;
-    this._listeners.set(this._id++, f);
+    this._id++;
+    if (this._cancelled) { f(); } else { this._listeners.set(id, f); }
     return id;
   }
   public removeListener(id: number) {

tests/Js/Main/AsyncTests.fs

@@ -6,6 +6,13 @@ open Util.Testing
 
 #if FABLE_COMPILER
 open Fable.Core
+
+type private Deferred =
+    [<ImportMember("./js/async-deferred.js")>]
+    static member startDeferred(computation: Async<int>, token: System.Threading.CancellationToken, onSuccess: int -> unit, onError: exn -> unit, onCancel: exn -> unit): unit -> unit = jsNative
+
+    [<ImportMember("./js/async-deferred.js")>]
+    static member handlerCount(event: Event<int>): int = jsNative
 #endif
 
 type DisposableAction(f) =
@@ -44,6 +51,30 @@ let tests =
         async { return () }
         |> Async.StartImmediate
 
+    testCase "Cancellation callback errors expose the first InnerException" <| fun () ->
+        for count in [1; 2] do
+            use cts = new System.Threading.CancellationTokenSource()
+            let errors = Array.init count (fun i -> exn (string i))
+            let mutable calls = 0
+            errors |> Array.iter (fun error ->
+                cts.Token.Register(fun () ->
+                    calls <- calls + 1
+                    raise error) |> ignore)
+            let caught =
+                try
+                    cts.Cancel()
+                    None
+                with ex -> Some ex
+            match caught with
+            | Some ex ->
+                equal errors[count - 1].Message ex.InnerException.Message
+                equal true (obj.ReferenceEquals(errors[count - 1], ex.InnerException))
+            | None -> failwith "Expected cancellation callback errors"
+            equal count calls
+            equal true cts.IsCancellationRequested
+            cts.Cancel()
+            equal count calls
+
     testCase "Async while binding works correctly" <| fun () ->
         let mutable result = 0
         async {
@@ -684,6 +715,21 @@ let tests =
         cts.Cancel()
         async { equal true cancelCalled }
 
+#if FABLE_COMPILER
+    testCase "Async.AwaitEvent cancelled before its body runs does not subscribe" <| fun () ->
+        let ev = Event<int>()
+        let cts = new System.Threading.CancellationTokenSource()
+        let calls = ResizeArray<string>()
+        let resume =
+            Deferred.startDeferred(Async.AwaitEvent ev.Publish, cts.Token,
+                (fun v -> calls.Add $"success {v}"), (fun _ -> calls.Add "error"), (fun _ -> calls.Add "cancel"))
+        cts.Cancel()
+        resume ()
+        ev.Trigger(42)
+        equal [ "cancel" ] (List.ofSeq calls)
+        equal 0 (Deferred.handlerCount ev)
+#endif
+
     testCase "Async try .. with does not run 'with' branch when body succeeds" <| fun () ->
         let work = async {
             try

tests/Js/Main/js/async-deferred.js

@@ -0,0 +1,15 @@
+export function startDeferred(computation, token, onSuccess, onError, onCancel) {
+    let body;
+    computation({
+        onSuccess,
+        onError,
+        onCancel,
+        cancelToken: token,
+        trampoline: { completed: false, incrementAndCheck: () => true, hijack: f => { body = f; } },
+    });
+    return () => body();
+}
+
+export function handlerCount(event) {
+    return event.delegates.length;
+}

tests/Python/TestAsync.fs

@@ -32,6 +32,31 @@ let ``test Simple async translates without exception`` () =
     async { return () }
     |> Async.StartImmediate
 
+[<Fact>]
+let ``test Cancellation callback errors expose the first InnerException`` () =
+    for count in [1; 2] do
+        use cts = new System.Threading.CancellationTokenSource()
+        let errors = Array.init count (fun i -> exn (string i))
+        let mutable calls = 0
+        errors |> Array.iter (fun error ->
+            cts.Token.Register(fun () ->
+                calls <- calls + 1
+                raise error) |> ignore)
+        let caught =
+            try
+                cts.Cancel()
+                None
+            with ex -> Some ex
+        match caught with
+        | Some ex ->
+            equal errors[count - 1].Message ex.InnerException.Message
+            equal true (obj.ReferenceEquals(errors[count - 1], ex.InnerException))
+        | None -> failwith "Expected cancellation callback errors"
+        equal count calls
+        equal true cts.IsCancellationRequested
+        cts.Cancel()
+        equal count calls
+
 
 [<Fact>]
 let ``test Async while binding works correctly`` () =