Merge pull request #309 from Chucks1093/feat/htlc-clients-matching-solana-15-23-35-55

feat: add HTLC tracking, chain clients, and order matching

Author: floxxih

.github/workflows/ci.yml

@@ -219,9 +219,15 @@ jobs:
           ${{ runner.os }}-cargo-tests-
           
     - name: Run integration tests
+      if: ${{ hashFiles('tests/Cargo.toml') != '' }}
       run: cargo test --manifest-path tests/Cargo.toml --verbose
-      
+
+    - name: Skip missing integration crate
+      if: ${{ hashFiles('tests/Cargo.toml') == '' }}
+      run: echo "tests/Cargo.toml not present; skipping integration-tests job."
+
     - name: Run stress tests
+      if: ${{ hashFiles('tests/Cargo.toml') != '' }}
       run: cargo test --manifest-path tests/Cargo.toml -- stress --test-threads=1
 
   code-quality:

backend/app/config/settings.py

@@ -1,3 +1,4 @@
+from pydantic import field_validator
 from pydantic_settings import BaseSettings
 
 
@@ -6,7 +7,9 @@ class Settings(BaseSettings):
     debug: bool = False
 
     # Database
-    database_url: str = "postgresql+asyncpg://chainbridge:password@localhost:5432/chainbridge"
+    database_url: str = (
+        "postgresql+asyncpg://chainbridge:password@localhost:5432/chainbridge"
+    )
 
     # Redis
     redis_url: str = "redis://localhost:6379/0"
@@ -25,6 +28,17 @@ class Settings(BaseSettings):
     soroban_rpc_url: str = "https://soroban-testnet.stellar.org"
     chainbridge_contract_id: str = ""
 
+    @field_validator("debug", mode="before")
+    @classmethod
+    def parse_debug(cls, value):
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in {"release", "prod", "production", "false", "0", "off"}:
+                return False
+            if normalized in {"debug", "dev", "development", "true", "1", "on"}:
+                return True
+        return value
+
     class Config:
         env_file = ".env"
 

backend/app/indexer/base.py

@@ -56,9 +56,7 @@ async def get_latest_block(self) -> int:
         """Get the latest confirmed block number on the chain."""
 
     @abstractmethod
-    async def fetch_events(
-        self, from_block: int, to_block: int
-    ) -> list[IndexedEvent]:
+    async def fetch_events(self, from_block: int, to_block: int) -> list[IndexedEvent]:
         """Fetch events from a block range."""
 
     @abstractmethod

backend/app/indexer/bitcoin_indexer.py

@@ -2,6 +2,7 @@
 
 import logging
 import os
+from asyncio import sleep
 from datetime import datetime
 
 import httpx
@@ -20,27 +21,41 @@ def __init__(self):
         self.rpc_user = os.getenv("BITCOIN_RPC_USER", "")
         self.rpc_password = os.getenv("BITCOIN_RPC_PASSWORD", "")
         self.confirmations = int(os.getenv("BITCOIN_CONFIRMATIONS", "6"))
+        self.max_retries = int(os.getenv("BITCOIN_RPC_RETRIES", "3"))
+        self._last_block_hash: str | None = None
 
     async def _rpc_call(self, method: str, params: list = None) -> dict:
         """Make a JSON-RPC call to the Bitcoin node."""
-        async with httpx.AsyncClient() as client:
-            response = await client.post(
-                self.rpc_url,
-                json={
-                    "jsonrpc": "2.0",
-                    "id": 1,
-                    "method": method,
-                    "params": params or [],
-                },
-                auth=(self.rpc_user, self.rpc_password)
-                if self.rpc_user
-                else None,
-                timeout=30,
-            )
-            data = response.json()
-            if "error" in data and data["error"]:
-                raise Exception(f"Bitcoin RPC error: {data['error']}")
-            return data.get("result", {})
+        last_error: Exception | None = None
+        for attempt in range(1, self.max_retries + 1):
+            try:
+                async with httpx.AsyncClient() as client:
+                    response = await client.post(
+                        self.rpc_url,
+                        json={
+                            "jsonrpc": "2.0",
+                            "id": 1,
+                            "method": method,
+                            "params": params or [],
+                        },
+                        auth=(
+                            (self.rpc_user, self.rpc_password)
+                            if self.rpc_user
+                            else None
+                        ),
+                        timeout=30,
+                    )
+                    response.raise_for_status()
+                    data = response.json()
+                    if "error" in data and data["error"]:
+                        raise RuntimeError(f"Bitcoin RPC error: {data['error']}")
+                    return data.get("result", {})
+            except Exception as exc:
+                last_error = exc
+                if attempt == self.max_retries:
+                    break
+                await sleep(attempt)
+        raise RuntimeError("Bitcoin RPC call failed") from last_error
 
     async def get_latest_block(self) -> int:
         try:
