ensure persisted messages are sent when no assignments (#4618)

Author: tim-smart

packages/cluster/src/Runners.ts

@@ -11,6 +11,7 @@ import * as Effect from "effect/Effect"
 import * as Exit from "effect/Exit"
 import * as FiberRef from "effect/FiberRef"
 import * as Layer from "effect/Layer"
+import * as Option from "effect/Option"
 import * as RcMap from "effect/RcMap"
 import * as Schema from "effect/Schema"
 import type { Scope } from "effect/Scope"
@@ -74,7 +75,7 @@ export class Runners extends Context.Tag("@effect/cluster/Runners")<Runners, {
    */
   readonly notify: <R extends Rpc.Any>(
     options: {
-      readonly address: RunnerAddress
+      readonly address: Option.Option<RunnerAddress>
       readonly message: Message.Outgoing<R>
       readonly discard: boolean
     }
@@ -497,8 +498,11 @@ export const makeRpc: Effect.Effect<
       })
     },
     notify({ address, message }) {
+      if (Option.isNone(address)) {
+        return Effect.void
+      }
       const envelope = message.envelope
-      return RcMap.get(clients, address).pipe(
+      return RcMap.get(clients, address.value).pipe(
         Effect.flatMap((client) => client.Notify({ envelope })),
         Effect.scoped,
         Effect.ignore

packages/cluster/src/Sharding.ts

@@ -668,19 +668,18 @@ export const make = Effect.gen(function*() {
       Effect.suspend(() => {
         const address = message.envelope.address
         const maybeRunner = MutableHashMap.get(shardAssignments, address.shardId)
-        if (Option.isNone(maybeRunner)) {
-          return Effect.fail(new EntityNotManagedByRunner({ address }))
-        }
-        const runner = maybeRunner.value
-        const rpc = message.rpc as any as Rpc.AnyWithProps
-        if (storageEnabled && Context.get(rpc.annotations, Persisted)) {
-          return isLocalRunner(runner)
+        const isPersisted = storageEnabled && Context.get(message.rpc.annotations, Persisted)
+        const runnerIsLocal = Option.isSome(maybeRunner) && isLocalRunner(maybeRunner.value)
+        if (isPersisted) {
+          return runnerIsLocal
             ? notifyLocal(message, discard)
-            : runners.notify({ address: runner, message, discard })
+            : runners.notify({ address: maybeRunner, message, discard })
+        } else if (Option.isNone(maybeRunner)) {
+          return Effect.fail(new EntityNotManagedByRunner({ address }))
         }
-        return isLocalRunner(runner)
+        return runnerIsLocal
           ? sendLocal(message)
-          : runners.send({ address: runner, message })
+          : runners.send({ address: maybeRunner.value, message })
       }),
       isTransientError,
       (error) => {