Add structured log for indexer lag threshold breach

Author: robertocarlous

src/modules/health/health.controllers.test.ts

@@ -19,14 +19,24 @@ jest.mock('../../utils/prisma.utils', () => ({
    },
 }));
 
+jest.mock('../../utils/indexer-cursor-staleness.utils', () => ({
+   checkIndexerCursorStalenessFromStore: jest.fn().mockResolvedValue(undefined),
+}));
+
 import {
    indexerHeartbeatCheck,
    recordIndexerHeartbeat,
    readinessCheck,
 } from './health.controllers';
 import { indexerHeartbeat } from '../../utils/heartbeat.service';
+import { checkIndexerCursorStalenessFromStore } from '../../utils/indexer-cursor-staleness.utils';
 import { prisma } from '../../utils/prisma.utils';
 
+const checkCursorStalenessMock =
+   checkIndexerCursorStalenessFromStore as jest.MockedFunction<
+      typeof checkIndexerCursorStalenessFromStore
+   >;
+
 const queryRawMock = prisma.$queryRaw as unknown as jest.Mock;
 
 // ---------------------------------------------------------------------------
@@ -118,13 +128,14 @@ describe('Indexer Heartbeat Controllers', () => {
    describe('recordIndexerHeartbeat', () => {
       beforeEach(() => {
          indexerHeartbeat.reset();
+         checkCursorStalenessMock.mockClear();
       });
 
-      it('records a heartbeat and returns 200', () => {
+      it('records a heartbeat and returns 200', async () => {
          const req = mockRequest();
          const res = mockResponse();
 
-         recordIndexerHeartbeat(req, res);
+         await recordIndexerHeartbeat(req, res);
 
          expect(res.statusCode).toBe(200);
          expect(res.body).toEqual(
@@ -139,14 +150,23 @@ describe('Indexer Heartbeat Controllers', () => {
          );
       });
 
-      it('makes the indexer status healthy', () => {
+      it('makes the indexer status healthy', async () => {
          const req = mockRequest();
          const res = mockResponse();
 
-         recordIndexerHeartbeat(req, res);
+         await recordIndexerHeartbeat(req, res);
 
          expect(indexerHeartbeat.getStatus().status).toBe('healthy');
       });
+
+      it('checks indexer cursor staleness after recording a heartbeat', async () => {
+         const req = mockRequest();
+         const res = mockResponse();
+
+         await recordIndexerHeartbeat(req, res);
+
+         expect(checkCursorStalenessMock).toHaveBeenCalledWith({ job: 'indexer' });
+      });
    });
 });
 

src/modules/health/health.controllers.ts

@@ -2,6 +2,7 @@ import { Request, Response } from 'express';
 import { prisma } from '../../utils/prisma.utils';
 import { envConfig } from '../../config';
 import { indexerHeartbeat } from '../../utils/heartbeat.service';
+import { checkIndexerCursorStalenessFromStore } from '../../utils/indexer-cursor-staleness.utils';
 import { sendSuccess } from '../../utils/api-response.utils';
 import { PUBLIC_ENDPOINT_CACHE_SECONDS } from '../../constants/public-endpoint-cache.constants';
 
@@ -173,8 +174,12 @@ export const indexerHeartbeatCheck = (_: Request, res: Response): void => {
  * POST /health/indexer/heartbeat
  * Called by the indexer worker to record a successful run.
  */
-export const recordIndexerHeartbeat = (_: Request, res: Response): void => {
+export const recordIndexerHeartbeat = async (
+  _: Request,
+  res: Response
+): Promise<void> => {
   const timestamp = indexerHeartbeat.recordHeartbeat();
+  await checkIndexerCursorStalenessFromStore({ job: 'indexer' });
   sendSuccess(
     res,
     { recorded: true, timestamp: timestamp.toISOString() },

src/utils/indexer-cursor-staleness.utils.ts

@@ -1,5 +1,16 @@
-import { logger } from './logger.utils';
 import { envConfig } from '../config';
+import { prisma } from './prisma.utils';
+import { logger } from './logger.utils';
+
+/** Correlates a staleness warning with the indexer job that observed it. */
+export interface IndexerCursorStalenessContext {
+   /** Indexer job or worker surface (e.g. `indexer`, `ledger-indexer`). */
+   job?: string;
+   /** Opaque cursor value from the backing store, when available. */
+   cursor?: string;
+   /** Latest indexed ledger sequence, when available. */
+   ledger?: number;
+}
 
 /**
  * Emits a structured warning when the indexer cursor has not been updated within the
@@ -11,20 +22,64 @@ import { envConfig } from '../config';
  * Default threshold: `INDEXER_CURSOR_STALE_AGE_WARNING_MS` env variable (300 000 ms / 5 min).
  * Override with the `thresholdMs` parameter for per-call control.
  *
+ * No log is emitted when `ENABLE_INDEXER_CURSOR_STALENESS_WARNING` is false or when lag
+ * is at or below the threshold.
+ *
  * @param lastUpdatedAt - Timestamp of the cursor's most recent update
  * @param thresholdMs   - Optional override; defaults to env config value
+ * @param context       - Optional fields to correlate the warning with an indexer job
  */
 export function warnIfIndexerCursorStale(
-  lastUpdatedAt: Date,
-  thresholdMs: number = envConfig.INDEXER_CURSOR_STALE_AGE_WARNING_MS
+   lastUpdatedAt: Date,
+   thresholdMs: number = envConfig.INDEXER_CURSOR_STALE_AGE_WARNING_MS,
+   context: IndexerCursorStalenessContext = {}
 ): void {
-  const ageMs = Date.now() - lastUpdatedAt.getTime();
-  if (ageMs > thresholdMs) {
-    logger.warn({
-      msg: 'Indexer cursor is stale',
-      lastUpdatedAt: lastUpdatedAt.toISOString(),
-      ageMs,
-      thresholdMs,
-    });
-  }
+   if (!envConfig.ENABLE_INDEXER_CURSOR_STALENESS_WARNING) {
+      return;
+   }
+
+   const lagMs = Date.now() - lastUpdatedAt.getTime();
+   if (lagMs > thresholdMs) {
+      logger.warn({
+         msg: 'Indexer cursor lag exceeded threshold',
+         job: context.job ?? 'indexer',
+         lagMs,
+         thresholdMs,
+         lastUpdatedAt: lastUpdatedAt.toISOString(),
+         ...(context.cursor !== undefined ? { cursor: context.cursor } : {}),
+         ...(context.ledger !== undefined ? { ledger: context.ledger } : {}),
+      });
+   }
+}
+
+/**
+ * Reads the latest indexed-ledger cursor from the database and emits a staleness
+ * warning when its age exceeds the configured threshold.
+ *
+ * Intended for the indexer worker heartbeat path after a successful run.
+ */
+export async function checkIndexerCursorStalenessFromStore(
+   context: IndexerCursorStalenessContext = {}
+): Promise<void> {
+   if (!envConfig.ENABLE_INDEXER_CURSOR_STALENESS_WARNING) {
+      return;
+   }
+
+   const status = await prisma.indexedLedger.findFirst({
+      orderBy: { updatedAt: 'desc' },
+   });
+
+   if (!status) {
+      return;
+   }
+
+   warnIfIndexerCursorStale(
+      status.updatedAt,
+      envConfig.INDEXER_CURSOR_STALE_AGE_WARNING_MS,
+      {
+         job: context.job ?? 'indexer',
+         cursor: status.cursor,
+         ledger: status.ledger,
+      }
+   );
 }

src/utils/test/indexer-cursor-staleness.utils.test.ts

@@ -1,54 +1,134 @@
-import { warnIfIndexerCursorStale } from '../indexer-cursor-staleness.utils';
+import {
+   checkIndexerCursorStalenessFromStore,
+   warnIfIndexerCursorStale,
+} from '../indexer-cursor-staleness.utils';
 import { logger } from '../logger.utils';
+import { prisma } from '../prisma.utils';
+import { envConfig } from '../../config';
 
 jest.mock('../logger.utils', () => ({
-  logger: { warn: jest.fn() },
+   logger: { warn: jest.fn() },
+}));
+
+jest.mock('../prisma.utils', () => ({
+   prisma: {
+      indexedLedger: {
+         findFirst: jest.fn(),
+      },
+   },
+}));
+
+jest.mock('../../config', () => ({
+   envConfig: {
+      ENABLE_INDEXER_CURSOR_STALENESS_WARNING: true,
+      INDEXER_CURSOR_STALE_AGE_WARNING_MS: 300_000,
+   },
 }));
 
 const warnMock = logger.warn as jest.Mock;
+const findFirstMock = prisma.indexedLedger.findFirst as jest.Mock;
 
 beforeEach(() => {
-  warnMock.mockClear();
+   warnMock.mockClear();
+   findFirstMock.mockReset();
+   (envConfig as { ENABLE_INDEXER_CURSOR_STALENESS_WARNING: boolean }).ENABLE_INDEXER_CURSOR_STALENESS_WARNING = true;
 });
 
 describe('warnIfIndexerCursorStale()', () => {
-  it('emits a warning when cursor age exceeds the threshold', () => {
-    const sixMinutesAgo = new Date(Date.now() - 360_000);
-    warnIfIndexerCursorStale(sixMinutesAgo, 300_000);
-    expect(warnMock).toHaveBeenCalledTimes(1);
-    expect(warnMock).toHaveBeenCalledWith(
-      expect.objectContaining({
-        msg: 'Indexer cursor is stale',
-        thresholdMs: 300_000,
-      })
-    );
-  });
-
-  it('does not emit a warning when cursor age is within the threshold', () => {
-    const oneMinuteAgo = new Date(Date.now() - 60_000);
-    warnIfIndexerCursorStale(oneMinuteAgo, 300_000);
-    expect(warnMock).not.toHaveBeenCalled();
-  });
-
-  it('does not emit a warning when cursor age exactly equals the threshold', () => {
-    const exactly = new Date(Date.now() - 300_000);
-    warnIfIndexerCursorStale(exactly, 300_000);
-    expect(warnMock).not.toHaveBeenCalled();
-  });
-
-  it('includes lastUpdatedAt, ageMs and thresholdMs in the warning payload', () => {
-    const ts = new Date(Date.now() - 400_000);
-    warnIfIndexerCursorStale(ts, 300_000);
-    const call = warnMock.mock.calls[0][0];
-    expect(call.lastUpdatedAt).toBe(ts.toISOString());
-    expect(typeof call.ageMs).toBe('number');
-    expect(call.ageMs).toBeGreaterThan(300_000);
-    expect(call.thresholdMs).toBe(300_000);
-  });
-
-  it('respects a custom threshold override', () => {
-    const twoSecondsAgo = new Date(Date.now() - 2_000);
-    warnIfIndexerCursorStale(twoSecondsAgo, 1_000);
-    expect(warnMock).toHaveBeenCalledTimes(1);
-  });
+   it('emits a warning when cursor lag exceeds the threshold', () => {
+      const sixMinutesAgo = new Date(Date.now() - 360_000);
+      warnIfIndexerCursorStale(sixMinutesAgo, 300_000, {
+         job: 'indexer',
+         cursor: 'cursor-1',
+         ledger: 42,
+      });
+      expect(warnMock).toHaveBeenCalledTimes(1);
+      expect(warnMock).toHaveBeenCalledWith(
+         expect.objectContaining({
+            msg: 'Indexer cursor lag exceeded threshold',
+            job: 'indexer',
+            cursor: 'cursor-1',
+            ledger: 42,
+            lagMs: expect.any(Number),
+            thresholdMs: 300_000,
+         })
+      );
+   });
+
+   it('does not emit a warning when cursor lag is within the threshold', () => {
+      const oneMinuteAgo = new Date(Date.now() - 60_000);
+      warnIfIndexerCursorStale(oneMinuteAgo, 300_000);
+      expect(warnMock).not.toHaveBeenCalled();
+   });
+
+   it('does not emit a warning when cursor lag exactly equals the threshold', () => {
+      const exactly = new Date(Date.now() - 300_000);
+      warnIfIndexerCursorStale(exactly, 300_000);
+      expect(warnMock).not.toHaveBeenCalled();
+   });
+
+   it('includes lastUpdatedAt, lagMs and thresholdMs in the warning payload', () => {
+      const ts = new Date(Date.now() - 400_000);
+      warnIfIndexerCursorStale(ts, 300_000);
+      const call = warnMock.mock.calls[0][0];
+      expect(call.lastUpdatedAt).toBe(ts.toISOString());
+      expect(typeof call.lagMs).toBe('number');
+      expect(call.lagMs).toBeGreaterThan(300_000);
+      expect(call.thresholdMs).toBe(300_000);
+   });
+
+   it('respects a custom threshold override', () => {
+      const twoSecondsAgo = new Date(Date.now() - 2_000);
+      warnIfIndexerCursorStale(twoSecondsAgo, 1_000);
+      expect(warnMock).toHaveBeenCalledTimes(1);
+   });
+
+   it('does not emit a warning when ENABLE_INDEXER_CURSOR_STALENESS_WARNING is false', () => {
+      (envConfig as { ENABLE_INDEXER_CURSOR_STALENESS_WARNING: boolean }).ENABLE_INDEXER_CURSOR_STALENESS_WARNING = false;
+      const sixMinutesAgo = new Date(Date.now() - 360_000);
+      warnIfIndexerCursorStale(sixMinutesAgo, 300_000);
+      expect(warnMock).not.toHaveBeenCalled();
+   });
+});
+
+describe('checkIndexerCursorStalenessFromStore()', () => {
+   it('emits a warning when the stored cursor is stale', async () => {
+      findFirstMock.mockResolvedValue({
+         ledger: 99,
+         cursor: 'abc',
+         updatedAt: new Date(Date.now() - 400_000),
+      });
+
+      await checkIndexerCursorStalenessFromStore({ job: 'indexer' });
+
+      expect(warnMock).toHaveBeenCalledWith(
+         expect.objectContaining({
+            job: 'indexer',
+            cursor: 'abc',
+            ledger: 99,
+            lagMs: expect.any(Number),
+            thresholdMs: 300_000,
+         })
+      );
+   });
+
+   it('does not emit a warning when no indexed ledger row exists', async () => {
+      findFirstMock.mockResolvedValue(null);
+
+      await checkIndexerCursorStalenessFromStore();
+
+      expect(warnMock).not.toHaveBeenCalled();
+   });
+
+   it('does not emit a warning when the stored cursor is fresh', async () => {
+      findFirstMock.mockResolvedValue({
+         ledger: 1,
+         cursor: 'fresh',
+         updatedAt: new Date(),
+      });
+
+      await checkIndexerCursorStalenessFromStore();
+
+      expect(warnMock).not.toHaveBeenCalled();
+   });
 });