usenet-indexer/app/services/leases.ts

import db from '@adonisjs/lucid/services/db'
import coordinatorConfig from '#config/coordinator'
import RuleSets from '#services/rule_sets'
import Spool from '#services/spool'
import Corpus from '#services/corpus'
import CatalogSpace from '#services/catalog_space'
import type { Lease } from '#lib/edge/index'

export class LeaseLost extends Error {}

/**
 * Ranges for workers (REBUILD.md, coordinator): the next range with chunks
 * spooled, by priority (oldest era first), else the next planned one (the
 * prefetcher fetches leased ranges first; the worker's stream waits). A
 * worker never waits for prefetching to choose its range: refused leases
 * left workers idle for 30 s, so fewer ranges were leased and fewer
 * 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()
    // The catalog at its cap on the NVMe: the tail and the stored ranges only (CatalogSpace).
    const history = (await CatalogSpace.full()) ? 'AND priority < 0' : ''
    const take = (spooled: boolean) =>
      db.rawQuery(
        `UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
         WHERE r.id = (
           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,
          coordinatorConfig.leaseTtlSeconds,
        ]
      )
    let { rows } = await take(true)
    if (rows.length === 0) ({ rows } = await take(false))
    const row = rows[0]
    if (!row) return null
    await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
    const group = await db.from('groups').where('id', row.group_id).select('name').firstOrFail()
    return {
      range: {
        id: Number(row.id),
        group: group.name,
        start: Number(row.start_id),
        end: Number(row.end_id),
        next: Number(row.next_id),
      },
      rules,
    }
  }

  /**
   * Ranges read to their end whose worker went quiet before finishing them: done. The lease query skips a range past
   * its end, so one whose finish never came stayed leased for good (a gap range since 2026-10-16 14:03, during the lock
   * pile-up; LOG.md 2026-10-17 (72)). Its batches are all stored; its spooled chunks go.
   */
  async finishAbandoned(): Promise<number> {
    const { rows } = await db.rawQuery(
      `UPDATE ranges SET status = 'done', done_at = now()
       WHERE status = 'leased' AND next_id > end_id AND heartbeat_at < now() - make_interval(secs => ?)
       RETURNING id`,
      [coordinatorConfig.leaseTtlSeconds]
    )
    for (const { id } of rows) await Spool.remove(Number(id))
    return rows.length
  }

  async leased(rangeId: number, nodeId: number) {
    const range = await db.from('ranges').where('id', rangeId).first()
    if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId)
      throw new LeaseLost(`Range ${rangeId} is not leased to this node`)
    return range
  }

  async heartbeat(rangeId: number, nodeId: number) {
    await this.leased(rangeId, nodeId)
    await db.from('ranges').where('id', rangeId).update({ heartbeat_at: new Date() })
    await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
  }

  /** The worker reached the range's end: done, and its spooled chunks go. */
  async finish(rangeId: number, nodeId: number) {
    const range = await this.leased(rangeId, nodeId)
    if (Number(range.next_id) <= Number(range.end_id))
      throw new LeaseLost(
        `Range ${rangeId} is at ${range.next_id}, not past its end ${range.end_id}`
      )
    await db
      .from('ranges')
      .where('id', rangeId)
      .update({ status: 'done', done_at: new Date(), heartbeat_at: new Date() })
    await Spool.remove(rangeId)
  }

  /**
   * A worker got a chunk that doesn't inflate: the spooled chunk goes and
   * the range back to the queue (the prefetcher fetches leased ranges'
   * missing chunks again; the next worker resumes at the range's resume point).
   */
  async corrupt(rangeId: number, nodeId: number, from: number) {
    const range = await this.leased(rangeId, nodeId)
    const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find(
      (s) => s.from === from
    )
    if (span) {
      await Spool.removeChunk(rangeId, span)
      if (range.capture) {
        const group = await db.from('groups').where('id', range.group_id).select('name').first()
        await Corpus.removeChunk(
          { group: group.name, start: Number(range.start_id), end: Number(range.end_id) },
          span
        )
        // Not complete in the corpus any more: the corpus fill fetches the chunk again.
        await db.from('ranges').where('id', rangeId).update({ stored_at: null })
      }
    }
    const plain: number[] = (range.plain_chunks ?? []).map(Number)
    // Reported once already: the compressed copy may have been in flight (or the prefetcher's targets not
    // refreshed yet) and spooled again (range 161820, LOG.md 2026-10-11), so a second try plain first.
    if (plain.filter((p) => p === from).length >= 2) {
      // Fetched plain (deflated here) and still not inflating: not the server's compression; set aside.
      await db
        .from('ranges')
        .where('id', rangeId)
        .update({
          status: 'failed',
          node_id: null,
          error: `chunk ${from} doesn't inflate, also fetched plain`,
        })
      return
    }
    // The server's compressed reply is cut short (the same bytes every time): fetched plain next time.
    await db.rawQuery(
      `UPDATE ranges SET status = 'planned', node_id = NULL, plain_chunks = array_append(plain_chunks, ?::bigint) WHERE id = ?`,
      [from, rangeId]
    )
  }

  /** A stopping worker gives its range back; the next one resumes it. */
  async release(rangeId: number, nodeId: number) {
    await this.leased(rangeId, nodeId)
    await db.from('ranges').where('id', rangeId).update({ status: 'planned', node_id: null })
  }
}

export default new Leases()