barbercollie/ReadMeABook
1
0
Fork 1
mirror of https://github.com/kikootwo/ReadMeABook.git synced 2026-06-03 21:00:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add DB pooling, throttling and monitor backoff

Browse Source 
Add connection pool params to DATABASE_URL and configure Prisma to use the pooled URL (connection_limit=20, pool_timeout=30) to reduce connection exhaustion. Introduce safeguards and throttling across processors: limit in-flight progress DB updates in direct-download, add short delays when processing RSS, retry-failed-imports, and retry-missing-torrents, and stagger scheduler triggers to avoid bursts. Implement adaptive monitor-download polling with stallCount/lastProgress and exponential backoff, and thread these fields through JobQueueService (including reduced worker concurrency for several queues). Batch audiobook enrichment queries to small parallel batches to limit DB load. Update tests to reflect new monitor payload parameters. Overall intent: reduce DB connection pool pressure and smooth load spikes during startup and heavy processing.
This commit is contained in:
kikootwo
2026-02-18 02:43:00 -05:00
parent 20798b3dc0
commit 3820b9b21d
10 changed files with 112 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/services/job-queue.service.ts
+14 -8
View File
@@ -63,6 +63,8 @@ export interface MonitorDownloadPayload extends JobPayload {
downloadHistoryId: string;
downloadClientId: string;
downloadClient: DownloadClientType;
lastProgress?: number; // Previous poll's progress (0-100) for stall detection
stallCount?: number; // Consecutive polls with no progress change (drives backoff)
}
export interface OrganizeFilesPayload extends JobPayload {
@@ -277,19 +279,19 @@ export class JobQueueService {
*/
private startProcessors(): void {
// Search indexers processor
this.queue.process('search_indexers', 3, async (job: BullJob<SearchIndexersPayload>) => {
this.queue.process('search_indexers', 2, async (job: BullJob<SearchIndexersPayload>) => {
const { processSearchIndexers } = await import('../processors/search-indexers.processor');
return await processSearchIndexers(job.data);
});
// Download torrent processor
this.queue.process('download_torrent', 3, async (job: BullJob<DownloadTorrentPayload>) => {
this.queue.process('download_torrent', 2, async (job: BullJob<DownloadTorrentPayload>) => {
const { processDownloadTorrent } = await import('../processors/download-torrent.processor');
return await processDownloadTorrent(job.data);
});
// Monitor download processor
this.queue.process('monitor_download', 5, async (job: BullJob<MonitorDownloadPayload>) => {
this.queue.process('monitor_download', 2, async (job: BullJob<MonitorDownloadPayload>) => {
const { processMonitorDownload } = await import('../processors/monitor-download.processor');
return await processMonitorDownload(job.data);
});
@@ -357,23 +359,23 @@ export class JobQueueService {
});
// Send notification processor
this.queue.process('send_notification', 5, async (job: BullJob<SendNotificationPayload>) => {
this.queue.process('send_notification', 2, async (job: BullJob<SendNotificationPayload>) => {
const { processSendNotification } = await import('../processors/send-notification.processor');
return await processSendNotification(job.data);
});
// Ebook-specific processors
this.queue.process('search_ebook', 3, async (job: BullJob<SearchEbookPayload>) => {
this.queue.process('search_ebook', 2, async (job: BullJob<SearchEbookPayload>) => {
const { processSearchEbook } = await import('../processors/search-ebook.processor');
return await processSearchEbook(job.data);
});
this.queue.process('start_direct_download', 3, async (job: BullJob<StartDirectDownloadPayload>) => {
this.queue.process('start_direct_download', 2, async (job: BullJob<StartDirectDownloadPayload>) => {
const { processStartDirectDownload } = await import('../processors/direct-download.processor');
return await processStartDirectDownload(job.data);
});
this.queue.process('monitor_direct_download', 5, async (job: BullJob<MonitorDirectDownloadPayload>) => {
this.queue.process('monitor_direct_download', 2, async (job: BullJob<MonitorDirectDownloadPayload>) => {
const { processMonitorDirectDownload } = await import('../processors/direct-download.processor');
return await processMonitorDirectDownload(job.data);
});
@@ -563,7 +565,9 @@ export class JobQueueService {
downloadHistoryId: string,
downloadClientId: string,
downloadClient: DownloadClientType,
delaySeconds: number = 0
delaySeconds: number = 0,
lastProgress?: number,
stallCount?: number
): Promise<string> {
return await this.addJob(
'monitor_download',
@@ -572,6 +576,8 @@ export class JobQueueService {
downloadHistoryId,
downloadClientId,
downloadClient,
lastProgress,
stallCount,
} as MonitorDownloadPayload,
{
priority: 5, // Medium priority