@@ -51,13 +66,13 @@ async def get_latest_block(self) -> int:
             logger.error("[bitcoin] Failed to get block count: %s", e)
             raise
 
-    async def fetch_events(
-        self, from_block: int, to_block: int
-    ) -> list[IndexedEvent]:
+    async def fetch_events(self, from_block: int, to_block: int) -> list[IndexedEvent]:
+        await self.detect_reorg(from_block)
         events = []
         for height in range(from_block, to_block + 1):
             try:
                 block_hash = await self._rpc_call("getblockhash", [height])
+                self._last_block_hash = block_hash
                 block = await self._rpc_call("getblock", [block_hash, 2])
 
                 for tx in block.get("tx", []):
@@ -95,6 +110,21 @@ async def handle_reorg(self, reorg_block: int) -> None:
         # In production: delete indexed events >= reorg_block from DB
         # and re-index from reorg_block
 
+    async def detect_reorg(self, height: int) -> None:
+        if height <= 0 or not self._last_block_hash:
+            return
+        current_hash = await self._rpc_call("getblockhash", [height - 1])
+        if current_hash != self._last_block_hash:
+            await self.handle_reorg(height - 1)
+
+    async def get_mempool_transactions(self) -> list[str]:
+        mempool = await self._rpc_call("getrawmempool")
+        return list(mempool or [])
+
+    async def broadcast_transaction(self, raw_tx: str) -> str:
+        tx_hash = await self._rpc_call("sendrawtransaction", [raw_tx])
+        return str(tx_hash)
+
     def _is_bridge_tx(self, asm: str) -> bool:
         """Check if an OP_RETURN contains a ChainBridge marker."""
         # ChainBridge transactions use a specific prefix in OP_RETURN

backend/app/indexer/ethereum_indexer.py

@@ -2,6 +2,7 @@
 
 import logging
 import os
+from asyncio import sleep
 from datetime import datetime
 
 import httpx
@@ -21,24 +22,36 @@ def __init__(self):
         )
         self.contract_address = os.getenv("ETHEREUM_BRIDGE_CONTRACT", "")
         self.confirmations = int(os.getenv("ETHEREUM_CONFIRMATIONS", "12"))
+        self.max_retries = int(os.getenv("ETHEREUM_RPC_RETRIES", "3"))
+        self._last_block_hash: str | None = None
 
     async def _rpc_call(self, method: str, params: list = None) -> dict:
         """Make a JSON-RPC call to the Ethereum node."""
-        async with httpx.AsyncClient() as client:
-            response = await client.post(
-                self.rpc_url,
-                json={
-                    "jsonrpc": "2.0",
-                    "id": 1,
-                    "method": method,
-                    "params": params or [],
-                },
-                timeout=30,
-            )
-            data = response.json()
-            if "error" in data and data["error"]:
-                raise Exception(f"Ethereum RPC error: {data['error']}")
-            return data.get("result")
+        last_error: Exception | None = None
+        for attempt in range(1, self.max_retries + 1):
+            try:
+                async with httpx.AsyncClient() as client:
+                    response = await client.post(
+                        self.rpc_url,
+                        json={
+                            "jsonrpc": "2.0",
+                            "id": 1,
+                            "method": method,
+                            "params": params or [],
+                        },
+                        timeout=30,
+                    )
+                    response.raise_for_status()
+                    data = response.json()
+                    if "error" in data and data["error"]:
+                        raise RuntimeError(f"Ethereum RPC error: {data['error']}")
+                    return data.get("result")
+            except Exception as exc:
+                last_error = exc
+                if attempt == self.max_retries:
+                    break
+                await sleep(attempt)
+        raise RuntimeError("Ethereum RPC call failed") from last_error
 
     async def get_latest_block(self) -> int:
         try:
@@ -49,12 +62,11 @@ async def get_latest_block(self) -> int:
             logger.error("[ethereum] Failed to get block number: %s", e)
             raise
 
-    async def fetch_events(
-        self, from_block: int, to_block: int
-    ) -> list[IndexedEvent]:
+    async def fetch_events(self, from_block: int, to_block: int) -> list[IndexedEvent]:
         if not self.contract_address:
             return []
 
