At most two leases per group: a group's batches wait for each other, more workers only queue

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/services/leases.ts

@@ -17,6 +17,13 @@ export class LeaseLost extends Error {}
  * prefetched (LOG.md 2026-10-09). A lease whose worker went quiet for
  * LEASE_TTL_SECONDS is taken over and resumes at its resume point.
  */
+/**
+ * A group's batches wait for each other (the batch store locks the group): more workers than this on one group only
+ * queue. Eight on one picture group's gap ranges (priority -1, in id order) held the node API's batches for minutes
+ * (LOG.md 2026-10-16 (68)).
+ */
+const MAX_LEASES_PER_GROUP = 2
+
 class Leases {
   async lease(nodeId: number): Promise<Lease | null> {
     const rules = await RuleSets.current()
@@ -29,9 +36,12 @@ class Leases {
            SELECT id FROM ranges
            WHERE ${spooled ? 'spooled_chunks > 0' : 'true'} AND next_id <= end_id ${history}
              AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))
+             AND (SELECT count(*) FROM ranges l
+                  WHERE l.status = 'leased' AND l.group_id = ranges.group_id AND l.id <> ranges.id
+                    AND l.heartbeat_at >= now() - make_interval(secs => ?)) < ${MAX_LEASES_PER_GROUP}
            ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED)
          RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`,
-        [nodeId, rules.version, coordinatorConfig.leaseTtlSeconds]
+        [nodeId, rules.version, coordinatorConfig.leaseTtlSeconds, coordinatorConfig.leaseTtlSeconds]
       )
     let { rows } = await take(true)
     if (rows.length === 0) ({ rows } = await take(false))

tests/functional/leases.spec.ts

@@ -0,0 +1,35 @@
+import { test } from '@japa/runner'
+import db from '@adonisjs/lucid/services/db'
+import testUtils from '@adonisjs/core/services/test_utils'
+import Leases from '#services/leases'
+
+test.group('Leases', (group) => {
+  group.each.setup(() => testUtils.db().withGlobalTransaction())
+
+  test('at most two ranges of a group are leased at a time; a stale lease is taken over', async ({ assert }) => {
+    const [a] = await db.table('groups').insert({ name: 'alt.binaries.lease.a', priority: 1 }).returning('id')
+    const [b] = await db.table('groups').insert({ name: 'alt.binaries.lease.b', priority: 2 }).returning('id')
+    const range = async (groupId: number, start: number) => {
+      const [row] = await db.table('ranges').insert({ group_id: groupId, start_id: start, end_id: start + 99, next_id: start, priority: -1 }).returning('id')
+      return Number(row.id)
+    }
+    // Four of group a first in id order, then one of b.
+    const a1 = await range(a.id, 1)
+    const a2 = await range(a.id, 101)
+    await range(a.id, 201)
+    await range(a.id, 301)
+    const b1 = await range(b.id, 1)
+    const node = async (name: string) => Number((await db.table('nodes').insert({ name }).returning('id'))[0].id)
+
+    const leased = async (n: number) => (await Leases.lease(await node(`lease-${n}`)))?.range.id ?? null
+    assert.equal(await leased(1), a1)
+    assert.equal(await leased(2), a2)
+    // A third of group a would only wait on the group's lock: group b's.
+    assert.equal(await leased(3), b1)
+    assert.isNull(await leased(4))
+
+    // a1's worker went quiet: its lease is taken over (a2 alone counts).
+    await db.from('ranges').where('id', a1).update({ heartbeat_at: new Date(Date.now() - 24 * 3600_000) })
+    assert.equal(await leased(5), a1)
+  })
+})