Merge pull request #8 from Apex-Fusion/fix/wallet-health-idempotence

fix(wallet): make wallet-health idempotent - structural health predicate + circuit breakers
2026-08-10 14:11:59 +02:00
parent 4d94aa5040 36c5cfc6be
commit 6ca7f2927a
4 changed files with 331 additions and 30 deletions
--- a/packages/shared/src/tx/wallet/planConsolidate.ts
+++ b/packages/shared/src/tx/wallet/planConsolidate.ts
@@ -2,9 +2,22 @@
 * planConsolidate.ts — pure pre-flight planner for the wallet-consolidate flow.
 *
 * Given a wallet's current UTxO set + a target collateral lovelace, decide
- * whether to consolidate (many → 2), split (1 → 2), or skip (already at
- * {≥collateral, working}). The same downstream tx-builder handles
- * consolidate and split — only the "reason" tag differs for logging.
+ * whether the wallet needs reshaping or is already usable for script spends.
+ *
+ * Healthy is a STRUCTURAL predicate, not a UTxO count (F7, 2026-08-08):
+ *   (a) some pure-lovelace UTxO >= collateralLovelace exists (the collateral
+ *       candidate lucid's setCollateral needs), AND
+ *   (b) at least MIN_WORKING_LOVELACE sits OUTSIDE that candidate (fee and
+ *       spend money that is not the candidate), AND
+ *   (c) fragmentation is bounded (utxos.length <= maxUtxos).
+ *
+ * The earlier revision returned `already-healthy` only for EXACTLY 2 UTxOs.
+ * The consolidate tx itself lands 3 (collateral + working + the fee-reserve
+ * remainder as change), so the shape consolidation produced could never
+ * satisfy its own health check — the wallet-health ticker re-consolidated
+ * every tick and burned 0.5 AP3X per pass (116 AP3X in one 24h soak).
+ * The health check also runs BEFORE the low-balance guard now: a wallet that
+ * can already transact must never be told "balance too low to consolidate".
 *
 * No lucid / WASM deps so this is unit-testable from a vanilla vitest.
 */
@@ -26,12 +39,24 @@ export interface PlannedConsolidate {

 export const DEFAULT_COLLATERAL_LOVELACE = 5_000_000n;
 export const DEFAULT_FEE_RESERVE = 2_000_000n;
+/** Reshape only when fragmentation passes this; receipt/change UTxOs
+ *  accumulating between jobs are fine as long as a candidate exists. */
+export const DEFAULT_MAX_UTXOS = 6;
 const MIN_WORKING_LOVELACE = 1_000_000n;

+/** Lovelace of a UTxO iff it carries no native assets; null otherwise. */
+function pureLovelace(u: UtxoLike): bigint | null {
+  for (const [unit, amount] of Object.entries(u.assets)) {
+    if (unit !== "lovelace" && (amount ?? 0n) > 0n) return null;
+  }
+  return u.assets.lovelace ?? 0n;
+}
+
 export function planConsolidate(
  utxos: UtxoLike[],
  collateralLovelace: bigint = DEFAULT_COLLATERAL_LOVELACE,
  feeReserve: bigint = DEFAULT_FEE_RESERVE,
+  maxUtxos: number = DEFAULT_MAX_UTXOS,
 ): PlannedConsolidate {
  if (utxos.length === 0) {
    throw new Error("wallet has no UTxOs — fund the address first");
@@ -42,6 +67,32 @@ export function planConsolidate(
    0n,
  );

+  // Structural health check first — reshaping a usable wallet wastes a fee.
+  if (utxos.length <= maxUtxos) {
+    // Smallest qualifying candidate: leave larger UTxOs as working funds.
+    let candidate: bigint | null = null;
+    for (const u of utxos) {
+      const pure = pureLovelace(u);
+      if (pure !== null && pure >= collateralLovelace) {
+        if (candidate === null || pure < candidate) candidate = pure;
+      }
+    }
+    if (candidate !== null) {
+      const workingOutside = totalLovelaceIn - candidate;
+      if (workingOutside >= MIN_WORKING_LOVELACE) {
+        return {
+          inputCount: utxos.length,
+          totalLovelaceIn,
+          collateralOutput: candidate,
+          workingOutput: workingOutside,
+          alreadyHealthy: true,
+          reason: "already-healthy",
+        };
+      }
+    }
+  }
+
+  // Reshape needed — only an actual reshape has a funds threshold.
  const required = collateralLovelace + feeReserve + MIN_WORKING_LOVELACE;
  if (totalLovelaceIn < required) {
    throw new Error(
@@ -51,23 +102,6 @@ export function planConsolidate(
    );
  }

-  if (utxos.length === 2) {
-    const lov0 = utxos[0].assets.lovelace ?? 0n;
-    const lov1 = utxos[1].assets.lovelace ?? 0n;
-    const smaller = lov0 < lov1 ? lov0 : lov1;
-    const larger = lov0 < lov1 ? lov1 : lov0;
-    if (smaller >= collateralLovelace) {
-      return {
-        inputCount: 2,
-        totalLovelaceIn,
-        collateralOutput: smaller,
-        workingOutput: larger,
-        alreadyHealthy: true,
-        reason: "already-healthy",
-      };
-    }
-  }
-
  const workingOutput = totalLovelaceIn - collateralLovelace - feeReserve;

  return {
--- a/supplier/src/walletHealth.ts
+++ b/supplier/src/walletHealth.ts
@@ -7,15 +7,26 @@
 * and the lane stalls until an operator runs `tx:consolidate-wallet`.
 *
 * Two layers of self-healing:
- *   startWalletHealthTicker — periodic background tick (every N ms). When
- *     the supplier is idle, runs runConsolidateWallet. planConsolidate
- *     returns `already-healthy` cheaply for clean 2-UTxO wallets, so this
- *     is a no-op in steady state — no tx, no fee.
+ *   startWalletHealthTicker — periodic background tick (every N ms). Runs
+ *     runConsolidateWallet; planConsolidate's structural health predicate
+ *     short-circuits without a tx when a collateral candidate exists and
+ *     fragmentation is bounded. A tick on a healthy wallet costs one UTxO
+ *     query and nothing else.
 *   triggerOnFailureConsolidate — fire-and-forget consolidate from the
 *     Claim-failure path. Debounced so back-to-back failures don't queue
 *     multiple consolidates. Returns 503 to the buyer for THIS request; the
 *     next buyer retry (~30-60s later) finds the wallet healthy.
 *
+ * F7 guards (2026-08-08 incident: a non-idempotent health check made every
+ * tick consolidate, burning 0.5 AP3X per 45s until the wallet hit the
+ * floor — 116 AP3X in one soak). Two independent circuit breakers protect
+ * against ANY future non-convergence:
+ *   - consecutive guard: 3 consolidations in a row without ever reaching
+ *     `already-healthy` halts wallet-health for this process.
+ *   - budget guard: more than 6 consolidations inside 24h halts it too.
+ * A halt logs loudly once; consolidate manually and restart the supplier
+ * to re-arm.
+ *
 * Wallet-lock discipline: consolidation spends EVERY wallet UTxO, so it runs
 * inside state.walletMutex — the same mutex that serializes Claims/Submits —
 * queueing behind any in-flight chain op instead of skipping. Session slots
@@ -63,6 +74,29 @@ export interface WalletHealthTicker {
 const inFlight = new WeakSet<SupplierState>();
 const lastOnFailureAt = new WeakMap<SupplierState, number>();

+// ── F7 circuit breakers ──────────────────────────────────────────────
+const MAX_CONSECUTIVE_CONSOLIDATIONS = 3;
+const MAX_CONSOLIDATIONS_PER_WINDOW = 6;
+const BUDGET_WINDOW_MS = 24 * 60 * 60 * 1000;
+
+interface GuardCounters {
+  consecutive: number;
+  windowStart: number;
+  windowCount: number;
+  halted: boolean;
+}
+
+const guardCounters = new WeakMap<SupplierState, GuardCounters>();
+
+function guardsFor(state: SupplierState): GuardCounters {
+  let g = guardCounters.get(state);
+  if (!g) {
+    g = { consecutive: 0, windowStart: Date.now(), windowCount: 0, halted: false };
+    guardCounters.set(state, g);
+  }
+  return g;
+}
+
 function defaultLog(line: string): void {
  // eslint-disable-next-line no-console
  console.log(`[wallet-health] ${line}`);
@@ -73,6 +107,8 @@ async function consolidateOnce(
  opts: { collateralLovelace: bigint; awaitTimeoutMs: number; log: (line: string) => void },
 ): Promise<void> {
  const { state } = deps;
+  const guards = guardsFor(state);
+  if (guards.halted) return; // the halt already logged loudly, once
  if (inFlight.has(state)) {
    opts.log("skip: another consolidate already in flight");
    return;
@@ -92,11 +128,36 @@ async function consolidateOnce(
      }),
    );
    if (result.reason === "already-healthy") {
+      guards.consecutive = 0;
      opts.log("already-healthy");
    } else {
+      const now = Date.now();
+      if (now - guards.windowStart > BUDGET_WINDOW_MS) {
+        guards.windowStart = now;
+        guards.windowCount = 0;
+      }
+      guards.consecutive += 1;
+      guards.windowCount += 1;
      opts.log(`consolidated: ${result.reason} txHash=${result.txHash}`);
+      if (guards.consecutive >= MAX_CONSECUTIVE_CONSOLIDATIONS) {
+        guards.halted = true;
+        opts.log(
+          `HALT: ${guards.consecutive} consecutive consolidations without reaching a healthy shape — ` +
+            `wallet-health stopped to protect funds (F7 guard). Investigate planConsolidate vs the ` +
+            `builder's output shape, consolidate manually, restart the supplier to re-arm.`,
+        );
+      } else if (guards.windowCount >= MAX_CONSOLIDATIONS_PER_WINDOW) {
+        guards.halted = true;
+        opts.log(
+          `HALT: consolidation budget exhausted (${guards.windowCount} in 24h) — wallet-health ` +
+            `stopped to protect funds (F7 guard). Restart the supplier to re-arm.`,
+        );
+      }
    }
  } catch (err) {
+    // Neither healthy nor consolidated (e.g. balance too low): no fee was
+    // spent, so the guards don't move — but nothing converged either, so
+    // the consecutive counter is NOT reset.
    opts.log(`consolidate failed: ${(err as Error).message}`);
  } finally {
    inFlight.delete(state);
@@ -143,8 +204,9 @@ export function triggerOnFailureConsolidate(
  });
 }

-/** Test-only: reset the in-flight + debounce maps between tests. */
+/** Test-only: reset the in-flight + debounce + guard maps between tests. */
 export function _resetWalletHealthForTests(state: SupplierState): void {
  inFlight.delete(state);
  lastOnFailureAt.delete(state);
+  guardCounters.delete(state);
 }
--- a/tests/unit/supplier-walletHealth.test.ts
+++ b/tests/unit/supplier-walletHealth.test.ts
@@ -242,3 +242,115 @@ describe("walletHealth — triggerOnFailureConsolidate", () => {
    expect(state.snapshot().status).toBe("working");
  });
 });
+
+describe("walletHealth — F7 circuit breakers", () => {
+  let state: SupplierState;
+  let logLines: string[];
+  let log: (line: string) => void;
+  let consolidate: ReturnType<typeof vi.fn> & ConsolidateFn;
+
+  beforeEach(() => {
+    vi.useFakeTimers();
+    state = new SupplierState();
+    _resetWalletHealthForTests(state);
+    logLines = [];
+    log = (line) => logLines.push(line);
+    consolidate = vi.fn() as never;
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("halts after 3 consecutive consolidations that never reach healthy", async () => {
+    // The 2026-08-07 pump: every tick consolidates, none lands healthy.
+    consolidate.mockResolvedValue(consolidatedResult);
+
+    const ticker = startWalletHealthTicker(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { intervalMs: 1_000, log },
+    );
+
+    await vi.advanceTimersByTimeAsync(3_000);
+    await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(3));
+    expect(logLines.some((l) => l.includes("HALT") && l.includes("consecutive"))).toBe(true);
+
+    // Two more ticks: the breaker holds, no further txs.
+    await vi.advanceTimersByTimeAsync(2_000);
+    expect(consolidate).toHaveBeenCalledTimes(3);
+
+    ticker.stop();
+  });
+
+  it("an already-healthy result resets the consecutive counter", async () => {
+    consolidate
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockResolvedValueOnce(healthyResult)
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockResolvedValue(healthyResult);
+
+    const ticker = startWalletHealthTicker(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { intervalMs: 1_000, log },
+    );
+
+    // 6 ticks: c,c,h,c,c,h — never 3 consecutive, so never halted.
+    await vi.advanceTimersByTimeAsync(6_000);
+    await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(6));
+    expect(logLines.some((l) => l.includes("HALT"))).toBe(false);
+
+    ticker.stop();
+  });
+
+  it("halts when the 24h consolidation budget is exhausted", async () => {
+    // Interleave healthy results so the consecutive guard never fires:
+    // c,c,h,c,c,h,c,c -> 6 consolidations inside the window at tick 8.
+    consolidate
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockResolvedValueOnce(healthyResult)
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockResolvedValueOnce(healthyResult)
+      .mockResolvedValue(consolidatedResult);
+
+    const ticker = startWalletHealthTicker(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { intervalMs: 1_000, log },
+    );
+
+    await vi.advanceTimersByTimeAsync(8_000);
+    await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(8));
+    expect(logLines.some((l) => l.includes("HALT") && l.includes("budget"))).toBe(true);
+
+    // Breaker holds on subsequent ticks.
+    await vi.advanceTimersByTimeAsync(2_000);
+    expect(consolidate).toHaveBeenCalledTimes(8);
+
+    ticker.stop();
+  });
+
+  it("failed consolidations neither trip nor reset the breakers", async () => {
+    // Errors spend no fee: they must not count as consolidations, but they
+    // must not clear the consecutive count either.
+    consolidate
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockResolvedValueOnce(consolidatedResult)
+      .mockRejectedValueOnce(new Error("balance too low to consolidate"))
+      .mockResolvedValue(consolidatedResult);
+
+    const ticker = startWalletHealthTicker(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { intervalMs: 1_000, log },
+    );
+
+    // c,c,error,c -> the 4th call is the 3rd consecutive consolidation.
+    await vi.advanceTimersByTimeAsync(4_000);
+    await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(4));
+    expect(logLines.some((l) => l.includes("HALT") && l.includes("consecutive"))).toBe(true);
+
+    ticker.stop();
+  });
+});
--- a/tests/unit/tx-consolidate-wallet.test.ts
+++ b/tests/unit/tx-consolidate-wallet.test.ts
@@ -64,14 +64,17 @@ describe("planConsolidate", () => {
    expect(plan.workingOutput).toBe(8_650_000n);
  });

-  it("consolidates a 2-UTxO wallet where the smaller is below collateral threshold", () => {
-    // Both fit budget but smaller (3 AP3X) is below 5 AP3X — needs reshape.
+  it("accepts a 2-UTxO wallet where the LARGER is the collateral candidate", () => {
+    // {3, 15}: the 15 qualifies as a pure >=5 candidate and the 3 covers
+    // fees — this wallet can run script txs as-is. The old exact-2-slot
+    // convention reshaped it for nothing (0.5 AP3X per pass).
    const utxos = [u(3_000_000n), u(15_000_000n)];
    const plan = planConsolidate(utxos);

-    expect(plan.reason).toBe("consolidate");
-    expect(plan.alreadyHealthy).toBe(false);
-    expect(plan.inputCount).toBe(2);
+    expect(plan.reason).toBe("already-healthy");
+    expect(plan.alreadyHealthy).toBe(true);
+    expect(plan.collateralOutput).toBe(15_000_000n);
+    expect(plan.workingOutput).toBe(3_000_000n);
  });

  it("throws on 0 UTxOs", () => {
@@ -93,3 +96,93 @@ describe("planConsolidate", () => {
    expect(plan.workingOutput).toBe(20_000_000n - 10_000_000n - DEFAULT_FEE_RESERVE);
  });
 });
+
+describe("planConsolidate — structural health predicate (F7)", () => {
+  // The tx buildConsolidateWalletTx actually lands: {collateral 5.0,
+  // working = total - 5.0 - 2.0 reserve, change = reserve - fee}. The fee
+  // observed on mainnet is exactly 501,628, so the change is 1,498,372.
+  const MAINNET_CONSOLIDATE_FEE = 501_628n;
+
+  it("REGRESSION: the exact shape consolidation produces is healthy", () => {
+    // This is the loop that burned 116 AP3X in the 2026-08-07 soak: the
+    // planner called its own output shape unhealthy and re-consolidated
+    // every 45s tick. Feed it precisely what the builder lands.
+    const utxos = [
+      u(5_000_000n),
+      u(91_000_000n),
+      u(DEFAULT_FEE_RESERVE - MAINNET_CONSOLIDATE_FEE), // 1,498,372 change
+    ];
+    const plan = planConsolidate(utxos);
+
+    expect(plan.reason).toBe("already-healthy");
+    expect(plan.alreadyHealthy).toBe(true);
+    expect(plan.collateralOutput).toBe(5_000_000n);
+  });
+
+  it("is idempotent: applying the builder's output model to any plan yields healthy", () => {
+    const fragmented = [u(2_400_750n), u(4_100_000n), u(1_679_301n), u(1_679_301n)];
+    const plan = planConsolidate(fragmented);
+    expect(plan.alreadyHealthy).toBe(false);
+
+    // Model of what buildConsolidateWalletTx lands for this plan.
+    const afterConsolidation = [
+      u(plan.collateralOutput),
+      u(plan.workingOutput),
+      u(DEFAULT_FEE_RESERVE - MAINNET_CONSOLIDATE_FEE),
+    ];
+    const replan = planConsolidate(afterConsolidation);
+
+    expect(replan.reason).toBe("already-healthy");
+    expect(replan.alreadyHealthy).toBe(true);
+  });
+
+  it("still consolidates when fragmentation exceeds the max UTxO bound", () => {
+    // Candidate present, but 7 UTxOs > DEFAULT_MAX_UTXOS (6).
+    const utxos = [
+      u(5_000_000n), u(1_100_000n), u(1_100_000n), u(1_100_000n),
+      u(1_100_000n), u(1_100_000n), u(1_100_000n),
+    ];
+    const plan = planConsolidate(utxos);
+    expect(plan.reason).toBe("consolidate");
+  });
+
+  it("still consolidates when no pure UTxO reaches the collateral threshold", () => {
+    // The 2026-08-07 batch-stall shape: fragments only, max 4,498,372.
+    const utxos = [u(4_498_372n), u(3_098_372n), u(1_498_372n)];
+    const plan = planConsolidate(utxos);
+    expect(plan.reason).toBe("consolidate");
+    expect(plan.alreadyHealthy).toBe(false);
+  });
+
+  it("a UTxO 1,628 lovelace short of collateral is not a candidate", () => {
+    // Seen live: 4,998,372 (= 5.0 minus a 501,628 fee twice-shuffled).
+    const utxos = [u(4_998_372n), u(9_000_000n - 4_998_372n)];
+    const plan = planConsolidate(utxos);
+    expect(plan.reason).toBe("consolidate");
+  });
+
+  it("a UTxO carrying native assets is not a collateral candidate", () => {
+    const withToken: UtxoLike = {
+      assets: { lovelace: 6_000_000n, "deadbeef.72737456": 1n },
+    };
+    const utxos = [withToken, u(4_000_000n)];
+    const plan = planConsolidate(utxos);
+    expect(plan.reason).toBe("consolidate");
+  });
+
+  it("healthy shape below the reshape threshold is healthy, not an error", () => {
+    // {5.0, 2.5} = 7.5 total: under the 8.0 reshape minimum but perfectly
+    // able to run script txs. The old order threw "balance too low" here
+    // on every tick (1,744 log lines in the soak).
+    const utxos = [u(5_000_000n), u(2_500_000n)];
+    const plan = planConsolidate(utxos);
+    expect(plan.reason).toBe("already-healthy");
+  });
+
+  it("candidate with only dust outside it needs a reshape it cannot afford", () => {
+    // {5.0, 0.4}: no working money outside the candidate, and 5.4 total is
+    // below the reshape minimum — loud error, funds needed.
+    const utxos = [u(5_000_000n), u(400_000n)];
+    expect(() => planConsolidate(utxos)).toThrow(/balance too low/i);
+  });
+});