From e8ca5d9c697399e295c4c26105e931a469b55028 Mon Sep 17 00:00:00 2001 From: siddhant Date: Sun, 30 Aug 2026 23:58:08 +0530 Subject: [PATCH] fix: make P2P networking actually work, and let a node mine on its own MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Swarm.listen() waits on a nursery that only exists while the swarm runs as a background service, so a node calling host.get_network().listen() directly never finished starting up: no multiaddr printed, no inbound connections possible. Listening now goes through host.run(). The dialing side had the same problem in reverse — it borrowed a nursery attribute the swarm doesn't expose, so a peer that dialed out never registered the connection or read from it. Both now use the nursery this module already opens for itself. Two mining bugs made a fresh chain effectively unusable: mining refused to produce a block when the mempool was empty, but the mining reward is the only source of new coins, so a chain that had never received one could never mine its first coin either. And two blocks minted within the same millisecond were rejected, since consensus requires strictly increasing timestamps. Both are fixed at the source: an empty block still carries the reward and proof-of-work, and the timestamp is clamped to be newer than the parent's instead of just read off the clock. A node that accepted a transaction or block only kept it — it never told any other peer, so a network more than one hop wide couldn't stay in sync. Accepted content is now relayed onward, excluding whoever sent it, with the existing dedup set stopping the echo from circulating. That set was unbounded, and now that it's load-bearing for stopping gossip loops rather than just avoiding duplicate work, its size is capped with an LRU eviction. Smaller fixes bundled in because they're cheap and adjacent: `peers` now lists the peer IDs it's connected to, not just a count; the JSON-RPC server's bind address is a flag instead of hardcoded to loopack; two unused constants (TRUSTED_PEERS, LOCALHOST_PEERS) are removed; and the README's --connect example, which predates the multiaddr-based CLI, is corrected. --- README.md | 4 +- main.py | 37 ++++++++----- minichain/node_config.py | 3 ++ minichain/p2p.py | 70 ++++++++++++++++++++----- tests/test_protocol_hardening.py | 90 ++++++++++++++++++++++++++++++++ 5 files changed, 176 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 4c41e20..e1f4433 100644 --- a/README.md +++ b/README.md @@ -89,9 +89,9 @@ python main.py --port 9000 --datadir ./node1_data *Note: Keep this terminal open to interact with the node via the CLI.* ### 2. Connecting to an Existing Chain -To connect a secondary node to the network, start a new instance on a different port and point it to the seed node using the `--connect` flag. +To connect a secondary node to the network, start a new instance on a different port and point it to the seed node using the `--connect` flag with the full multiaddress the seed node printed on startup (it includes the seed's peer ID, not just its host and port). ```bash -python main.py --port 9001 --connect 127.0.0.1:9000 --datadir ./node2_data +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. diff --git a/main.py b/main.py index b37fce4..2ead628 100644 --- a/main.py +++ b/main.py @@ -26,6 +26,7 @@ import sys import os import json +import time from nacl.signing import SigningKey from nacl.encoding import HexEncoder @@ -38,9 +39,6 @@ logger = logging.getLogger(__name__) -TRUSTED_PEERS = set() -LOCALHOST_PEERS = {"127.0.0.1", "::1", "localhost", "0:0:0:0:0:0:0:1"} - # ────────────────────────────────────────────── # Wallet helpers @@ -120,11 +118,15 @@ async def submit_and_broadcast(chain, mempool, network, tx, ok_msg, reject_msg): # ────────────────────────────────────────────── def mine_and_process_block(chain, mempool, miner_pk): - """Mine pending transactions into a new block.""" + """ + Mine pending transactions into a new block. + + A block with no transactions is still mined: it carries the mining reward and + adds proof-of-work, so the chain keeps advancing while the network is idle. + Requiring a transaction would also make a fresh chain unusable, since coins + only come into existence through the mining reward. + """ pending_txs = mempool.get_transactions_for_block() - if not pending_txs: - logger.info("Mempool is empty — nothing to mine.") - return None # Filter queue candidates against a temporary state snapshot. temp_state = chain.state.copy() @@ -147,17 +149,19 @@ def mine_and_process_block(chain, mempool, miner_pk): if stale_txs: mempool.remove_transactions(stale_txs) - if not mineable_txs: - logger.info("No mineable transactions in current queue window.") - return None - total_fees = sum(getattr(r, 'gas_used', 0) * getattr(tx, 'fee_per_gas', 0) for r, tx in zip(receipts, mineable_txs)) temp_state.credit_mining_reward(miner_pk, reward=temp_state.DEFAULT_MINING_REWARD + total_fees) + # Timestamps are milliseconds and a block must be strictly newer than its parent. + # Blocks can be mined inside the same millisecond, so clamp forward rather than + # letting the miner build a block the chain will reject. + timestamp = max(round(time.time() * 1000), chain.last_block.timestamp + 1) + block = Block( index=chain.last_block.index + 1, previous_hash=chain.last_block.hash, transactions=mineable_txs, + timestamp=timestamp, state_root=temp_state.state_root(), receipt_root=calculate_receipt_root(receipts), receipts=receipts, @@ -505,6 +509,8 @@ def print_prompt_info(current_pk): # ── peers ── elif cmd == "peers": print(f" Connected peers: {network.peer_count}") + for peer_id in network.peer_ids: + print(f" {peer_id}") # ── connect ── elif cmd == "connect": @@ -605,7 +611,8 @@ def print_prompt_info(current_pk): # Main entry point # ────────────────────────────────────────────── -async def run_node(port: int, host: str, connect_to: str | None, fund: int, datadir: str | None): +async def run_node(port: int, host: str, connect_to: str | None, fund: int, datadir: str | None, + rpc_host: str = "127.0.0.1"): """Boot the node, optionally connect to a peer, then enter the CLI.""" sk, pk = load_or_create_wallet(datadir) @@ -656,7 +663,7 @@ async def on_peer_connected(writer): # Start RPC server on a port correlated to the node port (e.g. 8545 if P2P is 9000) rpc_port = 8545 + (port - 9000) - await rpc_server.start(host="127.0.0.1", port=rpc_port) + await rpc_server.start(host=rpc_host, port=rpc_port) # Fund this node's wallet so it can transact in the demo if fund > 0: @@ -690,6 +697,7 @@ def main(): parser.add_argument("--connect", type=str, default=None, help="Peer address to connect to (multiaddr)") parser.add_argument("--fund", type=int, default=100, help="Initial coins to fund this wallet (default: 100)") parser.add_argument("--datadir", type=str, default=".minichain", help="Directory to save/load blockchain state (enables persistence)") + parser.add_argument("--rpc-host", type=str, default="127.0.0.1", help="Host/IP to bind the JSON-RPC server (default: 127.0.0.1, loopback-only)") args = parser.parse_args() logging.basicConfig( @@ -699,7 +707,8 @@ def main(): ) try: - asyncio.run(run_node(args.port, args.host, args.connect, args.fund, args.datadir)) + asyncio.run(run_node(args.port, args.host, args.connect, args.fund, args.datadir, + rpc_host=args.rpc_host)) except KeyboardInterrupt: print("\nNode shut down.") diff --git a/minichain/node_config.py b/minichain/node_config.py index 784b54e..0128139 100644 --- a/minichain/node_config.py +++ b/minichain/node_config.py @@ -8,6 +8,9 @@ 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. +SEEN_CACHE_MAX = 10_000 + # Mempool Config MEMPOOL_MAX_SIZE = 1000 MEMPOOL_TX_PER_BLOCK = 100 diff --git a/minichain/p2p.py b/minichain/p2p.py index 1171eb1..f44e16d 100644 --- a/minichain/p2p.py +++ b/minichain/p2p.py @@ -11,7 +11,12 @@ import struct import trio import queue -from .node_config import MALFORMED_THRESHOLD, FAILED_THRESHOLD, INVALID_THRESHOLD, DECAY_INTERVAL_MINUTES +from collections import OrderedDict + +from .node_config import ( + MALFORMED_THRESHOLD, FAILED_THRESHOLD, INVALID_THRESHOLD, DECAY_INTERVAL_MINUTES, + SEEN_CACHE_MAX, +) from .network_config import SUPPORTED_MESSAGE_TYPES, PROTOCOL_ID, MAX_FRAME_BYTES from libp2p import new_host @@ -48,12 +53,16 @@ def __init__( ): self._handler_callback = handler_callback self._on_peer_connected = None - self._seen_tx_ids = set() - self._seen_block_hashes = set() + # OrderedDict as an LRU: unbounded sets here would let a live node accumulate + # tx/block ids forever, and since relaying now depends on this set to stop + # echoes circulating, its size is no longer just a memory concern. + self._seen_tx_ids = OrderedDict() + self._seen_block_hashes = OrderedDict() self._to_trio = queue.Queue() self._to_asyncio = queue.Queue() self._peer_count = 0 self._peer_count_lock = threading.Lock() + self._peer_ids: set = set() # Misbehavior tracking, keyed directly by ValidationStatus so there is a # single vocabulary for statuses (no parallel string keys to convert). @@ -111,11 +120,16 @@ def _is_duplicate(self, msg_type, payload): def _mark_seen(self, msg_type, payload): mid = self._message_id(msg_type, payload) - if mid: - self._seen_set(msg_type).add(mid) + if not mid: + return + seen = self._seen_set(msg_type) + seen[mid] = True + seen.move_to_end(mid) + if len(seen) > SEEN_CACHE_MAX: + seen.popitem(last=False) - async def _broadcast_raw(self, payload: dict): - self._to_trio.put(("BROADCAST", payload)) + async def _broadcast_raw(self, payload: dict, exclude: str | None = None): + self._to_trio.put(("BROADCAST", (payload, exclude))) async def _unicast_raw(self, target_addr: str, payload: dict): self._to_trio.put(("UNICAST", (target_addr, payload))) @@ -144,6 +158,11 @@ def peer_count(self) -> int: with self._peer_count_lock: return self._peer_count + @property + def peer_ids(self) -> list: + with self._peer_count_lock: + return list(self._peer_ids) + # ── misbehavior helpers ────────────────────────────────────────────────── def _increment_counter(self, peer_id: str, status: ValidationStatus) -> bool: @@ -237,6 +256,17 @@ async def _asyncio_reader(self): except Exception: pass + # Relay accepted content onward, so gossip travels further than one + # hop. Only VALID is forwarded: rejected content is never amplified, + # and a None status (hello/chain_request/chain_response) is not + # relayable at all. Excluding the sender avoids a pointless echo, and + # _mark_seen above means an echo arriving from another peer is dropped + # as a duplicate, so cycles in the peer graph terminate. + if msg_type in ("tx", "block") and status == ValidationStatus.VALID: + await self._broadcast_raw( + {"type": msg_type, "data": payload}, exclude=peer_addr + ) + elif msg[0] == "MALFORMED": # JSON parse failure signalled from the Trio thread. peer_addr = msg[1] @@ -262,7 +292,14 @@ async def drain(self): pass async def _trio_main(self): host = new_host() listen_addr = Multiaddr(f"/ip4/{self.host_addr}/tcp/{self.port}") - await host.get_network().listen(listen_addr) + # Swarm.listen() waits on a nursery that only exists while the swarm runs as + # a background service, so listening must go through host.run(). Calling + # host.get_network().listen() directly blocks forever and the node comes up + # with no networking at all. + async with host.run(listen_addrs=[listen_addr]): + await self._serve(host, listen_addr) + + async def _serve(self, host, listen_addr): print(f" Network Multiaddr: {listen_addr}/p2p/{host.get_id().to_string()}") streams = [] @@ -283,6 +320,7 @@ async def stream_handler(stream): streams.append(stream) with self._peer_count_lock: self._peer_count += 1 + self._peer_ids.add(peer_id) self._to_asyncio.put(("PEER_CONNECTED", None)) try: @@ -313,10 +351,11 @@ async def stream_handler(stream): streams.remove(stream) with self._peer_count_lock: self._peer_count -= 1 + self._peer_ids.discard(peer_id) host.set_stream_handler(PROTOCOL_ID, stream_handler) - async def check_queue(): + async def check_queue(nursery): while True: try: while not self._to_trio.empty(): @@ -329,12 +368,18 @@ async def check_queue(): info = info_from_p2p_addr(maddr) await host.connect(info) stream = await host.new_stream(info.peer_id, [PROTOCOL_ID]) - host.get_network().nursery.start_soon(stream_handler, stream) + # Read the outbound stream in our own nursery: the swarm + # exposes no nursery to borrow, and without this the + # dialing side never registers the peer or reads from it. + nursery.start_soon(stream_handler, stream) except Exception as e: logger.error(f"Dial error: {e}") elif cmd == "BROADCAST": - msg = (canonical_json_dumps(arg) + "\n").encode() + payload, exclude = arg + msg = (canonical_json_dumps(payload) + "\n").encode() for s in list(streams): + if exclude and f"peer:{s.muxed_conn.peer_id}" == exclude: + continue try: await s.write(msg) except Exception: @@ -361,13 +406,14 @@ async def check_queue(): streams.remove(s) with self._peer_count_lock: self._peer_count -= 1 + self._peer_ids.discard(str(s.muxed_conn.peer_id)) except Exception: pass await trio.sleep(0.1) async with trio.open_nursery() as nursery: async def run_monitor(): - if await check_queue(): + if await check_queue(nursery): await host.close() nursery.cancel_scope.cancel() nursery.start_soon(run_monitor) diff --git a/tests/test_protocol_hardening.py b/tests/test_protocol_hardening.py index d34de14..a7786a2 100644 --- a/tests/test_protocol_hardening.py +++ b/tests/test_protocol_hardening.py @@ -1,10 +1,13 @@ +import asyncio import unittest from nacl.encoding import HexEncoder from nacl.signing import SigningKey from minichain import Block, Mempool, P2PNetwork, State, Transaction, calculate_hash +from minichain.node_config import SEEN_CACHE_MAX from minichain.serialization import canonical_json_dumps +from minichain.validators import ValidationStatus class TestDeterministicConsensus(unittest.TestCase): @@ -162,3 +165,90 @@ async def test_duplicate_tx_and_block_detection(self): self.assertFalse(network._is_duplicate("block", block_message["data"])) network._mark_seen("block", block_message["data"]) self.assertTrue(network._is_duplicate("block", block_message["data"])) + + + +class TestP2PRelayAndPeers(unittest.IsolatedAsyncioTestCase): + """ + A node relays content it accepts, so gossip travels past one hop, but never + back to the peer it came from and never anything it rejected. + """ + + SOURCE = "peer:source" + TX_DATA = { + "sender": "a" * 64, "receiver": "b" * 64, "amount": 1, "nonce": 0, + "data": None, "timestamp": 123, "signature": "c" * 128, + } + + def _network(self, status): + network = P2PNetwork(malformed_threshold=1000, failed_threshold=1000, invalid_threshold=1000) + network.loop = asyncio.get_running_loop() + + async def handler(_data): + return status + + network.register_handler(handler) + return network + + async def _deliver(self, network, msg_type, data): + network._to_asyncio.put(("MSG", {"type": msg_type, "data": data, "_peer_addr": self.SOURCE})) + task = asyncio.create_task(network._asyncio_reader()) + for _ in range(30): + await asyncio.sleep(0.01) + if not network._to_trio.empty(): + break + task.cancel() + network._to_asyncio.put(("MALFORMED", self.SOURCE)) + try: + await task + except asyncio.CancelledError: + pass + commands = [] + while not network._to_trio.empty(): + commands.append(network._to_trio.get_nowait()) + return [arg for cmd, arg in commands if cmd == "BROADCAST"] + + async def test_accepted_tx_is_relayed_excluding_sender(self): + network = self._network(ValidationStatus.VALID) + + broadcasts = await self._deliver(network, "tx", self.TX_DATA) + + self.assertEqual(len(broadcasts), 1) + payload, exclude = broadcasts[0] + self.assertEqual(payload, {"type": "tx", "data": self.TX_DATA}) + self.assertEqual(exclude, self.SOURCE) + + async def test_rejected_content_is_never_relayed(self): + network = self._network(ValidationStatus.INVALID) + + broadcasts = await self._deliver(network, "tx", self.TX_DATA) + + self.assertEqual(broadcasts, []) + + async def test_control_messages_are_never_relayed(self): + network = self._network(None) + + broadcasts = await self._deliver(network, "hello", {"some": "payload"}) + + self.assertEqual(broadcasts, []) + + async def test_duplicate_is_relayed_only_once(self): + network = self._network(ValidationStatus.VALID) + + first = await self._deliver(network, "tx", self.TX_DATA) + second = await self._deliver(network, "tx", self.TX_DATA) + + self.assertEqual(len(first), 1) + self.assertEqual(second, []) + + +class TestSeenCacheIsBounded(unittest.TestCase): + def test_oldest_entry_is_evicted_past_the_cap(self): + network = P2PNetwork() + + for i in range(SEEN_CACHE_MAX + 10): + network._mark_seen("block", {"hash": f"h{i}"}) + + self.assertEqual(len(network._seen_block_hashes), SEEN_CACHE_MAX) + self.assertFalse(network._is_duplicate("block", {"hash": "h0"})) + self.assertTrue(network._is_duplicate("block", {"hash": f"h{SEEN_CACHE_MAX + 9}"}))