refactor: simplify .env.example and add bulk sync capabilities
Some checks failed
AITBC CI/CD Pipeline / lint-and-test (3.11) (pull_request) Has been cancelled
AITBC CI/CD Pipeline / lint-and-test (3.12) (pull_request) Has been cancelled
AITBC CI/CD Pipeline / lint-and-test (3.13) (pull_request) Has been cancelled
AITBC CLI Level 1 Commands Test / test-cli-level1 (3.11) (pull_request) Has been cancelled
AITBC CLI Level 1 Commands Test / test-cli-level1 (3.12) (pull_request) Has been cancelled
AITBC CLI Level 1 Commands Test / test-cli-level1 (3.13) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (apps/coordinator-api/src) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (cli/aitbc_cli) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (packages/py/aitbc-core/src) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (packages/py/aitbc-crypto/src) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (packages/py/aitbc-sdk/src) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (tests) (pull_request) Has been cancelled
Security Scanning / CodeQL Security Analysis (javascript) (pull_request) Has been cancelled
Security Scanning / CodeQL Security Analysis (python) (pull_request) Has been cancelled
Security Scanning / Dependency Security Scan (pull_request) Has been cancelled
Security Scanning / Container Security Scan (pull_request) Has been cancelled
Security Scanning / OSSF Scorecard (pull_request) Has been cancelled
AITBC CI/CD Pipeline / test-cli (pull_request) Has been cancelled
AITBC CI/CD Pipeline / test-services (pull_request) Has been cancelled
AITBC CI/CD Pipeline / test-production-services (pull_request) Has been cancelled
AITBC CI/CD Pipeline / security-scan (pull_request) Has been cancelled
AITBC CI/CD Pipeline / build (pull_request) Has been cancelled
AITBC CI/CD Pipeline / deploy-staging (pull_request) Has been cancelled
AITBC CI/CD Pipeline / deploy-production (pull_request) Has been cancelled
AITBC CI/CD Pipeline / performance-test (pull_request) Has been cancelled
AITBC CI/CD Pipeline / docs (pull_request) Has been cancelled
AITBC CI/CD Pipeline / release (pull_request) Has been cancelled
AITBC CI/CD Pipeline / notify (pull_request) Has been cancelled
AITBC CLI Level 1 Commands Test / test-summary (pull_request) Has been cancelled
Security Scanning / Security Summary Report (pull_request) Has been cancelled
Some checks failed
AITBC CI/CD Pipeline / lint-and-test (3.11) (pull_request) Has been cancelled
AITBC CI/CD Pipeline / lint-and-test (3.12) (pull_request) Has been cancelled
AITBC CI/CD Pipeline / lint-and-test (3.13) (pull_request) Has been cancelled
AITBC CLI Level 1 Commands Test / test-cli-level1 (3.11) (pull_request) Has been cancelled
AITBC CLI Level 1 Commands Test / test-cli-level1 (3.12) (pull_request) Has been cancelled
AITBC CLI Level 1 Commands Test / test-cli-level1 (3.13) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (apps/coordinator-api/src) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (cli/aitbc_cli) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (packages/py/aitbc-core/src) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (packages/py/aitbc-crypto/src) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (packages/py/aitbc-sdk/src) (pull_request) Has been cancelled
Security Scanning / Bandit Security Scan (tests) (pull_request) Has been cancelled
Security Scanning / CodeQL Security Analysis (javascript) (pull_request) Has been cancelled
Security Scanning / CodeQL Security Analysis (python) (pull_request) Has been cancelled
Security Scanning / Dependency Security Scan (pull_request) Has been cancelled
Security Scanning / Container Security Scan (pull_request) Has been cancelled
Security Scanning / OSSF Scorecard (pull_request) Has been cancelled
AITBC CI/CD Pipeline / test-cli (pull_request) Has been cancelled
AITBC CI/CD Pipeline / test-services (pull_request) Has been cancelled
AITBC CI/CD Pipeline / test-production-services (pull_request) Has been cancelled
AITBC CI/CD Pipeline / security-scan (pull_request) Has been cancelled
AITBC CI/CD Pipeline / build (pull_request) Has been cancelled
AITBC CI/CD Pipeline / deploy-staging (pull_request) Has been cancelled
AITBC CI/CD Pipeline / deploy-production (pull_request) Has been cancelled
AITBC CI/CD Pipeline / performance-test (pull_request) Has been cancelled
AITBC CI/CD Pipeline / docs (pull_request) Has been cancelled
AITBC CI/CD Pipeline / release (pull_request) Has been cancelled
AITBC CI/CD Pipeline / notify (pull_request) Has been cancelled
AITBC CLI Level 1 Commands Test / test-summary (pull_request) Has been cancelled
Security Scanning / Security Summary Report (pull_request) Has been cancelled
- Simplify .env.example to show actual config values instead of instructions - Add blockchain core, coordinator API, and marketplace web example sections - Add bulk import functionality to ChainSync for faster catchup - Add fetch_blocks_range and bulk_import_from methods with batching - Add init-production command to genesis CLI for production chain setup - Add HTTP client management with configurable batch size and poll interval
This commit is contained in:
@@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import hmac
|
||||
import time
|
||||
@@ -9,6 +10,7 @@ from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
import httpx
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from .config import settings
|
||||
@@ -95,12 +97,92 @@ class ChainSync:
|
||||
max_reorg_depth: int = 10,
|
||||
validator: Optional[ProposerSignatureValidator] = None,
|
||||
validate_signatures: bool = True,
|
||||
batch_size: int = 50,
|
||||
poll_interval: float = 0.5,
|
||||
) -> None:
|
||||
self._session_factory = session_factory
|
||||
self._chain_id = chain_id or settings.chain_id
|
||||
self._max_reorg_depth = max_reorg_depth
|
||||
self._validator = validator or ProposerSignatureValidator()
|
||||
self._validate_signatures = validate_signatures
|
||||
self._batch_size = batch_size
|
||||
self._poll_interval = poll_interval
|
||||
self._client = httpx.AsyncClient(timeout=10.0)
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Close HTTP client."""
|
||||
await self._client.aclose()
|
||||
|
||||
async def fetch_blocks_range(self, start: int, end: int, source_url: str) -> List[Dict[str, Any]]:
|
||||
"""Fetch a range of blocks from a source RPC."""
|
||||
try:
|
||||
resp = await self._client.get(f"{source_url}/rpc/blocks-range", params={"start": start, "end": end})
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
if isinstance(data, list):
|
||||
return data
|
||||
elif isinstance(data, dict) and "blocks" in data:
|
||||
return data["blocks"]
|
||||
else:
|
||||
logger.error("Unexpected blocks-range response", extra={"data": data})
|
||||
return []
|
||||
except Exception as e:
|
||||
logger.error("Failed to fetch blocks range", extra={"start": start, "end": end, "error": str(e)})
|
||||
return []
|
||||
|
||||
async def bulk_import_from(self, source_url: str, import_url: Optional[str] = None) -> int:
|
||||
"""Bulk import missing blocks from source to catch up quickly."""
|
||||
if import_url is None:
|
||||
import_url = "http://127.0.0.1:8006" # default local RPC
|
||||
|
||||
# Get local head
|
||||
with self._session_factory() as session:
|
||||
local_head = session.exec(
|
||||
select(Block).where(Block.chain_id == self._chain_id).order_by(Block.height.desc()).limit(1)
|
||||
).first()
|
||||
local_height = local_head.height if local_head else -1
|
||||
|
||||
# Get remote head
|
||||
try:
|
||||
resp = await self._client.get(f"{source_url}/rpc/head")
|
||||
resp.raise_for_status()
|
||||
remote_head = resp.json()
|
||||
remote_height = remote_head.get("height", -1)
|
||||
except Exception as e:
|
||||
logger.error("Failed to fetch remote head", extra={"source_url": source_url, "error": str(e)})
|
||||
return 0
|
||||
|
||||
if remote_height <= local_height:
|
||||
logger.info("Already up to date", extra={"local_height": local_height, "remote_height": remote_height})
|
||||
return 0
|
||||
|
||||
logger.info("Starting bulk import", extra={"local_height": local_height, "remote_height": remote_height, "batch_size": self._batch_size})
|
||||
|
||||
imported = 0
|
||||
start_height = local_height + 1
|
||||
while start_height <= remote_height:
|
||||
end_height = min(start_height + self._batch_size - 1, remote_height)
|
||||
batch = await self.fetch_blocks_range(start_height, end_height, source_url)
|
||||
if not batch:
|
||||
logger.warning("No blocks returned for range", extra={"start": start_height, "end": end_height})
|
||||
break
|
||||
|
||||
# Import blocks in order
|
||||
for block_data in batch:
|
||||
result = self.import_block(block_data)
|
||||
if result.accepted:
|
||||
imported += 1
|
||||
else:
|
||||
logger.warning("Block import failed during bulk", extra={"height": block_data.get("height"), "reason": result.reason})
|
||||
# Stop on first failure to avoid gaps
|
||||
break
|
||||
|
||||
start_height = end_height + 1
|
||||
# Brief pause to avoid overwhelming the DB
|
||||
await asyncio.sleep(self._poll_interval)
|
||||
|
||||
logger.info("Bulk import completed", extra={"imported": imported, "final_height": remote_height})
|
||||
return imported
|
||||
|
||||
def import_block(self, block_data: Dict[str, Any], transactions: Optional[List[Dict[str, Any]]] = None) -> ImportResult:
|
||||
"""Import a block from a remote peer.
|
||||
|
||||
42
apps/blockchain-node/src/aitbc_chain/sync_cli.py
Normal file
42
apps/blockchain-node/src/aitbc_chain/sync_cli.py
Normal file
@@ -0,0 +1,42 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Standalone bulk sync utility for fast catch-up.
|
||||
Usage: python -m aitbc_chain.sync_cli --source http://10.1.223.40:8006 [--batch-size 100]
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add src to path for standalone execution
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
|
||||
from aitbc_chain.config import Settings, settings
|
||||
from aitbc_chain.database import session_scope
|
||||
from aitbc_chain.sync import ChainSync
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Bulk import blocks from a leader to catch up quickly")
|
||||
parser.add_argument("--source", default="http://10.1.223.40:8006", help="Source RPC URL")
|
||||
parser.add_argument("--import-url", default="http://127.0.0.1:8006", help="Local RPC URL for import")
|
||||
parser.add_argument("--batch-size", type=int, default=100, help="Blocks per batch")
|
||||
parser.add_argument("--poll-interval", type=float, default=0.2, help="Seconds between batches")
|
||||
args = parser.parse_args()
|
||||
|
||||
sync = ChainSync(
|
||||
session_factory=session_scope,
|
||||
chain_id=settings.chain_id,
|
||||
batch_size=args.batch_size,
|
||||
poll_interval=args.poll_interval,
|
||||
)
|
||||
try:
|
||||
imported = await sync.bulk_import_from(args.source, import_url=args.import_url)
|
||||
print(f"[+] Bulk sync complete: imported {imported} blocks")
|
||||
finally:
|
||||
await sync.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user