Implement block transfer logic

node.py

@@ -1,6 +1,7 @@
 #! /usr/bin/env python3
 
-import hashlib, random, socket, sys, threading, time
+import blockchain, hashlib, observer, random, socket, sys, threading, time
+from _queue import Empty
 
 DEFAULT_PORT = 62039
 
@@ -88,11 +89,11 @@ def empty_transaction_list_hash():
         current_hash = hashlib.sha256(entry).digest()
     return current_hash
 
-def send_heartbeat(node, peer):
+def send_heartbeat(node, peer, b):
     protocol_version = 2 * b"\0"
     capable_version = 2 * b"\0"
     msg_type = b"\0"
-    difficulty_sum = 32 * b"\0"
+    difficulty_sum = b.get_second_last_difficulty_sum().to_bytes(32, "big")
     hash_value = empty_transaction_list_hash()
 
     if peer.partner is None:
@@ -117,7 +118,7 @@ def define_partnership(peers):
     if pairing_count % 2 == 1:
         peers_to_pair[-1].partner = None
 
-def heartbeat(node):
+def heartbeat(node, b):
     while True:
         heartbeats, partners = node.get_events()
         heartbeats = set(heartbeats)
@@ -169,11 +170,61 @@ def heartbeat(node):
         define_partnership(node.peers)
         for i, peer in enumerate(node.peers):
             wait_until(start_time + 60 * (i+1) / peer_count)
-            send_heartbeat(node, peer)
+            send_heartbeat(node, peer, b)
         if len(node.peers) == 0:
             time.sleep(60)
 
-def receiver(node):
+class NoReponseException(Exception):
+    pass
+
+def request_retry(node, addr, request, subscription, condition):
+    for _ in range(10):
+        node.node_socket.sendto(request, addr)
+        try:
+            while True:
+                response = subscription.receive(1)
+                if condition(response):
+                    break
+            return response
+        except Empty:
+            pass
+    raise NoReponseException()
+
+def transfer_block(addr, node, receive_observer, b):
+    try:
+        block_list = []
+        request_block_hash = 32 * b"\0"
+        subscription = receive_observer.listen((addr[0:2], "block transfer"))
+        while True:
+            if request_block_hash != 32 * b"\0":
+                existing_block = b.get_block(request_block_hash)
+                if existing_block is not None:
+                    if existing_block.valid:
+                        break
+                    request_block_hash = existing_block.previous_hash
+                    if request_block_hash == 32*b"\0":
+                        break
+                    continue
+            request = b"\0\0\0\0\x01" + request_block_hash
+            def condition(response_hash):
+                if request_block_hash == 32*b"\0":
+                    return True
+                return response_hash == request_block_hash
+            block_hash = request_retry(node, addr, request, subscription, condition)
+            block_list.append(block_hash)
+            request_block_hash = b.get_block(block_hash).previous_hash
+            if request_block_hash == 32*b"\0":
+                break
+        for block_hash in reversed(block_list):
+            if not b.get_block(block_hash).validate(b):
+                return
+            if b.set_latest_block(block_hash):
+                log("Got a new block")
+    except NoReponseException:
+        pass
+
+def receiver(node, b):
+    receive_observer = observer.Observer()
     while True:
         msg, addr = node.node_socket.recvfrom(4096)
         sender = describe(addr[0], addr[1])
@@ -202,6 +253,36 @@ def receiver(node):
                 partner_ip = socket.inet_ntop(socket.AF_INET6, partner_info[0:16])
                 partner_port = int.from_bytes(partner_info[16:18], "big")
                 node.add_partner((partner_ip, partner_port))
+            contained_difficulty_sum = int.from_bytes(msg[5:37])
+            my_difficulty_sum = b.get_second_last_difficulty_sum()
+            if contained_difficulty_sum > my_difficulty_sum:
+                log("beginning a block transfer ...")
+                threading.Thread(target = transfer_block, args=(addr, node, receive_observer, b)).start()
+        elif msg_type == 1:
+            # block request
+            if msg_len != 37:
+                log(f"Got a block request of wrong length ({msg_len} bytes from {sender}, but expected 37 bytes)")
+            block_hash = msg[5:37]
+            if block_hash == 32 * b"\0":
+                block_to_send = b.get_latest_block()
+            else:
+                block_to_send = b.get_block(block_hash)
+            if block_to_send is None:
+                continue
+            block_raw = block_to_send.get_block_raw()
+            response_msg = b"\0\0\0\0\x02" + block_raw
+            node.node_socket.sendto(response_msg, addr)
+        elif msg_type == 2:
+            # block transfer
+            if msg_len != 297:
+                log(f"Got a block transfer of wrong length ({msg_len} bytes from {sender}, but expected 297 bytes)")
+            block_raw = msg[5:297]
+            new_block = b.add_block(block_raw)
+            block_hash = new_block.own_hash
+            if new_block.validate(b) and b.set_latest_block(block_hash):
+                log("Got a new block")
+            identifier = (addr[0:2], "block transfer")
+            receive_observer.publish(identifier, block_hash)
         else:
             log(f"Got a udp message of unknown type from {sender}. (type {msg_type})")
 
@@ -223,8 +304,9 @@ def main():
     log("Node is ready")
 
     node = Node(node_socket, peers)
-    heartbeat_thread = threading.Thread(target = heartbeat, args = (node,))
-    receiving_thread = threading.Thread(target = receiver, args = (node,))
+    b = blockchain.Blockchain()
+    heartbeat_thread = threading.Thread(target = heartbeat, args = (node, b))
+    receiving_thread = threading.Thread(target = receiver, args = (node, b))
     heartbeat_thread.start()
     receiving_thread.start()
     heartbeat_thread.join()