Reworked parts of internal error handling (#4145)

* Do not swallow errors

* Add test for #4004

* WIP fix tests?

* skip the unhandled rejection test

* wrap observer calls with reportUnhandledError

* move towards status unification

* update test

* roll back status changes

* add comment to reportUnhandledError

* make sure that root reports errors with no subscribers

* tweak the logic

* avoid rethrowing

* add more coverage

* remove redundant timeout

* Changeset

* Update .changeset/chilled-bobcats-lay.md

Co-authored-by: Mateusz Burzyński <mateuszburzynski@gmail.com>

---------

Co-authored-by: Mateusz Burzyński <mateuszburzynski@gmail.com>

.changeset/chilled-bobcats-lay.md

@@ -0,0 +1,16 @@
+---
+'xstate': minor
+---
+
+Significant improvements to error handling have been made:
+
+- Actors will no longer crash when an error is thrown in an observer (`actor.subscribe(observer)`).
+- Errors will be handled by observer's `.error()` handler:
+  ```ts
+  actor.subscribe({
+    error: (error) => {
+      // handle error
+    }
+  });
+  ```
+- If an observer does not have an error handler, the error will be thrown in a clear stack so bug tracking services can collect it.

jest.config.js

@@ -1,5 +1,4 @@
 const { constants } = require('jest-config');
-const os = require('os');
 
 /**
  * @type {import('@jest/types').Config.InitialOptions}

packages/core/src/State.ts

@@ -83,6 +83,7 @@ export class State<
    * The done data of the top-level finite state.
    */
   public output: any; // TODO: add an explicit type for `output`
+  public error: unknown;
   public context: TContext;
   public historyValue: Readonly<HistoryValue<TContext, TEvent>> = {};
   public _internalQueue: Array<TEvent>;
@@ -164,6 +165,7 @@ export class State<
     this.tags = new Set(flatten(this.configuration.map((sn) => sn.tags)));
     this.done = config.done ?? false;
     this.output = config.output;
+    this.error = config.error;
   }
 
   /**

packages/core/src/StateMachine.ts

@@ -1,6 +1,6 @@
 import { error, createInitEvent, assign } from './actions.ts';
 import { STATE_DELIMITER } from './constants.ts';
-import { getPersistedState, State } from './State.ts';
+import { cloneState, getPersistedState, State } from './State.ts';
 import { StateNode } from './StateNode.ts';
 import { interpret } from './interpreter.ts';
 import {
@@ -213,7 +213,9 @@ export class StateMachine<
       isErrorEvent(event) &&
       !state.nextEvents.some((nextEvent) => nextEvent === event.type)
     ) {
-      throw event.data;
+      return cloneState(state, {
+        error: event.data
+      });
     }
 
     const { state: nextState } = macrostep(state, event, actorCtx);
@@ -318,20 +320,11 @@ export class StateMachine<
   }
 
   public start(
-    state: State<TContext, TEvent, TActor, TResolvedTypesMeta>,
-    actorCtx: ActorContext<
-      TEvent,
-      State<TContext, TEvent, TActor, TResolvedTypesMeta>
-    >
+    state: State<TContext, TEvent, TActor, TResolvedTypesMeta>
   ): void {
     Object.values(state.children).forEach((child: any) => {
       if (child.status === 0) {
-        try {
-          child.start?.();
-        } catch (err) {
-          // TODO: unify error handling when child starts
-          actorCtx.self.send(error(child.id, err) as unknown as TEvent);
-        }
+        child.start?.();
       }
     });
   }
@@ -379,7 +372,9 @@ export class StateMachine<
   }
 
   public getStatus(state: State<TContext, TEvent, TActor, TResolvedTypesMeta>) {
-    return state.done
+    return state.error
+      ? { status: 'error', data: state.error }
+      : state.done
       ? { status: 'done', data: state.output }
       : { status: 'active' };
   }

packages/core/src/interpreter.ts

@@ -3,6 +3,7 @@ import { Mailbox } from './Mailbox.ts';
 import { doneInvoke, error } from './actions.ts';
 import { stopSignalType } from './actors/index.ts';
 import { devToolsAdapter } from './dev/index.ts';
+import { reportUnhandledError } from './reportUnhandledError.ts';
 import { symbolObservable } from './symbolObservable.ts';
 import { createSystem } from './system.ts';
 import {
@@ -201,22 +202,27 @@ export class Interpreter<
     }
 
     for (const observer of this.observers) {
-      observer.next?.(snapshot);
+      // TODO: should observers be notified in case of the error?
+      try {
+        observer.next?.(snapshot);
+      } catch (err) {
+        reportUnhandledError(err);
+      }
     }
 
     const status = this.logic.getStatus?.(state);
 
     switch (status?.status) {
       case 'done':
         this._stopProcedure();
+        this._complete();
         this._doneEvent = doneInvoke(this.id, status.data);
         this._parent?.send(this._doneEvent as any);
-        this._complete();
         break;
       case 'error':
         this._stopProcedure();
-        this._parent?.send(error(this.id, status.data));
         this._error(status.data);
+        this._parent?.send(error(this.id, status.data));
         break;
     }
   }
@@ -240,11 +246,14 @@ export class Interpreter<
       completeListener
     );
 
-    this.observers.add(observer);
-
-    if (this.status === ActorStatus.Stopped) {
-      observer.complete?.();
-      this.observers.delete(observer);
+    if (this.status !== ActorStatus.Stopped) {
+      this.observers.add(observer);
+    } else {
+      try {
+        observer.complete?.();
+      } catch (err) {
+        reportUnhandledError(err);
+      }
     }
 
     return {
@@ -269,8 +278,28 @@ export class Interpreter<
     }
     this.status = ActorStatus.Running;
 
+    const status = this.logic.getStatus?.(this._state);
+
+    switch (status?.status) {
+      case 'done':
+        // a state machine can be "done" upon intialization (it could reach a final state using initial microsteps)
+        // we still need to complete observers, flush deferreds etc
+        this.update(this._state);
+      // fallthrough
+      case 'error':
+        // TODO: rethink cleanup of observers, mailbox, etc
+        return this;
+    }
+
     if (this.logic.start) {
-      this.logic.start(this._state, this._actorContext);
+      try {
+        this.logic.start(this._state, this._actorContext);
+      } catch (err) {
+        this._stopProcedure();
+        this._error(err);
+        this._parent?.send(error(this.id, err));
+        return this;
+      }
     }
 
     // TODO: this notifies all subscribers but usually this is redundant
@@ -288,29 +317,29 @@ export class Interpreter<
   }
 
   private _process(event: TEvent) {
+    // TODO: reexamine what happens when an action (or a guard or smth) throws
+    let nextState;
+    let caughtError;
     try {
-      const nextState = this.logic.transition(
-        this._state,
-        event,
-        this._actorContext
-      );
+      nextState = this.logic.transition(this._state, event, this._actorContext);
+    } catch (err) {
+      // we wrap it in a box so we can rethrow it later even if falsy value gets caught here
+      caughtError = { err };
+    }
 
-      this.update(nextState);
+    if (caughtError) {
+      const { err } = caughtError;
 
-      if (event.type === stopSignalType) {
-        this._stopProcedure();
-        this._complete();
-      }
-    } catch (err) {
-      // TODO: properly handle errors
-      if (this.observers.size > 0) {
-        this.observers.forEach((observer) => {
-          observer.error?.(err);
-        });
-        this.stop();
-      } else {
-        throw err;
-      }
+      this._stopProcedure();
+      this._error(err);
+      this._parent?.send(error(this.id, err));
+      return;
+    }
+
+    this.update(nextState);
+    if (event.type === stopSignalType) {
+      this._stopProcedure();
+      this._complete();
     }
   }
 
@@ -339,15 +368,36 @@ export class Interpreter<
   }
   private _complete(): void {
     for (const observer of this.observers) {
-      observer.complete?.();
+      try {
+        observer.complete?.();
+      } catch (err) {
+        reportUnhandledError(err);
+      }
     }
     this.observers.clear();
   }
-  private _error(data: any): void {
+  private _error(err: unknown): void {
+    if (!this.observers.size) {
+      if (!this._parent) {
+        reportUnhandledError(err);
+      }
+      return;
+    }
+    let reportError = false;
+
     for (const observer of this.observers) {
-      observer.error?.(data);
+      const errorListener = observer.error;
+      reportError ||= !errorListener;
+      try {
+        errorListener?.(err);
+      } catch (err2) {
+        reportUnhandledError(err2);
+      }
     }
     this.observers.clear();
+    if (reportError) {
+      reportUnhandledError(err);
+    }
   }
   private _stopProcedure(): this {
     if (this.status !== ActorStatus.Running) {

packages/core/src/reportUnhandledError.ts

@@ -0,0 +1,12 @@
+/**
+ * This function makes sure that unhandled errors are thrown in a separate macrotask.
+ * It allows those errors to be detected by global error handlers and reported to bug tracking services
+ * without interrupting our own stack of execution.
+ *
+ * @param err error to be thrown
+ */
+export function reportUnhandledError(err: unknown) {
+  setTimeout(() => {
+    throw err;
+  });
+}

