From cc861ce8ae0a09386e427b0ad73b0e84275e0ce1 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Tue, 6 Oct 2026 16:19:37 +0200 Subject: [PATCH] Seed leases alternate between priorities; article relay can be off Half the live leases go to higher-priority ranges, the rest to the others, so a slow recent gap runs beside older history. With no article connections the article relay answers 503 instead of waiting forever. Co-Authored-By: Claude Opus 5.5 --- app/controllers/seed_controller.ts | 3 ++- app/services/RemoteContentService.ts | 4 ++++ app/services/SeedOrchestrator.ts | 11 ++++++++++- tests/functional/seed_api.spec.ts | 14 ++++++++++++++ 4 files changed, 30 insertions(+), 2 deletions(-) diff --git a/app/controllers/seed_controller.ts b/app/controllers/seed_controller.ts index 8ab224b..ee79f61 100644 --- a/app/controllers/seed_controller.ts +++ b/app/controllers/seed_controller.ts @@ -1,7 +1,7 @@ import type { HttpContext } from '@adonisjs/core/http' import SeedOrchestrator, { LeaseLost } from '#services/SeedOrchestrator' import SeedFetchPool from '#services/SeedFetchPool' -import RemoteContentService, { JobLost } from '#services/RemoteContentService' +import RemoteContentService, { ArticleRelayOff, JobLost } from '#services/RemoteContentService' import type { SeedBatch } from '#lib/seed/wire' /** @@ -123,6 +123,7 @@ export default class SeedController { response.header('content-type', 'application/octet-stream') return response.send(body) } catch (error: any) { + if (error instanceof ArticleRelayOff) return response.serviceUnavailable({ error: error.message }) return response.badGateway({ error: `NNTP: ${error.message}` }) } } diff --git a/app/services/RemoteContentService.ts b/app/services/RemoteContentService.ts index 23b3110..02182d1 100644 --- a/app/services/RemoteContentService.ts +++ b/app/services/RemoteContentService.ts @@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto' import { Job, Worker } from 'bullmq' import logger from '@adonisjs/core/services/logger' import queueConfig from '#config/queue' +import nntpConfig from '#config/nntp' import QueueService from '#services/QueueService' import NntpService from '#services/NntpService' import ContentService, { describeContent } from '#services/ContentService' @@ -24,6 +25,7 @@ export interface RemoteJob { } export class JobLost extends Error {} +export class ArticleRelayOff extends Error {} /** * Content inspection and archive walks on seeding nodes (CONCEPT.md D41). @@ -99,6 +101,8 @@ class RemoteContentService { /** One article body for a relay node, on this process's NNTP pool. Null when the server doesn't have it. */ async article(messageId: string): Promise { + // NNTP_POOL_SIZE 0 (SEED_ARTICLE_CONNECTIONS=0): no connections for this; acquire() would wait forever. + if (nntpConfig.poolSize <= 0) throw new ArticleRelayOff('Article relay is off (SEED_ARTICLE_CONNECTIONS=0)') const conn = await NntpService.acquire() let failed = false try { diff --git a/app/services/SeedOrchestrator.ts b/app/services/SeedOrchestrator.ts index 16923cd..c9476c6 100644 --- a/app/services/SeedOrchestrator.ts +++ b/app/services/SeedOrchestrator.ts @@ -73,12 +73,21 @@ class SeedOrchestrator { WHERE status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?) AND attempts >= ?`, [LEASE_TTL_SECONDS, MAX_ATTEMPTS] ) + // Priorities take turns: while ranges of a higher priority are leased for less than half of + // the work in progress they go first, otherwise the rest does (a 2026 gap fetches ~5x slower + // than older history; LOG.md 2026-10-06 (10)). + const { rows: mix } = await db.rawQuery( + `SELECT count(*) FILTER (WHERE priority > 0) AS high, count(*) AS all FROM seed_ranges + WHERE status = 'leased' AND heartbeat_at >= now() - make_interval(secs => ?)`, + [LEASE_TTL_SECONDS] + ) + const order = Number(mix[0].high) * 2 < Number(mix[0].all) + 1 ? 'priority DESC, id' : 'priority ASC, id' const { rows } = await db.rawQuery( `UPDATE seed_ranges SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1 WHERE id = ( SELECT id FROM seed_ranges WHERE status = 'queued' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)) - ORDER BY priority DESC, id LIMIT 1 FOR UPDATE SKIP LOCKED) + ORDER BY ${order} LIMIT 1 FOR UPDATE SKIP LOCKED) RETURNING id, group_name, next_id, end_id`, [nodeId, LEASE_TTL_SECONDS] ) diff --git a/tests/functional/seed_api.spec.ts b/tests/functional/seed_api.spec.ts index ff6fd00..cd4f5c5 100644 --- a/tests/functional/seed_api.spec.ts +++ b/tests/functional/seed_api.spec.ts @@ -144,6 +144,20 @@ test.group('Seed API', (group) => { assert.include(leased.body().range, { start: 101, end: 200 }) }) + test('priorities take turns: half the leases go to the higher priority', async ({ client, assert }) => { + const token = await SeedOrchestrator.issueToken('turns-node') + const auth = { authorization: `Bearer ${token}` } + await db.table('seed_ranges').insert([ + { group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 }, + { group_name: 'alt.binaries.test', start_id: 101, end_id: 200, next_id: 101 }, + { group_name: 'alt.binaries.test', start_id: 1001, end_id: 1100, next_id: 1001, priority: 1 }, + { group_name: 'alt.binaries.test', start_id: 1101, end_id: 1200, next_id: 1101, priority: 1 }, + ]) + const starts: number[] = [] + for (let i = 0; i < 4; i++) starts.push((await client.post('/api/v1/seed/lease').headers(auth)).body().range.start) + assert.deepEqual(starts, [1001, 1, 1101, 101]) + }) + test('a node inspects a file: lease, probe result, applied here (D41)', async ({ client, assert }) => { await QueueService.collectionQueue.drain() await QueueService.archiveWalkQueue.drain()