fix(asyncstatemachine): fixes not awaiting or asyncSpawning futures (#1033)

- adds a break in scheduler when CancelledError is caught
- tracks asyncSpawned state.run, so that it can be cancelled during stop
- removes usages of `then`
- ensures that no exceptions are leaked from async procs

codex/utils/asyncstatemachine.nim

@@ -1,12 +1,11 @@
 import std/sugar
 import pkg/questionable
 import pkg/chronos
-import pkg/upraises
 import ../logutils
 import ./then
 import ./trackedfutures
 
-push: {.upraises:[].}
+{.push raises:[].}
 
 type
   Machine* = ref object of RootObj
@@ -17,7 +16,7 @@ type
     trackedFutures: TrackedFutures
   State* = ref object of RootObj
   Query*[T] = proc(state: State): T
-  Event* = proc(state: State): ?State {.gcsafe, upraises:[].}
+  Event* = proc(state: State): ?State {.gcsafe, raises:[].}
 
 logScope:
   topics = "statemachine"
@@ -58,29 +57,31 @@ proc onError(machine: Machine, error: ref CatchableError): Event =
   return proc (state: State): ?State =
     state.onError(error)
 
-proc run(machine: Machine, state: State) {.async.} =
-  if next =? await state.run(machine):
-    machine.schedule(Event.transition(state, next))
+proc run(machine: Machine, state: State) {.async: (raises:[]).} =
+  try:
+    if next =? await state.run(machine):
+      machine.schedule(Event.transition(state, next))
+  except CancelledError:
+    discard # do not propagate
+  except CatchableError as e:
+    machine.schedule(machine.onError(e))
 
-proc scheduler(machine: Machine) {.async.} =
-  var running: Future[void]
+proc scheduler(machine: Machine) {.async: (raises: []).} =
+  var running: Future[void].Raising([])
   while machine.started:
-    let event = await machine.scheduled.get().track(machine)
-    if next =? event(machine.state):
-      if not running.isNil and not running.finished:
-        trace "cancelling current state", state = $machine.state
-        await running.cancelAndWait()
-      let fromState = if machine.state.isNil: "<none>" else: $machine.state
-      machine.state = next
-      debug "enter state", state = fromState & " => " & $machine.state
-      running = machine.run(machine.state)
-      running
-        .track(machine)
-        .cancelled(proc() = trace "state.run cancelled, swallowing", state = $machine.state)
-        .catch(proc(err: ref CatchableError) =
-          trace "error caught in state.run, calling state.onError", state = $machine.state
-          machine.schedule(machine.onError(err))
-        )
+    try:
+      let event = await machine.scheduled.get()
+      if next =? event(machine.state):
+        if not running.isNil and not running.finished:
+          trace "cancelling current state", state = $machine.state
+          await running.cancelAndWait()
+        let fromState = if machine.state.isNil: "<none>" else: $machine.state
+        machine.state = next
+        debug "enter state", state = fromState & " => " & $machine.state
+        running = machine.run(machine.state)
+        asyncSpawn running.track(machine)
+    except CancelledError:
+      break # do not propagate bc it is asyncSpawned
 
 proc start*(machine: Machine, initialState: State) =
   if machine.started:
@@ -90,13 +91,8 @@ proc start*(machine: Machine, initialState: State) =
     machine.scheduled = newAsyncQueue[Event]()
 
   machine.started = true
-  try:
-    discard machine.scheduler().track(machine)
-    machine.schedule(Event.transition(machine.state, initialState))
-  except CancelledError as e:
-    discard
-  except CatchableError as e:
-    error("Error in scheduler", error = e.msg)
+  asyncSpawn machine.scheduler().track(machine)
+  machine.schedule(Event.transition(machine.state, initialState))
 
 proc stop*(machine: Machine) {.async.} =
   if not machine.started: