From 0f493ac7e54c5a4e7139a17a791069c1addbdbd2 Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Tue, 13 Oct 2026 23:46:43 +0200 Subject: [PATCH 1/4] fix(async): install cleanup before cancellation checkpoints --- .../fable_library/async_builder.py | 24 +++- src/fable-library-ts/AsyncBuilder.ts | 19 ++- tests/Js/Main/AsyncTests.fs | 103 +++++++++++++++ tests/Python/TestAsync.fs | 117 ++++++++++++++++++ tests/Python/test_async_runtime.py | 39 ++++++ 5 files changed, 295 insertions(+), 7 deletions(-) diff --git a/src/fable-library-py/fable_library/async_builder.py b/src/fable-library-py/fable_library/async_builder.py index c3eef16f2d..a0dfcc04f6 100644 --- a/src/fable-library-py/fable_library/async_builder.py +++ b/src/fable-library-py/fable_library/async_builder.py @@ -350,7 +350,13 @@ 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: + settled = False + def on_success(x: T) -> None: + nonlocal settled + if settled: + return + settled = True try: compensation() except Exception as error: @@ -359,6 +365,11 @@ def on_success(x: T) -> None: ctx.on_success(x) def on_error(x: Exception) -> None: + nonlocal settled + # invariant: downstream continuation exceptions never rerun compensation + if settled: + raise x + settled = True try: compensation() except Exception as error: @@ -367,6 +378,10 @@ def on_error(x: Exception) -> None: ctx.on_error(x) def on_cancel(x: OperationCanceledError) -> None: + nonlocal settled + if settled: + return + settled = True # decision: ignores compensation failures during cancellation, matching .NET Async.TryFinally try: compensation() @@ -375,9 +390,11 @@ def on_cancel(x: OperationCanceledError) -> None: ctx.on_cancel(x) ctx_ = IAsyncContext.create(ctx.trampoline, ctx.cancel_token, on_success, on_error, on_cancel) - computation(ctx_) + # invariant: cleanup continuations are installed before the cancellation checkpoint + # decision: protect the body with the same trampoline to retain bounded stack growth + protected_cont(computation)(ctx_) - return protected_cont(cont) + return cont def TryWith[T](self, computation: Async[T], catch_handler: Callable[[Exception], Async[T]]) -> Async[T]: def fn(ctx: IAsyncContext[T]): @@ -403,7 +420,8 @@ def Using[D: IDisposable, U](self, resource: D, binder: Callable[[D], Async[U]]) def compensation() -> None: return resource.Dispose() - return self.TryFinally(binder(resource), compensation) + # decision: defer the resource binder until cleanup is installed and cancellation is checked + return self.TryFinally(self.Delay(lambda: binder(resource)), compensation) @overload def While(self, guard: Callable[[], bool], computation: Async[Literal[None]]) -> Async[None]: ... diff --git a/src/fable-library-ts/AsyncBuilder.ts b/src/fable-library-ts/AsyncBuilder.ts index 2a89be285e..a301f97b61 100644 --- a/src/fable-library-ts/AsyncBuilder.ts +++ b/src/fable-library-ts/AsyncBuilder.ts @@ -200,9 +200,14 @@ export class AsyncBuilder { } public TryFinally(computation: Async, compensation: () => void) { - return protectedCont((ctx: IAsyncContext) => { - computation({ + // invariant: cleanup continuations are installed before the cancellation checkpoint + // decision: protect the body with the shared trampoline so pre-cancellation still runs compensation + return (ctx: IAsyncContext) => { + let settled = false; + protectedCont(computation)({ onSuccess: (x: T) => { + if (settled) { return; } + settled = true; try { compensation(); } catch (error) { ctx.onError(ensureErrorOrException(error)); return; @@ -210,6 +215,9 @@ export class AsyncBuilder { ctx.onSuccess(x); }, onError: (x: any) => { + // invariant: downstream continuation exceptions never rerun compensation + if (settled) { throw x; } + settled = true; try { compensation(); } catch (error) { ctx.onError(ensureErrorOrException(error)); return; @@ -217,6 +225,8 @@ export class AsyncBuilder { ctx.onError(x); }, onCancel: (x: any) => { + if (settled) { return; } + settled = true; // decision: cancellation remains terminal when compensation throws, matching .NET Async.TryFinally try { compensation(); } catch { /* Cancellation takes precedence. */ } ctx.onCancel(x); @@ -224,7 +234,7 @@ export class AsyncBuilder { cancelToken: ctx.cancelToken, trampoline: ctx.trampoline, }); - }); + }; } public TryWith(computation: Async, catchHandler: (e: any) => Async) { @@ -250,7 +260,8 @@ export class AsyncBuilder { } public Using(resource: T, binder: (x: T) => Async) { - return this.TryFinally(binder(resource), () => resource.Dispose()); + // decision: defer the resource binder until cleanup is installed and cancellation is checked + return this.TryFinally(this.Delay(() => binder(resource)), () => resource.Dispose()); } public While(guard: () => boolean, computation: Async): Async { diff --git a/tests/Js/Main/AsyncTests.fs b/tests/Js/Main/AsyncTests.fs index 362207952e..4d7ffc6e92 100644 --- a/tests/Js/Main/AsyncTests.fs +++ b/tests/Js/Main/AsyncTests.fs @@ -4,6 +4,103 @@ open System open Util open Util.Testing +module private FinalizerRegressions = + open System + open System.Threading + + let private check expected actual = + if expected <> actual then failwithf "Expected %A, got %A" expected actual + + let private observe (token: CancellationToken) work = + let terminal = ResizeArray() + Async.StartWithContinuations(work, (fun _ -> terminal.Add "success"), (fun e -> terminal.Add e.Message), (fun _ -> terminal.Add "cancel"), token) + terminal + + let preCancelled () = + for throws in [false; true] do + use source = new CancellationTokenSource() + source.Cancel() + let mutable cleanups = 0 + let mutable entered = false + let work = async.TryFinally(async { entered <- true }, fun () -> + cleanups <- cleanups + 1 + if throws then failwith "cleanup") + let terminal = observe source.Token work + check false entered + check 1 cleanups + check [|"cancel"|] (terminal.ToArray()) + + let cancelledAcquisition () = + for acquisition in [0; 1; 2] do + use source = new CancellationTokenSource() + let mutable disposals = 0 + let mutable entered = false + let resource = { new IDisposable with member _.Dispose() = disposals <- disposals + 1 } + let acquire = Async.FromContinuations(fun (ok, _, _) -> + source.Cancel() + ok resource) + let work = + if acquisition = 1 then async { + use! owned = acquire + entered <- true + } + elif acquisition = 0 then async { + let! acquired = acquire + use owned = acquired + entered <- true + } + else async { + use owned = + source.Cancel() + resource + entered <- true + } + let terminal = observe source.Token work + check false entered + check 1 disposals + check [|"cancel"|] (terminal.ToArray()) + + let normalAndFailure () = + for fails in [false; true] do + let mutable disposals = 0 + let resource = { new IDisposable with member _.Dispose() = disposals <- disposals + 1 } + let work = async { + use owned = resource + if fails then failwith "body" + return 42 + } + let terminal = observe CancellationToken.None work + check 1 disposals + check [|if fails then "body" else "success"|] (terminal.ToArray()) + + let trampoline () = async { + let mutable cleanups = 0 + for _ in 1 .. 10000 do + do! async.TryFinally(async { return () }, fun () -> cleanups <- cleanups + 1) + check 10000 cleanups + } + + let downstreamExceptions () = + for channel in ["success"; "error"; "cancel"] do + let mutable cleanups = 0 + let calls = ResizeArray() + let body: Async = Async.FromContinuations(fun (ok, error, cancelled) -> + match channel with + | "success" -> ok () + | "error" -> error (Exception "body") + | _ -> cancelled (OperationCanceledException())) + let terminal name = calls.Add name; failwith "downstream" + let caught = + try + Async.StartWithContinuations( + async.TryFinally(body, fun () -> cleanups <- cleanups + 1), + (fun () -> terminal "success"), (fun _ -> terminal "error"), (fun _ -> terminal "cancel")) + "no exception" + with e -> e.Message + check "downstream" caught + check 1 cleanups + check [|channel|] (calls.ToArray()) + #if FABLE_COMPILER open Fable.Core @@ -59,6 +156,12 @@ let asyncMap f a = async { let tests = testList "Async" [ + testCase "TryFinally installs cleanup before pre-cancellation" <| FinalizerRegressions.preCancelled + testCase "Cancelled acquisition disposes use and use bang resources" <| FinalizerRegressions.cancelledAcquisition + testCase "Resource cleanup runs once on success and failure" <| FinalizerRegressions.normalAndFailure + testCase "TryFinally propagates downstream continuation errors once" <| FinalizerRegressions.downstreamExceptions + testCaseAsync "TryFinally preserves bounded trampoline execution" <| FinalizerRegressions.trampoline + testCase "FromContinuations unit and generic values" <| fun () -> let fromValue value = Async.FromContinuations(fun (ok, _, _) -> ok value) let mutable completed = 0 diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 713af3d5c6..5845467881 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -3,6 +3,103 @@ module Fable.Tests.Async open System open Util.Testing +module private FinalizerRegressions = + open System + open System.Threading + + let private check expected actual = + if expected <> actual then failwithf "Expected %A, got %A" expected actual + + let private observe (token: CancellationToken) work = + let terminal = ResizeArray() + Async.StartWithContinuations(work, (fun _ -> terminal.Add "success"), (fun e -> terminal.Add e.Message), (fun _ -> terminal.Add "cancel"), token) + terminal + + let preCancelled () = + for throws in [false; true] do + use source = new CancellationTokenSource() + source.Cancel() + let mutable cleanups = 0 + let mutable entered = false + let work = async.TryFinally(async { entered <- true }, fun () -> + cleanups <- cleanups + 1 + if throws then failwith "cleanup") + let terminal = observe source.Token work + check false entered + check 1 cleanups + check [|"cancel"|] (terminal.ToArray()) + + let cancelledAcquisition () = + for acquisition in [0; 1; 2] do + use source = new CancellationTokenSource() + let mutable disposals = 0 + let mutable entered = false + let resource = { new IDisposable with member _.Dispose() = disposals <- disposals + 1 } + let acquire = Async.FromContinuations(fun (ok, _, _) -> + source.Cancel() + ok resource) + let work = + if acquisition = 1 then async { + use! owned = acquire + entered <- true + } + elif acquisition = 0 then async { + let! acquired = acquire + use owned = acquired + entered <- true + } + else async { + use owned = + source.Cancel() + resource + entered <- true + } + let terminal = observe source.Token work + check false entered + check 1 disposals + check [|"cancel"|] (terminal.ToArray()) + + let normalAndFailure () = + for fails in [false; true] do + let mutable disposals = 0 + let resource = { new IDisposable with member _.Dispose() = disposals <- disposals + 1 } + let work = async { + use owned = resource + if fails then failwith "body" + return 42 + } + let terminal = observe CancellationToken.None work + check 1 disposals + check [|if fails then "body" else "success"|] (terminal.ToArray()) + + let trampoline () = async { + let mutable cleanups = 0 + for _ in 1 .. 10000 do + do! async.TryFinally(async { return () }, fun () -> cleanups <- cleanups + 1) + check 10000 cleanups + } + + let downstreamExceptions () = + for channel in ["success"; "error"; "cancel"] do + let mutable cleanups = 0 + let calls = ResizeArray() + let body: Async = Async.FromContinuations(fun (ok, error, cancelled) -> + match channel with + | "success" -> ok () + | "error" -> error (Exception "body") + | _ -> cancelled (OperationCanceledException())) + let terminal name = calls.Add name; failwith "downstream" + let caught = + try + Async.StartWithContinuations( + async.TryFinally(body, fun () -> cleanups <- cleanups + 1), + (fun () -> terminal "success"), (fun _ -> terminal "error"), (fun _ -> terminal "cancel")) + "no exception" + with e -> e.Message + check "downstream" caught + check 1 cleanups + check [|channel|] (calls.ToArray()) + #if FABLE_COMPILER module private Runtime = open Fable.Core.PyInterop @@ -957,3 +1054,23 @@ let ``test cancelling the mailbox token cancels a body waiting in Receive`` () = |> ignore) equal "cancelled" outcome } |> Async.RunSynchronously + +[] +let ``test TryFinally installs cleanup before pre-cancellation`` () = + FinalizerRegressions.preCancelled () + +[] +let ``test cancelled acquisition disposes use and use bang resources`` () = + FinalizerRegressions.cancelledAcquisition () + +[] +let ``test resource cleanup runs once on success and failure`` () = + FinalizerRegressions.normalAndFailure () + +[] +let ``test TryFinally preserves bounded trampoline execution`` () = + FinalizerRegressions.trampoline () |> Async.RunSynchronously + +[] +let ``test TryFinally propagates downstream continuation errors once`` () = + FinalizerRegressions.downstreamExceptions () diff --git a/tests/Python/test_async_runtime.py b/tests/Python/test_async_runtime.py index 4e04b809a1..2905221696 100644 --- a/tests/Python/test_async_runtime.py +++ b/tests/Python/test_async_runtime.py @@ -263,3 +263,42 @@ async def start() -> None: assert not token.listeners finally: loop.close() + + +@pytest.mark.parametrize("channel", ["success", "error", "cancel"]) +def test_try_finally_propagates_downstream_errors_once(channel: str) -> None: + from fable_library.async_builder import OperationCanceledError, singleton + + async def run() -> None: + cleanups = 0 + calls: list[str] = [] + trampoline = Trampoline() + token = CancellationToken() + defect = ValueError("downstream") + + def cleanup() -> None: + nonlocal cleanups + cleanups += 1 + + def terminal(value: object) -> None: + calls.append(channel) + raise defect + + def body(ctx: IAsyncContext[int]) -> None: + assert ctx.trampoline is trampoline + assert ctx.cancel_token is token + if channel == "success": + ctx.on_success(42) + elif channel == "error": + ctx.on_error(ValueError("body")) + else: + ctx.on_cancel(OperationCanceledError()) + + ctx = IAsyncContext.create(trampoline, token, terminal, terminal, terminal) + with pytest.raises(ValueError) as caught: + singleton.TryFinally(body, cleanup)(ctx) + assert caught.value is defect + assert cleanups == 1 + assert calls == [channel] + + asyncio.run(run()) From bfe9150284f6cfa93e3614ed0347cfdb27efc29d Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Tue, 13 Oct 2026 23:47:29 +0200 Subject: [PATCH 2/4] fix(python): adapt erased unit at generic async continuations --- .../Python/Fable2Python.Transforms.fs | 11 ++- .../Python/Fable2Python.Util.fs | 21 +++++ src/fable-library-py/fable_library/async_.py | 45 ++++++++-- src/fable-library-py/fable_library/util.py | 10 +++ tests/Js/Main/AsyncTests.fs | 16 ++++ tests/Python/TestAsync.fs | 26 ++++++ tests/Python/test_async_runtime.py | 90 +++++++++++++++++++ 7 files changed, 207 insertions(+), 12 deletions(-) diff --git a/src/Fable.Transforms/Python/Fable2Python.Transforms.fs b/src/Fable.Transforms/Python/Fable2Python.Transforms.fs index d0bb7480da..2b42fce334 100644 --- a/src/Fable.Transforms/Python/Fable2Python.Transforms.fs +++ b/src/Fable.Transforms/Python/Fable2Python.Transforms.fs @@ -854,13 +854,16 @@ let transformCallArgs let hasSpread = paramsInfo |> Option.map (fun i -> i.HasSpread) |> Option.defaultValue false - // Helper to transform an arg and wrap with widen() if needed - let transformArgWithWiden (sigType: Fable.Type option) (argExpr: Fable.Expr) = + // Adapt callback annotations at generic representation boundaries. + let transformArgWithAdapters (sigType: Fable.Type option) (argExpr: Fable.Expr) = let expr, stmts = com.TransformAsExpr(ctx, argExpr) if needsOptionWidenForArg sigType argExpr then let widen = com.TransformImport(ctx, "widen", getLibPath com "option") Expression.call (widen, [ expr ]), stmts + elif needsUnitCallableEraseForArg sigType argExpr then + let erase = com.TransformImport(ctx, "erase_unit_callable", getLibPath com "util") + Expression.call (erase, [ expr ]), stmts else expr, stmts @@ -884,11 +887,11 @@ let transformCallArgs let expr, stmts' = com.TransformAsExpr(ctx, last) rest @ [ Expression.starred expr ], stmts @ stmts' | args -> - // Transform args with widen() where needed based on signature types + // Use declaration signature types to identify specialization boundaries. args |> List.mapi (fun i e -> let sigType = List.tryItem i callInfo.SignatureArgTypes - transformArgWithWiden sigType e + transformArgWithAdapters sigType e ) |> Helpers.unzipArgs diff --git a/src/Fable.Transforms/Python/Fable2Python.Util.fs b/src/Fable.Transforms/Python/Fable2Python.Util.fs index e449751b98..81d39b82d8 100644 --- a/src/Fable.Transforms/Python/Fable2Python.Util.fs +++ b/src/Fable.Transforms/Python/Fable2Python.Util.fs @@ -319,6 +319,27 @@ module Util = | _ -> false | _ -> false + /// Detect an erased unit callable inside a callback parameter specialized from a generic input. + let needsUnitCallableEraseForArg (expectedType: Fable.Type option) (argExpr: Fable.Expr) = + let rec mismatches expected actual = + match FableTransforms.uncurryType expected, FableTransforms.uncurryType actual with + | Fable.DelegateType([ Fable.GenericParam _ ], _), Fable.DelegateType([ Fable.Unit ], _) -> true + | Fable.DelegateType(expectedArgs, _), Fable.DelegateType(actualArgs, _) + | Fable.Tuple(expectedArgs, _), Fable.Tuple(actualArgs, _) -> + List.sameLength expectedArgs actualArgs + && List.exists2 mismatches expectedArgs actualArgs + | _ -> false + + // decision: adapt only generic-to-unit callable parameters; ordinary unit and non-unit annotations stay precise + expectedType + |> Option.exists (fun expected -> + match FableTransforms.uncurryType expected, FableTransforms.uncurryType argExpr.Type with + | Fable.DelegateType(expectedArgs, _), Fable.DelegateType(actualArgs, _) -> + List.sameLength expectedArgs actualArgs + && List.exists2 mismatches expectedArgs actualArgs + | _ -> false + ) + /// Wraps None values in cast(type, None) for type safety. /// Skips if type annotation is also None (unit type). let wrapNoneInCast (com: IPythonCompiler) ctx (value: Expression) (typeAnnotation: Expression) : Expression = diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 0e06c6fffd..9359a97149 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -7,6 +7,7 @@ from threading import RLock, Timer from typing import ( Any, + cast, ) from .array_ import Array @@ -228,16 +229,44 @@ def on_error(err: Exception): return protected_cont(cont) -def from_continuations[T]( - f: Callable[ - [Continuations[T]], - None, - ], -) -> Callable[[IAsyncContext[T]], None]: +def from_continuations[T](f: Callable[[Continuations[T]], None]) -> Async[T]: def cont(ctx: IAsyncContext[T]) -> None: - f((ctx.on_success, ctx.on_error, ctx.on_cancel)) + settled = False + gate = RLock() - return protected_cont(cont) + def claim() -> None: + nonlocal settled + # invariant: settlement is claimed under the gate and user continuations run outside it + with gate: + if settled: + raise RuntimeError("A continuation was invoked more than once") + settled = True + + # decision: the default represents only an erased F# unit; Python cannot express that specialization of T + def success(value: T = cast(T, None)) -> None: + claim() + ctx.on_success(value) + + def error(value: Exception) -> None: + claim() + ctx.on_error(value) + + def cancelled(value: OperationCanceledError) -> None: + claim() + ctx.on_cancel(value) + + def on_exception(value: Exception) -> None: + # invariant: an exception after settlement belongs to continuation code and propagates unchanged + with gate: + completed = settled + if completed: + raise value + error(value) + + guarded = IAsyncContext.create(ctx.trampoline, ctx.cancel_token, success, on_exception, cancelled) + protected_cont(lambda _: f((success, error, cancelled)))(guarded) + + return cont def from_continuations_unit( diff --git a/src/fable-library-py/fable_library/util.py b/src/fable-library-py/fable_library/util.py index 1d3f5648ed..44bb5c9714 100644 --- a/src/fable-library-py/fable_library/util.py +++ b/src/fable-library-py/fable_library/util.py @@ -56,6 +56,15 @@ from .types import UNIT, Unit +def erase_unit_callable[**P, R](callback: Callable[P, R], /) -> Callable[..., R]: + """Adapt a callback annotation across an F# generic-to-unit boundary. + + decision: erase parameter arity for the type checker while retaining the callback's return type + invariant: the callback's identity and erased-unit invocation are unchanged at runtime + """ + return callback + + class nullable: """Returns a tuple of None values with the specified types. @@ -963,6 +972,7 @@ def range(start: int, stop: int, step: int = 1) -> Iterable[int]: "Unit", "array_hash", "copy_to_array", + "erase_unit_callable", "escape_data_string", "escape_uri_string", "create_random", diff --git a/tests/Js/Main/AsyncTests.fs b/tests/Js/Main/AsyncTests.fs index 4d7ffc6e92..0fbfb3c7d0 100644 --- a/tests/Js/Main/AsyncTests.fs +++ b/tests/Js/Main/AsyncTests.fs @@ -154,6 +154,9 @@ let asyncMap f a = async { return f a } +let private fromGenericCallback (callback: ('T -> unit) -> (exn -> unit) -> (OperationCanceledException -> unit) -> unit) : Async<'T> = + Async.FromContinuations(fun (ok, error, cancelled) -> callback ok error cancelled) + let tests = testList "Async" [ testCase "TryFinally installs cleanup before pre-cancellation" <| FinalizerRegressions.preCancelled @@ -162,6 +165,19 @@ let tests = testCase "TryFinally propagates downstream continuation errors once" <| FinalizerRegressions.downstreamExceptions testCaseAsync "TryFinally preserves bounded trampoline execution" <| FinalizerRegressions.trampoline + testCase "Generic FromContinuations accepts erased unit callbacks" <| fun () -> + let calls = ResizeArray() + let start work = Async.StartWithContinuations(work, (fun _ -> calls.Add "success"), (fun e -> calls.Add e.Message), (fun e -> calls.Add e.Message)) + let unitWork: Async = fromGenericCallback (fun ok _ _ -> ok ()) + start unitWork + let valueWork = fromGenericCallback (fun ok _ _ -> ok 42) + Async.StartWithContinuations(valueWork, (fun value -> equal 42 value; calls.Add "value"), raise, raise) + let errorWork: Async = fromGenericCallback (fun _ error _ -> error (exn "error")) + start errorWork + let cancelWork: Async = fromGenericCallback (fun _ _ cancelled -> cancelled (OperationCanceledException("cancel"))) + start cancelWork + equal [|"success"; "value"; "error"; "cancel"|] (calls.ToArray()) + testCase "FromContinuations unit and generic values" <| fun () -> let fromValue value = Async.FromContinuations(fun (ok, _, _) -> ok value) let mutable completed = 0 diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 5845467881..4e20a810d3 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -1074,3 +1074,29 @@ let ``test TryFinally preserves bounded trampoline execution`` () = [] let ``test TryFinally propagates downstream continuation errors once`` () = FinalizerRegressions.downstreamExceptions () + +let private fromGenericCallback (callback: ('T -> unit) -> (exn -> unit) -> (OperationCanceledException -> unit) -> unit) : Async<'T> = + Async.FromContinuations(fun (ok, error, cancelled) -> callback ok error cancelled) + +[] +let ``test generic FromContinuations accepts erased unit callbacks`` () = + let calls = ResizeArray() + let start work = Async.StartWithContinuations(work, (fun _ -> calls.Add "success"), (fun e -> calls.Add e.Message), (fun e -> calls.Add e.Message)) + let unitWork: Async = fromGenericCallback (fun ok _ _ -> ok ()) + start unitWork + let valueWork = fromGenericCallback (fun ok _ _ -> ok 42) + Async.StartWithContinuations(valueWork, (fun value -> equal 42 value; calls.Add "value"), raise, raise) + let errorWork: Async = fromGenericCallback (fun _ error _ -> error (exn "error")) + start errorWork + let cancelWork: Async = fromGenericCallback (fun _ _ cancelled -> cancelled (OperationCanceledException("cancel"))) + start cancelWork + equal [|"success"; "value"; "error"; "cancel"|] (calls.ToArray()) + +let private invokeGenericCallback (callback: ('T -> unit) -> 'Result) (value: 'T) : 'Result = + callback (fun received -> equal value received) + +[] +let ``test generic unit callback adaptation preserves return types`` () = + equal 42 (invokeGenericCallback (fun ok -> ok (); 42) ()) + equal "unit" (invokeGenericCallback (fun ok -> ok (); "unit") ()) + equal "value" (invokeGenericCallback (fun ok -> ok 7; "value") 7) diff --git a/tests/Python/test_async_runtime.py b/tests/Python/test_async_runtime.py index 2905221696..272e45380e 100644 --- a/tests/Python/test_async_runtime.py +++ b/tests/Python/test_async_runtime.py @@ -302,3 +302,93 @@ def body(ctx: IAsyncContext[int]) -> None: assert calls == [channel] asyncio.run(run()) + + +@pytest.mark.parametrize("channel", ["success", "error", "cancel"]) +def test_generic_from_continuations_propagates_terminal_exceptions(channel: str) -> None: + from fable_library.async_ import from_continuations + from fable_library.async_builder import Continuations, OperationCanceledError + + async def run() -> None: + calls: list[str] = [] + defect = ValueError("downstream") + trampoline = Trampoline() + token = CancellationToken() + + def terminal(value: object) -> None: + calls.append(channel) + raise defect + + def callback(conts: Continuations[int]) -> None: + ok, error, cancelled = conts + if channel == "success": + ok(42) + elif channel == "error": + error(ValueError("body")) + else: + cancelled(OperationCanceledError()) + + ctx = IAsyncContext.create(trampoline, token, terminal, terminal, terminal) + with pytest.raises(ValueError) as caught: + from_continuations(callback)(ctx) + assert caught.value is defect + assert calls == [channel] + assert ctx.trampoline is trampoline + + asyncio.run(run()) + + +def test_generic_from_continuations_rejects_duplicate_completion() -> None: + from fable_library.async_ import from_continuations + from fable_library.async_builder import Continuations + + calls: list[int] = [] + + def callback(conts: Continuations[int]) -> None: + conts[0](42) + conts[0](43) + + with pytest.raises(RuntimeError, match="invoked more than once"): + start_with_continuations(from_continuations(callback), calls.append) + assert calls == [42] + + +def test_generic_from_continuations_claims_concurrent_completion_once() -> None: + from fable_library.async_ import from_continuations + from fable_library.async_builder import Continuations + + callbacks: Continuations[int] | None = None + calls: list[str] = [] + rejected: list[Exception] = [] + barrier = Barrier(2) + + def callback(conts: Continuations[int]) -> None: + nonlocal callbacks + callbacks = conts + + start_with_continuations( + from_continuations(callback), + lambda _: calls.append("success"), + lambda _: calls.append("error"), + ) + assert callbacks is not None + ok, error, _ = callbacks + + def complete(success: bool) -> None: + barrier.wait(5) + try: + if success: + ok(42) + else: + error(ValueError("body")) + except RuntimeError as duplicate: + rejected.append(duplicate) + + threads = [Thread(target=complete, args=(success,)) for success in (True, False)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(5) + assert all(not thread.is_alive() for thread in threads) + assert len(calls) == 1 + assert len(rejected) == 1 From 3eb3cbed961fe0c480ec48db760aa7775b6b02d3 Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Wed, 14 Oct 2026 09:05:39 +0200 Subject: [PATCH 3/4] test(python): cover pre-cancelled nested async finalizers --- tests/Python/TestAsync.fs | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 4e20a810d3..3d92935be8 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -29,6 +29,16 @@ module private FinalizerRegressions = check 1 cleanups check [|"cancel"|] (terminal.ToArray()) + let nestedPreCancelled () = + use source = new CancellationTokenSource() + source.Cancel() + let cleanups = ResizeArray() + let inner = async.TryFinally(async.Return 42, fun () -> cleanups.Add "inner") + let outer = async.TryFinally(inner, fun () -> cleanups.Add "outer") + let terminal = observe source.Token outer + check [|"outer"|] (cleanups.ToArray()) + check [|"cancel"|] (terminal.ToArray()) + let cancelledAcquisition () = for acquisition in [0; 1; 2] do use source = new CancellationTokenSource() @@ -1059,6 +1069,10 @@ let ``test cancelling the mailbox token cancels a body waiting in Receive`` () = let ``test TryFinally installs cleanup before pre-cancellation`` () = FinalizerRegressions.preCancelled () +[] +let ``test pre-cancelled nested TryFinally only runs outer cleanup`` () = + FinalizerRegressions.nestedPreCancelled () + [] let ``test cancelled acquisition disposes use and use bang resources`` () = FinalizerRegressions.cancelledAcquisition () From efe486d1282c29d26a8d324945e26ce58c77e92f Mon Sep 17 00:00:00 2001 From: Maxime Mangel Date: Thu, 15 Oct 2026 20:34:27 +0200 Subject: [PATCH 4/4] fix(js/ts/python): settle async continuations once (#5081) --- .../Python/Fable2Python.Transforms.fs | 84 ++++++--- .../Python/Fable2Python.Types.fs | 1 + .../Python/Fable2Python.Util.fs | 28 ++- src/fable-library-py/fable_library/async_.py | 31 +-- .../fable_library/async_builder.py | 58 +++--- .../fable_library/mailbox_processor.py | 25 ++- src/fable-library-py/fable_library/util.py | 5 - src/fable-library-ts/Async.ts | 21 ++- src/fable-library-ts/AsyncBuilder.ts | 6 +- src/fable-library-ts/MailboxProcessor.ts | 2 + tests/Js/Main/AsyncTests.fs | 42 +++++ tests/Python/TestAsync.fs | 74 ++++++++ tests/Python/TestMailboxProcessor.fs | 14 ++ tests/Python/test_async_runtime.py | 178 ++++++++++++++++++ 14 files changed, 482 insertions(+), 87 deletions(-) diff --git a/src/Fable.Transforms/Python/Fable2Python.Transforms.fs b/src/Fable.Transforms/Python/Fable2Python.Transforms.fs index 2b42fce334..c445abfde4 100644 --- a/src/Fable.Transforms/Python/Fable2Python.Transforms.fs +++ b/src/Fable.Transforms/Python/Fable2Python.Transforms.fs @@ -757,11 +757,24 @@ let transformObjectExpr Expression.call (Expression.name name), thisCaptureStmts @ [ stmt ] @ stmts -let transformCallArgs +let transformArgWithAdapters (com: IPythonCompiler) ctx (sigType: Fable.Type option) (argExpr: Fable.Expr) = + let expr, stmts = com.TransformAsExpr(ctx, argExpr) + + if needsOptionWidenForArg sigType argExpr then + let widen = com.TransformImport(ctx, "widen", getLibPath com "option") + Expression.call (widen, [ expr ]), stmts + elif needsUnitCallableEraseForArg sigType argExpr then + let erase = com.TransformImport(ctx, "erase_unit_callable", getLibPath com "util") + Expression.call (erase, [ expr ]), stmts + else + expr, stmts + +let transformCallArgsWith (com: IPythonCompiler) ctx (callInfo: Fable.CallInfo) (isBaseConstructorCall: bool) + (fallbackSigTypes: Fable.Type list) : Expression list * Keyword list * Statement list = @@ -854,18 +867,15 @@ let transformCallArgs let hasSpread = paramsInfo |> Option.map (fun i -> i.HasSpread) |> Option.defaultValue false - // Adapt callback annotations at generic representation boundaries. - let transformArgWithAdapters (sigType: Fable.Type option) (argExpr: Fable.Expr) = - let expr, stmts = com.TransformAsExpr(ctx, argExpr) + let transformArgs args = + args + |> List.mapi (fun i e -> + let sigType = + List.tryItem i callInfo.SignatureArgTypes + |> Option.orElse (List.tryItem i fallbackSigTypes) - if needsOptionWidenForArg sigType argExpr then - let widen = com.TransformImport(ctx, "widen", getLibPath com "option") - Expression.call (widen, [ expr ]), stmts - elif needsUnitCallableEraseForArg sigType argExpr then - let erase = com.TransformImport(ctx, "erase_unit_callable", getLibPath com "util") - Expression.call (erase, [ expr ]), stmts - else - expr, stmts + transformArgWithAdapters com ctx sigType e + ) let args, stmts' = match args with @@ -874,31 +884,30 @@ let transformCallArgs match List.rev args with | [] -> [], [] | Replacements.Util.ArrayOrListLiteral(spreadArgs, _) :: rest -> - let rest = List.rev rest |> List.map (fun e -> com.TransformAsExpr(ctx, e)) + let rest = List.rev rest |> transformArgs rest @ List.map (fun e -> com.TransformAsExpr(ctx, e)) spreadArgs |> Helpers.unzipArgs | last :: rest -> - let rest, stmts = - List.rev rest - |> List.map (fun e -> com.TransformAsExpr(ctx, e)) - |> Helpers.unzipArgs + let rest, stmts = List.rev rest |> transformArgs |> Helpers.unzipArgs let expr, stmts' = com.TransformAsExpr(ctx, last) rest @ [ Expression.starred expr ], stmts @ stmts' - | args -> - // Use declaration signature types to identify specialization boundaries. - args - |> List.mapi (fun i e -> - let sigType = List.tryItem i callInfo.SignatureArgTypes - transformArgWithAdapters sigType e - ) - |> Helpers.unzipArgs + | args -> args |> transformArgs |> Helpers.unzipArgs match objArg with | None -> args, [], stmts @ stmts' | Some objArg -> args, objArg, stmts @ stmts' +let transformCallArgs + (com: IPythonCompiler) + ctx + (callInfo: Fable.CallInfo) + (isBaseConstructorCall: bool) + : Expression list * Keyword list * Statement list + = + transformCallArgsWith com ctx callInfo isBaseConstructorCall [] + let resolveExpr (ctx: Context) _t strategy pyExpr : Statement list = // printfn "resolveExpr: %A" (pyExpr, strategy) match strategy with @@ -1284,7 +1293,8 @@ let transformCall (com: IPythonCompiler) ctx range callee (callInfo: Fable.CallI let callee', stmts = com.TransformAsExpr(ctx, callee) - let args, kw, stmts' = transformCallArgs com ctx callInfo false + let args, kw, stmts' = + transformCallArgsWith com ctx callInfo false (calleeSignatureArgTypes ctx callee callInfo) match callee, callInfo.ThisArg with | Fable.Get(expr, Fable.FieldGet { Name = "Dispose" }, _, _), _ -> @@ -1307,9 +1317,17 @@ let transformCall (com: IPythonCompiler) ctx range callee (callInfo: Fable.CallI Expression.call (callee', args, kw, ?loc = range), stmts @ stmts' | _, None -> callFunction range callee' args kw, stmts @ stmts' -let transformCurriedApply com ctx range (TransformExpr com ctx (applied, stmts)) args = - ((applied, stmts), args) - ||> List.fold (fun (applied, stmts) arg -> +let transformCurriedApply (com: IPythonCompiler) ctx range (callee: Fable.Expr) args = + let rec expectedArgTypes typ args = + match typ, args with + | Fable.LambdaType(argType, returnType), _ :: rest -> Some argType :: expectedArgTypes returnType rest + | _, _ :: rest -> None :: expectedArgTypes typ rest + | _, [] -> [] + + let applied, stmts = com.TransformAsExpr(ctx, callee) + + ((applied, stmts), List.zip (expectedArgTypes callee.Type args) args) + ||> List.fold (fun (applied, stmts) (sigType, arg) -> let args, argStmts = match arg with // TODO: If arg type is unit but it's an expression with potential @@ -1319,7 +1337,7 @@ let transformCurriedApply com ctx range (TransformExpr com ctx (applied, stmts)) | Fable.Value(Fable.UnitConstant, _) -> [], [] | Fable.IdentExpr ident when ident.Type = Fable.Unit -> [], [] | arg -> - let argExpr, transformStmts = com.TransformAsExpr(ctx, arg) + let argExpr, transformStmts = transformArgWithAdapters com ctx sigType arg // When a Call or CurriedApply result is passed as an argument, // check if we need to erase Option[T] to T | None let argExpr = @@ -2832,6 +2850,8 @@ let rec transformAsExpr (com: IPythonCompiler) ctx (expr: Fable.Expr) : Expressi iife com ctx expr | Fable.LetRec(bindings, body) -> + let ctx = withDeclaredFunctionTypes ctx bindings + if ctx.HoistVars(List.map fst bindings) then Expression.withStmts { let values, stmts = @@ -3033,10 +3053,13 @@ let rec transformAsStatements (com: IPythonCompiler) ctx returnStrategy (expr: F | Fable.Let(ident, value, body) -> let binding = transformBindingAsStatements com ctx ident value + let ctx = withDeclaredFunctionTypes ctx [ ident, value ] List.append binding (transformAsStatements com ctx returnStrategy body) | Fable.LetRec(bindings, body) -> + let ctx = withDeclaredFunctionTypes ctx bindings + let bindings = bindings |> Seq.collect (fun (i, v) -> transformBindingAsStatements com ctx i v) @@ -5469,6 +5492,7 @@ let transformFile (com: IPythonCompiler) (file: Fable.File) = ScopedTypeParams = Set.empty TypeParamsScope = 0 NarrowedTypes = Map.empty + DeclaredFunctionTypes = Map.empty EnclosingUnionBaseClass = None InEmitMacroArgument = false } diff --git a/src/Fable.Transforms/Python/Fable2Python.Types.fs b/src/Fable.Transforms/Python/Fable2Python.Types.fs index 2e9536484a..93dad4224e 100644 --- a/src/Fable.Transforms/Python/Fable2Python.Types.fs +++ b/src/Fable.Transforms/Python/Fable2Python.Types.fs @@ -124,6 +124,7 @@ type Context = ScopedTypeParams: Set TypeParamsScope: int NarrowedTypes: Map + DeclaredFunctionTypes: Map /// When inside a union base class definition, this holds the entity name. /// Used to determine whether to use base class name or type alias for annotations. EnclosingUnionBaseClass: string option diff --git a/src/Fable.Transforms/Python/Fable2Python.Util.fs b/src/Fable.Transforms/Python/Fable2Python.Util.fs index 81d39b82d8..40fe337adb 100644 --- a/src/Fable.Transforms/Python/Fable2Python.Util.fs +++ b/src/Fable.Transforms/Python/Fable2Python.Util.fs @@ -319,7 +319,6 @@ module Util = | _ -> false | _ -> false - /// Detect an erased unit callable inside a callback parameter specialized from a generic input. let needsUnitCallableEraseForArg (expectedType: Fable.Type option) (argExpr: Fable.Expr) = let rec mismatches expected actual = match FableTransforms.uncurryType expected, FableTransforms.uncurryType actual with @@ -330,7 +329,6 @@ module Util = && List.exists2 mismatches expectedArgs actualArgs | _ -> false - // decision: adapt only generic-to-unit callable parameters; ordinary unit and non-unit annotations stay precise expectedType |> Option.exists (fun expected -> match FableTransforms.uncurryType expected, FableTransforms.uncurryType argExpr.Type with @@ -340,6 +338,32 @@ module Util = | _ -> false ) + // Identifiers at call sites carry the instantiated type, not the declared generic one + let withDeclaredFunctionTypes (ctx: Context) (bindings: (Fable.Ident * Fable.Expr) list) = + let declared = + (ctx.DeclaredFunctionTypes, bindings) + ||> List.fold (fun declared (ident, value) -> + match value with + | Fable.Lambda _ + | Fable.Delegate _ -> Map.add ident.Name ident.Type declared + | _ -> Map.remove ident.Name declared + ) + + { ctx with DeclaredFunctionTypes = declared } + + // uncurryApplications emits an empty SignatureArgTypes for local function calls + let calleeSignatureArgTypes (ctx: Context) (callee: Fable.Expr) (callInfo: Fable.CallInfo) = + let calleeType = + match callee with + | Fable.IdentExpr ident -> + Map.tryFind ident.Name ctx.DeclaredFunctionTypes + |> Option.defaultValue callee.Type + | _ -> callee.Type + + match callInfo.SignatureArgTypes, FableTransforms.uncurryType calleeType with + | [], Fable.DelegateType(argTypes, _) when List.sameLength argTypes callInfo.Args -> argTypes + | _ -> [] + /// Wraps None values in cast(type, None) for type safety. /// Skips if type annotation is also None (unit type). let wrapNoneInCast (com: IPythonCompiler) ctx (value: Expression) (typeAnnotation: Expression) : Expression = diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 9359a97149..4f82c01e59 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -34,7 +34,7 @@ ) from .event import IEvent_2 from .protocols import IEnumerable_1 -from .system import TimeoutException__ctor +from .system import InvalidOperationException__ctor_Z721C83C5, TimeoutException__ctor from .task import TaskCompletionSource from .time_span import TimeSpan, to_milliseconds from .types import UNIT, Unit @@ -236,13 +236,12 @@ def cont(ctx: IAsyncContext[T]) -> None: def claim() -> None: nonlocal settled - # invariant: settlement is claimed under the gate and user continuations run outside it with gate: if settled: - raise RuntimeError("A continuation was invoked more than once") + raise InvalidOperationException__ctor_Z721C83C5("A continuation was invoked more than once") settled = True - # decision: the default represents only an erased F# unit; Python cannot express that specialization of T + # Compiled F# calls the success continuation with no argument when T is unit def success(value: T = cast(T, None)) -> None: claim() ctx.on_success(value) @@ -256,7 +255,6 @@ def cancelled(value: OperationCanceledError) -> None: ctx.on_cancel(value) def on_exception(value: Exception) -> None: - # invariant: an exception after settlement belongs to continuation code and propagates unchanged with gate: completed = settled if completed: @@ -275,13 +273,11 @@ def from_continuations_unit( None, ], ) -> Async[None]: - def cont(ctx: IAsyncContext[None]) -> None: - def success(value: None = None) -> None: - ctx.on_success(value) + def callback(conts: Continuations[None]) -> None: + success, error, cancelled = conts + f((lambda: success(None), error, cancelled)) - f((success, ctx.on_error, ctx.on_cancel)) - - return protected_cont(cont) + return from_continuations(callback) def await_task[T](task: Awaitable[T]) -> Async[T]: @@ -356,12 +352,19 @@ def start_with_continuations[T]( token is used.""" trampoline = Trampoline() + def done[U](cont: Callable[[U], None]) -> Callable[[U], None]: + def terminal(value: U) -> None: + trampoline.completed = True + cont(value) + + return terminal + ctx = IAsyncContext.create( trampoline, cancellation_token or default_cancellation_token, - continuation or empty_continuation, - exception_continuation or empty_continuation, - cancellation_continuation or empty_continuation, + done(continuation or empty_continuation), + done(exception_continuation or empty_continuation), + done(cancellation_continuation or empty_continuation), ) def runner() -> None: diff --git a/src/fable-library-py/fable_library/async_builder.py b/src/fable-library-py/fable_library/async_builder.py index a0dfcc04f6..e3f43ead41 100644 --- a/src/fable-library-py/fable_library/async_builder.py +++ b/src/fable-library-py/fable_library/async_builder.py @@ -233,7 +233,7 @@ class ScheduledItem: class Trampoline: - __slots__ = "call_count", "lock", "running" + __slots__ = "call_count", "completed", "lock", "running" MaxTrampolineCallCount = 75 # Max recursion depth: 1000 @@ -241,6 +241,8 @@ def __init__(self) -> None: self.call_count: int = 0 self.lock = Lock() self.running = False + # After a terminal continuation runs, exceptions reaching protected_cont come from continuation code + self.completed = False def increment_and_check(self): with self.lock: @@ -256,13 +258,17 @@ def run_later( return loop.call_later(due_time, action) def run(self, action: Callable[[], None]): - loop = asyncio.get_running_loop() + if not self.increment_and_check(): + action() + return - if self.increment_and_check(): - self.call_count = 0 - loop.call_soon(action) - else: + self.call_count = 0 + try: + loop = asyncio.get_running_loop() + except RuntimeError: action() + return + loop.call_soon(action) def protected_cont[T](f: Async[T]) -> Async[T]: @@ -275,7 +281,8 @@ def fn(): try: return f(ctx) except Exception as err: - # print("Exception: ", err) + if ctx.trampoline.completed: + raise ctx.on_error(err) ctx.trampoline.run(fn) @@ -290,10 +297,11 @@ def protected_bind[T, U]( def cont(ctx: IAsyncContext[U]) -> None: def on_success(x: T) -> None: try: - binder(x)(ctx) + bound = binder(x) except Exception as err: - # print("Exception: ", err) ctx.on_error(err) + return + bound(ctx) ctx_ = IAsyncContext.create(ctx.trampoline, ctx.cancel_token, on_success, ctx.on_error, ctx.on_cancel) return computation(ctx_) @@ -351,12 +359,20 @@ 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: settled = False + gate = Lock() - def on_success(x: T) -> None: + def claim() -> bool: nonlocal settled - if settled: + # on_cancel may run on another thread while on_success runs on the loop + with gate: + if settled: + return False + settled = True + return True + + def on_success(x: T) -> None: + if not claim(): return - settled = True try: compensation() except Exception as error: @@ -365,11 +381,8 @@ def on_success(x: T) -> None: ctx.on_success(x) def on_error(x: Exception) -> None: - nonlocal settled - # invariant: downstream continuation exceptions never rerun compensation - if settled: + if not claim(): raise x - settled = True try: compensation() except Exception as error: @@ -378,10 +391,8 @@ def on_error(x: Exception) -> None: ctx.on_error(x) def on_cancel(x: OperationCanceledError) -> None: - nonlocal settled - if settled: + if not claim(): return - settled = True # decision: ignores compensation failures during cancellation, matching .NET Async.TryFinally try: compensation() @@ -390,8 +401,7 @@ def on_cancel(x: OperationCanceledError) -> None: ctx.on_cancel(x) ctx_ = IAsyncContext.create(ctx.trampoline, ctx.cancel_token, on_success, on_error, on_cancel) - # invariant: cleanup continuations are installed before the cancellation checkpoint - # decision: protect the body with the same trampoline to retain bounded stack growth + # The cancellation check in protected_cont must see the cleanup continuations protected_cont(computation)(ctx_) return cont @@ -400,9 +410,11 @@ def TryWith[T](self, computation: Async[T], catch_handler: Callable[[Exception], def fn(ctx: IAsyncContext[T]): def on_error(err: Exception) -> None: try: - catch_handler(err)(ctx) + handled = catch_handler(err) except Exception as ex2: ctx.on_error(ex2) + return + handled(ctx) ctx_ = IAsyncContext.create( on_success=ctx.on_success, @@ -420,7 +432,7 @@ def Using[D: IDisposable, U](self, resource: D, binder: Callable[[D], Async[U]]) def compensation() -> None: return resource.Dispose() - # decision: defer the resource binder until cleanup is installed and cancellation is checked + # Without Delay the binder runs before TryFinally installs Dispose return self.TryFinally(self.Delay(lambda: binder(resource)), compensation) @overload diff --git a/src/fable-library-py/fable_library/mailbox_processor.py b/src/fable-library-py/fable_library/mailbox_processor.py index cbdbe5f020..ef8bc9e734 100644 --- a/src/fable-library-py/fable_library/mailbox_processor.py +++ b/src/fable-library-py/fable_library/mailbox_processor.py @@ -3,7 +3,7 @@ import asyncio from collections.abc import Callable from queue import SimpleQueue -from threading import RLock +from threading import Lock, RLock from typing import Any from .async_ import call_on_owner_loop, from_continuations, start_immediate @@ -73,18 +73,30 @@ def post_and_async_reply[Reply](self, build_message: Callable[[AsyncReplyChannel # Track whether `reply` was actually called, since the reply value itself # may legitimately be `None` (e.g. `AsyncReplyChannel[None]` for unit). replied = False + delivered = False + gate = Lock() continuation: Continuations[Any] | None = ( None # This is the continuation for the `done` callback of the awaiting poster. ) def check_completion() -> None: - if replied and continuation is not None: - continuation[0](result) + nonlocal delivered + # reply may run on another thread before the poster's continuation is installed + with gate: + if delivered or not replied or continuation is None: + return + delivered = True + deliver = continuation[0] + deliver(result) def reply_callback(res: Reply): nonlocal result, replied - result = res - replied = True + # .NET AsyncReplyChannel.Reply ignores a second reply + with gate: + if replied: + return + result = res + replied = True check_completion() reply_channel: AsyncReplyChannel[Reply] = AsyncReplyChannel(reply_callback) @@ -93,7 +105,8 @@ def reply_callback(res: Reply): def callback(conts: Continuations[Any]) -> None: nonlocal continuation - continuation = conts + with gate: + continuation = conts check_completion() return from_continuations(callback) diff --git a/src/fable-library-py/fable_library/util.py b/src/fable-library-py/fable_library/util.py index 44bb5c9714..a808660b3b 100644 --- a/src/fable-library-py/fable_library/util.py +++ b/src/fable-library-py/fable_library/util.py @@ -57,11 +57,6 @@ def erase_unit_callable[**P, R](callback: Callable[P, R], /) -> Callable[..., R]: - """Adapt a callback annotation across an F# generic-to-unit boundary. - - decision: erase parameter arity for the type checker while retaining the callback's return type - invariant: the callback's identity and erased-unit invocation are unchanged at runtime - """ return callback diff --git a/src/fable-library-ts/Async.ts b/src/fable-library-ts/Async.ts index 23ab998b7e..09eaabee44 100644 --- a/src/fable-library-ts/Async.ts +++ b/src/fable-library-ts/Async.ts @@ -4,7 +4,7 @@ import { Async, IAsyncContext, CancellationToken } from "./AsyncBuilder.ts"; import { protectedCont, protectedBind, protectedReturn } from "./AsyncBuilder.ts"; import type { IEvent$2 } from "./Event.ts"; import { FSharpChoice$2_$union, Choice_makeChoice1Of2, Choice_makeChoice2Of2 } from "./Choice.ts"; -import { TimeoutException_$ctor } from "./System.ts"; +import { InvalidOperationException_$ctor_Z721C83C5, TimeoutException_$ctor } from "./System.ts"; import { Exception } from "./Util.ts"; function emptyContinuation(_x: T) { @@ -128,8 +128,23 @@ export function catchAsync(work: Async) { } export function fromContinuations(f: (conts: Continuations) => void) { - return protectedCont((ctx: IAsyncContext) => - f([ctx.onSuccess, ctx.onError, ctx.onCancel])); + return (ctx: IAsyncContext) => { + let settled = false; + const claim = () => { + if (settled) { throw InvalidOperationException_$ctor_Z721C83C5("A continuation was invoked more than once"); } + settled = true; + }; + const success = (x: T) => { claim(); ctx.onSuccess(x); }; + const error = (x: Exception) => { claim(); ctx.onError(x); }; + const cancelled = (x: OperationCanceledException) => { claim(); ctx.onCancel(x); }; + protectedCont(() => f([success, error, cancelled]))({ + onSuccess: success, + onError: (x: Exception) => { if (settled) { throw x; } error(x); }, + onCancel: cancelled, + cancelToken: ctx.cancelToken, + trampoline: ctx.trampoline, + }); + }; } export function ignore(computation: Async) { diff --git a/src/fable-library-ts/AsyncBuilder.ts b/src/fable-library-ts/AsyncBuilder.ts index a301f97b61..0a89c59e18 100644 --- a/src/fable-library-ts/AsyncBuilder.ts +++ b/src/fable-library-ts/AsyncBuilder.ts @@ -200,8 +200,7 @@ export class AsyncBuilder { } public TryFinally(computation: Async, compensation: () => void) { - // invariant: cleanup continuations are installed before the cancellation checkpoint - // decision: protect the body with the shared trampoline so pre-cancellation still runs compensation + // The cancellation check in protectedCont must see the cleanup continuations return (ctx: IAsyncContext) => { let settled = false; protectedCont(computation)({ @@ -215,7 +214,6 @@ export class AsyncBuilder { ctx.onSuccess(x); }, onError: (x: any) => { - // invariant: downstream continuation exceptions never rerun compensation if (settled) { throw x; } settled = true; try { compensation(); } catch (error) { @@ -260,7 +258,7 @@ export class AsyncBuilder { } public Using(resource: T, binder: (x: T) => Async) { - // decision: defer the resource binder until cleanup is installed and cancellation is checked + // Without Delay the binder runs before TryFinally installs Dispose return this.TryFinally(this.Delay(() => binder(resource)), () => resource.Dispose()); } diff --git a/src/fable-library-ts/MailboxProcessor.ts b/src/fable-library-ts/MailboxProcessor.ts index b1771bf70a..c4d08b3b24 100644 --- a/src/fable-library-ts/MailboxProcessor.ts +++ b/src/fable-library-ts/MailboxProcessor.ts @@ -123,6 +123,8 @@ export function postAndAsyncReply( } const reply = { reply: (res: Reply) => { + // .NET AsyncReplyChannel.Reply ignores a second reply + if (replied) { return; } result = res; replied = true; checkCompletion(); diff --git a/tests/Js/Main/AsyncTests.fs b/tests/Js/Main/AsyncTests.fs index 0fbfb3c7d0..fc17756520 100644 --- a/tests/Js/Main/AsyncTests.fs +++ b/tests/Js/Main/AsyncTests.fs @@ -30,6 +30,16 @@ module private FinalizerRegressions = check 1 cleanups check [|"cancel"|] (terminal.ToArray()) + let nestedPreCancelled () = + use source = new CancellationTokenSource() + source.Cancel() + let cleanups = ResizeArray() + let inner = async.TryFinally(async.Return 42, fun () -> cleanups.Add "inner") + let outer = async.TryFinally(inner, fun () -> cleanups.Add "outer") + let terminal = observe source.Token outer + check [|"outer"|] (cleanups.ToArray()) + check [|"cancel"|] (terminal.ToArray()) + let cancelledAcquisition () = for acquisition in [0; 1; 2] do use source = new CancellationTokenSource() @@ -160,6 +170,7 @@ let private fromGenericCallback (callback: ('T -> unit) -> (exn -> unit) -> (Ope let tests = testList "Async" [ testCase "TryFinally installs cleanup before pre-cancellation" <| FinalizerRegressions.preCancelled + testCase "Pre-cancelled nested TryFinally only runs outer cleanup" <| FinalizerRegressions.nestedPreCancelled testCase "Cancelled acquisition disposes use and use bang resources" <| FinalizerRegressions.cancelledAcquisition testCase "Resource cleanup runs once on success and failure" <| FinalizerRegressions.normalAndFailure testCase "TryFinally propagates downstream continuation errors once" <| FinalizerRegressions.downstreamExceptions @@ -187,6 +198,24 @@ let tests = Async.StartWithContinuations(fromValue 42, (fun value -> equal 42 value; completed <- completed + 1), raise, raise) equal 3 completed + testCase "FromContinuations rejects a second completion" <| fun () -> + let unitCalls = ResizeArray() + let unitWork: Async = Async.FromContinuations(fun (ok, _, _) -> ok (); ok ()) + throwsAnyError (fun () -> + Async.StartWithContinuations(unitWork, (fun () -> unitCalls.Add "success"), (fun _ -> unitCalls.Add "error"), (fun _ -> unitCalls.Add "cancel"))) + equal true (unitCalls.Count <= 1) + equal false (unitCalls.Contains "error") + equal false (unitCalls.Contains "cancel") + + let valueCalls = ResizeArray() + let valueWork = Async.FromContinuations(fun (ok, _, _) -> ok 42; ok 43) + throwsAnyError (fun () -> + Async.StartWithContinuations(valueWork, (fun v -> valueCalls.Add (string v)), (fun _ -> valueCalls.Add "error"), (fun _ -> valueCalls.Add "cancel"))) + equal true (valueCalls.Count <= 1) + equal false (valueCalls.Contains "43") + equal false (valueCalls.Contains "error") + equal false (valueCalls.Contains "cancel") + testCase "Generic unit continuations work through recursive bindings and aliases" <| fun () -> let mutable recurse = true let rec fromValue value = @@ -821,6 +850,19 @@ let tests = equal true ran } + testCaseAsync "MailboxProcessor second Reply is ignored" <| fun () -> + async { + let log = ResizeArray() + let agent = MailboxProcessor>.Start(fun inbox -> async { + let! channel = inbox.Receive() + channel.Reply 1 + channel.Reply 2 + log.Add "body done" }) + let! reply = agent.PostAndAsyncReply(id) + equal 1 reply + equal [|"body done"|] (log.ToArray()) + } + testCase "Async try .. with returns correctly from 'with' branch" <| fun () -> let work = async { try diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 3d92935be8..c41e74086c 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -1,6 +1,7 @@ module Fable.Tests.Async open System +open Util open Util.Testing module private FinalizerRegressions = @@ -1114,3 +1115,76 @@ let ``test generic unit callback adaptation preserves return types`` () = equal 42 (invokeGenericCallback (fun ok -> ok (); 42) ()) equal "unit" (invokeGenericCallback (fun ok -> ok (); "unit") ()) equal "value" (invokeGenericCallback (fun ok -> ok 7; "value") 7) + +type private CallbackRunner = + static member Run(callback: ('T -> unit) -> 'Result, value: 'T, [] extra: int[]) : 'Result = + callback (fun received -> equal value received) + +[] +let ``test generic unit callback adaptation covers local functions and param arrays`` () = + let run (callback: ('T -> unit) -> 'Result) (value: 'T) : 'Result = + callback (fun received -> equal value received) + equal 42 (run (fun ok -> ok (); 42) ()) + equal "value" (run (fun ok -> ok 7; "value") 7) + equal 42 (CallbackRunner.Run((fun ok -> ok (); 42), (), 1, 2)) + equal "value" (CallbackRunner.Run((fun ok -> ok 7; "value"), 7)) + +[] +let ``test Async try with does not run with branch when body succeeds`` () = + let work = async { + try + return 1 + with _ -> + return 99 } + let mutable result = 0 + Async.StartWithContinuations(work, (fun r -> result <- r), ignore, ignore) + equal 1 result + +[] +let ``test Exception thrown by the continuation is not redirected to the with branch`` () = + let work = async { + try + return 1 + with _ -> + return 99 } + let mutable calls = 0 + throwsAnyError (fun () -> + Async.StartWithContinuations( + work, + (fun _ -> calls <- calls + 1; failwith "boom from continuation"), + ignore, + ignore)) + equal 1 calls + +[] +let ``test Exception thrown by the continuation is not redirected through a bind`` () = + let work = async { + let! x = async { return 1 } + return x + 1 } + let calls = ResizeArray() + throwsAnyError (fun () -> + Async.StartWithContinuations( + work, + (fun value -> calls.Add (string value); failwith "boom from continuation"), + (fun _ -> calls.Add "error"), + (fun _ -> calls.Add "cancel"))) + equal [|"2"|] (calls.ToArray()) + +[] +let ``test FromContinuations rejects a second completion`` () = + let unitCalls = ResizeArray() + let unitWork: Async = Async.FromContinuations(fun (ok, _, _) -> ok (); ok ()) + throwsAnyError (fun () -> + Async.StartWithContinuations(unitWork, (fun () -> unitCalls.Add "success"), (fun _ -> unitCalls.Add "error"), (fun _ -> unitCalls.Add "cancel"))) + equal true (unitCalls.Count <= 1) + equal false (unitCalls.Contains "error") + equal false (unitCalls.Contains "cancel") + + let valueCalls = ResizeArray() + let valueWork = Async.FromContinuations(fun (ok, _, _) -> ok 42; ok 43) + throwsAnyError (fun () -> + Async.StartWithContinuations(valueWork, (fun v -> valueCalls.Add (string v)), (fun _ -> valueCalls.Add "error"), (fun _ -> valueCalls.Add "cancel"))) + equal true (valueCalls.Count <= 1) + equal false (valueCalls.Contains "43") + equal false (valueCalls.Contains "error") + equal false (valueCalls.Contains "cancel") diff --git a/tests/Python/TestMailboxProcessor.fs b/tests/Python/TestMailboxProcessor.fs index 6e6ce289b1..e085dbb11c 100644 --- a/tests/Python/TestMailboxProcessor.fs +++ b/tests/Python/TestMailboxProcessor.fs @@ -317,3 +317,17 @@ let ``test MailboxProcessor postAndAsyncReply can compute reply from state`` () let! sum = agent.PostAndAsyncReply(GetSum) equal 65 sum } |> Async.StartImmediate + +[] +let ``test MailboxProcessor second Reply is ignored`` () = + async { + let log = ResizeArray() + let agent = MailboxProcessor>.Start(fun inbox -> async { + let! channel = inbox.Receive() + channel.Reply 1 + channel.Reply 2 + log.Add "body done" }) + let! reply = agent.PostAndAsyncReply(id) + equal 1 reply + equal [|"body done"|] (log.ToArray()) + } |> Async.StartImmediate diff --git a/tests/Python/test_async_runtime.py b/tests/Python/test_async_runtime.py index 272e45380e..b3cf21a3c9 100644 --- a/tests/Python/test_async_runtime.py +++ b/tests/Python/test_async_runtime.py @@ -1,9 +1,11 @@ import asyncio +from collections.abc import Callable from threading import Barrier, Condition, Event, Thread import pytest from fable_library.async_ import await_event, await_task_with_cancellation, sleep, start_with_continuations from fable_library.async_builder import ( + Async, CancellationCallbackError, CancellationToken, IAsyncContext, @@ -392,3 +394,179 @@ def complete(success: bool) -> None: assert all(not thread.is_alive() for thread in threads) assert len(calls) == 1 assert len(rejected) == 1 + + +@pytest.mark.parametrize("combinator", ["try_with", "bind"]) +def test_completed_continuation_exceptions_are_not_redirected(combinator: str) -> None: + from fable_library.async_builder import singleton + + calls: list[str] = [] + defect = ValueError("downstream") + + if combinator == "try_with": + work = singleton.TryWith(singleton.Return(1), lambda _: singleton.Return(99)) + else: + work = singleton.Bind(singleton.Return(1), lambda x: singleton.Return(x + 1)) + + def success(value: int) -> None: + calls.append(f"success:{value}") + raise defect + + with pytest.raises(ValueError) as caught: + start_with_continuations(work, success, lambda _: calls.append("error"), lambda _: calls.append("cancel")) + assert caught.value is defect + assert len(calls) == 1 + + +def test_workflow_errors_still_reach_the_handler_before_completion() -> None: + from fable_library.async_builder import singleton + + def failing_binder(_: int) -> Async[str]: + raise ValueError("binder") + + def handler(error: Exception) -> Async[str]: + return singleton.Return(f"handled:{error}") + + results: list[object] = [] + work = singleton.TryWith(singleton.Bind(singleton.Return(1), failing_binder), handler) + start_with_continuations(work, results.append, lambda e: results.append(("error", e))) + assert results == ["handled:binder"] + + +@pytest.mark.parametrize("channel", ["success", "error", "cancel"]) +def test_unit_from_continuations_propagates_terminal_exceptions(channel: str) -> None: + from fable_library.async_ import from_continuations_unit + from fable_library.async_builder import OperationCanceledError + + calls: list[str] = [] + defect = ValueError("downstream") + + def terminal(value: object = None) -> None: + calls.append(channel) + raise defect + + def callback( + conts: tuple[Callable[[], None], Callable[[Exception], None], Callable[[OperationCanceledError], None]], + ) -> None: + ok, error, cancelled = conts + if channel == "success": + ok() + elif channel == "error": + error(ValueError("body")) + else: + cancelled(OperationCanceledError()) + + with pytest.raises(ValueError) as caught: + start_with_continuations(from_continuations_unit(callback), terminal, terminal, terminal) + assert caught.value is defect + assert calls == [channel] + + +def test_unit_from_continuations_rejects_duplicate_completion() -> None: + from fable_library.async_ import from_continuations_unit + + calls: list[object] = [] + + def callback(conts: tuple[Callable[[], None], object, object]) -> None: + conts[0]() + conts[0]() + + with pytest.raises(RuntimeError, match="invoked more than once"): + start_with_continuations(from_continuations_unit(callback), calls.append) + assert calls == [None] + + +def test_try_finally_claims_concurrent_completion_once() -> None: + from fable_library.async_builder import OperationCanceledError, singleton + + captured: list[IAsyncContext[int]] = [] + calls: list[str] = [] + cleanups: list[int] = [] + barrier = Barrier(2) + + def body(ctx: IAsyncContext[int]) -> None: + captured.append(ctx) + + start_with_continuations( + singleton.TryFinally(body, lambda: cleanups.append(1)), + lambda _: calls.append("success"), + lambda _: calls.append("error"), + lambda _: calls.append("cancel"), + ) + assert len(captured) == 1 + ctx = captured[0] + + def complete(success: bool) -> None: + barrier.wait(5) + if success: + ctx.on_success(42) + else: + ctx.on_cancel(OperationCanceledError()) + + threads = [Thread(target=complete, args=(success,)) for success in (True, False)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(5) + assert all(not thread.is_alive() for thread in threads) + assert len(cleanups) == 1 + assert len(calls) == 1 + + +def test_post_and_async_reply_delivers_once_when_reply_races_with_start() -> None: + from fable_library.mailbox_processor import AsyncReplyChannel, MailboxProcessor + + for _ in range(20): + channels: list[AsyncReplyChannel[int]] = [] + calls: list[object] = [] + failures: list[BaseException] = [] + barrier = Barrier(2) + mailbox: MailboxProcessor[AsyncReplyChannel[int]] = MailboxProcessor(lambda _: lambda _: None) + + def build(channel: AsyncReplyChannel[int]) -> AsyncReplyChannel[int]: + channels.append(channel) + return channel + + work = mailbox.post_and_async_reply(build) + assert len(channels) == 1 + + def start() -> None: + barrier.wait(5) + try: + start_with_continuations(work, calls.append, lambda e: failures.append(e), lambda e: failures.append(e)) + except BaseException as error: + failures.append(error) + + def reply() -> None: + barrier.wait(5) + try: + channels[0].reply(42) + except BaseException as error: + failures.append(error) + + threads = [Thread(target=start), Thread(target=reply)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(5) + assert all(not thread.is_alive() for thread in threads) + assert failures == [] + assert calls == [42] + + +def test_post_and_async_reply_ignores_a_second_reply() -> None: + from fable_library.mailbox_processor import AsyncReplyChannel, MailboxProcessor + + channels: list[AsyncReplyChannel[int]] = [] + calls: list[object] = [] + mailbox: MailboxProcessor[AsyncReplyChannel[int]] = MailboxProcessor(lambda _: lambda _: None) + + def build(channel: AsyncReplyChannel[int]) -> AsyncReplyChannel[int]: + channels.append(channel) + return channel + + work = mailbox.post_and_async_reply(build) + start_with_continuations(work, calls.append, lambda e: calls.append(("error", e))) + channels[0].reply(1) + channels[0].reply(2) + assert calls == [1]