Harden watcher event-source failure handling and scan metrics

Author: warmidris

packages/stackflow-agent/README.md

@@ -135,7 +135,9 @@ const result = await agent.acceptIncomingTransfer({
 4. Watcher retries are idempotent for already-disputed closures (same closure txid is skipped on later polls).
 5. Read-only polling isolates per-pipe failures (`getPipeState` errors on one pipe do not stop others).
 6. Event scan mode intentionally holds the cursor when any dispute submission errors occur, so failed disputes are retried on next run.
-7. `buildOutgoingTransfer(...)` defaults `actor` to the tracked local principal and rejects mismatched actor values.
-8. Incoming transfer validation enforces tracked contract/pipe/principals/token consistency; mismatched `pipeId`, `pipeKey`, `actor`, or token payloads are rejected.
-9. Incoming transfer validation also enforces sequential nonces and balance invariants against the latest stored local state (same total balance, and counterparty-actor updates must not reduce local balance).
-10. For production hardening, add alerting, signer balance checks, and idempotency audit logs.
+7. Event scan mode now reports `listErrors` (event source/indexer failures) and keeps the watcher cursor unchanged on those failures.
+8. Invalid closure event payloads are skipped and counted in `invalidEvents` so one malformed record does not abort a full scan.
+9. `buildOutgoingTransfer(...)` defaults `actor` to the tracked local principal and rejects mismatched actor values.
+10. Incoming transfer validation enforces tracked contract/pipe/principals/token consistency; mismatched `pipeId`, `pipeKey`, `actor`, or token payloads are rejected.
+11. Incoming transfer validation also enforces sequential nonces and balance invariants against the latest stored local state (same total balance, and counterparty-actor updates must not reduce local balance).
+12. For production hardening, add alerting, signer balance checks, and idempotency audit logs.

packages/stackflow-agent/src/watcher.js

@@ -315,16 +315,34 @@ export class HourlyClosureWatcher {
     this.running = true;
     try {
       const fromBlockHeight = this.agentService.stateStore.getWatcherCursor();
-      const events = await this.listClosureEvents({
-        fromBlockHeight,
-      });
+      let events;
+      try {
+        events = await this.listClosureEvents({
+          fromBlockHeight,
+        });
+      } catch (error) {
+        this.reportError(error, "listClosureEvents");
+        return {
+          ok: false,
+          scanned: 0,
+          invalidEvents: 0,
+          disputesSubmitted: 0,
+          skippedAlreadyDisputed: 0,
+          disputeErrors: 0,
+          listErrors: 1,
+          fromBlockHeight,
+          toBlockHeight: fromBlockHeight,
+        };
+      }
       if (!Array.isArray(events) || events.length === 0) {
         return {
           ok: true,
           scanned: 0,
+          invalidEvents: 0,
           disputesSubmitted: 0,
           skippedAlreadyDisputed: 0,
           disputeErrors: 0,
+          listErrors: 0,
           fromBlockHeight,
           toBlockHeight: fromBlockHeight,
         };
@@ -334,6 +352,7 @@ export class HourlyClosureWatcher {
       let disputesSubmitted = 0;
       let skippedAlreadyDisputed = 0;
       let disputeErrors = 0;
+      let invalidEvents = 0;
       let hasDisputeErrors = false;
       let scanned = 0;
 
@@ -342,6 +361,7 @@ export class HourlyClosureWatcher {
         try {
           closure = normalizeClosureEvent(rawEvent);
         } catch {
+          invalidEvents += 1;
           continue;
         }
         scanned += 1;
@@ -382,9 +402,11 @@ export class HourlyClosureWatcher {
       return {
         ok: true,
         scanned,
+        invalidEvents,
         disputesSubmitted,
         skippedAlreadyDisputed,
         disputeErrors,
+        listErrors: 0,
         fromBlockHeight,
         toBlockHeight,
       };

tests/stackflow-agent.test.ts

@@ -510,6 +510,133 @@ describe("stackflow agent", () => {
     store.close();
   });
 
+  it("keeps event cursor and reports list source failures", async () => {
+    const dbFile = tempDbFile("agent-event-source-error");
+    const store = new AgentStateStore({ dbFile });
+
+    const errors: Error[] = [];
+    const agent = new StackflowAgentService({
+      stateStore: store,
+      signer: {
+        async submitDispute() {
+          return { txid: "0x1" };
+        },
+        async callContract() {
+          return { ok: true };
+        },
+      },
+      network: "devnet",
+    });
+
+    const watcher = new HourlyClosureWatcher({
+      agentService: agent,
+      listClosureEvents: async () => {
+        throw new Error("indexer timeout");
+      },
+      onError: (error) => {
+        errors.push(error as Error);
+      },
+    });
+
+    const result = await watcher.runOnce();
+    expect(result.ok).toBe(false);
+    expect(result.listErrors).toBe(1);
+    expect(result.scanned).toBe(0);
+    expect(result.toBlockHeight).toBe("0");
+    expect(store.getWatcherCursor()).toBe("0");
+    expect(errors).toHaveLength(1);
+
+    watcher.stop();
+    store.close();
+  });
+
+  it("counts invalid closure events without aborting scan", async () => {
+    const dbFile = tempDbFile("agent-event-invalid-event");
+    const store = new AgentStateStore({ dbFile });
+
+    const contractId = "ST1TESTABC.contract";
+    const pipeKey = {
+      "principal-1": "ST1LOCAL",
+      "principal-2": "ST1OTHER",
+      token: null,
+    };
+    const pipeId = buildPipeId({ contractId, pipeKey });
+
+    store.upsertTrackedPipe({
+      pipeId,
+      contractId,
+      pipeKey,
+      localPrincipal: "ST1LOCAL",
+      counterpartyPrincipal: "ST1OTHER",
+      token: null,
+    });
+    store.upsertSignatureState({
+      contractId,
+      pipeKey,
+      forPrincipal: "ST1LOCAL",
+      withPrincipal: "ST1OTHER",
+      token: null,
+      myBalance: "90",
+      theirBalance: "10",
+      nonce: "8",
+      action: "1",
+      actor: "ST1LOCAL",
+      mySignature: "0x" + "11".repeat(65),
+      theirSignature: "0x" + "22".repeat(65),
+      secret: null,
+      validAfter: null,
+      beneficialOnly: false,
+    });
+
+    let disputeCalls = 0;
+    const agent = new StackflowAgentService({
+      stateStore: store,
+      signer: {
+        async submitDispute() {
+          disputeCalls += 1;
+          return { txid: "0xdispute-ok" };
+        },
+        async callContract() {
+          return { ok: true };
+        },
+      },
+      network: "devnet",
+      disputeOnlyBeneficial: true,
+    });
+
+    const watcher = new HourlyClosureWatcher({
+      agentService: agent,
+      listClosureEvents: async () => [
+        {
+          eventName: "invalid",
+        },
+        {
+          contractId,
+          pipeKey,
+          eventName: "force-close",
+          nonce: "5",
+          closer: "ST1OTHER",
+          txid: "0xtx-valid",
+          blockHeight: "123",
+          expiresAt: "200",
+          closureMyBalance: "20",
+        },
+      ],
+    });
+
+    const result = await watcher.runOnce();
+    expect(result.ok).toBe(true);
+    expect(result.invalidEvents).toBe(1);
+    expect(result.scanned).toBe(1);
+    expect(result.disputesSubmitted).toBe(1);
+    expect(result.toBlockHeight).toBe("123");
+    expect(store.getWatcherCursor()).toBe("123");
+    expect(disputeCalls).toBe(1);
+
+    watcher.stop();
+    store.close();
+  });
+
   it("skips overlapping readonly watcher runs", async () => {
     const dbFile = tempDbFile("agent-readonly-overlap");
     const store = new AgentStateStore({ dbFile });