Remove backup and temporary files from git
Deleted .bak, .backup, and .orig files: - 2 .orig files from blockchain-node - 9 .bak files from cli commands - 1 .bak file from dev scripts - 1 .backup file from docs - 1 .bak file from scripts These files add noise and should not be tracked in git.
This commit is contained in:
aitbc
@@ -1,229 +0,0 @@
|
||||
import asyncio
|
||||
import hashlib
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Callable, ContextManager, Optional
|
||||
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from ..logger import get_logger
|
||||
from ..metrics import metrics_registry
|
||||
from ..config import ProposerConfig
|
||||
from ..models import Block
|
||||
from ..gossip import gossip_broker
|
||||
|
||||
_METRIC_KEY_SANITIZE = re.compile(r"[^a-zA-Z0-9_]")
|
||||
|
||||
|
||||
def _sanitize_metric_suffix(value: str) -> str:
|
||||
sanitized = _METRIC_KEY_SANITIZE.sub("_", value).strip("_")
|
||||
return sanitized or "unknown"
|
||||
|
||||
|
||||
|
||||
import time
|
||||
|
||||
class CircuitBreaker:
|
||||
def __init__(self, threshold: int, timeout: int):
|
||||
self._threshold = threshold
|
||||
self._timeout = timeout
|
||||
self._failures = 0
|
||||
self._last_failure_time = 0.0
|
||||
self._state = "closed"
|
||||
|
||||
@property
|
||||
def state(self) -> str:
|
||||
if self._state == "open":
|
||||
if time.time() - self._last_failure_time > self._timeout:
|
||||
self._state = "half-open"
|
||||
return self._state
|
||||
|
||||
def allow_request(self) -> bool:
|
||||
state = self.state
|
||||
if state == "closed":
|
||||
return True
|
||||
if state == "half-open":
|
||||
return True
|
||||
return False
|
||||
|
||||
def record_failure(self) -> None:
|
||||
self._failures += 1
|
||||
self._last_failure_time = time.time()
|
||||
if self._failures >= self._threshold:
|
||||
self._state = "open"
|
||||
|
||||
def record_success(self) -> None:
|
||||
self._failures = 0
|
||||
self._state = "closed"
|
||||
|
||||
class PoAProposer:
|
||||
"""Proof-of-Authority block proposer.
|
||||
|
||||
Responsible for periodically proposing blocks if this node is configured as a proposer.
|
||||
In the real implementation, this would involve checking the mempool, validating transactions,
|
||||
and signing the block.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
config: ProposerConfig,
|
||||
session_factory: Callable[[], ContextManager[Session]],
|
||||
) -> None:
|
||||
self._config = config
|
||||
self._session_factory = session_factory
|
||||
self._logger = get_logger(__name__)
|
||||
self._stop_event = asyncio.Event()
|
||||
self._task: Optional[asyncio.Task[None]] = None
|
||||
self._last_proposer_id: Optional[str] = None
|
||||
|
||||
async def start(self) -> None:
|
||||
if self._task is not None:
|
||||
return
|
||||
self._logger.info("Starting PoA proposer loop", extra={"interval": self._config.interval_seconds})
|
||||
self._ensure_genesis_block()
|
||||
self._stop_event.clear()
|
||||
self._task = asyncio.create_task(self._run_loop())
|
||||
|
||||
async def stop(self) -> None:
|
||||
if self._task is None:
|
||||
return
|
||||
self._logger.info("Stopping PoA proposer loop")
|
||||
self._stop_event.set()
|
||||
await self._task
|
||||
self._task = None
|
||||
|
||||
async def _run_loop(self) -> None:
|
||||
while not self._stop_event.is_set():
|
||||
await self._wait_until_next_slot()
|
||||
if self._stop_event.is_set():
|
||||
break
|
||||
try:
|
||||
self._propose_block()
|
||||
except Exception as exc: # pragma: no cover - defensive logging
|
||||
self._logger.exception("Failed to propose block", extra={"error": str(exc)})
|
||||
|
||||
async def _wait_until_next_slot(self) -> None:
|
||||
head = self._fetch_chain_head()
|
||||
if head is None:
|
||||
return
|
||||
now = datetime.utcnow()
|
||||
elapsed = (now - head.timestamp).total_seconds()
|
||||
sleep_for = max(self._config.interval_seconds - elapsed, 0.1)
|
||||
if sleep_for <= 0:
|
||||
sleep_for = 0.1
|
||||
try:
|
||||
await asyncio.wait_for(self._stop_event.wait(), timeout=sleep_for)
|
||||
except asyncio.TimeoutError:
|
||||
return
|
||||
|
||||
async def _propose_block(self) -> None:
|
||||
# Check internal mempool
|
||||
from ..mempool import get_mempool
|
||||
if get_mempool().size(self._config.chain_id) == 0:
|
||||
return
|
||||
|
||||
with self._session_factory() as session:
|
||||
head = session.exec(select(Block).where(Block.chain_id == self._config.chain_id).order_by(Block.height.desc()).limit(1)).first()
|
||||
next_height = 0
|
||||
parent_hash = "0x00"
|
||||
interval_seconds: Optional[float] = None
|
||||
if head is not None:
|
||||
next_height = head.height + 1
|
||||
parent_hash = head.hash
|
||||
interval_seconds = (datetime.utcnow() - head.timestamp).total_seconds()
|
||||
|
||||
timestamp = datetime.utcnow()
|
||||
block_hash = self._compute_block_hash(next_height, parent_hash, timestamp)
|
||||
|
||||
block = Block(
|
||||
chain_id=self._config.chain_id,
|
||||
height=next_height,
|
||||
hash=block_hash,
|
||||
parent_hash=parent_hash,
|
||||
proposer=self._config.proposer_id,
|
||||
timestamp=timestamp,
|
||||
tx_count=0,
|
||||
state_root=None,
|
||||
)
|
||||
session.add(block)
|
||||
session.commit()
|
||||
|
||||
metrics_registry.increment("blocks_proposed_total")
|
||||
metrics_registry.set_gauge("chain_head_height", float(next_height))
|
||||
if interval_seconds is not None and interval_seconds >= 0:
|
||||
metrics_registry.observe("block_interval_seconds", interval_seconds)
|
||||
metrics_registry.set_gauge("poa_last_block_interval_seconds", float(interval_seconds))
|
||||
|
||||
proposer_suffix = _sanitize_metric_suffix(self._config.proposer_id)
|
||||
metrics_registry.increment(f"poa_blocks_proposed_total_{proposer_suffix}")
|
||||
if self._last_proposer_id is not None and self._last_proposer_id != self._config.proposer_id:
|
||||
metrics_registry.increment("poa_proposer_switches_total")
|
||||
self._last_proposer_id = self._config.proposer_id
|
||||
|
||||
self._logger.info(
|
||||
"Proposed block",
|
||||
extra={
|
||||
"height": block.height,
|
||||
"hash": block.hash,
|
||||
"proposer": block.proposer,
|
||||
},
|
||||
)
|
||||
|
||||
# Broadcast the new block
|
||||
await gossip_broker.publish(
|
||||
"blocks",
|
||||
{
|
||||
"height": block.height,
|
||||
"hash": block.hash,
|
||||
"parent_hash": block.parent_hash,
|
||||
"proposer": block.proposer,
|
||||
"timestamp": block.timestamp.isoformat(),
|
||||
"tx_count": block.tx_count,
|
||||
"state_root": block.state_root,
|
||||
}
|
||||
)
|
||||
|
||||
async def _ensure_genesis_block(self) -> None:
|
||||
with self._session_factory() as session:
|
||||
head = session.exec(select(Block).where(Block.chain_id == self._config.chain_id).order_by(Block.height.desc()).limit(1)).first()
|
||||
if head is not None:
|
||||
return
|
||||
|
||||
# Use a deterministic genesis timestamp so all nodes agree on the genesis block hash
|
||||
timestamp = datetime(2025, 1, 1, 0, 0, 0)
|
||||
block_hash = self._compute_block_hash(0, "0x00", timestamp)
|
||||
genesis = Block(
|
||||
chain_id=self._config.chain_id,
|
||||
height=0,
|
||||
hash=block_hash,
|
||||
parent_hash="0x00",
|
||||
proposer="genesis",
|
||||
timestamp=timestamp,
|
||||
tx_count=0,
|
||||
state_root=None,
|
||||
)
|
||||
session.add(genesis)
|
||||
session.commit()
|
||||
|
||||
# Broadcast genesis block for initial sync
|
||||
await gossip_broker.publish(
|
||||
"blocks",
|
||||
{
|
||||
"height": genesis.height,
|
||||
"hash": genesis.hash,
|
||||
"parent_hash": genesis.parent_hash,
|
||||
"proposer": genesis.proposer,
|
||||
"timestamp": genesis.timestamp.isoformat(),
|
||||
"tx_count": genesis.tx_count,
|
||||
"state_root": genesis.state_root,
|
||||
}
|
||||
)
|
||||
|
||||
def _fetch_chain_head(self) -> Optional[Block]:
|
||||
with self._session_factory() as session:
|
||||
return session.exec(select(Block).order_by(Block.height.desc()).limit(1)).first()
|
||||
|
||||
def _compute_block_hash(self, height: int, parent_hash: str, timestamp: datetime) -> str:
|
||||
payload = f"{self._config.chain_id}|{height}|{parent_hash}|{timestamp.isoformat()}".encode()
|
||||
return "0x" + hashlib.sha256(payload).hexdigest()
|
||||
@@ -1,307 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import warnings
|
||||
from collections import defaultdict
|
||||
from contextlib import suppress
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable, Dict, List, Optional, Set
|
||||
|
||||
warnings.filterwarnings("ignore", message="coroutine.* was never awaited", category=RuntimeWarning)
|
||||
|
||||
try:
|
||||
from broadcaster import Broadcast
|
||||
except ImportError: # pragma: no cover
|
||||
Broadcast = None # type: ignore[assignment]
|
||||
|
||||
from ..metrics import metrics_registry
|
||||
|
||||
|
||||
def _increment_publication(metric_prefix: str, topic: str) -> None:
|
||||
metrics_registry.increment(f"{metric_prefix}_total")
|
||||
metrics_registry.increment(f"{metric_prefix}_topic_{topic}")
|
||||
|
||||
|
||||
def _set_queue_gauge(topic: str, size: int) -> None:
|
||||
metrics_registry.set_gauge(f"gossip_queue_size_{topic}", float(size))
|
||||
|
||||
|
||||
def _update_subscriber_metrics(topics: Dict[str, List["asyncio.Queue[Any]"]]) -> None:
|
||||
for topic, queues in topics.items():
|
||||
metrics_registry.set_gauge(f"gossip_subscribers_topic_{topic}", float(len(queues)))
|
||||
total = sum(len(queues) for queues in topics.values())
|
||||
metrics_registry.set_gauge("gossip_subscribers_total", float(total))
|
||||
|
||||
|
||||
def _clear_topic_metrics(topic: str) -> None:
|
||||
metrics_registry.set_gauge(f"gossip_subscribers_topic_{topic}", 0.0)
|
||||
_set_queue_gauge(topic, 0)
|
||||
|
||||
@dataclass
|
||||
class TopicSubscription:
|
||||
topic: str
|
||||
queue: "asyncio.Queue[Any]"
|
||||
_unsubscribe: Callable[[], None]
|
||||
|
||||
def close(self) -> None:
|
||||
self._unsubscribe()
|
||||
|
||||
async def get(self) -> Any:
|
||||
return await self.queue.get()
|
||||
|
||||
async def __aiter__(self): # type: ignore[override]
|
||||
try:
|
||||
while True:
|
||||
yield await self.queue.get()
|
||||
finally:
|
||||
self.close()
|
||||
|
||||
|
||||
class GossipBackend:
|
||||
async def start(self) -> None: # pragma: no cover - overridden as needed
|
||||
return None
|
||||
|
||||
async def publish(self, topic: str, message: Any) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
async def subscribe(self, topic: str, max_queue_size: int = 100) -> TopicSubscription:
|
||||
raise NotImplementedError
|
||||
|
||||
async def shutdown(self) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class InMemoryGossipBackend(GossipBackend):
|
||||
def __init__(self) -> None:
|
||||
self._topics: Dict[str, List["asyncio.Queue[Any]"]] = defaultdict(list)
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
async def publish(self, topic: str, message: Any) -> None:
|
||||
async with self._lock:
|
||||
queues = list(self._topics.get(topic, []))
|
||||
for queue in queues:
|
||||
await queue.put(message)
|
||||
_set_queue_gauge(topic, queue.qsize())
|
||||
_increment_publication("gossip_publications", topic)
|
||||
|
||||
async def subscribe(self, topic: str, max_queue_size: int = 100) -> TopicSubscription:
|
||||
queue: "asyncio.Queue[Any]" = asyncio.Queue(maxsize=max_queue_size)
|
||||
|
||||
async with self._lock:
|
||||
self._topics[topic].append(queue)
|
||||
_update_subscriber_metrics(self._topics)
|
||||
|
||||
_set_queue_gauge(topic, queue.qsize())
|
||||
|
||||
def _unsubscribe() -> None:
|
||||
async def _remove() -> None:
|
||||
async with self._lock:
|
||||
queues = self._topics.get(topic)
|
||||
if queues is None:
|
||||
return
|
||||
if queue in queues:
|
||||
queues.remove(queue)
|
||||
if not queues:
|
||||
self._topics.pop(topic, None)
|
||||
_clear_topic_metrics(topic)
|
||||
_update_subscriber_metrics(self._topics)
|
||||
|
||||
asyncio.create_task(_remove())
|
||||
|
||||
return TopicSubscription(topic=topic, queue=queue, _unsubscribe=_unsubscribe)
|
||||
|
||||
async def shutdown(self) -> None:
|
||||
async with self._lock:
|
||||
topics = list(self._topics.keys())
|
||||
self._topics.clear()
|
||||
for topic in topics:
|
||||
_clear_topic_metrics(topic)
|
||||
_update_subscriber_metrics(self._topics)
|
||||
|
||||
|
||||
class BroadcastGossipBackend(GossipBackend):
|
||||
def __init__(self, url: str) -> None:
|
||||
if Broadcast is None: # provide in-process fallback when Broadcast is missing
|
||||
self._broadcast = _InProcessBroadcast()
|
||||
else:
|
||||
self._broadcast = Broadcast(url) # type: ignore[arg-type]
|
||||
self._tasks: Set[asyncio.Task[None]] = set()
|
||||
self._lock = asyncio.Lock()
|
||||
self._running = False
|
||||
|
||||
async def start(self) -> None:
|
||||
if not self._running:
|
||||
await self._broadcast.connect() # type: ignore[union-attr]
|
||||
self._running = True
|
||||
|
||||
async def publish(self, topic: str, message: Any) -> None:
|
||||
if not self._running:
|
||||
raise RuntimeError("Broadcast backend not started")
|
||||
payload = _encode_message(message)
|
||||
await self._broadcast.publish(topic, payload) # type: ignore[union-attr]
|
||||
_increment_publication("gossip_broadcast_publications", topic)
|
||||
|
||||
async def subscribe(self, topic: str, max_queue_size: int = 100) -> TopicSubscription:
|
||||
if not self._running:
|
||||
raise RuntimeError("Broadcast backend not started")
|
||||
|
||||
queue: "asyncio.Queue[Any]" = asyncio.Queue(maxsize=max_queue_size)
|
||||
stop_event = asyncio.Event()
|
||||
|
||||
async def _run_subscription() -> None:
|
||||
async with self._broadcast.subscribe(topic) as subscriber: # type: ignore[attr-defined,union-attr]
|
||||
async for event in subscriber: # type: ignore[union-attr]
|
||||
if stop_event.is_set():
|
||||
break
|
||||
data = _decode_message(getattr(event, "message", event))
|
||||
try:
|
||||
await queue.put(data)
|
||||
_set_queue_gauge(topic, queue.qsize())
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
|
||||
task = asyncio.create_task(_run_subscription(), name=f"broadcast-sub:{topic}")
|
||||
async with self._lock:
|
||||
self._tasks.add(task)
|
||||
metrics_registry.set_gauge("gossip_broadcast_subscribers_total", float(len(self._tasks)))
|
||||
|
||||
def _unsubscribe() -> None:
|
||||
async def _stop() -> None:
|
||||
stop_event.set()
|
||||
task.cancel()
|
||||
with suppress(asyncio.CancelledError):
|
||||
await task
|
||||
async with self._lock:
|
||||
self._tasks.discard(task)
|
||||
metrics_registry.set_gauge("gossip_broadcast_subscribers_total", float(len(self._tasks)))
|
||||
|
||||
asyncio.create_task(_stop())
|
||||
|
||||
return TopicSubscription(topic=topic, queue=queue, _unsubscribe=_unsubscribe)
|
||||
|
||||
async def shutdown(self) -> None:
|
||||
async with self._lock:
|
||||
tasks = list(self._tasks)
|
||||
self._tasks.clear()
|
||||
metrics_registry.set_gauge("gossip_broadcast_subscribers_total", 0.0)
|
||||
for task in tasks:
|
||||
task.cancel()
|
||||
with suppress(asyncio.CancelledError):
|
||||
await task
|
||||
if self._running:
|
||||
await self._broadcast.disconnect() # type: ignore[union-attr]
|
||||
self._running = False
|
||||
|
||||
|
||||
class GossipBroker:
|
||||
def __init__(self, backend: GossipBackend) -> None:
|
||||
self._backend = backend
|
||||
self._lock = asyncio.Lock()
|
||||
self._started = False
|
||||
|
||||
async def publish(self, topic: str, message: Any) -> None:
|
||||
if not self._started:
|
||||
await self._backend.start()
|
||||
self._started = True
|
||||
await self._backend.publish(topic, message)
|
||||
|
||||
async def subscribe(self, topic: str, max_queue_size: int = 100) -> TopicSubscription:
|
||||
if not self._started:
|
||||
await self._backend.start()
|
||||
self._started = True
|
||||
return await self._backend.subscribe(topic, max_queue_size=max_queue_size)
|
||||
|
||||
async def set_backend(self, backend: GossipBackend) -> None:
|
||||
await backend.start()
|
||||
async with self._lock:
|
||||
previous = self._backend
|
||||
self._backend = backend
|
||||
self._started = True
|
||||
await previous.shutdown()
|
||||
|
||||
async def shutdown(self) -> None:
|
||||
await self._backend.shutdown()
|
||||
|
||||
|
||||
class _InProcessSubscriber:
|
||||
def __init__(self, queue: "asyncio.Queue[Any]"):
|
||||
self._queue = queue
|
||||
|
||||
def __aiter__(self): # type: ignore[override]
|
||||
return self._iterator()
|
||||
|
||||
async def _iterator(self):
|
||||
while True:
|
||||
yield await self._queue.get()
|
||||
|
||||
|
||||
class _InProcessBroadcast:
|
||||
"""Minimal in-memory broadcast substitute for tests when Starlette Broadcast is absent."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._topics: Dict[str, List["asyncio.Queue[Any]"]] = defaultdict(list)
|
||||
self._lock = asyncio.Lock()
|
||||
self._running = False
|
||||
|
||||
async def connect(self) -> None:
|
||||
self._running = True
|
||||
|
||||
async def disconnect(self) -> None:
|
||||
async with self._lock:
|
||||
self._topics.clear()
|
||||
self._running = False
|
||||
|
||||
@asynccontextmanager
|
||||
async def subscribe(self, topic: str):
|
||||
queue: "asyncio.Queue[Any]" = asyncio.Queue()
|
||||
async with self._lock:
|
||||
self._topics[topic].append(queue)
|
||||
|
||||
try:
|
||||
yield _InProcessSubscriber(queue)
|
||||
finally:
|
||||
async with self._lock:
|
||||
queues = self._topics.get(topic)
|
||||
if queues and queue in queues:
|
||||
queues.remove(queue)
|
||||
|
||||
async def publish(self, topic: str, message: Any) -> None:
|
||||
if not self._running:
|
||||
raise RuntimeError("Broadcast backend not started")
|
||||
async with self._lock:
|
||||
queues = list(self._topics.get(topic, []))
|
||||
for queue in queues:
|
||||
await queue.put(message)
|
||||
|
||||
|
||||
def create_backend(backend_type: str, *, broadcast_url: Optional[str] = None) -> GossipBackend:
|
||||
backend = backend_type.lower()
|
||||
if backend in {"memory", "inmemory", "local"}:
|
||||
return InMemoryGossipBackend()
|
||||
if backend in {"broadcast", "starlette", "redis"}:
|
||||
if not broadcast_url:
|
||||
raise ValueError("Broadcast backend requires a gossip_broadcast_url setting")
|
||||
return BroadcastGossipBackend(broadcast_url)
|
||||
raise ValueError(f"Unsupported gossip backend '{backend_type}'")
|
||||
|
||||
|
||||
def _encode_message(message: Any) -> Any:
|
||||
if isinstance(message, (str, bytes, bytearray)):
|
||||
return message
|
||||
return json.dumps(message, separators=(",", ":"))
|
||||
|
||||
|
||||
def _decode_message(message: Any) -> Any:
|
||||
if isinstance(message, (bytes, bytearray)):
|
||||
message = message.decode("utf-8")
|
||||
if isinstance(message, str):
|
||||
try:
|
||||
return json.loads(message)
|
||||
except json.JSONDecodeError:
|
||||
return message
|
||||
return message
|
||||
|
||||
|
||||
gossip_broker = GossipBroker(InMemoryGossipBackend())
|
||||
|
||||
Reference in New Issue
Block a user