diff --git a/CHANGELOG.md b/CHANGELOG.md index 573c1629..9810cc35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ * issue#710: Fixing Typo in thold_daemons.service File * issue#714: Increase the Name column to 255 characters * issue#719: Plugin Disabled due to mix of string and int +* issue#784: Retry failed queued email notifications with bounded exponential backoff * issue#812: Recover stale notification claims, scope worker drains, and deprecate the ignored notification --thread option * issue: All Columns checkd on Thresholds page * issue: Special character previous value handling broken on data query indexes with special characters diff --git a/INFO b/INFO index b6c2c483..7099aeb2 100644 --- a/INFO +++ b/INFO @@ -21,7 +21,7 @@ [info] name = thold -version = 1.8.2 +version = 1.8.3 longname = Thresholds author = The Cacti Group email = diff --git a/README.md b/README.md index 6def6001..4f355898 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,12 @@ and become familiar with its settings. From there, you can provide overall control of thold, and set defaults for things like Email bodies, weekend exemptions, alert log retention, logging, etc. +When the notification queue is enabled, transient email failures are retried +up to five times with exponential backoff from one to eight minutes. The +Notification Queue page shows the attempt count and next eligible retry time. +After the fifth failed attempt the row becomes a terminal error so a permanent +SMTP or address failure cannot retry forever. + Notification workers claim queue rows with their process ID and drain only that claim. Unfinished rows are released when a worker stops or notifications are suspended, while claims left by a hard-killed worker are recovered after diff --git a/includes/database.php b/includes/database.php index 279aff65..6512e6f4 100644 --- a/includes/database.php +++ b/includes/database.php @@ -1755,6 +1755,27 @@ function thold_upgrade_database($force = false) { db_execute('UPDATE plugin_notification_lists SET enabled = "on"'); } + if (cacti_version_compare($oldv, '1.8.3', '<')) { + db_add_column('notification_queue', [ + 'name' => 'attempt_count', + 'type' => 'int', + 'unsigned' => true, + 'NULL' => false, + 'default' => '0', + 'after' => 'error_message'] + ); + + db_add_column('notification_queue', [ + 'name' => 'next_attempt', + 'type' => 'timestamp', + 'NULL' => true, + 'default' => null, + 'after' => 'attempt_count'] + ); + + db_add_index('notification_queue', 'INDEX', 'retry_ready', ['event_processed', 'process_id', 'next_attempt']); + } + db_add_column('thold_data', [ 'name' => 'external_id', 'type' => 'varchar(20)', @@ -2123,12 +2144,15 @@ function thold_setup_database() { $data['columns'][] = ['name' => 'event_data', 'type' => 'longblob', 'NULL' => false, 'default' => '']; $data['columns'][] = ['name' => 'error_code', 'type' => 'int', 'NULL' => false, 'default' => '0']; $data['columns'][] = ['name' => 'error_message', 'type' => 'varchar(128)', 'NULL' => false, 'default' => '']; + $data['columns'][] = ['name' => 'attempt_count', 'type' => 'int', 'unsigned' => true, 'NULL' => false, 'default' => '0']; + $data['columns'][] = ['name' => 'next_attempt', 'type' => 'timestamp', 'NULL' => true, 'default' => null]; $data['columns'][] = ['name' => 'process_id', 'type' => 'int', 'unsigned' => true, 'NULL' => false, 'default' => '0']; $data['columns'][] = ['name' => 'event_processed', 'type' => 'tinyint', 'unsigned' => true, 'NULL' => false, 'default' => '0']; $data['columns'][] = ['name' => 'event_processed_time', 'type' => 'timestamp', 'NULL' => false, 'default' => '0000-00-00']; $data['columns'][] = ['name' => 'event_processed_runtime', 'type' => 'double', 'unsigned' => true, 'NULL' => false, 'default' => '0']; $data['primary'] = 'id'; $data['keys'][] = ['name' => 'topic_processed', 'columns' => 'topic`, `event_processed']; + $data['keys'][] = ['name' => 'retry_ready', 'columns' => 'event_processed`, `process_id`, `next_attempt']; $data['keys'][] = ['name' => 'process_id', 'columns' => 'process_id']; $data['keys'][] = ['name' => 'object_id', 'columns' => 'object_id']; $data['keys'][] = ['name' => 'host_id', 'columns' => 'host_id']; diff --git a/notify_queue.php b/notify_queue.php index 12f24645..c288a213 100644 --- a/notify_queue.php +++ b/notify_queue.php @@ -463,6 +463,18 @@ function clearFilter() { 'sort' => 'DESC', 'tip' => __('Did this notification result in an error. Hover on the error column for details.', 'thold') ], + 'attempt_count' => [ + 'display' => __('Attempts', 'thold'), + 'align' => 'right', + 'sort' => 'DESC', + 'tip' => __('The number of delivery attempts made for this notification.', 'thold') + ], + 'next_attempt' => [ + 'display' => __('Next Attempt', 'thold'), + 'align' => 'right', + 'sort' => 'DESC', + 'tip' => __('When a failed notification is eligible for its next retry.', 'thold') + ], 'event_processed_runtime' => [ 'display' => __('Run Time', 'thold'), 'align' => 'right', @@ -491,14 +503,9 @@ function clearFilter() { form_selectable_cell($n['id'], $n['id'], '', 'right'); form_selectable_cell($n['event_time'], $n['id'], '', 'right'); - form_selectable_cell($n['event_processed'] == 0 ? __('Pending', 'thold') : __('Done', 'thold'), $n['id'], '', 'right'); - if ($n['event_processed'] > 0) { - form_selectable_cell($n['error_code'] > 0 ? __('Errored', 'thold') : __('Success', 'thold'), $n['id'], '', 'right'); - form_selectable_cell(number_format_i18n($n['event_processed_runtime'], 2), $n['id'], '', 'right'); - } else { - form_selectable_cell(__('N/A', 'thold'), $n['id'], '', 'right'); - form_selectable_cell(__('N/A', 'thold'), $n['id'], '', 'right'); + foreach (thold_notification_queue_status_cells($n) as $cell) { + form_selectable_cell($cell, $n['id'], '', 'right'); } form_checkbox_cell($n['object_name'], $n['id']); diff --git a/tests/Unit/NotificationQueueClaimTest.php b/tests/Unit/NotificationQueueClaimTest.php index fcd40148..75a639f1 100644 --- a/tests/Unit/NotificationQueueClaimTest.php +++ b/tests/Unit/NotificationQueueClaimTest.php @@ -130,6 +130,7 @@ public function testAClaimRecoversOrphansThenTakesOnlyUnheldRows(): void { $this->assertStringContainsString('LEFT JOIN processes', $calls[0]['sql']); $this->assertStringContainsString('p.pid IS NULL', $calls[0]['sql']); $this->assertSame(['thold_notify', 'child'], $calls[0]['params']); + $this->assertStringContainsString('(next_attempt IS NULL OR next_attempt <= NOW())', $calls[1]['sql']); $this->assertStringContainsString('AND process_id = 0', $calls[1]['sql']); $this->assertSame([4242], $calls[1]['params']); } @@ -681,7 +682,9 @@ public function testDeviceCommandAndGroupedMailComplete(): void { putenv('THOLD_DEVICE_TEST'); $updates = array_values(array_filter(CactiStubs::$calls, static function ($call) { - return $call['fn'] === 'db_execute_prepared' && strpos($call['sql'], 'event_processed = 1') !== false; + return $call['fn'] === 'db_execute_prepared' + && (strpos($call['sql'], 'event_processed = 1') !== false + || strpos($call['sql'], 'attempt_count = CASE id') !== false); })); $this->assertCount(2, $updates); @@ -690,7 +693,7 @@ public function testDeviceCommandAndGroupedMailComplete(): void { $this->assertSame(128, strlen($updates[0]['params'][1])); $this->assertSame(1, $updates[1]['params'][0]); $this->assertSame(128, mb_strlen($updates[1]['params'][1], 'UTF-8')); - $this->assertSame(77, $updates[1]['params'][3]); + $this->assertSame([102, 77], array_slice($updates[1]['params'], -2)); $this->assertStringContainsString('AND process_id = ?', $updates[0]['sql']); $this->assertStringContainsString('AND process_id = ?', $updates[1]['sql']); $this->assertSame(3, $heartbeats); @@ -723,8 +726,7 @@ public function testIndividualDeviceMailCompletionRequiresItsOwner(): void { $this->assertStringContainsString('AND process_id = ?', $call['sql']); $this->assertSame(128, mb_strlen($call['params'][1], 'UTF-8')); - $this->assertSame(104, $call['params'][3]); - $this->assertSame(77, $call['params'][4]); + $this->assertSame([104, 77], array_slice($call['params'], -2)); } /** @@ -780,8 +782,7 @@ public function testNonDeviceMailCompletionRequiresItsOwner(): void { $this->assertStringContainsString('AND process_id = ?', $call['sql']); $this->assertSame(128, mb_strlen($call['params'][1], 'UTF-8')); - $this->assertSame(105, $call['params'][3]); - $this->assertSame(77, $call['params'][4]); + $this->assertSame([105, 77], array_slice($call['params'], -2)); } /** diff --git a/tests/Unit/NotificationQueueRetryTest.php b/tests/Unit/NotificationQueueRetryTest.php new file mode 100644 index 00000000..6d2ee421 --- /dev/null +++ b/tests/Unit/NotificationQueueRetryTest.php @@ -0,0 +1,304 @@ + + */ + private function lastPreparedCall() { + $calls = array_values(array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared'; + })); + + $this->assertNotEmpty($calls); + + return end($calls); + } + + /** + * @return array + * @param mixed $id + * @param mixed $topic + * @param mixed $attempts + */ + private function mailRow($id, $topic = 'thold_mail', $attempts = 0) { + return [ + 'id' => $id, + 'topic' => $topic, + 'attempt_count' => $attempts, + 'event_data' => json_encode([ + 'from' => ['sender@example.com'], + 'to' => 'recipient@example.com', + 'cc' => '', + 'bcc' => '', + 'replyto' => '', + 'subject' => 'Threshold alert', + 'body' => 'Alert', + 'body_text' => 'Alert', + 'attachments' => [], + 'headers' => [], + 'html' => true, + ]), + ]; + } + + /** + * @return void + */ + public function testRetryDelayUsesBoundedExponentialBackoff(): void { + $this->assertSame(60, thold_notification_retry_delay(-1)); + $this->assertSame(60, thold_notification_retry_delay(0)); + $this->assertSame(60, thold_notification_retry_delay(1)); + $this->assertSame(120, thold_notification_retry_delay(2)); + $this->assertSame(480, thold_notification_retry_delay(4)); + $this->assertSame(1920, thold_notification_retry_delay(6)); + $this->assertSame(3600, thold_notification_retry_delay(7)); + } + + /** + * @return void + */ + public function testQueueStatusCellsFollowTheirHeaderOrder(): void { + $cells = thold_notification_queue_status_cells([ + 'event_processed' => 1, + 'error_code' => 1, + 'attempt_count' => 4, + 'next_attempt' => null, + 'event_processed_runtime' => 0.25, + ]); + + $this->assertSame( + ['event_processed', 'error_code', 'attempt_count', 'next_attempt', 'event_processed_runtime'], + array_keys($cells) + ); + $this->assertSame(['Done', 'Errored', 4, 'N/A', '0.25'], array_values($cells)); + + $pending = thold_notification_queue_status_cells(['event_processed' => 0]); + $this->assertSame(['Pending', 'N/A', 0, 'N/A', 'N/A'], array_values($pending)); + } + + /** + * @return void + */ + public function testSuccessfulDeliveryIsTerminal(): void { + thold_notification_record_delivery(42, 77, '', 0.25, 2); + + $call = $this->lastPreparedCall(); + $sql = preg_replace('/\s+/', ' ', $call['sql']); + + $this->assertSame(0, $call['params'][0]); + $this->assertSame('', $call['params'][1]); + $this->assertStringContainsString('next_attempt = CASE id', $sql); + $this->assertStringContainsString('THEN NULL', $sql); + $this->assertSame([42, 3], array_slice($call['params'], 2, 2)); + $this->assertSame(1, $call['params'][7]); + $this->assertSame([42, 0.25, 42, 77], array_slice($call['params'], -4)); + $this->assertStringContainsString('AND process_id = ?', $sql); + } + + /** + * @return void + */ + public function testTransientFailureReleasesTheClaimAndSchedulesRetry(): void { + thold_notification_record_delivery(42, 77, "smtp\ndown", 0.5); + + $call = $this->lastPreparedCall(); + $sql = preg_replace('/\s+/', ' ', $call['sql']); + + $this->assertStringContainsString('THEN FROM_UNIXTIME', $sql); + $this->assertStringContainsString('process_id = CASE id', $sql); + $this->assertStringContainsString('THEN 0', $sql); + $this->assertSame([1, 'smtp down', 42, 1, 42, 60], array_slice($call['params'], 0, 6)); + $this->assertSame(0, $call['params'][8]); + $this->assertSame([42, 0.5, 42, 77], array_slice($call['params'], -4)); + } + + /** + * @return void + */ + public function testFifthFailureIsTerminal(): void { + thold_notification_record_delivery(42, 77, 'permanent failure', 0.5, 4); + + $call = $this->lastPreparedCall(); + $sql = preg_replace('/\s+/', ' ', $call['sql']); + + $this->assertStringContainsString('THEN NULL', $sql); + $this->assertSame([1, 'permanent failure', 42, 5], array_slice($call['params'], 0, 4)); + $this->assertSame(1, $call['params'][7]); + $this->assertSame([42, 0.5, 42, 77], array_slice($call['params'], -4)); + } + + /** + * @return void + */ + public function testFailureMessageFitsTheQueueColumn(): void { + thold_notification_record_delivery(42, 77, str_repeat('x', 200), 0.5); + + $call = $this->lastPreparedCall(); + + $this->assertSame(128, strlen($call['params'][1])); + } + + /** + * @return void + */ + public function testClaimAndBothDrainsIgnoreRetriesThatAreNotReady(): void { + thold_notification_claim(77); + + $claims = array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared' && + strpos($call['sql'], 'SET process_id = ?') !== false; + }); + + $this->assertCount(1, $claims); + $claim = reset($claims); + $this->assertStringContainsString('(next_attempt IS NULL OR next_attempt <= NOW())', $claim['sql']); + + thold_notification_execute(77); + + $queries = array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_fetch_assoc_prepared' && + strpos($call['sql'], 'notification_queue') !== false; + }); + + $this->assertCount(2, $queries); + + foreach ($queries as $call) { + $this->assertStringContainsString('(next_attempt IS NULL OR next_attempt <= NOW())', $call['sql']); + } + } + + /** + * @return void + */ + public function testIndividualDeviceMailFailureUsesTheRetryRecorder(): void { + CactiStubs::$configOptions['alert_deadnotify_one_mail'] = ''; + CactiStubs::willReturnFor('db_fetch_assoc_prepared', "topic IN ('thold_dhost_mail'", [ + $this->mailRow(51, 'thold_dhost_mail', 2), + ]); + CactiStubs::willReturn('mailer', 'temporary SMTP failure'); + + process_device_notifications(77, 'all', 0); + + $call = $this->lastPreparedCall(); + + $this->assertSame([1, 'temporary SMTP failure', 51, 3, 51, 240], array_slice($call['params'], 0, 6)); + $this->assertSame([51, $call['params'][10], 51, 77], array_slice($call['params'], -4)); + $this->assertSame(0, $call['params'][8]); + $this->assertStringContainsString('process_id = CASE id', $call['sql']); + } + + /** + * @return void + */ + public function testGroupedDeviceMailRecordsEveryAttempt(): void { + CactiStubs::$configOptions['alert_deadnotify_one_mail'] = 'on'; + CactiStubs::$configOptions['alert_deadnotify_subject'] = 'Device alerts'; + CactiStubs::willReturnFor('db_fetch_assoc_prepared', "topic IN ('thold_dhost_mail'", [ + $this->mailRow(61, 'thold_dhost_mail', 0), + $this->mailRow(63, 'thold_dhost_mail', 0), + $this->mailRow(62, 'thold_uhost_mail', 3), + $this->mailRow(64, 'thold_uhost_mail', 4), + ]); + CactiStubs::willReturn('mailer', 'temporary SMTP failure'); + CactiStubs::willReturn('db_affected_rows', 4); + + process_device_notifications(77, 'all', 0); + + $calls = array_values(array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared' && strpos($call['sql'], 'attempt_count') !== false; + })); + + $this->assertCount(1, $calls); + $this->assertStringContainsString('attempt_count = CASE id', $calls[0]['sql']); + $this->assertStringContainsString('event_processed = CASE id', $calls[0]['sql']); + $this->assertSame([61, 1, 63, 1, 62, 4, 64, 5], array_slice($calls[0]['params'], 2, 8)); + $this->assertSame([61, 60, 63, 60, 62, 480, 64], array_slice($calls[0]['params'], 10, 7)); + $this->assertSame([61, 63, 62, 64], array_slice($calls[0]['params'], 17, 4)); + $this->assertSame([61, 0, 63, 0, 62, 0, 64, 1], array_slice($calls[0]['params'], 21, 8)); + $this->assertSame([61, 63, 62, 64], array_slice($calls[0]['params'], 29, 4)); + $this->assertSame([61, 63, 62, 64, 77], array_slice($calls[0]['params'], -5)); + $this->assertSame('temporary SMTP failure', $calls[0]['params'][1]); + } + + /** + * @return void + */ + public function testGroupedDeliveryHandlesEmptyInvalidAndSuccessfulBatches(): void { + $this->assertTrue(thold_notification_record_deliveries([], 77, '', 0.25)); + $this->assertTrue(thold_notification_record_deliveries([-1 => 0], 77, '', 0.25)); + $this->assertSame([], CactiStubs::$calls); + + CactiStubs::willReturn('db_affected_rows', 2); + $this->assertTrue(thold_notification_record_deliveries([81 => 0, 82 => 4], 77, '', 0.25)); + + $calls = array_values(array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared' && strpos($call['sql'], 'attempt_count = CASE id') !== false; + })); + + $this->assertCount(1, $calls); + $this->assertSame(0, $calls[0]['params'][0]); + $this->assertSame('', $calls[0]['params'][1]); + $this->assertSame([81, 82, 77], array_slice($calls[0]['params'], -3)); + $this->assertStringNotContainsString('FROM_UNIXTIME', $calls[0]['sql']); + $this->assertStringContainsString('process_id = CASE id', $calls[0]['sql']); + $this->assertStringContainsString('THEN process_id', $calls[0]['sql']); + $this->assertStringContainsString('THEN NOW()', $calls[0]['sql']); + $this->assertSame([81, 1, 82, 1], array_slice($calls[0]['params'], 10, 4)); + } + + /** + * @return void + */ + public function testGroupedTerminalFailuresStayClaimedAndComplete(): void { + CactiStubs::willReturn('db_affected_rows', 2); + $this->assertTrue(thold_notification_record_deliveries([91 => 4, 92 => 5], 77, 'permanent failure', 0.5)); + + $call = $this->lastPreparedCall(); + + $this->assertSame([91, 5, 92, 6], array_slice($call['params'], 2, 4)); + $this->assertSame([91, 1, 92, 1], array_slice($call['params'], 10, 4)); + $this->assertStringNotContainsString('FROM_UNIXTIME', $call['sql']); + $this->assertStringContainsString('THEN process_id', $call['sql']); + $this->assertStringContainsString('THEN NOW()', $call['sql']); + } + + /** + * @return void + */ + public function testNonDeviceMailFailureUsesTheRetryRecorder(): void { + CactiStubs::willReturnFor('db_fetch_assoc_prepared', "topic NOT IN ('thold_dhost_mail'", [ + $this->mailRow(71, 'thold_mail', 1), + ]); + CactiStubs::willReturn('mailer', 'temporary SMTP failure'); + + process_non_device_notifications(77, 'all', 0); + + $call = $this->lastPreparedCall(); + + $this->assertSame([1, 'temporary SMTP failure', 71, 2, 71, 120], array_slice($call['params'], 0, 6)); + $this->assertSame([71, $call['params'][10], 71, 77], array_slice($call['params'], -4)); + $this->assertSame(0, $call['params'][8]); + $this->assertStringContainsString('process_id = CASE id', $call['sql']); + } +} diff --git a/tests/bin/patch-coverage.php b/tests/bin/patch-coverage.php index 8b49345b..07d45250 100644 --- a/tests/bin/patch-coverage.php +++ b/tests/bin/patch-coverage.php @@ -157,6 +157,11 @@ function changed_lines($base_ref) { * here with reviewable justification. */ $unmeasured_allowlist = [ + // Database migration/schema declarations require a live Cacti database. + 'includes/database.php', + // Authenticated web entry point; its status mapping lives in the covered + // thold_notification_queue_status_cells() helper. + 'notify_queue.php', 'thold_notify.php', ]; $unmeasured = array_values(array_diff(array_keys($changed), array_keys($measured))); diff --git a/thold_functions.php b/thold_functions.php index 7213a827..affe9dc7 100644 --- a/thold_functions.php +++ b/thold_functions.php @@ -7584,6 +7584,7 @@ function thold_notification_claim($pid) { db_execute_prepared('UPDATE notification_queue SET process_id = ? WHERE event_processed = 0 + AND (next_attempt IS NULL OR next_attempt <= NOW()) AND process_id = 0', [$pid]); @@ -7894,6 +7895,145 @@ function thold_notification_execute($pid = 0, $max_records = 'all', $heartbeat = process_device_notifications($pid, $max_records, $prev_suspended, $heartbeat); } +function thold_notification_retry_delay($attempt) { + $attempt = max(1, (int) $attempt); + + return min(3600, 60 * (2 ** ($attempt - 1))); +} + +/** + * Values for the delivery-status columns, in the same order as their headers. + * + * @param array $notification + * + * @return array + */ +function thold_notification_queue_status_cells(array $notification) { + $processed = (int) ($notification['event_processed'] ?? 0); + + return [ + 'event_processed' => $processed === 0 ? __('Pending', 'thold') : __('Done', 'thold'), + 'error_code' => $processed === 0 ? __('N/A', 'thold') : ((int) ($notification['error_code'] ?? 0) > 0 ? __('Errored', 'thold') : __('Success', 'thold')), + 'attempt_count' => (int) ($notification['attempt_count'] ?? 0), + 'next_attempt' => !empty($notification['next_attempt']) ? $notification['next_attempt'] : __('N/A', 'thold'), + 'event_processed_runtime' => $processed === 0 ? __('N/A', 'thold') : number_format_i18n($notification['event_processed_runtime'] ?? 0, 2), + ]; +} + +/** + * Record one queued email delivery without losing transient failures. + * + * The fifth failed attempt is terminal. Earlier failures release the claim and + * schedule a bounded exponential retry, so a permanent SMTP error cannot spin + * every poller cycle forever. + * + * @param mixed $id + * @param int $pid + * @param mixed $error + * @param mixed $runtime + * @param mixed $previous_attempts + */ +function thold_notification_record_delivery($id, $pid, $error, $runtime, $previous_attempts = 0) { + return thold_notification_record_deliveries([(int) $id => $previous_attempts], $pid, $error, $runtime); +} + +/** + * Record one grouped mail result with a single prepared update. + * + * @param array $records Record ID => previous attempt count. + * @param int $pid Owning notification worker. + * @param string $error + * @param float $runtime + * + * @return bool + */ +function thold_notification_record_deliveries(array $records, $pid, $error, $runtime) { + $pid = (int) $pid; + + if ($pid <= 0 || !cacti_sizeof($records)) { + return true; + } + + $error = thold_notification_error_message($error); + $attempt_cases = []; + $attempt_params = []; + $next_cases = []; + $next_params = []; + $process_cases = []; + $process_params = []; + $done_cases = []; + $done_params = []; + $time_cases = []; + $time_params = []; + $ids = []; + + foreach ($records as $id => $previous_attempts) { + $id = (int) $id; + + if ($id <= 0) { + continue; + } + + $attempt = max(0, (int) $previous_attempts) + 1; + $retryable = $error !== '' && $attempt < 5; + $done = $retryable ? 0 : 1; + + $attempt_cases[] = 'WHEN ? THEN ?'; + $attempt_params[] = $id; + $attempt_params[] = $attempt; + + if ($retryable) { + $next_cases[] = 'WHEN ? THEN FROM_UNIXTIME(UNIX_TIMESTAMP() + ?)'; + $next_params[] = $id; + $next_params[] = thold_notification_retry_delay($attempt); + $process_cases[] = 'WHEN ? THEN 0'; + } else { + $next_cases[] = 'WHEN ? THEN NULL'; + $next_params[] = $id; + $process_cases[] = 'WHEN ? THEN process_id'; + } + + $process_params[] = $id; + $done_cases[] = 'WHEN ? THEN ?'; + $done_params[] = $id; + $done_params[] = $done; + $time_cases[] = $done ? 'WHEN ? THEN NOW()' : 'WHEN ? THEN event_processed_time'; + $time_params[] = $id; + $ids[] = $id; + } + + if (!cacti_sizeof($ids)) { + return true; + } + + $placeholders = implode(',', array_fill(0, cacti_sizeof($ids), '?')); + $params = array_merge( + [$error === '' ? 0 : 1, $error], + $attempt_params, + $next_params, + $process_params, + $done_params, + $time_params, + [$runtime], + $ids, + [$pid] + ); + + return thold_notification_complete('UPDATE notification_queue + SET error_code = ?, error_message = ?, + attempt_count = CASE id ' . implode(' ', $attempt_cases) . ' ELSE attempt_count END, + next_attempt = CASE id ' . implode(' ', $next_cases) . ' ELSE next_attempt END, + process_id = CASE id ' . implode(' ', $process_cases) . ' ELSE process_id END, + event_processed = CASE id ' . implode(' ', $done_cases) . ' ELSE event_processed END, + event_processed_time = CASE id ' . implode(' ', $time_cases) . ' ELSE event_processed_time END, + event_processed_runtime = ? + WHERE id IN (' . $placeholders . ') + AND process_id = ?', + $params, + $ids, + $pid); +} + function process_device_notifications($pid, $max_records, $prev_suspended, $heartbeat = null) { $one_email = read_config_option('alert_deadnotify_one_mail') == 'on' ? true : false; $emails = []; @@ -7911,6 +8051,7 @@ function process_device_notifications($pid, $max_records, $prev_suspended, $hear $records = db_fetch_assoc_prepared("SELECT * FROM notification_queue WHERE event_processed = 0 + AND (next_attempt IS NULL OR next_attempt <= NOW()) AND topic IN ('thold_dhost_mail', 'thold_uhost_mail', 'thold_dhost_cmd', 'thold_uhost_cmd') AND process_id = ? ORDER BY event_time ASC @@ -7976,22 +8117,11 @@ function process_device_notifications($pid, $max_records, $prev_suspended, $hear if ($error != '') { cacti_log("ERROR: Sending Email Failed To:$to Subject:$subject. Error was:'$error'", true, 'THOLD'); - - $any_error = $error; - $error_code = 1; - } else { - $error_code = 0; } $nend = microtime(true); - thold_notification_complete('UPDATE notification_queue - SET error_code = ?, error_message = ?, event_processed = 1, event_processed_time=NOW(), event_processed_runtime = ? - WHERE id = ? - AND process_id = ?', - [$error_code, thold_notification_error_message($error), $nend - $nstart, $r['id'], $pid], - [$r['id']], - $pid); + thold_notification_record_delivery($r['id'], $pid, $error, $nend - $nstart, $r['attempt_count'] ?? 0); } else { $id = md5(json_encode([$from, $to, $cc, $bcc, $replyto])); @@ -8025,7 +8155,7 @@ function process_device_notifications($pid, $max_records, $prev_suspended, $hear } } - $emails[$id]['ids'][] = $r['id']; + $emails[$id]['records'][$r['id']] = $r['attempt_count'] ?? 0; } break; @@ -8110,24 +8240,11 @@ function process_device_notifications($pid, $max_records, $prev_suspended, $hear if ($error != '') { cacti_log("ERROR: Sending Email Failed To:$to Subject:$subject. Error was:'$error'", true, 'THOLD'); - - $any_error = $error; - $error_code = 1; - } else { - $error_code = 0; } $nend = microtime(true); - $ids = implode(', ', array_map('intval', $email['ids'])); - - thold_notification_complete("UPDATE notification_queue - SET error_code = ?, error_message = ?, event_processed = 1, event_processed_time=NOW(), event_processed_runtime = ? - WHERE id IN ($ids) - AND process_id = ?", - [$error_code, thold_notification_error_message($error), $nend - $nstart, $pid], - $email['ids'], - $pid); + thold_notification_record_deliveries($email['records'], $pid, $error, $nend - $nstart); } } } else { @@ -8145,6 +8262,7 @@ function process_non_device_notifications($pid, $max_records, $prev_suspended, $ $records = db_fetch_assoc_prepared("SELECT * FROM notification_queue WHERE event_processed = 0 + AND (next_attempt IS NULL OR next_attempt <= NOW()) AND topic NOT IN ('thold_dhost_mail', 'thold_uhost_mail', 'thold_dhost_cmd', 'thold_uhost_cmd') AND process_id = ? ORDER BY event_time ASC @@ -8199,22 +8317,11 @@ function process_non_device_notifications($pid, $max_records, $prev_suspended, $ if ($error != '') { cacti_log("ERROR: Sending Email Failed To:$to Subject:$subject. Error was:'$error'", true, 'THOLD'); - - $any_error = $error; - $error_code = 1; - } else { - $error_code = 0; } $nend = microtime(true); - thold_notification_complete('UPDATE notification_queue - SET error_code = ?, error_message = ?, event_processed = 1, event_processed_time=NOW(), event_processed_runtime = ? - WHERE id = ? - AND process_id = ?', - [$error_code, thold_notification_error_message($error), $nend - $nstart, $r['id'], $pid], - [$r['id']], - $pid); + thold_notification_record_delivery($r['id'], $pid, $error, $nend - $nstart, $r['attempt_count'] ?? 0); break; case 'thold_cmd':