From 7fcab5d50d4f2376e811787864ac66203d0921e1 Mon Sep 17 00:00:00 2001 From: Mangel Maxime Date: Thu, 15 Oct 2026 19:52:48 +0200 Subject: [PATCH] fix(js/ts/python): settle async continuations once --- .../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 | 179 +++++++++++++++++- 14 files changed, 482 insertions(+), 88 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 74c13bfd9b..21adf2b390 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -33,7 +33,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 @@ -235,13 +235,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) @@ -255,7 +254,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: @@ -274,13 +272,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]: @@ -344,12 +340,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 a79e9c983a..70302d8516 100644 --- a/src/fable-library-py/fable_library/async_builder.py +++ b/src/fable-library-py/fable_library/async_builder.py @@ -227,7 +227,7 @@ class ScheduledItem: class Trampoline: - __slots__ = "call_count", "lock", "running" + __slots__ = "call_count", "completed", "lock", "running" MaxTrampolineCallCount = 75 # Max recursion depth: 1000 @@ -235,6 +235,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: @@ -250,13 +252,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]: @@ -269,7 +275,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) @@ -284,10 +291,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_) @@ -345,12 +353,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: @@ -359,11 +375,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: @@ -372,10 +385,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() @@ -384,8 +395,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 @@ -394,9 +404,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, @@ -414,7 +426,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 21e277d81c..a362819ff1 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 = @@ -1106,3 +1107,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 48432db188..a655e67685 100644 --- a/tests/Python/test_async_runtime.py +++ b/tests/Python/test_async_runtime.py @@ -1,9 +1,10 @@ import asyncio +from collections.abc import Callable from threading import Barrier, Condition, Event, Thread import pytest from fable_library.async_ import await_event, sleep, start_with_continuations -from fable_library.async_builder import CancellationCallbackError, CancellationToken, IAsyncContext, Trampoline +from fable_library.async_builder import Async, CancellationCallbackError, CancellationToken, IAsyncContext, Trampoline from fable_library.event import Event as FableEvent @@ -345,3 +346,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]