diff --git a/backend/src/__tests__/notification-suppression-routes-api.test.ts b/backend/src/__tests__/notification-suppression-routes-api.test.ts index ef6c54ce..4cd78162 100644 --- a/backend/src/__tests__/notification-suppression-routes-api.test.ts +++ b/backend/src/__tests__/notification-suppression-routes-api.test.ts @@ -506,7 +506,7 @@ describe('Notification suppression - CRUD', () => { DatabaseService.getInstance().deleteNotificationSuppressionRule(940005); }); - it('replica does not resurrect a rule after it was deleted, even with a newer updated_at', async () => { + it('omitted-body replica DELETE is permanent; delayed POST cannot resurrect', async () => { const jwt = await import('jsonwebtoken'); const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); @@ -538,9 +538,8 @@ describe('Notification suppression - CRUD', () => { expect(del.status).toBe(200); expect(DatabaseService.getInstance().getNotificationSuppressionRule(950006)).toBeUndefined(); - // A delayed POST arrives after the DELETE, reordered by the network. Even - // though its updated_at is newer than anything the sender ever sent before - // the delete, the delete is authoritative: this id must stay gone. + // Omitted DELETE body (old hub) fails closed as permanent. A delayed POST + // with any updated_at must stay blocked. const delayed = await request(app) .post('/api/notification-suppression-rules/replica') .set('Authorization', `Bearer ${token}`) @@ -612,21 +611,479 @@ describe('Notification suppression - CRUD', () => { .set('Authorization', `Bearer ${token}`) .send({ rule: replicaRule({ updated_at: 1000 }) }); expect(first.status).toBe(200); + expect(first.body.outcome).toBe('applied'); expect(DatabaseService.getInstance().getNotificationSuppressionRule(970008)?.schedule).not.toBeNull(); - // This is the worst case the fix protects: a capability-cleanup DELETE - // retracts an all-day/scheduled mute from a node that stopped supporting - // it. A delayed re-push of the scheduled rule must not undo that cleanup. + // Capability-cleanup DELETE is recoverable at the pushed version. A delayed + // re-push at the same or older watermark must not undo cleanup. const del = await request(app) .delete('/api/notification-suppression-rules/replica/970008') - .set('Authorization', `Bearer ${token}`); + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'recoverable', source_updated_at: 1000 }); expect(del.status).toBe(200); + expect(del.body.outcome).toBe('applied'); const delayed = await request(app) .post('/api/notification-suppression-rules/replica') .set('Authorization', `Bearer ${token}`) - .send({ rule: replicaRule({ updated_at: 2000 }) }); + .send({ rule: replicaRule({ updated_at: 1000 }) }); expect(delayed.status).toBe(200); + expect(delayed.body.outcome).toBe('ignored_recoverable_watermark'); expect(DatabaseService.getInstance().getNotificationSuppressionRule(970008)).toBeUndefined(); + expect(DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(970008)?.kind).toBe( + 'recoverable', + ); + }); + + it('recoverable soft-cleanup allows recreate when hub re-save is newer', async () => { + const jwt = await import('jsonwebtoken'); + const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); + const replicaRule = (overrides: Record) => ({ + id: 980009, + name: 'replica-soft-cleanup-resave', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + schedule: { days: [2], start_minute: 60, end_minute: 120, tz: 'UTC' }, + created_at: 1, + ...overrides, + }); + + // Receiver clock skew must not affect ordering. Sign the JWT under the same + // mocked clock so exp verification stays valid. + const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(9_000_000_000_000); + const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1h' }); + try { + const first = await request(app) + .post('/api/notification-suppression-rules/replica') + .set('Authorization', `Bearer ${token}`) + .send({ rule: replicaRule({ updated_at: 1000 }) }); + expect(first.status).toBe(200); + expect(first.body.outcome).toBe('applied'); + + const del = await request(app) + .delete('/api/notification-suppression-rules/replica/980009') + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'recoverable', source_updated_at: 1000 }); + expect(del.status).toBe(200); + expect(del.body.outcome).toBe('applied'); + expect(DatabaseService.getInstance().getNotificationSuppressionRule(980009)).toBeUndefined(); + + const tie = await request(app) + .post('/api/notification-suppression-rules/replica') + .set('Authorization', `Bearer ${token}`) + .send({ rule: replicaRule({ updated_at: 1000 }) }); + expect(tie.status).toBe(200); + expect(tie.body.outcome).toBe('ignored_recoverable_watermark'); + expect(DatabaseService.getInstance().getNotificationSuppressionRule(980009)).toBeUndefined(); + + const resave = await request(app) + .post('/api/notification-suppression-rules/replica') + .set('Authorization', `Bearer ${token}`) + .send({ rule: replicaRule({ name: 'replica-soft-cleanup-resave-v2', updated_at: 2000 }) }); + expect(resave.status).toBe(200); + expect(resave.body.outcome).toBe('applied'); + const restored = DatabaseService.getInstance().getNotificationSuppressionRule(980009); + expect(restored?.name).toBe('replica-soft-cleanup-resave-v2'); + expect(restored?.updated_at).toBe(2000); + expect(DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(980009)).toBeUndefined(); + } finally { + nowSpy.mockRestore(); + } + }); + + it('permanent DELETE blocks any later POST regardless of updated_at', async () => { + const jwt = await import('jsonwebtoken'); + const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); + const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + + const del = await request(app) + .delete('/api/notification-suppression-rules/replica/990010') + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'permanent', source_updated_at: 50 }); + expect(del.status).toBe(200); + expect(del.body.outcome).toBe('applied'); + + const post = await request(app) + .post('/api/notification-suppression-rules/replica') + .set('Authorization', `Bearer ${token}`) + .send({ + rule: { + id: 990010, + name: 'should-not-return', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + created_at: 1, + updated_at: Number.MAX_SAFE_INTEGER, + }, + }); + expect(post.status).toBe(200); + expect(post.body.outcome).toBe('ignored_permanent_tombstone'); + expect(DatabaseService.getInstance().getNotificationSuppressionRule(990010)).toBeUndefined(); + expect(DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(990010)?.kind).toBe( + 'permanent', + ); + }); + + it('stale recoverable DELETE does not remove a newer stored row', async () => { + const jwt = await import('jsonwebtoken'); + const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); + const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + const replicaRule = (overrides: Record) => ({ + id: 991011, + name: 'v200', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + created_at: 1, + ...overrides, + }); + + const post = await request(app) + .post('/api/notification-suppression-rules/replica') + .set('Authorization', `Bearer ${token}`) + .send({ rule: replicaRule({ updated_at: 200 }) }); + expect(post.status).toBe(200); + + const del = await request(app) + .delete('/api/notification-suppression-rules/replica/991011') + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'recoverable', source_updated_at: 100 }); + expect(del.status).toBe(200); + expect(del.body.outcome).toBe('ignored_stale'); + expect(DatabaseService.getInstance().getNotificationSuppressionRule(991011)?.updated_at).toBe(200); + expect(DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(991011)).toBeUndefined(); + }); + + it('recoverable DELETE at exact stored version deletes and tombstones', async () => { + const jwt = await import('jsonwebtoken'); + const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); + const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + + await request(app) + .post('/api/notification-suppression-rules/replica') + .set('Authorization', `Bearer ${token}`) + .send({ + rule: { + id: 992012, + name: 'exact-tie-delete', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + created_at: 1, + updated_at: 150, + }, + }); + + const del = await request(app) + .delete('/api/notification-suppression-rules/replica/992012') + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'recoverable', source_updated_at: 150 }); + expect(del.status).toBe(200); + expect(DatabaseService.getInstance().getNotificationSuppressionRule(992012)).toBeUndefined(); + const tomb = DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(992012); + expect(tomb?.kind).toBe('recoverable'); + expect(tomb?.source_updated_at).toBe(150); + }); + + it('reordered recoverable tombstones keep the max watermark; permanent wins', async () => { + const jwt = await import('jsonwebtoken'); + const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); + const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + + await request(app) + .delete('/api/notification-suppression-rules/replica/993013') + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'recoverable', source_updated_at: 200 }); + await request(app) + .delete('/api/notification-suppression-rules/replica/993013') + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'recoverable', source_updated_at: 100 }); + expect( + DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(993013)?.source_updated_at, + ).toBe(200); + + await request(app) + .delete('/api/notification-suppression-rules/replica/993013') + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'permanent', source_updated_at: 50 }); + const tomb = DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(993013); + expect(tomb?.kind).toBe('permanent'); + expect(tomb?.source_updated_at).toBe(200); + + // Later recoverable cannot weaken permanent. + await request(app) + .delete('/api/notification-suppression-rules/replica/993013') + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'recoverable', source_updated_at: 999 }); + expect(DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(993013)?.kind).toBe( + 'permanent', + ); + }); + + it('partial or invalid DELETE body returns 400 without mutation', async () => { + const jwt = await import('jsonwebtoken'); + const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); + const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + + await request(app) + .post('/api/notification-suppression-rules/replica') + .set('Authorization', `Bearer ${token}`) + .send({ + rule: { + id: 994014, + name: 'keep-me', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + created_at: 1, + updated_at: 10, + }, + }); + + const cases: object[] = [ + { kind: 'recoverable' }, + { source_updated_at: 1 }, + { kind: 'nope', source_updated_at: 1 }, + { kind: 'recoverable', source_updated_at: 1.5 }, + { kind: 'recoverable', source_updated_at: -1 }, + { kind: 'recoverable', source_updated_at: '1' }, + { kind: 'recoverable', source_updated_at: null }, + { kind: 'recoverable', source_updated_at: Number.MAX_SAFE_INTEGER + 1 }, + ]; + for (const body of cases) { + const res = await request(app) + .delete('/api/notification-suppression-rules/replica/994014') + .set('Authorization', `Bearer ${token}`) + .send(body); + expect(res.status).toBe(400); + } + expect(DatabaseService.getInstance().getNotificationSuppressionRule(994014)?.name).toBe('keep-me'); + expect(DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(994014)).toBeUndefined(); + }); + + it('replica POST rejects missing or invalid created_at and updated_at', async () => { + const jwt = await import('jsonwebtoken'); + const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); + const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + const base = { + id: 995015, + name: 'bad-ts', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + }; + + for (const rule of [ + { ...base, updated_at: 1 }, + { ...base, created_at: 1 }, + { ...base, created_at: -1, updated_at: 1 }, + { ...base, created_at: 1, updated_at: 1.5 }, + { ...base, created_at: 1, updated_at: '1' }, + ]) { + const res = await request(app) + .post('/api/notification-suppression-rules/replica') + .set('Authorization', `Bearer ${token}`) + .send({ rule }); + expect(res.status).toBe(400); + } + expect(DatabaseService.getInstance().getNotificationSuppressionRule(995015)).toBeUndefined(); + }); + + it('permanent DELETE removes a newer stored row', async () => { + const jwt = await import('jsonwebtoken'); + const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); + const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + + await request(app) + .post('/api/notification-suppression-rules/replica') + .set('Authorization', `Bearer ${token}`) + .send({ + rule: { + id: 999019, + name: 'newer-row', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + created_at: 1, + updated_at: 200, + }, + }); + + const del = await request(app) + .delete('/api/notification-suppression-rules/replica/999019') + .set('Authorization', `Bearer ${token}`) + .send({ kind: 'permanent', source_updated_at: 50 }); + expect(del.status).toBe(200); + expect(DatabaseService.getInstance().getNotificationSuppressionRule(999019)).toBeUndefined(); + expect(DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(999019)?.kind).toBe( + 'permanent', + ); + }); + + it('empty JSON DELETE body fails closed as permanent', async () => { + const jwt = await import('jsonwebtoken'); + const { TEST_JWT_SECRET } = await import('./helpers/testConstants'); + const token = jwt.default.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + + const del = await request(app) + .delete('/api/notification-suppression-rules/replica/999020') + .set('Authorization', `Bearer ${token}`) + .set('Content-Type', 'application/json') + .send({}); + expect(del.status).toBe(200); + expect(DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(999020)?.kind).toBe( + 'permanent', + ); + expect( + DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(999020)?.source_updated_at, + ).toBe(0); + }); + + it('one-arg deleteNotificationSuppressionRule defaults to permanent', () => { + DatabaseService.getInstance().upsertNotificationSuppressionRuleReplica({ + id: 996016, + name: 'one-arg', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + schedule: null, + scheduleInvalid: false, + created_at: 1, + updated_at: 5, + }); + DatabaseService.getInstance().deleteNotificationSuppressionRule(996016); + expect(DatabaseService.getInstance().getNotificationSuppressionRule(996016)).toBeUndefined(); + expect(DatabaseService.getInstance().getNotificationSuppressionRuleTombstone(996016)?.kind).toBe( + 'permanent', + ); + }); + + it('delete rolls back when tombstone upsert fails', () => { + const db = DatabaseService.getInstance(); + db.upsertNotificationSuppressionRuleReplica({ + id: 997017, + name: 'atomic-del', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + schedule: null, + scheduleInvalid: false, + created_at: 1, + updated_at: 7, + }); + const raw = db.getDb(); + const orig = raw.prepare.bind(raw); + const spy = vi.spyOn(raw, 'prepare').mockImplementation(((sql: string) => { + if ( + typeof sql === 'string' && + sql.includes('INSERT INTO notification_suppression_rule_tombstones') + ) { + throw new Error('forced tombstone upsert failure'); + } + return orig(sql); + }) as typeof raw.prepare); + + expect(() => + db.deleteNotificationSuppressionRule(997017, { + kind: 'recoverable', + source_updated_at: 7, + }), + ).toThrow(/forced tombstone upsert failure/); + spy.mockRestore(); + + expect(db.getNotificationSuppressionRule(997017)?.name).toBe('atomic-del'); + expect(db.getNotificationSuppressionRuleTombstone(997017)).toBeUndefined(); + }); + + it('recoverable recreate rolls back when insert fails after tombstone clear attempt', () => { + const db = DatabaseService.getInstance(); + db.deleteNotificationSuppressionRule(998018, { + kind: 'recoverable', + source_updated_at: 10, + }); + expect(db.getNotificationSuppressionRuleTombstone(998018)?.kind).toBe('recoverable'); + + const raw = db.getDb(); + const orig = raw.prepare.bind(raw); + const spy = vi.spyOn(raw, 'prepare').mockImplementation(((sql: string) => { + if ( + typeof sql === 'string' && + sql.includes('INSERT INTO notification_suppression_rules') + ) { + throw new Error('forced replica insert failure'); + } + return orig(sql); + }) as typeof raw.prepare); + + expect(() => + db.upsertNotificationSuppressionRuleReplica({ + id: 998018, + name: 'recreate-fail', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + schedule: null, + scheduleInvalid: false, + created_at: 1, + updated_at: 20, + }), + ).toThrow(/forced replica insert failure/); + spy.mockRestore(); + + expect(db.getNotificationSuppressionRule(998018)).toBeUndefined(); + expect(db.getNotificationSuppressionRuleTombstone(998018)?.kind).toBe('recoverable'); + expect(db.getNotificationSuppressionRuleTombstone(998018)?.source_updated_at).toBe(10); }); }); diff --git a/backend/src/__tests__/notification-suppression-sync.test.ts b/backend/src/__tests__/notification-suppression-sync.test.ts index 3a401972..d46ae8d9 100644 --- a/backend/src/__tests__/notification-suppression-sync.test.ts +++ b/backend/src/__tests__/notification-suppression-sync.test.ts @@ -1,5 +1,6 @@ /** - * Fleet sync for suppression rules: node_id normalize, capability gate, stale DELETE. + * Fleet sync for suppression rules: node_id normalize, capability gate, stale DELETE, + * durable pending retractions, permanent fan-out to all remotes. */ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; @@ -8,12 +9,18 @@ const mockGetProxyTarget = vi.fn(); const mockGetNodes = vi.fn(); const mockGetNode = vi.fn(); const mockRemoteAdvertises = vi.fn(); +const mockUpsertPending = vi.fn(); +const mockDeletePending = vi.fn(); +const mockListPending = vi.fn(); vi.mock('../services/DatabaseService', () => ({ DatabaseService: { getInstance: () => ({ getNodes: mockGetNodes, getNode: mockGetNode, + upsertNotificationSuppressionPendingRetraction: mockUpsertPending, + deleteNotificationSuppressionPendingRetraction: mockDeletePending, + listNotificationSuppressionPendingRetractions: mockListPending, }), }, })); @@ -46,6 +53,7 @@ import { syncSuppressionRuleToFleet, syncSuppressionRuleUpdateToFleet, replicationTargetIds, + flushPendingSuppressionRetractions, } from '../helpers/notificationSuppressionSync'; import type { NotificationSuppressionRule } from '../services/DatabaseService'; @@ -72,6 +80,21 @@ function makeRule(overrides: Partial = {}): Notific const remoteA = { id: 10, name: 'remote-a', type: 'remote' as const }; const remoteB = { id: 11, name: 'remote-b', type: 'remote' as const }; +const RETRACTION_CAP = 'notification-suppression-replica-retraction'; + +function metaWith(...caps: string[]) { + return { capabilities: caps, online: true }; +} + +function okDeleteResponse(outcome = 'applied') { + return { + ok: true, + status: 200, + text: async () => '', + json: async () => ({ success: true, outcome }), + }; +} + describe('notificationSuppressionSync', () => { beforeEach(() => { vi.clearAllMocks(); @@ -83,8 +106,10 @@ describe('notificationSuppressionSync', () => { apiUrl: `http://node-${id}.example:1852`, apiToken: 'tok', })); - mockFetch.mockResolvedValue({ ok: true, status: 200, text: async () => '' }); + mockFetch.mockResolvedValue(okDeleteResponse()); mockRemoteAdvertises.mockResolvedValue(true); + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP, 'notification-suppression-schedule')); + mockListPending.mockReturnValue([]); }); afterEach(() => { @@ -100,6 +125,7 @@ describe('notificationSuppressionSync', () => { syncSuppressionRuleToFleet(makeRule({ schedule: null, node_id: 10 })); await vi.waitFor(() => expect(mockFetch).toHaveBeenCalled()); expect(mockRemoteAdvertises).not.toHaveBeenCalled(); + expect(mockFetchMeta).not.toHaveBeenCalled(); const body = JSON.parse((mockFetch.mock.calls[0][1] as { body: string }).body); expect(body.rule.node_id).toBeNull(); expect(body.rule.schedule).toBeNull(); @@ -121,44 +147,99 @@ describe('notificationSuppressionSync', () => { expect(body.rule.schedule.days).toEqual([6]); }); - it('probe false + DELETE success: no POST; cleanup logged as removed', async () => { + it('schedule unsupported + retraction supported: recoverable DELETE, no POST', async () => { const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); const error = vi.spyOn(console, 'error').mockImplementation(() => {}); mockRemoteAdvertises.mockResolvedValue(false); - mockFetch.mockResolvedValue({ ok: true, status: 200, text: async () => '' }); + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); syncSuppressionRuleToFleet(makeRule({ node_id: 10, + updated_at: 555, schedule: { days: [1], start_minute: 0, end_minute: 60, tz: 'UTC' }, })); await vi.waitFor(() => expect(mockFetch).toHaveBeenCalled()); expect(mockFetch.mock.calls.every((c) => (c[1] as { method: string }).method === 'DELETE')).toBe(true); - expect(warn.mock.calls.some((c) => String(c[0]).includes('replica was removed'))).toBe(true); + expect(JSON.parse((mockFetch.mock.calls[0][1] as { body: string }).body)).toEqual({ + kind: 'recoverable', + source_updated_at: 555, + }); + expect(warn.mock.calls.some((c) => String(c[0]).includes('recoverable DELETE applied'))).toBe(true); expect(error.mock.calls.some((c) => String(c[0]).includes('cleanup pending'))).toBe(false); + expect(mockUpsertPending).not.toHaveBeenCalled(); + expect(mockDeletePending).toHaveBeenCalledWith(42, 10); }); - it('probe false + no proxy target: no successful-cleanup claim', async () => { + it('schedule unsupported + retraction unsupported: no DELETE; queues pending', async () => { const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + mockRemoteAdvertises.mockResolvedValue(false); + mockFetchMeta.mockResolvedValue(metaWith('notification-suppression')); + + syncSuppressionRuleToFleet(makeRule({ + node_id: 10, + updated_at: 555, + schedule: { days: [1], start_minute: 0, end_minute: 60, tz: 'UTC' }, + })); + await vi.waitFor(() => expect(mockUpsertPending).toHaveBeenCalled()); + + expect(mockFetch).not.toHaveBeenCalled(); + expect(warn.mock.calls.some((c) => String(c[0]).includes('queued pending retraction'))).toBe(true); + expect(mockUpsertPending).toHaveBeenCalledWith( + expect.objectContaining({ + rule_id: 42, + node_id: 10, + kind: 'recoverable', + source_updated_at: 555, + }), + ); + }); + + it('schedule unsupported + probe unreachable (offline meta): no DELETE; queues pending', async () => { + mockRemoteAdvertises.mockResolvedValue(false); + mockFetchMeta.mockResolvedValue({ capabilities: [], online: false }); + + syncSuppressionRuleToFleet(makeRule({ + node_id: 10, + schedule: { days: [1], start_minute: 0, end_minute: 60, tz: 'UTC' }, + })); + await vi.waitFor(() => expect(mockUpsertPending).toHaveBeenCalled()); + expect(mockFetch).not.toHaveBeenCalled(); + expect(mockUpsertPending.mock.calls[0][0].last_error).toMatch(/unreachable/); + }); + + it('schedule unsupported + probe unreachable (throw): no DELETE; queues pending', async () => { + mockRemoteAdvertises.mockResolvedValue(false); + mockFetchMeta.mockRejectedValue(new Error('timeout')); + + syncSuppressionRuleToFleet(makeRule({ + node_id: 10, + schedule: { days: [1], start_minute: 0, end_minute: 60, tz: 'UTC' }, + })); + await vi.waitFor(() => expect(mockUpsertPending).toHaveBeenCalled()); + expect(mockFetch).not.toHaveBeenCalled(); + }); + + it('schedule unsupported + no proxy after supported probe: queues pending, no POST', async () => { const error = vi.spyOn(console, 'error').mockImplementation(() => {}); mockRemoteAdvertises.mockResolvedValue(false); + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); mockGetProxyTarget.mockReturnValue(null); syncSuppressionRuleToFleet(makeRule({ node_id: 10, schedule: { days: [1], start_minute: 0, end_minute: 60, tz: 'UTC' }, })); - await vi.waitFor(() => expect(error).toHaveBeenCalled()); + await vi.waitFor(() => expect(mockUpsertPending).toHaveBeenCalled()); expect(mockFetch).not.toHaveBeenCalled(); - expect(warn.mock.calls.some((c) => String(c[0]).includes('replica was removed'))).toBe(false); expect(error.mock.calls.some((c) => String(c[0]).includes('cleanup pending'))).toBe(true); }); - it('scheduleInvalid: DELETE success, no POST', async () => { + it('scheduleInvalid: DELETE success when retraction supported, no POST', async () => { const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); const error = vi.spyOn(console, 'error').mockImplementation(() => {}); - mockFetch.mockResolvedValue({ ok: true, status: 200, text: async () => '' }); + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); syncSuppressionRuleToFleet(makeRule({ node_id: 10, @@ -166,31 +247,44 @@ describe('notificationSuppressionSync', () => { scheduleInvalid: true, })); await vi.waitFor(() => { - expect(warn.mock.calls.some((c) => String(c[0]).includes('replica removed'))).toBe(true); + expect(warn.mock.calls.some((c) => String(c[0]).includes('recoverable DELETE applied'))).toBe(true); }); expect(mockFetch.mock.calls.every((c) => (c[1] as { method: string }).method === 'DELETE')).toBe(true); expect(error.mock.calls.some((c) => String(c[0]).includes('cleanup pending'))).toBe(false); }); - it('scheduleInvalid: DELETE 404 counts as cleanup success, no POST', async () => { - const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); - mockFetch.mockResolvedValue({ ok: false, status: 404, text: async () => 'gone' }); - - syncSuppressionRuleToFleet(makeRule({ - node_id: 10, - schedule: null, - scheduleInvalid: true, - })); - await vi.waitFor(() => { - expect(warn.mock.calls.some((c) => String(c[0]).includes('replica removed'))).toBe(true); + it('scheduleInvalid: opaque DELETE 404 queues pending (not treated as success)', async () => { + const error = vi.spyOn(console, 'error').mockImplementation(() => {}); + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); + mockFetch.mockResolvedValue({ + ok: false, + status: 404, + text: async () => 'Not Found', + json: async () => { + throw new Error('no json'); + }, }); - expect(mockFetch.mock.calls.every((c) => (c[1] as { method: string }).method === 'DELETE')).toBe(true); + + syncSuppressionRuleToFleet(makeRule({ + node_id: 10, + schedule: null, + scheduleInvalid: true, + })); + await vi.waitFor(() => expect(mockUpsertPending).toHaveBeenCalled()); + expect(error.mock.calls.some((c) => String(c[0]).includes('cleanup pending'))).toBe(true); + expect(mockDeletePending).not.toHaveBeenCalled(); }); - it('scheduleInvalid: DELETE failure logs pending cleanup, no POST', async () => { + it('scheduleInvalid: DELETE failure queues pending, no POST', async () => { const error = vi.spyOn(console, 'error').mockImplementation(() => {}); - mockFetch.mockResolvedValue({ ok: false, status: 503, text: async () => 'down' }); + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); + mockFetch.mockResolvedValue({ + ok: false, + status: 503, + text: async () => 'down', + json: async () => ({}), + }); syncSuppressionRuleToFleet(makeRule({ node_id: 10, @@ -199,28 +293,32 @@ describe('notificationSuppressionSync', () => { })); await vi.waitFor(() => expect(error).toHaveBeenCalled()); expect(mockFetch.mock.calls.every((c) => (c[1] as { method: string }).method === 'DELETE')).toBe(true); + expect(mockUpsertPending).toHaveBeenCalled(); expect(error.mock.calls.some((c) => String(c[0]).includes('cleanup pending'))).toBe(true); }); - it('scheduleInvalid: no proxy target logs pending, no POST', async () => { - const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); - const error = vi.spyOn(console, 'error').mockImplementation(() => {}); - mockGetProxyTarget.mockReturnValue(null); + it('scheduleInvalid without retraction capability: no DELETE; queues pending', async () => { + mockFetchMeta.mockResolvedValue(metaWith()); syncSuppressionRuleToFleet(makeRule({ node_id: 10, + updated_at: 777, schedule: null, scheduleInvalid: true, })); - await vi.waitFor(() => expect(error).toHaveBeenCalled()); + await vi.waitFor(() => expect(mockUpsertPending).toHaveBeenCalled()); expect(mockFetch).not.toHaveBeenCalled(); - expect(warn.mock.calls.some((c) => String(c[0]).includes('replica removed'))).toBe(false); - expect(error.mock.calls.some((c) => String(c[0]).includes('cleanup pending'))).toBe(true); + expect(mockUpsertPending).toHaveBeenCalledWith( + expect.objectContaining({ + kind: 'recoverable', + source_updated_at: 777, + }), + ); }); - it('unscheduled-to-scheduled on unsupported target attempts DELETE', async () => { + it('unscheduled-to-scheduled on unsupported schedule target attempts recoverable path', async () => { mockRemoteAdvertises.mockResolvedValue(false); - mockFetch.mockResolvedValue({ ok: true, status: 200, text: async () => '' }); + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); const previous = makeRule({ node_id: 10, schedule: null }); const updated = makeRule({ node_id: 10, @@ -231,22 +329,6 @@ describe('notificationSuppressionSync', () => { expect(mockFetch.mock.calls.every((c) => (c[1] as { method: string }).method === 'DELETE')).toBe(true); }); - it('probe false + DELETE failure: no POST; logs cleanup pending', async () => { - const error = vi.spyOn(console, 'error').mockImplementation(() => {}); - mockRemoteAdvertises.mockResolvedValue(false); - mockFetch.mockResolvedValue({ ok: false, status: 503, text: async () => 'down' }); - - syncSuppressionRuleToFleet(makeRule({ - node_id: 10, - schedule: { days: [1], start_minute: 0, end_minute: 60, tz: 'UTC' }, - })); - await vi.waitFor(() => expect(error).toHaveBeenCalled()); - - expect(mockFetch.mock.calls.every((c) => (c[1] as { method: string }).method === 'DELETE')).toBe(true); - expect(error.mock.calls.some((c) => String(c[0]).includes('cleanup pending'))).toBe(true); - expect(error.mock.calls.some((c) => String(c[0]).includes('rule 42'))).toBe(true); - }); - it('scheduled-to-unscheduled POST refresh does not require capability', async () => { mockRemoteAdvertises.mockResolvedValue(false); const previous = makeRule({ @@ -261,7 +343,8 @@ describe('notificationSuppressionSync', () => { expect(body.rule.schedule).toBeNull(); }); - it('stale targets receive DELETE on scope change', async () => { + it('stale targets receive recoverable DELETE when retraction supported', async () => { + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); const previous = makeRule({ node_id: null, schedule: null }); const updated = makeRule({ node_id: 10, schedule: null }); syncSuppressionRuleUpdateToFleet(previous, updated); @@ -272,4 +355,147 @@ describe('notificationSuppressionSync', () => { const posts = mockFetch.mock.calls.filter((c) => (c[1] as { method: string }).method === 'POST'); expect(posts.some((c) => String(c[0]).includes('node-10'))).toBe(true); }); + + it('stale-target without retraction capability: no DELETE; queues pending', async () => { + mockFetchMeta.mockResolvedValue(metaWith()); + const previous = makeRule({ id: 42, node_id: null, schedule: null, updated_at: 10 }); + const updated = makeRule({ id: 42, node_id: 10, schedule: null, updated_at: 99 }); + syncSuppressionRuleUpdateToFleet(previous, updated); + await vi.waitFor(() => expect(mockUpsertPending).toHaveBeenCalled()); + + const deletes = mockFetch.mock.calls.filter((c) => (c[1] as { method: string }).method === 'DELETE'); + expect(deletes).toHaveLength(0); + expect(mockUpsertPending).toHaveBeenCalledWith( + expect.objectContaining({ + rule_id: 42, + node_id: 11, + kind: 'recoverable', + source_updated_at: 99, + }), + ); + }); + + it('stale-target DELETE sends recoverable watermark from updated rule', async () => { + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); + const previous = makeRule({ id: 42, node_id: null, schedule: null, updated_at: 10 }); + const updated = makeRule({ id: 42, node_id: 10, schedule: null, updated_at: 99 }); + syncSuppressionRuleUpdateToFleet(previous, updated); + await vi.waitFor(() => expect(mockFetch.mock.calls.length).toBeGreaterThanOrEqual(2)); + const deletes = mockFetch.mock.calls.filter((c) => (c[1] as { method: string }).method === 'DELETE'); + const stale = deletes.find((c) => String(c[0]).includes('node-11')); + expect(stale).toBeTruthy(); + expect(JSON.parse((stale![1] as { body: string }).body)).toEqual({ + kind: 'recoverable', + source_updated_at: 99, + }); + }); + + it('authoritative fleet delete fans permanent retraction to all remotes', async () => { + const { deleteSuppressionRuleFromFleet } = await import('../helpers/notificationSuppressionSync'); + deleteSuppressionRuleFromFleet(makeRule({ node_id: 10, updated_at: 321 })); + await vi.waitFor(() => expect(mockFetch.mock.calls.length).toBe(2)); + + for (const call of mockFetch.mock.calls) { + const [, init] = call as [string, { method: string; body: string }]; + expect(init.method).toBe('DELETE'); + expect(JSON.parse(init.body)).toEqual({ kind: 'permanent', source_updated_at: 321 }); + } + expect(mockFetch.mock.calls.some((c) => String(c[0]).includes('node-10'))).toBe(true); + expect(mockFetch.mock.calls.some((c) => String(c[0]).includes('node-11'))).toBe(true); + }); + + it('authoritative delete transport failure queues pending permanent row', async () => { + const { deleteSuppressionRuleFromFleet } = await import('../helpers/notificationSuppressionSync'); + mockFetch.mockResolvedValue({ + ok: false, + status: 503, + text: async () => 'down', + json: async () => ({}), + }); + deleteSuppressionRuleFromFleet(makeRule({ node_id: 10, updated_at: 321 })); + await vi.waitFor(() => expect(mockUpsertPending).toHaveBeenCalled()); + expect(mockUpsertPending).toHaveBeenCalledWith( + expect.objectContaining({ + kind: 'permanent', + source_updated_at: 321, + }), + ); + }); + + it('ignored_stale DELETE keeps pending and does not clear', async () => { + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); + mockFetch.mockResolvedValue({ + ok: true, + status: 200, + text: async () => '', + json: async () => ({ success: true, outcome: 'ignored_stale' }), + }); + + syncSuppressionRuleToFleet(makeRule({ + node_id: 10, + schedule: null, + scheduleInvalid: true, + })); + await vi.waitFor(() => expect(mockUpsertPending).toHaveBeenCalled()); + expect(mockDeletePending).not.toHaveBeenCalled(); + expect(mockUpsertPending).toHaveBeenCalledWith( + expect.objectContaining({ + last_error: expect.stringMatching(/ignored_stale/), + }), + ); + }); + + it('flushPendingSuppressionRetractions retries recoverable only when supported', async () => { + mockListPending.mockReturnValue([ + { + rule_id: 7, + node_id: 10, + kind: 'recoverable', + source_updated_at: 50, + created_at: 1, + updated_at: 2, + attempts: 1, + last_error: 'earlier', + }, + ]); + mockFetchMeta.mockResolvedValue(metaWith()); + + await flushPendingSuppressionRetractions(10); + expect(mockFetch).not.toHaveBeenCalled(); + expect(mockUpsertPending).toHaveBeenCalled(); + + mockUpsertPending.mockClear(); + mockFetchMeta.mockResolvedValue(metaWith(RETRACTION_CAP)); + await flushPendingSuppressionRetractions(10); + await vi.waitFor(() => expect(mockFetch).toHaveBeenCalled()); + expect(JSON.parse((mockFetch.mock.calls[0][1] as { body: string }).body)).toEqual({ + kind: 'recoverable', + source_updated_at: 50, + }); + expect(mockDeletePending).toHaveBeenCalledWith(7, 10); + }); + + it('flushPendingSuppressionRetractions sends permanent without retraction capability', async () => { + mockListPending.mockReturnValue([ + { + rule_id: 8, + node_id: 11, + kind: 'permanent', + source_updated_at: 90, + created_at: 1, + updated_at: 2, + attempts: 2, + last_error: 'offline', + }, + ]); + mockFetchMeta.mockResolvedValue(metaWith()); + + await flushPendingSuppressionRetractions(11); + await vi.waitFor(() => expect(mockFetch).toHaveBeenCalled()); + expect(JSON.parse((mockFetch.mock.calls[0][1] as { body: string }).body)).toEqual({ + kind: 'permanent', + source_updated_at: 90, + }); + expect(mockDeletePending).toHaveBeenCalledWith(8, 11); + }); }); diff --git a/backend/src/__tests__/notification-suppression-tombstone-migration.test.ts b/backend/src/__tests__/notification-suppression-tombstone-migration.test.ts new file mode 100644 index 00000000..eb39ab8e --- /dev/null +++ b/backend/src/__tests__/notification-suppression-tombstone-migration.test.ts @@ -0,0 +1,159 @@ +/** + * Additive kind + source_updated_at on notification_suppression_rule_tombstones. + */ +import { describe, it, expect, afterEach } from 'vitest'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; +import Database from 'better-sqlite3'; +import { DatabaseService } from '../services/DatabaseService'; + +function resetDatabaseSingleton(): void { + const holder = DatabaseService as unknown as { instance?: DatabaseService }; + const existing = holder.instance; + if (existing) { + try { + existing.getDb().close(); + } catch { + // already closed + } + holder.instance = undefined; + } +} + +describe('notification suppression tombstone migration', () => { + let scratchDir: string | null = null; + let prevDataDir: string | undefined; + + afterEach(() => { + resetDatabaseSingleton(); + if (prevDataDir === undefined) { + delete process.env.DATA_DIR; + } else { + process.env.DATA_DIR = prevDataDir; + } + if (scratchDir) { + try { + fs.rmSync(scratchDir, { recursive: true, force: true }); + } catch { + // best-effort + } + scratchDir = null; + } + }); + + it('adds kind and source_updated_at; legacy rows stay permanent', { timeout: 60_000 }, () => { + scratchDir = fs.mkdtempSync(path.join(os.tmpdir(), 'sencho-supp-tomb-mig-')); + const dbPath = path.join(scratchDir, 'sencho.db'); + const seed = new Database(dbPath); + try { + seed.exec(` + CREATE TABLE notification_suppression_rules ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + node_id INTEGER NULL, + stack_patterns TEXT NOT NULL, + label_ids TEXT NULL, + categories TEXT NULL, + levels TEXT NULL, + applies_to TEXT NOT NULL, + enabled INTEGER DEFAULT 1, + expires_at INTEGER NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + CREATE TABLE notification_suppression_rule_tombstones ( + id INTEGER PRIMARY KEY, + deleted_at INTEGER NOT NULL + ); + INSERT INTO notification_suppression_rule_tombstones (id, deleted_at) + VALUES (42, 1700000000000); + `); + } finally { + seed.close(); + } + + prevDataDir = process.env.DATA_DIR; + process.env.DATA_DIR = scratchDir; + resetDatabaseSingleton(); + const db = DatabaseService.getInstance(); + + const cols = db.getDb().prepare('PRAGMA table_info(notification_suppression_rule_tombstones)').all() as Array<{ + name: string; + }>; + const names = cols.map((c) => c.name); + expect(names).toContain('kind'); + expect(names).toContain('source_updated_at'); + + const tomb = db.getNotificationSuppressionRuleTombstone(42); + expect(tomb?.kind).toBe('permanent'); + expect(tomb?.source_updated_at).toBe(1700000000000); + + // Legacy permanent cannot be cleared by a newer hub POST. + db.upsertNotificationSuppressionRuleReplica({ + id: 42, + name: 'should-stay-gone', + applies_to: 'both', + stack_patterns: [], + node_id: null, + label_ids: null, + categories: null, + levels: null, + enabled: true, + expires_at: null, + schedule: null, + scheduleInvalid: false, + created_at: 1, + updated_at: 9_999_999_999_999, + }); + expect(db.getNotificationSuppressionRule(42)).toBeUndefined(); + expect(db.getNotificationSuppressionRuleTombstone(42)?.kind).toBe('permanent'); + }); + + it('creates pending retractions table and merges permanent over recoverable', { timeout: 60_000 }, () => { + scratchDir = fs.mkdtempSync(path.join(os.tmpdir(), 'sencho-supp-pending-')); + prevDataDir = process.env.DATA_DIR; + process.env.DATA_DIR = scratchDir; + resetDatabaseSingleton(); + const db = DatabaseService.getInstance(); + + const cols = db.getDb().prepare('PRAGMA table_info(notification_suppression_pending_retractions)').all() as Array<{ + name: string; + }>; + expect(cols.map((c) => c.name)).toEqual( + expect.arrayContaining([ + 'rule_id', + 'node_id', + 'kind', + 'source_updated_at', + 'attempts', + 'last_error', + ]), + ); + + db.upsertNotificationSuppressionPendingRetraction({ + rule_id: 7, + node_id: 3, + kind: 'recoverable', + source_updated_at: 100, + last_error: 'unsupported', + }); + db.upsertNotificationSuppressionPendingRetraction({ + rule_id: 7, + node_id: 3, + kind: 'permanent', + source_updated_at: 50, + last_error: 'offline', + }); + + const rows = db.listNotificationSuppressionPendingRetractions(3); + expect(rows).toHaveLength(1); + expect(rows[0].kind).toBe('permanent'); + expect(rows[0].source_updated_at).toBe(100); + expect(rows[0].attempts).toBe(2); + expect(rows[0].last_error).toBe('offline'); + + db.deleteNotificationSuppressionPendingRetraction(7, 3); + expect(db.listNotificationSuppressionPendingRetractions(3)).toHaveLength(0); + }); +}); diff --git a/backend/src/bootstrap/shutdown.ts b/backend/src/bootstrap/shutdown.ts index 10533ffa..fe2ba8af 100644 --- a/backend/src/bootstrap/shutdown.ts +++ b/backend/src/bootstrap/shutdown.ts @@ -6,6 +6,7 @@ import { AutoHealService } from '../services/AutoHealService'; import { HealthGateService } from '../services/HealthGateService'; import { ServiceUpdateRecoveryService } from '../services/ServiceUpdateRecoveryService'; import { FleetSyncRetryService } from '../services/FleetSyncRetryService'; +import { SuppressionRetractionRetryService } from '../services/SuppressionRetractionRetryService'; import { DockerEventManager } from '../services/DockerEventManager'; import { ImageUpdateService } from '../services/ImageUpdateService'; import { SchedulerService } from '../services/SchedulerService'; @@ -37,6 +38,9 @@ export function installShutdownHandlers(server: Server): void { try { HealthGateService.getInstance().stop(); } catch (e) { console.warn('[Shutdown] HealthGateService cleanup failed:', (e as Error).message); } try { ServiceUpdateRecoveryService.getInstance().stop(); } catch (e) { console.warn('[Shutdown] ServiceUpdateRecoveryService cleanup failed:', (e as Error).message); } try { FleetSyncRetryService.getInstance().stop(); } catch (e) { console.warn('[Shutdown] FleetSyncRetryService cleanup failed:', (e as Error).message); } + try { SuppressionRetractionRetryService.getInstance().stop(); } catch (e) { + console.warn('[Shutdown] SuppressionRetractionRetryService cleanup failed:', (e as Error).message); + } try { DockerEventManager.getInstance().stop(); } catch (e) { console.warn('[Shutdown] DockerEventManager cleanup failed:', (e as Error).message); } diff --git a/backend/src/bootstrap/startup.ts b/backend/src/bootstrap/startup.ts index f3415e6c..5b6bda80 100644 --- a/backend/src/bootstrap/startup.ts +++ b/backend/src/bootstrap/startup.ts @@ -15,6 +15,7 @@ import { ServiceUpdateRecoveryService } from '../services/ServiceUpdateRecoveryS import { DeployedStackDeletionService } from '../services/DeployedStackDeletionService'; import { StackUpdateRecoveryService } from '../services/StackUpdateRecoveryService'; import { FleetSyncRetryService } from '../services/FleetSyncRetryService'; +import { SuppressionRetractionRetryService } from '../services/SuppressionRetractionRetryService'; import { DockerEventManager } from '../services/DockerEventManager'; import TrivyService, { sweepStaleTrivyTempDirs } from '../services/TrivyService'; import { ImageUpdateService } from '../services/ImageUpdateService'; @@ -154,6 +155,7 @@ export async function startServer(server: Server): Promise { HealthGateService.getInstance().start(); ServiceUpdateRecoveryService.getInstance().start(); FleetSyncRetryService.getInstance().start(); + SuppressionRetractionRetryService.getInstance().start(); ImageUpdateService.getInstance().start(); SchedulerService.getInstance().start(); MfaService.getInstance().start(); @@ -166,7 +168,11 @@ export async function startServer(server: Server): Promise { // Drop the cached /api/meta entry on tunnel reconnect so the next // /api/nodes/:id/meta refetches fresh capabilities and version through // the live loopback bridge instead of waiting for the 3-minute TTL. - PilotTunnelManager.getInstance().on('tunnel-up', invalidateRemoteMetaCache); + // Also flush durable mute-replica retractions that waited for this node. + PilotTunnelManager.getInstance().on('tunnel-up', (nodeId: number) => { + invalidateRemoteMetaCache(nodeId); + void SuppressionRetractionRetryService.getInstance().flushNode(nodeId); + }); // Most async initializers still run in parallel. Docker event monitoring // is sequenced after self identity so it never classifies Sencho's own diff --git a/backend/src/helpers/notificationSuppressionSync.ts b/backend/src/helpers/notificationSuppressionSync.ts index a5f641c8..922dc526 100644 --- a/backend/src/helpers/notificationSuppressionSync.ts +++ b/backend/src/helpers/notificationSuppressionSync.ts @@ -1,9 +1,10 @@ -import { DatabaseService, type NotificationSuppressionRule, type Node } from '../services/DatabaseService'; +import { DatabaseService, type NotificationSuppressionRule, type NotificationSuppressionRetraction, type NotificationSuppressionRetractionKind, type Node } from '../services/DatabaseService'; import { NodeRegistry } from '../services/NodeRegistry'; import { LicenseService } from '../services/LicenseService'; import { PROXY_TIER_HEADER } from '../services/license-headers'; import { NOTIFICATION_SUPPRESSION_SCHEDULE_CAPABILITY, + NOTIFICATION_SUPPRESSION_REPLICA_RETRACTION_CAPABILITY, } from '../services/CapabilityRegistry'; import { remoteAdvertisesCapability } from './remoteCapabilities'; import { getErrorMessage } from '../utils/errors'; @@ -20,6 +21,13 @@ function buildRemoteHeaders(apiToken: string): Record { return headers; } +function retractionFor( + kind: NotificationSuppressionRetractionKind, + rule: Pick, +): NotificationSuppressionRetraction { + return { kind, source_updated_at: rule.updated_at }; +} + /** Hub node ids that should receive a replica for this rule (before wire identity normalize). */ export function replicationTargetIds(rule: NotificationSuppressionRule): number[] { return replicationTargets(rule).map((n) => n.id); @@ -35,46 +43,171 @@ export function replicationTargets(rule: NotificationSuppressionRule): Node[] { return remotes; } +/** Every registered remote: permanent deletes must reach historical replicas, not only current scope. */ +export function allRemoteNodes(): Node[] { + return DatabaseService.getInstance().getNodes().filter((n) => n.type === 'remote'); +} + function replicaPayload(rule: NotificationSuppressionRule): NotificationSuppressionRule { return { ...rule, node_id: null }; } -async function pushRuleToNode(node: Node, rule: NotificationSuppressionRule): Promise { - const target = NodeRegistry.getInstance().getProxyTarget(node.id); - if (!target?.apiUrl) { - console.warn(`[SuppressionSync] Skipping node "${node.name}": no proxy target`); - return; - } - const baseUrl = target.apiUrl.replace(/\/$/, ''); - const res = await fetch(`${baseUrl}/api/notification-suppression-rules/replica`, { - method: 'POST', - headers: buildRemoteHeaders(target.apiToken), - body: JSON.stringify({ rule: replicaPayload(rule) }), - signal: AbortSignal.timeout(SYNC_TIMEOUT_MS), - }); - if (!res.ok) { - const body = await res.text().catch(() => ''); - throw new Error(`HTTP ${res.status}${body ? `: ${body.slice(0, 200)}` : ''}`); +type RetractionSupport = 'supported' | 'unsupported' | 'unreachable'; + +async function probeRetractionSupport(nodeId: number): Promise { + try { + const meta = await NodeRegistry.getInstance().fetchMetaForNode(nodeId); + // fetchMetaForNode returns OFFLINE_META (online:false, capabilities:[]) on + // transport failure instead of throwing; treat that as unreachable. + if (meta.online === false) return 'unreachable'; + return meta.capabilities.includes(NOTIFICATION_SUPPRESSION_REPLICA_RETRACTION_CAPABILITY) + ? 'supported' + : 'unsupported'; + } catch (err) { + console.warn( + `[SuppressionSync] Retraction capability probe failed for node ${nodeId}; treating as unreachable:`, + getErrorMessage(err, 'unknown'), + ); + return 'unreachable'; } } -async function deleteRuleOnNode(node: Node, ruleId: number): Promise { +function enqueuePending( + node: Node, + ruleId: number, + retraction: NotificationSuppressionRetraction, + lastError: string, +): void { + DatabaseService.getInstance().upsertNotificationSuppressionPendingRetraction({ + rule_id: ruleId, + node_id: node.id, + kind: retraction.kind, + source_updated_at: retraction.source_updated_at, + last_error: lastError, + }); +} + +function clearPending(nodeId: number, ruleId: number): void { + DatabaseService.getInstance().deleteNotificationSuppressionPendingRetraction(ruleId, nodeId); +} + +function resolveRemoteApi(node: Node): { baseUrl: string; apiToken: string } | null { const target = NodeRegistry.getInstance().getProxyTarget(node.id); - if (!target?.apiUrl) { - throw new Error(`no proxy target for node "${node.name}" (id=${node.id})`); + if (!target?.apiUrl) return null; + return { baseUrl: target.apiUrl.replace(/\/$/, ''), apiToken: target.apiToken }; +} + +/** Non-2xx (including opaque 404) is always failure; never treat missing routes as applied. */ +async function throwHttpFailure(res: Response): Promise { + const body = await res.text().catch(() => ''); + throw new Error(`HTTP ${res.status}${body ? `: ${body.slice(0, 200)}` : ''}`); +} + +/** Prefer JSON outcome; fall back to applied for pre-outcome remotes (bare 2xx). */ +async function readOutcome(res: Response): Promise { + try { + const json = (await res.json()) as { outcome?: string }; + return typeof json.outcome === 'string' ? json.outcome : 'applied'; + } catch { + return 'applied'; } - const baseUrl = target.apiUrl.replace(/\/$/, ''); - const res = await fetch(`${baseUrl}/api/notification-suppression-rules/replica/${ruleId}`, { - method: 'DELETE', - headers: buildRemoteHeaders(target.apiToken), +} + +async function pushRuleToNode(node: Node, rule: NotificationSuppressionRule): Promise { + const remote = resolveRemoteApi(node); + if (!remote) { + console.warn(`[SuppressionSync] Skipping node "${node.name}": no proxy target`); + return; + } + const res = await fetch(`${remote.baseUrl}/api/notification-suppression-rules/replica`, { + method: 'POST', + headers: buildRemoteHeaders(remote.apiToken), + body: JSON.stringify({ rule: replicaPayload(rule) }), signal: AbortSignal.timeout(SYNC_TIMEOUT_MS), }); - if (!res.ok && res.status !== 404) { - const body = await res.text().catch(() => ''); - throw new Error(`HTTP ${res.status}${body ? `: ${body.slice(0, 200)}` : ''}`); + if (!res.ok) await throwHttpFailure(res); + const outcome = await readOutcome(res); + if (outcome !== 'applied') { + throw new Error(`replica POST outcome=${outcome}`); } } +/** + * Deliver a replica DELETE. On transport failure, durable-queues the retraction. + * Clears pending only when the remote reports outcome=applied (not ignored_stale). + */ +export async function deleteRuleOnNode( + node: Node, + ruleId: number, + retraction: NotificationSuppressionRetraction, +): Promise<{ outcome: string }> { + const remote = resolveRemoteApi(node); + if (!remote) { + const err = `no proxy target for node "${node.name}" (id=${node.id})`; + enqueuePending(node, ruleId, retraction, err); + throw new Error(err); + } + try { + const res = await fetch(`${remote.baseUrl}/api/notification-suppression-rules/replica/${ruleId}`, { + method: 'DELETE', + headers: buildRemoteHeaders(remote.apiToken), + body: JSON.stringify(retraction), + signal: AbortSignal.timeout(SYNC_TIMEOUT_MS), + }); + if (!res.ok) await throwHttpFailure(res); + const outcome = await readOutcome(res); + if (outcome !== 'applied') { + // ignored_stale and any other non-applied outcome keep the outbox row. + const reason = + outcome === 'ignored_stale' + ? 'remote ignored_stale; rule still present' + : `remote DELETE outcome=${outcome}`; + enqueuePending(node, ruleId, retraction, reason); + return { outcome }; + } + clearPending(node.id, ruleId); + return { outcome }; + } catch (err) { + try { + enqueuePending(node, ruleId, retraction, getErrorMessage(err, String(err))); + } catch (enqueueErr) { + console.error( + `[SuppressionSync] Failed to durable-queue retract rule=${ruleId} node=${node.id}:`, + getErrorMessage(enqueueErr, String(enqueueErr)), + ); + } + throw err; + } +} + +/** + * Recoverable DELETE only when the remote advertises versioned retractions. + * Otherwise enqueue pending and do not bare-delete on incompatible remotes. + * @returns applied when DELETE was accepted; deferred when queued without sending + * or when the remote ignored a stale watermark. + */ +async function deliverRecoverableDelete( + node: Node, + ruleId: number, + retraction: NotificationSuppressionRetraction, + context: string, +): Promise<'applied' | 'deferred'> { + const support = await probeRetractionSupport(node.id); + if (support !== 'supported') { + const reason = + support === 'unsupported' + ? 'remote lacks notification-suppression-replica-retraction; not sending recoverable DELETE' + : 'remote unreachable for retraction capability probe; not sending recoverable DELETE'; + enqueuePending(node, ruleId, retraction, reason); + console.warn( + `[SuppressionSync] ${context} on node "${node.name}" (id=${node.id}): ${reason}; queued pending retraction`, + ); + return 'deferred'; + } + const { outcome } = await deleteRuleOnNode(node, ruleId, retraction); + return outcome === 'applied' ? 'applied' : 'deferred'; +} + async function pushOrCleanupScheduled(node: Node, rule: NotificationSuppressionRule): Promise { const supportsSchedule = await remoteAdvertisesCapability( node.id, @@ -84,36 +217,46 @@ async function pushOrCleanupScheduled(node: Node, rule: NotificationSuppressionR await pushRuleToNode(node, rule); return; } - // Probe false means unsupported OR unreachable. Never POST a scheduled rule - // through the legacy contract (older remotes would mute all day). Attempt DELETE; - // only claim cleanup when DELETE succeeds. + // Never POST a scheduled rule through the legacy contract. Soft-cleanup DELETE + // requires versioned retraction support so incompatible remotes are not bare-deleted. try { - await deleteRuleOnNode(node, rule.id); - console.warn( - `[SuppressionSync] Scheduled rule ${rule.id} not applied on node "${node.name}" (id=${node.id}): ` + - `capability unsupported-or-unreachable; DELETE succeeded and replica was removed`, + const result = await deliverRecoverableDelete( + node, + rule.id, + retractionFor('recoverable', rule), + `Scheduled rule ${rule.id} not applied`, ); + if (result === 'applied') { + console.warn( + `[SuppressionSync] Scheduled rule ${rule.id} not applied on node "${node.name}" (id=${node.id}): ` + + `schedule unsupported-or-unreachable; recoverable DELETE applied`, + ); + } } catch (err) { console.error( `[SuppressionSync] Scheduled rule ${rule.id}: cleanup pending on node "${node.name}" (id=${node.id}); ` + - `capability unsupported-or-unreachable and DELETE failed (${getErrorMessage(err, String(err))}). ` + - `Prior replica may remain until connectivity returns and the rule is re-saved`, + `recoverable DELETE failed (${getErrorMessage(err, String(err))})`, ); } } async function cleanupInvalidScheduleReplica(node: Node, rule: NotificationSuppressionRule): Promise { - // Never POST an invalid schedule (would mute all day on remotes that ignore the field). - // Attempt DELETE so a prior valid/unscheduled replica cannot keep muting. try { - await deleteRuleOnNode(node, rule.id); - console.warn( - `[SuppressionSync] Corrupt schedule on rule ${rule.id}: replica removed on node "${node.name}" (id=${node.id}); not posting`, + const result = await deliverRecoverableDelete( + node, + rule.id, + retractionFor('recoverable', rule), + `Corrupt schedule on rule ${rule.id}`, ); + if (result === 'applied') { + console.warn( + `[SuppressionSync] Corrupt schedule on rule ${rule.id}: recoverable DELETE applied on node "${node.name}" (id=${node.id}); not posting`, + ); + } } catch (err) { console.error( `[SuppressionSync] Corrupt schedule on rule ${rule.id}: cleanup pending on node "${node.name}" (id=${node.id}); ` + - `DELETE failed (${getErrorMessage(err, String(err))}). Prior replica may remain until connectivity returns`, + `DELETE failed (${getErrorMessage(err, String(err))})`, ); } } @@ -160,13 +303,19 @@ export function syncSuppressionRuleUpdateToFleet( const newIds = new Set(replicationTargetIds(updated)); const db = DatabaseService.getInstance(); const staleIds = [...oldIds].filter((id) => !newIds.has(id)); + const staleRetraction = retractionFor('recoverable', updated); void Promise.allSettled([ ...staleIds.map(async (id) => { const node = db.getNode(id); if (!node || node.type !== 'remote') return; try { - await deleteRuleOnNode(node, previous.id); + await deliverRecoverableDelete( + node, + previous.id, + staleRetraction, + `Stale-target retract for rule ${previous.id}`, + ); } catch (err) { console.error( `[SuppressionSync] Failed to delete stale rule ${previous.id} on node "${node.name}":`, @@ -187,14 +336,18 @@ export function syncSuppressionRuleUpdateToFleet( ]); } -/** Best-effort delete of a replicated rule on fleet nodes. */ +/** + * Authoritative delete: fan out permanent retraction to every known remote + * (not only current scope), and durable-queue failures for retry. + */ export function deleteSuppressionRuleFromFleet(rule: NotificationSuppressionRule): void { - const targets = replicationTargets(rule); + const targets = allRemoteNodes(); if (targets.length === 0) return; + const retraction = retractionFor('permanent', rule); void Promise.allSettled( targets.map(async (node) => { try { - await deleteRuleOnNode(node, rule.id); + await deleteRuleOnNode(node, rule.id, retraction); } catch (err) { console.error( `[SuppressionSync] Failed to delete rule ${rule.id} on node "${node.name}":`, @@ -204,3 +357,40 @@ export function deleteSuppressionRuleFromFleet(rule: NotificationSuppressionRule }), ); } + +/** + * Retry durable pending retractions for one node (tunnel-up / reconnect) or all. + * Recoverable rows still require the remote to advertise retraction support. + */ +export async function flushPendingSuppressionRetractions(nodeId?: number): Promise { + const db = DatabaseService.getInstance(); + const pending = db.listNotificationSuppressionPendingRetractions(nodeId); + for (const row of pending) { + const node = db.getNode(row.node_id); + if (!node || node.type !== 'remote') { + db.deleteNotificationSuppressionPendingRetraction(row.rule_id, row.node_id); + continue; + } + const retraction: NotificationSuppressionRetraction = { + kind: row.kind, + source_updated_at: row.source_updated_at, + }; + try { + if (retraction.kind === 'recoverable') { + await deliverRecoverableDelete( + node, + row.rule_id, + retraction, + `Pending recoverable retract for rule ${row.rule_id}`, + ); + } else { + await deleteRuleOnNode(node, row.rule_id, retraction); + } + } catch (err) { + console.error( + `[SuppressionSync] Pending retract retry failed rule=${row.rule_id} node=${node.name}:`, + getErrorMessage(err, String(err)), + ); + } + } +} diff --git a/backend/src/routes/notifications.ts b/backend/src/routes/notifications.ts index d253dcbc..fa61b0a6 100644 --- a/backend/src/routes/notifications.ts +++ b/backend/src/routes/notifications.ts @@ -1,5 +1,10 @@ import { Router, type Request, type Response } from 'express'; -import { DatabaseService, type NotificationSuppressionAppliesTo, type NotificationSuppressionRule } from '../services/DatabaseService'; +import { + DatabaseService, + type NotificationSuppressionAppliesTo, + type NotificationSuppressionRetraction, + type NotificationSuppressionRule, +} from '../services/DatabaseService'; import { NotificationService, ALL_NOTIFICATION_CATEGORIES, ALL_SUPPRESSIBLE_CATEGORIES } from '../services/NotificationService'; import type { NotificationCategory } from '../services/NotificationService'; import { NodeRegistry } from '../services/NodeRegistry'; @@ -37,6 +42,42 @@ const VALID_SUPPRESSION_CATEGORIES: ReadonlySet = new Set( const VALID_LEVELS = new Set(['info', 'warning', 'error']); const VALID_APPLIES_TO = new Set(['bell', 'external', 'both']); +function isNonNegativeSafeInteger(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} + +/** + * Omitted / empty body (old hub) -> permanent watermark 0. + * Present body with any keys must be a complete valid retraction or 400. + */ +function parseReplicaRetractionBody( + body: unknown, + res: Response, +): NotificationSuppressionRetraction | false { + if ( + body == null || + (typeof body === 'object' && !Array.isArray(body) && Object.keys(body as object).length === 0) + ) { + return { kind: 'permanent', source_updated_at: 0 }; + } + if (typeof body !== 'object' || Array.isArray(body)) { + res.status(400).json({ error: 'retraction body must be an object' }); + return false; + } + const raw = body as Record; + const kind = raw.kind; + const source = raw.source_updated_at; + if (kind !== 'permanent' && kind !== 'recoverable') { + res.status(400).json({ error: 'kind must be permanent or recoverable' }); + return false; + } + if (!isNonNegativeSafeInteger(source)) { + res.status(400).json({ error: 'source_updated_at must be a non-negative safe integer' }); + return false; + } + return { kind, source_updated_at: source }; +} + function validateNodeId(nodeId: unknown, res: Response): number | null | false { if (nodeId === undefined || nodeId === null) return null; if (typeof nodeId !== 'number' || !Number.isInteger(nodeId)) { @@ -550,8 +591,16 @@ notificationSuppressionRouter.post('/replica', authMiddleware, (req: Request, re isCreate: true, }, res); if (scheduleResult === false) return; + if (!isNonNegativeSafeInteger(rule.created_at)) { + res.status(400).json({ error: 'created_at must be a non-negative safe integer' }); + return; + } + if (!isNonNegativeSafeInteger(rule.updated_at)) { + res.status(400).json({ error: 'updated_at must be a non-negative safe integer' }); + return; + } - DatabaseService.getInstance().upsertNotificationSuppressionRuleReplica({ + const outcome = DatabaseService.getInstance().upsertNotificationSuppressionRuleReplica({ ...rule, stack_patterns: patterns.patterns, label_ids: Array.isArray(rule.label_ids) && rule.label_ids.length > 0 ? rule.label_ids : null, @@ -561,11 +610,13 @@ notificationSuppressionRouter.post('/replica', authMiddleware, (req: Request, re scheduleInvalid: false, enabled: rule.enabled !== false, expires_at: rule.expires_at ?? null, + created_at: rule.created_at, + updated_at: rule.updated_at, // Replicas are always node-agnostic on the receiving node: the hub's node_id // is a hub-local scoping concept and never trustworthy as a foreign key here. node_id: null, }); - res.json({ success: true }); + res.json({ success: true, outcome }); } catch (error) { console.error('Failed to apply suppression rule replica:', error); res.status(500).json({ error: 'Failed to apply suppression rule replica' }); @@ -577,8 +628,10 @@ notificationSuppressionRouter.delete('/replica/:id', authMiddleware, (req: Reque try { const id = parseIntParam(req, res, 'id', 'suppression rule ID'); if (id === null) return; - DatabaseService.getInstance().deleteNotificationSuppressionRule(id); - res.json({ success: true }); + const retraction = parseReplicaRetractionBody(req.body, res); + if (retraction === false) return; + const { outcome } = DatabaseService.getInstance().deleteNotificationSuppressionRule(id, retraction); + res.json({ success: true, outcome }); } catch (error) { console.error('Failed to delete suppression rule replica:', error); res.status(500).json({ error: 'Failed to delete suppression rule replica' }); @@ -727,7 +780,10 @@ notificationSuppressionRouter.delete('/:id', authMiddleware, (req: Request, res: const existing = DatabaseService.getInstance().getNotificationSuppressionRule(id); if (!existing) { res.status(404).json({ error: 'Suppression rule not found' }); return; } - DatabaseService.getInstance().deleteNotificationSuppressionRule(id); + DatabaseService.getInstance().deleteNotificationSuppressionRule(id, { + kind: 'permanent', + source_updated_at: existing.updated_at, + }); deleteSuppressionRuleFromFleet(existing); console.log(`[Suppression] Rule ${id} deleted`); res.json({ success: true }); diff --git a/backend/src/services/CapabilityRegistry.ts b/backend/src/services/CapabilityRegistry.ts index ba806c5d..00937323 100644 --- a/backend/src/services/CapabilityRegistry.ts +++ b/backend/src/services/CapabilityRegistry.ts @@ -37,6 +37,7 @@ export const CAPABILITIES = [ 'notification-routing', 'notification-suppression', 'notification-suppression-schedule', + 'notification-suppression-replica-retraction', 'host-console', 'host-console-community', 'container-exec', @@ -83,6 +84,14 @@ export const HOST_CONSOLE_COMMUNITY_CAPABILITY = 'host-console-community' as con export const NOTIFICATION_SUPPRESSION_SCHEDULE_CAPABILITY = 'notification-suppression-schedule' as const satisfies Capability; +/** + * Remotes that accept hub-authored `{ kind, source_updated_at }` on replica DELETE + * and persist versioned tombstones. Without this, hubs must not send recoverable + * soft-cleanup DELETEs (pre-tombstone remotes would bare-delete with no guard). + */ +export const NOTIFICATION_SUPPRESSION_REPLICA_RETRACTION_CAPABILITY = + 'notification-suppression-replica-retraction' as const satisfies Capability; + /** Capability for optional `?removeVolumes=true` on POST /stacks/:name/down. */ export const STACK_DOWN_REMOVE_VOLUMES_CAPABILITY = 'stack-down-remove-volumes' as const satisfies Capability; diff --git a/backend/src/services/DatabaseService.ts b/backend/src/services/DatabaseService.ts index 4e09b840..fe868ee7 100644 --- a/backend/src/services/DatabaseService.ts +++ b/backend/src/services/DatabaseService.ts @@ -766,6 +766,52 @@ export interface NotificationRoute { export type NotificationSuppressionAppliesTo = 'bell' | 'external' | 'both'; +/** Hub-authored replica retraction: permanent never clears; recoverable is LWW-ordered. */ +export type NotificationSuppressionRetractionKind = 'permanent' | 'recoverable'; + +export interface NotificationSuppressionRetraction { + kind: NotificationSuppressionRetractionKind; + /** Hub rule.updated_at at retraction time; compared only to hub versions, never receiver clocks. */ + source_updated_at: number; +} + +export interface NotificationSuppressionRuleTombstone { + id: number; + deleted_at: number; + kind: NotificationSuppressionRetractionKind; + source_updated_at: number; +} + +export type SuppressionReplicaWriteOutcome = + | 'applied' + | 'ignored_stale' + | 'ignored_permanent_tombstone' + | 'ignored_recoverable_watermark'; + +export type SuppressionReplicaDeleteOutcome = 'applied' | 'ignored_stale'; + +export interface NotificationSuppressionPendingRetraction { + rule_id: number; + node_id: number; + kind: NotificationSuppressionRetractionKind; + source_updated_at: number; + created_at: number; + updated_at: number; + attempts: number; + last_error: string | null; +} + +/** Fail closed: anything other than recoverable is permanent. */ +function normalizeSuppressionRetractionKind(kind: unknown): NotificationSuppressionRetractionKind { + return kind === 'recoverable' ? 'recoverable' : 'permanent'; +} + +/** Coerce tombstone watermarks for merge/read; invalid values become 0. */ +function safeTombstoneSourceUpdatedAt(value: unknown): number { + const n = Number(value); + return Number.isSafeInteger(n) ? n : 0; +} + export interface NotificationSuppressionRule { id: number; name: string; @@ -2266,6 +2312,39 @@ export class DatabaseService { ); `); this.tryAddColumn('notification_suppression_rules', 'schedule', 'TEXT NULL'); + // kind + source_updated_at: hub-authored retract ordering. Legacy rows stay + // permanent (fail closed); source_updated_at backfills from deleted_at for + // audit continuity but is never compared to receiver wall clocks on write paths. + this.tryAddColumn( + 'notification_suppression_rule_tombstones', + 'kind', + "TEXT NOT NULL DEFAULT 'permanent'", + ); + this.tryAddColumn( + 'notification_suppression_rule_tombstones', + 'source_updated_at', + 'INTEGER', + ); + this.db.prepare( + `UPDATE notification_suppression_rule_tombstones + SET source_updated_at = deleted_at + WHERE source_updated_at IS NULL`, + ).run(); + this.db.exec(` + CREATE TABLE IF NOT EXISTS notification_suppression_pending_retractions ( + rule_id INTEGER NOT NULL, + node_id INTEGER NOT NULL, + kind TEXT NOT NULL, + source_updated_at INTEGER NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + PRIMARY KEY (rule_id, node_id) + ); + CREATE INDEX IF NOT EXISTS idx_supp_pending_retract_node + ON notification_suppression_pending_retractions(node_id); + `); } private migrateNotificationHistoryContext(): void { @@ -2940,7 +3019,32 @@ export class DatabaseService { return this.getNotificationSuppressionRule(result.lastInsertRowid as number)!; } - public upsertNotificationSuppressionRuleReplica(rule: NotificationSuppressionRule): void { + private insertNotificationSuppressionRuleReplicaRow( + rule: NotificationSuppressionRule, + scheduleJson: string | null, + ): void { + this.db.prepare( + `INSERT INTO notification_suppression_rules + (id, name, node_id, stack_patterns, label_ids, categories, levels, applies_to, enabled, expires_at, schedule, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ).run( + rule.id, + rule.name, + rule.node_id ?? null, + JSON.stringify(rule.stack_patterns), + rule.label_ids ? JSON.stringify(rule.label_ids) : null, + rule.categories ? JSON.stringify(rule.categories) : null, + rule.levels ? JSON.stringify(rule.levels) : null, + rule.applies_to, + rule.enabled ? 1 : 0, + rule.expires_at ?? null, + scheduleJson, + rule.created_at, + rule.updated_at, + ); + } + + public upsertNotificationSuppressionRuleReplica(rule: NotificationSuppressionRule): SuppressionReplicaWriteOutcome { const scheduleJson = rule.schedule ? JSON.stringify(rule.schedule) : null; const existing = this.getNotificationSuppressionRule(rule.id); if (existing) { @@ -2951,7 +3055,7 @@ export class DatabaseService { `[DatabaseService] Ignoring stale suppression replica write for rule id=${sanitizeForLog(rule.id)} ` + `(incoming updated_at=${sanitizeForLog(rule.updated_at)} <= stored updated_at=${sanitizeForLog(existing.updated_at)})`, ); - return; + return 'ignored_stale'; } this.db.prepare( `UPDATE notification_suppression_rules SET @@ -2972,37 +3076,46 @@ export class DatabaseService { rule.updated_at, rule.id, ); - return; + return 'applied'; } - const tombstone = this.db.prepare( - 'SELECT deleted_at FROM notification_suppression_rule_tombstones WHERE id = ?', - ).get(rule.id) as { deleted_at: number } | undefined; - if (tombstone) { - console.warn( - `[DatabaseService] Ignoring suppression replica write for rule id=${sanitizeForLog(rule.id)}: ` + - `this id was deleted at ${sanitizeForLog(tombstone.deleted_at)} and must not be recreated`, - ); - return; - } - this.db.prepare( - `INSERT INTO notification_suppression_rules - (id, name, node_id, stack_patterns, label_ids, categories, levels, applies_to, enabled, expires_at, schedule, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - ).run( - rule.id, - rule.name, - rule.node_id ?? null, - JSON.stringify(rule.stack_patterns), - rule.label_ids ? JSON.stringify(rule.label_ids) : null, - rule.categories ? JSON.stringify(rule.categories) : null, - rule.levels ? JSON.stringify(rule.levels) : null, - rule.applies_to, - rule.enabled ? 1 : 0, - rule.expires_at ?? null, - scheduleJson, - rule.created_at, - rule.updated_at, - ); + // Re-read tombstone inside the transaction so a concurrent permanent + // retract cannot be cleared after an outer eligibility check. + let outcome: SuppressionReplicaWriteOutcome = 'applied'; + this.transaction(() => { + const tombstone = this.db.prepare( + `SELECT id, deleted_at, kind, source_updated_at + FROM notification_suppression_rule_tombstones WHERE id = ?`, + ).get(rule.id) as NotificationSuppressionRuleTombstone | undefined; + if (tombstone) { + const kind = normalizeSuppressionRetractionKind(tombstone.kind); + // Keep raw Number here: invalid watermarks must fail closed (block recreate), + // not coerce to 0 the way safeTombstoneSourceUpdatedAt does for merges/reads. + const sourceUpdatedAt = Number(tombstone.source_updated_at); + if (kind === 'permanent') { + console.warn( + `[DatabaseService] Ignoring suppression replica write for rule id=${sanitizeForLog(rule.id)}: ` + + `permanent tombstone (source_updated_at=${sanitizeForLog(sourceUpdatedAt)})`, + ); + outcome = 'ignored_permanent_tombstone'; + return; + } + if (!Number.isSafeInteger(sourceUpdatedAt) || rule.updated_at <= sourceUpdatedAt) { + console.warn( + `[DatabaseService] Ignoring suppression replica write for rule id=${sanitizeForLog(rule.id)}: ` + + `recoverable tombstone source_updated_at=${sanitizeForLog(sourceUpdatedAt)}; ` + + `incoming updated_at=${sanitizeForLog(rule.updated_at)} is not newer`, + ); + outcome = 'ignored_recoverable_watermark'; + return; + } + this.db.prepare( + 'DELETE FROM notification_suppression_rule_tombstones WHERE id = ?', + ).run(rule.id); + } + this.insertNotificationSuppressionRuleReplicaRow(rule, scheduleJson); + outcome = 'applied'; + }); + return outcome; } public updateNotificationSuppressionRule( @@ -3032,19 +3145,164 @@ export class DatabaseService { this.db.prepare(`UPDATE notification_suppression_rules SET ${fields.join(', ')} WHERE id = ?`).run(...values); } - public deleteNotificationSuppressionRule(id: number): number { - const changes = this.db.prepare('DELETE FROM notification_suppression_rules WHERE id = ?').run(id).changes; - // Fleet sync has no delivery ordering guarantee: a replica POST for this id can - // still be in flight. Record the delete permanently (ids are AUTOINCREMENT and - // never reused) so upsertNotificationSuppressionRuleReplica refuses to resurrect it. - // Tombstone unconditionally, even when changes is 0: deleteRuleOnNode issues this - // same DELETE for capability/invalid-schedule cleanup against remotes that may never - // have received the rule yet, and a reordered POST behind that DELETE must not create it. + /** + * Delete a suppression rule and record a hub-authored retraction tombstone. + * Optional retraction defaults to permanent/0 (fail closed) for one-arg callers. + * Recoverable DELETEs that are strictly older than a stored row are ignored + * (no durable mutation). Permanent always applies. Row delete + tombstone + * upsert are one transaction. + */ + public deleteNotificationSuppressionRule( + id: number, + retraction: NotificationSuppressionRetraction = { kind: 'permanent', source_updated_at: 0 }, + ): { changes: number; outcome: SuppressionReplicaDeleteOutcome } { + const kind = normalizeSuppressionRetractionKind(retraction.kind); + const sourceUpdatedAt = retraction.source_updated_at; + + return this.transaction(() => { + const existing = this.getNotificationSuppressionRule(id); + if ( + kind === 'recoverable' && + existing && + existing.updated_at > sourceUpdatedAt + ) { + console.warn( + `[DatabaseService] Ignoring stale recoverable suppression DELETE for rule id=${sanitizeForLog(id)}: ` + + `source_updated_at=${sanitizeForLog(sourceUpdatedAt)} < stored updated_at=${sanitizeForLog(existing.updated_at)}`, + ); + return { changes: 0, outcome: 'ignored_stale' }; + } + + const changes = this.db.prepare( + 'DELETE FROM notification_suppression_rules WHERE id = ?', + ).run(id).changes; + + const prior = this.db.prepare( + `SELECT kind, source_updated_at FROM notification_suppression_rule_tombstones WHERE id = ?`, + ).get(id) as { kind: string; source_updated_at: number } | undefined; + + let mergedKind = kind; + let mergedSource = sourceUpdatedAt; + if (prior) { + const priorKind = normalizeSuppressionRetractionKind(prior.kind); + // Permanent wins over recoverable; watermark always takes the max. + mergedKind = + priorKind === 'permanent' || kind === 'permanent' ? 'permanent' : 'recoverable'; + mergedSource = Math.max( + safeTombstoneSourceUpdatedAt(prior.source_updated_at), + sourceUpdatedAt, + ); + } + + this.db.prepare( + `INSERT INTO notification_suppression_rule_tombstones (id, deleted_at, kind, source_updated_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + deleted_at = excluded.deleted_at, + kind = excluded.kind, + source_updated_at = excluded.source_updated_at`, + ).run(id, Date.now(), mergedKind, mergedSource); + + return { changes, outcome: 'applied' }; + }); + } + + public getNotificationSuppressionRuleTombstone( + id: number, + ): NotificationSuppressionRuleTombstone | undefined { + const row = this.db.prepare( + `SELECT id, deleted_at, kind, source_updated_at + FROM notification_suppression_rule_tombstones WHERE id = ?`, + ).get(id) as NotificationSuppressionRuleTombstone | undefined; + if (!row) return undefined; + return { + id: row.id, + deleted_at: row.deleted_at, + kind: normalizeSuppressionRetractionKind(row.kind), + // Keep || 0 (not safeTombstoneSourceUpdatedAt): public reads historically + // surface any truthy Number() result; merge/write paths coerce separately. + source_updated_at: Number(row.source_updated_at) || 0, + }; + } + + + public upsertNotificationSuppressionPendingRetraction(row: { + rule_id: number; + node_id: number; + kind: NotificationSuppressionRetractionKind; + source_updated_at: number; + last_error?: string; + }): void { + const now = Date.now(); + const kind = normalizeSuppressionRetractionKind(row.kind); + const prior = this.db.prepare( + `SELECT kind, source_updated_at, attempts FROM notification_suppression_pending_retractions + WHERE rule_id = ? AND node_id = ?`, + ).get(row.rule_id, row.node_id) as { kind: string; source_updated_at: number; attempts: number } | undefined; + let mergedKind = kind; + let mergedSource = row.source_updated_at; + let attempts = 1; + if (prior) { + const priorKind = normalizeSuppressionRetractionKind(prior.kind); + mergedKind = + priorKind === 'permanent' || kind === 'permanent' ? 'permanent' : 'recoverable'; + mergedSource = Math.max( + safeTombstoneSourceUpdatedAt(prior.source_updated_at), + row.source_updated_at, + ); + attempts = (prior.attempts || 0) + 1; + } this.db.prepare( - `INSERT INTO notification_suppression_rule_tombstones (id, deleted_at) VALUES (?, ?) - ON CONFLICT(id) DO UPDATE SET deleted_at = excluded.deleted_at`, - ).run(id, Date.now()); - return changes; + `INSERT INTO notification_suppression_pending_retractions + (rule_id, node_id, kind, source_updated_at, created_at, updated_at, attempts, last_error) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(rule_id, node_id) DO UPDATE SET + kind = excluded.kind, + source_updated_at = excluded.source_updated_at, + updated_at = excluded.updated_at, + attempts = excluded.attempts, + last_error = excluded.last_error`, + ).run( + row.rule_id, + row.node_id, + mergedKind, + mergedSource, + now, + now, + attempts, + row.last_error ?? null, + ); + } + + public deleteNotificationSuppressionPendingRetraction(ruleId: number, nodeId: number): void { + this.db.prepare( + 'DELETE FROM notification_suppression_pending_retractions WHERE rule_id = ? AND node_id = ?', + ).run(ruleId, nodeId); + } + + public listNotificationSuppressionPendingRetractions( + nodeId?: number, + ): NotificationSuppressionPendingRetraction[] { + const rows = ( + nodeId == null + ? this.db.prepare( + `SELECT * FROM notification_suppression_pending_retractions ORDER BY updated_at ASC`, + ).all() + : this.db.prepare( + `SELECT * FROM notification_suppression_pending_retractions + WHERE node_id = ? ORDER BY updated_at ASC`, + ).all(nodeId) + ) as Array>; + return rows.map((r) => ({ + rule_id: r.rule_id as number, + node_id: r.node_id as number, + kind: normalizeSuppressionRetractionKind(r.kind), + source_updated_at: Number(r.source_updated_at) || 0, + created_at: r.created_at as number, + updated_at: r.updated_at as number, + attempts: (r.attempts as number) || 0, + last_error: (r.last_error as string | null) ?? null, + })); } // --- Global Settings --- @@ -4450,6 +4708,9 @@ export class DatabaseService { this.deleteRoleAssignmentsByResource('node', String(id)); this.db.prepare('DELETE FROM fleet_sync_status WHERE node_id = ?').run(id); this.db.prepare('DELETE FROM node_update_skips WHERE node_id = ?').run(id); + this.db.prepare( + 'DELETE FROM notification_suppression_pending_retractions WHERE node_id = ?', + ).run(id); this.db.prepare('DELETE FROM nodes WHERE id = ?').run(id); })(); } diff --git a/backend/src/services/SuppressionRetractionRetryService.ts b/backend/src/services/SuppressionRetractionRetryService.ts new file mode 100644 index 00000000..7abdc8ed --- /dev/null +++ b/backend/src/services/SuppressionRetractionRetryService.ts @@ -0,0 +1,75 @@ +import { flushPendingSuppressionRetractions } from '../helpers/notificationSuppressionSync'; +import { isDebugEnabled } from '../utils/debug'; +import { getErrorMessage } from '../utils/errors'; + +const INITIAL_DELAY_MS = 30_000; +const EVAL_INTERVAL_MS = 5 * 60_000; + +/** + * Background retry for durable mute-replica retractions that failed or were + * deferred (offline Pilot/proxy, or remote lacking versioned retraction support). + */ +export class SuppressionRetractionRetryService { + private static instance: SuppressionRetractionRetryService; + private intervalId: NodeJS.Timeout | null = null; + private initialTimer: NodeJS.Timeout | null = null; + private isProcessing = false; + + private constructor() {} + + static getInstance(): SuppressionRetractionRetryService { + if (!SuppressionRetractionRetryService.instance) { + SuppressionRetractionRetryService.instance = new SuppressionRetractionRetryService(); + } + return SuppressionRetractionRetryService.instance; + } + + start(): void { + this.initialTimer = setTimeout(() => { + void this.evaluate(); + this.intervalId = setInterval(() => void this.evaluate(), EVAL_INTERVAL_MS); + }, INITIAL_DELAY_MS); + } + + stop(): void { + if (this.initialTimer) { + clearTimeout(this.initialTimer); + this.initialTimer = null; + } + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + } + } + + /** Full sweep of all pending rows. */ + async evaluate(): Promise { + if (this.isProcessing) return; + this.isProcessing = true; + try { + if (isDebugEnabled()) { + console.debug('[SuppressionRetractionRetry:debug] Evaluating pending retractions'); + } + await flushPendingSuppressionRetractions(); + } catch (err) { + console.error( + '[SuppressionRetractionRetry] evaluate error:', + getErrorMessage(err, String(err)), + ); + } finally { + this.isProcessing = false; + } + } + + /** Targeted flush when a Pilot tunnel or proxy node comes online. */ + async flushNode(nodeId: number): Promise { + try { + await flushPendingSuppressionRetractions(nodeId); + } catch (err) { + console.error( + `[SuppressionRetractionRetry] flushNode ${nodeId} failed:`, + getErrorMessage(err, String(err)), + ); + } + } +} diff --git a/frontend/src/lib/capabilities.ts b/frontend/src/lib/capabilities.ts index 887ccedd..15eccc9a 100644 --- a/frontend/src/lib/capabilities.ts +++ b/frontend/src/lib/capabilities.ts @@ -15,6 +15,7 @@ export const CAPABILITIES = [ 'notification-routing', 'notification-suppression', 'notification-suppression-schedule', + 'notification-suppression-replica-retraction', 'host-console', 'host-console-community', 'container-exec',