Fix metrics monitoring

scripts/grafana-dashboard.json

@@ -471,6 +471,104 @@
       ],
       "title": "Avg Batch Size",
       "type": "stat"
+    },
+    {
+      "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+      "fieldConfig": {
+        "defaults": {
+          "color": { "mode": "palette-classic" },
+          "custom": {
+            "axisBorderShow": false,
+            "axisCenteredZero": false,
+            "axisColorMode": "text",
+            "axisLabel": "",
+            "axisPlacement": "auto",
+            "barAlignment": 0,
+            "drawStyle": "bars",
+            "fillOpacity": 80,
+            "gradientMode": "none",
+            "hideFrom": { "legend": false, "tooltip": false, "viz": false },
+            "insertNulls": false,
+            "lineInterpolation": "linear",
+            "lineWidth": 1,
+            "pointSize": 5,
+            "scaleDistribution": { "type": "linear" },
+            "showPoints": "never",
+            "spanNulls": false,
+            "stacking": { "group": "A", "mode": "normal" },
+            "thresholdsStyle": { "mode": "off" }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              { "color": "green", "value": null }
+            ]
+          },
+          "unit": "short"
+        }
+      },
+      "gridPos": { "h": 10, "w": 12, "x": 0, "y": 30 },
+      "id": 9,
+      "options": {
+        "legend": { "calcs": ["sum"], "displayMode": "table", "placement": "bottom", "showLegend": true },
+        "tooltip": { "mode": "multi", "sort": "desc" }
+      },
+      "pluginVersion": "11.0.0",
+      "targets": [
+        {
+          "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+          "editorMode": "code",
+          "expr": "increase(keeper_charge_results_total[$__rate_interval])",
+          "legendFormat": "{{result}}",
+          "range": true,
+          "refId": "A"
+        }
+      ],
+      "title": "Charge Results Breakdown",
+      "type": "timeseries"
+    },
+    {
+      "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+      "fieldConfig": {
+        "defaults": {
+          "color": { "mode": "thresholds" },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              { "color": "green", "value": null },
+              { "color": "orange", "value": 1 },
+              { "color": "red", "value": 5 }
+            ]
+          },
+          "unit": "short"
+        }
+      },
+      "gridPos": { "h": 10, "w": 12, "x": 12, "y": 30 },
+      "id": 10,
+      "options": {
+        "colorMode": "background",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "percentChangeColorMode": "standard",
+        "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false },
+        "showPercentChange": false,
+        "textMode": "auto",
+        "wideLayout": true
+      },
+      "pluginVersion": "11.0.0",
+      "targets": [
+        {
+          "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+          "editorMode": "code",
+          "expr": "rate(keeper_rpc_failovers_total[$__rate_interval])",
+          "refId": "A"
+        }
+      ],
+      "title": "RPC Failover Rate",
+      "type": "stat"
     }
   ],
   "refresh": "10s",

scripts/keeper.ts

@@ -53,6 +53,13 @@ import { MultiEndpointServer } from "./rpc-client.js";
 import { buildOptimizedBatches } from "./batch-optimizer";
 import { Server, assembleTransaction } from "@stellar/stellar-sdk/rpc";
 import { buildOptimizedBatches } from "./batch-optimizer.js";
