feat: sidecar long pending deposits management

emily_cron/app/models/deposit.py

@@ -1,6 +1,5 @@
 from dataclasses import asdict, dataclass, field
 from enum import Enum
-import functools
 from typing import Any, Optional, Self
 
 from bitcoinlib.scripts import Script
@@ -63,10 +62,9 @@ def max_fee(self) -> int:
         max_fee_bytes = bytes.fromhex(script.view(as_list=True)[0])
         return int.from_bytes(max_fee_bytes[:8], byteorder="big", signed=False)
 
-    @functools.cached_property
-    def deposit_time(self) -> int:
+    def deposit_last_update(self) -> int:
         """Get the timestamp from the last update block hash."""
-        from ..clients import HiroAPI  # Moved import here to avoid circular import
+        from ..clients import HiroAPI  # avoid circular import
 
         return HiroAPI.get_stacks_block(self.last_update_block_hash).time
 

emily_cron/app/services/deposit_processor.py

@@ -1,7 +1,11 @@
 import logging
+from datetime import datetime, UTC
 from itertools import chain, groupby
+from json import JSONDecodeError
 from typing import Iterable
 
+from requests.exceptions import RequestException, JSONDecodeError
+
 from ..clients import PrivateEmilyAPI, MempoolAPI
 from ..models import (
     DepositUpdate,
@@ -69,7 +73,7 @@ def process_rbf_transactions(
                         bitcoin_chaintip_height
                         < tx.confirmed_height + settings.MIN_BLOCK_CONFIRMATIONS
                     ):
-                        logger.warning(
+                        logger.info(
                             f"Confirmed transaction {tx.bitcoin_txid} is not yet eligible for RBF replacement"
                         )
                     else:
@@ -203,6 +207,55 @@ def process_expired_locktime(
 
         return updates
 
+    def process_long_pending(
+        self,
+        enriched_deposits: list[EnrichedDepositInfo],
+    ) -> list[DepositUpdate]:
+        """Process long-pending transactions.
+
+        Args:
+            enriched_deposits: List of enriched deposit information
+
+        Returns:
+            list[DepositUpdate]: List of deposit updates
+        """
+        updates = []
+
+        current_time = int(datetime.now(UTC).timestamp())
+
+        long_pending_txs = []
+        for tx in enriched_deposits:
+            # Only check pending transactions
+            if tx.status != RequestStatus.PENDING.value:
+                continue
+            # that we can't find via the mempool API (it might have been dropped)
+            if tx.in_mempool:
+                continue
+
+            try:
+                # and have been pending for too long
+                if current_time - tx.deposit_last_update() > settings.MAX_UNCONFIRMED_TIME:
+                    long_pending_txs.append(tx)
+            except (RequestException, ValueError, JSONDecodeError) as e:
+                logger.warning(
+                    f"Could not check pending status for deposit {tx.bitcoin_txid} due to "
+                    f"API error fetching block {tx.last_update_block_hash}: {e}. Skipping."
+                )
+
+        for tx in long_pending_txs:
+            logger.info(f"Marking long-pending transaction {tx.bitcoin_txid} as FAILED")
+            updates.append(
+                DepositUpdate(
+                    bitcoin_txid=tx.bitcoin_txid,
+                    bitcoin_tx_output_index=tx.bitcoin_tx_output_index,
+                    status=RequestStatus.FAILED.value,
+                    status_message=f"Pending for too long ({settings.MAX_UNCONFIRMED_TIME} seconds)",
+                )
+            )
+
+        logger.info(f"Found {len(long_pending_txs)} long-pending transactions to mark as FAILED")
+        return updates
+
     def update_deposits(self) -> None:
         """Update deposit statuses.
 
@@ -237,6 +290,9 @@ def update_deposits(self) -> None:
         rbf_updates = self.process_rbf_transactions(enriched_deposits, bitcoin_chaintip_height)
         updates.extend(rbf_updates)
 
+        # Process long-pending transactions
+        pending_updates = self.process_long_pending(enriched_deposits)
+        updates.extend(pending_updates)
         # Apply updates
         if updates:
             logger.info(f"Updating {len(updates)} deposit statuses")

emily_cron/app/settings.py

@@ -14,3 +14,6 @@
 
 # The number of confirmations required for a deposit update to be considered final
 MIN_BLOCK_CONFIRMATIONS = int(os.getenv("MIN_BLOCK_CONFIRMATIONS", 6))
+
+# Maximum time (in seconds) a transaction can remain unconfirmed before being marked as FAILED
+MAX_UNCONFIRMED_TIME = int(os.getenv("MAX_UNCONFIRMED_TIME", 60 * 60 * 24))  # 24 hours in seconds

emily_cron/main.py

@@ -27,6 +27,7 @@
 logger.info(f"MEMPOOL_API_URL: {settings.MEMPOOL_API_URL}")
 logger.info(f"HIRO_API_URL: {settings.HIRO_API_URL}")
 logger.info(f"MIN_BLOCK_CONFIRMATIONS: {settings.MIN_BLOCK_CONFIRMATIONS}")
+logger.info(f"MAX_UNCONFIRMED_TIME: {settings.MAX_UNCONFIRMED_TIME}")
 logger.info("=" * 80)