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..6485a61 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> = @@ -24,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. @@ -43,159 +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 () - | _ -> - 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 -> - child.Actor.Post(unbox<'TSource> msg) - return! loop () - } - - 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 } - 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 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 (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 } 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 @@ +