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 <noreply@anthropic.com>
This commit is contained in:
1 parent
3cc69a617f
commit
cc861ce8ae
4 files changed
+30
-2
No files matched your search
@@ -1,7 +1,7 @@
|
|||||||
import type { HttpContext } from '@adonisjs/core/http'
|
import type { HttpContext } from '@adonisjs/core/http'
|
||||||
import SeedOrchestrator, { LeaseLost } from '#services/SeedOrchestrator'
|
import SeedOrchestrator, { LeaseLost } from '#services/SeedOrchestrator'
|
||||||
import SeedFetchPool from '#services/SeedFetchPool'
|
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'
|
import type { SeedBatch } from '#lib/seed/wire'
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -123,6 +123,7 @@ export default class SeedController {
|
|||||||
response.header('content-type', 'application/octet-stream')
|
response.header('content-type', 'application/octet-stream')
|
||||||
return response.send(body)
|
return response.send(body)
|
||||||
} catch (error: any) {
|
} catch (error: any) {
|
||||||
|
if (error instanceof ArticleRelayOff) return response.serviceUnavailable({ error: error.message })
|
||||||
return response.badGateway({ error: `NNTP: ${error.message}` })
|
return response.badGateway({ error: `NNTP: ${error.message}` })
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto'
|
|||||||
import { Job, Worker } from 'bullmq'
|
import { Job, Worker } from 'bullmq'
|
||||||
import logger from '@adonisjs/core/services/logger'
|
import logger from '@adonisjs/core/services/logger'
|
||||||
import queueConfig from '#config/queue'
|
import queueConfig from '#config/queue'
|
||||||
|
import nntpConfig from '#config/nntp'
|
||||||
import QueueService from '#services/QueueService'
|
import QueueService from '#services/QueueService'
|
||||||
import NntpService from '#services/NntpService'
|
import NntpService from '#services/NntpService'
|
||||||
import ContentService, { describeContent } from '#services/ContentService'
|
import ContentService, { describeContent } from '#services/ContentService'
|
||||||
@@ -24,6 +25,7 @@ export interface RemoteJob {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export class JobLost extends Error {}
|
export class JobLost extends Error {}
|
||||||
|
export class ArticleRelayOff extends Error {}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Content inspection and archive walks on seeding nodes (CONCEPT.md D41).
|
* 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. */
|
/** 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<Buffer | null> {
|
async article(messageId: string): Promise<Buffer | null> {
|
||||||
|
// 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()
|
const conn = await NntpService.acquire()
|
||||||
let failed = false
|
let failed = false
|
||||||
try {
|
try {
|
||||||
|
|||||||
@@ -73,12 +73,21 @@ class SeedOrchestrator {
|
|||||||
WHERE status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?) AND attempts >= ?`,
|
WHERE status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?) AND attempts >= ?`,
|
||||||
[LEASE_TTL_SECONDS, MAX_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(
|
const { rows } = await db.rawQuery(
|
||||||
`UPDATE seed_ranges SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1
|
`UPDATE seed_ranges SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1
|
||||||
WHERE id = (
|
WHERE id = (
|
||||||
SELECT id FROM seed_ranges
|
SELECT id FROM seed_ranges
|
||||||
WHERE status = 'queued' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?))
|
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`,
|
RETURNING id, group_name, next_id, end_id`,
|
||||||
[nodeId, LEASE_TTL_SECONDS]
|
[nodeId, LEASE_TTL_SECONDS]
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -144,6 +144,20 @@ test.group('Seed API', (group) => {
|
|||||||
assert.include(leased.body().range, { start: 101, end: 200 })
|
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 }) => {
|
test('a node inspects a file: lease, probe result, applied here (D41)', async ({ client, assert }) => {
|
||||||
await QueueService.collectionQueue.drain()
|
await QueueService.collectionQueue.drain()
|
||||||
await QueueService.archiveWalkQueue.drain()
|
await QueueService.archiveWalkQueue.drain()
|
||||||
|
|||||||
Reference in new issue
Block a user