packages/core/src/types.ts

@@ -1323,6 +1323,7 @@ export interface StateConfig<
   children: Record<string, ActorRef<any>>;
   done?: boolean;
   output?: any;
+  error?: unknown;
   tags?: Set<string>;
   machine?: StateMachine<TContext, TEvent, any, any, any, any>;
   _internalQueue?: Array<TEvent>;
@@ -1767,7 +1768,7 @@ export type AnyActorSystem = ActorSystem<any>;
 
 export type PersistedMachineState<TState extends AnyState> = Pick<
   TState,
-  'value' | 'output' | 'context' | 'done' | 'historyValue'
+  'value' | 'output' | 'error' | 'context' | 'done' | 'historyValue'
 > & {
   children: {
     [K in keyof TState['children']]: {

packages/core/src/utils.ts

@@ -423,16 +423,15 @@ export function toObserver<T>(
   errorHandler?: (error: any) => void,
   completionHandler?: () => void
 ): Observer<T> {
-  const noop = () => {};
   const isObserver = typeof nextHandler === 'object';
-  const self = isObserver ? nextHandler : null;
+  const self = isObserver ? nextHandler : undefined;
 
   return {
-    next: ((isObserver ? nextHandler.next : nextHandler) || noop).bind(self),
-    error: ((isObserver ? nextHandler.error : errorHandler) || noop).bind(self),
-    complete: (
-      (isObserver ? nextHandler.complete : completionHandler) || noop
-    ).bind(self)
+    next: (isObserver ? nextHandler.next : nextHandler)?.bind(self),
+    error: (isObserver ? nextHandler.error : errorHandler)?.bind(self),
+    complete: (isObserver ? nextHandler.complete : completionHandler)?.bind(
+      self
+    )
   };
 }
 

packages/core/test/actions.test.ts

@@ -2581,13 +2581,22 @@ describe('forwardTo()', () => {
       }
     });
 
-    const service = interpret(machine).start();
+    const errorSpy = jest.fn();
 
-    expect(() =>
-      service.send({ type: 'TEST' })
-    ).toThrowErrorMatchingInlineSnapshot(
-      `"Attempted to forward event to undefined actor. This risks an infinite loop in the sender."`
-    );
+    const actorRef = interpret(machine);
+    actorRef.subscribe({
+      error: errorSpy
+    });
+    actorRef.start();
+    actorRef.send({ type: 'TEST' });
+
+    expect(errorSpy).toMatchMockCallsInlineSnapshot(`
+      [
+        [
+          [Error: Attempted to forward event to undefined actor. This risks an infinite loop in the sender.],
+        ],
+      ]
+    `);
   });
 });
 

packages/core/test/actorLogic.test.ts

@@ -174,6 +174,7 @@ describe('promise logic (fromPromise)', () => {
       return Promise.reject(createdPromises);
     });
     const actor = interpret(promiseLogic);
+    actor.subscribe({ error: function preventUnhandledErrorListener() {} });
     actor.start();
 
     await new Promise((res) => setTimeout(res, 5));