diff --git a/src/alerts.js b/src/alerts.js index d18e92f..107aa16 100755 --- a/src/alerts.js +++ b/src/alerts.js @@ -797,21 +797,34 @@ async function alertSceneReleases() { const trx = await knex.transaction(); try { - await trx('notifications') - .insert(alerts.map((alert) => ({ - user_id: alert.user_id, - scene_id: alert.scene_id, - alert_id: alert.id, - notify: alert.notify, - email: alert.email, - }))); - - await trx('alerts') + // claim the alerts before notifying, so overlapping runs (or multiple app instances) can't notify the same release twice + // a concurrent run's update waits for this transaction's row locks, then re-checks is_expired and claims nothing + const claimedAlerts = await trx('alerts') .whereIn('id', alerts.map((alert) => alert.id)) - .update('is_expired', true); + .where('is_expired', false) + .update('is_expired', true) + .returning('id'); + + const claimedAlertIds = new Set(claimedAlerts.map((alert) => alert.id)); + const claimedReleases = alerts.filter((alert) => claimedAlertIds.has(alert.id)); + + if (claimedReleases.length > 0) { + await trx('notifications') + .insert(claimedReleases.map((alert) => ({ + user_id: alert.user_id, + scene_id: alert.scene_id, + alert_id: alert.id, + notify: alert.notify, + email: alert.email, + }))); + } await trx.commit(); + if (claimedReleases.length === 0) { + return; + } + await Promise.all([ knex.schema.refreshMaterializedView('alerts_users_actors'), knex.schema.refreshMaterializedView('alerts_users_tags'), @@ -871,5 +884,6 @@ export function initAlertCron() { }, start: config.alerts.enabled, runOnInit: true, + waitForCompletion: true, // skip ticks while the previous one is still running, i.e. refreshing the materialized views }); } diff --git a/src/app.js b/src/app.js index 15cfe0d..5423647 100644 --- a/src/app.js +++ b/src/app.js @@ -1,16 +1,25 @@ +import process from 'node:process'; + import { initAlertCron } from './alerts.js'; import { initCaches } from './cache.js'; import { initStashViewsCron } from './stashes.js'; import { initSyncCron } from './sync.js'; import initServer from './web/server.js'; +// pm2 cluster mode runs the app once per instance, numbered by NODE_APP_INSTANCE (unset outside pm2) +// the server and caches are needed in every instance, but the crons must only run once or they would sync and notify twice +const isPrimaryInstance = !process.env.NODE_APP_INSTANCE || process.env.NODE_APP_INSTANCE === '0'; + async function init() { await initCaches(); initServer(); - initSyncCron(); - initAlertCron(); - initStashViewsCron(); + + if (isPrimaryInstance) { + initSyncCron(); + initAlertCron(); + initStashViewsCron(); + } } init(); diff --git a/src/sync.js b/src/sync.js index e205dac..6ae35ba 100644 --- a/src/sync.js +++ b/src/sync.js @@ -1,4 +1,3 @@ -import process from 'node:process'; import config from 'config'; import { CronJob } from 'cron'; import { format } from 'date-fns'; @@ -495,30 +494,42 @@ function getQueueItemIds(tasks, domain) { } export async function syncQueue() { - const tasks = await knex('sync'); + // claim the queued tasks in a transaction, so a concurrent run (e.g. an API-triggered sync during the cron) skips them instead of syncing and notifying them twice + // a failed sync rolls back and leaves its tasks queued for the next run + await knex.transaction(async (trx) => { + const tasks = await trx('sync') + .forUpdate() + .skipLocked(); - const sceneIds = getQueueItemIds(tasks, 'scene'); - const movieIds = getQueueItemIds(tasks, 'movie'); - const actorIds = getQueueItemIds(tasks, 'actor'); + const sceneIds = getQueueItemIds(tasks, 'scene'); + const movieIds = getQueueItemIds(tasks, 'movie'); + const actorIds = getQueueItemIds(tasks, 'actor'); - await Promise.all([ - syncScenes(sceneIds), - syncMovies(movieIds), - syncActors(actorIds), - ]); + await Promise.all([ + syncScenes(sceneIds), + syncMovies(movieIds), + syncActors(actorIds), + ]); - await knex('sync') - .whereIn('id', tasks.map((task) => task.id)) - .delete(); + await trx('sync') + .whereIn('id', tasks.map((task) => task.id)) + .delete(); - logger[process.tasks > 0 ? 'info' : 'verbose'](`Processed ${tasks.length} sync items`); + logger[tasks.length > 0 ? 'info' : 'verbose'](`Processed ${tasks.length} sync items`); + }); } export function initSyncCron() { CronJob.from({ cronTime: config.sync.crontab, async onTick() { - syncQueue(); + try { + await syncQueue(); + } + catch (error) { + // unhandled, the rejection would crash the process + logger.error(`Failed to process sync queue: ${error.message}`); + } }, start: config.sync.enabled, runOnInit: true,