Merge pull request #716 from greatest0fallt1me/feat/670-backfill-lag-metrics

Add backfill lag gauge

Author: Jagadeeshftw

docs/routes/backfill-events.md

@@ -61,6 +61,16 @@ the persisted checkpoint.
   the DB transaction boundary — each batch and its checkpoint commit
   atomically.
 
+## Lag metric and alerting
+
+`backfill_lag_blocks{job,contract}` reports the difference between the current
+Starknet chain head and the last block persisted for each backfill job. The
+gauge is refreshed when progress is read and after each checkpoint advances.
+RPC failures are counted in `backfill_lag_rpc_errors_total` and do not fail the
+backfill itself. Alert when lag remains above the normal batch window (for
+example, more than 100 blocks for 15 minutes), adjusting the threshold for the
+configured indexing interval and expected chain throughput.
+
 ## Backward-compatibility contract (Issue #264)
 
 1. **Input alias stability** — `before`, `resumeToken`, and `cursor` are

src/db/migrations/0005_backfill_lag_columns.sql

@@ -0,0 +1,3 @@
+ALTER TABLE "backfill_progress"
+  ADD COLUMN "last_block_number" bigint,
+  ADD COLUMN "last_contract_address" text;

src/db/schema.ts

@@ -636,6 +636,8 @@ export const backfillProgress = pgTable(
     jobName: text("job_name").primaryKey(),
     status: text("status").notNull().default("idle"),
     lastCursor: timestamp("last_cursor"),
+    lastBlockNumber: bigint("last_block_number", { mode: "number" }),
+    lastContractAddress: text("last_contract_address"),
     totalScanned: integer("total_scanned").notNull().default(0),
     totalCreated: integer("total_created").notNull().default(0),
     lastError: text("last_error"),
@@ -795,4 +797,3 @@ export const SCHEMA_TABLES: Array<{ name: string; table: PgTableWithColumns<any>
   { name: "sessions", table: sessions },
   { name: "backfillProgress", table: backfillProgress },
 ];
-

src/routes/backfill-events.test.ts

@@ -33,6 +33,10 @@ vi.mock("../config.js", () => ({
   env: { ADMIN_ADDRESSES: ["0xabc1"] },
 }));
 
+vi.mock("../starknet/client.js", () => ({
+  provider: { getBlockNumber: vi.fn().mockResolvedValue(1200) },
+}));
+
 const { dbMock, schemaMock, store } = vi.hoisted(() => {
   interface AgreementEventRow {
     id: string;
@@ -48,6 +52,8 @@ const { dbMock, schemaMock, store } = vi.hoisted(() => {
     jobName: string;
     status: string;
     lastCursor: Date | null;
+    lastBlockNumber: number | null;
+    lastContractAddress: string | null;
     totalScanned: number;
     totalCreated: number;
     lastError: string | null;
@@ -245,6 +251,8 @@ import {
   getBackfillProgress,
 } from "./backfill-events.js";
 import { requireSession } from "../auth/session.js";
+import { provider } from "../starknet/client.js";
+import { getStarknetMetricsSnapshot, resetStarknetMetrics } from "../starknet/client-metrics.js";
 
 const ADMIN = "0xabc1";
 const NON_ADMIN = "0xdef2";
@@ -313,10 +321,28 @@ function makeDescendingRows(count: number, idOffset = 0) {
 
 beforeEach(() => {
   vi.clearAllMocks();
+  resetStarknetMetrics();
+  vi.mocked(provider.getBlockNumber).mockResolvedValue(1200);
   store.reset();
   vi.mocked(requireSession).mockResolvedValue(true);
 });
 
+describe("backfill lag metrics", () => {
+  it("records chain-head lag with job and contract labels", async () => {
+    queueRows([makeRow(1)]);
+    vi.mocked(provider.getBlockNumber).mockResolvedValue(1200);
+
+    const response = await request(makeApp())
+      .post("/api/v1/backfill/employee-events")
+      .set(authHeaders(ADMIN));
+
+    expect(response.status).toBe(200);
+    expect(getStarknetMetricsSnapshot().gauges[
+      'backfill_lag_blocks{job="employee-events",contract="0xcontract"}'
+    ]).toBe(199);
+  });
+});
+
 interface JobConfig {
   jobName: "employee-events" | "milestone-events";
   path: string;
@@ -790,4 +816,3 @@ describe("Edge cases", () => {
   });
 });
 
-

src/routes/backfill-events.ts

@@ -3,6 +3,12 @@ import { requireAuth, requireAdmin } from "../auth/middleware.js";
 import { z } from "zod";
 import { db, schema } from "../db/index.js";
 import { sql, eq } from "drizzle-orm";
+import { provider } from "../starknet/client.js";
+import {
+  incStarknetMetric,
+  labeledStarknetMetric,
+  setStarknetGauge,
+} from "../starknet/client-metrics.js";
 
 export const backfillEventsRouter = Router();
 
@@ -36,6 +42,7 @@ export const RESULTS_PREVIEW_SIZE = 10;
  * matches what is actually durable in `agreement_events`.
  */
 export const BACKFILL_CHECKPOINT_BATCH_SIZE = 100;
+export const BACKFILL_LAG_METRIC = "backfill_lag_blocks";
 
 // ---------------------------------------------------------------------------
 // Resume token freshness bounds (Issue #263)
@@ -61,6 +68,27 @@ export const CLOCK_SKEW_TOLERANCE_MS = 60 * 1000; // 60 seconds
 
 export type BackfillJobName = "employee-events" | "milestone-events";
 
+/** Refresh lag observability without allowing an RPC outage to stop a backfill. */
+async function updateBackfillLag(
+  jobName: BackfillJobName,
+  lastBlockNumber: number | null | undefined,
+  contractAddress: string | null | undefined,
+): Promise<void> {
+  try {
+    const chainHead = await provider.getBlockNumber();
+    const lag = Math.max(0, chainHead - (lastBlockNumber ?? 0));
+    setStarknetGauge(
+      labeledStarknetMetric(BACKFILL_LAG_METRIC, {
+        job: jobName,
+        contract: contractAddress ?? "unknown",
+      }),
+      lag,
+    );
+  } catch {
+    incStarknetMetric("backfill_lag_rpc_errors_total");
+  }
+}
+
 export const EMPLOYEE_BACKFILL_JOB: BackfillJobName = "employee-events";
 export const MILESTONE_BACKFILL_JOB: BackfillJobName = "milestone-events";
 export const BACKFILL_JOB_NAMES: readonly BackfillJobName[] = [
@@ -261,6 +289,8 @@ export async function upsertBackfillProgress(
   fields: {
     status?: string;
     lastCursor?: Date | null;
+    lastBlockNumber?: number | null;
+    lastContractAddress?: string | null;
     totalScanned?: number;
     totalCreated?: number;
     lastError?: string | null;
@@ -309,6 +339,8 @@ export async function getBackfillProgress(
 ): Promise<{
   status: string;
   lastCursor: Date | null;
+  lastBlockNumber: number | null;
+  lastContractAddress: string | null;
   totalScanned: number;
   totalCreated: number;
   lastError: string | null;
@@ -324,6 +356,8 @@ export async function getBackfillProgress(
   return {
     status: row.status,
     lastCursor: row.lastCursor,
+    lastBlockNumber: row.lastBlockNumber,
+    lastContractAddress: row.lastContractAddress,
     totalScanned: row.totalScanned,
     totalCreated: row.totalCreated,
     lastError: row.lastError,
@@ -384,6 +418,7 @@ export async function performBackfill(
       persistedTotalScanned = progress.totalScanned;
       persistedTotalCreated = progress.totalCreated;
     }
+    await updateBackfillLag(jobName, progress?.lastBlockNumber, progress?.lastContractAddress);
   }
 
   const conditions = sql`1=1`;
@@ -483,12 +518,20 @@ export async function performBackfill(
           totalScanned: batchTotalScanned + batch.length,
           totalCreated: batchCreatedCount + batchCreated,
           lastCursor: batchCursor,
+          lastBlockNumber: Number(lastRow.block_number),
+          lastContractAddress: String(lastRow.contract_address),
         });
 
         batchCreatedCount += batchCreated;
         batchTotalScanned += batch.length;
       });
 
+      await updateBackfillLag(
+        jobName,
+        Number(batch[batch.length - 1].block_number),
+        String(batch[batch.length - 1].contract_address),
+      );
+
       const insertedIds = new Set(insertedRows.map((r) => String(r.id)));
 
       for (const row of batch) {

src/starknet/client-metrics.ts

@@ -65,6 +65,14 @@ export function setStarknetGauge(name: string, value: number): void {
   gauges[name] = value;
 }
 
+/** Return a Prometheus-style metric key with safely escaped label values. */
+export function labeledStarknetMetric(name: string, labels: Record<string, string>): string {
+  const encoded = Object.entries(labels)
+    .map(([key, value]) => `${key}="${value.replaceAll("\\", "\\\\").replaceAll('"', '\\"')}"`)
+    .join(",");
+  return `${name}{${encoded}}`;
+}
+
 /**
  * Point-in-time snapshot of every Starknet metric counter and gauge.
  * Returns shallow copies so callers cannot mutate the internal state.