Scheduler and worker loops survive a database restart
index-scheduler's loop ended when Postgres restarted (the active-groups query was outside its try), leaving live indexing stopped for ~3 hours while the process stayed up. Each step's error is now caught and retried next round; worker:spots and worker:deferred wait 30 s and claim again. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
61192690a7
commit
e3e3ecc93d
3 files changed
+60
-41
No files matched your search
@@ -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) {
|
while (!Shutdown.requested) {
|
||||||
await schedule()
|
for (const step of [schedule, backfill, split]) {
|
||||||
await backfill()
|
try {
|
||||||
await split()
|
await step()
|
||||||
|
} catch (err: any) {
|
||||||
|
this.logger.error(`Scheduler round failed, retrying next round: ${err.message}`)
|
||||||
|
}
|
||||||
|
}
|
||||||
await Shutdown.sleep(indexerConfig.intervalMs)
|
await Shutdown.sleep(indexerConfig.intervalMs)
|
||||||
}
|
}
|
||||||
await this.pool.shutdown()
|
await this.pool.shutdown()
|
||||||
|
|||||||
@@ -31,23 +31,29 @@ export default class DeferredWorker extends BaseCommand {
|
|||||||
let done = 0
|
let done = 0
|
||||||
try {
|
try {
|
||||||
while (!Shutdown.requested && (!this.limit || done < this.limit)) {
|
while (!Shutdown.requested && (!this.limit || done < this.limit)) {
|
||||||
const batch = await DeferredService.claim(Math.min(concurrency * 4, this.limit ? this.limit - done : Infinity))
|
try {
|
||||||
if (batch.length === 0) {
|
const batch = await DeferredService.claim(Math.min(concurrency * 4, this.limit ? this.limit - done : Infinity))
|
||||||
if (this.limit) break
|
if (batch.length === 0) {
|
||||||
await Shutdown.sleep(IDLE_MS)
|
if (this.limit) break
|
||||||
continue
|
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 {
|
} finally {
|
||||||
await NntpService.shutdown()
|
await NntpService.shutdown()
|
||||||
|
|||||||
+27
-21
@@ -33,28 +33,34 @@ export default class SpotsWorker extends BaseCommand {
|
|||||||
let done = 0
|
let done = 0
|
||||||
try {
|
try {
|
||||||
while (!Shutdown.requested && (!this.limit || done < this.limit)) {
|
while (!Shutdown.requested && (!this.limit || done < this.limit)) {
|
||||||
const batch = await SpotService.claim(Math.min(BATCH, this.limit ? this.limit - done : Infinity))
|
try {
|
||||||
if (batch.length === 0) {
|
const batch = await SpotService.claim(Math.min(BATCH, this.limit ? this.limit - done : Infinity))
|
||||||
if (this.limit) break
|
if (batch.length === 0) {
|
||||||
// Idle: read the details of imported spots that don't have them yet (newest first).
|
if (this.limit) break
|
||||||
if ((await SpotService.fillDetails(DETAILS_BATCH)) === 0) await Shutdown.sleep(IDLE_MS)
|
// Idle: read the details of imported spots that don't have them yet (newest first).
|
||||||
continue
|
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 {
|
} finally {
|
||||||
await NntpService.shutdown()
|
await NntpService.shutdown()
|
||||||
|
|||||||
Reference in new issue
Block a user