mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-10 02:25:56 +00:00
fix notification terminal failure recovery
This commit is contained in:
rcourtman
@@ -600,6 +600,12 @@ Common reporting error codes:
|
||||
- `GET /api/notifications/dlq` (admin)
|
||||
- `POST /api/notifications/dlq/retry` (admin)
|
||||
- `POST /api/notifications/dlq/delete` (admin)
|
||||
- `POST /api/notifications/terminal-failures/retry` (admin, `settings:write`)
|
||||
- Returns every retained `failed` or `dlq` delivery to `pending` with a fresh
|
||||
retry budget. Existing per-attempt delivery history is preserved.
|
||||
- `POST /api/notifications/terminal-failures/dismiss` (admin, `settings:write`)
|
||||
- Marks every retained terminal delivery `cancelled`, clearing the active
|
||||
queue-health warning without deleting delivery history.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -151,6 +151,10 @@ first step before platform-local cleanup.
|
||||
- Open **Alerts → Notifications** first. Pulse shows a delivery warning when
|
||||
failed or dead-lettered notifications remain in the persistent queue; a
|
||||
missing queue-health read is shown as unavailable rather than healthy.
|
||||
- After correcting the destination, use **Retry retained deliveries**. Use
|
||||
**Dismiss retained failures** only when those deliveries should not be sent.
|
||||
Both actions preserve delivery history; do not delete `notification_queue.db`
|
||||
to clear the warning.
|
||||
- Check SMTP settings in **Alerts → Notifications**.
|
||||
- Check logs: `docker logs pulse | grep email`.
|
||||
- Ensure your SMTP provider allows the connection (e.g., Gmail App Passwords).
|
||||
@@ -159,6 +163,9 @@ first step before platform-local cleanup.
|
||||
- Check the delivery warning in **Alerts → Notifications** and use **Send test**
|
||||
after correcting the destination. Recoverable retries do not trigger the
|
||||
warning; retained terminal failures do.
|
||||
- If the test succeeds, use **Retry retained deliveries** to give the retained
|
||||
items a fresh retry budget. Dismiss them only when delivery is no longer
|
||||
wanted; neither action deletes the audit trail.
|
||||
- Verify the URL is reachable from the Pulse server.
|
||||
- If targeting private IPs, allow them in **Settings → System → Network → Webhook Security**.
|
||||
- Check Pulse logs for HTTP status codes and response bodies.
|
||||
|
||||
@@ -192,6 +192,10 @@ Docker/Podman module attached to that host identity. The removal must delete
|
||||
the associated runtime report surfaces and clear their active alerts instead
|
||||
of leaving container alerts or inventory detached from a removed agent.
|
||||
|
||||
Notification terminal-failure retry and dismissal are delivery-queue operator
|
||||
actions only. They do not renew agent enrollment, heartbeat, command-channel
|
||||
readiness, update observation, or any agent-lifecycle lease.
|
||||
|
||||
Proxmox node and backup-server setup may request an explicitly insecure
|
||||
Unified Agent installer command. The choice must remain off by default, be
|
||||
visible at the point where the command is generated, and affect both the
|
||||
|
||||
@@ -258,6 +258,11 @@ frontend-primitives: row, mobile-card, global-default, and bulk-selection
|
||||
icon-only actions must compose `ActionIconButton` for shared size, tone, focus,
|
||||
title, and accessible-name behavior instead of rendering local `<button>` plus
|
||||
inline SVG shells in alert-owned runtime components.
|
||||
The notification-delivery health card is an alerts-owned presentation over
|
||||
notification-owned queue truth. When retained terminal failures exist, the
|
||||
Destinations surface exposes explicit retry and dismiss actions with
|
||||
consequence confirmations, refreshes health and delivery history after either
|
||||
action, and never instructs the operator to delete queue storage.
|
||||
Alert runtime state has one explicit ownership boundary: `AlertConfig.enabled`
|
||||
controls detector evaluation and in-product alert visibility, while
|
||||
`AlertConfig.activationState` controls external notification delivery only.
|
||||
|
||||
@@ -72,6 +72,12 @@ The same live update applies all four `schedule.grouping` fields atomically.
|
||||
When grouping is disabled, pending and subsequent alerts are delivered
|
||||
individually; the API boundary must not reduce that setting to the window or
|
||||
grouping-key fields while silently ignoring `enabled`.
|
||||
`POST /api/notifications/terminal-failures/retry` and
|
||||
`POST /api/notifications/terminal-failures/dismiss` are admin,
|
||||
`settings:write` operator actions. Their strict response is
|
||||
`{"success": true, "affected": <non-negative integer>}`; frontend clients
|
||||
must normalize malformed counts to zero and must not consume the raw DLQ
|
||||
payload, which can contain notification configuration and content.
|
||||
`DELETE /api/ai/patrol/suppressions/finding_{findingID}` is the canonical
|
||||
reopen path for a dismissed Patrol finding. It removes the finding-backed
|
||||
suppression row, preserves the operator note, clears dismissal state in both
|
||||
|
||||
@@ -22,6 +22,10 @@
|
||||
|
||||
Own reusable frontend primitives and canonical page-shell patterns so feature
|
||||
work extends shared components instead of creating new local variants.
|
||||
Feature-owned warning cards, including notification delivery health, compose
|
||||
the shared `Button` variants for retry, dismiss, refresh, loading, and disabled
|
||||
states. Feature code owns the action copy and confirmation consequences, but
|
||||
must not recreate local button chrome for those controls.
|
||||
The alert schedule's initial-delivery selector composes `SettingsPanel` and
|
||||
`FormSelect`, uses the shared alert-configuration presentation vocabulary, and
|
||||
exposes the same email, webhook, Apprise, and all-destination labels used by
|
||||
|
||||
@@ -331,6 +331,15 @@ counts cannot be read. Pending retries do not degrade health. The response
|
||||
must expose fixed reason codes and retention metadata rather than raw queue
|
||||
errors or notification content.
|
||||
|
||||
Operator recovery is owned by the queue, not by database-file deletion. The
|
||||
settings-write retry action returns all retained `failed` and `dlq` rows to
|
||||
`pending`, resets their queue-attempt counters to a fresh retry budget, keeps
|
||||
their operational links, and wakes the processor; it never rewrites existing
|
||||
per-attempt audit rows. The settings-write dismiss action transitions those
|
||||
same rows to `cancelled`, preserves both queue and audit history, and clears
|
||||
the active health warning. Both actions are transactional across the selected
|
||||
terminal set and report the number of rows actually transitioned.
|
||||
|
||||
### User-facing delivery log and honest test sends
|
||||
|
||||
The queue owner exposes its retained per-attempt audit rows to the local
|
||||
|
||||
@@ -201,6 +201,9 @@ active-alert read model only while detector evaluation is enabled. They must
|
||||
not hide storage or recovery alerts when external notification delivery is
|
||||
pending, paused, snoozed, unavailable, or destination-disabled; those delivery
|
||||
states do not change the underlying storage evidence or active lifecycle.
|
||||
Likewise, retrying or dismissing retained notification failures changes only
|
||||
delivery-queue state and its warning. It does not alter storage health,
|
||||
protection posture, recovery-point evidence, or restore readiness.
|
||||
|
||||
## Extension Points
|
||||
|
||||
|
||||
@@ -1,17 +1,24 @@
|
||||
{
|
||||
"version": 1,
|
||||
"base_sha": "af6d4825152580779c21a48bc13a702c15a4a982",
|
||||
"verified_at": "2026-08-21T05:48:24Z",
|
||||
"base_sha": "434e1448ff70464bccec3e6ec46257d10da2162f",
|
||||
"verified_at": "2026-08-23T08:04:16Z",
|
||||
"result": "passed",
|
||||
"changed_paths": [
|
||||
"frontend-modern/src/components/Settings/useInfrastructureOperationsState.tsx"
|
||||
"frontend-modern/src/api/notifications.ts",
|
||||
"frontend-modern/src/features/alerts/AlertDeliveryHealthCard.tsx",
|
||||
"frontend-modern/src/features/alerts/tabs/DestinationsTab.tsx",
|
||||
"frontend-modern/src/features/alerts/useAlertDestinationsTabState.ts",
|
||||
"frontend-modern/src/utils/alertDestinationsPresentation.ts"
|
||||
],
|
||||
"content_sha256": {
|
||||
"frontend-modern/src/components/Settings/useInfrastructureOperationsState.tsx": "afcb6b556a0bb21e19b492744a62f97513206c27f3e8d643a2c8862e2e43fe13"
|
||||
"frontend-modern/src/api/notifications.ts": "df77323e526d5c528bd4fb7fcfdccb70b5823c5b0dd41525718067ffd441b0b5",
|
||||
"frontend-modern/src/features/alerts/AlertDeliveryHealthCard.tsx": "0e9c53191da45f5c81f336c1034bc0e0651d88f92fc89b268d9a997dcd7d1b20",
|
||||
"frontend-modern/src/features/alerts/tabs/DestinationsTab.tsx": "23cc5d7800028b35019359af2b46937b5fef308bdf82116240fa92c21e16177c",
|
||||
"frontend-modern/src/features/alerts/useAlertDestinationsTabState.ts": "6993829996d1acd877f310212f045f33ee9f92b56cb1fee62e293cf4c9b8f14e",
|
||||
"frontend-modern/src/utils/alertDestinationsPresentation.ts": "52bf726d4b5a081db84c1be05bce466231749c73c0f2b43acce4cec11b7ff80b"
|
||||
},
|
||||
"routes": [
|
||||
"/settings/infrastructure",
|
||||
"/settings/infrastructure (Agent Doctor view via Open Agent Doctor)"
|
||||
"/alerts/notifications"
|
||||
],
|
||||
"viewports": [
|
||||
{
|
||||
@@ -24,16 +31,15 @@
|
||||
}
|
||||
],
|
||||
"states": [
|
||||
"Agent Doctor fleet table with 1 Critical and 12 Healthy targets (mock stack against live dev backend)",
|
||||
"Expanded Critical row prod-euw1-k8s-03 showing stale-report diagnostic, no credential repair offered",
|
||||
"Expanded Healthy row apollo-114 showing no fleet-health issues",
|
||||
"Doctor table re-rendered at mobile width after reload",
|
||||
"No repair_authentication target exists in the mock fleet, so the replaceCredential command branch was verified by unit tests (InfrastructureOperationsModel.test.tsx source pin plus agentInstallCommand builder tests), not by a rendered command"
|
||||
"Healthy queue with no retained terminal deliveries: recovery actions correctly absent and delivery activity empty",
|
||||
"Notifications-paused warning remains distinct from queue health and explains that test sends bypass the activation gate",
|
||||
"Desktop and narrow layouts render the delivery activity card without horizontal overflow or console errors",
|
||||
"Retained-failure warning, retry/dismiss controls, confirmation consequences, loading states, API refresh, and success/error paths verified by focused component, state, API, and queue tests because the live development queue had no terminal failures"
|
||||
],
|
||||
"interactions": [
|
||||
"Clicked Open Agent Doctor from Infrastructure settings",
|
||||
"Expanded the Critical target row prod-euw1-k8s-03",
|
||||
"Expanded the Healthy target row apollo-114",
|
||||
"Reloaded the doctor view at 375x812 and confirmed clean render with zero console errors"
|
||||
"Navigated from Alerts overview to Notifications",
|
||||
"Refreshed healthy delivery status on the desktop layout",
|
||||
"Reloaded /alerts/notifications at 1280x800 and 375x812",
|
||||
"Confirmed Recent delivery activity and Refresh delivery status remain visible and usable at narrow width"
|
||||
]
|
||||
}
|
||||
|
||||
@@ -600,6 +600,12 @@ Common reporting error codes:
|
||||
- `GET /api/notifications/dlq` (admin)
|
||||
- `POST /api/notifications/dlq/retry` (admin)
|
||||
- `POST /api/notifications/dlq/delete` (admin)
|
||||
- `POST /api/notifications/terminal-failures/retry` (admin, `settings:write`)
|
||||
- Returns every retained `failed` or `dlq` delivery to `pending` with a fresh
|
||||
retry budget. Existing per-attempt delivery history is preserved.
|
||||
- `POST /api/notifications/terminal-failures/dismiss` (admin, `settings:write`)
|
||||
- Marks every retained terminal delivery `cancelled`, clearing the active
|
||||
queue-health warning without deleting delivery history.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -151,6 +151,10 @@ first step before platform-local cleanup.
|
||||
- Open **Alerts → Notifications** first. Pulse shows a delivery warning when
|
||||
failed or dead-lettered notifications remain in the persistent queue; a
|
||||
missing queue-health read is shown as unavailable rather than healthy.
|
||||
- After correcting the destination, use **Retry retained deliveries**. Use
|
||||
**Dismiss retained failures** only when those deliveries should not be sent.
|
||||
Both actions preserve delivery history; do not delete `notification_queue.db`
|
||||
to clear the warning.
|
||||
- Check SMTP settings in **Alerts → Notifications**.
|
||||
- Check logs: `docker logs pulse | grep email`.
|
||||
- Ensure your SMTP provider allows the connection (e.g., Gmail App Passwords).
|
||||
@@ -159,6 +163,9 @@ first step before platform-local cleanup.
|
||||
- Check the delivery warning in **Alerts → Notifications** and use **Send test**
|
||||
after correcting the destination. Recoverable retries do not trigger the
|
||||
warning; retained terminal failures do.
|
||||
- If the test succeeds, use **Retry retained deliveries** to give the retained
|
||||
items a fresh retry budget. Dismiss them only when delivery is no longer
|
||||
wanted; neither action deletes the audit trail.
|
||||
- Verify the URL is reachable from the Pulse server.
|
||||
- If targeting private IPs, allow them in **Settings → System → Network → Webhook Security**.
|
||||
- Check Pulse logs for HTTP status codes and response bodies.
|
||||
|
||||
@@ -373,6 +373,28 @@ describe('NotificationsAPI', () => {
|
||||
expect(log).toEqual({ entries: [], windowDays: 7 });
|
||||
});
|
||||
|
||||
it('retries retained terminal deliveries through the operator recovery endpoint', async () => {
|
||||
apiFetchJSONMock.mockResolvedValueOnce({ affected: 3, success: true } as any);
|
||||
|
||||
const result = await NotificationsAPI.retryTerminalFailures();
|
||||
|
||||
expect(apiFetchJSONMock).toHaveBeenCalledWith('/api/notifications/terminal-failures/retry', {
|
||||
method: 'POST',
|
||||
});
|
||||
expect(result).toEqual({ affected: 3, success: true });
|
||||
});
|
||||
|
||||
it('dismisses retained terminal failures and fails closed on malformed counts', async () => {
|
||||
apiFetchJSONMock.mockResolvedValueOnce({ affected: -1, success: 'yes' } as any);
|
||||
|
||||
const result = await NotificationsAPI.dismissTerminalFailures();
|
||||
|
||||
expect(apiFetchJSONMock).toHaveBeenCalledWith('/api/notifications/terminal-failures/dismiss', {
|
||||
method: 'POST',
|
||||
});
|
||||
expect(result).toEqual({ affected: 0, success: false });
|
||||
});
|
||||
|
||||
it('passes the deliveryPaused flag through from test-send responses', async () => {
|
||||
apiFetchJSONMock.mockResolvedValueOnce({
|
||||
status: 'success',
|
||||
|
||||
@@ -129,6 +129,11 @@ export interface NotificationHealth {
|
||||
queue: NotificationQueueHealth;
|
||||
}
|
||||
|
||||
export interface NotificationTerminalFailureActionResult {
|
||||
success: boolean;
|
||||
affected: number;
|
||||
}
|
||||
|
||||
export type NotificationDeliveryOutcome = 'sent' | 'retry' | 'failed' | 'dead_letter' | 'cancelled';
|
||||
|
||||
export interface NotificationDeliveryLogEntry {
|
||||
@@ -352,6 +357,28 @@ export class NotificationsAPI {
|
||||
};
|
||||
}
|
||||
|
||||
static async retryTerminalFailures(): Promise<NotificationTerminalFailureActionResult> {
|
||||
const payload = await apiFetchJSON<Record<string, unknown>>(
|
||||
`${this.baseUrl}/terminal-failures/retry`,
|
||||
{ method: 'POST' },
|
||||
);
|
||||
return {
|
||||
success: strictBoolean(payload.success),
|
||||
affected: nonNegativeCount(payload.affected) ?? 0,
|
||||
};
|
||||
}
|
||||
|
||||
static async dismissTerminalFailures(): Promise<NotificationTerminalFailureActionResult> {
|
||||
const payload = await apiFetchJSON<Record<string, unknown>>(
|
||||
`${this.baseUrl}/terminal-failures/dismiss`,
|
||||
{ method: 'POST' },
|
||||
);
|
||||
return {
|
||||
success: strictBoolean(payload.success),
|
||||
affected: nonNegativeCount(payload.affected) ?? 0,
|
||||
};
|
||||
}
|
||||
|
||||
static async updateAppriseConfig(config: AppriseConfig): Promise<AppriseConfig> {
|
||||
return apiFetchJSON(`${this.baseUrl}/apprise`, {
|
||||
method: 'PUT',
|
||||
|
||||
@@ -38,12 +38,16 @@ describe('AlertDeliveryHealthCard', () => {
|
||||
|
||||
it('tells operators that retained terminal deliveries were not delivered', () => {
|
||||
const onRefresh = vi.fn();
|
||||
const onRetryFailures = vi.fn();
|
||||
const onDismissFailures = vi.fn();
|
||||
render(() => (
|
||||
<AlertDeliveryHealthCard
|
||||
health={degradedHealth}
|
||||
unavailable={false}
|
||||
refreshing={false}
|
||||
onRefresh={onRefresh}
|
||||
onRetryFailures={onRetryFailures}
|
||||
onDismissFailures={onDismissFailures}
|
||||
/>
|
||||
));
|
||||
|
||||
@@ -62,6 +66,12 @@ describe('AlertDeliveryHealthCard', () => {
|
||||
|
||||
fireEvent.click(screen.getByRole('button', { name: 'Refresh delivery status' }));
|
||||
expect(onRefresh).toHaveBeenCalledTimes(1);
|
||||
|
||||
fireEvent.click(screen.getByRole('button', { name: 'Retry retained deliveries' }));
|
||||
expect(onRetryFailures).toHaveBeenCalledTimes(1);
|
||||
|
||||
fireEvent.click(screen.getByRole('button', { name: 'Dismiss retained failures' }));
|
||||
expect(onDismissFailures).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('fails closed when queue health cannot be verified', () => {
|
||||
|
||||
@@ -2,11 +2,14 @@ import AlertTriangleIcon from 'lucide-solid/icons/alert-triangle';
|
||||
import RefreshCwIcon from 'lucide-solid/icons/refresh-cw';
|
||||
|
||||
import type { NotificationQueueHealth } from '@/api/notifications';
|
||||
import { Button } from '@/components/shared/Button';
|
||||
import { Card } from '@/components/shared/Card';
|
||||
import {
|
||||
getAlertDestinationsDeliveryDismissLabel,
|
||||
getAlertDestinationsDeliveryHealthDescription,
|
||||
getAlertDestinationsDeliveryHealthTitle,
|
||||
getAlertDestinationsDeliveryRefreshLabel,
|
||||
getAlertDestinationsDeliveryRetryLabel,
|
||||
} from '@/utils/alertDestinationsPresentation';
|
||||
|
||||
interface AlertDeliveryHealthCardProps {
|
||||
@@ -14,6 +17,10 @@ interface AlertDeliveryHealthCardProps {
|
||||
unavailable: boolean;
|
||||
refreshing: boolean;
|
||||
onRefresh: () => void;
|
||||
retryingFailures?: boolean;
|
||||
dismissingFailures?: boolean;
|
||||
onRetryFailures?: () => void;
|
||||
onDismissFailures?: () => void;
|
||||
}
|
||||
|
||||
export function AlertDeliveryHealthCard(props: AlertDeliveryHealthCardProps) {
|
||||
@@ -46,15 +53,39 @@ export function AlertDeliveryHealthCard(props: AlertDeliveryHealthCardProps) {
|
||||
</p>
|
||||
</div>
|
||||
</div>
|
||||
<button
|
||||
type="button"
|
||||
class="inline-flex flex-shrink-0 items-center justify-center gap-2 rounded-md border border-red-300 bg-transparent px-3 py-1.5 text-sm font-medium text-red-800 transition hover:bg-red-100 disabled:cursor-not-allowed disabled:opacity-50 dark:border-red-700 dark:text-red-200 dark:hover:bg-red-900/30"
|
||||
disabled={props.refreshing}
|
||||
onClick={props.onRefresh}
|
||||
>
|
||||
<RefreshCwIcon class={`h-4 w-4 ${props.refreshing ? 'animate-spin' : ''}`} />
|
||||
{getAlertDestinationsDeliveryRefreshLabel()}
|
||||
</button>
|
||||
<div class="flex flex-shrink-0 flex-wrap items-center gap-2">
|
||||
{props.onRetryFailures ? (
|
||||
<Button
|
||||
variant="secondary"
|
||||
size="sm"
|
||||
isLoading={props.retryingFailures}
|
||||
disabled={props.dismissingFailures}
|
||||
onClick={props.onRetryFailures}
|
||||
>
|
||||
{getAlertDestinationsDeliveryRetryLabel()}
|
||||
</Button>
|
||||
) : null}
|
||||
{props.onDismissFailures ? (
|
||||
<Button
|
||||
variant="danger"
|
||||
size="sm"
|
||||
isLoading={props.dismissingFailures}
|
||||
disabled={props.retryingFailures}
|
||||
onClick={props.onDismissFailures}
|
||||
>
|
||||
{getAlertDestinationsDeliveryDismissLabel()}
|
||||
</Button>
|
||||
) : null}
|
||||
<Button
|
||||
variant="ghost"
|
||||
size="sm"
|
||||
disabled={props.refreshing || props.retryingFailures || props.dismissingFailures}
|
||||
onClick={props.onRefresh}
|
||||
>
|
||||
<RefreshCwIcon class={`mr-2 h-4 w-4 ${props.refreshing ? 'animate-spin' : ''}`} />
|
||||
{getAlertDestinationsDeliveryRefreshLabel()}
|
||||
</Button>
|
||||
</div>
|
||||
</div>
|
||||
</Card>
|
||||
);
|
||||
|
||||
@@ -16,6 +16,8 @@ vi.mock('@/api/notifications', () => ({
|
||||
getDeliveryLog: vi.fn(),
|
||||
getHealth: vi.fn(),
|
||||
getWebhooks: vi.fn(),
|
||||
dismissTerminalFailures: vi.fn(),
|
||||
retryTerminalFailures: vi.fn(),
|
||||
testNotification: vi.fn(),
|
||||
testWebhook: vi.fn(),
|
||||
updateWebhook: vi.fn(),
|
||||
@@ -77,6 +79,8 @@ describe('useAlertDestinationsTabState', () => {
|
||||
vi.mocked(NotificationsAPI.getDeliveryLog).mockReset();
|
||||
vi.mocked(NotificationsAPI.getHealth).mockReset();
|
||||
vi.mocked(NotificationsAPI.getWebhooks).mockReset();
|
||||
vi.mocked(NotificationsAPI.dismissTerminalFailures).mockReset();
|
||||
vi.mocked(NotificationsAPI.retryTerminalFailures).mockReset();
|
||||
vi.mocked(NotificationsAPI.testNotification).mockReset();
|
||||
vi.mocked(NotificationsAPI.testWebhook).mockReset();
|
||||
vi.mocked(NotificationsAPI.updateWebhook).mockReset();
|
||||
@@ -248,4 +252,82 @@ describe('useAlertDestinationsTabState', () => {
|
||||
expect(notificationStore.warning).toHaveBeenCalledTimes(2);
|
||||
expect(notificationStore.success).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('confirms and resolves retained terminal deliveries without deleting delivery history', async () => {
|
||||
const [emailConfig] = createSignal(buildEmailConfig());
|
||||
const [appriseConfig, setAppriseConfig] = createSignal(buildAppriseConfig());
|
||||
const [configLoadError] = createSignal<string | null>(null);
|
||||
const [isRetrying] = createSignal(false);
|
||||
const [isLoadingDestinations] = createSignal(false);
|
||||
const confirmSpy = vi.spyOn(globalThis, 'confirm').mockReturnValue(true);
|
||||
|
||||
vi.mocked(NotificationsAPI.getWebhooks).mockResolvedValue([]);
|
||||
vi.mocked(NotificationsAPI.getHealth).mockResolvedValue({
|
||||
overallHealthy: false,
|
||||
queue: {
|
||||
attentionRequired: 3,
|
||||
completedRetentionDays: 7,
|
||||
countsAreRetentionBounded: true,
|
||||
deadLetter: 2,
|
||||
deadLetterRetentionDays: 30,
|
||||
failed: 1,
|
||||
failureClasses7d: {
|
||||
authentication: 3,
|
||||
configuration: 0,
|
||||
connectivity: 0,
|
||||
rate_limited: 0,
|
||||
rejected: 0,
|
||||
tls: 0,
|
||||
unknown: 0,
|
||||
},
|
||||
failureClassesAvailable: true,
|
||||
failureClassWindowDays: 7,
|
||||
healthy: false,
|
||||
pending: 0,
|
||||
reasonCodes: ['retained_failed_deliveries', 'retained_dead_letter_deliveries'],
|
||||
retryAttemptsAffectHealth: false,
|
||||
sending: 0,
|
||||
sent: 0,
|
||||
status: 'degraded',
|
||||
terminalFailuresAffectHealth: true,
|
||||
},
|
||||
});
|
||||
vi.mocked(NotificationsAPI.getDeliveryLog).mockResolvedValue({ entries: [], windowDays: 7 });
|
||||
vi.mocked(NotificationsAPI.retryTerminalFailures).mockResolvedValue({
|
||||
affected: 3,
|
||||
success: true,
|
||||
});
|
||||
vi.mocked(NotificationsAPI.dismissTerminalFailures).mockResolvedValue({
|
||||
affected: 3,
|
||||
success: true,
|
||||
});
|
||||
|
||||
const { result } = renderHook(() =>
|
||||
useAlertDestinationsTabState({
|
||||
appriseConfig,
|
||||
configLoadError,
|
||||
emailConfig,
|
||||
isLoadingDestinations,
|
||||
isRetrying,
|
||||
onRetryLoad: vi.fn(),
|
||||
setAppriseConfig,
|
||||
}),
|
||||
);
|
||||
|
||||
await waitFor(() => expect(result.deliveryNeedsAttention()).toBe(true));
|
||||
await result.retryTerminalFailures();
|
||||
expect(confirmSpy).toHaveBeenCalledWith(expect.stringContaining('Retry 3 retained deliveries'));
|
||||
expect(NotificationsAPI.retryTerminalFailures).toHaveBeenCalledTimes(1);
|
||||
expect(notificationStore.success).toHaveBeenCalledWith(
|
||||
'3 retained deliveries queued for retry.',
|
||||
);
|
||||
|
||||
await result.dismissTerminalFailures();
|
||||
expect(confirmSpy).toHaveBeenCalledWith(expect.stringContaining('Dismiss 3 retained failures'));
|
||||
expect(NotificationsAPI.dismissTerminalFailures).toHaveBeenCalledTimes(1);
|
||||
expect(notificationStore.success).toHaveBeenCalledWith('3 retained failures dismissed.');
|
||||
expect(NotificationsAPI.getDeliveryLog).toHaveBeenCalledTimes(3);
|
||||
|
||||
confirmSpy.mockRestore();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -80,6 +80,10 @@ export function DestinationsTab(props: DestinationsTabProps) {
|
||||
unavailable={state.deliveryHealthUnavailable()}
|
||||
refreshing={state.refreshingDeliveryHealth()}
|
||||
onRefresh={() => void state.loadDeliveryHealth()}
|
||||
retryingFailures={state.retryingTerminalFailures()}
|
||||
dismissingFailures={state.dismissingTerminalFailures()}
|
||||
onRetryFailures={() => void state.retryTerminalFailures()}
|
||||
onDismissFailures={() => void state.dismissTerminalFailures()}
|
||||
/>
|
||||
</Show>
|
||||
|
||||
|
||||
@@ -8,6 +8,8 @@ import {
|
||||
getAlertDestinationsAppriseTestFailure,
|
||||
getAlertDestinationsAppriseTestSuccess,
|
||||
getAlertDestinationsAppriseValidationError,
|
||||
getAlertDestinationsDeliveryDismissConfirmation,
|
||||
getAlertDestinationsDeliveryRetryConfirmation,
|
||||
getAlertDestinationsEmailTestFailure,
|
||||
getAlertDestinationsEmailTestSuccess,
|
||||
getAlertDestinationsTestPausedWarning,
|
||||
@@ -32,6 +34,8 @@ export interface AlertDestinationsTabStateProps {
|
||||
export function useAlertDestinationsTabState(props: AlertDestinationsTabStateProps) {
|
||||
const [testingEmail, setTestingEmail] = createSignal(false);
|
||||
const [testingApprise, setTestingApprise] = createSignal(false);
|
||||
const [retryingTerminalFailures, setRetryingTerminalFailures] = createSignal(false);
|
||||
const [dismissingTerminalFailures, setDismissingTerminalFailures] = createSignal(false);
|
||||
const {
|
||||
deliveryHealth,
|
||||
deliveryHealthUnavailable,
|
||||
@@ -138,6 +142,46 @@ export function useAlertDestinationsTabState(props: AlertDestinationsTabStatePro
|
||||
void loadDeliveryLog();
|
||||
};
|
||||
|
||||
const retryTerminalFailures = async () => {
|
||||
const count = deliveryHealth()?.queue.attentionRequired ?? 0;
|
||||
if (count <= 0 || !confirm(getAlertDestinationsDeliveryRetryConfirmation(count))) {
|
||||
return;
|
||||
}
|
||||
setRetryingTerminalFailures(true);
|
||||
try {
|
||||
const result = await NotificationsAPI.retryTerminalFailures();
|
||||
notificationStore.success(
|
||||
`${result.affected} retained ${result.affected === 1 ? 'delivery' : 'deliveries'} queued for retry.`,
|
||||
);
|
||||
await Promise.all([loadDeliveryHealth(), loadDeliveryLog()]);
|
||||
} catch (error) {
|
||||
logger.error('Failed to retry retained notification deliveries', error);
|
||||
notificationStore.error('Unable to retry retained notification deliveries.');
|
||||
} finally {
|
||||
setRetryingTerminalFailures(false);
|
||||
}
|
||||
};
|
||||
|
||||
const dismissTerminalFailures = async () => {
|
||||
const count = deliveryHealth()?.queue.attentionRequired ?? 0;
|
||||
if (count <= 0 || !confirm(getAlertDestinationsDeliveryDismissConfirmation(count))) {
|
||||
return;
|
||||
}
|
||||
setDismissingTerminalFailures(true);
|
||||
try {
|
||||
const result = await NotificationsAPI.dismissTerminalFailures();
|
||||
notificationStore.success(
|
||||
`${result.affected} retained ${result.affected === 1 ? 'failure' : 'failures'} dismissed.`,
|
||||
);
|
||||
await Promise.all([loadDeliveryHealth(), loadDeliveryLog()]);
|
||||
} catch (error) {
|
||||
logger.error('Failed to dismiss retained notification failures', error);
|
||||
notificationStore.error('Unable to dismiss retained notification failures.');
|
||||
} finally {
|
||||
setDismissingTerminalFailures(false);
|
||||
}
|
||||
};
|
||||
|
||||
onMount(() => {
|
||||
void loadDeliveryHealth();
|
||||
void loadDeliveryLog();
|
||||
@@ -150,6 +194,8 @@ export function useAlertDestinationsTabState(props: AlertDestinationsTabStatePro
|
||||
deliveryLog,
|
||||
deliveryLogUnavailable,
|
||||
deliveryNeedsAttention,
|
||||
dismissTerminalFailures,
|
||||
dismissingTerminalFailures,
|
||||
handleRetry,
|
||||
hasLoadError,
|
||||
isLoading,
|
||||
@@ -157,6 +203,8 @@ export function useAlertDestinationsTabState(props: AlertDestinationsTabStatePro
|
||||
loadDeliveryLog,
|
||||
refreshingDeliveryHealth,
|
||||
refreshingDeliveryLog,
|
||||
retryTerminalFailures,
|
||||
retryingTerminalFailures,
|
||||
testApprise,
|
||||
testEmailConfig,
|
||||
testingApprise,
|
||||
|
||||
@@ -42,7 +42,11 @@ import {
|
||||
getAlertDestinationsConfigLoadError,
|
||||
getAlertDestinationsDeliveryHealthDescription,
|
||||
getAlertDestinationsDeliveryHealthTitle,
|
||||
getAlertDestinationsDeliveryDismissConfirmation,
|
||||
getAlertDestinationsDeliveryDismissLabel,
|
||||
getAlertDestinationsDeliveryRefreshLabel,
|
||||
getAlertDestinationsDeliveryRetryConfirmation,
|
||||
getAlertDestinationsDeliveryRetryLabel,
|
||||
getAlertDestinationsEmailTestFailure,
|
||||
getAlertDestinationsEmailTestSuccess,
|
||||
getAlertDestinationsLoadErrorBanner,
|
||||
@@ -170,6 +174,14 @@ describe('alertDestinationsPresentation', () => {
|
||||
}),
|
||||
).toContain('could not verify the notification queue');
|
||||
expect(getAlertDestinationsDeliveryRefreshLabel()).toBe('Refresh delivery status');
|
||||
expect(getAlertDestinationsDeliveryRetryLabel()).toBe('Retry retained deliveries');
|
||||
expect(getAlertDestinationsDeliveryDismissLabel()).toBe('Dismiss retained failures');
|
||||
expect(getAlertDestinationsDeliveryRetryConfirmation(1)).toContain(
|
||||
'A destination that accepted an earlier attempt may receive a duplicate',
|
||||
);
|
||||
expect(getAlertDestinationsDeliveryDismissConfirmation(2)).toContain(
|
||||
'Delivery history remains available',
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -153,6 +153,20 @@ describe('docsLinks', () => {
|
||||
expect(importContract).toContain('settings:write');
|
||||
});
|
||||
|
||||
it('ships the retained notification recovery API contract', () => {
|
||||
const apiReference = readFileSync(path.join(repoRoot, 'docs', 'API.md'), 'utf8');
|
||||
const shippedAPIReference = readFileSync(
|
||||
path.join(frontendRoot, 'public', 'docs', 'API.md'),
|
||||
'utf8',
|
||||
);
|
||||
|
||||
expect(shippedAPIReference).toBe(apiReference);
|
||||
expect(apiReference).toContain('`POST /api/notifications/terminal-failures/retry`');
|
||||
expect(apiReference).toContain('`POST /api/notifications/terminal-failures/dismiss`');
|
||||
expect(apiReference).toContain('Existing per-attempt delivery history is preserved');
|
||||
expect(apiReference).toContain('without deleting delivery history');
|
||||
});
|
||||
|
||||
it('ships the truthful Patrol objective API contract', () => {
|
||||
const apiReference = readFileSync(path.join(repoRoot, 'docs', 'API.md'), 'utf8');
|
||||
const shippedAPIReference = readFileSync(
|
||||
|
||||
@@ -79,6 +79,8 @@ export const ALERT_DESTINATIONS_DELIVERY_DEGRADED_TITLE = 'Notification delivery
|
||||
export const ALERT_DESTINATIONS_DELIVERY_UNAVAILABLE_TITLE =
|
||||
'Notification delivery status is unavailable';
|
||||
export const ALERT_DESTINATIONS_DELIVERY_REFRESH_LABEL = 'Refresh delivery status';
|
||||
export const ALERT_DESTINATIONS_DELIVERY_RETRY_LABEL = 'Retry retained deliveries';
|
||||
export const ALERT_DESTINATIONS_DELIVERY_DISMISS_LABEL = 'Dismiss retained failures';
|
||||
|
||||
export function getAlertDestinationsConfigLoadError() {
|
||||
return ALERT_DESTINATIONS_CONFIG_LOAD_ERROR;
|
||||
@@ -199,6 +201,22 @@ export function getAlertDestinationsDeliveryRefreshLabel() {
|
||||
return ALERT_DESTINATIONS_DELIVERY_REFRESH_LABEL;
|
||||
}
|
||||
|
||||
export function getAlertDestinationsDeliveryRetryLabel() {
|
||||
return ALERT_DESTINATIONS_DELIVERY_RETRY_LABEL;
|
||||
}
|
||||
|
||||
export function getAlertDestinationsDeliveryDismissLabel() {
|
||||
return ALERT_DESTINATIONS_DELIVERY_DISMISS_LABEL;
|
||||
}
|
||||
|
||||
export function getAlertDestinationsDeliveryRetryConfirmation(count: number) {
|
||||
return `Retry ${count} retained ${count === 1 ? 'delivery' : 'deliveries'} now? A destination that accepted an earlier attempt may receive a duplicate.`;
|
||||
}
|
||||
|
||||
export function getAlertDestinationsDeliveryDismissConfirmation(count: number) {
|
||||
return `Dismiss ${count} retained ${count === 1 ? 'failure' : 'failures'}? Pulse will clear the warning and will not retry them. Delivery history remains available.`;
|
||||
}
|
||||
|
||||
// Shown instead of the plain test-success toast when the backend reports the
|
||||
// test went out while real alert delivery is paused. Without this, a
|
||||
// successful test is exactly how installs come to believe delivery works
|
||||
|
||||
@@ -82,7 +82,7 @@ func (h *NotificationQueueHandlers) GetQueueStats(w http.ResponseWriter, r *http
|
||||
|
||||
// RetryDLQItem retries a specific notification from the DLQ
|
||||
func (h *NotificationQueueHandlers) RetryDLQItem(w http.ResponseWriter, r *http.Request) {
|
||||
if !apihttp.EnsureScope(w, r, config.ScopeMonitoringWrite) {
|
||||
if !apihttp.EnsureScope(w, r, config.ScopeSettingsWrite) {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -129,7 +129,7 @@ func (h *NotificationQueueHandlers) RetryDLQItem(w http.ResponseWriter, r *http.
|
||||
|
||||
// DeleteDLQItem removes a notification from the DLQ permanently
|
||||
func (h *NotificationQueueHandlers) DeleteDLQItem(w http.ResponseWriter, r *http.Request) {
|
||||
if !apihttp.EnsureScope(w, r, config.ScopeMonitoringWrite) {
|
||||
if !apihttp.EnsureScope(w, r, config.ScopeSettingsWrite) {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -174,6 +174,56 @@ func (h *NotificationQueueHandlers) DeleteDLQItem(w http.ResponseWriter, r *http
|
||||
}
|
||||
}
|
||||
|
||||
// RetryTerminalFailures returns all retained terminal failures to the queue
|
||||
// after an operator has repaired the destination.
|
||||
func (h *NotificationQueueHandlers) RetryTerminalFailures(w http.ResponseWriter, r *http.Request) {
|
||||
if !apihttp.EnsureScope(w, r, config.ScopeSettingsWrite) {
|
||||
return
|
||||
}
|
||||
queue := h.monitor.GetNotificationManager().GetQueue()
|
||||
if queue == nil {
|
||||
http.Error(w, "Notification queue not initialized", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
affected, err := queue.RetryTerminalFailures()
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("Failed to retry retained terminal notifications")
|
||||
http.Error(w, "Failed to retry retained notification failures", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if err := utils.WriteJSONResponse(w, map[string]interface{}{
|
||||
"success": true,
|
||||
"affected": affected,
|
||||
}); err != nil {
|
||||
log.Error().Err(err).Msg("Failed to write terminal retry response")
|
||||
}
|
||||
}
|
||||
|
||||
// DismissTerminalFailures cancels all retained terminal queue rows without
|
||||
// deleting their immutable delivery audit.
|
||||
func (h *NotificationQueueHandlers) DismissTerminalFailures(w http.ResponseWriter, r *http.Request) {
|
||||
if !apihttp.EnsureScope(w, r, config.ScopeSettingsWrite) {
|
||||
return
|
||||
}
|
||||
queue := h.monitor.GetNotificationManager().GetQueue()
|
||||
if queue == nil {
|
||||
http.Error(w, "Notification queue not initialized", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
affected, err := queue.DismissTerminalFailures()
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("Failed to dismiss retained terminal notifications")
|
||||
http.Error(w, "Failed to dismiss retained notification failures", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if err := utils.WriteJSONResponse(w, map[string]interface{}{
|
||||
"success": true,
|
||||
"affected": affected,
|
||||
}); err != nil {
|
||||
log.Error().Err(err).Msg("Failed to write terminal dismissal response")
|
||||
}
|
||||
}
|
||||
|
||||
// HandleNotificationQueue routes notification queue requests
|
||||
func (h *NotificationQueueHandlers) HandleNotificationQueue(w http.ResponseWriter, r *http.Request) {
|
||||
path := r.URL.Path
|
||||
@@ -187,6 +237,10 @@ func (h *NotificationQueueHandlers) HandleNotificationQueue(w http.ResponseWrite
|
||||
h.RetryDLQItem(w, r)
|
||||
case path == "/api/notifications/dlq/delete" && r.Method == http.MethodPost:
|
||||
h.DeleteDLQItem(w, r)
|
||||
case path == "/api/notifications/terminal-failures/retry" && r.Method == http.MethodPost:
|
||||
h.RetryTerminalFailures(w, r)
|
||||
case path == "/api/notifications/terminal-failures/dismiss" && r.Method == http.MethodPost:
|
||||
h.DismissTerminalFailures(w, r)
|
||||
default:
|
||||
http.Error(w, "Not found", http.StatusNotFound)
|
||||
}
|
||||
|
||||
@@ -116,6 +116,43 @@ func TestNotificationQueueHandlers_RetryAndDelete(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestNotificationQueueHandlers_BulkTerminalFailureRecovery(t *testing.T) {
|
||||
handler, queue := newNotificationQueueHandlers(t)
|
||||
enqueueDLQNotification(t, queue, "notif-bulk")
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/api/notifications/terminal-failures/retry", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
handler.RetryTerminalFailures(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("RetryTerminalFailures status = %d, want 200: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
var response struct {
|
||||
Affected int `json:"affected"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &response); err != nil {
|
||||
t.Fatalf("decode retry response: %v", err)
|
||||
}
|
||||
if response.Affected != 1 {
|
||||
t.Fatalf("retry affected = %d, want 1", response.Affected)
|
||||
}
|
||||
|
||||
if err := queue.UpdateStatus("notif-bulk", notifications.QueueStatusDLQ, "still unavailable"); err != nil {
|
||||
t.Fatalf("mark retried notification DLQ: %v", err)
|
||||
}
|
||||
req = httptest.NewRequest(http.MethodPost, "/api/notifications/terminal-failures/dismiss", nil)
|
||||
rec = httptest.NewRecorder()
|
||||
handler.DismissTerminalFailures(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("DismissTerminalFailures status = %d, want 200: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &response); err != nil {
|
||||
t.Fatalf("decode dismiss response: %v", err)
|
||||
}
|
||||
if response.Affected != 1 {
|
||||
t.Fatalf("dismiss affected = %d, want 1", response.Affected)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNotificationQueueHandlers_HandleNotificationQueue(t *testing.T) {
|
||||
handler, queue := newNotificationQueueHandlers(t)
|
||||
enqueueDLQNotification(t, queue, "notif-3")
|
||||
|
||||
@@ -594,6 +594,8 @@ var allRouteAllowlist = []string{
|
||||
"/api/notifications/queue/stats",
|
||||
"/api/notifications/dlq/retry",
|
||||
"/api/notifications/dlq/delete",
|
||||
"/api/notifications/terminal-failures/retry",
|
||||
"/api/notifications/terminal-failures/dismiss",
|
||||
"/api/system/settings",
|
||||
"/api/system/settings/update",
|
||||
"/api/system/ssh-config",
|
||||
|
||||
@@ -335,6 +335,20 @@ func (r *Router) registerMonitoringResourceRoutes(
|
||||
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
|
||||
}
|
||||
})))
|
||||
r.mux.HandleFunc("/api/notifications/terminal-failures/retry", RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, func(w http.ResponseWriter, req *http.Request) {
|
||||
if req.Method == http.MethodPost {
|
||||
r.notificationQueueHandlers.RetryTerminalFailures(w, req)
|
||||
} else {
|
||||
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
|
||||
}
|
||||
})))
|
||||
r.mux.HandleFunc("/api/notifications/terminal-failures/dismiss", RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, func(w http.ResponseWriter, req *http.Request) {
|
||||
if req.Method == http.MethodPost {
|
||||
r.notificationQueueHandlers.DismissTerminalFailures(w, req)
|
||||
} else {
|
||||
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
|
||||
}
|
||||
})))
|
||||
// AI-powered infrastructure discovery endpoints
|
||||
r.mux.HandleFunc("/api/discovery", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.discoveryHandlers.HandleListDiscoveries)))
|
||||
r.mux.HandleFunc("/api/discovery/status", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.discoveryHandlers.HandleGetStatus)))
|
||||
|
||||
@@ -508,6 +508,8 @@ func TestNotificationsDLQMutationsRequireSettingsWriteScope(t *testing.T) {
|
||||
cases := []string{
|
||||
"/api/notifications/dlq/retry",
|
||||
"/api/notifications/dlq/delete",
|
||||
"/api/notifications/terminal-failures/retry",
|
||||
"/api/notifications/terminal-failures/dismiss",
|
||||
}
|
||||
|
||||
for _, path := range cases {
|
||||
@@ -2897,6 +2899,8 @@ func TestProxyAuthNonAdminDeniedAdminEndpoints(t *testing.T) {
|
||||
{method: http.MethodGet, path: "/api/notifications/dlq", body: ""},
|
||||
{method: http.MethodPost, path: "/api/notifications/dlq/retry", body: `{}`},
|
||||
{method: http.MethodPost, path: "/api/notifications/dlq/delete", body: `{}`},
|
||||
{method: http.MethodPost, path: "/api/notifications/terminal-failures/retry", body: `{}`},
|
||||
{method: http.MethodPost, path: "/api/notifications/terminal-failures/dismiss", body: `{}`},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
|
||||
@@ -1278,6 +1278,122 @@ func (nq *NotificationQueue) GetDLQ(limit int) ([]*QueuedNotification, error) {
|
||||
return notifications, rows.Err()
|
||||
}
|
||||
|
||||
// RetryTerminalFailures returns every retained failed or dead-lettered
|
||||
// delivery to the pending queue with a fresh retry budget. The per-attempt
|
||||
// audit rows remain intact, so this operator recovery never rewrites delivery
|
||||
// history or makes an earlier failure look successful.
|
||||
func (nq *NotificationQueue) RetryTerminalFailures() (int, error) {
|
||||
return nq.resolveTerminalFailures(true)
|
||||
}
|
||||
|
||||
// DismissTerminalFailures marks every retained failed or dead-lettered
|
||||
// delivery as cancelled. This clears the active queue-health warning while
|
||||
// retaining the immutable delivery audit instead of asking operators to
|
||||
// delete the queue database.
|
||||
func (nq *NotificationQueue) DismissTerminalFailures() (int, error) {
|
||||
return nq.resolveTerminalFailures(false)
|
||||
}
|
||||
|
||||
func (nq *NotificationQueue) resolveTerminalFailures(retry bool) (int, error) {
|
||||
if nq == nil || nq.db == nil {
|
||||
return 0, fmt.Errorf("notification queue not initialized")
|
||||
}
|
||||
|
||||
nq.mu.Lock()
|
||||
defer nq.mu.Unlock()
|
||||
|
||||
rows, err := nq.db.Query(`
|
||||
SELECT id, operational_links
|
||||
FROM notification_queue
|
||||
WHERE status IN ('failed', 'dlq')
|
||||
ORDER BY completed_at, id
|
||||
`)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("read retained terminal notifications: %w", err)
|
||||
}
|
||||
type terminalNotification struct {
|
||||
id string
|
||||
links []operationaltrust.NotificationLink
|
||||
}
|
||||
terminal := make([]terminalNotification, 0)
|
||||
for rows.Next() {
|
||||
var id, rawLinks string
|
||||
if err := rows.Scan(&id, &rawLinks); err != nil {
|
||||
_ = rows.Close()
|
||||
return 0, fmt.Errorf("scan retained terminal notification: %w", err)
|
||||
}
|
||||
var links []operationaltrust.NotificationLink
|
||||
if err := json.Unmarshal([]byte(rawLinks), &links); err != nil {
|
||||
_ = rows.Close()
|
||||
return 0, fmt.Errorf("decode retained terminal notification links: %w", err)
|
||||
}
|
||||
terminal = append(terminal, terminalNotification{id: id, links: links})
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return 0, fmt.Errorf("close retained terminal notification rows: %w", err)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, fmt.Errorf("iterate retained terminal notifications: %w", err)
|
||||
}
|
||||
if len(terminal) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
tx, err := nq.db.Begin()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("begin terminal notification recovery: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
now := time.Now()
|
||||
affected := 0
|
||||
for _, notification := range terminal {
|
||||
state := operationaltrust.NotificationCancelled
|
||||
if retry {
|
||||
state = operationaltrust.NotificationRetrying
|
||||
}
|
||||
linksJSON, err := json.Marshal(transitionNotificationLinks(notification.links, state, now))
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("encode terminal notification links: %w", err)
|
||||
}
|
||||
|
||||
var result sql.Result
|
||||
if retry {
|
||||
result, err = tx.Exec(`
|
||||
UPDATE notification_queue
|
||||
SET status = 'pending', attempts = 0, last_attempt = NULL,
|
||||
last_error = NULL, next_retry_at = ?, completed_at = NULL,
|
||||
operational_links = ?
|
||||
WHERE id = ? AND status IN ('failed', 'dlq')
|
||||
`, now.Unix(), string(linksJSON), notification.id)
|
||||
} else {
|
||||
result, err = tx.Exec(`
|
||||
UPDATE notification_queue
|
||||
SET status = 'cancelled', last_attempt = ?,
|
||||
last_error = 'Dismissed by operator', next_retry_at = NULL,
|
||||
completed_at = ?, operational_links = ?
|
||||
WHERE id = ? AND status IN ('failed', 'dlq')
|
||||
`, now.Unix(), now.Unix(), string(linksJSON), notification.id)
|
||||
}
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("resolve terminal notification %s: %w", notification.id, err)
|
||||
}
|
||||
if count, countErr := result.RowsAffected(); countErr == nil {
|
||||
affected += int(count)
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return 0, fmt.Errorf("commit terminal notification recovery: %w", err)
|
||||
}
|
||||
if retry && affected > 0 {
|
||||
select {
|
||||
case nq.notifyChan <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
return affected, nil
|
||||
}
|
||||
|
||||
// RecordAudit records a notification delivery attempt in the audit log
|
||||
func (nq *NotificationQueue) RecordAudit(notif *QueuedNotification, success bool, errorMsg string) error {
|
||||
nq.mu.Lock()
|
||||
|
||||
@@ -1381,6 +1381,73 @@ func TestGetQueueStats(t *testing.T) {
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("operator recovery retries or dismisses every terminal row without deleting history", func(t *testing.T) {
|
||||
nq, err := NewNotificationQueue(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("NewNotificationQueue: %v", err)
|
||||
}
|
||||
defer func() { _ = nq.Stop() }()
|
||||
|
||||
for _, notif := range []*QueuedNotification{
|
||||
{ID: "failed-recovery", Type: "email", Status: QueueStatusPending, Attempts: 8, MaxAttempts: 8, Config: []byte(`{}`)},
|
||||
{ID: "dlq-recovery", Type: "webhook", Status: QueueStatusPending, Attempts: 8, MaxAttempts: 8, Config: []byte(`{}`)},
|
||||
{ID: "pending-unrelated", Type: "email", Status: QueueStatusPending, MaxAttempts: 8, Config: []byte(`{}`)},
|
||||
} {
|
||||
if err := nq.Enqueue(notif); err != nil {
|
||||
t.Fatalf("enqueue %s: %v", notif.ID, err)
|
||||
}
|
||||
}
|
||||
if err := nq.UpdateStatus("failed-recovery", QueueStatusFailed, "authentication failed"); err != nil {
|
||||
t.Fatalf("mark failed: %v", err)
|
||||
}
|
||||
if err := nq.UpdateStatus("dlq-recovery", QueueStatusDLQ, "retries exhausted"); err != nil {
|
||||
t.Fatalf("mark dlq: %v", err)
|
||||
}
|
||||
|
||||
affected, err := nq.RetryTerminalFailures()
|
||||
if err != nil {
|
||||
t.Fatalf("RetryTerminalFailures: %v", err)
|
||||
}
|
||||
if affected != 2 {
|
||||
t.Fatalf("retry affected = %d, want 2", affected)
|
||||
}
|
||||
for _, id := range []string{"failed-recovery", "dlq-recovery"} {
|
||||
var status string
|
||||
var attempts int
|
||||
var lastAttempt, lastError, completedAt *int64
|
||||
if err := nq.db.QueryRow(`
|
||||
SELECT status, attempts, last_attempt, last_error, completed_at
|
||||
FROM notification_queue WHERE id = ?
|
||||
`, id).Scan(&status, &attempts, &lastAttempt, &lastError, &completedAt); err != nil {
|
||||
t.Fatalf("read retried %s: %v", id, err)
|
||||
}
|
||||
if status != string(QueueStatusPending) || attempts != 0 || lastAttempt != nil || lastError != nil || completedAt != nil {
|
||||
t.Fatalf("retried %s = status %q attempts %d lastAttempt %v lastError %v completedAt %v", id, status, attempts, lastAttempt, lastError, completedAt)
|
||||
}
|
||||
}
|
||||
|
||||
if err := nq.UpdateStatus("failed-recovery", QueueStatusFailed, "still invalid"); err != nil {
|
||||
t.Fatalf("mark failed again: %v", err)
|
||||
}
|
||||
if err := nq.UpdateStatus("dlq-recovery", QueueStatusDLQ, "still unavailable"); err != nil {
|
||||
t.Fatalf("mark dlq again: %v", err)
|
||||
}
|
||||
affected, err = nq.DismissTerminalFailures()
|
||||
if err != nil {
|
||||
t.Fatalf("DismissTerminalFailures: %v", err)
|
||||
}
|
||||
if affected != 2 {
|
||||
t.Fatalf("dismiss affected = %d, want 2", affected)
|
||||
}
|
||||
stats, err := nq.GetQueueStats()
|
||||
if err != nil {
|
||||
t.Fatalf("GetQueueStats: %v", err)
|
||||
}
|
||||
if stats[string(QueueStatusFailed)] != 0 || stats[string(QueueStatusDLQ)] != 0 || stats[string(QueueStatusCancelled)] != 2 || stats[string(QueueStatusPending)] != 1 {
|
||||
t.Fatalf("stats after dismissal = %#v", stats)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("UpdateStatus returns error for non-existent notification", func(t *testing.T) {
|
||||
tempDir := t.TempDir()
|
||||
nq, err := NewNotificationQueue(tempDir)
|
||||
|
||||
Reference in New Issue
Block a user