+        await self.detect_reorg(from_block)
         events = []
         try:
             logs = await self._rpc_call(
@@ -71,10 +83,19 @@ async def fetch_events(
             for log in logs or []:
                 try:
                     block_num = int(log["blockNumber"], 16)
+                    block_data = await self._rpc_call(
+                        "eth_getBlockByNumber",
+                        [log["blockNumber"], False],
+                    )
+                    self._last_block_hash = (
+                        block_data.get("hash") if block_data else None
+                    )
                     events.append(
                         IndexedEvent(
                             chain="ethereum",
-                            event_type=log["topics"][0] if log.get("topics") else "unknown",
+                            event_type=(
+                                log["topics"][0] if log.get("topics") else "unknown"
+                            ),
                             tx_hash=log.get("transactionHash", ""),
                             block_number=block_num,
                             contract_address=log.get("address"),
@@ -101,3 +122,23 @@ async def handle_reorg(self, reorg_block: int) -> None:
             reorg_block,
         )
         # In production: delete indexed events >= reorg_block from DB
+
+    async def detect_reorg(self, from_block: int) -> None:
+        if from_block <= 0 or not self._last_block_hash:
+            return
+        previous_block = await self._rpc_call(
+            "eth_getBlockByNumber", [hex(from_block - 1), False]
+        )
+        if previous_block and previous_block.get("hash") != self._last_block_hash:
+            await self.handle_reorg(from_block - 1)
+
+    async def get_mempool_transactions(self) -> list[str]:
+        txpool = await self._rpc_call("txpool_content")
+        if not isinstance(txpool, dict):
+            return []
+        pending = txpool.get("pending", {})
+        return list(pending.keys())
+
+    async def broadcast_transaction(self, raw_tx: str) -> str:
+        tx_hash = await self._rpc_call("eth_sendRawTransaction", [raw_tx])
+        return str(tx_hash)

backend/app/indexer/manager.py

@@ -7,7 +7,8 @@
 
 import asyncio
 import logging
-from datetime import datetime
+
+from app.config.redis import CacheService, get_redis
 
 from .base import BaseIndexer, IndexerStatus
 from .stellar_indexer import StellarIndexer
@@ -35,6 +36,7 @@ def __init__(self):
             "ethereum": EthereumIndexer(),
         }
         self._tasks: dict[str, asyncio.Task] = {}
+        self._status_task: asyncio.Task | None = None
 
     async def start_all(
         self,
@@ -51,6 +53,7 @@ async def start_all(
             )
             self._tasks[chain] = task
             logger.info("Started %s indexer from block %d", chain, from_block)
+        self._status_task = asyncio.create_task(self._publish_status_loop())
 
     async def stop_all(self) -> None:
         """Stop all running indexers."""
@@ -64,6 +67,14 @@ async def stop_all(self) -> None:
             except asyncio.CancelledError:
                 pass
 
+        if self._status_task:
+            self._status_task.cancel()
+            try:
+                await self._status_task
+            except asyncio.CancelledError:
+                pass
+            self._status_task = None
+
         self._tasks.clear()
         logger.info("All indexers stopped")
 
@@ -80,9 +91,7 @@ def get_all_status(self) -> dict[str, dict]:
                 "blocks_behind": s.blocks_behind,
                 "events_processed": s.events_processed,
                 "last_error": s.last_error,
-                "last_sync_at": s.last_sync_at.isoformat()
-                if s.last_sync_at
-                else None,
+                "last_sync_at": s.last_sync_at.isoformat() if s.last_sync_at else None,
             }
         return statuses
 
@@ -109,3 +118,12 @@ async def catch_up(self, chain: str, from_block: int, to_block: int) -> int:
             to_block,
         )
         return len(events)
+
+    async def _publish_status_loop(self) -> None:
+        cache = CacheService(get_redis())
+        while True:
+            statuses = self.get_all_status()
+            for chain, status in statuses.items():
+                status["last_updated"] = status["last_sync_at"]
+                await cache.set(f"indexer:status:{chain}", status, ttl=60)
+            await asyncio.sleep(5)

backend/app/indexer/stellar_indexer.py

@@ -17,9 +17,7 @@ class StellarIndexer(BaseIndexer):
 
     def __init__(self):
         super().__init__(chain="stellar")
-        rpc_url = os.getenv(
-            "SOROBAN_RPC_URL", "https://soroban-testnet.stellar.org"
-        )
+        rpc_url = os.getenv("SOROBAN_RPC_URL", "https://soroban-testnet.stellar.org")
         self.server = SorobanServer(rpc_url)
         self.contract_id = os.getenv("CHAINBRIDGE_CONTRACT_ID", "")
 
@@ -31,21 +29,21 @@ async def get_latest_block(self) -> int:
             logger.error("[stellar] Failed to get latest ledger: %s", e)
             raise
 
-    async def fetch_events(
-        self, from_block: int, to_block: int
-    ) -> list[IndexedEvent]:
+    async def fetch_events(self, from_block: int, to_block: int) -> list[IndexedEvent]:
         events = []
         try:
             response = self.server.get_events(
                 start_ledger=from_block,
-                filters=[
-                    {
-                        "type": "contract",
-                        "contractIds": [self.contract_id],
-                    }
-                ]
-                if self.contract_id
-                else [],
+                filters=(
+                    [
+                        {
+                            "type": "contract",
+                            "contractIds": [self.contract_id],
+                        }
+                    ]
+                    if self.contract_id
+                    else []
+                ),
                 limit=200,
             )