diff --git a/commands/IndexScheduler.ts b/commands/IndexScheduler.ts index 14acb82..a7d46a0 100644 --- a/commands/IndexScheduler.ts +++ b/commands/IndexScheduler.ts @@ -117,11 +117,18 @@ export default class IndexScheduler extends BaseCommand { } } - // Run immediately, then wait a full interval after each run so runs never overlap. + // Run immediately, then wait a full interval after each run so runs never overlap. A failed + // round (e.g. Postgres restarting) is logged and the next one tries again: an error that escaped + // here once ended the loop while the process stayed up, and live indexing stopped for ~3 hours + // (LOG.md 2026-10-07 (3)). while (!Shutdown.requested) { - await schedule() - await backfill() - await split() + for (const step of [schedule, backfill, split]) { + try { + await step() + } catch (err: any) { + this.logger.error(`Scheduler round failed, retrying next round: ${err.message}`) + } + } await Shutdown.sleep(indexerConfig.intervalMs) } await this.pool.shutdown() diff --git a/commands/workers/Deferred.ts b/commands/workers/Deferred.ts index 008c2ca..ed90891 100644 --- a/commands/workers/Deferred.ts +++ b/commands/workers/Deferred.ts @@ -31,23 +31,29 @@ export default class DeferredWorker extends BaseCommand { let done = 0 try { while (!Shutdown.requested && (!this.limit || done < this.limit)) { - const batch = await DeferredService.claim(Math.min(concurrency * 4, this.limit ? this.limit - done : Infinity)) - if (batch.length === 0) { - if (this.limit) break - await Shutdown.sleep(IDLE_MS) - continue + try { + const batch = await DeferredService.claim(Math.min(concurrency * 4, this.limit ? this.limit - done : Infinity)) + if (batch.length === 0) { + if (this.limit) break + await Shutdown.sleep(IDLE_MS) + continue + } + let next = 0 + await Promise.all( + Array.from({ length: concurrency }, async () => { + while (next < batch.length) { + const status = await DeferredService.process(batch[next++]) + counts[status] = (counts[status] ?? 0) + 1 + } + }) + ) + done += batch.length + this.logger.info(`${done} deferred files processed: ${Object.entries(counts).map(([k, v]) => `${v} ${k}`).join(', ')}.`) + } catch (err: any) { + // E.g. Postgres restarting: wait and claim again, instead of ending the loop (LOG.md 2026-10-07 (3)). + this.logger.error(`Round failed, retrying in 30 s: ${err.message}`) + await Shutdown.sleep(30_000) } - let next = 0 - await Promise.all( - Array.from({ length: concurrency }, async () => { - while (next < batch.length) { - const status = await DeferredService.process(batch[next++]) - counts[status] = (counts[status] ?? 0) + 1 - } - }) - ) - done += batch.length - this.logger.info(`${done} deferred files processed: ${Object.entries(counts).map(([k, v]) => `${v} ${k}`).join(', ')}.`) } } finally { await NntpService.shutdown() diff --git a/commands/workers/Spots.ts b/commands/workers/Spots.ts index 3614325..7738d2c 100644 --- a/commands/workers/Spots.ts +++ b/commands/workers/Spots.ts @@ -33,28 +33,34 @@ export default class SpotsWorker extends BaseCommand { let done = 0 try { while (!Shutdown.requested && (!this.limit || done < this.limit)) { - const batch = await SpotService.claim(Math.min(BATCH, this.limit ? this.limit - done : Infinity)) - if (batch.length === 0) { - if (this.limit) break - // Idle: read the details of imported spots that don't have them yet (newest first). - if ((await SpotService.fillDetails(DETAILS_BATCH)) === 0) await Shutdown.sleep(IDLE_MS) - continue + try { + const batch = await SpotService.claim(Math.min(BATCH, this.limit ? this.limit - done : Infinity)) + if (batch.length === 0) { + if (this.limit) break + // Idle: read the details of imported spots that don't have them yet (newest first). + if ((await SpotService.fillDetails(DETAILS_BATCH)) === 0) await Shutdown.sleep(IDLE_MS) + continue + } + // Old, verified spots are listed in bulk; the rest need their header (and maybe NZB) read. + const listable = batch.filter((spot) => SpotService.isListable(spot)) + const single = batch.filter((spot) => !SpotService.isListable(spot)) + counts.listed = (counts.listed ?? 0) + (await SpotService.list(listable)) + let next = 0 + await Promise.all( + Array.from({ length: concurrency }, async () => { + while (next < single.length) { + const status = await SpotService.process(single[next++]) + counts[status] = (counts[status] ?? 0) + 1 + } + }) + ) + done += batch.length + this.logger.info(`${done} spots: ${Object.entries(counts).map(([k, v]) => `${v} ${k}`).join(', ')}.`) + } catch (err: any) { + // E.g. Postgres restarting: wait and claim again, instead of ending the loop (LOG.md 2026-10-07 (3)). + this.logger.error(`Round failed, retrying in 30 s: ${err.message}`) + await Shutdown.sleep(30_000) } - // Old, verified spots are listed in bulk; the rest need their header (and maybe NZB) read. - const listable = batch.filter((spot) => SpotService.isListable(spot)) - const single = batch.filter((spot) => !SpotService.isListable(spot)) - counts.listed = (counts.listed ?? 0) + (await SpotService.list(listable)) - let next = 0 - await Promise.all( - Array.from({ length: concurrency }, async () => { - while (next < single.length) { - const status = await SpotService.process(single[next++]) - counts[status] = (counts[status] ?? 0) + 1 - } - }) - ) - done += batch.length - this.logger.info(`${done} spots: ${Object.entries(counts).map(([k, v]) => `${v} ${k}`).join(', ')}.`) } } finally { await NntpService.shutdown()