Release the replication lag alert when the master link drops

apps/api/src/prometheus/prometheus.replication-lag.spec.ts

@@ -0,0 +1,168 @@
+import { ConfigService } from '@nestjs/config';
+import { WebhookEventType, type Webhook } from '@betterdb/shared';
+import { PrometheusService } from './prometheus.service';
+import { ConnectionRegistry } from '../connections/connection-registry.service';
+import { RuntimeCapabilityTracker } from '../connections/runtime-capability-tracker.service';
+import { SlowLogAnalyticsService } from '../slowlog-analytics/slowlog-analytics.service';
+import { CommandLogAnalyticsService } from '../commandlog-analytics/commandlog-analytics.service';
+import { HealthService } from '../health/health.service';
+import { StoragePort } from '../common/interfaces/storage-port.interface';
+import { WebhookDispatcherService } from '../webhooks/webhook-dispatcher.service';
+import { WebhookEventsProService } from '@proprietary/webhook-pro/webhook-events-pro.service';
+
+const CONNECTION_ID = 'conn-1';
+const OTHER_CONNECTION_ID = 'conn-2';
+const INSTANCE = { host: '10.0.0.1', port: 6379 };
+
+function webhook(id: string): Webhook {
+  return {
+    id,
+    name: id,
+    url: `https://hooks.example.com/${id}`,
+    enabled: true,
+    events: [WebhookEventType.REPLICATION_LAG],
+    retryPolicy: { maxRetries: 0, backoffMultiplier: 2, initialDelayMs: 1, maxDelayMs: 1 },
+    thresholds: {},
+    createdAt: 0,
+    updatedAt: 0,
+  };
+}
+
+interface Harness {
+  service: PrometheusService;
+  fired: string[];
+}
+
+function buildService(): Harness {
+  const config = {
+    get: jest.fn((_key: string, fallback?: unknown) => fallback),
+  } as unknown as ConfigService;
+  const dispatcher = new WebhookDispatcherService(
+    {} as never,
+    { getWebhooksByEvent: jest.fn().mockResolvedValue([webhook('wh-1'), webhook('wh-2')]) } as never,
+    config,
+  );
+  const fired: string[] = [];
+  jest
+    .spyOn(dispatcher as never, 'dispatchPendingEvent')
+    .mockImplementation((async (event: { connectionId?: string }, targets?: Webhook[]) => {
+      fired.push(`${event.connectionId}:${targets?.[0]?.id}`);
+      return true;
+    }) as never);
+  const pro = new WebhookEventsProService(dispatcher, { getLicenseTier: () => 'pro' } as never);
+  const registry = {
+    getConfig: jest.fn().mockReturnValue(INSTANCE),
+    getName: jest.fn().mockReturnValue('primary'),
+    list: jest.fn().mockReturnValue([]),
+  } as unknown as ConnectionRegistry;
+  const service = new PrometheusService(
+    {} as StoragePort,
+    registry,
+    config,
+    {} as RuntimeCapabilityTracker,
+    {} as SlowLogAnalyticsService,
+    {} as CommandLogAnalyticsService,
+    {} as HealthService,
+    undefined,
+    pro,
+  );
+  return { service, fired };
+}
+
+async function poll(
+  service: PrometheusService,
+  replication: Record<string, string>,
+  connectionId = CONNECTION_ID,
+): Promise<void> {
+  service['updateReplicationMetrics'](
+    { replication } as never,
+    'primary',
+    connectionId,
+    INSTANCE,
+    false,
+  );
+  for (let i = 0; i < 5; i++) {
+    await new Promise((resolve) => setImmediate(resolve));
+  }
+}
+
+const LAGGING = { role: 'slave', master_link_status: 'up', master_last_io_seconds_ago: '30' };
+const LINK_DOWN = { role: 'slave', master_link_status: 'down', master_last_io_seconds_ago: '30' };
+const FIRED_ONCE = [`${CONNECTION_ID}:wh-1`, `${CONNECTION_ID}:wh-2`];
+const FIRED_TWICE = [...FIRED_ONCE, ...FIRED_ONCE];
+
+describe('PrometheusService replication lag alert', () => {
+  it('re-fires after the master link drops and comes back', async () => {
+    const { service, fired } = buildService();
+
+    await poll(service, LAGGING);
+    await poll(service, LAGGING);
+    expect(fired).toEqual(FIRED_ONCE);
+
+    await poll(service, LINK_DOWN);
+    await poll(service, LAGGING);
+    expect(fired).toEqual(FIRED_TWICE);
+  });
+
+  it('keeps one connection latch from suppressing another', async () => {
+    const { service, fired } = buildService();
+
+    await poll(service, LAGGING, CONNECTION_ID);
+    await poll(service, LAGGING, OTHER_CONNECTION_ID);
+    await poll(service, LINK_DOWN, OTHER_CONNECTION_ID);
+    await poll(service, LAGGING, CONNECTION_ID);
+
+    expect(fired).toEqual([...FIRED_ONCE, `${OTHER_CONNECTION_ID}:wh-1`, `${OTHER_CONNECTION_ID}:wh-2`]);
+  });
+
+  it('keeps the latch when the link is up but the lag field is unparsable', async () => {
+    const { service, fired } = buildService();
+
+    await poll(service, LAGGING);
+    await poll(service, { role: 'slave', master_link_status: 'up', master_last_io_seconds_ago: '' });
+    await poll(service, { role: 'slave', master_link_status: 'up' });
+    await poll(service, LAGGING);
+
+    expect(fired).toEqual(FIRED_ONCE);
+  });
+
+  it('keeps the latch when INFO omits the link status', async () => {
+    const { service, fired } = buildService();
+
+    await poll(service, LAGGING);
+    await poll(service, { role: 'slave' });
+    await poll(service, LAGGING);
+
+    expect(fired).toEqual(FIRED_ONCE);
+  });
+
+  it('re-arms the alert when the replica is promoted', async () => {
+    const { service, fired } = buildService();
+
+    await poll(service, LAGGING);
+    await poll(service, { role: 'master', connected_slaves: '0', master_repl_offset: '0' });
+    await poll(service, LAGGING);
+
+    expect(fired).toEqual(FIRED_TWICE);
+  });
+
+  it('re-arms the alert when the node reports an unknown role', async () => {
+    const { service, fired } = buildService();
+
+    await poll(service, LAGGING);
+    await poll(service, { role: 'sentinel' });
+    await poll(service, LAGGING);
+
+    expect(fired).toEqual(FIRED_TWICE);
+  });
+
+  it('re-arms the alert when the connection is removed', async () => {
+    const { service, fired } = buildService();
+
+    await poll(service, LAGGING);
+    service.cleanupConnectionMetrics(CONNECTION_ID);
+    await poll(service, LAGGING);
+
+    expect(fired).toEqual(FIRED_TWICE);
+  });
+});

