From 3cd95305cf1d612869777e40402cdc957a5bcbfa Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Sun, 18 Oct 2026 23:06:16 +0200 Subject: [PATCH 1/2] fix: distinguish supervised payloads from child exits --- CLAUDE.md | 10 ++--- src/ActorInterop.fs | 21 ++++++---- test/Main.fs | 1 + test/SupervisedPayloadTest.fs | 75 +++++++++++++++++++++++++++++++++++ test/Tests.fsproj | 1 + 5 files changed, 96 insertions(+), 12 deletions(-) create mode 100644 test/SupervisedPayloadTest.fs diff --git a/CLAUDE.md b/CLAUDE.md index ae58cfb..aa40da2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -104,14 +104,14 @@ CI executes the shared Scriptorium suite on .NET, JavaScript, Python, and BEAM. Every target is a required runtime gate; successful Fable compilation alone is insufficient. The Fable builds exclude the seven .NET-only AsyncSeq integration tests. -Runtime results measured on 2026-10-17 with Fable 5.22.0 and Fable.Actor 5.0.0-rc.18: +Runtime results measured on 2026-10-18 with Fable 5.22.0 and Fable.Actor 5.0.0-rc.18: | Target | Passed | Failed | Total | | ------ | ------ | ------ | ----- | -| .NET | 111 | 0 | 111 | -| JavaScript | 104 | 0 | 104 | -| Python | 104 | 0 | 104 | -| BEAM | 104 | 0 | 104 | +| .NET | 119 | 0 | 119 | +| JavaScript | 112 | 0 | 112 | +| Python | 112 | 0 | 112 | +| BEAM | 112 | 0 | 112 | Fable 5.22.0 includes the BEAM lazy-sequence portability fix. Direct merge, concat, and mapped-sequence actor tests now run against the published runtime without deferred rebuilding. diff --git a/src/ActorInterop.fs b/src/ActorInterop.fs index 905f6c9..92354d0 100644 --- a/src/ActorInterop.fs +++ b/src/ActorInterop.fs @@ -10,6 +10,10 @@ module internal ActorInterop = type private SupervisorStop = SupervisorStop + // decision: tags forwarded values before the shared monitor sees them so BEAM never tests user records as child exits + // invariant: the monitor classifies only the outer envelope and never inspects a source value for exit markers + type private ForwardedValue = ReactiveForwardedValue of obj + /// Wraps an Actor> as an IAsyncObserver<'T>. /// Posts each notification to the actor. The actor owns its own lifecycle. let toObserver (actor: Actor>) : IAsyncObserver<'T> = @@ -98,6 +102,10 @@ module internal ActorInterop = match msg with | :? SupervisorStop -> () | _ when closed.Value <> 0 -> return! loop () + | :? ForwardedValue as forwarded -> + let (ReactiveForwardedValue value) = forwarded + child.Actor.Post(unbox<'TSource> value) + return! loop () | _ -> match Actor.tryAsChildExited msg with | Some exited -> @@ -116,9 +124,7 @@ module internal ActorInterop = child <- spawnChild () return! loop () - | None -> - child.Actor.Post(unbox<'TSource> msg) - return! loop () + | None -> failwith "Unexpected supervised actor message" } actor { @@ -157,7 +163,8 @@ module internal ActorInterop = failwith "Supervisor child cleanup failed" } - let cleanup (sourceDisp: IReactiveDisposable) () = + // decision: passes an explicit unit closure to disposal so Python does not call a partially applied function + let cleanup (sourceDisp: IReactiveDisposable) = ActorLifetime.cleanupAll [ stopMonitor @@ -174,7 +181,7 @@ module internal ActorInterop = member _.OnNextAsync x = async { if closed.Value = 0 then - monitor.Actor.Post(box x) + monitor.Actor.Post(box (ReactiveForwardedValue(box x))) } member _.OnErrorAsync err = @@ -192,10 +199,10 @@ module internal ActorInterop = try let! sourceDisp = source.SubscribeAsync obv - return ActorLifetime.disposable (cleanupTimeout * 4) close (cleanup sourceDisp) + return ActorLifetime.disposable (cleanupTimeout * 4) close (fun () -> cleanup sourceDisp) with ex -> close () - do! cleanup AsyncDisposable.Empty () + do! cleanup AsyncDisposable.Empty return raise ex } diff --git a/test/Main.fs b/test/Main.fs index 4398853..960b51c 100644 --- a/test/Main.fs +++ b/test/Main.fs @@ -42,6 +42,7 @@ let main _argv = Tests.Timeshift.tests Tests.Debounce.tests Tests.ActorInterop.tests + Tests.SupervisedPayload.tests Tests.Probe.tests ] ) diff --git a/test/SupervisedPayloadTest.fs b/test/SupervisedPayloadTest.fs new file mode 100644 index 0000000..528b078 --- /dev/null +++ b/test/SupervisedPayloadTest.fs @@ -0,0 +1,75 @@ +module Tests.SupervisedPayload + +open Scriptorium.Quill +open Scriptorium.Nib.Assertion +open type Scriptorium.Quill.Test + +open Fable.Actor +open Fable.Actor.Types +open Fable.Reactive +open Tests.Utils + +type private ExitShapedValue = { Pid: int; Reason: string } + +let private verifyPayloads (values: obj list) = + async { + let source = + Reactive.create (fun observer -> + async { + for value in values do + do! observer.OnNextAsync value + + return AsyncDisposable.Empty + }) + + let result = + source + |> Reactive.flatMapActorSupervised (fun _ -> Directive.Escalate) (fun emit inbox -> + let rec loop () = + actor { + let! value = inbox.Receive() + emit value + return! loop () + } + + loop ()) + + let observer = TestObserver() + let! subscription = result.SubscribeAsync observer + + do! + observer.WaitUntil(fun notifications -> + List.filter isOnNext notifications |> List.length = List.length values + || List.exists isOnError notifications) + + let actual = + observer.Notifications + |> Seq.choose (function + | OnNext value -> Some value + | _ -> None) + |> Seq.toList + + assertThat (observer.Notifications |> Seq.exists isOnError) isFalse + assertThat actual (isEqualTo values) + + do! subscription.DisposeAsync() + } + +let tests = + testList ( + "SupervisedPayload", + [ + testAsync ( + "exit-shaped records reach the child as values", + verifyPayloads [ box { Pid = 42; Reason = "payload" }; box { Pid = 43; Reason = "next" } ] + ) + testAsync ( + "ChildExited records can be upstream values", + verifyPayloads [ box ({ Pid = box 42; Reason = box "payload" }: ChildExited) ] + ) + testAsync ( + "null, primitive, and tuple payloads retain order", + verifyPayloads [ null; box 0; box false; box "payload"; box ("reactive_forwarded_value", 42) ] + ) + ] + ) diff --git a/test/Tests.fsproj b/test/Tests.fsproj index cf8f8dc..d54b0a2 100644 --- a/test/Tests.fsproj +++ b/test/Tests.fsproj @@ -27,6 +27,7 @@ + From a01f2cecf95138226d0f3c0c856c69e627c3f59d Mon Sep 17 00:00:00 2001 From: Dag Brattli Date: Sun, 18 Oct 2026 23:10:11 +0200 Subject: [PATCH 2/2] refactor: split supervised actor routing and cleanup --- src/ActorInterop.fs | 356 +++++++++++++++++++++++++------------------- 1 file changed, 206 insertions(+), 150 deletions(-) diff --git a/src/ActorInterop.fs b/src/ActorInterop.fs index 92354d0..6485a61 100644 --- a/src/ActorInterop.fs +++ b/src/ActorInterop.fs @@ -28,6 +28,195 @@ module internal ActorInterop = let subscribeActor (actor: Actor>) (source: IAsyncObservable<'T>) : Async = source.SubscribeAsync(toObserver actor) + let private cleanupTimeout = 5000 + + type private SupervisedContext<'TSource, 'TResult> = + { + Closed: ActorLifetime.Flag + MonitorFailed: ActorLifetime.Flag + ChildCancellation: System.Threading.CancellationTokenSource + Output: ActorLifetime.Worker<'TResult> + Downstream: IAsyncObserver<'TResult> + Decide: exn -> Directive + Handler: ('TResult -> unit) -> ActorInbox<'TSource> -> ActorOp + } + + let private spawnOutput (closed: ActorLifetime.Flag) (downstream: IAsyncObserver<'T>) = + ActorLifetime.spawn None ignore (fun inbox -> + let rec loop () = + actor { + let! value = inbox.Receive() + + if closed.Value = 0 then + try + do! downstream.OnNextAsync value + with ex -> + do! downstream.OnErrorAsync ex + + return! loop () + } + + loop ()) + + // invariant: escalation never throws back into the unsupervised monitor + let private escalate (context: SupervisedContext<_, _>) ex = + async { + if context.Closed.Value = 0 then + try + do! context.Downstream.OnErrorAsync ex + with _ -> + () + } + + let private childExitError (exited: ChildExited) = + match exited.Reason with + | :? exn as ex -> ex + | reason -> ProcessExitException(sprintf "%A" reason) + + let private handleChildExit + (context: SupervisedContext<'TSource, 'TResult>) + (spawnChild: unit -> ActorLifetime.Worker<'TSource>) + (child: ActorLifetime.Worker<'TSource>) + (exited: ChildExited) + : ActorOp option> = + actor { + let ex = childExitError exited + + match context.Decide ex with + | Directive.Escalate -> + do! escalate context ex + return None + | Directive.Stop -> return Some child + | Directive.Restart -> + do! child.Stop cleanupTimeout + + if context.Closed.Value = 0 then + return Some(spawnChild ()) + else + return Some child + } + + let private routeMonitorMessage + (context: SupervisedContext<'TSource, 'TResult>) + (spawnChild: unit -> ActorLifetime.Worker<'TSource>) + (child: ActorLifetime.Worker<'TSource>) + (msg: obj) + : ActorOp option> = + actor { + match msg with + | :? SupervisorStop -> return None + | _ when context.Closed.Value <> 0 -> return Some child + | :? ForwardedValue as forwarded -> + let (ReactiveForwardedValue value) = forwarded + child.Actor.Post(unbox<'TSource> value) + return Some child + | _ -> + match Actor.tryAsChildExited msg with + | Some exited -> return! handleChildExit context spawnChild child exited + | None -> return failwith "Unexpected supervised actor message" + } + + let private startSupervisedMonitor (context: SupervisedContext<'TSource, 'TResult>) = + ActorLifetime.spawn None (fun _ -> context.MonitorFailed.Set 1) (fun inbox -> + Actor.trapExits () + + let emit value = + if context.Closed.Value = 0 then + context.Output.Actor.Post value + + let spawnChild () = + ActorLifetime.spawnWithToken + context.ChildCancellation.Token + (Some inbox.Reference) + (fun ex -> inbox.Post(box ({ Pid = null; Reason = box ex }: ChildExited))) + (context.Handler emit) + + let mutable child = spawnChild () + + let rec loop () = + actor { + let! msg = inbox.Receive() + let! nextChild = routeMonitorMessage context spawnChild child msg + + match nextChild with + | None -> () + | Some next -> + child <- next + return! loop () + } + + actor { + try + do! loop () + with ex -> + do! escalate context ex + + try + do! child.Stop cleanupTimeout + with _ -> + context.MonitorFailed.Set 1 + }) + + let private closeSupervision (context: SupervisedContext<_, _>) = + context.Closed.Set 1 + context.ChildCancellation.Cancel() + + let private stopMonitor (context: SupervisedContext<_, _>) (monitor: ActorLifetime.Worker) = + async { + monitor.Actor.Post(box SupervisorStop) + + try + do! ActorLifetime.waitUntil cleanupTimeout "Supervisor did not acknowledge shutdown" monitor.IsStopped + with ex -> + do! monitor.Stop cleanupTimeout + return raise ex + + do! monitor.Stop cleanupTimeout + + if context.MonitorFailed.Value <> 0 then + failwith "Supervisor child cleanup failed" + } + + let private cleanupSupervision + (context: SupervisedContext<_, _>) + (monitor: ActorLifetime.Worker) + (sourceDisp: IReactiveDisposable) + = + ActorLifetime.cleanupAll + [ + (fun () -> stopMonitor context monitor) + (fun () -> context.Output.Stop cleanupTimeout) + (fun () -> + ActorLifetime.cleanupStep + cleanupTimeout + "Source disposal failed or timed out" + (sourceDisp.DisposeAsync())) + ] + + let private supervisedObserver + (context: SupervisedContext<'TSource, 'TResult>) + (monitor: ActorLifetime.Worker) + = + { new IAsyncObserver<'TSource> with + member _.OnNextAsync x = + async { + if context.Closed.Value = 0 then + monitor.Actor.Post(box (ReactiveForwardedValue(box x))) + } + + member _.OnErrorAsync err = + async { + if context.Closed.Value = 0 then + do! context.Downstream.OnErrorAsync err + } + + member _.OnCompletedAsync() = + async { + if context.Closed.Value = 0 then + do! context.Downstream.OnCompletedAsync() + } + } + /// For each upstream item, posts it to a supervised per-subscription actor. /// The actor emits downstream via an emit callback provided at spawn time. /// Terminal events bypass the actor and go directly to downstream. @@ -47,162 +236,29 @@ module internal ActorInterop = let subscribeAsync (aobv: IAsyncObserver<'TResult>) = async { let closed = ActorLifetime.Flag(0) - let monitorFailed = ActorLifetime.Flag(0) - let childCancellation = new System.Threading.CancellationTokenSource() - let cleanupTimeout = 5000 - - let output = - ActorLifetime.spawn None ignore (fun inbox -> - let rec loop () = - actor { - let! value = inbox.Receive() - - if closed.Value = 0 then - try - do! aobv.OnNextAsync value - with ex -> - do! aobv.OnErrorAsync ex - - return! loop () - } - - loop ()) - - let emit value = - if closed.Value = 0 then - output.Actor.Post value - - // invariant: escalation never throws back into the unsupervised monitor - let escalate ex = - async { - if closed.Value = 0 then - try - do! aobv.OnErrorAsync ex - with _ -> - () - } - - let monitor = - ActorLifetime.spawn None (fun _ -> monitorFailed.Set 1) (fun inbox -> - Actor.trapExits () - - let spawnChild () = - ActorLifetime.spawnWithToken - childCancellation.Token - (Some inbox.Reference) - (fun ex -> inbox.Post(box ({ Pid = null; Reason = box ex }: ChildExited))) - (handler emit) - - let mutable child = spawnChild () - - let rec loop () = - actor { - let! msg = inbox.Receive() - - match msg with - | :? SupervisorStop -> () - | _ when closed.Value <> 0 -> return! loop () - | :? ForwardedValue as forwarded -> - let (ReactiveForwardedValue value) = forwarded - child.Actor.Post(unbox<'TSource> value) - return! loop () - | _ -> - match Actor.tryAsChildExited msg with - | Some exited -> - let ex = - match exited.Reason with - | :? exn as e -> e - | r -> ProcessExitException(sprintf "%A" r) - - match decide ex with - | Directive.Escalate -> do! escalate ex - | Directive.Stop -> return! loop () - | Directive.Restart -> - do! child.Stop cleanupTimeout - - if closed.Value = 0 then - child <- spawnChild () - - return! loop () - | None -> failwith "Unexpected supervised actor message" - } - - actor { - try - do! loop () - with ex -> - do! escalate ex - - try - do! child.Stop cleanupTimeout - with _ -> - monitorFailed.Set 1 - }) - - let close () = - closed.Set 1 - childCancellation.Cancel() - - let stopMonitor () = - async { - monitor.Actor.Post(box SupervisorStop) - - try - do! - ActorLifetime.waitUntil - cleanupTimeout - "Supervisor did not acknowledge shutdown" - monitor.IsStopped - with ex -> - do! monitor.Stop cleanupTimeout - return raise ex - - do! monitor.Stop cleanupTimeout - if monitorFailed.Value <> 0 then - failwith "Supervisor child cleanup failed" + let context = + { + Closed = closed + MonitorFailed = ActorLifetime.Flag(0) + ChildCancellation = new System.Threading.CancellationTokenSource() + Output = spawnOutput closed aobv + Downstream = aobv + Decide = decide + Handler = handler } - // decision: passes an explicit unit closure to disposal so Python does not call a partially applied function - let cleanup (sourceDisp: IReactiveDisposable) = - ActorLifetime.cleanupAll - [ - stopMonitor - (fun () -> output.Stop cleanupTimeout) - (fun () -> - ActorLifetime.cleanupStep - cleanupTimeout - "Source disposal failed or timed out" - (sourceDisp.DisposeAsync())) - ] - - let obv = - { new IAsyncObserver<'TSource> with - member _.OnNextAsync x = - async { - if closed.Value = 0 then - monitor.Actor.Post(box (ReactiveForwardedValue(box x))) - } - - member _.OnErrorAsync err = - async { - if closed.Value = 0 then - do! aobv.OnErrorAsync err - } - - member _.OnCompletedAsync() = - async { - if closed.Value = 0 then - do! aobv.OnCompletedAsync() - } - } + let monitor = startSupervisedMonitor context try - let! sourceDisp = source.SubscribeAsync obv - return ActorLifetime.disposable (cleanupTimeout * 4) close (fun () -> cleanup sourceDisp) + let! sourceDisp = source.SubscribeAsync(supervisedObserver context monitor) + + return + ActorLifetime.disposable (cleanupTimeout * 4) (fun () -> closeSupervision context) (fun () -> + cleanupSupervision context monitor sourceDisp) with ex -> - close () - do! cleanup AsyncDisposable.Empty + closeSupervision context + do! cleanupSupervision context monitor AsyncDisposable.Empty return raise ex }