Add sync package (#543)

* Move network sync modules to new sync package

* Run sync worker thread from main module

* Got sync worker working in admin handler

* Use squeak_controller in sync_controller

* Use squeak_controller in peer task

* Remove old commented args replaced by squeak_controller

* Use squeak_controller method to get best block height

* Improve constructor for network sync task

* Improve constructor for network sync task

config.ini

@@ -1,6 +1,5 @@
 [squeaknode]
 price_msat=1000000
-sync_interval_s=10
 
 [lnd]
 host=localhost

itests/config.ini

@@ -21,3 +21,6 @@ password=postgres
 
 [admin]
 rpc_enabled=true
+
+[sync]
+enabled=false

squeaknode/admin/squeak_admin_server_handler.py

@@ -13,6 +13,7 @@ from squeaknode.admin.util import squeak_profile_to_message
 from squeaknode.admin.util import sync_result_to_message
 from squeaknode.lightning.lnd_lightning_client import LNDLightningClient
 from squeaknode.node.squeak_controller import SqueakController
+from squeaknode.sync.squeak_sync_status import SqueakSyncController
 
 logger = logging.getLogger(__name__)
 
@@ -24,9 +25,11 @@ class SqueakAdminServerHandler(object):
         self,
         lightning_client: LNDLightningClient,
         squeak_controller: SqueakController,
+        sync_controller: SqueakSyncController,
     ):
         self.lightning_client = lightning_client
         self.squeak_controller = squeak_controller
+        self.sync_controller = sync_controller
 
     def handle_lnd_get_info(self, request):
         logger.info("Handle lnd get info")
@@ -380,7 +383,8 @@ class SqueakAdminServerHandler(object):
 
     def handle_sync_squeaks(self, request):
         logger.info("Handle sync squeaks")
-        sync_result = self.squeak_controller.sync_squeaks()
+        # sync_result = self.squeak_controller.sync_squeaks()
+        sync_result = self.sync_controller.sync_timeline()
         sync_result_msg = sync_result_to_message(sync_result)
         return squeak_admin_pb2.SyncSqueaksReply(
             sync_result=sync_result_msg,
@@ -389,7 +393,8 @@ class SqueakAdminServerHandler(object):
     def handle_sync_squeak(self, request):
         squeak_hash = request.squeak_hash
         logger.info("Handle download squeak with hash: {}".format(squeak_hash))
-        sync_result = self.squeak_controller.sync_squeak(squeak_hash)
+        # sync_result = self.squeak_controller.sync_squeak(squeak_hash)
+        sync_result = self.sync_controller.sync_single_squeak(squeak_hash)
         sync_result_msg = sync_result_to_message(sync_result)
         return squeak_admin_pb2.SyncSqueakReply(
             sync_result=sync_result_msg,

squeaknode/config/config.py

@@ -28,6 +28,7 @@ DEFAULT_LND_TLS_CERT_NAME = "tls.cert"
 DEFAULT_LND_MACAROON_NAME = "admin.macaroon"
 DEFAULT_LND_DIR_PATH = str(Path.home() / DEFAULT_LND_DIR)
 DEFAULT_LND_HOST = "localhost"
+DEFAULT_SYNC_INTERVAL_S = 10
 
 
 class Config:
@@ -81,11 +82,12 @@ class Config:
             "squeaknode_max_squeaks_per_address_per_hour"
         ] = self._get_squeaknode_max_squeaks_per_address_per_hour()
         self._configs["squeaknode_sqk_dir"] = self._get_squeaknode_sqk_dir()
-        self._configs[
-            "squeaknode_sync_interval_s"
-        ] = self._get_squeaknode_sync_interval_s()
         self._configs["squeaknode_log_level"] = self._get_squeaknode_log_level()
 
+        # sync
+        self._configs["sync_enabled"] = self._get_sync_enabled()
+        self._configs["sync_interval_s"] = self._get_sync_interval_s()
+
         # db
         self._configs["db_connection_string"] = self._get_db_connection_string()
 
@@ -250,10 +252,14 @@ class Config:
             "squeaknode", "sqk_dir", fallback=DEFAULT_SQK_DIR_PATH
         )
 