apps/api/src/prometheus/prometheus.service.ts

@@ -1954,6 +1954,7 @@ export class PrometheusService extends MultiConnectionPoller implements OnModule
     }
 
     if (role !== 'master' && !isReplica) {
+      this.webhookEventsProService?.clearReplicationLag(connectionId);
       this.clearRoleSpecificReplicationSeries(connLabel);
       if (!absentAsZero) {
         this.replicationOffset.remove(connLabel);
@@ -1962,6 +1963,7 @@ export class PrometheusService extends MultiConnectionPoller implements OnModule
     }
 
     if (role === 'master') {
+      this.webhookEventsProService?.clearReplicationLag(connectionId);
       this.masterLinkUp.remove(connLabel);
       this.masterLastIoSecondsAgo.remove(connLabel);
       this.setInfoGauge(
@@ -2013,6 +2015,8 @@ export class PrometheusService extends MultiConnectionPoller implements OnModule
           .catch((err) => {
             this.logger.error('Failed to dispatch replication.lag webhook', err);
           });
+      } else if (masterLinkStatus !== undefined && masterLinkStatus !== 'up') {
+        this.webhookEventsProService?.clearReplicationLag(connectionId);
       }
     }
   }
@@ -3573,6 +3577,7 @@ export class PrometheusService extends MultiConnectionPoller implements OnModule
       this.removeCveSeries(cveLabel);
     }
     this.perConnectionState.delete(connectionId);
