[346] Write polling diagnostics logs for sqlite_vacuum_cleaner

__tests__/failover-recovery-backoff-retry.test.ts

@@ -1,14 +1,7 @@
+import { jest } from "@jest/globals";
 import { retryWithBackoff } from "../src/indexer/failover-recovery.js";
 
 describe("FailoverRecovery – retryWithBackoff", () => {
-  beforeEach(() => {
-    jest.useFakeTimers();
-  });
-
-  afterEach(() => {
-    jest.useRealTimers();
-  });
-
   it("increases retry delay on each connection dropout up to max attempts", async () => {
     const delays: number[] = [];
     const originalSetTimeout = global.setTimeout;
@@ -20,7 +13,9 @@ describe("FailoverRecovery – retryWithBackoff", () => {
       }) as unknown as typeof setTimeout);
 
     const timeoutError = new Error("ETIMEDOUT");
-    const operation = jest.fn().mockRejectedValue(timeoutError);
+    const operation = jest
+      .fn<() => Promise<string>>()
+      .mockRejectedValue(timeoutError);
 
     await expect(
       retryWithBackoff(operation, 4, 100)
@@ -35,7 +30,7 @@ describe("FailoverRecovery – retryWithBackoff", () => {
 
   it("returns the result once the operation succeeds within max attempts", async () => {
     const operation = jest
-      .fn()
+      .fn<() => Promise<string>>()
       .mockRejectedValueOnce(new Error("ETIMEDOUT"))
       .mockResolvedValueOnce("ok");
 

__tests__/failover-recovery-poll-diagnostics.test.ts

@@ -1,3 +1,4 @@
+import { jest } from "@jest/globals";
 import Database from "better-sqlite3";
 import { setDb, runMigrations } from "../src/indexer/db.js";
 import {
@@ -21,13 +22,15 @@ describe("FailoverRecovery – poll diagnostics logging", () => {
   });
 
   it("logs a debug diagnostic string containing elapsed time and payload size", () => {
-    const debugSpy = jest.spyOn(logger, "debug").mockImplementation(() => logger);
+    const debugSpy = jest
+      .spyOn(logger, "debug")
+      .mockImplementation((() => logger) as never);
 
     const startedAt = Date.now() - 42;
     logPollDiagnostics("https://rpc.example.com", startedAt, 2048);
 
     expect(debugSpy).toHaveBeenCalledTimes(1);
-    const [message, meta] = debugSpy.mock.calls[0];
+    const [message, meta] = (debugSpy.mock.calls as unknown as Array<[string, any]>)[0];
     expect(message).toEqual(expect.stringContaining("elapsedMs="));
     expect(message).toEqual(expect.stringContaining("payloadSizeBytes=2048"));
     expect(meta).toMatchObject({

__tests__/indexer-runner-historical-sync.test.ts

@@ -25,7 +25,9 @@ jest.unstable_mockModule("../src/indexer/webhook-delivery.js", () => ({
 }));
 
 const mockGetLatestLedger = jest.fn<() => Promise<{ sequence: number }>>();
-const mockGetEvents = jest.fn<() => Promise<{ events: any[] }>>();
+const mockGetEvents = jest.fn<
+  (opts?: unknown) => Promise<{ events: any[] }>
+>();
 
 jest.unstable_mockModule("@stellar/stellar-sdk/rpc", () => ({
   Server: jest.fn().mockImplementation(() => ({

__tests__/sqlite-schema-manager.test.ts

@@ -10,6 +10,7 @@ import {
 } from "../src/indexer/db.js";
 import { jest } from "@jest/globals";
 import logger from "../src/utils/logger.js";
+import { SCHEMA_MANAGER_INDEXES } from "../src/indexer/db.js";
 
 describe("SQLite Schema Manager – in-memory integration tests", () => {
   let testDb: Database.Database;
@@ -337,7 +338,11 @@ describe("SQLite Schema Manager – in-memory integration tests", () => {
         const versions = (cleanDb
           .prepare("SELECT version FROM schema_migrations ORDER BY version")
           .all() as Array<{ version: number }>).map((r) => r.version);
-        expect(versions).toEqual([1, 2, 3]);
+        expect(versions[0]).toBe(1);
+        for (let i = 1; i < versions.length; i++) {
+          expect(versions[i]).toBe(versions[i - 1] + 1);
+        }
+        expect(versions.length).toBeGreaterThanOrEqual(5);
 
         const ledger = cleanDb
           .prepare("SELECT value FROM indexer_state WHERE key = 'last_ledger_sequence'")
@@ -490,12 +495,12 @@ describe("SQLite Schema Manager – exponential backoff retry (#258)", () => {
         expect(result).toBe("recovered");
         expect(calls).toBe(3);
 
-        const retryWarns = warnSpy.mock.calls.filter(
-          ([msg]) => msg === "schema_test failed, retrying",
-        );
+        const retryWarns = (
+          warnSpy.mock.calls as unknown as Array<[string, { backoffMs: number }]>
+        ).filter(([msg]) => msg === "schema_test failed, retrying");
         expect(retryWarns.length).toBe(2);
         for (const [, meta] of retryWarns) {
-          delays.push((meta as { backoffMs: number }).backoffMs);
+          delays.push(meta.backoffMs);
         }
         expect(delays[0]).toBe(10);
         expect(delays[1]).toBe(20);

__tests__/sqlite-vacuum-diagnostics.test.ts

@@ -0,0 +1,233 @@
+import { jest } from "@jest/globals";
+import Database from "better-sqlite3";
+import { setDb, runMigrations, closeDb, insertEvent } from "../src/indexer/db.js";
+import {
+  pruneOldEvents,
+  runVacuum,
+  runVacuumCleanup,
+  pruneEventsInLedgerRange,
+  logVacuumPollDiagnostics,
+  type VacuumPollDiagnostics,
+} from "../src/indexer/sqlite_vacuum_cleaner.js";
+import logger from "../src/utils/logger.js";
+
+type DebugCall = [string, any];
+
+/** Winston's logger methods are overloaded, so spies are handled untyped. */
+function spyOnLogger(method: "debug" | "info" | "warn" | "error"): any {
+  return jest
+    .spyOn(logger, method)
+    .mockImplementation((() => logger) as never);
+}
+
+function debugCalls(spy: any): DebugCall[] {
+  return (spy.mock.calls as DebugCall[]).filter((call) =>
+    String(call[0]).includes("poll diagnostics"),
+  );
+}
+
+function callFor(spy: any, operation: string): DebugCall[] {
+  return debugCalls(spy).filter((call) => call[1]?.operation === operation);
+}
+
+/** Pull a `key=value` token out of a diagnostic message string. */
+function readTag(message: string, key: string): string | undefined {
+  const match = new RegExp(`${key}=([^\\s]+)`).exec(message);
+  return match ? match[1] : undefined;
+}
+
+describe("sqlite_vacuum_cleaner – polling diagnostics (#346)", () => {
+  let debugSpy: any;
+
+  beforeEach(() => {
+    debugSpy = spyOnLogger("debug");
+  });
+
+  afterEach(() => {
+    debugSpy.mockRestore();
+  });
+
+  describe("logVacuumPollDiagnostics", () => {
+    it("logs a debug string containing elapsed time", () => {
+      logVacuumPollDiagnostics({
+        component: "sqlite_vacuum_cleaner",
+        operation: "run_vacuum",
+        status: "success",
+        elapsedMs: 12.345,
+      });
+
+      expect(debugSpy).toHaveBeenCalledTimes(1);
+      const [message, meta] = (debugSpy.mock.calls as DebugCall[])[0];
+
+      expect(message).toEqual(expect.stringContaining("elapsedMs=12.345"));
+      expect(message).toEqual(
+        expect.stringContaining("sqlite_vacuum_cleaner poll diagnostics"),
+      );
+      expect(message).toEqual(expect.stringContaining("operation=run_vacuum"));
+      expect(message).toEqual(expect.stringContaining("status=success"));
+      expect(meta).toMatchObject({
+        component: "sqlite_vacuum_cleaner",
+        operation: "run_vacuum",
+        status: "success",
+        elapsedMs: 12.345,
+      });
+    });
+
+    it("includes the pruned row count when one is supplied", () => {
+      logVacuumPollDiagnostics({
+        component: "sqlite_vacuum_cleaner",
+        operation: "prune_old_events",
+        status: "success",
+        elapsedMs: 1,
+        prunedEvents: 42,
+        retentionDays: 90,
+      });
+
+      const [message, meta] = (debugSpy.mock.calls as DebugCall[])[0];
+      expect(message).toEqual(expect.stringContaining("prunedEvents=42"));
+      expect(message).toEqual(expect.stringContaining("retentionDays=90"));
+      expect(meta.prunedEvents).toBe(42);
+    });
+
+    it("carries the error text on a failure diagnostic", () => {
+      logVacuumPollDiagnostics({
+        component: "sqlite_vacuum_cleaner",
+        operation: "prune_old_events",
+        status: "failure",
+        elapsedMs: 3,
+        error: "cannot VACUUM from within a transaction",
+      });
+
+      const [message, meta] = (debugSpy.mock.calls as DebugCall[])[0];
+      expect(message).toEqual(expect.stringContaining("status=failure"));
+      expect(meta.error).toBe("cannot VACUUM from within a transaction");
+    });
+  });
+
+  describe("cleanup cycle diagnostics", () => {
+    let testDb: Database.Database;
+    let beforeInsert: () => void;
+
+    beforeEach(() => {
+      testDb = new Database(":memory:");
+      setDb(testDb);
+      runMigrations();
+      beforeInsert = () => {
+        insertEvent("contract-1", "funded", 1, 1_600_000_000, "{}");
+        insertEvent("contract-1", "funded", 2, 1_700_000_000, "{}");
+      };
+      // Seed one old row (well past the retention window) and one fresh row.
+      testDb
+        .prepare(
+          `INSERT INTO events (contract_id, event_type, ledger_sequence, timestamp, data_json, created_at)
+           VALUES (?, ?, ?, ?, ?, datetime('now', '-999 days'))`,
+        )
+        .run("old-contract", "initialized", 1, 1_000_000_000, "{}");
+    });
+
+    afterEach(() => {
+      closeDb();
+    });
+
+    it("emits a started and a success diagnostic for prune_old_events", () => {
+      pruneOldEvents(testDb, 90);
+
+      const calls = callFor(debugSpy, "prune_old_events");
+      expect(calls).toHaveLength(1);
+      const [message, meta] = calls[0];
+      expect(meta.status).toBe("success");
+      expect(message).toEqual(expect.stringContaining("elapsedMs="));
+      expect(meta.prunedEvents).toBe(1);
+      expect(meta.retentionDays).toBe(90);
+    });
+
+    it("emits a success diagnostic for run_vacuum", () => {
+      runVacuum(testDb);
+
+      const [message, meta] = callFor(debugSpy, "run_vacuum")[0];
+      expect(meta.status).toBe("success");
+      expect(message).toEqual(expect.stringContaining("elapsedMs="));
+      expect(Number(readTag(message, "elapsedMs"))).toBeGreaterThanOrEqual(0);
+    });
+
+    it("emits a success diagnostic for prune_ledger_range", () => {
+      beforeInsert();
+      pruneEventsInLedgerRange(testDb, { startLedger: 1, endLedger: 1 });
+
+      const [message, meta] = callFor(debugSpy, "prune_ledger_range")[0];
+      expect(meta.status).toBe("success");
+      expect(message).toEqual(expect.stringContaining("elapsedMs="));
+      expect(message).toEqual(expect.stringContaining("startLedger=1"));
+      expect(message).toEqual(expect.stringContaining("endLedger=1"));
+      expect(meta.prunedEvents).toBe(2);
+    });
+
+    it("runVacuumCleanup emits the per-stage diagnostics plus a boundary", () => {
+      beforeInsert();
+      const result = runVacuumCleanup(testDb, { retentionDays: 90 });
+
+      expect(result.prunedEvents).toBe(1);
+      expect(result.vacuumed).toBe(true);
+
+      expect(callFor(debugSpy, "prune_old_events")).toHaveLength(1);
+      expect(callFor(debugSpy, "run_vacuum")).toHaveLength(1);
+
+      const boundary = callFor(debugSpy, "vacuum_cleanup");
+      expect(boundary).toHaveLength(2);
+      expect(boundary[0][1].status).toBe("started");
+      expect(boundary[1][1].status).toBe("success");
+      expect(boundary[1][0]).toEqual(expect.stringContaining("elapsedMs="));
+      expect(boundary[1][1].prunedEvents).toBe(1);
+      expect(Number(readTag(boundary[1][0], "elapsedMs"))).toBeGreaterThanOrEqual(0);
+    });
+
+    it("skips run_vacuum and emits failure diagnostics when pruning fails", () => {
+      testDb.exec("DROP TABLE events");
+
+      expect(() => runVacuumCleanup(testDb, { retentionDays: 90 })).toThrow();
+
+      const stage = callFor(debugSpy, "prune_old_events")[0];
+      expect(stage[1].status).toBe("failure");
+      expect(stage[0]).toEqual(expect.stringContaining("elapsedMs="));
+      expect(stage[1].error).toEqual(expect.stringContaining("events"));
+
+      expect(callFor(debugSpy, "run_vacuum")).toHaveLength(0);
+
+      const boundary = callFor(debugSpy, "vacuum_cleanup");
+      const failure = boundary[boundary.length - 1];
+      expect(failure[1].status).toBe("failure");
+      expect(failure[0]).toEqual(expect.stringContaining("elapsedMs="));
+      expect(failure[1].retentionDays).toBe(90);
+    });
+
+    it("every diagnostic message carries a numeric elapsedMs", () => {
+      beforeInsert();
+      runVacuumCleanup(testDb, { retentionDays: 90 });
+
+      const calls = debugCalls(debugSpy);
+      expect(calls.length).toBeGreaterThanOrEqual(4);
+      for (const [message, meta] of calls) {
+        const tag = readTag(message, "elapsedMs");
+        expect(tag).toBeDefined();
+        expect(Number.isNaN(Number(tag))).toBe(false);
+        expect((meta as VacuumPollDiagnostics).elapsedMs).toBeGreaterThanOrEqual(0);
+        expect(meta.component).toBe("sqlite_vacuum_cleaner");
+      }
+    });
+
+    it("keeps diagnostics at debug level so normal runs stay quiet", () => {
+      const warnSpy = spyOnLogger("warn");
+      const errorSpy = spyOnLogger("error");
+
+      beforeInsert();
+      runVacuumCleanup(testDb, { retentionDays: 90 });
+
+      expect(debugCalls(debugSpy).length).toBeGreaterThan(0);
+      expect(warnSpy).not.toHaveBeenCalled();
+      expect(errorSpy).not.toHaveBeenCalled();
+
+      warnSpy.mockRestore();
+      errorSpy.mockRestore();
+    });
+  });
+});

src/indexer/db.ts

@@ -439,6 +439,23 @@ export function verifySchemaUpToDate(): void {
 // Schema verification hooks (#264)
 // ---------------------------------------------------------------------------
 
+/**
+ * Index names created by the schema manager migrations. Exported so modules
+ * and tests can assert the exact lookup indexes the schema manager relies on
+ * without hardcoding names (#259).
+ */
+export const SCHEMA_MANAGER_INDEXES = [
+  "idx_events_contract_id",
+  "idx_events_ledger_sequence",
+  "idx_events_contract_ledger",
+  "idx_events_contract_type",
+  "idx_webhook_subscriptions_contract",
+  "idx_events_ledger_event_type",
+  "idx_monitored_contracts_active",
+  "idx_events_created_at",
+  "idx_events_contract_type_ledger",
+] as const;
+
 export interface SchemaVerificationResult {
   valid: boolean;
   missingTables: string[];