feat: scan claim candidates

Author: cardmagic

CHANGELOG.md

@@ -56,6 +56,8 @@
   and mailbox blockers.
 - Reject committed calls and message waits inside an ambient Solid Objects
   database transaction before they can self-deadlock.
+- Scan a bounded set of ready actor candidates so a lost lease race does not
+  make a worker report idle while other actors are ready.
 
 ## 0.1.0 - 2026-08-13
 

README.md

@@ -763,6 +763,8 @@ IDs and observable values are not authorization.
 - Different actor identities may execute concurrently.
 - A worker drains at most `maxMessagesPerActivationPass` turns from one actor,
   then yields its still-due work behind actors that were already waiting.
+- A global claim scans at most `claimScanLimit` ordered candidates, continuing
+  to another ready actor after a lost lease race.
 - Long-running workers reuse hydrated actors for
   `idleDeactivationTimeoutMilliseconds` while renewing the same fenced lease.
 - State, completion, staged messages, effects, reminders, commit actions, and

docs/architecture.md

@@ -70,6 +70,11 @@ moves only that actor's already-due ready memberships to current database time.
 Actors with older ready work therefore win the next global claim; delayed work
 keeps its original future availability.
 
+The global claim reads at most `claimScanLimit` ordered candidates. If another
+worker acquires the first candidate's lease, the transaction continues through
+that bounded set instead of returning idle and sacrificing parallelism across
+independent actor identities.
+
 When a pass becomes idle, a long-running worker keeps the hydrated actor and
 continues renewing the same fenced lease until its idle timeout. A later turn
 on that actor reuses both its persisted public fields and process-local private

docs/operations.md

@@ -23,6 +23,10 @@ identities stay continuously busy; higher values reduce claim overhead for
 isolated backlogs. `solid_objects.activation.yielded` reports the actor
 identity, turns processed, and remaining due membership count.
 
+`claimScanLimit` defaults to 100. Global claims inspect a bounded ordered set of
+actor identities and continue after a lost lease race, preserving worker
+parallelism without an unbounded scan.
+
 Workers retain a hydrated actor and its fenced lease for
 `idleDeactivationTimeoutMilliseconds`, which defaults to 30 seconds. Idle
 leases renew at `leaseRenewalIntervalMilliseconds`; the worker polling cadence

docs/parity.md

@@ -26,6 +26,7 @@ Reference: Ruby `solid_objects` 0.12.0 at commit `a01b6f5`.
 | Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native  | Relational ready/claimed membership tables, durable history, and adapter-appropriate sequence locking.                                                                                                       |
 | Domain rejection and strict poison ordering                                                                   | Native  | Rejections roll back without retry; retryable failures block later operations until completion or dead-lettering.                                                                                            |
 | Bounded activation passes and hot-actor fairness                                                              | Native  | Configurable turn-count and elapsed-time budgets bound each pass, then move only that actor's already-due memberships behind actors already waiting.                                                         |
+| Bounded claim candidate scan                                                                                  | Native  | A configurable ordered scan continues to another ready actor when a worker loses the first candidate's lease race.                                                                                           |
 | Idle activation cache                                                                                         | Native  | Long-running workers retain hydrated actors under renewable fenced leases, restore public state after failed turns, and release on timeout, fairness yield, lease loss, or shutdown.                         |
 | Transactional effects and outcome operations                                                                  | Native  | At-least-once effect handlers with stable IDs and success/failure actor operations.                                                                                                                          |
 | Actor-to-actor delivery                                                                                       | Native  | `sendTo(reference).operation()` stages delivery in the source actor commit.                                                                                                                                  |

src/configuration.ts

