feat(cluster): bound runner entity residency and storage reads

.changeset/wild-donuts-brake.md

@@ -0,0 +1,31 @@
+---
+"effect": patch
+---
+
+Bound cluster runner entity residency and storage reads.
+
+`ShardingConfig` gains two knobs:
+
+- `maxResidentEntities` (default `10_000`): the maximum number of entities
+  that can be resident on a runner at the same time. At the cap, the storage
+  read loop stops admitting messages for new entity addresses (they stay in
+  storage until a slot frees up) and volatile sends to new addresses fail with
+  `MailboxFull`. Persisted sends still succeed. `"unbounded"` restores the
+  previous behaviour and can only be set programmatically.
+- `unprocessedMessageBatchSize` (default `1024`): the maximum number of
+  unprocessed messages read from storage in a single poll.
+
+`MessageStorage.unprocessedMessages` accepts an optional
+`{ limit, addresses }` argument, and only claims the messages it actually
+returns. The memory implementation now applies the same ten-minute claim
+window as SQL, so bounded reads advance past in-flight requests; resetting an
+address or shard makes its claimed messages immediately eligible again.
+
+The encoded driver contract replaces `Encoded.resetAddress` with the batched
+`Encoded.resetAddresses` operation. `SqlMessageStorage.makeEncoded` constructs
+the SQL encoded driver directly for custom storage composition.
+
+`ClusterWorkflowEngine` entities (workflows and the durable clock) now use a
+fixed ten-second idle time, so completed and suspended executions release their
+entity slots quickly. Their state is durable, so an evicted execution is
+rebuilt from storage when its next message arrives.

packages/effect/src/unstable/cluster/ClusterError.ts

