diff --git a/listener/src/services/scheduled-notification-repository.ts b/listener/src/services/scheduled-notification-repository.ts index b326707e..592273e5 100644 --- a/listener/src/services/scheduled-notification-repository.ts +++ b/listener/src/services/scheduled-notification-repository.ts @@ -122,29 +122,73 @@ export class ScheduledNotificationRepository { */ async recoverStaleLocks(requestId?: string): Promise { const now = new Date(); + let recoveredCount = 0; - const sql = ` - UPDATE scheduled_notifications - SET - status = ?, - processor_id = NULL, - lock_expires_at = NULL - WHERE status = ? - AND lock_expires_at IS NOT NULL - AND lock_expires_at < ? - `; - - const result = await this.db.run(sql, [ - NotificationStatus.PENDING, - NotificationStatus.PROCESSING, - now.toISOString(), - ]); + await this.db.transaction(async () => { + const selectSql = ` + SELECT * FROM scheduled_notifications + WHERE status = ? + AND lock_expires_at IS NOT NULL + AND lock_expires_at < ? + `; + + const rows = await this.db.all(selectSql, [ + NotificationStatus.PROCESSING, + now.toISOString(), + ]); + + recoveredCount = rows.length; + + for (const row of rows) { + const model = this.rowToModel(row); + const newRetryCount = model.retryCount + 1; + const isFailed = newRetryCount >= model.maxRetries; + const newStatus = isFailed ? NotificationStatus.FAILED : NotificationStatus.PENDING; + + const updateSql = ` + UPDATE scheduled_notifications + SET + status = ?, + retry_count = ?, + last_error = ?, + error_details = ?, + processing_completed_at = ?, + processor_id = NULL, + lock_expires_at = NULL + WHERE id = ? + `; + + const errorMsg = 'Lock expired/Processor timeout'; + const errorDetails = JSON.stringify({ + message: errorMsg, + timestamp: now.toISOString(), + }); + + await this.db.run(updateSql, [ + newStatus, + newRetryCount, + errorMsg, + errorDetails, + isFailed ? now.toISOString() : null, + model.id, + ]); + + // Log execution attempt + await this.logExecution({ + scheduledNotificationId: model.id!, + executionAttempt: newRetryCount, + executionTime: now, + status: isFailed ? 'FAILED' : 'RETRY', + errorMessage: errorMsg, + }); + } + }); - if (result.changes > 0) { - logger.warn('Recovered stale locks', { requestId, count: result.changes }); + if (recoveredCount > 0) { + logger.warn('Recovered stale locks', { requestId, count: recoveredCount }); } - return result.changes; + return recoveredCount; } /** @@ -283,23 +327,28 @@ export class ScheduledNotificationRepository { failed: number; overdue: number; }> { + const now = new Date().toISOString(); + const countBySql = ` - SELECT status, COUNT(*) as count + SELECT + CASE + WHEN status = 'PROCESSING' AND lock_expires_at IS NOT NULL AND lock_expires_at < ? THEN 'PENDING' + ELSE status + END AS adjusted_status, + COUNT(*) as count FROM scheduled_notifications - GROUP BY status + GROUP BY adjusted_status `; const overdueSql = ` SELECT COUNT(*) as count FROM scheduled_notifications - WHERE status = ? AND execute_at < ? + WHERE (status = 'PENDING' OR (status = 'PROCESSING' AND lock_expires_at IS NOT NULL AND lock_expires_at < ?)) + AND execute_at < ? `; - const counts = await this.db.all<{ status: string; count: number }>(countBySql); - const overdueResult = await this.db.get<{ count: number }>(overdueSql, [ - NotificationStatus.PENDING, - new Date().toISOString(), - ]); + const counts = await this.db.all<{ adjusted_status: string; count: number }>(countBySql, [now]); + const overdueResult = await this.db.get<{ count: number }>(overdueSql, [now, now]); const stats = { pending: 0, @@ -310,7 +359,7 @@ export class ScheduledNotificationRepository { }; counts.forEach((row) => { - const status = row.status.toLowerCase(); + const status = row.adjusted_status.toLowerCase(); if (status in stats) { (stats as any)[status] = row.count; } diff --git a/listener/src/tests/notification-scheduler-refactored.test.ts b/listener/src/tests/notification-scheduler-refactored.test.ts index a06dd001..51f5897f 100644 --- a/listener/src/tests/notification-scheduler-refactored.test.ts +++ b/listener/src/tests/notification-scheduler-refactored.test.ts @@ -230,6 +230,111 @@ describe('NotificationScheduler (Refactored)', () => { expect(stats.pending).toBe(2); expect(stats.overdue).toBe(1); }); + + test('should increment retry_count and log attempt on lock recovery', async () => { + const input = NotificationFixtureBuilder + .aScheduledNotificationInput() + .forImmediateExecution() + .withMaxRetries(3) + .build(); + + const id = await repository.create(input); + + // Lock the notification + await repository.fetchAndLockPendingNotifications('processor-1', 30000, 10); + + // Manually expire the lock + const pastLock = NotificationFixtureBuilder.dates.past(1000); + await db.run('UPDATE scheduled_notifications SET lock_expires_at = ? WHERE id = ?', [ + pastLock.toISOString(), + id, + ]); + + const recovered = await repository.recoverStaleLocks(); + expect(recovered).toBe(1); + + const notification = await repository.getById(id); + expect(notification!.status).toBe(NotificationStatus.PENDING); + expect(notification!.retryCount).toBe(1); + expect(notification!.lastError).toBe('Lock expired/Processor timeout'); + + // Verify execution log + const logs = await db.all('SELECT * FROM notification_execution_log WHERE scheduled_notification_id = ?', [id]); + expect(logs.length).toBe(1); + expect(logs[0].status).toBe('RETRY'); + expect(logs[0].error_message).toBe('Lock expired/Processor timeout'); + }); + + test('should mark as failed if retry_count reaches max_retries on lock recovery', async () => { + const input = NotificationFixtureBuilder + .aScheduledNotificationInput() + .forImmediateExecution() + .withMaxRetries(1) + .build(); + + const id = await repository.create(input); + + // Lock the notification + await repository.fetchAndLockPendingNotifications('processor-1', 30000, 10); + + // Manually expire the lock + const pastLock = NotificationFixtureBuilder.dates.past(1000); + await db.run('UPDATE scheduled_notifications SET lock_expires_at = ? WHERE id = ?', [ + pastLock.toISOString(), + id, + ]); + + const recovered = await repository.recoverStaleLocks(); + expect(recovered).toBe(1); + + const notification = await repository.getById(id); + expect(notification!.status).toBe(NotificationStatus.FAILED); + expect(notification!.retryCount).toBe(1); + + // Verify execution log + const logs = await db.all('SELECT * FROM notification_execution_log WHERE scheduled_notification_id = ?', [id]); + expect(logs.length).toBe(1); + expect(logs[0].status).toBe('FAILED'); + }); + + test('should return correct statistics accounting for stale locks', async () => { + // 1. Create a notification in the future (pending, not overdue) + await repository.create( + NotificationFixtureBuilder + .aScheduledNotificationInput() + .withExecuteAt(new Date(Date.now() + 3600000)) + .build() + ); + + // 2. Create a notification in the past (overdue, pending) + await repository.create( + NotificationFixtureBuilder + .aScheduledNotificationInput() + .forImmediateExecution() + .build() + ); + + // 3. Create a notification in the past that is currently PROCESSING but lock is expired + const staleId = await repository.create( + NotificationFixtureBuilder + .aScheduledNotificationInput() + .forImmediateExecution() + .build() + ); + await repository.fetchAndLockPendingNotifications('processor-1', 30000, 10); + const pastLock = NotificationFixtureBuilder.dates.past(1000); + await db.run('UPDATE scheduled_notifications SET lock_expires_at = ? WHERE id = ?', [ + pastLock.toISOString(), + staleId, + ]); + + // Get stats BEFORE recovery + const stats = await repository.getStats(); + + expect(stats.pending).toBe(3); + expect(stats.overdue).toBe(2); + expect(stats.processing).toBe(0); + }); }); describe('NotificationAPI', () => {