fix(indexer): count ledgers advanced in indexer_ledgers_processed_total

indexer/src/index.test.ts

@@ -1,5 +1,12 @@
 import { describe, expect, it } from "vitest";
-import { backoffDelay, commitPage, healthReport, seedLastLedger } from "./index.js";
+import {
+  backoffDelay,
+  commitPage,
+  healthReport,
+  metricLedgersProcessed,
+  metricPagesProcessed,
+  seedLastLedger,
+} from "./index.js";
 
 describe("backoffDelay", () => {
   it("doubles the delay for each consecutive failure", () => {
@@ -126,4 +133,36 @@ describe("commitPage", () => {
     expect(committed.gaps).toEqual([]);
     expect(committed.cursor).toEqual({ lastLedger: 100, paginationCursor: null });
   });
+
+  it("records the number of ledgers advanced (40 when advancing from 100 to 140)", async () => {
+    metricLedgersProcessed.reset();
+    metricPagesProcessed.reset();
+    const { client } = fakeClient();
+
+    await commitPage([], { lastLedger: 140, paginationCursor: null }, client, 100);
+
+    const ledgersMetric = await metricLedgersProcessed.get();
+    const pagesMetric = await metricPagesProcessed.get();
+    expect(ledgersMetric.values[0]?.value ?? 0).toBe(40);
+    expect(pagesMetric.values[0]?.value ?? 0).toBe(0);
+  });
+
+  it("counts a ledger split across three pages once in ledgers_processed and three times in pages_processed", async () => {
+    metricLedgersProcessed.reset();
+    metricPagesProcessed.reset();
+
+    const nopClient = {
+      $transaction: async () => {},
+    } as unknown as Client;
+    const dummyTx = {} as never;
+
+    await commitPage([dummyTx], { lastLedger: 105, paginationCursor: "c1" }, nopClient, 104);
+    await commitPage([dummyTx], { lastLedger: 105, paginationCursor: "c2" }, nopClient, 105);
+    await commitPage([dummyTx], { lastLedger: 105, paginationCursor: null }, nopClient, 105);
+
+    const ledgersMetric = await metricLedgersProcessed.get();
+    const pagesMetric = await metricPagesProcessed.get();
+    expect(ledgersMetric.values[0]?.value ?? 0).toBe(1);
+    expect(pagesMetric.values[0]?.value ?? 0).toBe(3);
+  });
 });

indexer/src/index.ts

@@ -56,6 +56,27 @@ export const metricLedgersProcessed = new Counter({
   registers: [metricsRegistry],
 });
 
+export const metricPagesProcessed = new Counter({
+  name: "indexer_pages_processed_total",
+  help: "Total number of transaction pages processed by the indexer",
+  registers: [metricsRegistry],
+});
+
+export function recordPageMetrics(
+  previousCursor: number,
+  nextCursor: number,
+  transactionCount = 0
+): number {
+  const ledgersAdvanced = Math.max(0, nextCursor - previousCursor);
+  if (ledgersAdvanced > 0) {
+    metricLedgersProcessed.inc(ledgersAdvanced);
+  }
+  if (transactionCount > 0) {
+    metricPagesProcessed.inc();
+  }
+  return ledgersAdvanced;
+}
+
 export const metricRowsWritten = new Counter({
   name: "indexer_rows_written_total",
   help: "Total rows written, partitioned by table",
@@ -350,7 +371,8 @@ async function fetchPage(
 export async function commitPage(
   transactions: rpc.Api.TransactionInfo[],
   progress: PageProgress,
-  client: Pick<typeof prisma, "$transaction"> = prisma
+  client: Pick<typeof prisma, "$transaction"> = prisma,
+  previousCursor?: number
 ): Promise<void> {
   await client.$transaction(
     async (db) => {
@@ -364,6 +386,9 @@ export async function commitPage(
     },
     { timeout: 30_000 }
   );
+  if (previousCursor !== undefined) {
+    recordPageMetrics(previousCursor, progress.lastLedger, transactions.length);
+  }
 }
 
 async function main(): Promise<void> {
@@ -403,16 +428,13 @@ async function main(): Promise<void> {
           progress.lastLedger !== cursor ||
           progress.paginationCursor !== paginationCursor
         ) {
-          await commitPage(page.transactions, progress);
+          await commitPage(page.transactions, progress, prisma, cursor);
           cursor = progress.lastLedger;
           paginationCursor = progress.paginationCursor;
           health.cursor = cursor;
           console.log(`[indexer] processed through ledger ${cursor}`);
         }
         caughtUp = cursor >= latestNow.sequence && !paginationCursor;
-        if (page.transactions.length > 0) {
-          metricLedgersProcessed.inc();
-        }
       }
       consecutiveFailures = 0;
       health.lastPollFailed = false;