+import {
+  startMetricsServer,
+  recordBatchCharge,
+  recordChargeResults,
+  incrementCycles,
+  setActiveSubscribers,
+} from "./metrics-server";
 import {
   Address,
   Contract,
@@ -656,14 +663,26 @@ async function processPageDryRun(
 
   if (users.length === 0) return result;
 
-  const { results, amounts } = await simulateBatchCharge(users);
+  const startMs = Date.now();
+  try {
+    const { results, amounts } = await simulateBatchCharge(users);
 
-  // results is index-aligned with users (same order, one entry per input).
-  let amountIdx = 0;
-  for (let i = 0; i < results.length; i++) {
-    const variant = results[i];
-    const user = i < users.length ? users[i] : "unknown";
+    // results is index-aligned with users (same order, one entry per input).
+    let amountIdx = 0;
+    for (let i = 0; i < results.length; i++) {
+      const variant = results[i];
+      const user = i < users.length ? users[i] : "unknown";
 
+      if (variant === "Charged") {
+        const amt = amountIdx < amounts.length ? amounts[amountIdx] : 0n;
+        result.wouldCharge++;
+        result.totalVolume += amt;
+        result.candidates.push({ user, result: variant, amountStroops: amt.toString() });
+        amountIdx++;
+      } else {
+        result.skipCounts[variant] = (result.skipCounts[variant] || 0) + 1;
+        result.candidates.push({ user, result: variant, amountStroops: "0" });
+      }
     if (variant === "Charged") {
       const amt = amountIdx < amounts.length ? amounts[amountIdx] : 0n;
       result.wouldCharge++;
@@ -678,6 +697,26 @@ async function processPageDryRun(
       result.skipCounts[variant] = (result.skipCounts[variant] || 0) + 1;
       result.candidates.push({ user, result: variant, amountStroops: "0" });
     }
+
+    const durationMs = Date.now() - startMs;
+    recordBatchCharge({
+      status: "success",
+      charged: result.wouldCharge,
+      skipped: users.length - result.wouldCharge,
+      durationMs,
+    });
+    recordChargeResults({ Charged: result.wouldCharge, ...result.skipCounts });
+  } catch (err) {
+    const durationMs = Date.now() - startMs;
+    recordBatchCharge({
+      status: "failed",
+      charged: 0,
+      skipped: 0,
+      durationMs,
+      rpcError: true,
+    });
+    const errorStr = err instanceof Error ? err.message : String(err);
+    result.errors.push(`Page ${pageOffset}: ${errorStr}`);
   }
 
   return result;
@@ -697,6 +736,8 @@ async function processPageLive(
 
   if (users.length === 0) return result;
 
+  const startMs = Date.now();
+
   try {
     const { results, amounts, txHash } = await submitBatchCharge(users);
     result.txHash = txHash;
@@ -722,7 +763,26 @@ async function processPageLive(
         result.candidates.push({ user, result: variant, amountStroops: "0" });
       }
     }
+
+    const durationMs = Date.now() - startMs;
+    recordBatchCharge({
+      status: "success",
+      charged: result.charged,
+      skipped: users.length - result.charged,
+      durationMs,
+    });
+    recordChargeResults({ Charged: result.charged, ...result.skipCounts });
+
   } catch (err) {
+    const durationMs = Date.now() - startMs;
+    recordBatchCharge({
+      status: "failed",
+      charged: 0,
+      skipped: 0,
+      durationMs,
+      rpcError: true,
+    });
+
     const errorStr = err instanceof Error ? err.message : String(err);
     result.errors.push(`Page ${pageOffset}: ${errorStr}`);
     let ledgerSeq: number | undefined;
@@ -825,6 +885,8 @@ async function runCycle(): Promise<CycleReport> {
 
   // ── Post-cycle reporting ───────────────────────────────────────────────────
 
+  setActiveSubscribers(report.totalChecked);
+
   if (!isDryRun) {
     writeLatestLive(report);
   } else {
@@ -1149,17 +1211,20 @@ async function main(): Promise<void> {
     log(false, "Keeper started in LIVE mode");
   }
 
+  startMetricsServer();
   BATCH_SIZE = await resolveBatchSize();
   log(DRY_RUN, `Effective legacy page size (BATCH_SIZE): ${BATCH_SIZE}`);
 
   if (once) {
     const report = await runCycle();
+    incrementCycles();
     process.exit(report.errors.length > 0 && report.totalCharged === 0 ? 1 : 0);
   }
 
   // Loop mode
   while (true) {
     const report = await runCycle();
+    incrementCycles();
     const nextRun = new Date(Date.now() + INTERVAL_SECONDS * 1000);
     log(
       DRY_RUN,

scripts/metrics-server.ts

@@ -115,6 +115,21 @@ const indexerDedupEvictionsTotal = new Counter({
   registers: [registry],
 });
 
+/** Total RPC failovers across multiple endpoints. */
+const rpcFailoversTotal = new Counter({
+  name: "keeper_rpc_failovers_total",
+  help: "Total number of RPC failovers triggered",
+  registers: [registry],
+});
+
+/** Granular outcomes for each subscriber checked. */
+const chargeResultsTotal = new Counter({
+  name: "keeper_charge_results_total",
+  help: "Total number of charge outcomes labeled by specific contract result",
+  labelNames: ["result"] as const,
+  registers: [registry],
+});
+
 // ── Public API for the keeper run loop ───────────────────────────────────────
 
 /**
@@ -149,11 +164,23 @@ export function recordBatchCharge(params: {
   }
 }
 
+/** Record specific granular charge results. */
+export function recordChargeResults(results: Record<string, number>): void {
+  for (const [result, count] of Object.entries(results)) {
+    chargeResultsTotal.inc({ result }, count);
+  }
+}
+
 /** Increment the RPC error counter. */
 export function incrementRpcErrors(): void {
   rpcErrorsTotal.inc(1);
 }
 
+/** Increment the RPC failovers counter. */
+export function incrementRpcFailovers(): void {
+  rpcFailoversTotal.inc(1);
+}
+
 /** Set the current active subscriber count gauge. */
 export function setActiveSubscribers(count: number): void {
   activeSubscribers.set(count);

scripts/rpc-client.ts

@@ -8,6 +8,7 @@ import {
   Networks,
 } from "@stellar/stellar-sdk";
 import { Server, Durability, Api } from "@stellar/stellar-sdk/rpc";
+import { incrementRpcFailovers } from "./metrics-server";
 
 /**
  * Interface representing state of a single RPC endpoint.
@@ -167,6 +168,7 @@ export class MultiEndpointServer {
         console.warn(
           `[RPC Failover] Endpoint ${failedUrl} failed: ${err?.message || err}. Retrying with ${nextUrl}...`,
         );
+        incrementRpcFailovers();
       }
     }