fix(runtime): commit client sends eagerly

.changeset/eager-client-dispatch.md

@@ -0,0 +1,5 @@
+---
+"effect-machine": patch
+---
+
+Commit synchronous client transitions before `ActorClient.send` returns and notify Actor Atom subscribers without a timer delay.

src/atom.ts

@@ -36,6 +36,14 @@ export type ActorAcquireAtom<State extends { readonly _tag: string }, Event, Out
   AsyncResult.AsyncResult<ActorRef<State, Event, Output>, Cause.NoSuchElementError>
 >;
 
+const stateAtom = <State extends { readonly _tag: string }, Event, Output>(
+  actor: ActorRef<State, Event, Output>,
+): Atom.Atom<State> =>
+  Atom.readable((get) => {
+    get.addFinalizer(actor.subscribe((state) => get.setSelf(state)));
+    return actor.client.getSnapshot();
+  });
+
 /** Observe the current ActorRef for one ActorSystem key. */
 export const fromSystem: {
   <State extends { readonly _tag: string }, Event, Output>(
@@ -102,7 +110,7 @@ export const acquire: {
 export const make = <State extends { readonly _tag: string }, Event, Output>(
   actor: ActorRef<State, Event, Output>,
 ): ActorAtom<State, Event> => {
-  const state = Atom.subscriptionRef(actor.state);
+  const state = stateAtom(actor);
   return Atom.writable(
     (get) => get(state),
     (_ctx, event) => actor.client.send(event),
@@ -170,7 +178,7 @@ export const can: {
     actor: ActorRef<State, Event, Output>,
     event: Event,
   ): CanAtom => {
-    const state = Atom.subscriptionRef(actor.state);
+    const state = stateAtom(actor);
     return Atom.make((get) => {
       get(state);
       return actor.can(event);

src/internal/runtime.ts

@@ -550,6 +550,7 @@ export const createRuntime = Effect.fn("effect-machine.runtime.create")(function
       eventQueue,
       pendingRequests,
       exitDeferred,
+      () => loopFiberRef.current?.currentDispatcher?.flush(),
     ),
     stop: stop.pipe(Effect.provide(services)),
     start: start.pipe(Effect.provide(services)),
@@ -568,6 +569,7 @@ const makeHandle = <S extends { readonly _tag: string }, E extends { readonly _t
   eventQueue: Queue.Queue<RuntimeQueuedEvent<S, E>>,
   pendingRequests: Set<(error: ActorStoppedError) => Effect.Effect<void>>,
   exitDeferred: Deferred.Deferred<RuntimeExit<S>>,
+  flushLoop: () => void,
 ): RuntimeHandle<S, E> => {
   const track = <A, RequestError>(
     deferred: Deferred.Deferred<A, RequestError>,
@@ -625,7 +627,10 @@ const makeHandle = <S extends { readonly _tag: string }, E extends { readonly _t
     }).pipe(Effect.asVoid),
     sendSync: (event: E) => {
       const stopped = Effect.runSync(Ref.get(stoppedRef));
-      if (!stopped) Effect.runSync(Queue.offer(eventQueue, { _tag: "send", event }));
+      if (stopped) return;
+      Effect.runSync(Queue.offer(eventQueue, { _tag: "send", event }));
+      eventQueue.dispatcher.flush();
+      flushLoop();
     },
     getState: SubscriptionRef.get(stateRef),
     stateRef,

test/actor.test.ts

@@ -1055,7 +1055,7 @@ describe("ActorRef", () => {
   // ============================================================================
 
   describe("sendSync", () => {
-    it.scopedLive("sendSync sends event synchronously", () =>
+    it.scopedLive("sendSync commits a synchronous transition before returning", () =>
       Effect.gen(function* () {
         const machine = Machine.make({
           state: TestState,
@@ -1071,9 +1071,8 @@ describe("ActorRef", () => {
         const actor = yield* Machine.spawn(machine);
         yield* actor.start;
         actor.client.send(TestEvent.Start({ value: 7 }));
-        yield* yieldFibers;
 
-        const state = yield* actor.snapshot;
+        const state = actor.client.getSnapshot();
         expect(state._tag).toBe("Active");
         if (state._tag === "Active") {
           expect(state.value).toBe(7);
@@ -1102,6 +1101,54 @@ describe("ActorRef", () => {
         expect(state._tag).toBe("Idle");
       }),
     );
+
+    it.scopedLive("sendSync commits a transition with a synchronous Effect guard", () =>
+      Effect.gen(function* () {
+        const machine = Machine.make({
+          state: TestState,
+          event: TestEvent,
+          initial: TestState.Idle,
+        }).when(
+          TestState.Idle,
+          TestEvent.Start,
+          () => Effect.succeed(true),
+          ({ event }) => TestState.Active({ value: event.value }),
+        );
+        const actor = yield* Machine.spawn(machine);
+        yield* actor.start;
+
+        actor.client.send(TestEvent.Start({ value: 9 }));
+
+        expect(actor.client.getSnapshot()).toEqual(TestState.Active({ value: 9 }));
+      }),
+    );
+
+    it.scopedLive("sendSync preserves nested event order", () =>
+      Effect.gen(function* () {
+        const machine = Machine.make({
+          state: TestState,
+          event: TestEvent,
+          initial: TestState.Idle,
+        })
+          .on(TestState.Idle, TestEvent.Start, ({ event }) =>
+            TestState.Active({ value: event.value }),
+          )
+          .on(TestState.Active, TestEvent.Stop, () => TestState.Done);
+        const actor = yield* Machine.spawn(machine);
+        yield* actor.start;
+        const states: Array<string> = [];
+        const unsubscribe = actor.subscribe((state) => {
+          states.push(state._tag);
+          if (state._tag === "Active") actor.client.send(TestEvent.Stop);
+        });
+
+        actor.client.send(TestEvent.Start({ value: 1 }));
+
+        expect(actor.client.getSnapshot()._tag).toBe("Done");
+        expect(states).toEqual(["Active", "Done"]);
+        unsubscribe();
+      }),
+    );
   });
 
   // ============================================================================

test/atom.test.ts

@@ -43,8 +43,6 @@ it.scopedLive("make exposes actor state and sends events through Atom writes", (
     });
 
     registry.set(stateAtom, CounterEvent.Increment);
-    yield* actor.waitFor((state) => state.count === 1);
-    yield* Effect.yieldNow;
 
     expect(registry.get(stateAtom).count).toBe(1);
     expect(values).toEqual([0, 1]);