From 3bb15220979afefdefdc402864e54b5cbd74bf8d Mon Sep 17 00:00:00 2001 From: anshulchikhale30-p Date: Sun, 8 Nov 2026 14:59:31 +0530 Subject: [PATCH 1/4] fix: sync with validated headers before downloading bodies (#178) Replace the full 500-block chain_request sync path with a small getheaders/getblocks protocol. - Add BlockHeader and validate hash, linkage, target bounds, proof-of-work, and timestamps before requesting any body. - Download bodies by header hash in bounded batches, respecting the P2P frame size, and rate-limit sync requests per peer. - Apply linear batches incrementally; buffer only fork branches for resolve_conflicts and return orphaned transactions to the mempool. - Keep a bounded legacy chain_request path for old peers; ignore unsolicited legacy chain_response messages. - Document the protocol and cover locators, validation failures, response bounds, pagination, rate limiting, and fork/reorg sync with tests. --- README.md | 2 +- docs/sync.md | 155 +++++++++ main.py | 128 ++++---- minichain/__init__.py | 5 +- minichain/block.py | 110 ++++++- minichain/network_config.py | 16 +- minichain/node_config.py | 12 + minichain/sync.py | 607 ++++++++++++++++++++++++++++++++++++ tests/test_sync.py | 491 +++++++++++++++++++++++++++++ 9 files changed, 1434 insertions(+), 92 deletions(-) create mode 100644 docs/sync.md create mode 100644 minichain/sync.py create mode 100644 tests/test_sync.py diff --git a/README.md b/README.md index c3fc105..42d7780 100644 --- a/README.md +++ b/README.md @@ -93,7 +93,7 @@ To connect a secondary node to the network, start a new instance on a different ```bash python main.py --port 9001 --connect /ip4/127.0.0.1/tcp/9000/p2p/ --datadir ./node2_data ``` -The node will automatically sync the blockchain state via the P2P network using the Fork-Choice rule. +The node will automatically sync the blockchain state via the P2P network using headers-first sync and the Fork-Choice rule (see [`docs/sync.md`](docs/sync.md)). ### 3. Mining Blocks To confirm pending transactions, you need to mine blocks. In the interactive CLI of your node, simply type: diff --git a/docs/sync.md b/docs/sync.md new file mode 100644 index 0000000..55ac026 --- /dev/null +++ b/docs/sync.md @@ -0,0 +1,155 @@ +# Headers-first sync + +MiniChain syncs in two stages: + +1. Ask a peer for the **headers** that follow a block we already know. +2. Validate those headers, then download their **bodies** in bounded batches. + +Headers are small and cheap to verify. A peer must prove chain linkage, +proof-of-work, and plausible timestamps before we spend bandwidth on +transactions and receipts. Bodies are applied exactly as before, so consensus +rules have not changed. + +## Messages + +The sync protocol uses four message types. All payloads are JSON, and responses +echo the requester's `request_id`. + +### `getheaders` → `headers` + +```json +{ + "type": "getheaders", + "data": { + "request_id": "5f0c...", + "locator": ["", "...", ""], + "max": 1000 + } +} +``` + +```json +{ + "type": "headers", + "data": { + "request_id": "5f0c...", + "headers": [ + { + "index": 42, + "previous_hash": "...", + "merkle_root": "...", + "state_root": "...", + "receipt_root": "...", + "timestamp": 1716880000000, + "target": "0x00ff...", + "nonce": 12345, + "miner": "...", + "hash": "..." + } + ] + } +} +``` + +The responder finds the first hash in `locator` that it knows and returns the +headers after it. A locator is newest-first with exponentially spaced entries, +anchored at genesis, so it stays small even on a long chain. + +### `getblocks` → `blocks` + +```json +{ + "type": "getblocks", + "data": { + "request_id": "9ab1...", + "hashes": ["
", "..."] + } +} +``` + +```json +{ + "type": "blocks", + "data": { + "request_id": "9ab1...", + "blocks": [ + { + "index": 42, + "previous_hash": "...", + "target": "0x00ff...", + "transactions": [], + "receipts": [], + "hash": "..." + } + ] + } +} +``` + +A response is always a contiguous prefix of the requested hashes. It may be +shorter than the request when a block is unknown or when the response would +exceed the frame budget. The requester continues from the last received block. + +## Validation + +Header validation happens before any body is requested: + +- header hash matches the header fields; +- `index` and `previous_hash` link to the known parent; +- `target` is within `[MIN_TARGET, MAX_TARGET]`; +- `int(hash, 16) < target` proves the declared proof-of-work; +- timestamps strictly increase and are not too far in the future; +- the advertised headers represent more work than our current suffix. + +Body application then enforces all remaining consensus rules: transactions, +receipts, state roots, and the exact EMA difficulty target for each height. + +MiniChain does not persist the EMA target and average block time for every +historical block. A fork's exact target sequence therefore cannot be +reconstructed at the header stage, so it is checked when the fork bodies are +applied. This keeps the header stage cheap without weakening consensus. + +## Limits and peer behaviour + +Defaults live in [`minichain/node_config.py`](../minichain/node_config.py): + +| Setting | Default | Purpose | +| --- | --- | --- | +| `MAX_HEADERS_PER_REQUEST` | 1000 | Header batch cap (about 540 KB of JSON) | +| `MAX_BLOCKS_PER_REQUEST` | 64 | Body batch cap | +| `MAX_LOCATOR_HASHES` | 32 | Locator cap | +| `MAX_SYNC_RESPONSE_BYTES` | 768 KiB | Response byte budget, below the 1 MiB frame limit | +| `SYNC_REQUEST_TIMEOUT_SECONDS` | 30 | Drop a stalled sync session | +| `SYNC_REQUESTS_PER_SECOND` / `SYNC_REQUEST_BURST` | 5 / 10 | Per-peer token bucket for serving sync requests | + +Out-of-range requests and malformed sync payloads are returned to the P2P layer +as validation statuses, so existing peer scoring and banning apply. A peer that +advertises a chain with no more work than ours is ignored rather than punished. + +## Reorgs + +When the common ancestor is not our tip, headers describe a competing branch. +Bodies for that branch are buffered and handed to `Blockchain.resolve_conflicts`. +If the branch is heavier and valid, the chain switches and transactions that are +no longer confirmed are returned to the mempool. Otherwise the session is +dropped. + +## Compatibility + +`chain_request` and `chain_response` are still accepted for older peers: + +- A new node never sends them. +- A legacy `chain_request` is answered with at most `MAX_BLOCKS_PER_REQUEST` + blocks, so an old peer cannot ask for an unbounded response. +- A legacy `chain_response` is ignored because this node no longer initiates + that request. + +Old nodes cannot serve `getheaders` and therefore cannot be used as sync +sources by a new node. They can still download blocks from a new node through +the legacy path. + +## Tests + +`tests/test_sync.py` covers header serialization and validation, locator +spacing, request and response bounds, rate limiting, linear sync with header and +body pagination, early rejection of invalid headers, and fork/reorg sync. diff --git a/main.py b/main.py index 27597d2..a521dc5 100644 --- a/main.py +++ b/main.py @@ -33,6 +33,7 @@ from minichain import Transaction, Blockchain, Block, State, Mempool, P2PNetwork, mine_block from minichain.rpc import JSONRPCServer +from minichain.sync import SyncManager from minichain.validators import is_valid_receiver, ValidationStatus from minichain.block import calculate_receipt_root from minichain.persistence import remember_peer, get_known_peers, is_peer_banned @@ -80,12 +81,6 @@ def load_or_create_wallet(datadir: str | None): return sk, pk -def request_chain(network, start_index, limit): - """Broadcast a chain_request for blocks starting at start_index.""" - req = {"type": "chain_request", "data": {"start_index": start_index, "limit": limit}} - asyncio.create_task(network._broadcast_raw(req)) - - def parse_tx_params(parts, start): """Parse optional non-negative amount, gas_limit, fee_per_gas. Returns (amount, gas_limit, fee_per_gas), or None if invalid.""" @@ -190,16 +185,32 @@ def mine_and_process_block(chain, mempool, miner_pk): # Network message handler # ────────────────────────────────────────────── -def make_network_handler(chain, mempool, network): +def make_sync_manager(chain, mempool, network): + """Wire headers-first sync to the network and mempool.""" + return SyncManager( + chain, + send=lambda peer, payload: asyncio.create_task( + network._unicast_raw(peer, payload) + ), + on_block_applied=lambda block: mempool.remove_transactions(block.transactions), + on_orphans=lambda orphans: [mempool.add_transaction(tx) for tx in orphans], + ) + + +def make_network_handler(chain, mempool, network, sync_manager=None): """Return an async callback that processes incoming P2P messages.""" + sync = sync_manager or make_sync_manager(chain, mempool, network) async def handler(data): msg_type = data.get("type") payload = data.get("data") peer_addr = data.get("_peer_addr", "unknown") - if payload is None and msg_type in ("hello", "chain_request", "chain_response", "getaddr", "addr"): - return + if payload is None: + if sync.handles(msg_type): + return ValidationStatus.MALFORMED + if msg_type in ("hello", "chain_request", "chain_response", "getaddr", "addr"): + return if msg_type == "hello": peer_chain_id = payload.get("chain_id") @@ -216,13 +227,19 @@ async def handler(data): logger.info("🔄 Handshake successful with %s", peer_addr) peer_tip = payload.get("latest_block_index", 0) if peer_tip > chain.last_block.index: - logger.info("📡 Peer %s is ahead (%d > %d). Initiating chunked sync...", peer_addr, peer_tip, chain.last_block.index) - request_chain(network, chain.last_block.index + 1, 500) + logger.info( + "📡 Peer %s is ahead (%d > %d). Starting headers-first sync...", + peer_addr, peer_tip, chain.last_block.index, + ) + sync.request_sync(peer_addr) # Ask for more peers, so the network stays reachable after whoever we # were told about goes away, instead of only ever knowing our seed. asyncio.create_task(network._unicast_raw(peer_addr, {"type": "getaddr", "data": {}})) + elif sync.handles(msg_type): + return sync.handle_message(peer_addr, msg_type, payload) + elif msg_type == "tx": try: tx = Transaction.from_dict(payload) @@ -257,77 +274,37 @@ async def handler(data): mempool.remove_transactions(block.transactions) else: if block.index > chain.last_block.index + 1: - logger.warning("📥 Received Block #%s — ahead of us (tip: %s). Requesting chunked sync...", block.index, chain.last_block.index) - request_chain(network, chain.last_block.index + 1, 500) + logger.warning( + "📥 Received Block #%s — ahead of us (tip: %s). Starting headers-first sync...", + block.index, chain.last_block.index, + ) + sync.request_sync(peer_addr) else: - logger.warning("📥 Received Block #%s — rejected. Fork detected, trigger reorg sync.", block.index) - request_chain(network, max(0, block.index - 50), 50) + logger.warning( + "📥 Received Block #%s — rejected. Fork detected, starting headers-first sync.", + block.index, + ) + sync.request_sync(peer_addr) return status elif msg_type == "chain_request": - start_index = payload.get("start_index", 0) - limit = payload.get("limit", 500) + # Deprecated compatibility path. New nodes use getheaders/getblocks; + # this path is still bounded so an old peer cannot request an + # arbitrarily large response. + try: + start_index = max(0, int(payload.get("start_index", 0))) + limit = max(1, min(int(payload.get("limit", sync.max_blocks)), sync.max_blocks)) + except (TypeError, ValueError): + return ValidationStatus.MALFORMED logger.info("📡 Peer requested blocks from %d (limit %d).", start_index, limit) - - if start_index < len(chain.chain): - blocks_slice = chain.chain[start_index : start_index + limit] - blocks_dicts = [b.to_dict() for b in blocks_slice] - else: - blocks_dicts = [] - - resp_payload = {"type": "chain_response", "data": {"blocks": blocks_dicts, "requested_limit": limit}} + resp_payload = sync.build_legacy_blocks_response(start_index, limit) asyncio.create_task(network._unicast_raw(peer_addr, resp_payload)) elif msg_type == "chain_response": - blocks_payload = payload.get("blocks", []) - requested_limit = payload.get("requested_limit", 500) - if not blocks_payload: - return - - new_chain = [] - try: - new_chain = [Block.from_dict(b) for b in blocks_payload] - except Exception as e: - logger.warning("❌ Failed to parse chain_response: %s", e) - return - - if new_chain: - # Differentiate between linear catchup and reorg - if new_chain[0].index == chain.last_block.index + 1 and new_chain[0].previous_hash == chain.last_block.hash: - # Linear Catch-up - all_added = True - for block in new_chain: - if block.index <= chain.last_block.index: - continue # Ignore already known blocks - if chain.add_block(block) == ValidationStatus.VALID: - logger.info("📥 Synced Block #%d", block.index) - mempool.remove_transactions(block.transactions) - else: - logger.warning("❌ Sync failed at Block #%d. Fork detected. Requesting chunk backward.", block.index) - request_chain(network, max(0, block.index - 50), 50) - all_added = False - break - - if all_added and len(new_chain) == requested_limit: - next_index = chain.last_block.index + 1 - logger.info("📡 Requesting next batch from index %d", next_index) - request_chain(network, next_index, requested_limit) - else: - # Fork / Reorg sync - success, orphans = chain.resolve_conflicts(new_chain) - if success: - logger.info("🔄 Reorg complete! Restoring %d orphaned txs to mempool.", len(orphans)) - for tx in orphans: - mempool.add_transaction(tx) - - if len(new_chain) == requested_limit: - next_index = chain.last_block.index + 1 - request_chain(network, next_index, requested_limit) - else: - logger.warning("❌ Reorg failed. Fetching earlier blocks to find fork point...") - earlier_start = max(0, new_chain[0].index - 50) - if new_chain[0].index > 0: - request_chain(network, earlier_start, 50) + # Deprecated: this node no longer sends chain_request. Ignoring an + # unsolicited response avoids keeping two competing sync paths alive. + logger.debug("Ignoring deprecated chain_response from %s", peer_addr) + return elif msg_type == "getaddr": # Share only peers we have ourselves successfully dialed: an inbound @@ -662,7 +639,8 @@ async def run_node(port: int, host: str, connect_to: str | None, fund: int, data mempool = Mempool() network = P2PNetwork(data_path=datadir or ".") - handler = make_network_handler(chain, mempool, network) + sync = make_sync_manager(chain, mempool, network) + handler = make_network_handler(chain, mempool, network, sync) network.register_handler(handler) rpc_server = JSONRPCServer(chain, mempool, network) diff --git a/minichain/__init__.py b/minichain/__init__.py index ae52604..2cd156e 100644 --- a/minichain/__init__.py +++ b/minichain/__init__.py @@ -1,5 +1,5 @@ from .pow import mine_block, calculate_hash, MiningExceededError -from .block import Block +from .block import Block, BlockHeader from .chain import Blockchain from .transaction import Transaction from .state import State @@ -7,18 +7,21 @@ from .p2p import P2PNetwork from .mempool import Mempool from .persistence import save, load +from .sync import SyncManager __all__ = [ "mine_block", "calculate_hash", "MiningExceededError", "Block", + "BlockHeader", "Blockchain", "Transaction", "State", "ContractMachine", "P2PNetwork", "Mempool", + "SyncManager", "save", "load", ] diff --git a/minichain/block.py b/minichain/block.py index ee00551..c4503a1 100644 --- a/minichain/block.py +++ b/minichain/block.py @@ -1,5 +1,6 @@ import time import hashlib +from dataclasses import dataclass, replace from typing import Sequence, Optional from .transaction import Transaction @@ -33,6 +34,100 @@ def calculate_receipt_root(receipts: Sequence[Receipt]) -> Optional[str]: return None return _calculate_merkle_tree([canonical_json_hash(r.to_dict()) for r in receipts]) + +@dataclass(frozen=True) +class BlockHeader: + """The consensus-critical header of a block, without its body. + + Headers are what make headers-first sync cheap: a peer can prove proof-of-work + and chain linkage before this node downloads transactions and receipts. + """ + + index: int + previous_hash: str + merkle_root: Optional[str] + state_root: Optional[str] + receipt_root: Optional[str] + timestamp: int + target: int + nonce: int + miner: Optional[str] = None + hash: Optional[str] = None + + @classmethod + def from_block(cls, block: "Block") -> "BlockHeader": + return cls( + index=block.index, + previous_hash=block.previous_hash, + merkle_root=block.merkle_root, + state_root=block.state_root, + receipt_root=block.receipt_root, + timestamp=block.timestamp, + target=block.target, + nonce=block.nonce, + miner=block.miner, + hash=block.hash, + ) + + @classmethod + def from_dict(cls, payload: dict) -> "BlockHeader": + if not isinstance(payload, dict): + raise ValueError("header must be an object") + + try: + raw_target = payload["target"] + target = int(raw_target, 16) if isinstance(raw_target, str) else int(raw_target) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("invalid or missing header target") from exc + + from .network_config import MAX_TARGET + if not isinstance(target, int) or target <= 0 or target > MAX_TARGET: + raise ValueError(f"invalid header target: {target}") + + try: + header = cls( + index=int(payload["index"]), + previous_hash=str(payload["previous_hash"]), + merkle_root=payload.get("merkle_root"), + state_root=payload.get("state_root"), + receipt_root=payload.get("receipt_root"), + timestamp=int(payload["timestamp"]), + target=target, + nonce=int(payload.get("nonce") or 0), + miner=payload.get("miner"), + hash=payload.get("hash"), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("invalid header fields") from exc + + expected_hash = header.compute_hash() + if header.hash is not None and header.hash != expected_hash: + raise ValueError("header hash does not match header") + if header.hash is None: + header = replace(header, hash=expected_hash) + return header + + def to_dict(self, *, include_hash: bool = True) -> dict: + header = { + "index": self.index, + "previous_hash": self.previous_hash, + "merkle_root": self.merkle_root, + "state_root": self.state_root, + "receipt_root": self.receipt_root, + "timestamp": self.timestamp, + "target": hex(self.target), + "nonce": self.nonce, + } + if self.miner is not None: + header["miner"] = self.miner + if include_hash and self.hash is not None: + header["hash"] = self.hash + return header + + def compute_hash(self) -> str: + return canonical_json_hash(self.to_dict(include_hash=False)) + + class Block: def __init__( self, @@ -75,20 +170,7 @@ def __init__( # HEADER (used for mining) # ------------------------- def to_header_dict(self): - header = { - "index": self.index, - "previous_hash": self.previous_hash, - "merkle_root": self.merkle_root, - "state_root": self.state_root, - "receipt_root": self.receipt_root, - "timestamp": self.timestamp, - "target": hex(self.target), - "nonce": self.nonce, - } - # Include miner in header only when present (optional field) - if self.miner is not None: - header["miner"] = self.miner - return header + return BlockHeader.from_block(self).to_dict(include_hash=False) # ------------------------- # BODY (transactions only) diff --git a/minichain/network_config.py b/minichain/network_config.py index 3ffbf7a..427f1e6 100644 --- a/minichain/network_config.py +++ b/minichain/network_config.py @@ -5,7 +5,21 @@ # to maintain consensus. # P2P Network Rules -SUPPORTED_MESSAGE_TYPES = {"hello", "tx", "block", "chain_request", "chain_response", "getaddr", "addr"} +SUPPORTED_MESSAGE_TYPES = { + "hello", + "tx", + "block", + "getheaders", + "headers", + "getblocks", + "blocks", + # Retained for backward compatibility with nodes that predate headers-first + # sync. New nodes never send these messages. + "chain_request", + "chain_response", + "getaddr", + "addr", +} PROTOCOL_ID = TProtocol("/minichain/1.0.0") MAX_FRAME_BYTES = 1 * 1024 * 1024 # 1 MB diff --git a/minichain/node_config.py b/minichain/node_config.py index 0128139..fe7e7fc 100644 --- a/minichain/node_config.py +++ b/minichain/node_config.py @@ -29,3 +29,15 @@ # State Config MAX_STATE_SNAPSHOTS = 10 # Number of recent block states to keep in memory for reorg optimization + +# Sync Config (node-local policy; not consensus) +# 1000 JSON headers are roughly 540 KB, comfortably below MAX_FRAME_BYTES. +MAX_HEADERS_PER_REQUEST = 1000 +MAX_BLOCKS_PER_REQUEST = 64 +MAX_LOCATOR_HASHES = 32 +# Leave headroom below MAX_FRAME_BYTES (1 MB) for envelope overhead and +# multiple small blocks sharing one response frame. +MAX_SYNC_RESPONSE_BYTES = 768 * 1024 +SYNC_REQUEST_TIMEOUT_SECONDS = 30.0 +SYNC_REQUESTS_PER_SECOND = 5.0 +SYNC_REQUEST_BURST = 10 diff --git a/minichain/sync.py b/minichain/sync.py new file mode 100644 index 0000000..b52a7b3 --- /dev/null +++ b/minichain/sync.py @@ -0,0 +1,607 @@ +"""Headers-first block synchronization. + +The node used to ask for a range of full blocks and only discover an invalid +chain after downloading every body. This module separates the cheap consensus +checks (header hash, linkage, proof-of-work, timestamp) from body download, so a +peer must prove its chain is plausible and heavier before we request blocks. + +The protocol is intentionally small: + + getheaders -> headers + getblocks -> blocks + +A locator is a list of recent block hashes, newest first. The responding node +finds the first hash it knows and returns the headers that follow it. Bodies are +then requested by hash in bounded batches. This is the same shape as Bitcoin's +headers-first sync, without trying to reproduce the rest of Bitcoin's protocol. +""" + +from __future__ import annotations + +import logging +import time +import uuid +from collections.abc import Callable, Sequence +from dataclasses import dataclass, field + +from .block import Block, BlockHeader +from .network_config import MAX_FUTURE_BLOCK_TIME_MS, MAX_TARGET, MIN_TARGET +from .node_config import ( + MAX_BLOCKS_PER_REQUEST, + MAX_HEADERS_PER_REQUEST, + MAX_LOCATOR_HASHES, + MAX_SYNC_RESPONSE_BYTES, + SYNC_REQUEST_BURST, + SYNC_REQUEST_TIMEOUT_SECONDS, + SYNC_REQUESTS_PER_SECOND, +) +from .serialization import canonical_json_dumps +from .validators import ValidationStatus + +logger = logging.getLogger(__name__) + +SYNC_MESSAGE_TYPES = frozenset({"getheaders", "headers", "getblocks", "blocks"}) + + +class SyncError(ValueError): + """A peer sent a sync message that cannot be processed safely.""" + + def __init__(self, message: str, status: ValidationStatus = ValidationStatus.MALFORMED): + super().__init__(message) + self.status = status + + +class PeerRateLimiter: + """Small token bucket, one per peer, for expensive sync requests.""" + + def __init__( + self, + rate_per_second: float = SYNC_REQUESTS_PER_SECOND, + burst: int = SYNC_REQUEST_BURST, + clock: Callable[[], float] = time.monotonic, + ): + if rate_per_second <= 0: + raise ValueError("rate_per_second must be positive") + if burst <= 0: + raise ValueError("burst must be positive") + self.rate_per_second = float(rate_per_second) + self.burst = float(burst) + self._clock = clock + self._buckets: dict[str, tuple[float, float]] = {} + + def allow(self, peer_addr: str) -> bool: + now = self._clock() + tokens, last_seen = self._buckets.get(peer_addr, (self.burst, now)) + elapsed = max(0.0, now - last_seen) + tokens = min(self.burst, tokens + elapsed * self.rate_per_second) + if tokens < 1.0: + self._buckets[peer_addr] = (tokens, now) + return False + self._buckets[peer_addr] = (tokens - 1.0, now) + return True + + +@dataclass +class _PendingRequest: + peer_addr: str + kind: str + created_at: float + + +@dataclass +class _SyncSession: + peer_addr: str + headers: list[BlockHeader] + fork: bool + next_index: int = 0 + blocks: list[Block] = field(default_factory=list) + + +def _bounded_int(value, *, default: int, minimum: int, maximum: int) -> int: + if value is None: + return default + if isinstance(value, bool): + raise SyncError("boolean is not a valid integer") + try: + parsed = int(value) + except (TypeError, ValueError) as exc: + raise SyncError("invalid integer value") from exc + return max(minimum, min(maximum, parsed)) + + +def _validate_hashes(value, *, maximum: int, field_name: str) -> list[str]: + if value is None: + return [] + if not isinstance(value, list): + raise SyncError(f"{field_name} must be a list") + if len(value) > maximum: + raise SyncError(f"{field_name} exceeds the maximum of {maximum}") + if not all(isinstance(item, str) and item for item in value): + raise SyncError(f"{field_name} must contain non-empty strings") + return value + + +def _total_work(blocks: Sequence[Block | BlockHeader]) -> int: + return sum((1 << 256) // block.target for block in blocks) + + +def validate_header_chain( + headers: Sequence[BlockHeader], + start_block: Block, + *, + now_ms: int | None = None, +) -> None: + """Validate a contiguous header range that starts after ``start_block``. + + This deliberately mirrors the cheap checks in ``chain.add_block``. The + expected difficulty for each height is still enforced when the bodies are + applied: MiniChain does not persist the EMA value per block, so a fork's + target sequence cannot be reconstructed without replaying state. Header + validation still stops bad linkage, bad proof-of-work and timestamps before + any body is downloaded. + """ + + previous: Block | BlockHeader = start_block + max_allowed_time = ( + int(time.time() * 1000) if now_ms is None else now_ms + ) + MAX_FUTURE_BLOCK_TIME_MS + + for header in headers: + if header.hash is None: + raise SyncError("header is missing its hash") + if header.index != previous.index + 1: + raise SyncError( + f"invalid header index {header.index}, expected {previous.index + 1}" + ) + if header.previous_hash != previous.hash: + raise SyncError( + f"invalid previous hash {header.previous_hash} != {previous.hash}" + ) + if header.hash != header.compute_hash(): + raise SyncError(f"invalid header hash {header.hash}") + if not isinstance(header.target, int) or not (MIN_TARGET <= header.target <= MAX_TARGET): + raise SyncError(f"invalid target {header.target}") + try: + hash_value = int(header.hash, 16) + except ValueError as exc: + raise SyncError(f"invalid header hash {header.hash}") from exc + if hash_value >= header.target: + raise SyncError( + f"invalid proof of work: hash {header.hash} does not satisfy target {header.target}" + ) + if header.timestamp <= previous.timestamp: + raise SyncError( + f"invalid timestamp {header.timestamp} <= previous {previous.timestamp}" + ) + if header.timestamp > max_allowed_time: + raise SyncError(f"timestamp {header.timestamp} is too far in the future") + previous = header + + +class SyncManager: + """Drives one headers-first sync per peer and answers sync requests.""" + + def __init__( + self, + chain, + send: Callable[[str, dict], None], + *, + on_block_applied: Callable[[Block], None] | None = None, + on_orphans: Callable[[Sequence], None] | None = None, + max_headers: int = MAX_HEADERS_PER_REQUEST, + max_blocks: int = MAX_BLOCKS_PER_REQUEST, + max_response_bytes: int = MAX_SYNC_RESPONSE_BYTES, + locator_size: int = MAX_LOCATOR_HASHES, + request_timeout: float = SYNC_REQUEST_TIMEOUT_SECONDS, + rate_limiter: PeerRateLimiter | None = None, + clock: Callable[[], float] = time.monotonic, + request_id_factory: Callable[[], str] = lambda: uuid.uuid4().hex, + ): + if max_headers <= 0 or max_blocks <= 0: + raise ValueError("sync batch sizes must be positive") + if max_response_bytes <= 0: + raise ValueError("max_response_bytes must be positive") + if locator_size <= 0: + raise ValueError("locator_size must be positive") + self.chain = chain + self._send = send + self._on_block_applied = on_block_applied + self._on_orphans = on_orphans + self.max_headers = max_headers + self.max_blocks = max_blocks + self.max_response_bytes = max_response_bytes + self.locator_size = locator_size + self.request_timeout = request_timeout + self._rate_limiter = rate_limiter or PeerRateLimiter(clock=clock) + self._clock = clock + self._request_id_factory = request_id_factory + self._pending: dict[str, _PendingRequest] = {} + self._sessions: dict[str, _SyncSession] = {} + + @staticmethod + def handles(message_type: str) -> bool: + return message_type in SYNC_MESSAGE_TYPES + + # ── outbound sync ──────────────────────────────────────────────────────── + + def request_sync(self, peer_addr: str) -> None: + """Ask a peer for headers that follow our current best block.""" + if peer_addr in self._sessions: + return + if any( + pending.peer_addr == peer_addr and pending.kind == "headers" + for pending in self._pending.values() + ): + return + + request_id = self._new_request_id() + self._pending[request_id] = _PendingRequest( + peer_addr=peer_addr, + kind="headers", + created_at=self._clock(), + ) + self._send( + peer_addr, + { + "type": "getheaders", + "data": { + "request_id": request_id, + "locator": self.locator(), + "max": self.max_headers, + }, + }, + ) + + def locator(self) -> list[str]: + """Return recent block hashes, newest first, with exponential spacing.""" + chain = self.chain.chain + if not chain: + return [] + + hashes: list[str] = [] + index = len(chain) - 1 + step = 1 + while index >= 0 and len(hashes) < self.locator_size: + hashes.append(chain[index].hash) + if len(hashes) >= 10: + step *= 2 + index -= step + + genesis_hash = chain[0].hash + if genesis_hash not in hashes: + if len(hashes) >= self.locator_size: + hashes[-1] = genesis_hash + else: + hashes.append(genesis_hash) + return hashes + + # ── inbound message handling ───────────────────────────────────────────── + + def handle_message(self, peer_addr: str, message_type: str, payload) -> ValidationStatus | None: + """Handle one sync message. Returns a validation status for P2P scoring.""" + self._prune_pending() + if not isinstance(payload, dict): + return ValidationStatus.MALFORMED + + if message_type in ("getheaders", "getblocks"): + if not self._rate_limiter.allow(peer_addr): + logger.warning("Rate-limited sync request from %s", peer_addr) + return ValidationStatus.FAILED + try: + response = ( + self.build_headers_response(payload) + if message_type == "getheaders" + else self.build_blocks_response(payload) + ) + except SyncError as exc: + logger.warning("Rejected %s from %s: %s", message_type, peer_addr, exc) + return exc.status + self._send(peer_addr, response) + return None + + if message_type == "headers": + try: + return self._handle_headers(peer_addr, payload) + except SyncError as exc: + logger.warning("Rejected headers from %s: %s", peer_addr, exc) + self._drop_session(peer_addr) + return exc.status + + if message_type == "blocks": + try: + return self._handle_blocks(peer_addr, payload) + except SyncError as exc: + logger.warning("Rejected blocks from %s: %s", peer_addr, exc) + self._drop_session(peer_addr) + return exc.status + + return ValidationStatus.MALFORMED + + def build_headers_response(self, payload: dict) -> dict: + locator = _validate_hashes( + payload.get("locator"), + maximum=self.locator_size, + field_name="locator", + ) + max_headers = _bounded_int( + payload.get("max"), + default=self.max_headers, + minimum=1, + maximum=self.max_headers, + ) + stop_hash = payload.get("stop_hash") + if stop_hash is not None and not isinstance(stop_hash, str): + raise SyncError("stop_hash must be a string or null") + + start_block = self.find_common_ancestor(locator) + start_index = self._index_of(start_block) if start_block is not None else -1 + budget = max(1, self.max_response_bytes - 512) + used = 0 + headers: list[dict] = [] + for block in self.chain.chain[start_index + 1 : start_index + 1 + max_headers]: + if stop_hash is not None and block.hash == stop_hash: + break + header = BlockHeader.from_block(block).to_dict() + size = len(canonical_json_dumps(header)) + 1 + if used + size > budget: + break + headers.append(header) + used += size + + response = {"type": "headers", "data": {"headers": headers}} + if isinstance(payload.get("request_id"), str): + response["data"]["request_id"] = payload["request_id"] + return response + + def build_blocks_response(self, payload: dict) -> dict: + hashes = _validate_hashes( + payload.get("hashes"), + maximum=self.max_blocks, + field_name="hashes", + ) + by_hash = {block.hash: block for block in self.chain.chain} + selected = [] + for block_hash in hashes: + block = by_hash.get(block_hash) + if block is None: + # Stop at the first gap: the requester applies a contiguous + # prefix of its requested batch. + break + selected.append(block) + + response = { + "type": "blocks", + "data": {"blocks": self._bounded_block_payloads(selected)}, + } + if isinstance(payload.get("request_id"), str): + response["data"]["request_id"] = payload["request_id"] + return response + + def build_legacy_blocks_response(self, start_index: int, limit: int) -> dict: + """Build a bounded ``chain_response`` for pre-headers-first peers.""" + blocks = self.chain.chain[start_index : start_index + limit] + payloads = self._bounded_block_payloads(blocks) + return { + "type": "chain_response", + "data": { + "blocks": payloads, + # Legacy peers continue only when the batch is exactly + # requested_limit, so report what we actually sent. + "requested_limit": len(payloads), + }, + } + + def _bounded_block_payloads(self, blocks: Sequence[Block]) -> list[dict]: + budget = max(1, self.max_response_bytes - 512) + used = 0 + payloads: list[dict] = [] + for block in blocks: + payload = block.to_dict() + size = len(canonical_json_dumps(payload)) + 1 + if used + size > budget: + break + payloads.append(payload) + used += size + return payloads + + # ── inbound responses ──────────────────────────────────────────────────── + + def _handle_headers(self, peer_addr: str, payload: dict) -> ValidationStatus | None: + pending = self._take_pending(peer_addr, "headers", payload.get("request_id")) + if pending is None: + # Unsolicited responses are ignored rather than punished: a peer may + # have answered a request that timed out locally. + return None + + raw_headers = payload.get("headers") + if not isinstance(raw_headers, list): + raise SyncError("headers must be a list") + if len(raw_headers) > self.max_headers: + raise SyncError(f"too many headers: {len(raw_headers)}") + + if not raw_headers: + self._drop_session(peer_addr) + return None + + try: + headers = [BlockHeader.from_dict(raw) for raw in raw_headers] + except (TypeError, ValueError) as exc: + raise SyncError(f"invalid header payload: {exc}") from exc + + start_block = self._find_block(headers[0].previous_hash) + if start_block is None: + raise SyncError("headers do not connect to a known block") + + validate_header_chain(headers, start_block) + + if _total_work(headers) <= self._suffix_work(start_block): + logger.info("Peer %s advertised a chain with no more work; not syncing", peer_addr) + self._drop_session(peer_addr) + return None + + session = _SyncSession( + peer_addr=peer_addr, + headers=headers, + fork=start_block.hash != self.chain.last_block.hash, + ) + self._sessions[peer_addr] = session + self._request_next_blocks(session) + return None + + def _handle_blocks(self, peer_addr: str, payload: dict) -> ValidationStatus | None: + pending = self._take_pending(peer_addr, "blocks", payload.get("request_id")) + if pending is None: + return None + + session = self._sessions.get(peer_addr) + if session is None: + return None + + raw_blocks = payload.get("blocks") + if not isinstance(raw_blocks, list): + raise SyncError("blocks must be a list") + if len(raw_blocks) > self.max_blocks: + raise SyncError(f"too many blocks: {len(raw_blocks)}") + + expected_headers = session.headers[session.next_index : session.next_index + self.max_blocks] + if not raw_blocks: + raise SyncError("empty blocks response") + if len(raw_blocks) > len(expected_headers): + raise SyncError( + f"expected at most {len(expected_headers)} blocks, received {len(raw_blocks)}" + ) + + try: + blocks = [Block.from_dict(raw) for raw in raw_blocks] + except (KeyError, TypeError, ValueError) as exc: + raise SyncError(f"invalid block payload: {exc}") from exc + + for block, header in zip(blocks, expected_headers): + if block.hash != header.hash: + raise SyncError(f"received block {block.hash}, expected {header.hash}") + + if session.fork: + session.blocks.extend(blocks) + else: + for block in blocks: + status = self.chain.add_block(block) + if status != ValidationStatus.VALID: + self._drop_session(peer_addr) + return status + self._notify_block_applied(block) + + session.next_index += len(blocks) + + if session.next_index < len(session.headers): + self._request_next_blocks(session) + return None + + if session.fork: + success, orphans = self.chain.resolve_conflicts(session.blocks) + if not success: + self._drop_session(peer_addr) + return ValidationStatus.INVALID + for block in session.blocks: + self._notify_block_applied(block) + self._notify_orphans(orphans) + + self._finish_session(session) + return None + + # ── session helpers ────────────────────────────────────────────────────── + + def _request_next_blocks(self, session: _SyncSession) -> None: + batch = session.headers[session.next_index : session.next_index + self.max_blocks] + if not batch: + self._finish_session(session) + return + + request_id = self._new_request_id() + self._pending[request_id] = _PendingRequest( + peer_addr=session.peer_addr, + kind="blocks", + created_at=self._clock(), + ) + self._send( + session.peer_addr, + { + "type": "getblocks", + "data": { + "request_id": request_id, + "hashes": [header.hash for header in batch], + }, + }, + ) + + def _finish_session(self, session: _SyncSession) -> None: + self._sessions.pop(session.peer_addr, None) + # A full header batch may be truncated; ask for the next one from our + # new tip. A short batch means the peer has caught us up. + if len(session.headers) == self.max_headers: + self.request_sync(session.peer_addr) + + def _drop_session(self, peer_addr: str) -> None: + self._sessions.pop(peer_addr, None) + for request_id, pending in list(self._pending.items()): + if pending.peer_addr == peer_addr: + self._pending.pop(request_id, None) + + def _take_pending(self, peer_addr: str, kind: str, request_id) -> _PendingRequest | None: + if not isinstance(request_id, str): + return None + pending = self._pending.get(request_id) + if pending is None or pending.peer_addr != peer_addr or pending.kind != kind: + return None + return self._pending.pop(request_id) + + def _prune_pending(self) -> None: + now = self._clock() + for request_id, pending in list(self._pending.items()): + if now - pending.created_at > self.request_timeout: + self._pending.pop(request_id, None) + self._drop_session(pending.peer_addr) + + def _new_request_id(self) -> str: + return self._request_id_factory() + + # ── chain helpers ──────────────────────────────────────────────────────── + + def _find_block(self, block_hash: str) -> Block | None: + for block in reversed(self.chain.chain): + if block.hash == block_hash: + return block + return None + + def _index_of(self, block: Block) -> int: + for index, candidate in enumerate(self.chain.chain): + if candidate.hash == block.hash: + return index + raise SyncError("block is not part of the local chain") + + def _suffix_work(self, start_block: Block) -> int: + start_index = self._index_of(start_block) + return _total_work(self.chain.chain[start_index + 1 :]) + + def find_common_ancestor(self, locator: Sequence[str]) -> Block | None: + by_hash = {block.hash: block for block in self.chain.chain} + for block_hash in locator: + block = by_hash.get(block_hash) + if block is not None: + return block + return None + + def _notify_block_applied(self, block: Block) -> None: + if self._on_block_applied is None: + return + try: + self._on_block_applied(block) + except Exception: + logger.exception("on_block_applied callback failed") + + def _notify_orphans(self, orphans: Sequence) -> None: + if self._on_orphans is None: + return + try: + self._on_orphans(orphans) + except Exception: + logger.exception("on_orphans callback failed") diff --git a/tests/test_sync.py b/tests/test_sync.py new file mode 100644 index 0000000..c255aab --- /dev/null +++ b/tests/test_sync.py @@ -0,0 +1,491 @@ +"""Tests for headers-first synchronization and the wire format it uses.""" + +import asyncio +import json +from dataclasses import replace +from types import SimpleNamespace + +import pytest +from nacl.encoding import HexEncoder +from nacl.signing import SigningKey + +from main import make_network_handler, make_sync_manager, mine_and_process_block +from minichain.block import Block, BlockHeader +from minichain.chain import Blockchain +from minichain.mempool import Mempool +from minichain.network_config import ( + MAX_FRAME_BYTES, + MAX_FUTURE_BLOCK_TIME_MS, + MAX_TARGET, +) +from minichain.serialization import canonical_json_dumps +from minichain.sync import ( + PeerRateLimiter, + SyncError, + SyncManager, + validate_header_chain, +) +from minichain.transaction import Transaction +from minichain.validators import ValidationStatus + + +@pytest.fixture +def genesis_file(tmp_path): + sk = SigningKey.generate() + pk = sk.verify_key.encode(encoder=HexEncoder).decode() + data = { + "chain_id": "minichain-default", + "timestamp": 1_700_000_000_000, + "target": hex(MAX_TARGET), + "target_block_time": 1000, + "alloc": {pk: {"balance": 10_000}}, + } + path = tmp_path / "genesis_sync.json" + path.write_text(json.dumps(data), encoding="utf-8") + return str(path), sk, pk + + +def _rehash(header, **changes): + """Return ``header`` with ``changes`` applied and a matching hash.""" + updated = replace(header, hash=None, **changes) + return replace(updated, hash=updated.compute_hash()) + + +def _fake_chain(length): + blocks = [] + for index in range(length): + block = Block( + index=index, + previous_hash="0" * 64 if index == 0 else blocks[-1].hash, + target=MAX_TARGET, + timestamp=1_700_000_000_000 + index, + ) + block.hash = block.compute_hash() + blocks.append(block) + return SimpleNamespace(chain=blocks) + + +def _make_sync_pair(receiver_chain, sender_chain, **client_kwargs): + receiver_outbox = [] + sender_outbox = [] + receiver_sent = [] + + def receiver_send(peer, message): + receiver_outbox.append((peer, message)) + receiver_sent.append(message) + + def sender_send(peer, message): + sender_outbox.append((peer, message)) + + client = SyncManager(receiver_chain, receiver_send, **client_kwargs) + server = SyncManager(sender_chain, sender_send) + return client, server, receiver_outbox, sender_outbox, receiver_sent + + +def _pump(client, server, receiver_outbox, sender_outbox, *, max_steps=500): + """Deliver queued protocol messages until both sides are idle.""" + statuses = [] + for _ in range(max_steps): + if not receiver_outbox and not sender_outbox: + return statuses + if receiver_outbox: + peer, message = receiver_outbox.pop(0) + status = server.handle_message(peer, message["type"], message["data"]) + else: + peer, message = sender_outbox.pop(0) + status = client.handle_message(peer, message["type"], message["data"]) + statuses.append(status) + raise AssertionError("sync did not terminate") + + +class TestBlockHeader: + def test_round_trip_matches_block_hash(self, genesis_file): + path, _, pk = genesis_file + chain = Blockchain(genesis_path=path) + block = mine_and_process_block(chain, Mempool(), pk) + + header = BlockHeader.from_dict(block.to_dict()) + + assert header.hash == block.hash + assert header.to_dict(include_hash=False) == block.to_header_dict() + assert header.compute_hash() == block.hash + + def test_missing_hash_is_computed(self, genesis_file): + path, _, pk = genesis_file + chain = Blockchain(genesis_path=path) + block = mine_and_process_block(chain, Mempool(), pk) + payload = block.to_dict() + expected_hash = payload.pop("hash") + + header = BlockHeader.from_dict(payload) + + assert header.hash == expected_hash + + def test_tampered_header_is_rejected(self, genesis_file): + path, _, pk = genesis_file + chain = Blockchain(genesis_path=path) + block = mine_and_process_block(chain, Mempool(), pk) + payload = block.to_dict() + payload["nonce"] += 1 + + with pytest.raises(ValueError, match="hash"): + BlockHeader.from_dict(payload) + + +class TestLocator: + def test_exponential_spacing_and_genesis_anchor(self): + chain = _fake_chain(100) + manager = SyncManager(chain, lambda *_: None, locator_size=32) + + locator = manager.locator() + + expected = [chain.chain[i].hash for i in (99, 98, 97, 96, 95, 94, 93, 92, 91, 90)] + expected += [chain.chain[i].hash for i in (88, 84, 76, 60, 28, 0)] + assert locator == expected + assert len(locator) <= 32 + + def test_single_block_chain_locator(self): + chain = _fake_chain(1) + manager = SyncManager(chain, lambda *_: None) + + assert manager.locator() == [chain.chain[0].hash] + + +class TestHeaderValidation: + def _headers(self): + chain = _fake_chain(4) + headers = [BlockHeader.from_block(block) for block in chain.chain[1:]] + return chain.chain[0], headers + + def test_valid_chain_passes(self): + start, headers = self._headers() + + validate_header_chain(headers, start, now_ms=headers[-1].timestamp + 1) + + @pytest.mark.parametrize( + "changes,message", + [ + ({"index": 99}, "index"), + ({"previous_hash": "f" * 64}, "previous hash"), + ({"timestamp": 1_700_000_000_000}, "timestamp"), + ], + ) + def test_bad_linkage_is_rejected(self, changes, message): + start, headers = self._headers() + + with pytest.raises(SyncError, match=message): + validate_header_chain([_rehash(headers[0], **changes)], start) + + def test_bad_hash_is_rejected(self): + start, headers = self._headers() + + with pytest.raises(SyncError, match="hash"): + validate_header_chain( + [replace(headers[0], hash="0" * 64)], + start, + ) + + def test_missing_hash_is_rejected(self): + start, headers = self._headers() + + with pytest.raises(SyncError, match="hash"): + validate_header_chain([replace(headers[0], hash=None)], start) + + def test_invalid_proof_of_work_is_rejected(self): + start, headers = self._headers() + impossible = _rehash(headers[0], target=1) + + with pytest.raises(SyncError, match="proof of work"): + validate_header_chain([impossible], start) + + def test_far_future_timestamp_is_rejected(self): + start, headers = self._headers() + now_ms = headers[0].timestamp + future = _rehash( + headers[0], + timestamp=now_ms + MAX_FUTURE_BLOCK_TIME_MS + 1, + ) + + with pytest.raises(SyncError, match="future"): + validate_header_chain([future], start, now_ms=now_ms) + + +class TestPeerRateLimiter: + def test_token_bucket_allows_burst_then_refills(self): + now = [0.0] + limiter = PeerRateLimiter( + rate_per_second=1.0, + burst=2, + clock=lambda: now[0], + ) + + assert limiter.allow("peer:a") + assert limiter.allow("peer:a") + assert not limiter.allow("peer:a") + + assert limiter.allow("peer:b") + + now[0] = 1.0 + assert limiter.allow("peer:a") + + +class TestSyncResponses: + def test_default_header_batch_fits_in_one_frame(self): + chain = _fake_chain(1100) + manager = SyncManager(chain, lambda *_: None) + + response = manager.build_headers_response( + {"locator": [chain.chain[0].hash], "max": 1000} + ) + + assert len(response["data"]["headers"]) == 1000 + assert len(canonical_json_dumps(response).encode()) < MAX_FRAME_BYTES + + def test_header_response_honours_count_and_byte_limits(self): + chain = _fake_chain(10) + header_size = len( + canonical_json_dumps(BlockHeader.from_block(chain.chain[1]).to_dict()) + ) + 1 + manager = SyncManager( + chain, + lambda *_: None, + max_headers=5, + max_response_bytes=512 + header_size * 2 + 10, + ) + + response = manager.build_headers_response( + {"locator": [chain.chain[0].hash], "max": 5} + ) + + assert len(response["data"]["headers"]) == 2 + assert [h["index"] for h in response["data"]["headers"]] == [1, 2] + + def test_block_response_is_an_ordered_contiguous_prefix(self): + chain = _fake_chain(5) + block_size = len(canonical_json_dumps(chain.chain[1].to_dict())) + 1 + manager = SyncManager( + chain, + lambda *_: None, + max_blocks=4, + max_response_bytes=512 + block_size * 3 + 10, + ) + + response = manager.build_blocks_response( + { + "hashes": [ + chain.chain[3].hash, + chain.chain[1].hash, + "f" * 64, + chain.chain[4].hash, + ] + } + ) + + assert [b["hash"] for b in response["data"]["blocks"]] == [ + chain.chain[3].hash, + chain.chain[1].hash, + ] + + def test_request_bounds_are_enforced(self): + chain = _fake_chain(5) + manager = SyncManager(chain, lambda *_: None, max_blocks=2, locator_size=2) + + with pytest.raises(SyncError, match="locator"): + manager.build_headers_response( + {"locator": [chain.chain[i].hash for i in range(3)]} + ) + + with pytest.raises(SyncError, match="hashes"): + manager.build_blocks_response( + {"hashes": [chain.chain[i].hash for i in range(3)]} + ) + + def test_rate_limited_peer_gets_no_response(self): + now = [0.0] + sent = [] + manager = SyncManager( + _fake_chain(5), + lambda peer, message: sent.append(message), + rate_limiter=PeerRateLimiter( + rate_per_second=1.0, + burst=1, + clock=lambda: now[0], + ), + ) + request = {"locator": [], "max": 1} + + assert manager.handle_message("peer:a", "getheaders", request) is None + assert manager.handle_message("peer:a", "getheaders", request) == ( + ValidationStatus.FAILED + ) + assert len(sent) == 1 + + +class TestNetworkHandlerIntegration: + class FakeNetwork: + _peer_id = "self" + data_path = "." + + def __init__(self): + self.sent = [] + + async def _unicast_raw(self, peer, payload): + self.sent.append((peer, payload)) + + def test_legacy_messages_are_bounded_and_ignored(self, tmp_path): + genesis = tmp_path / "genesis_handler.json" + genesis.write_text( + json.dumps( + { + "chain_id": "minichain-default", + "timestamp": 1_700_000_000_000, + "target": hex(MAX_TARGET), + "target_block_time": 1000, + "alloc": {}, + } + ), + encoding="utf-8", + ) + chain = Blockchain(genesis_path=str(genesis)) + + async def scenario(): + network = self.FakeNetwork() + sync = make_sync_manager(chain, Mempool(), network) + handler = make_network_handler(chain, Mempool(), network, sync) + + status = await handler( + { + "type": "chain_request", + "data": {"start_index": 0, "limit": 10_000}, + "_peer_addr": "peer:a", + } + ) + assert status is None + await asyncio.sleep(0) + + assert len(network.sent) == 1 + _, response = network.sent[0] + assert response["type"] == "chain_response" + assert len(response["data"]["blocks"]) == 1 + assert response["data"]["requested_limit"] == 1 + + status = await handler( + { + "type": "chain_response", + "data": {"blocks": [{"invalid": True}]}, + "_peer_addr": "peer:a", + } + ) + assert status is None + await asyncio.sleep(0) + assert len(network.sent) == 1 + + status = await handler( + {"type": "getheaders", "data": None, "_peer_addr": "peer:a"} + ) + assert status == ValidationStatus.MALFORMED + + asyncio.run(scenario()) + + +class TestHeadersFirstSync: + def test_linear_sync_paginates_headers_and_bodies(self, genesis_file): + path, _, pk = genesis_file + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + for _ in range(5): + assert mine_and_process_block(sender, Mempool(), pk) is not None + + applied = [] + client, server, receiver_out, sender_out, receiver_sent = _make_sync_pair( + receiver, + sender, + max_headers=4, + max_blocks=2, + on_block_applied=applied.append, + ) + client.request_sync("peer:test") + + statuses = _pump(client, server, receiver_out, sender_out) + + assert all(status is None for status in statuses) + assert receiver.last_block.hash == sender.last_block.hash + assert [block.index for block in applied] == [1, 2, 3, 4, 5] + assert [m["type"] for m in receiver_sent].count("getheaders") == 2 + assert [ + len(m["data"]["hashes"]) + for m in receiver_sent + if m["type"] == "getblocks" + ] == [2, 2, 1] + + def test_invalid_headers_stop_before_any_body_download(self, genesis_file): + path, _, pk = genesis_file + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + for _ in range(2): + assert mine_and_process_block(sender, Mempool(), pk) is not None + sender.chain[1].hash = "f" * 64 + + client, server, receiver_out, sender_out, receiver_sent = _make_sync_pair( + receiver, + sender, + ) + client.request_sync("peer:test") + + peer, request = receiver_out.pop(0) + assert server.handle_message(peer, request["type"], request["data"]) is None + + peer, response = sender_out.pop(0) + status = client.handle_message(peer, response["type"], response["data"]) + + assert status == ValidationStatus.MALFORMED + assert [m["type"] for m in receiver_sent] == ["getheaders"] + assert receiver.last_block.index == 0 + assert not client._sessions + + def test_fork_sync_reorgs_and_returns_orphans(self, genesis_file): + path, sk, pk = genesis_file + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + + receiver_pool = Mempool() + tx_a = Transaction( + sender=pk, receiver="a" * 64, amount=10, nonce=0, + gas_limit=1, fee_per_gas=1, + ) + tx_a.sign(sk) + receiver_pool.add_transaction(tx_a) + assert mine_and_process_block(receiver, receiver_pool, pk) is not None + + sender_pool = Mempool() + tx_b = Transaction( + sender=pk, receiver="b" * 64, amount=20, nonce=0, + gas_limit=1, fee_per_gas=1, + ) + tx_b.sign(sk) + sender_pool.add_transaction(tx_b) + assert mine_and_process_block(sender, sender_pool, pk) is not None + + tx_c = Transaction( + sender=pk, receiver="c" * 64, amount=30, nonce=1, + gas_limit=1, fee_per_gas=1, + ) + tx_c.sign(sk) + sender_pool.add_transaction(tx_c) + assert mine_and_process_block(sender, sender_pool, pk) is not None + + orphans = [] + client, server, receiver_out, sender_out, _ = _make_sync_pair( + receiver, + sender, + on_orphans=lambda txs: orphans.extend(txs), + ) + client.request_sync("peer:test") + + statuses = _pump(client, server, receiver_out, sender_out) + + assert all(status is None for status in statuses) + assert receiver.get_total_work() > 0 + assert receiver.last_block.hash == sender.last_block.hash + assert receiver.state.accounts == sender.state.accounts + assert [tx.tx_id for tx in orphans] == [tx_a.tx_id] From 9239330f04e07ec78adc39ee28e246e92801e1a4 Mon Sep 17 00:00:00 2001 From: anshulchikhale30-p Date: Sun, 8 Nov 2026 17:25:56 +0530 Subject: [PATCH 2/4] fix: accumulate headers for deep fork sync A fork deeper than one header batch was rejected after the first batch because its partial work was compared with our full suffix. Keep requesting headers from the last received hash until the branch proves more work, then download and apply its bodies. Add an end-to-end test with a fork spanning four header batches. --- docs/sync.md | 10 +++-- minichain/sync.py | 94 ++++++++++++++++++++++++++++++++++------------ tests/test_sync.py | 69 +++++++++++++++++++++++++++------- 3 files changed, 133 insertions(+), 40 deletions(-) diff --git a/docs/sync.md b/docs/sync.md index 55ac026..8381c0d 100644 --- a/docs/sync.md +++ b/docs/sync.md @@ -129,10 +129,12 @@ advertises a chain with no more work than ours is ignored rather than punished. ## Reorgs When the common ancestor is not our tip, headers describe a competing branch. -Bodies for that branch are buffered and handed to `Blockchain.resolve_conflicts`. -If the branch is heavier and valid, the chain switches and transactions that are -no longer confirmed are returned to the mempool. Otherwise the session is -dropped. +If that branch spans more than one header batch, the headers are accumulated +across requests until the branch proves more work than our suffix. Bodies are +only downloaded after that point. They are buffered and handed to +`Blockchain.resolve_conflicts`. If the branch is heavier and valid, the chain +switches and transactions that are no longer confirmed are returned to the +mempool. Otherwise the session is dropped. ## Compatibility diff --git a/minichain/sync.py b/minichain/sync.py index b52a7b3..5a0f1e0 100644 --- a/minichain/sync.py +++ b/minichain/sync.py @@ -46,7 +46,9 @@ class SyncError(ValueError): """A peer sent a sync message that cannot be processed safely.""" - def __init__(self, message: str, status: ValidationStatus = ValidationStatus.MALFORMED): + def __init__( + self, message: str, status: ValidationStatus = ValidationStatus.MALFORMED + ): super().__init__(message) self.status = status @@ -92,6 +94,7 @@ class _PendingRequest: class _SyncSession: peer_addr: str headers: list[BlockHeader] + common_ancestor: Block fork: bool next_index: int = 0 blocks: list[Block] = field(default_factory=list) @@ -159,7 +162,9 @@ def validate_header_chain( ) if header.hash != header.compute_hash(): raise SyncError(f"invalid header hash {header.hash}") - if not isinstance(header.target, int) or not (MIN_TARGET <= header.target <= MAX_TARGET): + if not isinstance(header.target, int) or not ( + MIN_TARGET <= header.target <= MAX_TARGET + ): raise SyncError(f"invalid target {header.target}") try: hash_value = int(header.hash, 16) @@ -277,7 +282,9 @@ def locator(self) -> list[str]: # ── inbound message handling ───────────────────────────────────────────── - def handle_message(self, peer_addr: str, message_type: str, payload) -> ValidationStatus | None: + def handle_message( + self, peer_addr: str, message_type: str, payload + ) -> ValidationStatus | None: """Handle one sync message. Returns a validation status for P2P scoring.""" self._prune_pending() if not isinstance(payload, dict): @@ -428,23 +435,37 @@ def _handle_headers(self, peer_addr: str, payload: dict) -> ValidationStatus | N except (TypeError, ValueError) as exc: raise SyncError(f"invalid header payload: {exc}") from exc - start_block = self._find_block(headers[0].previous_hash) - if start_block is None: - raise SyncError("headers do not connect to a known block") - - validate_header_chain(headers, start_block) + session = self._sessions.get(peer_addr) + if session is not None: + # A fork with more than one header batch keeps requesting headers + # until the branch proves more work than our own suffix. Otherwise a + # deep reorg would be rejected after the first 1,000 headers. + if not session.fork or session.next_index != 0: + raise SyncError("unexpected headers response during body download") + previous = session.headers[-1] + if headers[0].previous_hash != previous.hash: + raise SyncError("headers do not continue the advertised fork") + validate_header_chain(headers, previous) + session.headers.extend(headers) + else: + start_block = self._find_block(headers[0].previous_hash) + if start_block is None: + raise SyncError("headers do not connect to a known block") + validate_header_chain(headers, start_block) + session = _SyncSession( + peer_addr=peer_addr, + headers=headers, + common_ancestor=start_block, + fork=start_block.hash != self.chain.last_block.hash, + ) + self._sessions[peer_addr] = session - if _total_work(headers) <= self._suffix_work(start_block): - logger.info("Peer %s advertised a chain with no more work; not syncing", peer_addr) - self._drop_session(peer_addr) + if session.fork and _total_work(session.headers) <= self._suffix_work( + session.common_ancestor + ): + self._request_next_headers(session) return None - session = _SyncSession( - peer_addr=peer_addr, - headers=headers, - fork=start_block.hash != self.chain.last_block.hash, - ) - self._sessions[peer_addr] = session self._request_next_blocks(session) return None @@ -463,7 +484,9 @@ def _handle_blocks(self, peer_addr: str, payload: dict) -> ValidationStatus | No if len(raw_blocks) > self.max_blocks: raise SyncError(f"too many blocks: {len(raw_blocks)}") - expected_headers = session.headers[session.next_index : session.next_index + self.max_blocks] + expected_headers = session.headers[ + session.next_index : session.next_index + self.max_blocks + ] if not raw_blocks: raise SyncError("empty blocks response") if len(raw_blocks) > len(expected_headers): @@ -510,8 +533,30 @@ def _handle_blocks(self, peer_addr: str, payload: dict) -> ValidationStatus | No # ── session helpers ────────────────────────────────────────────────────── + def _request_next_headers(self, session: _SyncSession) -> None: + """Continue a fork's header range from its last known hash.""" + request_id = self._new_request_id() + self._pending[request_id] = _PendingRequest( + peer_addr=session.peer_addr, + kind="headers", + created_at=self._clock(), + ) + self._send( + session.peer_addr, + { + "type": "getheaders", + "data": { + "request_id": request_id, + "locator": [session.headers[-1].hash], + "max": self.max_headers, + }, + }, + ) + def _request_next_blocks(self, session: _SyncSession) -> None: - batch = session.headers[session.next_index : session.next_index + self.max_blocks] + batch = session.headers[ + session.next_index : session.next_index + self.max_blocks + ] if not batch: self._finish_session(session) return @@ -535,9 +580,10 @@ def _request_next_blocks(self, session: _SyncSession) -> None: def _finish_session(self, session: _SyncSession) -> None: self._sessions.pop(session.peer_addr, None) - # A full header batch may be truncated; ask for the next one from our - # new tip. A short batch means the peer has caught us up. - if len(session.headers) == self.max_headers: + # A fork may have been adopted after only the prefix needed to prove more + # work, so always ask for what follows the new tip. A linear sync only + # needs another request when the peer filled a whole header batch. + if session.fork or len(session.headers) == self.max_headers: self.request_sync(session.peer_addr) def _drop_session(self, peer_addr: str) -> None: @@ -546,7 +592,9 @@ def _drop_session(self, peer_addr: str) -> None: if pending.peer_addr == peer_addr: self._pending.pop(request_id, None) - def _take_pending(self, peer_addr: str, kind: str, request_id) -> _PendingRequest | None: + def _take_pending( + self, peer_addr: str, kind: str, request_id + ) -> _PendingRequest | None: if not isinstance(request_id, str): return None pending = self._pending.get(request_id) diff --git a/tests/test_sync.py b/tests/test_sync.py index c255aab..dd421c2 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -139,7 +139,9 @@ def test_exponential_spacing_and_genesis_anchor(self): locator = manager.locator() - expected = [chain.chain[i].hash for i in (99, 98, 97, 96, 95, 94, 93, 92, 91, 90)] + expected = [ + chain.chain[i].hash for i in (99, 98, 97, 96, 95, 94, 93, 92, 91, 90) + ] expected += [chain.chain[i].hash for i in (88, 84, 76, 60, 28, 0)] assert locator == expected assert len(locator) <= 32 @@ -243,9 +245,10 @@ def test_default_header_batch_fits_in_one_frame(self): def test_header_response_honours_count_and_byte_limits(self): chain = _fake_chain(10) - header_size = len( - canonical_json_dumps(BlockHeader.from_block(chain.chain[1]).to_dict()) - ) + 1 + header_size = ( + len(canonical_json_dumps(BlockHeader.from_block(chain.chain[1]).to_dict())) + + 1 + ) manager = SyncManager( chain, lambda *_: None, @@ -413,9 +416,7 @@ def test_linear_sync_paginates_headers_and_bodies(self, genesis_file): assert [block.index for block in applied] == [1, 2, 3, 4, 5] assert [m["type"] for m in receiver_sent].count("getheaders") == 2 assert [ - len(m["data"]["hashes"]) - for m in receiver_sent - if m["type"] == "getblocks" + len(m["data"]["hashes"]) for m in receiver_sent if m["type"] == "getblocks" ] == [2, 2, 1] def test_invalid_headers_stop_before_any_body_download(self, genesis_file): @@ -450,8 +451,12 @@ def test_fork_sync_reorgs_and_returns_orphans(self, genesis_file): receiver_pool = Mempool() tx_a = Transaction( - sender=pk, receiver="a" * 64, amount=10, nonce=0, - gas_limit=1, fee_per_gas=1, + sender=pk, + receiver="a" * 64, + amount=10, + nonce=0, + gas_limit=1, + fee_per_gas=1, ) tx_a.sign(sk) receiver_pool.add_transaction(tx_a) @@ -459,16 +464,24 @@ def test_fork_sync_reorgs_and_returns_orphans(self, genesis_file): sender_pool = Mempool() tx_b = Transaction( - sender=pk, receiver="b" * 64, amount=20, nonce=0, - gas_limit=1, fee_per_gas=1, + sender=pk, + receiver="b" * 64, + amount=20, + nonce=0, + gas_limit=1, + fee_per_gas=1, ) tx_b.sign(sk) sender_pool.add_transaction(tx_b) assert mine_and_process_block(sender, sender_pool, pk) is not None tx_c = Transaction( - sender=pk, receiver="c" * 64, amount=30, nonce=1, - gas_limit=1, fee_per_gas=1, + sender=pk, + receiver="c" * 64, + amount=30, + nonce=1, + gas_limit=1, + fee_per_gas=1, ) tx_c.sign(sk) sender_pool.add_transaction(tx_c) @@ -489,3 +502,33 @@ def test_fork_sync_reorgs_and_returns_orphans(self, genesis_file): assert receiver.last_block.hash == sender.last_block.hash assert receiver.state.accounts == sender.state.accounts assert [tx.tx_id for tx in orphans] == [tx_a.tx_id] + + def test_deep_fork_accumulates_headers_across_batches(self, genesis_file): + path, _, pk = genesis_file + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + for _ in range(6): + assert mine_and_process_block(receiver, Mempool(), pk) is not None + for _ in range(8): + assert mine_and_process_block(sender, Mempool(), pk) is not None + + client, server, receiver_out, sender_out, receiver_sent = _make_sync_pair( + receiver, + sender, + max_headers=2, + max_blocks=3, + ) + client.request_sync("peer:test") + + statuses = _pump(client, server, receiver_out, sender_out) + + assert all(status is None for status in statuses) + assert receiver.last_block.hash == sender.last_block.hash + assert receiver.get_total_work() == sender.get_total_work() + assert receiver.state.accounts == sender.state.accounts + assert [m["type"] for m in receiver_sent].count("getheaders") == 5 + assert [ + len(m["data"]["hashes"]) for m in receiver_sent if m["type"] == "getblocks" + ] == [3, 3, 2] + assert not client._sessions + assert not client._pending From ec353ec757810f8a9475a5df6f60f5d87b1e3305 Mon Sep 17 00:00:00 2001 From: anshulchikhale30-p Date: Mon, 9 Nov 2026 09:07:13 +0530 Subject: [PATCH 3/4] fix: address sync review feedback - pace outbound sync requests and retry unanswered ones through maintenance - bound rate-limiter state and apply the limiter to legacy chain requests - cap retained fork headers and track fork work incrementally - include request IDs in response budgets and bound their size - reuse the chain hash index and rebuild it after reorgs - restart sync when the local tip changes mid-download - continue linear sync after short header batches - add review-driven test coverage --- docs/sync.md | 30 +- main.py | 19 +- minichain/network_config.py | 3 +- minichain/node_config.py | 17 + minichain/sync.py | 443 +++++++++++++++++++++---- tests/test_sync.py | 635 +++++++++++++++++++++++++++++++++++- 6 files changed, 1051 insertions(+), 96 deletions(-) diff --git a/docs/sync.md b/docs/sync.md index 8381c0d..e4d6b98 100644 --- a/docs/sync.md +++ b/docs/sync.md @@ -120,11 +120,24 @@ Defaults live in [`minichain/node_config.py`](../minichain/node_config.py): | `MAX_LOCATOR_HASHES` | 32 | Locator cap | | `MAX_SYNC_RESPONSE_BYTES` | 768 KiB | Response byte budget, below the 1 MiB frame limit | | `SYNC_REQUEST_TIMEOUT_SECONDS` | 30 | Drop a stalled sync session | -| `SYNC_REQUESTS_PER_SECOND` / `SYNC_REQUEST_BURST` | 5 / 10 | Per-peer token bucket for serving sync requests | - -Out-of-range requests and malformed sync payloads are returned to the P2P layer -as validation statuses, so existing peer scoring and banning apply. A peer that -advertises a chain with no more work than ours is ignored rather than punished. +| `SYNC_REQUESTS_PER_SECOND` / `SYNC_REQUEST_BURST` | 5 / 10 | Per-peer token bucket for serving requests and pacing outbound requests | +| `SYNC_REQUEST_RETRY_SECONDS` / `SYNC_REQUEST_MAX_RETRIES` | 5 / 2 | Retry an unanswered request before dropping the session | +| `MAX_FORK_HEADERS` | 16,000 | Cap on headers retained while following a fork | + +A request's `max` field and a legacy `chain_request.limit` are clamped to their +configured ranges, so an out-of-range value alone is not a validation failure. +Oversized locator or hash lists and malformed sync payloads are returned to the +P2P layer as validation statuses, so existing peer scoring and banning apply. A +peer that advertises a chain with no more work than ours is ignored rather than +punished. + +Requests are answered in bounded batches. A response may be shorter than the +requested count because of the byte budget, and the requester continues from the +last received item; an empty header batch ends the sync. Outbound requests are +paced against the same per-peer budget, and a request that would exceed it is +queued briefly instead of being dropped. Unanswered requests are retried before +a stalled session is dropped. A single valid block may exceed the soft response +budget when its complete frame still fits within `MAX_FRAME_BYTES`. ## Reorgs @@ -140,9 +153,10 @@ mempool. Otherwise the session is dropped. `chain_request` and `chain_response` are still accepted for older peers: -- A new node never sends them. -- A legacy `chain_request` is answered with at most `MAX_BLOCKS_PER_REQUEST` - blocks, so an old peer cannot ask for an unbounded response. +- A new node does not initiate legacy sync. +- When a new node receives a legacy `chain_request`, it answers with a + `chain_response` containing at most `MAX_BLOCKS_PER_REQUEST` blocks, so an old + peer cannot ask for an unbounded response. - A legacy `chain_response` is ignored because this node no longer initiates that request. diff --git a/main.py b/main.py index a521dc5..c7f7306 100644 --- a/main.py +++ b/main.py @@ -32,6 +32,7 @@ from nacl.encoding import HexEncoder from minichain import Transaction, Blockchain, Block, State, Mempool, P2PNetwork, mine_block +from minichain.node_config import SYNC_MAINTENANCE_INTERVAL_SECONDS from minichain.rpc import JSONRPCServer from minichain.sync import SyncManager from minichain.validators import is_valid_receiver, ValidationStatus @@ -289,12 +290,15 @@ async def handler(data): elif msg_type == "chain_request": # Deprecated compatibility path. New nodes use getheaders/getblocks; - # this path is still bounded so an old peer cannot request an - # arbitrarily large response. + # this path is still bounded and rate-limited so an old peer cannot + # request an arbitrarily large response or flood the node. + if not sync.allow_request(peer_addr): + logger.warning("Rate-limited legacy chain_request from %s", peer_addr) + return None try: start_index = max(0, int(payload.get("start_index", 0))) limit = max(1, min(int(payload.get("limit", sync.max_blocks)), sync.max_blocks)) - except (TypeError, ValueError): + except (TypeError, ValueError, OverflowError): return ValidationStatus.MALFORMED logger.info("📡 Peer requested blocks from %d (limit %d).", start_index, limit) resp_payload = sync.build_legacy_blocks_response(start_index, limit) @@ -642,6 +646,15 @@ async def run_node(port: int, host: str, connect_to: str | None, fund: int, data sync = make_sync_manager(chain, mempool, network) handler = make_network_handler(chain, mempool, network, sync) network.register_handler(handler) + + async def maintain_sync(): + """Recover unanswered requests and expire idle limiter buckets.""" + while True: + await asyncio.sleep(SYNC_MAINTENANCE_INTERVAL_SECONDS) + sync.maintain() + + # The event loop keeps the task alive while it sleeps between ticks. + asyncio.create_task(maintain_sync()) rpc_server = JSONRPCServer(chain, mempool, network) diff --git a/minichain/network_config.py b/minichain/network_config.py index 427f1e6..e12d343 100644 --- a/minichain/network_config.py +++ b/minichain/network_config.py @@ -14,7 +14,8 @@ "getblocks", "blocks", # Retained for backward compatibility with nodes that predate headers-first - # sync. New nodes never send these messages. + # sync. New nodes do not initiate these messages, but still answer a legacy + # chain_request with a chain_response. "chain_request", "chain_response", "getaddr", diff --git a/minichain/node_config.py b/minichain/node_config.py index fe7e7fc..55876ae 100644 --- a/minichain/node_config.py +++ b/minichain/node_config.py @@ -41,3 +41,20 @@ SYNC_REQUEST_TIMEOUT_SECONDS = 30.0 SYNC_REQUESTS_PER_SECOND = 5.0 SYNC_REQUEST_BURST = 10 +# Hard cap on peers tracked by the per-peer sync token bucket. Fully refilled +# buckets are dropped first; if every retained bucket is still active, new +# peers are refused rather than resetting an active rate limit. +SYNC_RATE_LIMIT_MAX_PEERS = 1024 +# A fork deeper than one header batch keeps accumulating headers until it +# proves more work than our suffix. Cap that retention so a peer cannot make +# us hold an unbounded branch in memory. +MAX_FORK_HEADERS = MAX_HEADERS_PER_REQUEST * 16 +# Retry an unanswered sync request before dropping the session. This recovers +# from a receiver that rate-limited (and therefore ignored) one request. +SYNC_REQUEST_RETRY_SECONDS = 5.0 +SYNC_REQUEST_MAX_RETRIES = 2 +# Request IDs are opaque and echoed back; bound them so a response cannot be +# inflated by a large ID that was not part of the payload budget. +MAX_SYNC_REQUEST_ID_BYTES = 128 +# How often the node checks for stalled requests and expires idle limiter state. +SYNC_MAINTENANCE_INTERVAL_SECONDS = 1.0 diff --git a/minichain/sync.py b/minichain/sync.py index 5a0f1e0..54022c7 100644 --- a/minichain/sync.py +++ b/minichain/sync.py @@ -25,13 +25,23 @@ from dataclasses import dataclass, field from .block import Block, BlockHeader -from .network_config import MAX_FUTURE_BLOCK_TIME_MS, MAX_TARGET, MIN_TARGET +from .network_config import ( + MAX_FRAME_BYTES, + MAX_FUTURE_BLOCK_TIME_MS, + MAX_TARGET, + MIN_TARGET, +) from .node_config import ( MAX_BLOCKS_PER_REQUEST, + MAX_FORK_HEADERS, MAX_HEADERS_PER_REQUEST, MAX_LOCATOR_HASHES, + MAX_SYNC_REQUEST_ID_BYTES, MAX_SYNC_RESPONSE_BYTES, + SYNC_RATE_LIMIT_MAX_PEERS, SYNC_REQUEST_BURST, + SYNC_REQUEST_MAX_RETRIES, + SYNC_REQUEST_RETRY_SECONDS, SYNC_REQUEST_TIMEOUT_SECONDS, SYNC_REQUESTS_PER_SECOND, ) @@ -61,18 +71,30 @@ def __init__( rate_per_second: float = SYNC_REQUESTS_PER_SECOND, burst: int = SYNC_REQUEST_BURST, clock: Callable[[], float] = time.monotonic, + max_peers: int = SYNC_RATE_LIMIT_MAX_PEERS, ): if rate_per_second <= 0: raise ValueError("rate_per_second must be positive") if burst <= 0: raise ValueError("burst must be positive") + if max_peers <= 0: + raise ValueError("max_peers must be positive") self.rate_per_second = float(rate_per_second) self.burst = float(burst) + self.max_peers = max_peers self._clock = clock self._buckets: dict[str, tuple[float, float]] = {} def allow(self, peer_addr: str) -> bool: now = self._clock() + if peer_addr not in self._buckets and len(self._buckets) >= self.max_peers: + # Buckets that have refilled to the burst are indistinguishable + # from absent ones, so dropping them cannot change any active + # limit. If every retained bucket is still in use, refuse the new + # peer rather than evicting one and resetting a live rate limit. + self._prune(now) + if len(self._buckets) >= self.max_peers: + return False tokens, last_seen = self._buckets.get(peer_addr, (self.burst, now)) elapsed = max(0.0, now - last_seen) tokens = min(self.burst, tokens + elapsed * self.rate_per_second) @@ -82,12 +104,45 @@ def allow(self, peer_addr: str) -> bool: self._buckets[peer_addr] = (tokens - 1.0, now) return True + def prune(self) -> None: + """Drop buckets that have refilled completely, bounding retained state.""" + self._prune(self._clock()) + + def retry_after(self, peer_addr: str) -> float: + """Seconds until this peer has a token, or zero if one is available.""" + now = self._clock() + tokens, last_seen = self._buckets.get(peer_addr, (self.burst, now)) + tokens = min( + self.burst, + tokens + max(0.0, now - last_seen) * self.rate_per_second, + ) + if tokens >= 1.0: + return 0.0 + return (1.0 - tokens) / self.rate_per_second + + def _prune(self, now: float) -> None: + refilled = [ + peer + for peer, (tokens, last_seen) in self._buckets.items() + if tokens + (now - last_seen) * self.rate_per_second >= self.burst + ] + for peer in refilled: + self._buckets.pop(peer, None) + @dataclass class _PendingRequest: peer_addr: str kind: str created_at: float + expires_at: float = 0.0 + attempts: int = 1 + + +@dataclass +class _DeferredRequest: + message: dict + not_before: float @dataclass @@ -96,10 +151,17 @@ class _SyncSession: headers: list[BlockHeader] common_ancestor: Block fork: bool + work: int = 0 + local_tip: str | None = None next_index: int = 0 blocks: list[Block] = field(default_factory=list) +def _encoded_size(payload) -> int: + """Serialized size in bytes, matching the P2P layer's UTF-8 framing.""" + return len(canonical_json_dumps(payload).encode("utf-8")) + + def _bounded_int(value, *, default: int, minimum: int, maximum: int) -> int: if value is None: return default @@ -197,8 +259,13 @@ def __init__( max_blocks: int = MAX_BLOCKS_PER_REQUEST, max_response_bytes: int = MAX_SYNC_RESPONSE_BYTES, locator_size: int = MAX_LOCATOR_HASHES, + max_fork_headers: int = MAX_FORK_HEADERS, + max_request_id_bytes: int = MAX_SYNC_REQUEST_ID_BYTES, request_timeout: float = SYNC_REQUEST_TIMEOUT_SECONDS, + retry_seconds: float = SYNC_REQUEST_RETRY_SECONDS, + max_retries: int = SYNC_REQUEST_MAX_RETRIES, rate_limiter: PeerRateLimiter | None = None, + outgoing_rate_limiter: PeerRateLimiter | None = None, clock: Callable[[], float] = time.monotonic, request_id_factory: Callable[[], str] = lambda: uuid.uuid4().hex, ): @@ -208,6 +275,12 @@ def __init__( raise ValueError("max_response_bytes must be positive") if locator_size <= 0: raise ValueError("locator_size must be positive") + if max_fork_headers <= 0: + raise ValueError("max_fork_headers must be positive") + if max_request_id_bytes <= 0: + raise ValueError("max_request_id_bytes must be positive") + if request_timeout <= 0 or retry_seconds <= 0 or max_retries < 0: + raise ValueError("invalid sync retry policy") self.chain = chain self._send = send self._on_block_applied = on_block_applied @@ -216,12 +289,28 @@ def __init__( self.max_blocks = max_blocks self.max_response_bytes = max_response_bytes self.locator_size = locator_size + self.max_fork_headers = max_fork_headers + self.max_request_id_bytes = max_request_id_bytes self.request_timeout = request_timeout + self.retry_seconds = retry_seconds + self.max_retries = max_retries self._rate_limiter = rate_limiter or PeerRateLimiter(clock=clock) + # Request pacing is separate from the inbound limiter: one bounds what + # we serve, the other keeps our requests inside the peer's budget. + self._outgoing_rate_limiter = outgoing_rate_limiter or PeerRateLimiter( + clock=clock + ) self._clock = clock self._request_id_factory = request_id_factory self._pending: dict[str, _PendingRequest] = {} + self._deferred: dict[str, _DeferredRequest] = {} self._sessions: dict[str, _SyncSession] = {} + # Cached hash -> block/position maps for the current main chain. They + # are extended as blocks are appended and rebuilt after a reorg. + self._hash_index: dict[str, Block] = {} + self._hash_positions: dict[str, int] = {} + self._indexed_length = 0 + self._indexed_tip: str | None = None @staticmethod def handles(message_type: str) -> bool: @@ -231,32 +320,90 @@ def handles(message_type: str) -> bool: def request_sync(self, peer_addr: str) -> None: """Ask a peer for headers that follow our current best block.""" + # Expire stale requests first: otherwise a request that timed out (for + # example because the peer rate-limited it) blocks every retry until + # some unrelated message happens to arrive. + self._prune_pending() if peer_addr in self._sessions: return - if any( - pending.peer_addr == peer_addr and pending.kind == "headers" - for pending in self._pending.values() - ): + if self._has_pending(peer_addr, "headers"): return + self._send_headers_request(peer_addr) + + def allow_request(self, peer_addr: str) -> bool: + """Apply the per-peer sync request budget to any sync-style request.""" + return self._rate_limiter.allow(peer_addr) + def maintain(self) -> None: + """Bound limiter state and recover requests that were never answered.""" + self._rate_limiter.prune() + self._outgoing_rate_limiter.prune() + self._prune_pending() + self._flush_deferred() + + def _has_pending(self, peer_addr: str, kind: str) -> bool: + return any( + pending.peer_addr == peer_addr and pending.kind == kind + for pending in self._pending.values() + ) + + def _send_headers_request( + self, + peer_addr: str, + *, + locator: Sequence[str] | None = None, + attempts: int = 1, + expires_at: float | None = None, + ) -> None: + now = self._clock() request_id = self._new_request_id() self._pending[request_id] = _PendingRequest( peer_addr=peer_addr, kind="headers", - created_at=self._clock(), + created_at=now, + expires_at=( + now + self.request_timeout if expires_at is None else expires_at + ), + attempts=attempts, ) - self._send( + self._dispatch_sync( peer_addr, { "type": "getheaders", "data": { "request_id": request_id, - "locator": self.locator(), + "locator": list(locator) if locator is not None else self.locator(), "max": self.max_headers, }, }, ) + def _dispatch_sync(self, peer_addr: str, message: dict) -> None: + """Send a sync request, or queue it until the peer has a token.""" + if self._outgoing_rate_limiter.allow(peer_addr): + self._deferred.pop(peer_addr, None) + self._send(peer_addr, message) + return + delay = self._outgoing_rate_limiter.retry_after(peer_addr) + self._deferred[peer_addr] = _DeferredRequest( + message=message, + not_before=self._clock() + delay, + ) + logger.debug("Deferring %s to %s for %.3fs", message["type"], peer_addr, delay) + + def _flush_deferred(self) -> None: + now = self._clock() + for peer_addr, deferred in list(self._deferred.items()): + if now < deferred.not_before: + continue + if not self._outgoing_rate_limiter.allow(peer_addr): + deferred.not_before = now + self._outgoing_rate_limiter.retry_after( + peer_addr + ) + continue + self._deferred.pop(peer_addr, None) + self._send(peer_addr, deferred.message) + def locator(self) -> list[str]: """Return recent block hashes, newest first, with exponential spacing.""" chain = self.chain.chain @@ -291,9 +438,13 @@ def handle_message( return ValidationStatus.MALFORMED if message_type in ("getheaders", "getblocks"): - if not self._rate_limiter.allow(peer_addr): + if not self.allow_request(peer_addr): + # A request can be dropped during a fast, honest bulk sync + # (for example after the burst is exhausted), so do not score + # the requester as failed. The client retries unanswered + # requests through maintain(). logger.warning("Rate-limited sync request from %s", peer_addr) - return ValidationStatus.FAILED + return None try: response = ( self.build_headers_response(payload) @@ -339,26 +490,26 @@ def build_headers_response(self, payload: dict) -> dict: stop_hash = payload.get("stop_hash") if stop_hash is not None and not isinstance(stop_hash, str): raise SyncError("stop_hash must be a string or null") + request_id = self._validated_request_id(payload) start_block = self.find_common_ancestor(locator) start_index = self._index_of(start_block) if start_block is not None else -1 - budget = max(1, self.max_response_bytes - 512) + budget = self._payload_budget("headers", "headers", request_id) used = 0 headers: list[dict] = [] for block in self.chain.chain[start_index + 1 : start_index + 1 + max_headers]: if stop_hash is not None and block.hash == stop_hash: break header = BlockHeader.from_block(block).to_dict() - size = len(canonical_json_dumps(header)) + 1 + size = _encoded_size(header) + 1 if used + size > budget: break headers.append(header) used += size - response = {"type": "headers", "data": {"headers": headers}} - if isinstance(payload.get("request_id"), str): - response["data"]["request_id"] = payload["request_id"] - return response + return self._with_request_id( + {"type": "headers", "data": {"headers": headers}}, request_id + ) def build_blocks_response(self, payload: dict) -> dict: hashes = _validate_hashes( @@ -366,7 +517,8 @@ def build_blocks_response(self, payload: dict) -> dict: maximum=self.max_blocks, field_name="hashes", ) - by_hash = {block.hash: block for block in self.chain.chain} + request_id = self._validated_request_id(payload) + by_hash = self._chain_hash_index()[0] selected = [] for block_hash in hashes: block = by_hash.get(block_hash) @@ -376,18 +528,26 @@ def build_blocks_response(self, payload: dict) -> dict: break selected.append(block) - response = { - "type": "blocks", - "data": {"blocks": self._bounded_block_payloads(selected)}, - } - if isinstance(payload.get("request_id"), str): - response["data"]["request_id"] = payload["request_id"] - return response + return self._with_request_id( + { + "type": "blocks", + "data": { + "blocks": self._bounded_block_payloads( + selected, response_type="blocks", request_id=request_id + ) + }, + }, + request_id, + ) def build_legacy_blocks_response(self, start_index: int, limit: int) -> dict: """Build a bounded ``chain_response`` for pre-headers-first peers.""" blocks = self.chain.chain[start_index : start_index + limit] - payloads = self._bounded_block_payloads(blocks) + payloads = self._bounded_block_payloads( + blocks, + response_type="chain_response", + extra={"requested_limit": 0}, + ) return { "type": "chain_response", "data": { @@ -398,19 +558,91 @@ def build_legacy_blocks_response(self, start_index: int, limit: int) -> dict: }, } - def _bounded_block_payloads(self, blocks: Sequence[Block]) -> list[dict]: - budget = max(1, self.max_response_bytes - 512) + def _validated_request_id(self, payload: dict) -> str | None: + request_id = payload.get("request_id") + if request_id is None: + return None + if not isinstance(request_id, str) or not request_id: + raise SyncError("request_id must be a non-empty string") + if len(request_id.encode("utf-8")) > self.max_request_id_bytes: + raise SyncError( + f"request_id exceeds the maximum of {self.max_request_id_bytes} bytes" + ) + return request_id + + def _response_base( + self, + response_type: str, + collection_key: str, + request_id: str | None, + extra: dict | None = None, + ) -> int: + base = {"type": response_type, "data": {collection_key: []}} + if request_id is not None: + base["data"]["request_id"] = request_id + if extra: + base["data"].update(extra) + return _encoded_size(base) + + def _payload_budget( + self, + response_type: str, + collection_key: str, + request_id: str | None, + extra: dict | None = None, + ) -> int: + envelope = self._response_base(response_type, collection_key, request_id, extra) + return max(0, self.max_response_bytes - envelope) + + @staticmethod + def _with_request_id(response: dict, request_id: str | None) -> dict: + if request_id is not None: + response["data"]["request_id"] = request_id + return response + + def _bounded_block_payloads( + self, + blocks: Sequence[Block], + *, + response_type: str = "blocks", + request_id: str | None = None, + extra: dict | None = None, + ) -> list[dict]: + budget = self._payload_budget(response_type, "blocks", request_id, extra=extra) used = 0 payloads: list[dict] = [] for block in blocks: payload = block.to_dict() - size = len(canonical_json_dumps(payload)) + 1 + size = _encoded_size(payload) + 1 if used + size > budget: + # A valid block can be larger than the soft response budget + # yet still fit in one framed message. Keep it rather than + # stalling sync on a block that fits the network limit. The + # encoded overhead is only needed for this single payload. + if not payloads and self._fits_frame( + response_type, payload, request_id, extra + ): + payloads.append(payload) break payloads.append(payload) used += size return payloads + def _fits_frame( + self, + response_type: str, + payload: dict, + request_id: str | None, + extra: dict | None, + ) -> bool: + response = {"type": response_type, "data": {"blocks": [payload]}} + if request_id is not None: + response["data"]["request_id"] = request_id + if extra: + response["data"].update(extra) + # +1 for the newline that frames the message on the wire. + return _encoded_size(response) + 1 <= MAX_FRAME_BYTES + # ── inbound responses ──────────────────────────────────────────────────── def _handle_headers(self, peer_addr: str, payload: dict) -> ValidationStatus | None: @@ -432,9 +664,10 @@ def _handle_headers(self, peer_addr: str, payload: dict) -> ValidationStatus | N try: headers = [BlockHeader.from_dict(raw) for raw in raw_headers] - except (TypeError, ValueError) as exc: + except (TypeError, ValueError, OverflowError) as exc: raise SyncError(f"invalid header payload: {exc}") from exc + batch_work = _total_work(headers) session = self._sessions.get(peer_addr) if session is not None: # A fork with more than one header batch keeps requesting headers @@ -446,7 +679,12 @@ def _handle_headers(self, peer_addr: str, payload: dict) -> ValidationStatus | N if headers[0].previous_hash != previous.hash: raise SyncError("headers do not continue the advertised fork") validate_header_chain(headers, previous) + if len(session.headers) + len(headers) > self.max_fork_headers: + raise SyncError( + f"fork headers exceed the maximum of {self.max_fork_headers}" + ) session.headers.extend(headers) + session.work += batch_work else: start_block = self._find_block(headers[0].previous_hash) if start_block is None: @@ -456,13 +694,12 @@ def _handle_headers(self, peer_addr: str, payload: dict) -> ValidationStatus | N peer_addr=peer_addr, headers=headers, common_ancestor=start_block, - fork=start_block.hash != self.chain.last_block.hash, + fork=start_block.hash != self._tip_hash(), + work=batch_work, ) self._sessions[peer_addr] = session - if session.fork and _total_work(session.headers) <= self._suffix_work( - session.common_ancestor - ): + if session.fork and session.work <= self._suffix_work(session.common_ancestor): self._request_next_headers(session) return None @@ -496,13 +733,22 @@ def _handle_blocks(self, peer_addr: str, payload: dict) -> ValidationStatus | No try: blocks = [Block.from_dict(raw) for raw in raw_blocks] - except (KeyError, TypeError, ValueError) as exc: + except (KeyError, TypeError, ValueError, AttributeError, OverflowError) as exc: raise SyncError(f"invalid block payload: {exc}") from exc for block, header in zip(blocks, expected_headers): if block.hash != header.hash: raise SyncError(f"received block {block.hash}, expected {header.hash}") + if not session.fork and session.local_tip != self._tip_hash(): + # A broadcast advanced our chain while this body request was in + # flight, so the response is validated against an old tip. Start + # over from the current chain instead of punishing an honest peer. + logger.info("Local tip changed during sync with %s; restarting", peer_addr) + self._drop_session(peer_addr) + self.request_sync(peer_addr) + return None + if session.fork: session.blocks.extend(blocks) else: @@ -533,27 +779,28 @@ def _handle_blocks(self, peer_addr: str, payload: dict) -> ValidationStatus | No # ── session helpers ────────────────────────────────────────────────────── - def _request_next_headers(self, session: _SyncSession) -> None: + def _request_next_headers( + self, + session: _SyncSession, + *, + attempts: int = 1, + expires_at: float | None = None, + ) -> None: """Continue a fork's header range from its last known hash.""" - request_id = self._new_request_id() - self._pending[request_id] = _PendingRequest( - peer_addr=session.peer_addr, - kind="headers", - created_at=self._clock(), - ) - self._send( + self._send_headers_request( session.peer_addr, - { - "type": "getheaders", - "data": { - "request_id": request_id, - "locator": [session.headers[-1].hash], - "max": self.max_headers, - }, - }, + locator=[session.headers[-1].hash], + attempts=attempts, + expires_at=expires_at, ) - def _request_next_blocks(self, session: _SyncSession) -> None: + def _request_next_blocks( + self, + session: _SyncSession, + *, + attempts: int = 1, + expires_at: float | None = None, + ) -> None: batch = session.headers[ session.next_index : session.next_index + self.max_blocks ] @@ -561,13 +808,19 @@ def _request_next_blocks(self, session: _SyncSession) -> None: self._finish_session(session) return + now = self._clock() + session.local_tip = self._tip_hash() request_id = self._new_request_id() self._pending[request_id] = _PendingRequest( peer_addr=session.peer_addr, kind="blocks", - created_at=self._clock(), + created_at=now, + expires_at=( + now + self.request_timeout if expires_at is None else expires_at + ), + attempts=attempts, ) - self._send( + self._dispatch_sync( session.peer_addr, { "type": "getblocks", @@ -580,14 +833,16 @@ def _request_next_blocks(self, session: _SyncSession) -> None: def _finish_session(self, session: _SyncSession) -> None: self._sessions.pop(session.peer_addr, None) - # A fork may have been adopted after only the prefix needed to prove more - # work, so always ask for what follows the new tip. A linear sync only - # needs another request when the peer filled a whole header batch. - if session.fork or len(session.headers) == self.max_headers: + # After every non-empty header batch, ask for what follows the (possibly + # newly adopted) tip. The peer returns an empty batch when we are caught + # up, which stops the loop. This also covers batches shortened by the + # peer's response byte budget rather than its header count. + if session.headers: self.request_sync(session.peer_addr) def _drop_session(self, peer_addr: str) -> None: self._sessions.pop(peer_addr, None) + self._deferred.pop(peer_addr, None) for request_id, pending in list(self._pending.items()): if pending.peer_addr == peer_addr: self._pending.pop(request_id, None) @@ -605,33 +860,87 @@ def _take_pending( def _prune_pending(self) -> None: now = self._clock() for request_id, pending in list(self._pending.items()): - if now - pending.created_at > self.request_timeout: + if now - pending.created_at <= self.retry_seconds: + continue + if pending.attempts <= self.max_retries: + self._pending.pop(request_id, None) + self._retry_request(pending) + elif now >= pending.expires_at: self._pending.pop(request_id, None) self._drop_session(pending.peer_addr) + def _retry_request(self, pending: _PendingRequest) -> None: + session = self._sessions.get(pending.peer_addr) + attempts = pending.attempts + 1 + if pending.kind == "headers": + if session is None: + self._send_headers_request( + pending.peer_addr, + attempts=attempts, + expires_at=pending.expires_at, + ) + elif session.fork and session.next_index == 0: + self._request_next_headers( + session, + attempts=attempts, + expires_at=pending.expires_at, + ) + else: + self._drop_session(pending.peer_addr) + elif pending.kind == "blocks" and session is not None: + self._request_next_blocks( + session, + attempts=attempts, + expires_at=pending.expires_at, + ) + else: + self._drop_session(pending.peer_addr) + def _new_request_id(self) -> str: return self._request_id_factory() # ── chain helpers ──────────────────────────────────────────────────────── + def _tip_hash(self) -> str | None: + chain = self.chain.chain + return chain[-1].hash if chain else None + + def _chain_hash_index(self) -> tuple[dict[str, Block], dict[str, int]]: + """Return cached hash/position maps for the current main chain.""" + chain = self.chain.chain + reorged = self._indexed_length > 0 and ( + len(chain) < self._indexed_length + or chain[self._indexed_length - 1].hash != self._indexed_tip + ) + if reorged: + self._hash_index = {} + self._hash_positions = {} + self._indexed_length = 0 + self._indexed_tip = None + for index in range(self._indexed_length, len(chain)): + block = chain[index] + self._hash_index[block.hash] = block + self._hash_positions[block.hash] = index + if len(chain) != self._indexed_length: + self._indexed_length = len(chain) + self._indexed_tip = chain[-1].hash if chain else None + return self._hash_index, self._hash_positions + def _find_block(self, block_hash: str) -> Block | None: - for block in reversed(self.chain.chain): - if block.hash == block_hash: - return block - return None + return self._chain_hash_index()[0].get(block_hash) def _index_of(self, block: Block) -> int: - for index, candidate in enumerate(self.chain.chain): - if candidate.hash == block.hash: - return index - raise SyncError("block is not part of the local chain") + position = self._chain_hash_index()[1].get(block.hash) + if position is None: + raise SyncError("block is not part of the local chain") + return position def _suffix_work(self, start_block: Block) -> int: start_index = self._index_of(start_block) return _total_work(self.chain.chain[start_index + 1 :]) def find_common_ancestor(self, locator: Sequence[str]) -> Block | None: - by_hash = {block.hash: block for block in self.chain.chain} + by_hash = self._chain_hash_index()[0] for block_hash in locator: block = by_hash.get(block_hash) if block is not None: diff --git a/tests/test_sync.py b/tests/test_sync.py index dd421c2..1c1a5b4 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -51,21 +51,61 @@ def _rehash(header, **changes): return replace(updated, hash=updated.compute_hash()) -def _fake_chain(length): +def _fake_chain(length, *, timestamp_base=1_700_000_000_000): blocks = [] for index in range(length): block = Block( index=index, previous_hash="0" * 64 if index == 0 else blocks[-1].hash, target=MAX_TARGET, - timestamp=1_700_000_000_000 + index, + timestamp=timestamp_base + index, ) block.hash = block.compute_hash() blocks.append(block) return SimpleNamespace(chain=blocks) -def _make_sync_pair(receiver_chain, sender_chain, **client_kwargs): +def _fake_branch(genesis, length): + """A competing chain from ``genesis`` with a different timestamp sequence.""" + blocks = [genesis] + for index in range(1, length + 1): + block = Block( + index=index, + previous_hash=blocks[-1].hash, + target=MAX_TARGET, + timestamp=genesis.timestamp + 1000 + index, + ) + block.hash = block.compute_hash() + blocks.append(block) + return SimpleNamespace(chain=blocks) + + +def _mine_empty_block(chain, miner, timestamp): + """Mine an empty block with an explicit timestamp for deterministic EMA.""" + temp_state = chain.state.copy() + temp_state.chain_id = chain.chain_id + temp_state.credit_mining_reward(miner, reward=temp_state.DEFAULT_MINING_REWARD) + block = Block( + index=chain.last_block.index + 1, + previous_hash=chain.last_block.hash, + transactions=[], + timestamp=timestamp, + state_root=temp_state.state_root(), + receipt_root=None, + receipts=[], + miner=miner, + target=chain.current_target, + ) + from minichain.pow import mine_block + + mined = mine_block(block) + assert chain.add_block(mined) == ValidationStatus.VALID + return mined + + +def _make_sync_pair( + receiver_chain, sender_chain, *, server_kwargs=None, **client_kwargs +): receiver_outbox = [] sender_outbox = [] receiver_sent = [] @@ -78,7 +118,7 @@ def sender_send(peer, message): sender_outbox.append((peer, message)) client = SyncManager(receiver_chain, receiver_send, **client_kwargs) - server = SyncManager(sender_chain, sender_send) + server = SyncManager(sender_chain, sender_send, **(server_kwargs or {})) return client, server, receiver_outbox, sender_outbox, receiver_sent @@ -121,6 +161,67 @@ def test_missing_hash_is_computed(self, genesis_file): assert header.hash == expected_hash + @pytest.mark.parametrize( + "receipt_root,miner,expected_hash", + [ + ( + None, + None, + "ba34a5340ad135c7820c508ea37bbc82f858298a33f8e2abfa26d7d86d1d0085", + ), + ( + "33" * 32, + None, + "855a744df5013d3c1cbdbe065cdb7cb65d69ba0252474f7a12c3516c51e8e6c4", + ), + ( + None, + "44" * 32, + "7587aca78c935ccbee746674ac34eddc4d53e61f000441440b4b1718a3d2e653", + ), + ( + "33" * 32, + "44" * 32, + "b164445a7b61128bb334ffdf8cd11ee4a6eb1b06633964f6df657ba11441d60d", + ), + ], + ids=[ + "no-receipt-no-miner", + "receipt-no-miner", + "no-receipt-miner", + "receipt-miner", + ], + ) + def test_legacy_header_vectors_are_pinned(self, receipt_root, miner, expected_hash): + expected = { + "index": 7, + "previous_hash": "11" * 32, + "merkle_root": None, + "state_root": "22" * 32, + "receipt_root": receipt_root, + "timestamp": 1_700_000_000_123, + "target": hex(MAX_TARGET), + "nonce": 42, + } + if miner is not None: + expected["miner"] = miner + + header = BlockHeader( + index=7, + previous_hash="11" * 32, + merkle_root=None, + state_root="22" * 32, + receipt_root=receipt_root, + timestamp=1_700_000_000_123, + target=MAX_TARGET, + nonce=42, + miner=miner, + ) + + assert header.to_dict(include_hash=False) == expected + assert header.compute_hash() == expected_hash + assert BlockHeader.from_dict(expected).hash == expected_hash + def test_tampered_header_is_rejected(self, genesis_file): path, _, pk = genesis_file chain = Blockchain(genesis_path=path) @@ -230,6 +331,130 @@ def test_token_bucket_allows_burst_then_refills(self): now[0] = 1.0 assert limiter.allow("peer:a") + def test_bucket_state_is_bounded_and_idle_peers_expire(self): + now = [0.0] + limiter = PeerRateLimiter( + rate_per_second=1.0, + burst=1, + clock=lambda: now[0], + max_peers=2, + ) + + assert limiter.allow("peer:a") + assert limiter.allow("peer:b") + assert not limiter.allow("peer:c") + assert len(limiter._buckets) <= 2 + + # After a full refill window every remaining bucket is indistinguishable + # from an absent one, so maintenance can drop them all. + now[0] = 2.0 + limiter.prune() + assert limiter._buckets == {} + + +class TestSyncRecovery: + def test_unanswered_request_is_retried_then_dropped(self): + now = [0.0] + sent = [] + manager = SyncManager( + _fake_chain(3), + lambda peer, message: sent.append(message), + clock=lambda: now[0], + request_timeout=30.0, + retry_seconds=5.0, + max_retries=1, + ) + + manager.request_sync("peer:a") + assert len(sent) == 1 + + now[0] = 6.0 + manager.maintain() + assert len(sent) == 2 + assert sent[-1]["type"] == "getheaders" + + # The retry budget is spent, but the session deadline has not passed. + manager.maintain() + assert len(sent) == 2 + assert manager._pending + + now[0] = 31.0 + manager.maintain() + assert not manager._pending + assert not manager._sessions + + def test_expired_request_does_not_block_request_sync(self): + now = [0.0] + sent = [] + manager = SyncManager( + _fake_chain(3), + lambda peer, message: sent.append(message), + clock=lambda: now[0], + request_timeout=30.0, + retry_seconds=5.0, + max_retries=0, + ) + + manager.request_sync("peer:a") + assert len(sent) == 1 + + now[0] = 31.0 + manager.request_sync("peer:a") + assert len(sent) == 2 + + def test_outgoing_requests_wait_for_a_peer_token(self): + now = [0.0] + sent = [] + manager = SyncManager( + _fake_chain(3), + lambda peer, message: sent.append(message), + clock=lambda: now[0], + outgoing_rate_limiter=PeerRateLimiter( + rate_per_second=1.0, burst=1, clock=lambda: now[0] + ), + ) + + manager.request_sync("peer:a") + assert [message["type"] for message in sent] == ["getheaders"] + + manager._pending.clear() + manager.request_sync("peer:a") + assert len(sent) == 1 + assert "peer:a" in manager._deferred + + now[0] = 0.5 + manager.maintain() + assert len(sent) == 1 + + now[0] = 1.0 + manager.maintain() + assert [message["type"] for message in sent] == ["getheaders", "getheaders"] + assert manager._deferred == {} + + def test_dropping_a_session_discards_its_deferred_request(self): + now = [0.0] + sent = [] + manager = SyncManager( + _fake_chain(3), + lambda peer, message: sent.append(message), + clock=lambda: now[0], + outgoing_rate_limiter=PeerRateLimiter( + rate_per_second=1.0, burst=1, clock=lambda: now[0] + ), + ) + + manager.request_sync("peer:a") + manager._pending.clear() + manager.request_sync("peer:a") + assert "peer:a" in manager._deferred + + manager._drop_session("peer:a") + now[0] = 1.0 + manager.maintain() + + assert manager._deferred == {} + assert [message["type"] for message in sent] == ["getheaders"] + class TestSyncResponses: def test_default_header_batch_fits_in_one_frame(self): @@ -246,14 +471,23 @@ def test_default_header_batch_fits_in_one_frame(self): def test_header_response_honours_count_and_byte_limits(self): chain = _fake_chain(10) header_size = ( - len(canonical_json_dumps(BlockHeader.from_block(chain.chain[1]).to_dict())) + len( + canonical_json_dumps( + BlockHeader.from_block(chain.chain[1]).to_dict() + ).encode("utf-8") + ) + 1 ) + envelope = len( + canonical_json_dumps({"type": "headers", "data": {"headers": []}}).encode( + "utf-8" + ) + ) manager = SyncManager( chain, lambda *_: None, max_headers=5, - max_response_bytes=512 + header_size * 2 + 10, + max_response_bytes=envelope + header_size * 2, ) response = manager.build_headers_response( @@ -265,13 +499,7 @@ def test_header_response_honours_count_and_byte_limits(self): def test_block_response_is_an_ordered_contiguous_prefix(self): chain = _fake_chain(5) - block_size = len(canonical_json_dumps(chain.chain[1].to_dict())) + 1 - manager = SyncManager( - chain, - lambda *_: None, - max_blocks=4, - max_response_bytes=512 + block_size * 3 + 10, - ) + manager = SyncManager(chain, lambda *_: None, max_blocks=4) response = manager.build_blocks_response( { @@ -289,6 +517,52 @@ def test_block_response_is_an_ordered_contiguous_prefix(self): chain.chain[1].hash, ] + def test_block_response_honours_byte_limit(self): + chain = _fake_chain(6) + block_size = ( + len(canonical_json_dumps(chain.chain[1].to_dict()).encode("utf-8")) + 1 + ) + envelope = len( + canonical_json_dumps({"type": "blocks", "data": {"blocks": []}}).encode( + "utf-8" + ) + ) + manager = SyncManager( + chain, + lambda *_: None, + max_blocks=4, + max_response_bytes=envelope + block_size * 2, + ) + + response = manager.build_blocks_response( + { + "hashes": [ + chain.chain[1].hash, + chain.chain[2].hash, + chain.chain[3].hash, + chain.chain[4].hash, + ] + } + ) + + assert [b["index"] for b in response["data"]["blocks"]] == [1, 2] + assert len(canonical_json_dumps(response).encode("utf-8")) <= ( + manager.max_response_bytes + ) + + def test_single_block_may_use_the_soft_budget(self): + chain = _fake_chain(2) + manager = SyncManager( + chain, + lambda *_: None, + max_response_bytes=64, + ) + + response = manager.build_blocks_response({"hashes": [chain.chain[1].hash]}) + + assert [b["index"] for b in response["data"]["blocks"]] == [1] + assert len(canonical_json_dumps(response).encode("utf-8")) < MAX_FRAME_BYTES + def test_request_bounds_are_enforced(self): chain = _fake_chain(5) manager = SyncManager(chain, lambda *_: None, max_blocks=2, locator_size=2) @@ -303,6 +577,97 @@ def test_request_bounds_are_enforced(self): {"hashes": [chain.chain[i].hash for i in range(3)]} ) + def test_request_id_is_bounded_and_counted_in_the_budget(self): + chain = _fake_chain(5) + manager = SyncManager(chain, lambda *_: None) + oversized = "a" * (manager.max_request_id_bytes + 1) + + with pytest.raises(SyncError, match="request_id"): + manager.build_headers_response({"locator": [], "request_id": oversized}) + with pytest.raises(SyncError, match="request_id"): + manager.build_blocks_response( + {"hashes": [chain.chain[1].hash], "request_id": oversized} + ) + + request_id = "b" * 32 + header_size = ( + len( + canonical_json_dumps( + BlockHeader.from_block(chain.chain[1]).to_dict() + ).encode("utf-8") + ) + + 1 + ) + envelope = len( + canonical_json_dumps( + { + "type": "headers", + "data": {"headers": [], "request_id": request_id}, + } + ).encode("utf-8") + ) + tight = SyncManager( + chain, + lambda *_: None, + max_response_bytes=envelope + header_size * 2, + ) + + response = tight.build_headers_response( + {"locator": [chain.chain[0].hash], "max": 5, "request_id": request_id} + ) + + assert response["data"]["request_id"] == request_id + assert len(response["data"]["headers"]) == 2 + assert len(canonical_json_dumps(response).encode("utf-8")) <= ( + tight.max_response_bytes + ) + + block_response = manager.build_blocks_response( + {"hashes": [chain.chain[1].hash], "request_id": request_id} + ) + assert block_response["data"]["request_id"] == request_id + assert len(canonical_json_dumps(block_response).encode("utf-8")) <= ( + manager.max_response_bytes + ) + + def test_block_hash_index_is_reused_and_rebuilt_after_reorg(self): + chain = _fake_chain(5) + manager = SyncManager(chain, lambda *_: None) + + assert manager.build_blocks_response({"hashes": [chain.chain[4].hash]})["data"][ + "blocks" + ] + cached = manager._hash_index + + # Appending extends the cached index in place. + extension = Block( + index=5, + previous_hash=chain.chain[4].hash, + target=MAX_TARGET, + timestamp=chain.chain[4].timestamp + 1, + ) + extension.hash = extension.compute_hash() + chain.chain.append(extension) + + response = manager.build_blocks_response({"hashes": [extension.hash]}) + assert [block["index"] for block in response["data"]["blocks"]] == [5] + assert manager._hash_index is cached + + # Replacing the indexed suffix invalidates every stale entry. + replacement = Block( + index=4, + previous_hash=chain.chain[3].hash, + target=MAX_TARGET, + timestamp=chain.chain[3].timestamp + 2, + ) + replacement.hash = replacement.compute_hash() + chain.chain[4:] = [replacement] + + response = manager.build_blocks_response({"hashes": [replacement.hash]}) + assert [block["index"] for block in response["data"]["blocks"]] == [4] + response = manager.build_blocks_response({"hashes": [extension.hash]}) + assert response["data"]["blocks"] == [] + def test_rate_limited_peer_gets_no_response(self): now = [0.0] sent = [] @@ -318,9 +683,9 @@ def test_rate_limited_peer_gets_no_response(self): request = {"locator": [], "max": 1} assert manager.handle_message("peer:a", "getheaders", request) is None - assert manager.handle_message("peer:a", "getheaders", request) == ( - ValidationStatus.FAILED - ) + # A dropped request is not a protocol failure: the client retries it, + # so an honest peer must not accumulate FAILED score for a burst. + assert manager.handle_message("peer:a", "getheaders", request) is None assert len(sent) == 1 @@ -390,6 +755,79 @@ async def scenario(): asyncio.run(scenario()) + def _genesis(self, tmp_path): + genesis = tmp_path / "genesis_handler.json" + genesis.write_text( + json.dumps( + { + "chain_id": "minichain-default", + "timestamp": 1_700_000_000_000, + "target": hex(MAX_TARGET), + "target_block_time": 1000, + "alloc": {}, + } + ), + encoding="utf-8", + ) + return Blockchain(genesis_path=str(genesis)) + + def test_legacy_response_is_capped_by_count(self, tmp_path): + chain = self._genesis(tmp_path) + pk = SigningKey.generate().verify_key.encode(encoder=HexEncoder).decode() + for _ in range(3): + assert mine_and_process_block(chain, Mempool(), pk) is not None + + async def scenario(): + network = self.FakeNetwork() + sync = SyncManager(chain, lambda *_: None, max_blocks=2) + handler = make_network_handler(chain, Mempool(), network, sync) + + status = await handler( + { + "type": "chain_request", + "data": {"start_index": 0, "limit": 10_000}, + "_peer_addr": "peer:a", + } + ) + assert status is None + await asyncio.sleep(0) + + _, response = network.sent[0] + assert len(response["data"]["blocks"]) == 2 + assert response["data"]["requested_limit"] == 2 + + asyncio.run(scenario()) + + def test_legacy_requests_share_the_per_peer_rate_limit(self, tmp_path): + chain = self._genesis(tmp_path) + now = [0.0] + + async def scenario(): + network = self.FakeNetwork() + sync = SyncManager( + chain, + lambda *_: None, + rate_limiter=PeerRateLimiter( + rate_per_second=1.0, burst=1, clock=lambda: now[0] + ), + ) + handler = make_network_handler(chain, Mempool(), network, sync) + request = { + "type": "chain_request", + "data": {"start_index": 0, "limit": 1}, + "_peer_addr": "peer:a", + } + + assert await handler(request) is None + await asyncio.sleep(0) + assert len(network.sent) == 1 + + assert await handler(request) is None + await asyncio.sleep(0) + assert len(network.sent) == 1 + + asyncio.run(scenario()) + class TestHeadersFirstSync: def test_linear_sync_paginates_headers_and_bodies(self, genesis_file): @@ -414,11 +852,56 @@ def test_linear_sync_paginates_headers_and_bodies(self, genesis_file): assert all(status is None for status in statuses) assert receiver.last_block.hash == sender.last_block.hash assert [block.index for block in applied] == [1, 2, 3, 4, 5] - assert [m["type"] for m in receiver_sent].count("getheaders") == 2 + # The final getheaders returns an empty batch, which ends the sync. + assert [m["type"] for m in receiver_sent].count("getheaders") == 3 assert [ len(m["data"]["hashes"]) for m in receiver_sent if m["type"] == "getblocks" ] == [2, 2, 1] + def test_short_header_batches_continue_until_empty(self, genesis_file): + path, _, pk = genesis_file + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + for _ in range(5): + assert mine_and_process_block(sender, Mempool(), pk) is not None + + header_size = max( + len( + canonical_json_dumps(BlockHeader.from_block(block).to_dict()).encode( + "utf-8" + ) + ) + + 1 + for block in sender.chain[1:] + ) + request_id = "0" * 32 + envelope = len( + canonical_json_dumps( + { + "type": "headers", + "data": {"headers": [], "request_id": request_id}, + } + ).encode("utf-8") + ) + client, server, receiver_out, sender_out, receiver_sent = _make_sync_pair( + receiver, + sender, + max_headers=10, + max_blocks=2, + server_kwargs={ + "max_headers": 10, + "max_response_bytes": envelope + header_size * 2, + }, + ) + client.request_sync("peer:test") + + statuses = _pump(client, server, receiver_out, sender_out) + + assert all(status is None for status in statuses) + assert receiver.last_block.hash == sender.last_block.hash + # Two headers fit per batch, so 5 blocks need 2 + 2 + 1 + an empty end. + assert [m["type"] for m in receiver_sent].count("getheaders") == 4 + def test_invalid_headers_stop_before_any_body_download(self, genesis_file): path, _, pk = genesis_file receiver = Blockchain(genesis_path=path) @@ -444,6 +927,65 @@ def test_invalid_headers_stop_before_any_body_download(self, genesis_file): assert receiver.last_block.index == 0 assert not client._sessions + def test_local_tip_change_restarts_sync_without_penalty(self, genesis_file): + path, _, pk = genesis_file + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + for _ in range(2): + assert mine_and_process_block(sender, Mempool(), pk) is not None + + client, server, receiver_out, sender_out, receiver_sent = _make_sync_pair( + receiver, + sender, + max_blocks=2, + ) + client.request_sync("peer:test") + + peer, request = receiver_out.pop(0) + assert server.handle_message(peer, request["type"], request["data"]) is None + peer, response = sender_out.pop(0) + assert client.handle_message(peer, response["type"], response["data"]) is None + + # A broadcast advances our tip while the body request is outstanding. + assert mine_and_process_block(receiver, Mempool(), pk) is not None + + peer, request = receiver_out.pop(0) + assert server.handle_message(peer, request["type"], request["data"]) is None + peer, response = sender_out.pop(0) + status = client.handle_message(peer, response["type"], response["data"]) + + assert status is None + assert not client._sessions + assert [m["type"] for m in receiver_sent].count("getheaders") == 2 + assert receiver_out[-1][1]["type"] == "getheaders" + + def test_null_block_drops_the_session(self, genesis_file): + path, _, pk = genesis_file + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + assert mine_and_process_block(sender, Mempool(), pk) is not None + + client, server, receiver_out, sender_out, _ = _make_sync_pair( + receiver, + sender, + ) + client.request_sync("peer:test") + + peer, request = receiver_out.pop(0) + assert server.handle_message(peer, request["type"], request["data"]) is None + peer, response = sender_out.pop(0) + assert client.handle_message(peer, response["type"], response["data"]) is None + + peer, request = receiver_out.pop(0) + assert server.handle_message(peer, request["type"], request["data"]) is None + peer, response = sender_out.pop(0) + response["data"]["blocks"] = [None] + status = client.handle_message(peer, response["type"], response["data"]) + + assert status == ValidationStatus.MALFORMED + assert not client._sessions + assert not client._pending + def test_fork_sync_reorgs_and_returns_orphans(self, genesis_file): path, sk, pk = genesis_file receiver = Blockchain(genesis_path=path) @@ -532,3 +1074,62 @@ def test_deep_fork_accumulates_headers_across_batches(self, genesis_file): ] == [3, 3, 2] assert not client._sessions assert not client._pending + + def test_fork_header_accumulation_is_bounded(self): + receiver = _fake_chain(10) + sender = _fake_branch(receiver.chain[0], 12) + client, server, receiver_out, sender_out, _ = _make_sync_pair( + receiver, + sender, + max_headers=1, + max_blocks=2, + max_fork_headers=2, + ) + client.request_sync("peer:test") + + statuses = _pump(client, server, receiver_out, sender_out) + + assert ValidationStatus.MALFORMED in statuses + assert not client._sessions + assert not client._pending + + def test_fork_sync_uses_the_forks_target_sequence(self, genesis_file): + path, _, pk = genesis_file + template = Blockchain(genesis_path=path) + shared = _mine_empty_block(template, pk, template.last_block.timestamp + 1000) + + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + assert receiver.add_block(shared) == ValidationStatus.VALID + assert sender.add_block(shared) == ValidationStatus.VALID + + # The local branch mines faster than the target, lowering its target. + receiver_2 = _mine_empty_block(receiver, pk, shared.timestamp + 500) + _mine_empty_block(receiver, pk, receiver_2.timestamp + 1000) + + # The competing branch mines slower, keeping the maximum target, and + # ends up both longer and heavier. + sender_2 = _mine_empty_block(sender, pk, shared.timestamp + 2000) + sender_3 = _mine_empty_block(sender, pk, sender_2.timestamp + 1000) + _mine_empty_block(sender, pk, sender_3.timestamp + 1000) + + local_target = receiver.current_target + local_avg = receiver.avg_block_time + assert local_target != sender.current_target + assert local_avg != sender.avg_block_time + + client, server, receiver_out, sender_out, _ = _make_sync_pair( + receiver, + sender, + ) + client.request_sync("peer:test") + + statuses = _pump(client, server, receiver_out, sender_out) + + assert all(status is None for status in statuses) + assert receiver.last_block.hash == sender.last_block.hash + assert receiver.current_target == sender.current_target + assert receiver.avg_block_time == sender.avg_block_time + assert receiver.current_target != local_target + assert receiver.avg_block_time != local_avg + assert receiver.state.accounts == sender.state.accounts From d4260d6307dcf6b263aedcde6222fbae10de2522 Mon Sep 17 00:00:00 2001 From: anshulchikhale30-p Date: Mon, 9 Nov 2026 09:31:41 +0530 Subject: [PATCH 4/4] fix: harden sync retries and fork retention - keep earlier attempt IDs valid until the session deadline - preserve the batch tip across body retries and restart on changes - restart stale fork comparisons instead of penalizing the peer - cap first-batch and accumulated fork headers without scoring - bound buffered fork bodies with MAX_FORK_BODY_BYTES - allow one frame-fitting header over the soft response budget - test UTF-8 response sizing and the shared legacy/modern budget --- docs/sync.md | 12 +- minichain/node_config.py | 20 ++- minichain/sync.py | 105 ++++++++++++- tests/test_sync.py | 313 ++++++++++++++++++++++++++++++++++++++- 4 files changed, 433 insertions(+), 17 deletions(-) diff --git a/docs/sync.md b/docs/sync.md index e4d6b98..7650d35 100644 --- a/docs/sync.md +++ b/docs/sync.md @@ -123,6 +123,7 @@ Defaults live in [`minichain/node_config.py`](../minichain/node_config.py): | `SYNC_REQUESTS_PER_SECOND` / `SYNC_REQUEST_BURST` | 5 / 10 | Per-peer token bucket for serving requests and pacing outbound requests | | `SYNC_REQUEST_RETRY_SECONDS` / `SYNC_REQUEST_MAX_RETRIES` | 5 / 2 | Retry an unanswered request before dropping the session | | `MAX_FORK_HEADERS` | 16,000 | Cap on headers retained while following a fork | +| `MAX_FORK_BODY_BYTES` | 64 MiB | Cap on bodies buffered while following a fork | A request's `max` field and a legacy `chain_request.limit` are clamped to their configured ranges, so an out-of-range value alone is not a validation failure. @@ -136,8 +137,15 @@ requested count because of the byte budget, and the requester continues from the last received item; an empty header batch ends the sync. Outbound requests are paced against the same per-peer budget, and a request that would exceed it is queued briefly instead of being dropped. Unanswered requests are retried before -a stalled session is dropped. A single valid block may exceed the soft response -budget when its complete frame still fits within `MAX_FRAME_BYTES`. +a stalled session is dropped; earlier attempt IDs stay valid until the session +deadline so a late response can still complete the sync. A single valid header +or block may exceed the soft response budget when its complete frame still fits +within `MAX_FRAME_BYTES`. + +A fork that exceeds `MAX_FORK_HEADERS` or `MAX_FORK_BODY_BYTES` is dropped +without penalizing the peer: the caps are local retention policy, not protocol +violations. If the local chain gains work while fork bodies are downloading, the +stale comparison restarts sync instead of treating the peer as invalid. ## Reorgs diff --git a/minichain/node_config.py b/minichain/node_config.py index 55876ae..810edcb 100644 --- a/minichain/node_config.py +++ b/minichain/node_config.py @@ -1,11 +1,11 @@ # node_config.py -# This file contains parameters that individual node operators can tweak +# This file contains parameters that individual node operators can tweak # without causing consensus forks. # P2P Banning Thresholds -MALFORMED_THRESHOLD = 15 # N: accumulated malformed messages before ban -FAILED_THRESHOLD = 15 # M: accumulated failed messages before ban -INVALID_THRESHOLD = 1 # L: accumulated invalid messages before ban (1 = immediate) +MALFORMED_THRESHOLD = 15 # N: accumulated malformed messages before ban +FAILED_THRESHOLD = 15 # M: accumulated failed messages before ban +INVALID_THRESHOLD = 1 # L: accumulated invalid messages before ban (1 = immediate) DECAY_INTERVAL_MINUTES = 10 # T: counter half-life period in minutes # Gossip dedup: max tx/block ids remembered per set before the oldest is evicted. @@ -16,7 +16,9 @@ MEMPOOL_TX_PER_BLOCK = 100 # Mining Config -MINING_MAX_NONCE = 10_000_000 # Number of hashes to attempt before yielding the mining thread +MINING_MAX_NONCE = ( + 10_000_000 # Number of hashes to attempt before yielding the mining thread +) # Initial nonce range parameters (A and B). # Miners can configure these to create unique search ranges and avoid overlapping work. @@ -28,7 +30,9 @@ MINING_INITIAL_NONCE_MAX = 2**32 - 1 # State Config -MAX_STATE_SNAPSHOTS = 10 # Number of recent block states to keep in memory for reorg optimization +MAX_STATE_SNAPSHOTS = ( + 10 # Number of recent block states to keep in memory for reorg optimization +) # Sync Config (node-local policy; not consensus) # 1000 JSON headers are roughly 540 KB, comfortably below MAX_FRAME_BYTES. @@ -49,6 +53,10 @@ # proves more work than our suffix. Cap that retention so a peer cannot make # us hold an unbounded branch in memory. MAX_FORK_HEADERS = MAX_HEADERS_PER_REQUEST * 16 +# Fork bodies are buffered until resolve_conflicts can validate the branch. +# Cap that buffer independently of the header count so large bodies cannot +# exhaust memory during a deep fork. +MAX_FORK_BODY_BYTES = 64 * 1024 * 1024 # Retry an unanswered sync request before dropping the session. This recovers # from a receiver that rate-limited (and therefore ignored) one request. SYNC_REQUEST_RETRY_SECONDS = 5.0 diff --git a/minichain/sync.py b/minichain/sync.py index 54022c7..79af721 100644 --- a/minichain/sync.py +++ b/minichain/sync.py @@ -33,6 +33,7 @@ ) from .node_config import ( MAX_BLOCKS_PER_REQUEST, + MAX_FORK_BODY_BYTES, MAX_FORK_HEADERS, MAX_HEADERS_PER_REQUEST, MAX_LOCATOR_HASHES, @@ -137,6 +138,8 @@ class _PendingRequest: created_at: float expires_at: float = 0.0 attempts: int = 1 + group: str = "" + retired: bool = False @dataclass @@ -153,6 +156,8 @@ class _SyncSession: fork: bool work: int = 0 local_tip: str | None = None + start_tip: str | None = None + body_bytes: int = 0 next_index: int = 0 blocks: list[Block] = field(default_factory=list) @@ -260,6 +265,7 @@ def __init__( max_response_bytes: int = MAX_SYNC_RESPONSE_BYTES, locator_size: int = MAX_LOCATOR_HASHES, max_fork_headers: int = MAX_FORK_HEADERS, + max_fork_body_bytes: int = MAX_FORK_BODY_BYTES, max_request_id_bytes: int = MAX_SYNC_REQUEST_ID_BYTES, request_timeout: float = SYNC_REQUEST_TIMEOUT_SECONDS, retry_seconds: float = SYNC_REQUEST_RETRY_SECONDS, @@ -277,6 +283,8 @@ def __init__( raise ValueError("locator_size must be positive") if max_fork_headers <= 0: raise ValueError("max_fork_headers must be positive") + if max_fork_body_bytes <= 0: + raise ValueError("max_fork_body_bytes must be positive") if max_request_id_bytes <= 0: raise ValueError("max_request_id_bytes must be positive") if request_timeout <= 0 or retry_seconds <= 0 or max_retries < 0: @@ -290,6 +298,7 @@ def __init__( self.max_response_bytes = max_response_bytes self.locator_size = locator_size self.max_fork_headers = max_fork_headers + self.max_fork_body_bytes = max_fork_body_bytes self.max_request_id_bytes = max_request_id_bytes self.request_timeout = request_timeout self.retry_seconds = retry_seconds @@ -354,6 +363,7 @@ def _send_headers_request( locator: Sequence[str] | None = None, attempts: int = 1, expires_at: float | None = None, + group: str | None = None, ) -> None: now = self._clock() request_id = self._new_request_id() @@ -365,6 +375,7 @@ def _send_headers_request( now + self.request_timeout if expires_at is None else expires_at ), attempts=attempts, + group=group or request_id, ) self._dispatch_sync( peer_addr, @@ -503,6 +514,13 @@ def build_headers_response(self, payload: dict) -> dict: header = BlockHeader.from_block(block).to_dict() size = _encoded_size(header) + 1 if used + size > budget: + if not headers: + if self._fits_frame("headers", "headers", header, request_id, None): + headers.append(header) + else: + logger.error( + "Header %s does not fit in a response frame", block.hash + ) break headers.append(header) used += size @@ -620,7 +638,7 @@ def _bounded_block_payloads( # stalling sync on a block that fits the network limit. The # encoded overhead is only needed for this single payload. if not payloads and self._fits_frame( - response_type, payload, request_id, extra + response_type, "blocks", payload, request_id, extra ): payloads.append(payload) break @@ -631,11 +649,12 @@ def _bounded_block_payloads( def _fits_frame( self, response_type: str, + collection_key: str, payload: dict, request_id: str | None, extra: dict | None, ) -> bool: - response = {"type": response_type, "data": {"blocks": [payload]}} + response = {"type": response_type, "data": {collection_key: [payload]}} if request_id is not None: response["data"]["request_id"] = request_id if extra: @@ -680,9 +699,15 @@ def _handle_headers(self, peer_addr: str, payload: dict) -> ValidationStatus | N raise SyncError("headers do not continue the advertised fork") validate_header_chain(headers, previous) if len(session.headers) + len(headers) > self.max_fork_headers: - raise SyncError( - f"fork headers exceed the maximum of {self.max_fork_headers}" + # The retention cap is local policy, not a protocol violation, + # so drop the session without scoring an honest peer as bad. + logger.warning( + "Fork headers from %s exceed the retention cap of %d", + peer_addr, + self.max_fork_headers, ) + self._drop_session(peer_addr) + return None session.headers.extend(headers) session.work += batch_work else: @@ -690,12 +715,21 @@ def _handle_headers(self, peer_addr: str, payload: dict) -> ValidationStatus | N if start_block is None: raise SyncError("headers do not connect to a known block") validate_header_chain(headers, start_block) + fork = start_block.hash != self._tip_hash() + if fork and len(headers) > self.max_fork_headers: + logger.warning( + "Fork headers from %s exceed the retention cap of %d", + peer_addr, + self.max_fork_headers, + ) + return None session = _SyncSession( peer_addr=peer_addr, headers=headers, common_ancestor=start_block, - fork=start_block.hash != self._tip_hash(), + fork=fork, work=batch_work, + start_tip=self._tip_hash(), ) self._sessions[peer_addr] = session @@ -740,6 +774,20 @@ def _handle_blocks(self, peer_addr: str, payload: dict) -> ValidationStatus | No if block.hash != header.hash: raise SyncError(f"received block {block.hash}, expected {header.hash}") + batch_bytes = 0 + if session.fork: + batch_bytes = sum(_encoded_size(raw) + 1 for raw in raw_blocks) + if session.body_bytes + batch_bytes > self.max_fork_body_bytes: + # The branch may be valid but too large for our retention + # policy, so drop it without scoring the peer as bad. + logger.warning( + "Fork bodies from %s exceed the retention cap of %d bytes", + peer_addr, + self.max_fork_body_bytes, + ) + self._drop_session(peer_addr) + return None + if not session.fork and session.local_tip != self._tip_hash(): # A broadcast advanced our chain while this body request was in # flight, so the response is validated against an old tip. Start @@ -751,6 +799,7 @@ def _handle_blocks(self, peer_addr: str, payload: dict) -> ValidationStatus | No if session.fork: session.blocks.extend(blocks) + session.body_bytes += batch_bytes else: for block in blocks: status = self.chain.add_block(block) @@ -768,6 +817,17 @@ def _handle_blocks(self, peer_addr: str, payload: dict) -> ValidationStatus | No if session.fork: success, orphans = self.chain.resolve_conflicts(session.blocks) if not success: + if session.start_tip != self._tip_hash(): + # The local chain gained work while the fork bodies were + # downloading, so the comparison is stale rather than the + # fork being invalid. Restart without penalizing the peer. + logger.info( + "Local tip changed during fork sync with %s; restarting", + peer_addr, + ) + self._drop_session(peer_addr) + self.request_sync(peer_addr) + return None self._drop_session(peer_addr) return ValidationStatus.INVALID for block in session.blocks: @@ -785,6 +845,7 @@ def _request_next_headers( *, attempts: int = 1, expires_at: float | None = None, + group: str | None = None, ) -> None: """Continue a fork's header range from its last known hash.""" self._send_headers_request( @@ -792,6 +853,7 @@ def _request_next_headers( locator=[session.headers[-1].hash], attempts=attempts, expires_at=expires_at, + group=group, ) def _request_next_blocks( @@ -800,6 +862,7 @@ def _request_next_blocks( *, attempts: int = 1, expires_at: float | None = None, + group: str | None = None, ) -> None: batch = session.headers[ session.next_index : session.next_index + self.max_blocks @@ -809,7 +872,12 @@ def _request_next_blocks( return now = self._clock() - session.local_tip = self._tip_hash() + if attempts == 1: + # Retries resend the same batch, so keep the tip captured when the + # batch was first requested. If a broadcast changes the tip before + # the retry, the response must still trigger a restart instead of + # being applied on top of the new tip. + session.local_tip = self._tip_hash() request_id = self._new_request_id() self._pending[request_id] = _PendingRequest( peer_addr=session.peer_addr, @@ -819,6 +887,7 @@ def _request_next_blocks( now + self.request_timeout if expires_at is None else expires_at ), attempts=attempts, + group=group or request_id, ) self._dispatch_sync( session.peer_addr, @@ -855,15 +924,32 @@ def _take_pending( pending = self._pending.get(request_id) if pending is None or pending.peer_addr != peer_addr or pending.kind != kind: return None - return self._pending.pop(request_id) + # A response to any attempt satisfies the whole group: stop waiting for + # the other attempts and retire their IDs so they cannot match again. + if pending.group: + for other_id, other in list(self._pending.items()): + if ( + other.group == pending.group + and other.peer_addr == peer_addr + and other.kind == kind + ): + self._pending.pop(other_id, None) + else: + self._pending.pop(request_id, None) + return pending def _prune_pending(self) -> None: now = self._clock() for request_id, pending in list(self._pending.items()): + if pending.retired: + # Keep the ID valid for a late response until the deadline. + if now >= pending.expires_at: + self._pending.pop(request_id, None) + continue if now - pending.created_at <= self.retry_seconds: continue if pending.attempts <= self.max_retries: - self._pending.pop(request_id, None) + pending.retired = True self._retry_request(pending) elif now >= pending.expires_at: self._pending.pop(request_id, None) @@ -878,12 +964,14 @@ def _retry_request(self, pending: _PendingRequest) -> None: pending.peer_addr, attempts=attempts, expires_at=pending.expires_at, + group=pending.group, ) elif session.fork and session.next_index == 0: self._request_next_headers( session, attempts=attempts, expires_at=pending.expires_at, + group=pending.group, ) else: self._drop_session(pending.peer_addr) @@ -892,6 +980,7 @@ def _retry_request(self, pending: _PendingRequest) -> None: session, attempts=attempts, expires_at=pending.expires_at, + group=pending.group, ) else: self._drop_session(pending.peer_addr) diff --git a/tests/test_sync.py b/tests/test_sync.py index 1c1a5b4..2876eb4 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -402,6 +402,78 @@ def test_expired_request_does_not_block_request_sync(self): manager.request_sync("peer:a") assert len(sent) == 2 + def test_response_to_an_earlier_attempt_is_accepted(self): + now = [0.0] + sent = [] + manager = SyncManager( + _fake_chain(3), + lambda peer, message: sent.append(message), + clock=lambda: now[0], + retry_seconds=5.0, + max_retries=1, + ) + manager.request_sync("peer:a") + first = sent[0] + first_id = first["data"]["request_id"] + + now[0] = 6.0 + manager.maintain() + second = sent[1] + second_id = second["data"]["request_id"] + assert first_id != second_id + assert manager._pending[first_id].retired + + previous = manager.chain.chain[-1] + header = BlockHeader( + index=previous.index + 1, + previous_hash=previous.hash, + merkle_root=None, + state_root=None, + receipt_root=None, + timestamp=previous.timestamp + 1, + target=MAX_TARGET, + nonce=0, + ) + header = replace(header, hash=header.compute_hash()) + status = manager.handle_message( + "peer:a", + "headers", + { + "headers": [header.to_dict()], + "request_id": first_id, + }, + ) + + assert status is None + assert manager._sessions + # A response to either attempt retires the whole request group. + assert first_id not in manager._pending + assert second_id not in manager._pending + assert [p.kind for p in manager._pending.values()] == ["blocks"] + assert sent[-1]["type"] == "getblocks" + + def test_retired_attempts_expire_with_the_session(self): + now = [0.0] + manager = SyncManager( + _fake_chain(3), + lambda *_: None, + clock=lambda: now[0], + retry_seconds=5.0, + max_retries=1, + ) + manager.request_sync("peer:a") + first_id = next(iter(manager._pending)) + + now[0] = 6.0 + manager.maintain() + assert len(manager._pending) == 2 + assert manager._pending[first_id].retired + + now[0] = 31.0 + manager.maintain() + assert not manager._pending + assert not manager._sessions + def test_outgoing_requests_wait_for_a_peer_token(self): now = [0.0] sent = [] @@ -563,6 +635,65 @@ def test_single_block_may_use_the_soft_budget(self): assert [b["index"] for b in response["data"]["blocks"]] == [1] assert len(canonical_json_dumps(response).encode("utf-8")) < MAX_FRAME_BYTES + def test_single_header_may_use_the_soft_budget(self): + chain = _fake_chain(2) + manager = SyncManager( + chain, + lambda *_: None, + max_response_bytes=64, + ) + + response = manager.build_headers_response({"locator": [chain.chain[0].hash]}) + + assert [h["index"] for h in response["data"]["headers"]] == [1] + assert len(canonical_json_dumps(response).encode("utf-8")) < MAX_FRAME_BYTES + + def test_block_byte_limit_counts_utf8(self): + transaction = Transaction( + sender="11" * 32, + receiver="22" * 32, + amount=1, + nonce=0, + data="\u00e9" * 100, + ) + blocks = [] + for index in range(3): + block = Block( + index=index, + previous_hash="0" * 64 if index == 0 else blocks[-1].hash, + target=MAX_TARGET, + timestamp=1_700_000_000_000 + index, + transactions=[transaction], + ) + block.hash = block.compute_hash() + blocks.append(block) + chain = SimpleNamespace(chain=blocks) + block_size = max( + len(canonical_json_dumps(block.to_dict()).encode("utf-8")) + 1 + for block in blocks[1:] + ) + envelope = len( + canonical_json_dumps({"type": "blocks", "data": {"blocks": []}}).encode( + "utf-8" + ) + ) + manager = SyncManager( + chain, + lambda *_: None, + max_blocks=3, + max_response_bytes=envelope + block_size, + ) + + response = manager.build_blocks_response( + {"hashes": [block.hash for block in blocks[1:]]} + ) + + assert [block["index"] for block in response["data"]["blocks"]] == [1] + encoded = canonical_json_dumps(response).encode("utf-8") + assert len(encoded) <= manager.max_response_bytes + # Character-based sizing would miss the extra UTF-8 bytes. + assert len(canonical_json_dumps(response)) < len(encoded) + def test_request_bounds_are_enforced(self): chain = _fake_chain(5) manager = SyncManager(chain, lambda *_: None, max_blocks=2, locator_size=2) @@ -828,6 +959,51 @@ async def scenario(): asyncio.run(scenario()) + def test_modern_and_legacy_requests_share_the_rate_limit(self, tmp_path): + chain = self._genesis(tmp_path) + now = [0.0] + + async def scenario(): + network = self.FakeNetwork() + sync = SyncManager( + chain, + lambda peer, payload: asyncio.create_task( + network._unicast_raw(peer, payload) + ), + rate_limiter=PeerRateLimiter( + rate_per_second=1.0, burst=1, clock=lambda: now[0] + ), + ) + handler = make_network_handler(chain, Mempool(), network, sync) + + assert ( + await handler( + { + "type": "getheaders", + "data": {"locator": [chain.chain[0].hash]}, + "_peer_addr": "peer:a", + } + ) + is None + ) + await asyncio.sleep(0) + assert network.sent[0][1]["type"] == "headers" + + assert ( + await handler( + { + "type": "chain_request", + "data": {"start_index": 0, "limit": 1}, + "_peer_addr": "peer:a", + } + ) + is None + ) + await asyncio.sleep(0) + assert len(network.sent) == 1 + + asyncio.run(scenario()) + class TestHeadersFirstSync: def test_linear_sync_paginates_headers_and_bodies(self, genesis_file): @@ -1089,9 +1265,144 @@ def test_fork_header_accumulation_is_bounded(self): statuses = _pump(client, server, receiver_out, sender_out) - assert ValidationStatus.MALFORMED in statuses + # Exceeding the local retention cap is not a peer protocol violation. + assert all(status is None for status in statuses) + assert not client._sessions + assert not client._pending + + def test_fork_header_cap_applies_to_the_first_batch(self): + receiver = _fake_chain(10) + sender = _fake_branch(receiver.chain[0], 12) + client, server, receiver_out, sender_out, receiver_sent = _make_sync_pair( + receiver, + sender, + max_headers=2, + max_fork_headers=1, + ) + client.request_sync("peer:test") + + statuses = _pump(client, server, receiver_out, sender_out) + + assert all(status is None for status in statuses) + assert not client._sessions + assert not [m for m in receiver_sent if m["type"] == "getblocks"] + + def test_fork_body_memory_is_bounded(self, genesis_file): + path, _, pk = genesis_file + template = Blockchain(genesis_path=path) + shared = _mine_empty_block(template, pk, template.last_block.timestamp + 1000) + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + assert receiver.add_block(shared) == ValidationStatus.VALID + assert sender.add_block(shared) == ValidationStatus.VALID + local = _mine_empty_block(receiver, pk, shared.timestamp + 1000) + sender_2 = _mine_empty_block(sender, pk, shared.timestamp + 1000) + _mine_empty_block(sender, pk, sender_2.timestamp + 1000) + + client, server, receiver_out, sender_out, receiver_sent = _make_sync_pair( + receiver, + sender, + max_blocks=2, + max_fork_body_bytes=1, + ) + client.request_sync("peer:test") + + statuses = _pump(client, server, receiver_out, sender_out) + + assert all(status is None for status in statuses) assert not client._sessions assert not client._pending + assert receiver.last_block.hash == local.hash + assert [m["type"] for m in receiver_sent].count("getblocks") == 1 + + def test_fork_race_with_local_tip_restarts_without_penalty(self, genesis_file): + path, _, pk = genesis_file + template = Blockchain(genesis_path=path) + shared = _mine_empty_block(template, pk, template.last_block.timestamp + 1000) + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + assert receiver.add_block(shared) == ValidationStatus.VALID + assert sender.add_block(shared) == ValidationStatus.VALID + + local = _mine_empty_block(receiver, pk, shared.timestamp + 1000) + sender_2 = _mine_empty_block(sender, pk, shared.timestamp + 1000) + _mine_empty_block(sender, pk, sender_2.timestamp + 1000) + + client, server, receiver_out, sender_out, receiver_sent = _make_sync_pair( + receiver, + sender, + max_blocks=1, + ) + client.request_sync("peer:test") + + peer, request = receiver_out.pop(0) + assert server.handle_message(peer, request["type"], request["data"]) is None + peer, response = sender_out.pop(0) + assert client.handle_message(peer, response["type"], response["data"]) is None + + peer, request = receiver_out.pop(0) + assert request["type"] == "getblocks" + assert server.handle_message(peer, request["type"], request["data"]) is None + peer, response = sender_out.pop(0) + assert client.handle_message(peer, response["type"], response["data"]) is None + + # The local chain gains more work while the fork bodies download. + next_local = _mine_empty_block(receiver, pk, local.timestamp + 1000) + final_local = _mine_empty_block(receiver, pk, next_local.timestamp + 1000) + + peer, request = receiver_out.pop(0) + assert request["type"] == "getblocks" + assert server.handle_message(peer, request["type"], request["data"]) is None + peer, response = sender_out.pop(0) + status = client.handle_message(peer, response["type"], response["data"]) + + assert status is None + assert not client._sessions + assert receiver.last_block.hash == final_local.hash + assert [m["type"] for m in receiver_sent].count("getheaders") == 2 + assert receiver_out[-1][1]["type"] == "getheaders" + + def test_block_retry_preserves_the_batch_tip_and_restarts(self, genesis_file): + path, _, pk = genesis_file + receiver = Blockchain(genesis_path=path) + sender = Blockchain(genesis_path=path) + for _ in range(2): + assert mine_and_process_block(sender, Mempool(), pk) is not None + + now = [0.0] + client, server, receiver_out, sender_out, _ = _make_sync_pair( + receiver, + sender, + clock=lambda: now[0], + ) + client.request_sync("peer:test") + + peer, request = receiver_out.pop(0) + assert server.handle_message(peer, request["type"], request["data"]) is None + peer, response = sender_out.pop(0) + assert client.handle_message(peer, response["type"], response["data"]) is None + peer, request = receiver_out.pop(0) + assert request["type"] == "getblocks" + batch = list(request["data"]["hashes"]) + + # A broadcast adds the first requested block before the retry. + assert receiver.add_block(sender.chain[1]) == ValidationStatus.VALID + + now[0] = 6.0 + client.maintain() + peer, retry = receiver_out.pop(0) + assert retry["type"] == "getblocks" + assert retry["data"]["hashes"] == batch + + # Answer the retried batch with the original request's response. + assert server.handle_message(peer, request["type"], request["data"]) is None + peer, response = sender_out.pop(0) + status = client.handle_message(peer, response["type"], response["data"]) + + assert status is None + assert not client._sessions + assert receiver.last_block.hash == sender.chain[1].hash + assert receiver_out[-1][1]["type"] == "getheaders" def test_fork_sync_uses_the_forks_target_sequence(self, genesis_file): path, _, pk = genesis_file