@@ -51,6 +51,7 @@ export interface SolidObjectsConfiguration {
   maxAttempts?: number
   maxMessagesPerActivationPass?: number
   maxActivationDurationMilliseconds?: number
+  claimScanLimit?: number
   retryDelayMilliseconds?: (attempt: number) => number
   processHeartbeatIntervalMilliseconds?: number
   processAliveThresholdMilliseconds?: number
@@ -120,6 +121,7 @@ export function buildSettings(configuration: SolidObjectsConfiguration): Runtime
     maxAttempts: configuration.maxAttempts ?? 5,
     maxMessagesPerActivationPass: configuration.maxMessagesPerActivationPass ?? 50,
     maxActivationDurationMilliseconds: configuration.maxActivationDurationMilliseconds ?? 5_000,
+    claimScanLimit: configuration.claimScanLimit ?? 100,
     retryDelayMilliseconds:
       configuration.retryDelayMilliseconds ??
       ((attempt) => Math.min(2 ** (attempt - 1), 60) * 1_000),
@@ -210,11 +212,10 @@ function validateSettings(settings: RuntimeSettings): void {
     messageRetentionMilliseconds: settings.messageRetentionMilliseconds,
     processRetentionMilliseconds: settings.processRetentionMilliseconds,
   }
-  if (
-    !Number.isSafeInteger(settings.maxMessagesPerActivationPass) ||
-    settings.maxMessagesPerActivationPass < 1
-  ) {
-    throw new TypeError("maxMessagesPerActivationPass must be a positive safe integer")
+  for (const name of ["maxMessagesPerActivationPass", "claimScanLimit"] as const) {
+    if (!Number.isSafeInteger(settings[name]) || settings[name] < 1) {
+      throw new TypeError(`${name} must be a positive safe integer`)
+    }
   }
   for (const [name, value] of Object.entries(positive)) {
     if (!Number.isFinite(value) || value <= 0) throw new TypeError(`${name} must be positive`)

src/repository.ts

@@ -320,7 +320,7 @@ export class Repository {
         if (options.instanceId !== undefined) parameters.push(options.instanceId)
         parameters.push(now)
       }
-      const candidate = await connection.get<{ message_id: string; instance_id: string }>(
+      const candidates = await connection.all<{ message_id: string; instance_id: string }>(
         `SELECT ready.message_id, ready.instance_id
          FROM ${this.table("ready_messages")} ready
          JOIN ${this.table("instances")} instances ON instances.id = ready.instance_id
@@ -335,14 +335,18 @@ export class Repository {
              WHERE earlier.instance_id = ready.instance_id AND earlier.sequence < ready.sequence
            )
            ${leaseCondition}
-         ORDER BY ready.available_at_ms, ready.sequence
-         LIMIT 1`,
-        parameters,
+         ORDER BY ready.available_at_ms, ready.sequence, ready.message_id
+         LIMIT ?`,
+        [
+          ...parameters,
+          options.activation || options.instanceId ? 1 : this.settings.claimScanLimit,
+        ],
       )
-      if (!candidate) return undefined
-
-      const token = options.activation?.activationToken ?? randomUUID()
-      if (!options.activation) {
+      for (const candidate of candidates) {
+        const token = options.activation?.activationToken ?? randomUUID()
+        if (options.activation) {
+          return this.claimCandidate({ connection, candidate, processId, token, now })
+        }
         const lease = await connection.run(
           `UPDATE ${this.table("instances")}
            SET activation_owner_id = ?, activation_token = ?, activation_generation = activation_generation + 1,
@@ -357,52 +361,63 @@ export class Repository {
             now,
           ],
         )
-        if (lease.changes !== 1) return undefined
+        if (lease.changes !== 1) continue
+        return this.claimCandidate({ connection, candidate, processId, token, now })
       }
+      return undefined
+    })
+  }
 
