fix: make P2P networking work, and let a node mine on its own

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/<seed-peer-id> --datadir ./node2_data
 ```
 The node will automatically sync the blockchain state via the P2P network using the Fork-Choice rule.
 

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.")
 

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

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)

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}"}))