From f4e26525c3a844d5c4819ff29d95b753230dab25 Mon Sep 17 00:00:00 2001 From: Vladimir Rogojin Date: Sun, 14 Jun 2026 16:29:52 +0200 Subject: [PATCH] fix(trader)(#30): short-TTL tarpit on proposal-timeout counterparty MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously, when a counterparty went silent during NP-0 negotiation (np.propose_deal sent, np.accept_deal never arrived), the matcher would re-pick the same dead candidate on the next scan tick, burn another 30s on the proposal timeout, and starve the live candidates in the feed. Issue #30 documents the soak failure mode where a single dead acceptor monopolised alice's entire 15-minute match cycle. Fix: thread a CancellationReason through transitionDeal -> onDealCancelled so trader-main can react reason-specifically. On proposal_timeout and proposal_send_failed, call IntentEngine.recordCounterpartyFailure() to increment a per-intent failure counter; after K=2 consecutive failures the counterparty is tarpitted for 5 minutes (long enough to skip ~10 scan cycles, short enough that a recovered peer rejoins quickly). AGENT_BUSY rejection and sibling-cancellation keep the historical no-blacklist semantics — they are race-only and fully recoverable. The existing markCounterpartyFailed() (used by the W2 yield-timeout fall-through) becomes the strong-signal immediate-tarpit primitive, sharing the same failedCounterparties Map. A new clearCounterpartyFailures() resets the counter when the counterparty proves responsive (called from onDealAccepted on the proposer side) so "K consecutive" really means consecutive, not "K within the decay window regardless of intermediate successes." Tests cover: threshold gating, immediate tarpit, TTL expiry, per-intent isolation, counter decay, ID resolution (local vs market intent_id), success-reset contract, and forwarding of each cancellation reason through onDealCancelled. --- src/trader/intent-engine.test.ts | 340 ++++++++++++++++++++++++- src/trader/intent-engine.ts | 220 +++++++++++++--- src/trader/negotiation-handler.test.ts | 71 +++++- src/trader/negotiation-handler.ts | 30 ++- src/trader/trader-main.ts | 68 +++-- src/trader/types.ts | 25 +- 6 files changed, 693 insertions(+), 61 deletions(-) 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,