diff --git a/app/services/leases.ts b/app/services/leases.ts index cd29ecc..5c34f2a 100644 --- a/app/services/leases.ts +++ b/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 { 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)) diff --git a/tests/functional/leases.spec.ts b/tests/functional/leases.spec.ts new file mode 100644 index 0000000..6779711 --- /dev/null +++ b/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) + }) +})