add entityRegistrationTimeout to ShardingConfig (#5765)

Author: tim-smart

.changeset/chubby-wolves-scream.md

@@ -0,0 +1,5 @@
+---
+"@effect/cluster": patch
+---
+
+add entityRegistrationTimeout to ShardingConfig

packages/cluster/src/Sharding.ts

@@ -199,6 +199,7 @@ interface EntityManagerState {
 
 const make = Effect.gen(function*() {
   const config = yield* ShardingConfig
+  const clock = yield* Effect.clock
 
   const runnersService = yield* Runners
   const runnerHealth = yield* RunnerHealth.RunnerHealth
@@ -444,6 +445,8 @@ const make = Effect.gen(function*() {
 
   if (storageEnabled && Option.isSome(config.runnerAddress)) {
     const selfAddress = config.runnerAddress.value
+    const entityRegistrationTimeoutMillis = Duration.toMillis(config.entityRegistrationTimeout)
+    const storageStartMillis = clock.unsafeCurrentTimeMillis()
 
     yield* Effect.gen(function*() {
       yield* Effect.logDebug("Starting")
@@ -469,9 +472,14 @@ const make = Effect.gen(function*() {
           }
           const state = entityManagers.get(address.entityType)
           if (!state) {
-            // reset address in the case that the entity is slow to register
-            MutableHashSet.add(resetAddresses, address)
-            return Effect.void
+            const sinceStart = clock.unsafeCurrentTimeMillis() - storageStartMillis
+            if (sinceStart < entityRegistrationTimeoutMillis) {
+              // reset address in the case that the entity is slow to register
+              MutableHashSet.add(resetAddresses, address)
+              return Effect.void
+            }
+            // if the entity did not register in time, we save a defect reply
+            return Effect.die(new Error(`Entity type '${address.entityType}' not registered`))
           } else if (state.status === "closed") {
             return Effect.void
           }

packages/cluster/src/ShardingConfig.ts

@@ -78,6 +78,13 @@ export class ShardingConfig extends Context.Tag("@effect/cluster/ShardingConfig"
    * after which an entity will be interrupted.
    */
   readonly entityMaxIdleTime: DurationInput
+  /**
+   * If an entity does not register itself within this time after a message is
+   * sent to it, the message will be marked as failed.
+   *
+   * Defaults to 1 minute.
+   */
+  readonly entityRegistrationTimeout: DurationInput
   /**
    * The maximum duration of time to wait for an entity to terminate.
    *
@@ -129,6 +136,7 @@ export const defaults: ShardingConfig["Type"] = {
   shardLockExpiration: Duration.seconds(35),
   entityMailboxCapacity: 4096,
   entityMaxIdleTime: Duration.minutes(1),
+  entityRegistrationTimeout: Duration.minutes(1),
   entityTerminationTimeout: Duration.seconds(15),
   entityMessagePollInterval: Duration.seconds(10),
   entityReplyPollInterval: Duration.millis(200),
@@ -208,6 +216,12 @@ export const config: Config.Config<ShardingConfig["Type"]> = Config.all({
       "The maximum duration of inactivity (i.e. without receiving a message) after which an entity will be interrupted."
     )
   ),
+  entityRegistrationTimeout: Config.duration("entityRegistrationTimeout").pipe(
+    Config.withDefault(defaults.entityRegistrationTimeout),
+    Config.withDescription(
+      "If an entity does not register itself within this time after a message is sent to it, the message will be marked as failed."
+    )
+  ),
   entityTerminationTimeout: Config.duration("entityTerminationTimeout").pipe(
     Config.withDefault(defaults.entityTerminationTimeout),
     Config.withDescription("The maximum duration of time to wait for an entity to terminate.")