+    this.webhookEventsProService?.clearReplicationLag(connectionId);
     // Drop any in-flight metric update so a reused connection ID cannot join
     // stale work. The promise itself still settles; its compare-and-delete
     // finally then no-ops because the entry is already gone.

apps/api/src/webhooks/__tests__/webhook-dispatcher.service.spec.ts

@@ -221,6 +221,82 @@ describe('WebhookDispatcherService', () => {
     });
   });
 
+  describe('Clearing a threshold alert', () => {
+    it('re-arms every webhook of one connection without touching others', () => {
+      const fire = (key: string) => service.shouldFireAlert(key, 30, 10, true);
+      const keys = [
+        'replication_lag:conn-1:wh-1',
+        'replication_lag:conn-1:wh-2',
+        'replication_lag:conn-10:wh-1',
+        'replication_lag:conn-2:wh-1',
+        'slowlog_threshold:conn-1:wh-1',
+      ];
+
+      keys.forEach((key) => expect(fire(key)).toBe(true));
+      service.clearThresholdAlertPerWebhook('replication_lag', 'conn-1');
+
+      expect(keys.map(fire)).toEqual([true, true, false, false, false]);
+    });
+
+    it('drops a dispatch whose webhook lookup was still pending when the alert was cleared', async () => {
+      let resolveLookup: (webhooks: any[]) => void = () => undefined;
+      webhooksService.getWebhooksByEvent.mockReturnValue(new Promise((resolve) => (resolveLookup = resolve)));
+      const dispatched = jest.spyOn(service as any, 'dispatchPendingEvent').mockResolvedValue(undefined);
+
+      const pending = service.dispatchThresholdAlertPerWebhook(
+        WebhookEventType.REPLICATION_LAG,
+        'replication_lag',
+        30,
+        'replicationLagSeconds',
+        true,
+        { message: 'lag' },
+        'conn-1',
+      );
+      service.clearThresholdAlertPerWebhook('replication_lag', 'conn-1');
+      resolveLookup([{ id: 'wh-1', thresholds: {} }]);
+
+      await expect(pending).resolves.toBe(false);
+      expect(dispatched).not.toHaveBeenCalled();
+      expect(service.shouldFireAlert('replication_lag:conn-1:wh-1', 30, 10, true)).toBe(true);
+    });
+
+    it('does not restore a latch cleared in the microtask right after the lookup resolves', async () => {
+      webhooksService.getWebhooksByEvent.mockResolvedValue([{ id: 'wh-1', thresholds: {} }] as any);
+      jest.spyOn(service as any, 'dispatchPendingEvent').mockResolvedValue(undefined);
+
+      const pending = service.dispatchThresholdAlertPerWebhook(
+        WebhookEventType.REPLICATION_LAG,
+        'replication_lag',
+        30,
+        'replicationLagSeconds',
+        true,
+        { message: 'lag' },
+        'conn-1',
+      );
+      const cleared = Promise.resolve().then(() => service.clearThresholdAlertPerWebhook('replication_lag', 'conn-1'));
+      await Promise.all([pending, cleared]);
+
+      expect(service.shouldFireAlert('replication_lag:conn-1:wh-1', 30, 10, true)).toBe(true);
+    });
+
+    it('keeps no generation state once lookups settle, however many connections are cleared', async () => {
+      webhooksService.getWebhooksByEvent.mockResolvedValue([]);
+      for (let i = 0; i < 50; i++) {
+        service.clearThresholdAlertPerWebhook('replication_lag', `conn-${i}`);
+        await service.dispatchThresholdAlertPerWebhook(
+          WebhookEventType.REPLICATION_LAG,
+          'replication_lag',
+          30,
+          'replicationLagSeconds',
+          true,
+          { message: 'lag' },
+          `conn-${i}`,
+        );
+      }
+      expect((service as any).pendingLookups.size).toBe(0);
+    });
+  });
+
   describe('Signature Generation', () => {
     it('should generate signature with timestamp', () => {
       webhooksService.generateSignature.mockReturnValue('test-signature');

apps/api/src/webhooks/webhook-dispatcher.service.ts

@@ -94,6 +94,7 @@ export class WebhookDispatcherService implements OnModuleDestroy {
 
   // Track alert states with LRU cache to prevent memory leak
   private alertStates: LRUCache<string, AlertState>;
+  private readonly pendingLookups = new Map<string, { count: number; generation: number }>();
 
   // Instance context
   private readonly sourceHost: string;
@@ -412,6 +413,17 @@ export class WebhookDispatcherService implements OnModuleDestroy {
     return false;
   }
 
+  clearThresholdAlertPerWebhook(alertKeyPrefix: string, connectionId: string): void {
+    const generationKey = `${alertKeyPrefix}:${connectionId}`;
+    const pending = this.pendingLookups.get(generationKey);
+    if (pending) pending.generation += 1;
+    const connectionPrefix = `${generationKey}:`;
+    const keys = [...this.alertStates.keys()].filter((key) => key.startsWith(connectionPrefix));
+    for (const key of keys) {
+      this.alertStates.delete(key);
+    }
+  }
+
   /**
    * Dispatch threshold-based alert (e.g., memory.critical, connection.critical)
    * Uses global threshold - consider using dispatchThresholdAlertPerWebhook for per-webhook thresholds
@@ -448,6 +460,24 @@ export class WebhookDispatcherService implements OnModuleDestroy {
    * @param data Event data payload
    * @param connectionId Optional connection ID to filter webhooks and include in payload
    */
+  private trackLookup(alertKeyPrefix: string, connectionId?: string): () => boolean {
+    if (!connectionId) return () => true;
+    const key = `${alertKeyPrefix}:${connectionId}`;
+    const pending = this.pendingLookups.get(key) ?? { count: 0, generation: 0 };
+    pending.count += 1;
+    this.pendingLookups.set(key, pending);
+    const generation = pending.generation;
+    let settled = false;
+    return () => {
+      if (!settled) {
+        settled = true;
+        pending.count -= 1;
+        if (pending.count === 0) this.pendingLookups.delete(key);
+      }
+      return pending.generation === generation;
+    };
+  }
+
   async dispatchThresholdAlertPerWebhook(
     eventType: WebhookEventType,
     alertKeyPrefix: string,
@@ -458,9 +488,13 @@ export class WebhookDispatcherService implements OnModuleDestroy {
     connectionId?: string,
   ): Promise<boolean> {
     let firedAny = false;
+    const settleLookup = this.trackLookup(alertKeyPrefix, connectionId);
     try {
       // Get webhooks subscribed to this event, filtered by connectionId
       const webhooks = await this.webhooksService.getWebhooksByEvent(eventType, connectionId);
+      if (!settleLookup()) {
+        return false;
+      }
 
       if (webhooks.length === 0) {
         this.logger.debug(
@@ -514,6 +548,8 @@ export class WebhookDispatcherService implements OnModuleDestroy {
       );
     } catch (error) {
       this.logger.error(`Failed to dispatch per-webhook threshold alert ${eventType}:`, error);
+    } finally {
+      settleLookup();
     }
     return firedAny;
   }

packages/shared/src/webhooks/types.ts

@@ -372,9 +372,11 @@ export interface IWebhookEventsProService {
     masterLinkStatus: string;
     timestamp: number;
     instance: WebhookInstanceInfo;
-    connectionId?: string;
+    connectionId: string;
   }): Promise<void>;
 
+  clearReplicationLag(connectionId: string): void;
+
   dispatchClusterFailover(data: {
     clusterState: string;
     previousState?: string;