Two structural fixes for demo concurrency (3 single-slot suppliers thrashed
under Open WebUI's ~3 conversations per message):
Multi-slot suppliers (MAX_CHAT_SESSIONS, default 1):
- SupplierState is now capacity-aware: a Set of active escrow refs with
free-until-full status semantics, ref-scoped release(ref) (also fixes the
latent bug where handler catch-alls released another request's slot), and
a walletMutex serializing every wallet-spending chain op — Claim, Submit,
and wallet-health consolidation queue instead of racing the single
working/collateral UTxO. /status gains active_sessions/max_sessions.
Entry-ticket settle mode (CHAT_SETTLE_MODE=ticket, default full, both sides):
- The buyer's escrow post at session open is the ONLY chain op. The supplier
verifies the Open escrow as an entry ticket but never Claims/Submits
(/v1/chat/start → {status:"ticket"}; /v1/chat/end → receipt-less
{status:"ended"}); no hard-cap timer, so sessions may outlive deliver_by.
Used tickets can't reopen sessions (records retained; 409 session_exists).
- Buyer SDK: startChat/endChat are settle-mode aware, driven by the
supplier's response — mixed-mode rollout windows settle correctly.
- Gateway: ticket closes bill cost 0; the sweeper Reclaims Open tickets
after deliver_by WITHOUT closing live sessions (the conversation and its
affinity mirror survive their reclaimed ticket), gates reclaims on
deliver_by, and keeps an in-process settled-refs memory so terminal
indexer rows aren't retried every sweep forever.
Deployment (kimi-chat trio + gateway): ticket mode, 8 slots per supplier,
idle policy gateway 10 min / supplier 15 min (fixes the inversion where the
supplier idle-closed before the gateway janitor).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
245 lines
7.5 KiB
TypeScript
245 lines
7.5 KiB
TypeScript
/**
|
|
* supplier-walletHealth.test.ts — Tests for the periodic + on-failure
|
|
* wallet auto-consolidation hooks.
|
|
*
|
|
* Coverage:
|
|
* - Ticker calls consolidate when idle, logs already-healthy
|
|
* - Consolidation runs even while session slots are held (wallet ops don't
|
|
* consume session slots) but queues behind a held walletMutex
|
|
* - Failure paths never leave the walletMutex poisoned
|
|
* - On-failure trigger fires consolidate the first time
|
|
* - On-failure trigger debounces back-to-back invocations
|
|
*
|
|
* The consolidate function is injected via deps so the test never touches
|
|
* a real Ogmios.
|
|
*/
|
|
|
|
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
|
|
|
import { SupplierState } from "../../supplier/src/state.js";
|
|
import {
|
|
startWalletHealthTicker,
|
|
triggerOnFailureConsolidate,
|
|
_resetWalletHealthForTests,
|
|
type ConsolidateFn,
|
|
} from "../../supplier/src/walletHealth.js";
|
|
|
|
const FAKE_CHAIN = {} as never;
|
|
const FAKE_KEY = {
|
|
privateKeyHex: "0".repeat(64),
|
|
pubKeyHex: "1".repeat(64),
|
|
pubKeyHash: "2".repeat(56),
|
|
address: "addr1_fake",
|
|
} as never;
|
|
const ESCROW_REF = `${"f".repeat(64)}#0`;
|
|
|
|
const healthyResult = {
|
|
txHash: null,
|
|
collateralRef: null,
|
|
workingRef: null,
|
|
reason: "already-healthy" as const,
|
|
inputCount: 2,
|
|
totalLovelaceIn: 200_000_000n,
|
|
collateralOutputLovelace: 5_000_000n,
|
|
workingOutputLovelace: 195_000_000n,
|
|
};
|
|
|
|
const consolidatedResult = {
|
|
txHash: "a".repeat(64),
|
|
collateralRef: `${"a".repeat(64)}#0`,
|
|
workingRef: `${"a".repeat(64)}#1`,
|
|
reason: "consolidate" as const,
|
|
inputCount: 14,
|
|
totalLovelaceIn: 328_530_235n,
|
|
collateralOutputLovelace: 5_000_000n,
|
|
workingOutputLovelace: 322_530_235n,
|
|
};
|
|
|
|
describe("walletHealth — ticker", () => {
|
|
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("calls consolidate when idle and logs already-healthy", async () => {
|
|
consolidate.mockResolvedValue(healthyResult);
|
|
|
|
const ticker = startWalletHealthTicker(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ intervalMs: 1_000, log },
|
|
);
|
|
|
|
await vi.advanceTimersByTimeAsync(1_000);
|
|
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
|
|
|
|
expect(logLines).toContain("already-healthy");
|
|
expect(state.snapshot().status).toBe("free");
|
|
|
|
ticker.stop();
|
|
});
|
|
|
|
it("runs even while a session slot is held (wallet ops don't consume slots)", async () => {
|
|
state.tryAcquire(ESCROW_REF);
|
|
consolidate.mockResolvedValue(healthyResult);
|
|
|
|
const ticker = startWalletHealthTicker(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ intervalMs: 1_000, log },
|
|
);
|
|
|
|
await vi.advanceTimersByTimeAsync(1_000);
|
|
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
|
|
|
|
// The held session slot is untouched.
|
|
expect(state.snapshot().status).toBe("working");
|
|
expect(state.snapshot().activeEscrowRefs).toEqual([ESCROW_REF]);
|
|
|
|
ticker.stop();
|
|
});
|
|
|
|
it("queues behind a held walletMutex instead of racing it", async () => {
|
|
consolidate.mockResolvedValue(healthyResult);
|
|
let releaseChainOp: () => void = () => undefined;
|
|
const chainOp = state.walletMutex.run(
|
|
() => new Promise<void>((resolve) => { releaseChainOp = resolve; }),
|
|
);
|
|
|
|
const ticker = startWalletHealthTicker(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ intervalMs: 1_000, log },
|
|
);
|
|
|
|
await vi.advanceTimersByTimeAsync(1_000);
|
|
// The mutex is held by the fake chain op — consolidate must wait.
|
|
expect(consolidate).not.toHaveBeenCalled();
|
|
|
|
releaseChainOp();
|
|
await chainOp;
|
|
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
|
|
|
|
ticker.stop();
|
|
});
|
|
|
|
it("logs success after a real consolidate", async () => {
|
|
consolidate.mockResolvedValue(consolidatedResult);
|
|
|
|
const ticker = startWalletHealthTicker(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ intervalMs: 1_000, log },
|
|
);
|
|
|
|
await vi.advanceTimersByTimeAsync(1_000);
|
|
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
|
|
|
|
expect(state.snapshot().status).toBe("free");
|
|
expect(logLines.some((l) => l.startsWith("consolidated:"))).toBe(true);
|
|
|
|
ticker.stop();
|
|
});
|
|
|
|
it("does not poison the walletMutex when consolidate throws", async () => {
|
|
consolidate.mockRejectedValue(new Error("ogmios unreachable"));
|
|
|
|
const ticker = startWalletHealthTicker(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ intervalMs: 1_000, log },
|
|
);
|
|
|
|
await vi.advanceTimersByTimeAsync(1_000);
|
|
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
|
|
|
|
expect(logLines.some((l) => l.includes("consolidate failed"))).toBe(true);
|
|
// The mutex still serves the next wallet op.
|
|
expect(await state.walletMutex.run(async () => "ok")).toBe("ok");
|
|
|
|
ticker.stop();
|
|
});
|
|
|
|
it("stop() prevents further ticks", async () => {
|
|
consolidate.mockResolvedValue(healthyResult);
|
|
|
|
const ticker = startWalletHealthTicker(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ intervalMs: 1_000, log },
|
|
);
|
|
|
|
ticker.stop();
|
|
await vi.advanceTimersByTimeAsync(5_000);
|
|
|
|
expect(consolidate).not.toHaveBeenCalled();
|
|
});
|
|
});
|
|
|
|
describe("walletHealth — triggerOnFailureConsolidate", () => {
|
|
let state: SupplierState;
|
|
let logLines: string[];
|
|
let log: (line: string) => void;
|
|
let consolidate: ReturnType<typeof vi.fn> & ConsolidateFn;
|
|
|
|
beforeEach(() => {
|
|
state = new SupplierState();
|
|
_resetWalletHealthForTests(state);
|
|
logLines = [];
|
|
log = (line) => logLines.push(line);
|
|
consolidate = vi.fn() as never;
|
|
});
|
|
|
|
it("fires consolidate on first call", async () => {
|
|
consolidate.mockResolvedValue(consolidatedResult);
|
|
|
|
triggerOnFailureConsolidate(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ log, debounceMs: 60_000 },
|
|
);
|
|
|
|
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
|
|
expect(state.snapshot().status).toBe("free");
|
|
});
|
|
|
|
it("debounces back-to-back invocations", async () => {
|
|
consolidate.mockResolvedValue(healthyResult);
|
|
|
|
triggerOnFailureConsolidate(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ log, debounceMs: 60_000 },
|
|
);
|
|
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
|
|
|
|
triggerOnFailureConsolidate(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ log, debounceMs: 60_000 },
|
|
);
|
|
// No second call — second invocation was inside the debounce window.
|
|
expect(consolidate).toHaveBeenCalledTimes(1);
|
|
});
|
|
|
|
it("runs even when session slots are held (queues on the wallet mutex)", async () => {
|
|
state.tryAcquire(ESCROW_REF);
|
|
consolidate.mockResolvedValue(healthyResult);
|
|
|
|
expect(() =>
|
|
triggerOnFailureConsolidate(
|
|
{ chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
|
|
{ log, debounceMs: 60_000 },
|
|
),
|
|
).not.toThrow();
|
|
|
|
await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
|
|
// The held session slot is untouched.
|
|
expect(state.snapshot().status).toBe("working");
|
|
});
|
|
});
|