fix(miner): retain unmarked send cache

allways/miner/fulfillment.py

@@ -62,6 +62,7 @@ def __init__(
         self.sent: Dict[int, SentSwap] = {}
         self.mark_fulfilled_attempts: Dict[int, int] = {}
         self.cushion_warned: Set[int] = set()
+        self.unmarked_stale_warned: Set[int] = set()
         self.sent_cache_path = sent_cache_path
         self.load_sent_cache()
 
@@ -96,21 +97,32 @@ def save_sent_cache(self):
             bt.logging.error(f'CRITICAL: Failed to persist sent cache: {e}')
 
     def cleanup_stale_sends(self, active_swap_ids: Set[int]):
-        """Remove cached send results for swaps no longer active."""
+        """Remove completed cached send results for swaps no longer active.
+
+        Unmarked send records are retained even when the poller no longer sees
+        the swap. A temporary contract/indexer gap can make an active swap look
+        gone; dropping its cached destination tx would allow a rediscovered
+        swap to send funds a second time.
+        """
         stale = [sid for sid in self.sent if sid not in active_swap_ids]
-        unmarked = [sid for sid in stale if not self.sent[sid].marked_fulfilled]
-        for sid in stale:
+        removable = [sid for sid in stale if self.sent[sid].marked_fulfilled]
+        retained_unmarked = [sid for sid in stale if not self.sent[sid].marked_fulfilled]
+        for sid in removable:
             self.sent.pop(sid)
             self.mark_fulfilled_attempts.pop(sid, None)
+            self.unmarked_stale_warned.discard(sid)
         self.cushion_warned -= self.cushion_warned - active_swap_ids
-        if stale:
-            bt.logging.info(f'Cleaned up stale send cache for {len(stale)} swap(s): {stale}')
-            if unmarked:
-                bt.logging.warning(
-                    f'Stale send(s) without confirmed mark_fulfilled — funds may have been sent without '
-                    f'on-chain credit: {unmarked}'
-                )
+        self.unmarked_stale_warned -= active_swap_ids
+        if removable:
+            bt.logging.info(f'Cleaned up stale send cache for {len(removable)} marked swap(s): {removable}')
             self.save_sent_cache()
+        newly_retained = [sid for sid in retained_unmarked if sid not in self.unmarked_stale_warned]
+        if newly_retained:
+            bt.logging.warning(
+                f'Retaining stale send(s) without confirmed mark_fulfilled to avoid duplicate destination sends: '
+                f'{newly_retained}'
+            )
+            self.unmarked_stale_warned.update(newly_retained)
 
     def verify_swap_safety(self, swap: Swap) -> Optional[Tuple[int, str]]:
         """Verify the swap is safe to fulfill.
@@ -226,9 +238,10 @@ def process_swap(self, swap: Swap) -> bool:
 
         Idempotent across forward steps — the ``sent`` cache tracks both the
         dest-tx outcome and whether ``mark_fulfilled`` has landed, so retry
-        polls never double-send and never double-call the contract. Cache
+        polls never double-send and never double-call the contract. Marked
         entries live until ``cleanup_stale_sends`` drops them once the swap
-        leaves the active set.
+        leaves the active set; unmarked entries are retained to keep a
+        rediscovered swap from sending destination funds again.
 
         Three possible starting states when this runs:
           - no prior record → send dest funds, then mark fulfilled

tests/test_fulfillment.py

@@ -1,7 +1,8 @@
-"""SwapFulfiller — timeout cushion, sender verification, send-path behavior.
+"""SwapFulfiller - timeout cushion, send-cache, and send-path behavior.
 
 These tests stay at the verify_swap_safety layer, which is the only part
-of SwapFulfiller that's exercised on every forward step.
+of SwapFulfiller that's exercised on every forward step. The send-cache
+regressions cover the idempotency invariant for already-sent destination funds.
 """
 
 from unittest.mock import MagicMock
@@ -10,7 +11,8 @@
 
 from allways.classes import Swap, SwapStatus
 from allways.constants import MINER_TIMEOUT_CUSHION_BLOCKS
-from allways.miner.fulfillment import SwapFulfiller
+from allways.miner.fulfillment import SentSwap, SwapFulfiller
+from allways.miner.swap_poller import MAX_REFRESH_MISSES, SwapPoller
 
 
 def make_fulfiller() -> SwapFulfiller:
@@ -96,5 +98,64 @@ def test_return_is_post_fee_not_pre_fee(self):
         assert user_receives_amount == 3_415_500_000
 
 
+class TestSentCacheCleanup:
+    def test_unmarked_stale_sends_are_retained(self):
+        fulfiller = make_fulfiller()
+        fulfiller.sent = {
+            1: SentSwap('unmarked-stale-tx', 101, marked_fulfilled=False),
+            2: SentSwap('marked-stale-tx', 102, marked_fulfilled=True),
+            3: SentSwap('active-unmarked-tx', 103, marked_fulfilled=False),
+        }
+        fulfiller.mark_fulfilled_attempts = {1: 2, 2: 3, 3: 1}
+
+        fulfiller.cleanup_stale_sends(active_swap_ids={3})
+
+        assert fulfiller.sent == {
+            1: SentSwap('unmarked-stale-tx', 101, marked_fulfilled=False),
+            3: SentSwap('active-unmarked-tx', 103, marked_fulfilled=False),
+        }
+        assert fulfiller.mark_fulfilled_attempts == {1: 2, 3: 1}
+
+    def test_retained_send_cache_blocks_resend_after_poller_misses_and_rediscovery(self):
+        swap = make_swap()
+        poll_client = MagicMock()
+        poll_client.get_next_swap_id.return_value = swap.id + 1
+        poll_client.get_swap.return_value = None
+        poller = SwapPoller(contract_client=poll_client, miner_hotkey=swap.miner_hotkey)
+        poller.active[swap.id] = swap
+        poller.last_scanned_id = swap.id
+
+        fulfiller = make_fulfiller()
+        fulfiller.sent[swap.id] = SentSwap('already-sent-dest-tx', 777, marked_fulfilled=False)
+
+        for _ in range(MAX_REFRESH_MISSES):
+            poller.poll()
+
+        assert poller.active == {}
+
+        fulfiller.cleanup_stale_sends(active_swap_ids=set(poller.active))
+        assert fulfiller.sent[swap.id] == SentSwap('already-sent-dest-tx', 777, marked_fulfilled=False)
+
+        poll_client.get_swap.return_value = swap
+        poller.poll()
+        assert poller.active == {swap.id: swap}
+
+        fulfiller.verify_swap_safety = MagicMock(return_value=(3_415_500_000, swap.miner_from_address))
+        fulfiller.verify_user_sent_funds = MagicMock(return_value=True)
+        fulfiller.send_dest_funds = MagicMock(return_value=('second-dest-tx', 888))
+
+        assert fulfiller.process_swap(swap) is True
+
+        fulfiller.send_dest_funds.assert_not_called()
+        fulfiller.client.mark_fulfilled.assert_called_once_with(
+            wallet=fulfiller.wallet,
+            swap_id=swap.id,
+            to_tx_hash='already-sent-dest-tx',
+            to_amount=3_415_500_000,
+            to_tx_block=777,
+        )
+        assert fulfiller.sent[swap.id].marked_fulfilled is True
+
+
 if __name__ == '__main__':
     pytest.main([__file__, '-v'])