diff --git a/.changeset/wild-donuts-brake.md b/.changeset/wild-donuts-brake.md
new file mode 100644
index 00000000000..4bfeecea7f9
--- /dev/null
+++ b/.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.
diff --git a/packages/effect/src/unstable/cluster/ClusterError.ts b/packages/effect/src/unstable/cluster/ClusterError.ts
index e8faae7fea3..6e74365ee7b 100644
--- a/packages/effect/src/unstable/cluster/ClusterError.ts
+++ b/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**
  *
diff --git a/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts b/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts
index 797a996506a..db1c5f95dbe 100644
--- a/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts
+++ b/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")
 
diff --git a/packages/effect/src/unstable/cluster/MessageStorage.ts b/packages/effect/src/unstable/cluster/MessageStorage.ts
index 8d33454a50d..79490b90316 100644
--- a/packages/effect/src/unstable/cluster/MessageStorage.ts
+++ b/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)
     }
 
diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts
index ab40a7cfecb..33486152398 100644
--- a/packages/effect/src/unstable/cluster/Sharding.ts
+++ b/packages/effect/src/unstable/cluster/Sharding.ts
@@ -588,6 +588,33 @@ const make = Effect.gen(function*() {
   const storageReadLock = Semaphore.makeUnsafe(1)
   const withStorageReadLock = storageReadLock.withPermits(1)
 
+  // --- Entity residency ---
+  //
+  // A runner-wide counter that bounds how many entities can be resident at
+  // the same time. Entity managers reserve a slot before spawning an entity,
+  // so the sequential storage read loop and concurrent volatile sends share
+  // the same cap.
+
+  const maxResidentEntities = config.maxResidentEntities
+  let residentEntityCount = 0
+  const residencyAtCapacityUnsafe = () =>
+    maxResidentEntities !== "unbounded" && residentEntityCount >= maxResidentEntities
+  const residency: EntityManager.Residency = {
+    admitUnsafe() {
+      if (residencyAtCapacityUnsafe()) return false
+      residentEntityCount++
+      return true
+    },
+    releaseUnsafe() {
+      residentEntityCount--
+      // a slot has freed up, so the storage read loop can admit messages for
+      // entities it previously had to skip
+      if (maxResidentEntities !== "unbounded") {
+        storageReadLatch.openUnsafe()
+      }
+    }
+  }
+
   if (storageEnabled && initialRunnerAddress) {
     const selfAddress = initialRunnerAddress
     const entityRegistrationTimeoutMillis = Duration.toMillis(
@@ -600,8 +627,14 @@ const make = Effect.gen(function*() {
 
       let index = 0
       let messages: Array<Message.Incoming<any>> = []
+      let deliveredThisRead = false
       const removableNotifications = new Set<PendingNotification>()
       const resetAddresses = MutableHashSet.empty<EntityAddress>()
+      const cappedAddresses = MutableHashSet.empty<EntityAddress>()
+
+      const markDelivered = Effect.sync(() => {
+        deliveredThisRead = true
+      })
 
       const processMessages = Effect.whileLoop({
         while: () => index < messages.length,
@@ -609,6 +642,16 @@ const make = Effect.gen(function*() {
         body: () => send
       })
 
+      const readAndProcess = Effect.fnUntraced(function*(options?: {
+        readonly limit?: number | undefined
+        readonly addresses?: ReadonlyArray<EntityAddress> | undefined
+      }) {
+        messages = yield* storage.unprocessedMessages(acquiredShards, options)
+        index = 0
+        yield* processMessages
+        return messages.length
+      })
+
       const send = Effect.catchCause(
         Effect.suspend(() => {
           const message = messages[index]
@@ -642,7 +685,7 @@ const make = Effect.gen(function*() {
           if (message._tag === "IncomingEnvelope" && isProcessing) {
             // If the message might affect a currently processing request, we
             // send it to the entity manager to be processed.
-            return state.manager.send(message)
+            return Effect.tap(state.manager.send(message), markDelivered)
           } else if (isProcessing || state.status === "closing") {
             // If the request is already processing, we skip it.
             // Or if the entity is closing, we skip all incoming messages.
@@ -654,6 +697,14 @@ const make = Effect.gen(function*() {
             entry.resume(Effect.void)
           }
 
+          // The runner is at entity capacity, so messages for entities that
+          // are not already resident are skipped. Their claims are released
+          // after the read, so they stay eligible for once a slot frees up.
+          if (residencyAtCapacityUnsafe() && !state.manager.isResidentUnsafe(address)) {
+            MutableHashSet.add(cappedAddresses, address)
+            return Effect.void
+          }
+
           // If the entity was resuming in another fiber, we add the message
           // id to the unprocessed set.
           const resumptionState = MutableHashMap.get(entityResumptionState, address)
@@ -664,7 +715,7 @@ const make = Effect.gen(function*() {
             }
             return Effect.void
           }
-          return state.manager.send(message)
+          return Effect.tap(state.manager.send(message), markDelivered)
         }),
         (cause) => {
           const message = messages[index]
@@ -681,13 +732,25 @@ const make = Effect.gen(function*() {
             }))
           }
           if (error.success._tag === "MailboxFull") {
+            const address = message.envelope.address
+            const state = entityManagers.get(address.entityType)
+            // A resident entity has a full per-entity mailbox; its messages
+            // are resumed from storage once there is capacity again.
             // MailboxFull can only happen for requests, so this cast is safe
-            return resumeEntityFromStorage(message as Message.IncomingRequest<any>)
+            if (message._tag === "IncomingRequest" && state?.manager.isResidentUnsafe(address)) {
+              return resumeEntityFromStorage(message as Message.IncomingRequest<any>)
+            }
+            // Otherwise the runner ran out of entity slots while the message
+            // was in flight; leave it in storage.
+            MutableHashSet.add(cappedAddresses, address)
+            return Effect.void
           }
           return Effect.void
         }
       )
 
+      const batchSize = Math.max(1, config.unprocessedMessageBatchSize)
+
       while (true) {
         // wait for the next poll interval, or if we get notified of a change
         yield* storageReadLatch.await
@@ -706,15 +769,58 @@ const make = Effect.gen(function*() {
           pendingNotifications.forEach((entry) => removableNotifications.add(entry))
         }
 
-        messages = yield* storage.unprocessedMessages(acquiredShards)
-        index = 0
-        yield* processMessages
+        deliveredThisRead = false
+        let readCount = 0
+        let fullBatch = false
+        let exhaustive = false
+
+        // First deliver messages for entities that are already resident, so
+        // they keep making progress even when the runner is at entity
+        // capacity.
+        if (residencyAtCapacityUnsafe()) {
+          const residentAddresses: Array<EntityAddress> = []
+          for (const state of entityManagers.values()) {
+            if (state.status === "closed") continue
+            for (const address of state.manager.residentAddressesUnsafe()) {
+              residentAddresses.push(address)
+            }
+          }
+          if (residentAddresses.length > 0) {
+            readCount = yield* readAndProcess({ limit: batchSize, addresses: residentAddresses })
+            fullBatch = readCount >= batchSize
+          }
+        }
+
+        // Then walk the remaining messages, spawning new entities while the
+        // runner has entity slots left.
+        if (!residencyAtCapacityUnsafe() && readCount < batchSize) {
+          const limit = batchSize - readCount
+          const read = yield* readAndProcess({ limit })
+          if (read >= limit) {
+            fullBatch = true
+          } else if (MutableHashSet.size(cappedAddresses) === 0) {
+            exhaustive = true
+          }
+        }
 
         if (removableNotifications.size > 0) {
-          removableNotifications.forEach(({ message, resume }) => {
-            pendingNotifications.delete(message.envelope.requestId)
-            resume(Effect.fail(new EntityNotAssignedToRunner({ address: message.envelope.address })))
-          })
+          // On an exhaustive read, a message that was not seen is no longer
+          // processable by this runner. At the entity cap, the messages are
+          // safely persisted and delivered once a slot frees up, so persisted
+          // senders succeed instead of failing. Otherwise the read was
+          // truncated by the batch size and the notifications stay registered
+          // for the immediately following read.
+          const capped = MutableHashSet.size(cappedAddresses) > 0 || residencyAtCapacityUnsafe()
+          if (exhaustive || capped) {
+            removableNotifications.forEach(({ message, resume }) => {
+              pendingNotifications.delete(message.envelope.requestId)
+              resume(
+                exhaustive
+                  ? Effect.fail(new EntityNotAssignedToRunner({ address: message.envelope.address }))
+                  : Effect.void
+              )
+            })
+          }
           removableNotifications.clear()
         }
         if (MutableHashSet.size(resetAddresses) > 0) {
@@ -726,9 +832,29 @@ const make = Effect.gen(function*() {
           }
           MutableHashSet.clear(resetAddresses)
         }
+        // Capture claims skipped because of the entity cap. They are reset in
+        // one storage operation after releasing the read lock.
+        let cappedAddressesToReset: Array<EntityAddress> | undefined
+        if (MutableHashSet.size(cappedAddresses) > 0) {
+          cappedAddressesToReset = Arr.fromIterable(cappedAddresses)
+          MutableHashSet.clear(cappedAddresses)
+        }
 
         // let the resuming entities check if they are done
         yield* storageReadLock.release(1)
+
+        if (cappedAddressesToReset !== undefined) {
+          yield* Effect.ignore(storage.resetAddresses(cappedAddressesToReset))
+        }
+
+        // A full batch means more messages could be waiting; start the next
+        // read immediately, as long as this read made progress (a full batch
+        // of skipped messages must not spin the loop). When the runner is at
+        // capacity, the next read waits for an entity to be removed or the
+        // next poll interval instead.
+        if (fullBatch && deliveredThisRead) {
+          storageReadLatch.openUnsafe()
+        }
       }
     }).pipe(
       Effect.scoped,
@@ -1475,6 +1601,7 @@ const make = Effect.gen(function*() {
         ...options,
         storage,
         runnerAddress,
+        residency,
         sharding
       }).pipe(
         Effect.provideContext(services.pipe(
diff --git a/packages/effect/src/unstable/cluster/ShardingConfig.ts b/packages/effect/src/unstable/cluster/ShardingConfig.ts
index e9a465d6375..1c35f200b33 100644
--- a/packages/effect/src/unstable/cluster/ShardingConfig.ts
+++ b/packages/effect/src/unstable/cluster/ShardingConfig.ts
@@ -95,6 +95,28 @@ export class ShardingConfig extends Context.Service<ShardingConfig, {
    * The default capacity of the mailbox for entities.
    */
   readonly entityMailboxCapacity: number | "unbounded"
+  /**
+   * The maximum number of entities that can be resident on this runner at the
+   * same time.
+   *
+   * When the limit is reached, no new entities are spawned: 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`.
+   *
+   * `"unbounded"` can only be set programmatically; the environment
+   * configuration only accepts integers.
+   *
+   * Defaults to `10_000`.
+   */
+  readonly maxResidentEntities: number | "unbounded"
+  /**
+   * The maximum number of unprocessed messages read from storage in a single
+   * poll.
+   *
+   * Defaults to `1024`.
+   */
+  readonly unprocessedMessageBatchSize: number
   /**
    * The maximum duration of inactivity (i.e. without receiving a message)
    * after which an entity will be interrupted.
@@ -163,6 +185,8 @@ export const defaults: ShardingConfig["Service"] = {
   shardLockExpiration: Duration.seconds(35),
   shardLockDisableAdvisory: false,
   entityMailboxCapacity: 4096,
+  maxResidentEntities: 10_000,
+  unprocessedMessageBatchSize: 1024,
   entityMaxIdleTime: Duration.minutes(1),
   entityRegistrationTimeout: Duration.minutes(1),
   entityTerminationTimeout: Duration.seconds(15),
@@ -277,6 +301,20 @@ export const config: Config.Config<ShardingConfig["Service"]> = Config.all({
     Config.withDefault(defaults.entityMailboxCapacity)
     // Config.withDescription("The default capacity of the mailbox for entities.")
   ),
+  maxResidentEntities: Config.schema(
+    Schema.Int.check(Schema.isGreaterThan(0)),
+    "maxResidentEntities"
+  ).pipe(
+    Config.withDefault(defaults.maxResidentEntities)
+    // Config.withDescription("The maximum number of entities that can be resident on this runner at the same time.")
+  ),
+  unprocessedMessageBatchSize: Config.schema(
+    Schema.Int.check(Schema.isGreaterThan(0)),
+    "unprocessedMessageBatchSize"
+  ).pipe(
+    Config.withDefault(defaults.unprocessedMessageBatchSize)
+    // Config.withDescription("The maximum number of unprocessed messages read from storage in a single poll.")
+  ),
   entityMaxIdleTime: Config.duration("entityMaxIdleTime").pipe(
     Config.withDefault(defaults.entityMaxIdleTime)
     // Config.withDescription(
diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
index e8c2ed4c0d2..cdcef650b0a 100644
--- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
+++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts
@@ -29,6 +29,7 @@ import * as SqlClient from "../sql/SqlClient.ts"
 import type { Row } from "../sql/SqlConnection.ts"
 import { isSqlError, type SqlError } from "../sql/SqlError.ts"
 import { PersistenceError } from "./ClusterError.ts"
+import type * as EntityAddress from "./EntityAddress.ts"
 import type * as Envelope from "./Envelope.ts"
 import * as MessageStorage from "./MessageStorage.ts"
 import { SaveResultEncoded } from "./MessageStorage.ts"
@@ -40,13 +41,13 @@ import * as Snowflake from "./Snowflake.ts"
 const withTracerDisabled = Effect.withTracerEnabled(false)
 
 /**
- * Creates a SQL-backed `MessageStorage` implementation, running its migrations
+ * Creates a SQL-backed encoded message storage driver, running its migrations
  * and using the optional table prefix.
  *
  * **When to use**
  *
- * Use when you need the SQL-backed `MessageStorage` service directly, such as
- * when composing a custom layer or providing your own `Snowflake.Generator`.
+ * Use when you need the SQL-backed encoded driver directly, such as when
+ * composing a custom message storage adapter.
  *
  * **Details**
  *
@@ -58,18 +59,17 @@ const withTracerDisabled = Effect.withTracerEnabled(false)
  * Changing `prefix` after deployment points the runtime at a different set of
  * tables, including the migration history table.
  *
- * @see {@link layer} for a ready-made layer using the default prefix and generator
- * @see {@link layerWith} for a ready-made layer with a custom table prefix
+ * @see {@link make} for the decoded `MessageStorage` constructor
  *
  * @category constructors
  * @since 4.0.0
  */
-export const make: (options?: {
+export const makeEncoded: (options?: {
   readonly prefix?: string | undefined
 }) => Effect.Effect<
-  MessageStorage.MessageStorage["Service"],
+  MessageStorage.Encoded,
   never,
-  SqlClient.SqlClient | Snowflake.Generator | Crypto.Crypto
+  SqlClient.SqlClient | Crypto.Crypto
 > = Effect.fnUntraced(function*(options) {
   const sql = (yield* SqlClient.SqlClient).withoutTransforms()
   const crypto = yield* Crypto.Crypto
@@ -372,8 +372,58 @@ export const make: (options?: {
     orElse: () => sql.literal("FOR UPDATE")
   })
 
-  const getUnprocessedMessages = sql.onDialectOrElse({
-    pg: () => (shardIds: ReadonlyArray<string>, now: number) =>
+  const emptyFragment = sql.literal("")
+  // mssql only limits with OFFSET/FETCH, which requires the ORDER BY that
+  // both queries already have
+  const limitFragment = sql.onDialectOrElse({
+    mssql: () => (limit: number) => sql.literal(`OFFSET 0 ROWS FETCH NEXT ${Math.floor(limit)} ROWS ONLY`),
+    orElse: () => (limit: number) => sql.literal(`LIMIT ${Math.floor(limit)}`)
+  })
+  const groupAddresses = (addresses: ReadonlyArray<EntityAddress.EntityAddress>) => {
+    const byShard = new Map<string, Map<string, Array<EntityAddress.EntityAddress>>>()
+    for (const address of addresses) {
+      const shardId = address.shardId.toString()
+      let byEntityType = byShard.get(shardId)
+      if (byEntityType === undefined) {
+        byShard.set(shardId, byEntityType = new Map())
+      }
+      const group = byEntityType.get(address.entityType)
+      if (group === undefined) {
+        byEntityType.set(address.entityType, [address])
+      } else {
+        group.push(address)
+      }
+    }
+    return Array.from(
+      byShard,
+      ([shardId, byEntityType]) =>
+        Array.from(byEntityType, ([entityType, addresses]) => ({ shardId, entityType, addresses }))
+    ).flat()
+  }
+  type UnprocessedOptions = {
+    readonly limit?: number | undefined
+    readonly addresses?: ReadonlyArray<EntityAddress.EntityAddress> | undefined
+  }
+  const unprocessedFilters = (options?: UnprocessedOptions | undefined) => ({
+    addressFilter: options?.addresses !== undefined
+      ? sql`AND (${
+        sql.or(
+          groupAddresses(options.addresses).map((group) =>
+            sql.and([
+              sql`m.shard_id = ${group.shardId}`,
+              sql`m.entity_type = ${group.entityType}`,
+              sql.in("m.entity_id", group.addresses.map((address) => address.entityId))
+            ])
+          )
+        )
+      })`
+      : emptyFragment,
+    limit: options?.limit !== undefined ? limitFragment(options.limit) : emptyFragment
+  })
+  type UnprocessedFilters = ReturnType<typeof unprocessedFilters>
+
+  const getUnprocessedMessagesForDialect = sql.onDialectOrElse({
+    pg: () => (shardIds: ReadonlyArray<string>, now: number, filters: UnprocessedFilters) =>
       sql<MessageJoinRow>`
         WITH messages AS (
           UPDATE ${messagesTableSql} m
@@ -382,6 +432,7 @@ export const make: (options?: {
             SELECT m.*
             FROM ${messagesTableSql} m
             WHERE m.shard_id IN (${sql.literal(shardIds.map(wrapString).join(","))})
+            ${filters.addressFilter}
             AND NOT EXISTS (
               SELECT 1 FROM ${repliesTableSql}
               WHERE request_id = m.request_id
@@ -390,6 +441,8 @@ export const make: (options?: {
             AND m.processed = ${sqlFalse}
             AND (m.last_read IS NULL OR m.last_read < ${tenMinutesAgo})
             AND (m.deliver_at IS NULL OR m.deliver_at <= ${sql.literal(String(now))})
+            ORDER BY m.rowid ASC
+            ${filters.limit}
             FOR UPDATE
           ) AS ids
           LEFT JOIN ${repliesTableSql} r ON r.id = ids.last_reply_id
@@ -398,12 +451,13 @@ export const make: (options?: {
         )
         SELECT * FROM messages ORDER BY rowid ASC
       `,
-    orElse: () => (shardIds: ReadonlyArray<string>, now: number) =>
+    orElse: () => (shardIds: ReadonlyArray<string>, now: number, filters: UnprocessedFilters) =>
       sql<MessageJoinRow>`
         SELECT m.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence
         FROM ${messagesTableSql} m
         LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id
         WHERE m.shard_id IN (${sql.literal(shardIds.map(wrapString).join(","))})
+        ${filters.addressFilter}
         AND NOT EXISTS (
           SELECT 1 FROM ${repliesTableSql}
           WHERE request_id = m.request_id
@@ -413,6 +467,7 @@ export const make: (options?: {
         AND (m.last_read IS NULL OR m.last_read < ${tenMinutesAgo})
         AND (m.deliver_at IS NULL OR m.deliver_at <= ${sql.literal(String(now))})
         ORDER BY m.rowid ASC
+        ${filters.limit}
         ${forUpdate}
       `.unprepared.pipe(
         Effect.tap((rows) => {
@@ -428,8 +483,16 @@ export const make: (options?: {
         sql.withTransaction
       )
   })
+  const getUnprocessedMessages = (
+    shardIds: ReadonlyArray<string>,
+    now: number,
+    options?: UnprocessedOptions | undefined
+  ) =>
+    options?.addresses?.length === 0
+      ? Effect.succeed([])
+      : getUnprocessedMessagesForDialect(shardIds, now, unprocessedFilters(options))
 
-  return yield* MessageStorage.makeEncoded({
+  const encoded: MessageStorage.Encoded = {
     saveEnvelope: ({ deliverAt, envelope, primaryKey }) =>
       Effect.suspend(() => {
         let insert: Effect.Effect<ReadonlyArray<Row>, SqlError | PlatformError.PlatformError>
@@ -579,8 +642,8 @@ export const make: (options?: {
       ),
 
     unprocessedMessages: Effect.fnUntraced(
-      function*(shardIds, now) {
-        const rows = yield* getUnprocessedMessages(shardIds, now)
+      function*(shardIds, now, options) {
+        const rows = yield* getUnprocessedMessages(shardIds, now, options)
         if (rows.length === 0) {
           return []
         }
@@ -623,19 +686,30 @@ export const make: (options?: {
       )
     },
 
-    resetAddress: (address) =>
-      sql`
+    resetAddresses: (addresses) =>
+      addresses.length === 0
+        ? Effect.void
+        : sql`
         UPDATE ${messagesTableSql}
         SET last_read = NULL
         WHERE processed = ${sqlFalse}
-        AND shard_id = ${address.shardId.toString()}
-        AND entity_type = ${address.entityType}
-        AND entity_id = ${address.entityId}
-      `.pipe(
-        Effect.asVoid,
-        PersistenceError.refail,
-        withTracerDisabled
-      ),
+        AND (${
+          sql.or(
+            groupAddresses(addresses).map(
+              (group) =>
+                sql.and([
+                  sql`shard_id = ${group.shardId}`,
+                  sql`entity_type = ${group.entityType}`,
+                  sql.in("entity_id", group.addresses.map((address) => address.entityId))
+                ])
+            )
+          )
+        })
+        `.pipe(
+          Effect.asVoid,
+          PersistenceError.refail,
+          withTracerDisabled
+        ),
 
     clearAddress: (address) =>
       sql`
@@ -675,9 +749,25 @@ export const make: (options?: {
       sql.withTransaction(effect).pipe(
         Effect.catchIf(isSqlError, Effect.die)
       )
-  })
+  }
+  return encoded
 }, withTracerDisabled)
 
+/**
+ * Creates a SQL-backed `MessageStorage` implementation, running its migrations
+ * and using the optional table prefix.
+ *
+ * @category constructors
+ * @since 4.0.0
+ */
+export const make: (options?: {
+  readonly prefix?: string | undefined
+}) => Effect.Effect<
+  MessageStorage.MessageStorage["Service"],
+  never,
+  SqlClient.SqlClient | Snowflake.Generator | Crypto.Crypto
+> = (options) => Effect.flatMap(makeEncoded(options), MessageStorage.makeEncoded)
+
 /**
  * Layer that provides SQL-backed `MessageStorage` using the default table prefix
  * and the default snowflake generator.
@@ -1025,6 +1115,20 @@ const migrations = (options?: {
           // sqlite
           Effect.void
       })
+    }),
+    "0003_pg_messages_rowid_index": Effect.gen(function*() {
+      const sql = (yield* SqlClient.SqlClient).withoutTransforms()
+      const messagesTableSql = sql(messagesTable)
+      const rowIdIndex = `${messagesTable}_rowid_idx`
+
+      yield* sql.onDialectOrElse({
+        pg: () =>
+          sql`
+            CREATE INDEX IF NOT EXISTS ${sql(rowIdIndex)}
+            ON ${messagesTableSql} (rowid)
+          `,
+        orElse: () => Effect.void
+      })
     })
   })
 }
diff --git a/packages/effect/src/unstable/cluster/internal/entityManager.ts b/packages/effect/src/unstable/cluster/internal/entityManager.ts
index e568d3d07e2..b0f62f937d4 100644
--- a/packages/effect/src/unstable/cluster/internal/entityManager.ts
+++ b/packages/effect/src/unstable/cluster/internal/entityManager.ts
@@ -56,6 +56,9 @@ export interface EntityManager {
   }) => boolean
   readonly clearProcessed: () => void
 
+  readonly isResidentUnsafe: (address: EntityAddress) => boolean
+  readonly residentAddressesUnsafe: () => Array<EntityAddress>
+
   readonly interruptShard: (shardId: ShardId, options?: {
     readonly force?: boolean
   }) => Effect.Effect<void>
@@ -63,6 +66,19 @@ export interface EntityManager {
   readonly activeEntityCount: Effect.Effect<number>
 }
 
+// Tracks how many entities are resident on the runner across all entity
+// managers, so the spawn of new entities can be gated by
+// `ShardingConfig.maxResidentEntities`.
+/** @internal */
+export interface Residency {
+  /**
+   * Reserve a slot for a new entity. Returns `false` when the runner is at
+   * capacity.
+   */
+  readonly admitUnsafe: () => boolean
+  readonly releaseUnsafe: () => void
+}
+
 // Represents the entities managed by this entity manager
 /** @internal */
 export type EntityState = {
@@ -94,6 +110,7 @@ export const make = Effect.fnUntraced(function*<
     readonly sharding: Sharding["Service"]
     readonly storage: MessageStorage.MessageStorage["Service"]
     readonly runnerAddress: RunnerAddress
+    readonly residency: Residency
     readonly maxIdleTime?: Input | undefined
     readonly concurrency?: number | "unbounded" | undefined
     readonly mailboxCapacity?: number | "unbounded" | undefined
@@ -128,13 +145,22 @@ export const make = Effect.fnUntraced(function*<
   const entities: ResourceMap<
     EntityAddress,
     EntityState,
-    EntityNotAssignedToRunner
+    EntityNotAssignedToRunner | MailboxFull
   > = yield* ResourceMap.make(Effect.fnUntraced(function*(address: EntityAddress) {
     if (!options.sharding.hasShardId(address.shardId)) {
       return yield* new EntityNotAssignedToRunner({ address })
     }
 
     const scope = yield* Effect.scope
+
+    // Gate the spawn on the runner-wide entity cap. Registering the release
+    // must be atomic with taking the slot, otherwise an interrupt in between
+    // would leak it.
+    yield* Effect.uninterruptible(Effect.suspend(() =>
+      options.residency.admitUnsafe()
+        ? Scope.addFinalizer(scope, Effect.sync(options.residency.releaseUnsafe))
+        : Effect.fail(new MailboxFull({ address }))
+    ))
     const endLatch = Latch.makeUnsafe()
     const keepAliveLatch = Latch.makeUnsafe()
     const closeLatches = {
@@ -590,6 +616,8 @@ export const make = Effect.fnUntraced(function*<
     clearProcessed() {
       processedRequestIds.clear()
     },
+    isResidentUnsafe: (address) => entities.hasUnsafe(address),
+    residentAddressesUnsafe: () => entities.keysUnsafe(),
     sendLocal,
     send: (message) =>
       decodeMessage(message).pipe(
diff --git a/packages/effect/src/unstable/cluster/internal/entityReaper.ts b/packages/effect/src/unstable/cluster/internal/entityReaper.ts
index 0c3c974b813..d50c602fc24 100644
--- a/packages/effect/src/unstable/cluster/internal/entityReaper.ts
+++ b/packages/effect/src/unstable/cluster/internal/entityReaper.ts
@@ -3,7 +3,6 @@ import * as Context from "../../../Context.ts"
 import * as Effect from "../../../Effect.ts"
 import * as Latch from "../../../Latch.ts"
 import * as Layer from "../../../Layer.ts"
-import type { EntityNotAssignedToRunner } from "../ClusterError.ts"
 import type { EntityAddress } from "../EntityAddress.ts"
 import type { EntityId } from "../EntityId.ts"
 import type { EntityState } from "./entityManager.ts"
@@ -16,14 +15,14 @@ export class EntityReaper extends Context.Service<EntityReaper>()("effect/cluste
     const registered: Array<{
       readonly maxIdleTime: number
       readonly servers: Map<EntityId, EntityState>
-      readonly entities: ResourceMap<EntityAddress, EntityState, EntityNotAssignedToRunner>
+      readonly entities: ResourceMap<EntityAddress, EntityState, any>
     }> = []
     const latch = yield* Latch.make()
 
     const register = (options: {
       readonly maxIdleTime: number
       readonly servers: Map<EntityId, EntityState>
-      readonly entities: ResourceMap<EntityAddress, EntityState, EntityNotAssignedToRunner>
+      readonly entities: ResourceMap<EntityAddress, EntityState, any>
     }) =>
       Effect.suspend(() => {
         currentResolution = Math.max(Math.min(currentResolution, options.maxIdleTime), 5000)
diff --git a/packages/effect/src/unstable/cluster/internal/resourceMap.ts b/packages/effect/src/unstable/cluster/internal/resourceMap.ts
index 13377199b60..e09e1f67b41 100644
--- a/packages/effect/src/unstable/cluster/internal/resourceMap.ts
+++ b/packages/effect/src/unstable/cluster/internal/resourceMap.ts
@@ -56,6 +56,14 @@ export class ResourceMap<K, A, E> {
     )
   })
 
+  hasUnsafe(key: K): boolean {
+    return backingGet(this.entries, key) !== undefined
+  }
+
+  keysUnsafe(): Array<K> {
+    return Array.from(this.entries.map, ([key]) => key)
+  }
+
   get(key: K): Effect.Effect<A, E> {
     return Effect.suspend(() => {
       if (MutableRef.get(this.isClosed)) {
diff --git a/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts b/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts
index 150e6c0434e..1232a5aad17 100644
--- a/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts
+++ b/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts
@@ -736,24 +736,26 @@ describe.concurrent("ClusterWorkflowEngine", () => {
     }).pipe(Effect.provide(TestWorkflowLayer)))
 })
 
-const TestShardingConfig = ShardingConfig.layer({
-  shardsPerGroup: 300,
-  availableShardGroups: ["default", "workflow"],
-  assignedShardGroups: ["default", "workflow"],
-  entityMailboxCapacity: 10,
-  entityTerminationTimeout: 0,
-  entityMessagePollInterval: 5000,
-  sendRetryInterval: 100
-})
+const makeTestWorkflowEngine = (config?: Partial<ShardingConfig.ShardingConfig["Service"]>) =>
+  ClusterWorkflowEngine.layer.pipe(
+    Layer.provideMerge(Sharding.layer),
+    Layer.provide(Runners.layerNoop),
+    Layer.provideMerge(MessageStorage.layerMemory),
+    Layer.provide(RunnerStorage.layerMemory),
+    Layer.provide(RunnerHealth.layerNoop),
+    Layer.provide(ShardingConfig.layer({
+      shardsPerGroup: 300,
+      availableShardGroups: ["default", "workflow"],
+      assignedShardGroups: ["default", "workflow"],
+      entityMailboxCapacity: 10,
+      entityTerminationTimeout: 0,
+      entityMessagePollInterval: 5000,
+      sendRetryInterval: 100,
+      ...config
+    }))
+  )
 
-const TestWorkflowEngine = ClusterWorkflowEngine.layer.pipe(
-  Layer.provideMerge(Sharding.layer),
-  Layer.provide(Runners.layerNoop),
-  Layer.provideMerge(MessageStorage.layerMemory),
-  Layer.provide(RunnerStorage.layerMemory),
-  Layer.provide(RunnerHealth.layerNoop),
-  Layer.provide(TestShardingConfig)
-)
+const TestWorkflowEngine = makeTestWorkflowEngine()
 
 class SendEmailError extends Schema.Error<SendEmailError>("SendEmailError")({
   _tag: Schema.tag("SendEmailError"),
diff --git a/packages/effect/test/cluster/MessageStorage.test.ts b/packages/effect/test/cluster/MessageStorage.test.ts
index ba9b39061e3..e85096cf822 100644
--- a/packages/effect/test/cluster/MessageStorage.test.ts
+++ b/packages/effect/test/cluster/MessageStorage.test.ts
@@ -49,6 +49,41 @@ describe("MessageStorage", () => {
         expect(result._tag).toEqual("Success")
       }).pipe(Effect.provide(MessageStorage.MemoryDriver.layer)))
 
+    it.effect("encoded unprocessedMessages fails closed for an empty address filter", () =>
+      Effect.gen(function*() {
+        const driver = yield* MessageStorage.MemoryDriver
+        const envelope: Envelope.PartialRequestEncoded = {
+          _tag: "Request",
+          requestId: "1",
+          address: { shardId: { group: "default", id: 1 }, entityType: "Repro", entityId: "one" },
+          tag: "Repro",
+          payload: {},
+          headers: {}
+        }
+        yield* driver.encoded.saveEnvelope({ envelope, primaryKey: null, deliverAt: null })
+        const messages = yield* driver.encoded.unprocessedMessages(["default:1"], 0, { addresses: [] })
+        expect(messages).toHaveLength(0)
+      }).pipe(Effect.provide(MessageStorage.MemoryDriver.layer)))
+
+    it.effect("encoded resetAddresses fails closed for an empty address list", () =>
+      Effect.gen(function*() {
+        const driver = yield* MessageStorage.MemoryDriver
+        const envelope: Envelope.PartialRequestEncoded = {
+          _tag: "Request",
+          requestId: "1",
+          address: { shardId: { group: "default", id: 1 }, entityType: "Repro", entityId: "one" },
+          tag: "Repro",
+          payload: {},
+          headers: {}
+        }
+        yield* driver.encoded.saveEnvelope({ envelope, primaryKey: null, deliverAt: null })
+        const claimed = yield* driver.encoded.unprocessedMessages(["default:1"], 0)
+        expect(claimed).toHaveLength(1)
+        yield* driver.encoded.resetAddresses([])
+        const messages = yield* driver.encoded.unprocessedMessages(["default:1"], 1)
+        expect(messages).toHaveLength(0)
+      }).pipe(Effect.provide(MessageStorage.MemoryDriver.layer)))
+
     it.effect("saves a request", () =>
       Effect.gen(function*() {
         const storage = yield* MessageStorage.MessageStorage
@@ -86,6 +121,61 @@ describe("MessageStorage", () => {
         const messages = yield* storage.unprocessedMessages([request.envelope.address.shardId])
         expect(messages).toHaveLength(0)
       }).pipe(Effect.provide(MemoryLayer)))
+    it.effect("unprocessedMessages honors the limit option", () =>
+      Effect.gen(function*() {
+        const storage = yield* MessageStorage.MessageStorage
+        for (let i = 1; i <= 5; i++) {
+          yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) }))
+        }
+        const shardId = ShardId.make("default", 1)
+        const messages = yield* storage.unprocessedMessages([shardId], { limit: 3 })
+        expect(messages).toHaveLength(3)
+        expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 2, 3])
+        const remaining = yield* storage.unprocessedMessages([shardId])
+        expect(remaining.map((m: any) => m.envelope.payload.id)).toEqual([4, 5])
+      }).pipe(Effect.provide(MemoryLayer)))
+
+    it.effect("unprocessedMessages filters by address", () =>
+      Effect.gen(function*() {
+        const storage = yield* MessageStorage.MessageStorage
+        for (let i = 1; i <= 4; i++) {
+          yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) }))
+        }
+        const shardId = ShardId.make("default", 1)
+        const address = (entityId: string) =>
+          EntityAddress.make({
+            shardId,
+            entityType: EntityType.make("test"),
+            entityId: EntityId.make(entityId)
+          })
+        const messages = yield* storage.unprocessedMessages([shardId], {
+          addresses: [address("2"), address("4")]
+        })
+        expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([2, 4])
+        // an empty address filter returns nothing
+        const none = yield* storage.unprocessedMessages([shardId], { addresses: [] })
+        expect(none).toHaveLength(0)
+      }).pipe(Effect.provide(MemoryLayer)))
+
+    it.effect("resetAddresses makes claimed messages eligible again", () =>
+      Effect.gen(function*() {
+        const storage = yield* MessageStorage.MessageStorage
+        for (let i = 1; i <= 4; i++) {
+          yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) }))
+        }
+        const shardId = ShardId.make("default", 1)
+        yield* storage.unprocessedMessages([shardId], { limit: 3 })
+        yield* storage.resetAddresses(["1", "3"].map((entityId) =>
+          EntityAddress.make({
+            shardId,
+            entityType: EntityType.make("test"),
+            entityId: EntityId.make(entityId)
+          })
+        ))
+
+        const messages = yield* storage.unprocessedMessages([shardId])
+        expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 3, 4])
+      }).pipe(Effect.provide(MemoryLayer)))
 
     it.effect("repliesFor", () =>
       Effect.gen(function*() {
@@ -127,6 +217,7 @@ export const GetUserRpc = Rpc.make("GetUser", {
 export const makeRequest = Effect.fnUntraced(function*(options?: {
   readonly rpc?: Rpc.AnyWithProps
   readonly payload?: any
+  readonly entityId?: string
 }) {
   const snowflake = yield* Snowflake.Generator
   const rpc = options?.rpc ?? GetUserRpc
@@ -136,7 +227,7 @@ export const makeRequest = Effect.fnUntraced(function*(options?: {
       address: EntityAddress.make({
         shardId: ShardId.make("default", 1),
         entityType: EntityType.make("test"),
-        entityId: EntityId.make("1")
+        entityId: EntityId.make(options?.entityId ?? "1")
       }),
       tag: rpc._tag,
       payload: options?.payload ?? { id: 123 },
diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts
index 651bee01225..f7ed7ec6eb5 100644
--- a/packages/effect/test/cluster/Sharding.test.ts
+++ b/packages/effect/test/cluster/Sharding.test.ts
@@ -23,8 +23,12 @@ import {
   ClusterMetrics,
   ClusterSchema,
   Entity,
+  EntityAddress,
   EntityId,
+  EntityType,
+  Envelope,
   MachineId,
+  Message,
   MessageStorage,
   Runner,
   RunnerAddress,
@@ -36,6 +40,7 @@ import {
   ShardingConfig,
   Snowflake
 } from "effect/unstable/cluster"
+import { Headers } from "effect/unstable/http"
 import { Rpc } from "effect/unstable/rpc"
 import {
   CallerId,
@@ -1070,6 +1075,198 @@ describe.concurrent("Sharding", () => {
     }))
 })
 
+describe.concurrent("Sharding residency cap", () => {
+  it.effect("bounds resident entities to maxResidentEntities", () =>
+    Effect.gen(function*() {
+      yield* TestClock.adjust(1)
+      const sharding = yield* Sharding.Sharding
+      const makeClient = yield* TestEntity.client
+      const fibers: Array<Fiber.Fiber<User, any>> = []
+      for (let i = 1; i <= 6; i++) {
+        fibers.push(
+          yield* makeClient(String(i)).GetUser({ id: i }).pipe(
+            Effect.forkChild({ startImmediately: true })
+          )
+        )
+      }
+      yield* TestClock.adjust(1)
+      assert.isAtMost(yield* sharding.activeEntityCount, 2)
+
+      // idle entities are reaped over time, freeing slots for the backlog
+      for (let i = 0; i < 12; i++) {
+        yield* TestClock.adjust(5000)
+        assert.isAtMost(yield* sharding.activeEntityCount, 2)
+      }
+
+      const users = yield* Fiber.joinAll(fibers)
+      assert.deepStrictEqual(users.map((user) => user.id), [1, 2, 3, 4, 5, 6])
+    }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 2, entityMaxIdleTime: 1000 }))))
+
+  it.effect("volatile sends to new entities fail with MailboxFull at the cap", () =>
+    Effect.gen(function*() {
+      yield* TestClock.adjust(1)
+      const driver = yield* MessageStorage.MemoryDriver
+      const sharding = yield* Sharding.Sharding
+      const makeClient = yield* TestEntity.client
+      // occupy the only slot
+      yield* makeClient("1").NeverVolatile().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust(1)
+      assert.strictEqual(yield* sharding.activeEntityCount, 1)
+
+      const error = yield* makeClient("2").GetUserVolatile({ id: 2 }).pipe(Effect.flip)
+      assert.strictEqual(error._tag, "MailboxFull")
+
+      // persisted sends still succeed and wait in storage
+      yield* makeClient("2").GetUser({ id: 2 }, { discard: true })
+      yield* TestClock.adjust(5000)
+      assert.strictEqual(yield* sharding.activeEntityCount, 1)
+      assert.strictEqual(driver.unprocessed.size, 1)
+      assert.strictEqual(driver.replyIds.size, 0)
+    }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 1 }))))
+
+  it.effect("keeps delivering to resident entities at the cap", () =>
+    Effect.gen(function*() {
+      yield* TestClock.adjust(1)
+      const driver = yield* MessageStorage.MemoryDriver
+      const state = yield* TestEntityState
+      const sharding = yield* Sharding.Sharding
+      const makeClient = yield* TestEntity.client
+      // make entity "1" resident and keep it busy
+      yield* makeClient("1").NeverFork().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust(1)
+      assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1)
+
+      // a backlog of new entity ids in front of the resident's next message
+      for (let i = 2; i <= 4; i++) {
+        yield* makeClient(String(i)).GetUser({ id: i }, { discard: true })
+      }
+      const fiber = yield* makeClient("1").GetUser({ id: 1 }).pipe(
+        Effect.forkChild({ startImmediately: true })
+      )
+      yield* TestClock.adjust(5000)
+
+      // the resident entity received its message despite the backlog in front
+      assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 1, name: "User 1" }))
+      // the new ids were not admitted and their requests stay in storage
+      assert.strictEqual(yield* sharding.activeEntityCount, 1)
+      assert.strictEqual(driver.unprocessed.size, 4)
+      assert.strictEqual(driver.replyIds.size, 1)
+    }).pipe(Effect.provide(CappedSharding({
+      maxResidentEntities: 1,
+      unprocessedMessageBatchSize: 2
+    }))))
+
+  it.effect("drains a full batch without waiting for the poll interval", () =>
+    Effect.gen(function*() {
+      yield* TestClock.adjust(1)
+      const driver = yield* MessageStorage.MemoryDriver
+
+      for (let i = 1; i <= 5; i++) {
+        yield* saveGetUserRequest(String(i), i)
+      }
+      assert.strictEqual(driver.replyIds.size, 0)
+
+      // a single poll drains the whole backlog in batches of 2
+      yield* TestClock.adjust(5000)
+      assert.strictEqual(driver.replyIds.size, 5)
+    }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize: 2 }))))
+
+  it.effect("advances past in-flight memory requests when reading bounded batches", () =>
+    Effect.gen(function*() {
+      yield* TestClock.adjust(1)
+      const makeClient = yield* TestEntity.client
+      yield* makeClient("1").Never().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* makeClient("2").Never().pipe(Effect.forkChild({ startImmediately: true }))
+      const fiber = yield* makeClient("3").GetUser({ id: 3 }).pipe(
+        Effect.forkChild({ startImmediately: true })
+      )
+
+      yield* TestClock.adjust(5000)
+      assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 3, name: "User 3" }))
+    }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize: 2 }))))
+
+  it.effect("keeps full-sized storage reads when approaching the entity cap", () =>
+    Effect.gen(function*() {
+      const limits: Array<number> = []
+      yield* Effect.gen(function*() {
+        yield* TestClock.adjust(1)
+        const sharding = yield* Sharding.Sharding
+        const makeClient = yield* TestEntity.client
+        for (let i = 1; i <= 4; i++) {
+          yield* makeClient(String(i)).NeverVolatile().pipe(Effect.forkChild({ startImmediately: true }))
+        }
+        yield* TestClock.adjust(1)
+        assert.strictEqual(yield* sharding.activeEntityCount, 4)
+        limits.length = 0
+
+        for (let i = 0; i < 10; i++) {
+          yield* saveGetUserRequest(String(i % 4 + 1), i)
+        }
+        yield* TestClock.adjust(5000)
+        assert.isNotEmpty(limits)
+        assert.deepStrictEqual([...new Set(limits)], [100])
+      }).pipe(Effect.provide(CappedSharding({
+        maxResidentEntities: 5,
+        unprocessedMessageBatchSize: 100
+      }, (storage) => ({
+        ...storage,
+        unprocessedMessages(shardIds, options) {
+          if (options?.limit !== undefined) limits.push(options.limit)
+          return storage.unprocessedMessages(shardIds, options)
+        }
+      }))))
+    }))
+
+  it.effect("normalizes programmatic batch sizes below one", () =>
+    Effect.forEach([0, -1], (unprocessedMessageBatchSize) =>
+      Effect.gen(function*() {
+        yield* TestClock.adjust(1)
+        const makeClient = yield* TestEntity.client
+        const fiber = yield* makeClient("1").GetUser({ id: 1 }).pipe(
+          Effect.forkChild({ startImmediately: true })
+        )
+
+        yield* TestClock.adjust(5000)
+        assert.isDefined(fiber.pollUnsafe())
+        assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 1, name: "User 1" }))
+      }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize }))), { discard: true }))
+
+  it.effect("does not busy-poll storage at the cap", () =>
+    Effect.gen(function*() {
+      let reads = 0
+      yield* Effect.gen(function*() {
+        yield* TestClock.adjust(1)
+        const makeClient = yield* TestEntity.client
+        yield* makeClient("1").NeverVolatile().pipe(Effect.forkChild({ startImmediately: true }))
+        yield* TestClock.adjust(1)
+        // a backlog for a new id that cannot be admitted
+        yield* makeClient("2").GetUser({ id: 2 }, { discard: true })
+        yield* TestClock.adjust(1)
+        const before = reads
+        yield* TestClock.adjust(4000) // less than the poll interval
+        assert.isAtMost(reads - before, 1)
+      }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 1 }, (storage) => ({
+        ...storage,
+        unprocessedMessages(shardIds, options) {
+          reads++
+          return storage.unprocessedMessages(shardIds, options)
+        }
+      }))))
+    }))
+
+  it.effect("unbounded maxResidentEntities does not limit spawning", () =>
+    Effect.gen(function*() {
+      yield* TestClock.adjust(1)
+      const sharding = yield* Sharding.Sharding
+      const makeClient = yield* TestEntity.client
+      for (let i = 1; i <= 15; i++) {
+        yield* makeClient(String(i)).NeverVolatile().pipe(Effect.forkChild({ startImmediately: true }))
+      }
+      yield* TestClock.adjust(1)
+      assert.strictEqual(yield* sharding.activeEntityCount, 15)
+    }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: "unbounded" }))))
+})
+
 describe("Sharding shard lock failover", () => {
   it.effect("interrupts entities and reacquires shards after lock storage recovers", () =>
     Effect.gen(function*() {
@@ -1463,13 +1660,15 @@ const otherRunner = Runner.make({
   weight: 1
 })
 
-const TestShardingConfig = ShardingConfig.layer({
+const testConfigDefaults: Partial<ShardingConfig.ShardingConfig["Service"]> = {
   entityMailboxCapacity: 10,
   entityTerminationTimeout: 0,
   entityMessagePollInterval: 5000,
   sendRetryInterval: 100,
   refreshAssignmentsInterval: 0
-})
+}
+
+const TestShardingConfig = ShardingConfig.layer(testConfigDefaults)
 
 const TestShardingWithoutState = TestEntityNoState.pipe(
   Layer.provideMerge(Sharding.layer),
@@ -1493,4 +1692,53 @@ const TestSharding = TestShardingWithoutStorage.pipe(
   Layer.provide(TestShardingConfig)
 )
 
+const CappedSharding = (
+  config: Partial<ShardingConfig.ShardingConfig["Service"]>,
+  transformStorage?: (
+    storage: MessageStorage.MessageStorage["Service"]
+  ) => MessageStorage.MessageStorage["Service"]
+) => {
+  const configLayer = ShardingConfig.layer({ ...testConfigDefaults, ...config })
+  let layer = TestShardingWithoutRunners.pipe(
+    Layer.provide(Runners.layerNoop),
+    Layer.provide(configLayer)
+  )
+  if (transformStorage) {
+    layer = layer.pipe(Layer.updateService(MessageStorage.MessageStorage, transformStorage))
+  }
+  return layer.pipe(
+    Layer.provideMerge(MessageStorage.layerMemory),
+    Layer.provide(configLayer)
+  )
+}
+
 const ContextBleedSharding = ContextBleedLayer.pipe(Layer.provideMerge(TestSharding))
+
+// saves a persisted GetUser request directly to storage, bypassing the client,
+// so the storage read loop only learns about it from the next poll
+const saveGetUserRequest = Effect.fnUntraced(function*(entityId: string, id: number) {
+  const storage = yield* MessageStorage.MessageStorage
+  const sharding = yield* Sharding.Sharding
+  const rpc = TestEntity.protocol.requests.get("GetUser")! as any
+  const entity = EntityId.make(entityId)
+  yield* storage.saveRequest(
+    new Message.OutgoingRequest({
+      envelope: Envelope.makeRequest<any>({
+        requestId: yield* sharding.getSnowflake,
+        address: EntityAddress.make({
+          shardId: sharding.getShardId(entity, "default"),
+          entityType: EntityType.make(TestEntity.type),
+          entityId: entity
+        }),
+        tag: "GetUser",
+        payload: { id },
+        headers: Headers.empty
+      }),
+      annotations: rpc.annotations,
+      context: Context.empty() as any,
+      rpc,
+      lastReceivedReply: Option.none(),
+      respond: () => Effect.void
+    })
+  )
+})
diff --git a/packages/effect/test/cluster/ShardingConfig.test.ts b/packages/effect/test/cluster/ShardingConfig.test.ts
index 44ef9a893f2..bf8762f39b7 100644
--- a/packages/effect/test/cluster/ShardingConfig.test.ts
+++ b/packages/effect/test/cluster/ShardingConfig.test.ts
@@ -3,6 +3,33 @@ import { ConfigProvider, Effect, Option } from "effect"
 import { RunnerAddress, ShardingConfig } from "effect/unstable/cluster"
 
 describe("ShardingConfig", () => {
+  it.effect("bounds runner residency and storage reads by default", () =>
+    Effect.gen(function*() {
+      assert.strictEqual(ShardingConfig.defaults.maxResidentEntities, 10_000)
+      assert.strictEqual(ShardingConfig.defaults.unprocessedMessageBatchSize, 1024)
+
+      const defaults = yield* ShardingConfig.config.parse(ConfigProvider.fromUnknown({}))
+      assert.strictEqual(defaults.maxResidentEntities, 10_000)
+      assert.strictEqual(defaults.unprocessedMessageBatchSize, 1024)
+
+      // environment values are integers only
+      const fromEnv = yield* ShardingConfig.config.parse(
+        ConfigProvider.fromUnknown({ maxResidentEntities: "500", unprocessedMessageBatchSize: "128" })
+      )
+      assert.strictEqual(fromEnv.maxResidentEntities, 500)
+      assert.strictEqual(fromEnv.unprocessedMessageBatchSize, 128)
+
+      const invalid = yield* ShardingConfig.config.parse(
+        ConfigProvider.fromUnknown({ maxResidentEntities: "unbounded" })
+      ).pipe(Effect.flip)
+      assert.strictEqual(invalid._tag, "ConfigError")
+
+      // "unbounded" is only available programmatically
+      const layer = ShardingConfig.layer({ maxResidentEntities: "unbounded" })
+      const config = yield* Effect.provide(ShardingConfig.ShardingConfig, layer)
+      assert.strictEqual(config.maxResidentEntities, "unbounded")
+    }))
+
   it.effect("treats the optional listen address as an atomic group", () =>
     Effect.gen(function*() {
       const defaults = yield* ShardingConfig.config.parse(ConfigProvider.fromUnknown({}))
@@ -34,4 +61,19 @@ describe("ShardingConfig", () => {
   at ["listenPort"]`
       )
     }))
+
+  it.effect("rejects non-positive residency and batch limits", () =>
+    Effect.gen(function*() {
+      for (
+        const values of [
+          { maxResidentEntities: "0" },
+          { maxResidentEntities: "-1" },
+          { unprocessedMessageBatchSize: "0" },
+          { unprocessedMessageBatchSize: "-1" }
+        ]
+      ) {
+        const error = yield* ShardingConfig.config.parse(ConfigProvider.fromUnknown(values)).pipe(Effect.flip)
+        assert.strictEqual(error._tag, "ConfigError")
+      }
+    }))
 })
diff --git a/packages/platform/node/test/cluster-integration/Residency.test.ts b/packages/platform/node/test/cluster-integration/Residency.test.ts
new file mode 100644
index 00000000000..509c176b332
--- /dev/null
+++ b/packages/platform/node/test/cluster-integration/Residency.test.ts
@@ -0,0 +1,76 @@
+import { assert, describe, it } from "@effect/vitest"
+import { Effect, Fiber, Latch, Schema } from "effect"
+import { ClusterSchema, Entity } from "effect/unstable/cluster"
+import { Rpc } from "effect/unstable/rpc"
+import { type Backend, make } from "./harness.ts"
+
+const ResidencyEntity = Entity.make("ClusterIntegrationResidency", [
+  Rpc.make("Ping", {
+    payload: { id: Schema.String },
+    success: Schema.String
+  }),
+  Rpc.make("Hold", {
+    payload: { id: Schema.String }
+  })
+]).annotateRpcs(ClusterSchema.Persisted, true)
+
+let holdGate = Latch.makeUnsafe()
+
+const ResidencyEntityLayer = ResidencyEntity.toLayer({
+  Ping: ({ payload }) => Effect.succeed(payload.id),
+  Hold: () => holdGate.await
+}, { maxIdleTime: "1 second", concurrency: 2 })
+
+describe("cluster residency integration", () => {
+  for (const backend of ["pg", "mysql"] satisfies ReadonlyArray<Backend>) {
+    it.live(`${backend}: stays at the entity cap and keeps residents moving`, () =>
+      Effect.gen(function*() {
+        holdGate = Latch.makeUnsafe()
+        const cluster = yield* make({
+          backend,
+          entities: ResidencyEntityLayer,
+          config: { maxResidentEntities: 3 }
+        })
+        const [runner] = yield* cluster.start(1)
+        yield* cluster.waitForStableAssignments()
+        const client = yield* cluster.getClient(ResidencyEntity)
+
+        // fill every entity slot with a held request
+        const held = yield* Effect.forEach(
+          ["held-1", "held-2", "held-3"],
+          (id) => client(id).Hold({ id }).pipe(Effect.forkChild({ startImmediately: true }))
+        )
+        yield* cluster.waitUntil(
+          "The held entities did not become resident",
+          Effect.map(runner.sharding.activeEntityCount, (count) => count === 3)
+        )
+
+        // a backlog of messages for new entity ids stays in storage
+        for (let i = 1; i <= 8; i++) {
+          yield* client(`backlog-${i}`).Ping({ id: `backlog-${i}` }, { discard: true })
+        }
+
+        // resident entities keep receiving messages while at the cap
+        assert.strictEqual(yield* client("held-1").Ping({ id: "first" }), "first")
+        assert.strictEqual(yield* client("held-2").Ping({ id: "second" }), "second")
+
+        // the runner never spawned entities for the backlog
+        assert.strictEqual(yield* runner.sharding.activeEntityCount, 3)
+        assert.isAtLeast(yield* cluster.unprocessedMessageCount, 8)
+
+        // release the held entities: as slots free up, the backlog drains in
+        // waves without ever exceeding the cap
+        holdGate.openUnsafe()
+        yield* Fiber.joinAll(held)
+        yield* cluster.waitUntil(
+          "The backlog did not drain after entity slots freed up",
+          Effect.gen(function*() {
+            assert.isAtMost(yield* runner.sharding.activeEntityCount, 3)
+            const counts = yield* cluster.messageCounts()
+            return counts.unprocessed === 0 && counts.failed === 0
+          }),
+          "45 seconds"
+        )
+      }))
+  }
+})
diff --git a/packages/platform/node/test/cluster/MessageStorageTest.ts b/packages/platform/node/test/cluster/MessageStorageTest.ts
index 3dc448009f1..33c415e482e 100644
--- a/packages/platform/node/test/cluster/MessageStorageTest.ts
+++ b/packages/platform/node/test/cluster/MessageStorageTest.ts
@@ -101,6 +101,8 @@ export const GetUserRpc = Rpc.make("GetUser", {
 export const makeRequest = Effect.fnUntraced(function*(options?: {
   readonly rpc?: Rpc.AnyWithProps
   readonly payload?: any
+  readonly entityId?: string
+  readonly shardId?: ShardId.ShardId
 }) {
   const snowflake = yield* Snowflake.Generator
   const rpc = options?.rpc ?? GetUserRpc
@@ -108,9 +110,9 @@ export const makeRequest = Effect.fnUntraced(function*(options?: {
     envelope: Envelope.makeRequest<any>({
       requestId: snowflake.nextUnsafe(),
       address: EntityAddress.make({
-        shardId: ShardId.make("default", 1),
+        shardId: options?.shardId ?? ShardId.make("default", 1),
         entityType: EntityType.make("test"),
-        entityId: EntityId.make("1")
+        entityId: EntityId.make(options?.entityId ?? "1")
       }),
       tag: rpc._tag,
       payload: options?.payload ?? { id: 123 },
diff --git a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts
index 46a8f6a00e0..762ad924d08 100644
--- a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts
+++ b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts
@@ -5,12 +5,16 @@ import { Effect, Fiber, FileSystem, Latch, Layer, Option } from "effect"
 import { TestClock } from "effect/testing"
 import {
   Entity,
+  EntityAddress,
+  EntityId,
+  EntityType,
   Envelope,
   Message,
   MessageStorage,
   RunnerHealth,
   Runners,
   RunnerStorage,
+  ShardId,
   Sharding,
   ShardingConfig,
   Snowflake,
@@ -55,6 +59,20 @@ describe("SqlMessageStorage", () => {
     it.layer(StorageLayer.pipe(Layer.provideMerge(layer)), {
       timeout: 120000
     })(label, (it) => {
+      if (label === "pg") {
+        it.effect("creates an index for insertion-ordered message reads", () =>
+          Effect.gen(function*() {
+            const sql = yield* SqlClient.SqlClient
+            const indexes = yield* sql<{ indexname: string }>`
+              SELECT indexname
+              FROM pg_indexes
+              WHERE tablename = 'cluster_messages'
+              AND indexname = 'cluster_messages_rowid_idx'
+            `
+            expect(indexes).toHaveLength(1)
+          }))
+      }
+
       it.effect("saveRequest", () =>
         Effect.gen(function*() {
           const storage = yield* MessageStorage.MessageStorage
@@ -282,6 +300,128 @@ describe("SqlMessageStorage", () => {
           expect(messages).toHaveLength(1)
         }))
 
+      it.effect("unprocessedMessages honors the limit and claims only returned rows", () =>
+        Effect.gen(function*() {
+          yield* truncate
+
+          const storage = yield* MessageStorage.MessageStorage
+          const shardId = ShardId.make("default", 1)
+          for (let i = 1; i <= 5; i++) {
+            yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) }))
+          }
+          const limited = yield* storage.unprocessedMessages([shardId], { limit: 3 })
+          expect(limited).toHaveLength(3)
+          expect(limited.map((m: any) => m.envelope.payload.id)).toEqual([1, 2, 3])
+
+          // rows beyond the limit were not claimed and are returned by the
+          // next read
+          const rest = yield* storage.unprocessedMessages([shardId])
+          expect(rest.map((m: any) => m.envelope.payload.id)).toEqual([4, 5])
+        }))
+
+      it.effect("unprocessedMessages filters by address", () =>
+        Effect.gen(function*() {
+          yield* truncate
+
+          const storage = yield* MessageStorage.MessageStorage
+          const shardId = ShardId.make("default", 1)
+          const address = (entityId: string) =>
+            EntityAddress.make({
+              shardId,
+              entityType: EntityType.make("test"),
+              entityId: EntityId.make(entityId)
+            })
+          for (let i = 1; i <= 4; i++) {
+            yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) }))
+          }
+          const filtered = yield* storage.unprocessedMessages([shardId], {
+            addresses: [address("2"), address("4")]
+          })
+          expect(filtered.map((m: any) => m.envelope.payload.id)).toEqual([2, 4])
+
+          // the filtered read must not claim the other addresses
+          const rest = yield* storage.unprocessedMessages([shardId])
+          expect(rest.map((m: any) => m.envelope.payload.id)).toEqual([1, 3])
+        }))
+
+      it.effect("unprocessedMessages filters addresses by shard", () =>
+        Effect.gen(function*() {
+          yield* truncate
+
+          const storage = yield* MessageStorage.MessageStorage
+          const shardOne = ShardId.make("default", 1)
+          const shardTwo = ShardId.make("default", 2)
+          yield* storage.saveRequest(yield* makeRequest({ payload: { id: 1 }, shardId: shardOne }))
+          yield* storage.saveRequest(yield* makeRequest({ payload: { id: 2 }, shardId: shardTwo }))
+
+          const messages = yield* storage.unprocessedMessages([shardOne, shardTwo], {
+            addresses: [EntityAddress.make({
+              shardId: shardOne,
+              entityType: EntityType.make("test"),
+              entityId: EntityId.make("1")
+            })]
+          })
+          expect(messages.map((message: any) => message.envelope.payload.id)).toEqual([1])
+
+          const rest = yield* storage.unprocessedMessages([shardOne, shardTwo])
+          expect(rest.map((message: any) => message.envelope.payload.id)).toEqual([2])
+        }))
+
+      it.effect("encoded unprocessedMessages fails closed for empty addresses", () =>
+        Effect.gen(function*() {
+          yield* truncate
+
+          const storage = yield* MessageStorage.MessageStorage
+          const request = yield* makeRequest()
+          yield* storage.saveRequest(request)
+          const encoded = yield* SqlMessageStorage.makeEncoded().pipe(Effect.provide(NodeCrypto.layer))
+          const shardId = request.envelope.address.shardId.toString()
+          const messages = yield* encoded.unprocessedMessages([shardId], Date.now(), { addresses: [] })
+          expect(messages).toHaveLength(0)
+
+          const unfiltered = yield* encoded.unprocessedMessages([shardId], Date.now())
+          expect(unfiltered).toHaveLength(1)
+        }))
+
+      it.effect("encoded resetAddresses fails closed for an empty address list", () =>
+        Effect.gen(function*() {
+          yield* truncate
+
+          const storage = yield* MessageStorage.MessageStorage
+          const request = yield* makeRequest()
+          yield* storage.saveRequest(request)
+          const encoded = yield* SqlMessageStorage.makeEncoded().pipe(Effect.provide(NodeCrypto.layer))
+          const shardId = request.envelope.address.shardId.toString()
+          const claimed = yield* encoded.unprocessedMessages([shardId], Date.now())
+          expect(claimed).toHaveLength(1)
+
+          yield* encoded.resetAddresses([])
+          const messages = yield* encoded.unprocessedMessages([shardId], Date.now())
+          expect(messages).toHaveLength(0)
+        }))
+
+      it.effect("resetAddresses releases claims in one batch", () =>
+        Effect.gen(function*() {
+          yield* truncate
+
+          const storage = yield* MessageStorage.MessageStorage
+          const shardId = ShardId.make("default", 1)
+          const address = (entityId: string) =>
+            EntityAddress.make({
+              shardId,
+              entityType: EntityType.make("test"),
+              entityId: EntityId.make(entityId)
+            })
+          for (let i = 1; i <= 4; i++) {
+            yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) }))
+          }
+          yield* storage.unprocessedMessages([shardId], { limit: 3 })
+          yield* storage.resetAddresses([address("1"), address("3")])
+
+          const messages = yield* storage.unprocessedMessages([shardId])
+          expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 3, 4])
+        }))
+
       it.effect("unprocessedMessages excludes complete requests", () =>
         Effect.gen(function*() {
           yield* truncate