fix: sync with validated headers before downloading bodies (#178)

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/<seed-peer-id> --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:

docs/sync.md

@@ -0,0 +1,179 @@
+# 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": ["<newest known block>", "...", "<genesis>"],
+    "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": ["<header hash>", "..."]
+  }
+}
+```
+
+```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 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.
+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; 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
+
+When the common ancestor is not our tip, headers describe a competing branch.
+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
+
+`chain_request` and `chain_response` are still accepted for older peers:
+
+- 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.
+
+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.

main.py

@@ -32,7 +32,9 @@
 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
 from minichain.block import calculate_receipt_root
 from minichain.persistence import remember_peer, get_known_peers, is_peer_banned
@@ -80,12 +82,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 +186,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 +228,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 +275,40 @@ 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 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, OverflowError):
+                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,8 +643,18 @@ 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)
+
+    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)