-    def _get_squeaknode_sync_interval_s(self):
+    def _get_sync_enabled(self):
+        return environ.get("SQUEAKNODE_SYNC_ENABLED") or self.parser.getboolean(
+            "sync", "enabled", fallback=True)
+
+    def _get_sync_interval_s(self):
         return int(
             environ.get("SQUEAKNODE_SYNC_INTERVAL_S") or 0
-        ) or self.parser.getint("squeaknode", "sync_interval_s", fallback=None)
+        ) or self.parser.getint("sync", "interval_s", fallback=DEFAULT_SYNC_INTERVAL_S)
 
     def _get_squeaknode_log_level(self):
         return environ.get("LOG_LEVEL") or self.parser.get(

squeaknode/main.py

@@ -20,6 +20,8 @@ from squeaknode.node.squeak_controller import SqueakController
 from squeaknode.node.squeak_node import SqueakNode
 from squeaknode.server.squeak_server_handler import SqueakServerHandler
 from squeaknode.server.squeak_server_servicer import SqueakServerServicer
+from squeaknode.sync.squeak_peer_sync_worker import SqueakPeerSyncWorker
+from squeaknode.sync.squeak_sync_status import SqueakSyncController
 
 logger = logging.getLogger(__name__)
 
@@ -69,14 +71,22 @@ def load_admin_web_server(config, handler) -> SqueakAdminWebServer:
     )
 
 
+def load_sync_worker(config, sync_controller) -> SqueakPeerSyncWorker:
+    return SqueakPeerSyncWorker(
+        sync_controller,
+        config.sync_interval_s,
+    )
+
+
 def load_handler(squeak_controller):
     return SqueakServerHandler(squeak_controller)
 
 
-def load_admin_handler(lightning_client, squeak_controller):
+def load_admin_handler(lightning_client, squeak_controller, sync_controller):
     return SqueakAdminServerHandler(
         lightning_client,
         squeak_controller,
+        sync_controller,
     )
 
 
@@ -134,6 +144,16 @@ def start_admin_web_server(admin_web_server):
     thread.start()
 
 
+def start_sync_worker(sync_worker):
+    logger.info("Starting sync worker...")
+    thread = threading.Thread(
+        target=sync_worker.start_running,
+        args=(),
+    )
+    thread.daemon = True
+    thread.start()
+
+
 def parse_args():
     parser = argparse.ArgumentParser(
         description="squeaknode runs a node using squeak protocol. ",
@@ -208,11 +228,16 @@ def run_server(config):
     # Create and start the squeak node
     squeak_node = SqueakNode(
         squeak_controller,
-        config.squeaknode_sync_interval_s,
+        # config.sync_interval_s,
     )
     squeak_node.start_running()
 
-    admin_handler = load_admin_handler(lightning_client, squeak_controller)
+    sync_controller = SqueakSyncController(
+        squeak_controller,
+    )
+
+    admin_handler = load_admin_handler(
+        lightning_client, squeak_controller, sync_controller)
 
     # start admin rpc server
     if config.admin_rpc_enabled:
@@ -224,6 +249,11 @@ def run_server(config):
         admin_web_server = load_admin_web_server(config, admin_handler)
         start_admin_web_server(admin_web_server)
 
+    # start sync worker
+    if config.sync_enabled:
+        sync_worker = load_sync_worker(config, sync_controller)
+        start_sync_worker(sync_worker)
+
     # start rpc server
     handler = load_handler(squeak_controller)
     server = load_rpc_server(config, handler)

squeaknode/node/squeak_controller.py

@@ -21,8 +21,8 @@ from squeaknode.node.squeak_block_verifier import SqueakBlockVerifier
 from squeaknode.node.squeak_maker import SqueakMaker
 from squeaknode.node.squeak_rate_limiter import SqueakRateLimiter
 from squeaknode.node.squeak_store import SqueakStore
-from squeaknode.node.squeak_sync_status import SqueakSyncController
 from squeaknode.node.squeak_whitelist import SqueakWhitelist
+from squeaknode.sync.squeak_sync_status import SqueakSyncController
 
 logger = logging.getLogger(__name__)
 
@@ -58,12 +58,12 @@ class SqueakController:
             self.squeak_rate_limiter,
             self.squeak_whitelist,
         )
