mirror of
https://github.com/Studio-Saelix/sencho.git
synced 2026-07-26 11:49:16 +00:00
fix(notifications): version mute replica retractions for soft-cleanup restore (#1703)
* fix(notifications): version mute replica retractions for soft-cleanup restore Soft cleanup and authoritative delete shared an unversioned permanent tombstone, so a later hub re-save could not restore scheduled mutes on a remote. Carry hub-authored kind and source_updated_at on replica DELETE, allow recoverable recreate when updated_at is newer, keep permanent deletes fail-closed, and reject stale recoverable DELETEs against newer rows. * fix(notifications): durable mute retractions across mixed-version fleets Gate recoverable replica DELETEs on a new capability, durable-queue failures and incompatible remotes for retry, fan permanent deletes to every known remote, and return applied vs ignored outcomes on replica writes.
This commit is contained in:
Anso
@@ -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<string, unknown>) => ({
|
||||
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<string, unknown>) => ({
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<NotificationSuppressionRule> = {}): 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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
// 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
|
||||
|
||||
@@ -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<string, string> {
|
||||
return headers;
|
||||
}
|
||||
|
||||
function retractionFor(
|
||||
kind: NotificationSuppressionRetractionKind,
|
||||
rule: Pick<NotificationSuppressionRule, 'updated_at'>,
|
||||
): 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<void> {
|
||||
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<RetractionSupport> {
|
||||
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<void> {
|
||||
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<never> {
|
||||
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<string> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
// 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<void> {
|
||||
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)),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<NotificationCategory> = new Set(
|
||||
const VALID_LEVELS = new Set(['info', 'warning', 'error']);
|
||||
const VALID_APPLIES_TO = new Set<NotificationSuppressionAppliesTo>(['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<string, unknown>;
|
||||
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 });
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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<Record<string, unknown>>;
|
||||
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);
|
||||
})();
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
try {
|
||||
await flushPendingSuppressionRetractions(nodeId);
|
||||
} catch (err) {
|
||||
console.error(
|
||||
`[SuppressionRetractionRetry] flushNode ${nodeId} failed:`,
|
||||
getErrorMessage(err, String(err)),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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',
|
||||
|
||||
Reference in New Issue
Block a user