Fixed alerts handled by all cluster processes, resulting in double notifications.

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
 	});
 }

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();

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,