-        self.squeak_sync_controller = SqueakSyncController(
-            self.blockchain_client,
-            self.squeak_store,
-            self.squeak_db,
-            self.lightning_client,
-        )
+        # self.squeak_sync_controller = SqueakSyncController(
+        #     self.blockchain_client,
+        #     self.squeak_store,
+        #     self.squeak_db,
+        #     self.lightning_client,
+        # )
         self.sent_offers_verifier = SentOffersVerifier(
             self.squeak_db,
             self.lightning_client,
@@ -341,12 +341,12 @@ class SqueakController:
             secret_key,
         )
 
-    def sync_squeaks(self):
-        return self.squeak_sync_controller.sync_timeline()
+    # def sync_squeaks(self):
+    #     return self.squeak_sync_controller.sync_timeline()
 
-    def sync_squeak(self, squeak_hash):
-        peers = self.squeak_db.get_peers()
-        return self.squeak_sync_controller.sync_single_squeak(squeak_hash, peers)
+    # def sync_squeak(self, squeak_hash):
+    #     peers = self.squeak_db.get_peers()
+    #     return self.squeak_sync_controller.sync_single_squeak(squeak_hash, peers)
 
     def get_sent_payments(self):
         return self.squeak_db.get_sent_payments()
@@ -386,3 +386,7 @@ class SqueakController:
         ) as client:
             for payment in client.get_received_payments():
                 yield payment
+
+    def get_best_block_height(self):
+        block_info = self.blockchain_client.get_best_block_info()
+        return block_info.block_height

squeaknode/node/squeak_node.py

@@ -2,7 +2,6 @@ import logging
 
 from squeaknode.node.sent_offers_worker import SentOffersWorker
 from squeaknode.node.squeak_offer_expiry_worker import SqueakOfferExpiryWorker
-from squeaknode.node.squeak_peer_sync_worker import SqueakPeerSyncWorker
 
 logger = logging.getLogger(__name__)
 
@@ -11,13 +10,13 @@ class SqueakNode:
     def __init__(
         self,
         squeak_controller,
-        sync_interval_s,
+        # sync_interval_s,
     ):
         self.squeak_controller = squeak_controller
-        self.squeak_peer_sync_worker = SqueakPeerSyncWorker(
-            self.squeak_controller,
-            sync_interval_s,
-        )
+        # self.squeak_peer_sync_worker = SqueakPeerSyncWorker(
+        #     self.squeak_controller,
+        #     sync_interval_s,
+        # )
         self.squeak_offer_expiry_worker = SqueakOfferExpiryWorker(
             self.squeak_controller,
         )
@@ -26,6 +25,6 @@ class SqueakNode:
         )
 
     def start_running(self):
-        self.squeak_peer_sync_worker.start_running()
+        # self.squeak_peer_sync_worker.start_running()
         self.squeak_offer_expiry_worker.start_running()
         self.sent_offers_worker.start_running()

squeaknode/node/squeak_sync_status.py

