fix: escalate supervisor-monitor failures instead of hanging (#36)

Closes #28.

The monitor in `flatMapActorSupervised` is the top of its supervision tree —
`Actor.spawn`, not `spawnLinked`, so nothing watches it in turn — and its loop
body was bare. A throw from the user's `decide` callback, from `aobv.OnErrorAsync`
on the Escalate path, or from the `unbox`/`Post` forward killed the loop outright.

The stream then hung rather than failed: upstream kept calling `OnNextAsync`, the
posts landed in a mailbox nobody read, and the observer saw neither OnCompleted
nor OnError. On .NET the exception surfaced only on MailboxProcessor's unobserved
Error event; on BEAM not at all.

Guard the body, report the failure downstream, and stop looping — after an
OnError the Rx grammar is closed, so there is nothing left to forward. The
escalation path itself swallows secondary failures, since an exception thrown
while reporting would reproduce the exact bug being fixed.

The guard is a nested `actor { }` rather than `async { }` on purpose: it keeps
`child <- ...` inside the actor process. Confirmed in the generated Erlang — the
loop uses the actor builder's try_with and `child` stays process-dictionary
backed via erlang:get/put.

Test asserts the decider's own exception reaches downstream. Verified red-green:
without the fix it fails on Quill's 5003ms timeout — the silent hang itself —
and passes in 45ms with it.

Adds `isOnError` to Tests.Utils.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: dbrattli

src/ActorInterop.fs

@@ -41,6 +41,17 @@ module internal ActorInterop =
                 let emit value =
                     dispatch.OnNextAsync value |> Async.Start'
 
+                // Terminating downstream must never itself throw: nothing supervises the
+                // monitor, so an exception escaping here would kill it silently — the very
+                // failure this is here to report.
+                let escalate (ex: exn) =
+                    async {
+                        try
+                            do! aobv.OnErrorAsync ex
+                        with _ ->
+                            ()
+                    }
+
                 // Monitor actor: spawns and supervises the child, swaps it on
                 // Restart, and forwards upstream values to it. Routing values
                 // through the monitor means the child is always set before the
@@ -61,25 +72,42 @@ module internal ActorInterop =
                             actor {
                                 let! msg = inbox.Receive()
 
-                                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! aobv.OnErrorAsync ex
-                                    | Directive.Stop -> ()
-                                    | Directive.Restart -> child <- Actor.spawnLinked inbox (handler emit)
-                                // Anything that is not a ChildExited is an upstream value: both
-                                // backends deliver exits pre-converted (BEAM's receive_msg turns
-                                // trapped EXIT signals into ChildExited and drops `normal` ones,
-                                // .NET's spawnLinked posts one from its crash handler), so nothing
-                                // else can reach this mailbox.
-                                | None -> child.Post(unbox<'TSource> msg)
-
-                                return! loop ()
+                                // The monitor is the top of this supervision tree — nothing
+                                // watches it in turn. Left bare, a throw from the user's `decide`,
+                                // from `OnErrorAsync`, or from the cast would kill the loop and
+                                // leave the subscription hanging with no terminal notification:
+                                // upstream keeps posting into a mailbox nobody reads. Report the
+                                // failure downstream and stop instead.
+                                let! alive =
+                                    actor {
+                                        try
+                                            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 -> ()
+                                                | Directive.Restart -> child <- Actor.spawnLinked inbox (handler emit)
+                                            // Anything that is not a ChildExited is an upstream
+                                            // value: both backends deliver exits pre-converted
+                                            // (BEAM's receive_msg turns trapped EXIT signals into
+                                            // ChildExited and drops `normal` ones, .NET's
+                                            // spawnLinked posts one from its crash handler), so
+                                            // nothing else can reach this mailbox.
+                                            | None -> child.Post(unbox<'TSource> msg)
+
+                                            return true
+                                        with ex ->
+                                            do! escalate ex
+                                            return false
+                                    }
+
+                                if alive then
+                                    return! loop ()
                             }
 
                         loop ())

test/ActorInteropTest.fs

@@ -208,6 +208,44 @@ let tests =
               }
           )
 
+          testAsync (
+              "flatMapActorSupervised escalates a throwing decider instead of hanging",
+              async {
+                  // The decider is user code running inside the monitor loop. Before the monitor
+                  // caught its own failures, a throw here killed the supervisor silently and the
+                  // subscription hung forever with no terminal notification — this test would
+                  // then fail on Quill's timeout rather than on the assertion.
+                  let xs = fromNotification [ OnNext 1 ]
+                  let deciderError = System.InvalidOperationException("decider boom")
+
+                  let result =
+                      xs
+                      |> Reactive.flatMapActorSupervised (fun _ -> raise deciderError) (fun emit inbox ->
+                          let rec loop () =
+                              actor {
+                                  let! x = inbox.Receive()
+                                  failwith "child boom"
+                                  emit x
+                                  return! loop ()
+                              }
+
+                          loop ())
+
+                  let obv = TestObserver<int>()
+                  let! _sub = result.SubscribeAsync obv
+                  do! obv.WaitUntil(List.exists isOnError)
+
+                  // The decider's own failure is what reaches downstream, not the child's.
+                  let reported =
+                      obv.Notifications
+                      |> Seq.tryPick (function
+                          | OnError e -> Some e.Message
+                          | _ -> None)
+
+                  assertThat reported (isEqualTo (Some "decider boom"))
+              }
+          )
+
           testAsync (
               "flatMapActor forwards actor crash as OnError",
               async {

test/Utils.fs

@@ -30,6 +30,11 @@ let isOnCompleted (n: Notification<'a>) =
     | OnCompleted -> true
     | _ -> false
 
+let isOnError (n: Notification<'a>) =
+    match n with
+    | OnError _ -> true
+    | _ -> false
+
 /// Immutable snapshot of the observer state, returned across the process boundary.
 type private Snapshot<'a> =
     { Notifications: Notification<'a> list