@@ -171,7 +171,9 @@ export class RunnerUnavailable extends Schema.Error<RunnerUnavailable>(`${TypeId
  *
  * **Details**
  *
- * Carries the `address` whose bounded mailbox is at capacity.
+ * Carries the `address` whose bounded mailbox is at capacity. It is also used
+ * when the runner has reached `ShardingConfig.maxResidentEntities` and cannot
+ * spawn a new entity for the `address`.
  *
  * **Gotchas**
  *

packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts

@@ -444,7 +444,7 @@ export const make = Effect.gen(function*() {
             }
           }),
           // Reserve a slot for deferred completions to wake the active run.
-          { concurrency: 2 }
+          { concurrency: 2, maxIdleTime: entityMaxIdleTime }
         ) as Effect.Effect<void, never, Scope.Scope>
       ),
 
@@ -702,6 +702,12 @@ const ResumeRpc = Rpc.make("resume", {
 
 const payloadParentKey = "~effect/cluster/ClusterWorkflowEngine/payloadParentKey"
 
+// Workflow state is durable, so an idle entity (completed or suspended) can
+// be released quickly and is rebuilt from storage when the next message
+// arrives. This keeps completed and suspended executions from holding
+// `maxResidentEntities` slots for the full `entityMaxIdleTime`.
+const entityMaxIdleTime = Duration.seconds(10)
+
 const makeWorkflowEntity = (workflow: Workflow.Any) =>
   Entity.make(`Workflow/${workflow._tag}`, [
     Rpc.make("run", {
@@ -756,22 +762,25 @@ const ClockEntity = Entity.make("Workflow/-/DurableClock", [
   ClockRpc
 ])
 
-const ClockEntityLayer = ClockEntity.toLayer(Effect.gen(function*() {
-  const engine = yield* WorkflowEngine.WorkflowEngine
-  const address = yield* Entity.CurrentAddress
-  const executionId = address.entityId
-  return {
-    run(request) {
-      const deferred = DurableClock.make({ name: request.payload.name, duration: Duration.zero }).deferred
-      return ensureSuccess(engine.deferredDone(deferred, {
-        workflowName: request.payload.workflowName,
-        executionId,
-        deferredName: deferred.name,
-        exit: Exit.void
-      }))
+const ClockEntityLayer = ClockEntity.toLayer(
+  Effect.gen(function*() {
+    const engine = yield* WorkflowEngine.WorkflowEngine
+    const address = yield* Entity.CurrentAddress
+    const executionId = address.entityId
+    return {
+      run(request) {
+        const deferred = DurableClock.make({ name: request.payload.name, duration: Duration.zero }).deferred
+        return ensureSuccess(engine.deferredDone(deferred, {
+          workflowName: request.payload.workflowName,
+          executionId,
+          deferredName: deferred.name,
+          exit: Exit.void
+        }))
+      }
     }
-  }
-}))
+  }),
+  { maxIdleTime: entityMaxIdleTime }
+)
 
 const InterruptSignal = DurableDeferred.make("Workflow/InterruptSignal")
 

packages/effect/src/unstable/cluster/MessageStorage.ts

@@ -131,9 +131,18 @@ export class MessageStorage extends Context.Service<MessageStorage, {
    * - Requests that have no `WithExit` replies or no unacknowledged chunk replies
    * - The latest `AckChunk` envelope
    * - All `Interrupt` envelopes for unprocessed requests
+   *
+   * The `limit` option bounds the number of messages returned in a single
+   * read, and the `addresses` option restricts the read to the provided
+   * entity addresses. Only the returned messages are claimed; other messages
+   * stay eligible for later reads.
    */
   readonly unprocessedMessages: (
-    shardIds: Iterable<ShardId.ShardId>
+    shardIds: Iterable<ShardId.ShardId>,
+    options?: {
+      readonly limit?: number | undefined
+      readonly addresses?: ReadonlyArray<EntityAddress> | undefined
+    } | undefined
   ) => Effect.Effect<Array<Message.Incoming<any>>, PersistenceError>
 
   /**
@@ -157,6 +166,13 @@ export class MessageStorage extends Context.Service<MessageStorage, {
     address: EntityAddress
   ) => Effect.Effect<void, PersistenceError>
 
+  /**
+   * Reset the mailbox state for the provided addresses.
+   */
+  readonly resetAddresses: (
+    addresses: ReadonlyArray<EntityAddress>
+  ) => Effect.Effect<void, PersistenceError>
+
   /**
    * Clear all messages and replies for the provided address.
    */
@@ -349,10 +365,19 @@ export type Encoded = {
    * - Requests that have no `WithExit` replies or no unacknowledged chunk replies
    * - The latest `AckChunk` envelope
    * - All `Interrupt` envelopes for unprocessed requests
+   *
+   * The `limit` option bounds the number of rows returned by a single read,
+   * and the `addresses` option restricts the read to the provided entity
+   * addresses. Only the returned rows are claimed; other rows stay eligible
+   * for later reads.
    */
   readonly unprocessedMessages: (
     shardIds: Arr.NonEmptyArray<string>,
-    now: number
+    now: number,
+    options?: {
+      readonly limit?: number | undefined
+      readonly addresses?: ReadonlyArray<EntityAddress> | undefined
+    } | undefined
   ) => Effect.Effect<
     Array<{
       readonly envelope: Envelope.Encoded
@@ -376,10 +401,10 @@ export type Encoded = {
   >
 
   /**
-   * Reset the mailbox state for the provided address.
+   * Reset the mailbox state for the provided addresses.
    */
-  readonly resetAddress: (
-    address: EntityAddress
+  readonly resetAddresses: (
+    addresses: ReadonlyArray<EntityAddress>
   ) => Effect.Effect<void, PersistenceError>
 
   /**
@@ -629,12 +654,13 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect<
       const primaryKey = Envelope.primaryKeyByAddress(options)
       return encoded.requestIdForPrimaryKey(primaryKey)
     },
-    unprocessedMessages(shardIds) {
+    unprocessedMessages(shardIds, options) {
       const storage = this as MessageStorage["Service"]
       const shards = Array.from(shardIds, (id) => id.toString())
       if (!Arr.isArrayNonEmpty(shards)) return Effect.succeed([])
+      if (options?.addresses !== undefined && options.addresses.length === 0) return Effect.succeed([])
       return Effect.flatMap(
-        Effect.suspend(() => encoded.unprocessedMessages(shards, clock.currentTimeMillisUnsafe())),
+        Effect.suspend(() => encoded.unprocessedMessages(shards, clock.currentTimeMillisUnsafe(), options)),
         (messages) => decodeMessages(storage, messages)
       )
     },
@@ -647,7 +673,8 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect<
         (messages) => decodeMessages(storage, messages)
       )
     },
-    resetAddress: encoded.resetAddress,
+    resetAddress: (address) => encoded.resetAddresses([address]),
+    resetAddresses: (addresses) => addresses.length === 0 ? Effect.void : encoded.resetAddresses(addresses),
     clearAddress: encoded.clearAddress,
     resetShards: (shardIds) => {
       const shards = Array.from(shardIds, (id) => id.toString())
@@ -776,6 +803,7 @@ export const noop: MessageStorage["Service"] = Effect.runSync(make({
   unprocessedMessages: () => Effect.succeed([]),
   unprocessedMessagesById: () => Effect.succeed([]),
   resetAddress: () => Effect.void,
+  resetAddresses: () => Effect.void,
   clearAddress: () => Effect.void,
   resetShards: () => Effect.void,
   withTransaction: identity
@@ -809,6 +837,9 @@ export const MemoryTransaction = Context.Reference<boolean>("effect/cluster/Mess
   defaultValue: constFalse
 })
 
+// the same claim window the SQL driver uses (`last_read < ten minutes ago`)
+const claimExpirationMillis = 10 * 60 * 1000
+
 /**
  * Service that provides an in-memory message storage driver with inspectable backing state.
  *
@@ -828,9 +859,25 @@ export class MemoryDriver extends Context.Service<MemoryDriver>()("effect/cluste
     const requestsByPrimaryKey = new Map<string, MemoryEntry>()
     const unprocessed = new Set<Envelope.Encoded>()
     const replyIds = new Set<string>()
+    const lastRead = new Map<Envelope.Encoded, number>()
 
     const journal: Array<Envelope.Encoded> = []
 
+    const addressKey = (address: Envelope.Encoded["address"] | EntityAddress) =>
+      `${address.shardId.group}/${address.shardId.id}/${address.entityType}/${address.entityId}`
+
+    const resetAddresses = (addresses: ReadonlyArray<EntityAddress>) =>
+      addresses.length === 0
+        ? Effect.void
+        : Effect.sync(() => {
+          const keys = new Set(addresses.map(addressKey))
+          for (const envelope of journal) {
+            if (keys.has(addressKey(envelope.address))) {
+              lastRead.delete(envelope)
+            }
+          }
+        })
+
     const cursors = new WeakMap<{}, number>()
 
     const unprocessedWith = (predicate: Predicate<Envelope.Encoded>) => {
@@ -926,6 +973,7 @@ export class MemoryDriver extends Context.Service<MemoryDriver>()("effect/cluste
           if (!entry || replyIds.has(reply.id)) return
           if (reply._tag === "WithExit") {
             unprocessed.delete(entry.envelope)
+            lastRead.delete(entry.envelope)
           }
           entry.replies.push(reply)
           replyIds.add(reply.id)
@@ -938,6 +986,7 @@ export class MemoryDriver extends Context.Service<MemoryDriver>()("effect/cluste
           entry.replies = []
           entry.lastReceivedChunk = undefined
           unprocessed.add(entry.envelope)
+          lastRead.delete(entry.envelope)
         }),
       requestIdForPrimaryKey: (primaryKey) =>
         Effect.sync(() => {
@@ -947,29 +996,39 @@ export class MemoryDriver extends Context.Service<MemoryDriver>()("effect/cluste
       repliesFor: (requestIds) => Effect.sync(() => repliesFor(requestIds)),
       repliesForUnfiltered: (requestIds) =>
         Effect.sync(() => requestIds.flatMap((id) => requests.get(String(id))?.replies ?? [])),
-      unprocessedMessages: (shardIds) =>
-        Effect.sync(() => {
+      unprocessedMessages: (shardIds, now, options) =>
+        options?.addresses?.length === 0 ? Effect.succeed([]) : Effect.sync(() => {
           if (unprocessed.size === 0) return []
-          const now = clock.currentTimeMillisUnsafe()
+          const limit = options?.limit ?? Infinity
+          const addressFilter = options?.addresses && new Set(options.addresses.map(addressKey))
           const messages = Arr.empty<{
             envelope: Envelope.Encoded
             lastSentReply: Option.Option<Reply.Encoded>
           }>()
           for (let index = 0; index < journal.length; index++) {
+            if (messages.length >= limit) break
             const envelope = journal[index]
             const shardId = ShardId.make(envelope.address.shardId.group, envelope.address.shardId.id)
             if (!unprocessed.has(envelope as any) || !shardIds.includes(shardId.toString())) {
               continue
             }
+            if (addressFilter && !addressFilter.has(addressKey(envelope.address))) {
+              continue
+            }
             if (envelope._tag === "Request") {
               const entry = requests.get(envelope.requestId)!
               if (entry.deliverAt && entry.deliverAt > now) {
                 continue
               }
+              const claimedAt = lastRead.get(envelope)
+              if (claimedAt !== undefined && claimedAt > now - claimExpirationMillis) {
+                continue
+              }
               messages.push({
                 envelope,
                 lastSentReply: Option.fromNullishOr(entry.replies[entry.replies.length - 1])
               })
+              lastRead.set(envelope, now)
             } else {
               messages.push({
                 envelope,
@@ -988,7 +1047,7 @@ export class MemoryDriver extends Context.Service<MemoryDriver>()("effect/cluste
           }
           return unprocessedWith((envelope) => envelopeIds.has(envelope.requestId))
         }),
-      resetAddress: () => Effect.void,
+      resetAddresses,
       clearAddress: (address) =>
         Effect.sync(() => {
           for (const [primaryKey, entry] of requestsByPrimaryKey) {
@@ -1007,11 +1066,21 @@ export class MemoryDriver extends Context.Service<MemoryDriver>()("effect/cluste
               continue
             }
             unprocessed.delete(envelope)
+            lastRead.delete(envelope)
             requests.delete(envelope.requestId)
             journal.splice(i, 1)
           }
         }),
-      resetShards: () => Effect.void,
+      resetShards: (shardIds) =>
+        Effect.sync(() => {
+          const shards = new Set(shardIds)
+          for (const envelope of journal) {
+            const shardId = ShardId.make(envelope.address.shardId.group, envelope.address.shardId.id)
+            if (shards.has(shardId.toString())) {
+              lastRead.delete(envelope)
+            }
+          }
+        }),
       withTransaction: Effect.provideService(MemoryTransaction, true)
     }