@@ -1,54 +0,0 @@
-import logging
-
-from squeaknode.node.network_sync import NetworkSync
-from squeaknode.node.network_task import SingleSqueakNetworkSyncTask
-from squeaknode.node.network_task import TimelineNetworkSyncTask
-
-logger = logging.getLogger(__name__)
-
-
-LOOKUP_BLOCK_INTERVAL = 1008  # 1 week
-
-
-class SqueakSyncController:
-    def __init__(self, blockchain_client, squeak_store, squeak_db, lightning_client):
-        self.blockchain_client = blockchain_client
-        self.squeak_store = squeak_store
-        self.squeak_db = squeak_db
-        self.lightning_client = lightning_client
-        self.network_sync = NetworkSync(
-            squeak_store, squeak_db, lightning_client)
-
-    def sync_timeline(self):
-        try:
-            block_info = self.blockchain_client.get_best_block_info()
-            block_height = block_info.block_height
-        except Exception:
-            logger.error(
-                "Failed to sync because unable to get blockchain info.", exc_info=False
-            )
-            return
-        min_block = block_height - LOOKUP_BLOCK_INTERVAL
-        max_block = block_height
-        peers = self.squeak_db.get_peers()
-        dowload_timeline_task = TimelineNetworkSyncTask(
-            self.network_sync,
-            min_block,
-            max_block,
-        )
-        network_sync_result = dowload_timeline_task.sync(peers)
-        logger.info("Upload network_sync_result: {}".format(
-            network_sync_result))
-        return network_sync_result
-
-    def sync_single_squeak(self, squeak_hash, peers):
-        timeline_sync_task = SingleSqueakNetworkSyncTask(
-            self.network_sync,
-            squeak_hash,
-        )
-        network_sync_result = timeline_sync_task.sync(peers)
-        logger.info(
-            "Download single squeak network_sync_result: {}".format(
-                network_sync_result)
-        )
-        return network_sync_result

squeaknode/sync/network_sync.py

@@ -1,7 +1,7 @@
 import logging
 
-from squeaknode.node.peer_connection import PeerConnection
-from squeaknode.node.peer_task import PeerSyncTask
+from squeaknode.sync.peer_connection import PeerConnection
+from squeaknode.sync.peer_task import PeerSyncTask
 
 logger = logging.getLogger(__name__)
 
@@ -9,13 +9,9 @@ logger = logging.getLogger(__name__)
 class NetworkSync:
     def __init__(
         self,
-        squeak_store,
-        squeak_db,
-        lightning_client,
+        squeak_controller,
     ):
-        self.squeak_store = squeak_store
-        self.squeak_db = squeak_db
-        self.lightning_client = lightning_client
+        self.squeak_controller = squeak_controller
 
     def sync_timeline(self, peer, min_block, max_block):
         if not peer.downloading:
@@ -23,9 +19,7 @@ class NetworkSync:
         peer_connection = PeerConnection(peer)
         peer_sync_task = PeerSyncTask(
             peer_connection,
-            self.squeak_store,
-            self.squeak_db,
-            self.lightning_client,
+            self.squeak_controller,
         )
         if peer.uploading:
             peer_sync_task.upload(min_block, max_block)
@@ -38,9 +32,7 @@ class NetworkSync:
         peer_connection = PeerConnection(peer)
         peer_sync_task = PeerSyncTask(
             peer_connection,
-            self.squeak_store,
-            self.squeak_db,
-            self.lightning_client,
+            self.squeak_controller,
         )
         if peer.uploading:
             peer_sync_task.upload_single_squeak(squeak_hash)

squeaknode/sync/network_task.py

@@ -5,6 +5,8 @@ from dataclasses import dataclass
 from typing import Any
 from typing import List
 
+from squeaknode.sync.network_sync import NetworkSync
+
 logger = logging.getLogger(__name__)
 
 
@@ -25,15 +27,17 @@ class NetworkSyncResult:
 class NetworkSyncTask:
     def __init__(
         self,
-        network_sync,
+        squeak_controller,
     ):
-        self.network_sync = network_sync
+        self.squeak_controller = squeak_controller
         self.queue = queue.Queue()
 