-      const instance = await connection.get<InstanceRow>(
-        `SELECT * FROM ${this.table("instances")} WHERE id = ?`,
-        [candidate.instance_id],
-      )
-      const message = await connection.get<MessageRow>(
-        `SELECT * FROM ${this.table("messages")} WHERE id = ?`,
-        [candidate.message_id],
-      )
-      if (!instance || !message) return undefined
+  private async claimCandidate(options: {
+    connection: DatabaseConnection
+    candidate: { message_id: string; instance_id: string }
+    processId: string
+    token: string
+    now: number
+  }): Promise<ClaimedTurn | undefined> {
+    const { connection, candidate, processId, token, now } = options
+    const instance = await connection.get<InstanceRow>(
+      `SELECT * FROM ${this.table("instances")} WHERE id = ?`,
+      [candidate.instance_id],
+    )
+    const message = await connection.get<MessageRow>(
+      `SELECT * FROM ${this.table("messages")} WHERE id = ?`,
+      [candidate.message_id],
+    )
+    if (!instance || !message) return undefined
 
-      const removed = await connection.run(
-        `DELETE FROM ${this.table("ready_messages")} WHERE message_id = ?`,
-        [message.id],
-      )
-      if (removed.changes !== 1) return undefined
-      await connection.run(
-        `INSERT INTO ${this.table("claimed_messages")}
+    const removed = await connection.run(
+      `DELETE FROM ${this.table("ready_messages")} WHERE message_id = ?`,
+      [message.id],
+    )
+    if (removed.changes !== 1) return undefined
+    await connection.run(
+      `INSERT INTO ${this.table("claimed_messages")}
          (message_id, instance_id, sequence, process_id, activation_token, activation_generation, claimed_at_ms)
          VALUES (?, ?, ?, ?, ?, ?, ?)`,
-        [
-          message.id,
-          instance.id,
-          message.sequence,
-          processId,
-          token,
-          instance.activation_generation,
-          now,
-        ],
-      )
-      await connection.run(
-        `UPDATE ${this.table("messages")} SET attempt_count = attempt_count + 1, updated_at_ms = ? WHERE id = ?`,
-        [now, message.id],
-      )
-      message.attempt_count = Number(message.attempt_count) + 1
-      return {
-        instance,
-        message,
+      [
+        message.id,
+        instance.id,
+        message.sequence,
         processId,
-        activationToken: token,
-        activationGeneration: BigInt(instance.activation_generation),
-        nowMilliseconds: now,
-      }
-    })
+        token,
+        instance.activation_generation,
+        now,
+      ],
+    )
+    await connection.run(
+      `UPDATE ${this.table("messages")} SET attempt_count = attempt_count + 1, updated_at_ms = ? WHERE id = ?`,
+      [now, message.id],
+    )
+    message.attempt_count = Number(message.attempt_count) + 1
+    return {
+      instance,
+      message,
+      processId,
+      activationToken: token,
+      activationGeneration: BigInt(instance.activation_generation),
+      nowMilliseconds: now,
+    }
   }
 
   async renewTurn(turn: ClaimedTurn): Promise<void> {

test/lifecycle.test.ts

@@ -304,6 +304,51 @@ describe("runtime lifecycle", () => {
     expect(CachedActor.deactivations).toBe(1)
   })
 
+  it("claims another actor when an earlier candidate loses its lease race", async () => {
+    runtime = configuredRuntime({ claimScanLimit: 2 })
+    await runtime.install()
+    const first = await LifecycleCounter.ref("first-candidate").send.increment()
+    const second = await LifecycleCounter.ref("second-candidate").send.increment()
+    await runtime.settings.database.connection(async (connection) => {
+      await connection.run(
+        `UPDATE ${runtime?.repository.table("ready_messages")} SET available_at_ms = 0 WHERE message_id = ?`,
+        [first.id],
+      )
+      await connection.run(
+        `UPDATE ${runtime?.repository.table("ready_messages")} SET available_at_ms = 1 WHERE message_id = ?`,
+        [second.id],
+      )
+    })
+    await runtime.repository.registerProcess("race-winner", "worker")
+    const originalRun = runtime.settings.database.transaction.bind(runtime.settings.database)
+    let leaseUpdates = 0
+    const transaction = vi
+      .spyOn(runtime.settings.database, "transaction")
+      .mockImplementation((callback) =>
+        originalRun((connection) =>
+          callback({
+            get: connection.get.bind(connection),
+            all: connection.all.bind(connection),
+            nowMilliseconds: connection.nowMilliseconds.bind(connection),
+            run: async (sql, parameters) => {
+              if (sql.includes("SET activation_owner_id")) {
+                leaseUpdates += 1
+                if (leaseUpdates === 1) return { changes: 0 }
+              }
+              return connection.run(sql, parameters)
+            },
+          }),
+        ),
+      )
+
+    const turn = await runtime.repository.claim("race-winner")
+
+    expect(turn?.message.id).toBe(second.id)
+    expect(await first.status()).toBe("ready")
+    expect(await second.status()).toBe("claimed")
+    transaction.mockRestore()
+  })
+
   it("restores cached state after a failed turn", async () => {
     runtime = configuredRuntime({ maxAttempts: 1 })
     runtime.register(CachedActor)