diff --git a/src/trader/intent-engine.test.ts b/src/trader/intent-engine.test.ts index 273851b..710adac 100644 --- a/src/trader/intent-engine.test.ts +++ b/src/trader/intent-engine.test.ts @@ -740,8 +740,19 @@ describe('IntentEngine', () => { // This is the regression test for the basic-roundtrip flake. // Pre-fix observation: 80 proposals fired against ONE peer in 11.5 // min because firstYieldAt never reset. Post-fix: at most 1 per - // yield session (the blacklist also kicks in to filter the same + // yield session (the tarpit also kicks in to filter the same // peer out of subsequent scan results). + // + // Note (issue #30 update): the tarpit now uses a 5-min TTL instead + // of a permanent LRU-bounded blacklist, but this test still sees + // ≤ 1 proposal because: + // - The first fall-through transitions the intent to MATCHING. + // - The default mock onMatchFound resolves (no rejection), so + // `succeeded > 0` and the intent stays in MATCHING. + // - Subsequent scans skip MATCHING intents entirely (only ACTIVE/ + // PARTIALLY_FILLED are matchable). So TTL expiry has no effect + // here — the intent is never re-evaluated regardless of tarpit + // state. const { engine, market, onMatchFound } = createTestEngine({ strategy: { scan_interval_ms: 1000, auto_match: true }, agentPubkey: 'z'.repeat(64), @@ -753,7 +764,7 @@ describe('IntentEngine', () => { })]); engine.start(); // Run for 10 minutes of simulated time — pre-fix this would emit - // ~70 proposals; post-fix at most 1 (blacklisted after fall-through). + // ~70 proposals; post-fix at most 1 (intent stuck in MATCHING). await vi.advanceTimersByTimeAsync(10 * 60_000); engine.stop(); expect(onMatchFound.mock.calls.length).toBeLessThanOrEqual(1); @@ -777,6 +788,331 @@ describe('IntentEngine', () => { }); }); + // ========================================================================= + // Counterparty tarpit (issue #30) + // + // After a proposal timeout (np.propose_deal sent, np.accept_deal never + // arrived) the matcher must avoid re-picking the same dead counterparty on + // the next scan tick. The tarpit is: + // - per-intent (matches existing failedCounterparties shape) + // - threshold-based (K=2 consecutive transient failures before tripping) + // - TTL-bounded (~5 min, so recovered peers rejoin the candidate pool) + // + // The strong-signal path (markCounterpartyFailed) tarpits immediately on + // the first call; the soft-threshold path (recordCounterpartyFailure) + // increments a counter and only tarpits at the threshold. + // ========================================================================= + + describe('Counterparty tarpit (issue #30)', () => { + const TARPIT_TTL_MS = 5 * 60 * 1000; + const TARPIT_COUNTER_DECAY_MS = 10 * 60 * 1000; + + // 'a' lex-sorts before 'd' — we are the proposer in every test below. + const OWN_PUBKEY = 'a'.repeat(64); + const DEAD_COUNTERPARTY = 'd'.repeat(64); + const LIVE_COUNTERPARTY_1 = 'e'.repeat(64); + const LIVE_COUNTERPARTY_2 = 'f'.repeat(64); + + async function createIntentReturningId(engine: IntentEngine): Promise { + const record = await engine.createIntent(defaultParams(), OWN_PUBKEY, AGENT_ADDRESS); + return record.intent.intent_id; + } + + it('does NOT tarpit after a single transient failure (threshold = 2)', async () => { + const { engine, market, onMatchFound } = createTestEngine({ + strategy: { scan_interval_ms: 1000, auto_match: true }, + agentPubkey: OWN_PUBKEY, + }); + const intentId = await createIntentReturningId(engine); + + // One isolated proposal-timeout shouldn't burn the counterparty — + // a lost Nostr DM should be tolerable without 5 minutes of exclusion. + engine.recordCounterpartyFailure(intentId, DEAD_COUNTERPARTY); + + market.setSearchResults([ + buildSearchResult({ direction: 'sell', agentPublicKey: DEAD_COUNTERPARTY }), + ]); + + engine.start(); + await vi.advanceTimersByTimeAsync(1100); + engine.stop(); + + // The dead counterparty is still selectable — count was incremented to 1, + // below the K=2 threshold, so no tarpit applied. + expect(onMatchFound).toHaveBeenCalledTimes(1); + const [, counterparty] = onMatchFound.mock.calls[0]!; + expect(counterparty.agentPublicKey).toBe(DEAD_COUNTERPARTY); + }); + + it('DOES tarpit after K consecutive transient failures (threshold met)', async () => { + const { engine, market, onMatchFound } = createTestEngine({ + strategy: { scan_interval_ms: 1000, auto_match: true }, + agentPubkey: OWN_PUBKEY, + }); + const intentId = await createIntentReturningId(engine); + + // Two consecutive proposal-timeouts on the same peer → tarpit trips. + engine.recordCounterpartyFailure(intentId, DEAD_COUNTERPARTY); + engine.recordCounterpartyFailure(intentId, DEAD_COUNTERPARTY); + + market.setSearchResults([ + buildSearchResult({ direction: 'sell', agentPublicKey: DEAD_COUNTERPARTY }), + ]); + + engine.start(); + await vi.advanceTimersByTimeAsync(1100); + engine.stop(); + + // Counterparty is filtered out — no proposal attempt this scan. + expect(onMatchFound).not.toHaveBeenCalled(); + }); + + it('lets the matcher pick OTHER live candidates when one is tarpitted', async () => { + // Soak-failure regression: the issue described alice re-picking the + // same dead counterparty every scan tick despite a live one being + // available. With the tarpit, the live candidate must take their turn. + const { engine, market, onMatchFound } = createTestEngine({ + strategy: { scan_interval_ms: 1000, auto_match: true }, + agentPubkey: OWN_PUBKEY, + }); + const intentId = await createIntentReturningId(engine); + + // Tarpit the dead one (immediate path — equivalent to 2 transient). + engine.markCounterpartyFailed(intentId, DEAD_COUNTERPARTY); + + market.setSearchResults([ + buildSearchResult({ direction: 'sell', agentPublicKey: DEAD_COUNTERPARTY }), + buildSearchResult({ + direction: 'sell', + agentPublicKey: LIVE_COUNTERPARTY_1, + id: 'sr-live-1', + }), + ]); + + engine.start(); + await vi.advanceTimersByTimeAsync(1100); + engine.stop(); + + // Only the live candidate gets a proposal. + const proposedTo = onMatchFound.mock.calls.map((c) => c[1].agentPublicKey); + expect(proposedTo).toEqual([LIVE_COUNTERPARTY_1]); + }); + + it('tarpit EXPIRES after TTL — counterparty rejoins the candidate pool', async () => { + // Recovery contract: a peer that gets a fresh address / restarts / + // unblocks should resume matching within the TTL window, not require + // operator intervention. Without this, a transient outage would + // permanently sever otherwise-compatible trading pairs. + const { engine, market, onMatchFound } = createTestEngine({ + strategy: { scan_interval_ms: 60_000, auto_match: true }, + agentPubkey: OWN_PUBKEY, + }); + const intentId = await createIntentReturningId(engine); + + engine.markCounterpartyFailed(intentId, DEAD_COUNTERPARTY); + + market.setSearchResults([ + buildSearchResult({ direction: 'sell', agentPublicKey: DEAD_COUNTERPARTY }), + ]); + + engine.start(); + // Advance well past TARPIT_TTL_MS but stay under the counter-decay + // window so the entry is still in the map (just expired tarpit). + await vi.advanceTimersByTimeAsync(TARPIT_TTL_MS + 60_000); + engine.stop(); + + // Counterparty rejoins. At least one proposal fires. + expect(onMatchFound.mock.calls.length).toBeGreaterThanOrEqual(1); + const [, counterparty] = onMatchFound.mock.calls[0]!; + expect(counterparty.agentPublicKey).toBe(DEAD_COUNTERPARTY); + }); + + it('tarpit is PER-INTENT: a tarpit on intent A does not affect intent B', async () => { + // Two intents with overlapping candidate sets each maintain independent + // tarpit state. Otherwise, a counterparty unreachable on one trading + // pair would be silently excluded from an unrelated one. + const { engine, market, onMatchFound } = createTestEngine({ + strategy: { scan_interval_ms: 1000, auto_match: true }, + agentPubkey: OWN_PUBKEY, + }); + const recordA = await engine.createIntent( + defaultParams({ base_asset: 'ALPHA', quote_asset: 'USD' }), + OWN_PUBKEY, + AGENT_ADDRESS, + ); + const recordB = await engine.createIntent( + defaultParams({ base_asset: 'BETA', quote_asset: 'USD' }), + OWN_PUBKEY, + AGENT_ADDRESS, + ); + + // Tarpit the counterparty only for intent A. + engine.markCounterpartyFailed(recordA.intent.intent_id, DEAD_COUNTERPARTY); + void recordB; + + // The market returns the same counterparty for both queries (different + // base assets, so it's two separate matches against the same pubkey). + market.setSearchResults([ + buildSearchResult({ + direction: 'sell', + agentPublicKey: DEAD_COUNTERPARTY, + base_asset: 'ALPHA', + id: 'sr-a', + }), + buildSearchResult({ + direction: 'sell', + agentPublicKey: DEAD_COUNTERPARTY, + base_asset: 'BETA', + id: 'sr-b', + }), + ]); + + engine.start(); + await vi.advanceTimersByTimeAsync(1100); + engine.stop(); + + // Only intent B should reach the counterparty — A's tarpit excludes them. + const targetedIntents = onMatchFound.mock.calls.map( + (c) => c[0].intent.base_asset, + ); + expect(targetedIntents).toEqual(['BETA']); + }); + + it('counter DECAYS after a long quiet period — single failure does not insta-tarpit', async () => { + // A peer that flapped hours ago and ran cleanly since shouldn't be + // insta-tarpitted on a single fresh miss. The decay window prevents + // accumulating failureCount from a stale era. + const { engine, market, onMatchFound } = createTestEngine({ + strategy: { scan_interval_ms: 60_000, auto_match: true }, + agentPubkey: OWN_PUBKEY, + }); + const intentId = await createIntentReturningId(engine); + + // One old failure (count=1, no tarpit). + engine.recordCounterpartyFailure(intentId, DEAD_COUNTERPARTY); + + // Quiet period exceeds the decay window. + await vi.advanceTimersByTimeAsync(TARPIT_COUNTER_DECAY_MS + 1000); + + // A single fresh failure must NOT be enough to tarpit — counter + // should have decayed and reset to 1 again. + engine.recordCounterpartyFailure(intentId, DEAD_COUNTERPARTY); + + market.setSearchResults([ + buildSearchResult({ direction: 'sell', agentPublicKey: DEAD_COUNTERPARTY }), + ]); + + engine.start(); + await vi.advanceTimersByTimeAsync(60_100); + engine.stop(); + + // Counterparty is still selectable — counter is at 1 (post-decay). + expect(onMatchFound).toHaveBeenCalledTimes(1); + }); + + it('accepts EITHER local intent_id OR market_intent_id (DealTerms compatibility)', async () => { + // DealTerms carry market IDs by necessity (peers exchange their + // market-visible IDs in np.propose_deal). Trader-main passes whichever + // ID came in on the deal record. The tarpit map must resolve both. + const { engine, market, onMatchFound } = createTestEngine({ + strategy: { scan_interval_ms: 1000, auto_match: true }, + agentPubkey: OWN_PUBKEY, + }); + const record = await engine.createIntent(defaultParams(), OWN_PUBKEY, AGENT_ADDRESS); + const marketIntentId = record.intent.market_intent_id; + expect(marketIntentId).not.toBe(''); + expect(marketIntentId).not.toBe(record.intent.intent_id); + + // Tarpit via the market ID — should resolve to the same local entry. + engine.markCounterpartyFailed(marketIntentId, DEAD_COUNTERPARTY); + + market.setSearchResults([ + buildSearchResult({ direction: 'sell', agentPublicKey: DEAD_COUNTERPARTY }), + buildSearchResult({ + direction: 'sell', + agentPublicKey: LIVE_COUNTERPARTY_2, + id: 'sr-live-2', + }), + ]); + + engine.start(); + await vi.advanceTimersByTimeAsync(1100); + engine.stop(); + + const proposedTo = onMatchFound.mock.calls.map((c) => c[1].agentPublicKey); + expect(proposedTo).toEqual([LIVE_COUNTERPARTY_2]); + }); + + it('clearCounterpartyFailures resets the counter (success-reset contract)', async () => { + // Adversarial review (issue #30 follow-up): "K consecutive" must + // really mean *consecutive* across the deal lifecycle, not "K within + // the 10-min decay window regardless of intermediate successes." + // trader-main calls this when the acceptor's np.accept_deal arrives, + // proving the counterparty is responsive. Without the reset, a peer + // that fails 1 + succeeds 1 + fails 1 + succeeds 1 + ... would + // eventually trip the tarpit on a single fresh failure because the + // counter never decays under the noise floor. + const { engine, market, onMatchFound } = createTestEngine({ + strategy: { scan_interval_ms: 1000, auto_match: true }, + agentPubkey: OWN_PUBKEY, + }); + const intentId = await createIntentReturningId(engine); + + engine.recordCounterpartyFailure(intentId, DEAD_COUNTERPARTY); // count=1 + engine.clearCounterpartyFailures(intentId, DEAD_COUNTERPARTY); // reset + engine.recordCounterpartyFailure(intentId, DEAD_COUNTERPARTY); // count=1 (fresh) + + market.setSearchResults([ + buildSearchResult({ direction: 'sell', agentPublicKey: DEAD_COUNTERPARTY }), + ]); + + engine.start(); + await vi.advanceTimersByTimeAsync(1100); + engine.stop(); + + // Counter is at 1 — below threshold — so candidate is still selectable. + expect(onMatchFound).toHaveBeenCalledTimes(1); + }); + + it('clearCounterpartyFailures on absent entry is a no-op (defensive)', async () => { + // Called from onDealAccepted, which runs after any deal handshake. + // There may not be a prior failure record (clean first interaction). + // Must not throw. + const { engine } = createTestEngine({ + strategy: { scan_interval_ms: 60_000, auto_match: false }, + agentPubkey: OWN_PUBKEY, + }); + const intentId = await createIntentReturningId(engine); + expect(() => + engine.clearCounterpartyFailures(intentId, DEAD_COUNTERPARTY), + ).not.toThrow(); + }); + + it('immediate markCounterpartyFailed tarpits on the FIRST call (strong signal)', async () => { + // The W2 yield-timeout fall-through path uses this — 45s of silence + // after election-yielding is strong enough evidence to tarpit + // without waiting for two confirmations. + const { engine, market, onMatchFound } = createTestEngine({ + strategy: { scan_interval_ms: 1000, auto_match: true }, + agentPubkey: OWN_PUBKEY, + }); + const intentId = await createIntentReturningId(engine); + + engine.markCounterpartyFailed(intentId, DEAD_COUNTERPARTY); + + market.setSearchResults([ + buildSearchResult({ direction: 'sell', agentPublicKey: DEAD_COUNTERPARTY }), + ]); + + engine.start(); + await vi.advanceTimersByTimeAsync(1100); + engine.stop(); + + // Tarpitted on first call — no proposal attempt. + expect(onMatchFound).not.toHaveBeenCalled(); + }); + }); + // ========================================================================= // 5. Feed subscription // ========================================================================= diff --git a/src/trader/intent-engine.ts b/src/trader/intent-engine.ts index ec1a0cf..3d3c8c3 100644 --- a/src/trader/intent-engine.ts +++ b/src/trader/intent-engine.ts @@ -46,8 +46,29 @@ export interface IntentEngine { getIntent(intentId: string): IntentRecord | null; /** Restore an intent to ACTIVE after a failed deal (e.g., negotiation timeout). */ restoreToActive(intentId: string): void; - /** Mark a counterparty as failed for a given intent so it won't be matched again. */ + /** + * Mark a counterparty as failed for a given intent with an *immediate* + * tarpit. Used by strong-signal paths (W2 yield-timeout fall-through) where + * one observation is enough to suspect a dead peer. The entry expires after + * `TARPIT_TTL_MS` so a counterparty returning to life resumes matching. + */ markCounterpartyFailed(intentId: string, counterpartyPubkey: string): void; + /** + * Record a *transient* counterparty failure (proposal timeout / send + * failure) for the given intent. The counterparty is only tarpitted after + * `TARPIT_FAILURE_THRESHOLD` consecutive failures, so a single dropped DM + * does not penalise an otherwise live peer. See issue #30 for the soak + * failure mode this fixes. + */ + recordCounterpartyFailure(intentId: string, counterpartyPubkey: string): void; + /** + * Clear the recorded failure history for a counterparty on this intent. + * Called when the counterparty proves responsive (e.g. on receiving their + * `np.accept_deal`) so that "K consecutive failures" really means + * consecutive — not "K within the decay window regardless of successes + * in between." + */ + clearCounterpartyFailures(intentId: string, counterpartyPubkey: string): void; /** Record a partial or full fill after a successful swap. */ recordFill(intentId: string, filledVolume: string): void; /** Look up an intent by its MarketModule ID (UUID). */ @@ -158,9 +179,51 @@ export function createIntentEngine(deps: IntentEngineDeps): IntentEngine { // In-memory intent store, keyed by intent_id const intents = new Map(); - // Per-intent set of counterparty pubkeys that failed (timed out / rejected). - // Prevents repeatedly matching against the same dead counterparty. - const failedCounterparties = new Map>(); + // Per-intent map of counterparty pubkeys that failed (timed out / rejected). + // Prevents the matcher from re-picking the same dead counterparty on every + // 30s scan tick (issue #30). + // + // Each entry tracks: + // - failureCount: consecutive transient-failure observations (proposal + // timeout / send failure). The counterparty is only excluded once + // this reaches TARPIT_FAILURE_THRESHOLD, so a single dropped DM + // does not penalise an otherwise live peer. + // - tarpitUntil: ms timestamp until which matchesCriteria() filters + // this counterparty out. Set to `nowMs() + TARPIT_TTL_MS` once the + // threshold trips, OR immediately on strong-signal paths (W2 yield + // fall-through via markCounterpartyFailed). 0 means "counter only, + // no active tarpit yet". + // - lastFailureAt: ms timestamp of last failure observation. Used to + // decay failureCount after TARPIT_COUNTER_DECAY_MS of quiet so a + // counterparty that recovers does not carry old strikes forever. + // + // Capacity is bounded per-intent by MAX_FAILED_PER_INTENT; eviction is + // LRU-by-lastFailureAt (insertion order of the Map under the existing + // delete+re-add idiom). + interface TarpitEntry { + failureCount: number; + tarpitUntil: number; + lastFailureAt: number; + } + const failedCounterparties = new Map>(); + + // Tarpit policy constants (see issue #30). + // - THRESHOLD = 2: tolerate one transient miss (lost Nostr DM, jittered + // relay) without penalising the peer; two consecutive misses are strong + // evidence they are unresponsive. + // - TTL_MS = 5 min: long enough to skip ~10 scan cycles (default + // scan_interval_ms = 30s), short enough that a recovered peer re-enters + // the candidate pool quickly. + // - COUNTER_DECAY_MS = 10 min: if no failure was observed in this window, + // the counter resets to 0 on the next observation. Prevents a peer + // that flapped 2 hours ago from being insta-tarpitted on a single + // modern miss. + // - MAX_FAILED_PER_INTENT = 1000: capacity bound. Issue #30 explicitly + // asks the new TTL semantics to live alongside the existing LRU bound. + const TARPIT_FAILURE_THRESHOLD = 2; + const TARPIT_TTL_MS = 5 * 60 * 1000; + const TARPIT_COUNTER_DECAY_MS = 10 * 60 * 1000; + const MAX_FAILED_PER_INTENT = 1000; // Per-intent timestamp of the FIRST scan-cycle yield under spec 5.7. When // every match candidate is a lower-pubkey-priority peer (so we should @@ -330,11 +393,17 @@ export function createIntentEngine(deps: IntentEngineDeps): IntentEngine { return false; } - // 7b. Not a recently-failed counterparty for this intent. + // 7b. Not a tarpitted counterparty for this intent. // Normalize to a format-independent key so the same peer can't bypass the - // failed-list by reconnecting with a different pubkey encoding. + // tarpit list by reconnecting with a different pubkey encoding. + // + // A counterparty is excluded only while `tarpitUntil > now`. Once the + // window elapses the entry is treated as absent (and is cleared on the + // next failure observation so the counter starts fresh). This matches + // the issue #30 contract: short-TTL tarpit, not a permanent blacklist. const failed = failedCounterparties.get(own.intent_id); - if (failed?.has(canonicalPubkeyKey(result.agentPublicKey))) { + const entry = failed?.get(canonicalPubkeyKey(result.agentPublicKey)); + if (entry !== undefined && entry.tarpitUntil > nowMs()) { return false; } @@ -801,6 +870,92 @@ export function createIntentEngine(deps: IntentEngineDeps): IntentEngine { } } + // --------------------------------------------------------------------------- + // Tarpit bookkeeping (issue #30) + // --------------------------------------------------------------------------- + + /** + * Shared insert/update path for `failedCounterparties`. + * + * @param immediateTarpit - true: set tarpitUntil to now+TTL regardless of + * failureCount (strong-signal path). + * - false: increment failureCount; only set + * tarpitUntil once it reaches the threshold + * (soft-threshold path). + * + * Behaviour: + * - LRU eviction by `lastFailureAt` (delete+re-set keeps Map iteration + * insertion-order; eviction drops the front). + * - Counter decay: if the previous failure was >= TARPIT_COUNTER_DECAY_MS + * ago, treat this observation as the first of a new cycle (reset to 1). + * - Resolves either local intent_id or market_intent_id (DealTerms carry + * market IDs) so callers don't have to translate. + */ + function writeTarpitEntry( + intentId: string, + counterpartyPubkey: string, + immediateTarpit: boolean, + ): void { + const record = resolveIntentByEitherId(intentId); + const localId = record?.intent.intent_id ?? intentId; + + let perIntent = failedCounterparties.get(localId); + if (!perIntent) { + perIntent = new Map(); + failedCounterparties.set(localId, perIntent); + } + + const key = canonicalPubkeyKey(counterpartyPubkey); + const now = nowMs(); + const existing = perIntent.get(key); + + // Decay rule: a counterparty that hasn't failed recently starts fresh. + // Without this, a peer that flapped, recovered, ran cleanly for an + // hour, then dropped a single DM would be insta-tarpitted because the + // old failureCount is still at THRESHOLD. + const counterShouldDecay = + existing !== undefined && + now - existing.lastFailureAt >= TARPIT_COUNTER_DECAY_MS; + + let nextCount: number; + if (existing === undefined || counterShouldDecay) { + nextCount = 1; + } else { + nextCount = existing.failureCount + 1; + } + + const nextTarpit = immediateTarpit || nextCount >= TARPIT_FAILURE_THRESHOLD + ? now + TARPIT_TTL_MS + : 0; + + const nextEntry: TarpitEntry = { + failureCount: nextCount, + tarpitUntil: nextTarpit, + lastFailureAt: now, + }; + + // LRU semantics: delete+re-add moves a refreshed peer to the most- + // recently-failed position so they survive eviction longer than peers + // that have gone quiet. + if (existing !== undefined) { + perIntent.delete(key); + } else if (perIntent.size >= MAX_FAILED_PER_INTENT) { + // Capacity eviction: drop the least-recently-failed entry. Map + // iteration order is insertion order; the front is the oldest. + const oldestKey = perIntent.keys().next().value; + if (oldestKey !== undefined) perIntent.delete(oldestKey); + } + perIntent.set(key, nextEntry); + + logger.info('counterparty_failure_recorded', { + intent_id: localId, + counterparty: counterpartyPubkey, + failure_count: nextCount, + tarpitted: nextTarpit > 0, + immediate: immediateTarpit, + }); + } + // --------------------------------------------------------------------------- // Public API // --------------------------------------------------------------------------- @@ -1042,35 +1197,34 @@ export function createIntentEngine(deps: IntentEngineDeps): IntentEngine { }, markCounterpartyFailed(intentId: string, counterpartyPubkey: string): void { - // Resolve to the LOCAL intent_id so the failed-counterparty Set is - // keyed consistently regardless of whether the caller passed a local - // or a market intent_id (DealTerms carry market IDs). + // Strong-signal path: immediately tarpit. Used by the W2 yield-timeout + // fall-through where 45s of silence is sufficient evidence the + // counterparty is dead. See `recordCounterpartyFailure` for the + // soft-threshold path used by single-deal proposal failures. + writeTarpitEntry(intentId, counterpartyPubkey, /* immediateTarpit */ true); + }, + + recordCounterpartyFailure(intentId: string, counterpartyPubkey: string): void { + // Soft-threshold path: increment the failure counter; only tarpit + // once we cross TARPIT_FAILURE_THRESHOLD. One dropped DM is not + // enough to penalise an otherwise live peer (issue #30). + writeTarpitEntry(intentId, counterpartyPubkey, /* immediateTarpit */ false); + }, + + clearCounterpartyFailures(intentId: string, counterpartyPubkey: string): void { + // Resolve to the LOCAL intent_id so the lookup matches what the + // failure-recording path stored (it also resolves to local). const record = resolveIntentByEitherId(intentId); const localId = record?.intent.intent_id ?? intentId; - - const MAX_FAILED_PER_INTENT = 1000; - let set = failedCounterparties.get(localId); - if (!set) { - set = new Set(); - failedCounterparties.set(localId, set); - } - // Use canonical form so a peer can't re-engage by presenting a different - // pubkey encoding of the same identity. + const perIntent = failedCounterparties.get(localId); + if (!perIntent) return; const key = canonicalPubkeyKey(counterpartyPubkey); - // LRU semantics: if this peer is already on the list, delete+re-add - // refreshes their position (most-recently-failed). This ensures actively - // failing peers stay blocked even as new failures push the oldest out. - if (set.has(key)) { - set.delete(key); - } else if (set.size >= MAX_FAILED_PER_INTENT) { - // Capacity eviction: drop the LEAST-recently-failed (first-inserted under - // LRU ordering) to make room. Set iteration order is insertion order, and - // the delete+re-add pattern above moves refreshed peers to the end. - const oldest = set.values().next().value; - if (oldest !== undefined) set.delete(oldest); - } - set.add(key); - logger.info('counterparty_marked_failed', { intent_id: localId, counterparty: counterpartyPubkey }); + if (!perIntent.has(key)) return; + perIntent.delete(key); + logger.info('counterparty_failures_cleared', { + intent_id: localId, + counterparty: counterpartyPubkey, + }); }, recordFill(intentId: string, filledVolume: string): void { diff --git a/src/trader/negotiation-handler.test.ts b/src/trader/negotiation-handler.test.ts index bc969e2..cbf60c7 100644 --- a/src/trader/negotiation-handler.test.ts +++ b/src/trader/negotiation-handler.test.ts @@ -3,6 +3,7 @@ import { createHash } from 'node:crypto'; import { createNegotiationHandler } from './negotiation-handler.js'; import type { NegotiationHandler, NegotiationHandlerDeps } from './negotiation-handler.js'; import type { + CancellationReason, DealRecord, DealTerms, IntentRecord, @@ -45,7 +46,7 @@ function createDeps(overrides: Partial = {}): Negotiatio signMessage: vi.fn<(msg: string) => string>().mockReturnValue(FIXED_SIGNATURE), verifySignature: vi.fn<(sig: string, msg: string, pubkey: string) => boolean>().mockReturnValue(true), onDealAccepted: vi.fn<(deal: DealRecord) => Promise>().mockResolvedValue(undefined), - onDealCancelled: vi.fn<(deal: DealRecord) => void>(), + onDealCancelled: vi.fn<(deal: DealRecord, reason: CancellationReason) => void>(), agentPubkey: AGENT_PUBKEY, agentAddress: AGENT_ADDRESS, logger: createMockLogger(), @@ -684,6 +685,74 @@ describe('NegotiationHandler', () => { } }); + it('forwards reason="proposal_timeout" to onDealCancelled (issue #30)', async () => { + // Issue #30: trader-main needs the cancellation reason to decide + // whether to tarpit the counterparty. Proposal-timeout means the + // counterparty went silent — exactly the signal that justifies + // tarpit. + vi.useFakeTimers(); + try { + const localDeps = createDeps(); + const localHandler = createNegotiationHandler(localDeps); + const deal = await localHandler.proposeDeal( + makeOwnIntent(), makeCounterparty(), '150', '50', 'escrow-1', + ); + + // Async variant flushes microtasks between timer iterations so the + // fire-and-forget transitionDeal()→persistDeal()→onDealCancelled + // chain resolves before we assert. + await vi.advanceTimersByTimeAsync(30_000); + await localHandler.drainPersistChains(); + + expect(localDeps.onDealCancelled).toHaveBeenCalledTimes(1); + const [cancelledDeal, reason] = (localDeps.onDealCancelled as ReturnType).mock.calls[0]!; + expect(cancelledDeal.terms.deal_id).toBe(deal.terms.deal_id); + expect(reason).toBe('proposal_timeout'); + + localHandler.stop(); + } finally { + vi.useRealTimers(); + } + }); + + it('forwards reason="counterparty_rejected" on incoming np.reject_deal', async () => { + // Rejection from the peer is recoverable (AGENT_BUSY race, etc.) — + // trader-main uses this distinct reason to skip the tarpit branch + // even though the deal is also CANCELLED. + const deal = await handler.proposeDeal( + makeOwnIntent(), makeCounterparty(), '150', '50', 'escrow-1', + ); + + const rejectMsg = buildNpMessage( + deal.terms.deal_id, + 'np.reject_deal', + COUNTERPARTY_PUBKEY, + { reason_code: 'AGENT_BUSY', message: '' }, + ); + await handler.handleIncomingDm(COUNTERPARTY_PUBKEY, COUNTERPARTY_ADDRESS, JSON.stringify(rejectMsg)); + + expect(deps.onDealCancelled).toHaveBeenCalledTimes(1); + const [, reason] = (deps.onDealCancelled as ReturnType).mock.calls[0]!; + expect(reason).toBe('counterparty_rejected'); + }); + + it('forwards reason="shutdown" on cancelPending()', async () => { + // Shutdown sweeps must NOT propagate into the tarpit — these are + // local-state cancellations, not counterparty signals. + await handler.proposeDeal( + makeOwnIntent(), makeCounterparty(), '150', '50', 'escrow-1', + ); + + handler.cancelPending(); + // Drain pending state-change persistence triggered by cancelPending(). + await handler.drainPersistChains(); + + expect(deps.onDealCancelled).toHaveBeenCalled(); + const calls = (deps.onDealCancelled as ReturnType).mock.calls; + const reasons = calls.map((c: unknown[]) => c[1]); + expect(reasons.every((r) => r === 'shutdown')).toBe(true); + }); + it('ACCEPTED deal stays ACCEPTED (SwapExecutor owns execution timeouts)', async () => { vi.useFakeTimers(); try { diff --git a/src/trader/negotiation-handler.ts b/src/trader/negotiation-handler.ts index dedaae0..827c842 100644 --- a/src/trader/negotiation-handler.ts +++ b/src/trader/negotiation-handler.ts @@ -15,6 +15,7 @@ import { MAX_MESSAGE_SIZE } from '../protocols/envelope.js'; import { hasDangerousKeys, canonicalJson } from './utils.js'; import { validateDealTerms } from './utils.js'; import type { + CancellationReason, DealTerms, DealRecord, DealState, @@ -593,7 +594,7 @@ export function createNegotiationHandler(deps: NegotiationHandlerDeps): Negotiat async function transitionDeal( dealId: string, newState: DealState, - options?: { errorCode?: string }, + options?: { errorCode?: string; reason?: CancellationReason }, ): Promise { const deal = deals.get(dealId); if (!deal) return null; @@ -639,8 +640,13 @@ export function createNegotiationHandler(deps: NegotiationHandlerDeps): Negotiat (d.state === 'ACCEPTED' || d.state === 'EXECUTING'), ); if (!hasSiblingAccepted) { + // Forward the cancellation reason. Callers without a specific signal + // (legacy paths, defensive fall-throughs) get 'unknown', which the + // trader-level handler treats as "restore intent, do not tarpit" — + // matching the historical behaviour from before reasons were threaded. + const reason: CancellationReason = options?.reason ?? 'unknown'; try { - onDealCancelled(updated); + onDealCancelled(updated, reason); } catch (err: unknown) { const message = err instanceof Error ? err.message : String(err); logger.error('on_deal_cancelled_callback_failed', { deal_id: dealId, error: message }); @@ -672,7 +678,13 @@ export function createNegotiationHandler(deps: NegotiationHandlerDeps): Negotiat // is now async. Errors are handled inside persistDeal; any uncaught // rejection here is logged so the process doesn't die from an // unhandled rejection during background deal expiry. - transitionDeal(dealId, 'CANCELLED').catch((err: unknown) => { + // + // Reason: 'proposal_timeout' — startTimer is only invoked from + // proposeDeal() (see PROPOSE_TIMEOUT_MS callsite), so the only state + // a still-live timer can fire against is PROPOSED. The trader-level + // onDealCancelled handler uses this signal to tarpit the unresponsive + // counterparty (issue #30). + transitionDeal(dealId, 'CANCELLED', { reason: 'proposal_timeout' }).catch((err: unknown) => { const message = err instanceof Error ? err.message : String(err); logger.warn('deal_timeout_transition_failed', { deal_id: dealId, error: message }); }); @@ -1231,7 +1243,7 @@ export function createNegotiationHandler(deps: NegotiationHandlerDeps): Negotiat // only reachable from ACCEPTED/EXECUTING per the state machine. // The auto-accept gate treats anything not-in-ACCEPTED-or-EXECUTING // as "blocked", so CANCELLED is sufficient protection. - await transitionDeal(msg.deal_id, 'CANCELLED'); + await transitionDeal(msg.deal_id, 'CANCELLED', { reason: 'accept_send_failed' }); return; } @@ -1341,7 +1353,7 @@ export function createNegotiationHandler(deps: NegotiationHandlerDeps): Negotiat deal_id: msg.deal_id, winning_deal_id: otherId, }); - await transitionDeal(msg.deal_id, 'CANCELLED'); + await transitionDeal(msg.deal_id, 'CANCELLED', { reason: 'sibling_cancelled' }); return; } } @@ -1392,7 +1404,7 @@ export function createNegotiationHandler(deps: NegotiationHandlerDeps): Negotiat otherDeal.terms.proposer_intent_id === intentId ) { clearTimer(otherId); - await transitionDeal(otherId, 'CANCELLED'); + await transitionDeal(otherId, 'CANCELLED', { reason: 'sibling_cancelled' }); logger.info('sibling_proposal_cancelled', { cancelled_deal_id: otherId, winning_deal_id: msg.deal_id, @@ -1486,7 +1498,7 @@ export function createNegotiationHandler(deps: NegotiationHandlerDeps): Negotiat } clearTimer(msg.deal_id); - await transitionDeal(msg.deal_id, 'CANCELLED'); + await transitionDeal(msg.deal_id, 'CANCELLED', { reason: 'counterparty_rejected' }); const payload = msg.payload as Record; logger.info('np_deal_rejected', { @@ -1581,7 +1593,7 @@ export function createNegotiationHandler(deps: NegotiationHandlerDeps): Negotiat error: message, }); clearTimer(dealId); - await transitionDeal(dealId, 'CANCELLED'); + await transitionDeal(dealId, 'CANCELLED', { reason: 'proposal_send_failed' }); throw err; } finally { // Always clear the send timeout to prevent timer leak (steelman #3) @@ -1750,7 +1762,7 @@ export function createNegotiationHandler(deps: NegotiationHandlerDeps): Negotiat for (const [dealId, deal] of deals) { if (!(TERMINAL_DEAL_STATES as readonly string[]).includes(deal.state)) { clearTimer(dealId); - transitionDeal(dealId, 'CANCELLED').catch((err: unknown) => { + transitionDeal(dealId, 'CANCELLED', { reason: 'shutdown' }).catch((err: unknown) => { const message = err instanceof Error ? err.message : String(err); logger.warn('cancel_pending_transition_failed', { deal_id: dealId, error: message }); }); diff --git a/src/trader/trader-main.ts b/src/trader/trader-main.ts index c75e678..be184b9 100644 --- a/src/trader/trader-main.ts +++ b/src/trader/trader-main.ts @@ -469,6 +469,23 @@ export function createTraderAgent(deps: TraderMainDeps): TraderAgent { const onDealAccepted: OnDealAccepted = async (deal: DealRecord) => { if (!swapExecutor || !ledger || !stateStore) return; + // Issue #30 success-reset: if we are the proposer and the acceptor's + // np.accept_deal just arrived, they have proved responsive at the + // NP-0 layer — so the tarpit failure counter for this counterparty + // on our intent must be cleared. Without this reset, intermittent + // network jitter could accumulate stale strikes across long-running + // intents and produce spurious 5-min exclusions even after a + // successful handshake. Acceptor side has nothing to clear here + // (it never records failures against the proposer in this callback + // path). + const proposerHere = pubkeysEqual(deal.terms.proposer_pubkey, agentPubkey); + if (proposerHere && intentEngine) { + const ourId = ourIntentId(deal); + if (ourId !== null) { + intentEngine.clearCounterpartyFailures(ourId, deal.terms.acceptor_pubkey); + } + } + // Warning fix — volume reservation failure must ABORT the deal rather // than proceed. Previously a failed reserve logged a warning and fell // through to proposeSwap, potentially over-committing volume across @@ -717,23 +734,33 @@ export function createTraderAgent(deps: TraderMainDeps): TraderAgent { return null; }, getTrustedEscrows: () => strategy.trusted_escrows, - onDealCancelled: (deal) => { + onDealCancelled: (deal, reason) => { // Restore the intent to ACTIVE so the next scan cycle can re-match. // - // We deliberately DO NOT call markCounterpartyFailed() on a plain - // CANCELLED deal. Cancellation reasons include: - // - proposal timeout (counterparty unreachable — could be transient) - // - AGENT_BUSY rejection (proposer-election race, fully recoverable) - // - sibling-cancellation when another deal won proposer-election - // None of these justify permanently blacklisting the counterparty — - // and doing so caused a real deadlock when both sides happened to - // race fan-outs (carol's outgoing proposal got AGENT_BUSY-rejected - // because dave's incoming arrived first; carol then permanently - // blacklisted dave; carol's intent could never re-match dave). The - // 30s NP-0 PROPOSED-state timeout + spec 5.7 proposer-election are - // sufficient to break the race within one scan cycle. Permanent - // blacklisting belongs to higher-level signals (e.g. repeated - // verification failures) that aren't surfaced through this callback. + // We deliberately DO NOT permanently blacklist a counterparty on + // CANCELLED. The original reasoning (preserved from the no-blacklist + // era): some cancellations are race-only and fully recoverable — + // AGENT_BUSY rejection from a proposer-election race, sibling- + // cancellation when another deal won proposer-election — and + // permanently blacklisting on those caused real deadlocks (carol's + // outgoing proposal got AGENT_BUSY-rejected because dave's incoming + // arrived first; carol then permanently blacklisted dave; carol's + // intent could never re-match dave). Permanent blacklisting belongs + // to higher-level signals (e.g. repeated verification failures) that + // aren't surfaced through this callback. + // + // But for the proposal-timeout and propose-send-failed cases, a + // *short-TTL tarpit* is justified: the counterparty silently failed + // to respond to our NP-0 proposal. Without the tarpit, the next + // scan tick will deterministically re-pick them (sort order is + // stable, the dead candidate is still in the market feed), waste + // another 30s timeout cycle on the same corpse, and starve the + // other live candidates. Issue #30 documents this exact failure + // mode from a trader-roundtrip soak. The tarpit: + // - is per-intent (matches the existing failedCounterparties shape) + // - requires K=2 consecutive transient failures before tripping + // (so one lost Nostr DM doesn't penalise the peer) + // - expires after ~5 min so a recovered peer rejoins the pool. if (!intentEngine) return; const weAreProposer = pubkeysEqual(deal.terms.proposer_pubkey, agentPubkey); const weAreAcceptor = pubkeysEqual(deal.terms.acceptor_pubkey, agentPubkey); @@ -749,6 +776,17 @@ export function createTraderAgent(deps: TraderMainDeps): TraderAgent { const ourIntentId = weAreProposer ? deal.terms.proposer_intent_id : deal.terms.acceptor_intent_id; + + // Only the *proposer* side tarpits, and only on cancellation + // reasons that indicate counterparty unresponsiveness. The + // acceptor side has no recovery to do here — they were on the + // receiving end of a DM that didn't lead anywhere; the next + // scan tick will re-evaluate their own match candidate set + // afresh. + if (weAreProposer && (reason === 'proposal_timeout' || reason === 'proposal_send_failed')) { + const counterpartyPubkey = deal.terms.acceptor_pubkey; + intentEngine.recordCounterpartyFailure(ourIntentId, counterpartyPubkey); + } intentEngine.restoreToActive(ourIntentId); }, agentPubkey, diff --git a/src/trader/types.ts b/src/trader/types.ts index 4eaa9a4..634ddaa 100644 --- a/src/trader/types.ts +++ b/src/trader/types.ts @@ -377,7 +377,30 @@ export type OnMatchFound = ( export type OnDealAccepted = (deal: DealRecord) => Promise; -export type OnDealCancelled = (deal: DealRecord) => void; +/** + * Why a deal moved to CANCELLED. Threaded through `transitionDeal` and into + * the `OnDealCancelled` callback so the trader-level reaction can be reason- + * specific (e.g. tarpit a counterparty that proposal-timed-out vs. let an + * AGENT_BUSY proposer-election race recover on the next scan). + * + * The reason is transient — it is NOT persisted onto DealRecord. CANCELLED + * records on disk carry the counterparty-signed `np.reject_deal` envelope + * (when one was received) as the canonical source of truth; this transient + * reason exists purely to drive the local recovery policy. + */ +export type CancellationReason = + | 'proposal_timeout' // PROPOSED-state timer fired without np.accept_deal arriving + | 'proposal_send_failed' // sendDm failed/timed out — np.propose_deal never reached counterparty + | 'accept_send_failed' // our np.accept_deal sendDm failed — we cannot proceed as acceptor + | 'counterparty_rejected' // np.reject_deal received from counterparty (reason inside payload) + | 'sibling_cancelled' // a sibling deal on the same intent already won proposer-election + | 'shutdown' // engine stopping — cancelPending sweep + | 'unknown'; // legacy / unspecified path + +export type OnDealCancelled = ( + deal: DealRecord, + reason: CancellationReason, +) => void; export type OnSwapCompleted = ( deal: DealRecord,