-    def sync(self, peers):
+    def sync(self):
+        peers = self.squeak_controller.get_peers()
         logger.debug(
-            "Network sync for class {}".format(
+            "Network sync for class {} with peers: {}".format(
                 self.__class__,
+                peers,
             )
         )
         run_sync_thread = threading.Thread(
@@ -88,26 +92,32 @@ class NetworkSyncTask:
 class TimelineNetworkSyncTask(NetworkSyncTask):
     def __init__(
         self,
-        network_sync,
+        squeak_controller,
         min_block,
         max_block,
     ):
-        super().__init__(network_sync)
+        super().__init__(squeak_controller)
         self.min_block = min_block
         self.max_block = max_block
 
     def sync_peer(self, peer):
-        self.network_sync.sync_timeline(peer, self.min_block, self.max_block)
+        network_sync = NetworkSync(
+            self.squeak_controller,
+        )
+        network_sync.sync_timeline(peer, self.min_block, self.max_block)
 
 
 class SingleSqueakNetworkSyncTask(NetworkSyncTask):
     def __init__(
         self,
-        network_sync,
+        squeak_controller,
         squeak_hash,
     ):
-        super().__init__(network_sync)
+        super().__init__(squeak_controller)
         self.squeak_hash = squeak_hash
 
     def sync_peer(self, peer):
-        self.network_sync.sync_single_squeak(peer, self.squeak_hash)
+        network_sync = NetworkSync(
+            self.squeak_controller,
+        )
+        network_sync.sync_single_squeak(peer, self.squeak_hash)

squeaknode/sync/peer_task.py

@@ -10,19 +10,27 @@ class PeerSyncTask:
     def __init__(
         self,
         peer_connection,
-        squeak_store,
-        squeak_db,
-        lightning_client,
+        squeak_controller,
     ):
         self.peer_connection = peer_connection
-        self.squeak_store = squeak_store
-        self.squeak_db = squeak_db
-        self.lightning_client = lightning_client
+        self.squeak_controller = squeak_controller
 
     @property
     def peer(self):
         return self.peer_connection.peer
 
+    @property
+    def squeak_store(self):
+        return self.squeak_controller.squeak_store
+
+    @property
+    def squeak_db(self):
+        return self.squeak_controller.squeak_db
+
+    @property
+    def lightning_client(self):
+        return self.squeak_controller.lightning_client
+
     @property
     def peer_client(self):
         return self.peer_connection.peer_client

squeaknode/sync/squeak_peer_sync_worker.py

@@ -7,15 +7,16 @@ logger = logging.getLogger(__name__)
 class SqueakPeerSyncWorker:
     def __init__(
         self,
-        squeak_controller,
+        squeak_sync_controller,
         sync_interval_s,
     ):
-        self.squeak_controller = squeak_controller
+        self.squeak_sync_controller = squeak_sync_controller
         self.sync_interval_s = sync_interval_s
 
     def sync_timeline(self):
         logger.info("Syncing timeline with peers...")
-        self.squeak_controller.sync_squeaks()
+        # self.squeak_sync_controller.sync_squeaks()
+        self.squeak_sync_controller.sync_timeline()
 
     def start_running(self):
         if self.sync_interval_s:

squeaknode/sync/squeak_sync_status.py

@@ -0,0 +1,47 @@
+import logging
+
+from squeaknode.sync.network_sync import NetworkSync
+from squeaknode.sync.network_task import SingleSqueakNetworkSyncTask
+from squeaknode.sync.network_task import TimelineNetworkSyncTask
+
+logger = logging.getLogger(__name__)
+
+
+LOOKUP_BLOCK_INTERVAL = 1008  # 1 week
+
+
+class SqueakSyncController:
+    def __init__(self, squeak_controller):
+        self.squeak_controller = squeak_controller
+
+    def sync_timeline(self):
+        try:
+            block_height = self.squeak_controller.get_best_block_height()
+        except Exception:
+            logger.error(
+                "Failed to sync timeline because unable to get best block height.", exc_info=False
+            )
+            return
+        min_block = block_height - LOOKUP_BLOCK_INTERVAL
+        max_block = block_height
+        dowload_timeline_task = TimelineNetworkSyncTask(
+            self.squeak_controller,
+            min_block,
+            max_block,
+        )
+        network_sync_result = dowload_timeline_task.sync()
+        logger.info("Upload network_sync_result: {}".format(
+            network_sync_result))
+        return network_sync_result
+
+    def sync_single_squeak(self, squeak_hash):
+        timeline_sync_task = SingleSqueakNetworkSyncTask(
+            self.squeak_controller,
+            squeak_hash,
+        )
+        network_sync_result = timeline_sync_task.sync()
+        logger.info(
+            "Download single squeak network_sync_result: {}".format(
+                network_sync_result)
+        )
+        return network_sync_result