diff --git a/config.ini b/config.ini index de684fd5..ac5e515f 100644 --- a/config.ini +++ b/config.ini @@ -1,6 +1,5 @@ [squeaknode] price_msat=1000000 -sync_interval_s=10 [lnd] host=localhost diff --git a/itests/config.ini b/itests/config.ini index e6beb7cf..ff0b9e01 100644 --- a/itests/config.ini +++ b/itests/config.ini @@ -21,3 +21,6 @@ password=postgres [admin] rpc_enabled=true + +[sync] +enabled=false diff --git a/squeaknode/admin/squeak_admin_server_handler.py b/squeaknode/admin/squeak_admin_server_handler.py index 30015a48..1213e2dd 100644 --- a/squeaknode/admin/squeak_admin_server_handler.py +++ b/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, diff --git a/squeaknode/config/config.py b/squeaknode/config/config.py index 1f652085..1a2360c9 100644 --- a/squeaknode/config/config.py +++ b/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( diff --git a/squeaknode/main.py b/squeaknode/main.py index 9b8038bc..a7e3838e 100644 --- a/squeaknode/main.py +++ b/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) diff --git a/squeaknode/node/squeak_controller.py b/squeaknode/node/squeak_controller.py index 4defff4a..78362eb1 100644 --- a/squeaknode/node/squeak_controller.py +++ b/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 diff --git a/squeaknode/node/squeak_node.py b/squeaknode/node/squeak_node.py index 16844814..bf124d6a 100644 --- a/squeaknode/node/squeak_node.py +++ b/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() diff --git a/squeaknode/node/squeak_sync_status.py b/squeaknode/node/squeak_sync_status.py deleted file mode 100644 index 136bcb2a..00000000 --- a/squeaknode/node/squeak_sync_status.py +++ /dev/null @@ -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 diff --git a/squeaknode/sync/__init__.py b/squeaknode/sync/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/squeaknode/node/network_sync.py b/squeaknode/sync/network_sync.py similarity index 65% rename from squeaknode/node/network_sync.py rename to squeaknode/sync/network_sync.py index 292992a8..5efbd927 100644 --- a/squeaknode/node/network_sync.py +++ b/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) diff --git a/squeaknode/node/network_task.py b/squeaknode/sync/network_task.py similarity index 78% rename from squeaknode/node/network_task.py rename to squeaknode/sync/network_task.py index 839ee321..272e97a0 100644 --- a/squeaknode/node/network_task.py +++ b/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) diff --git a/squeaknode/node/peer_connection.py b/squeaknode/sync/peer_connection.py similarity index 100% rename from squeaknode/node/peer_connection.py rename to squeaknode/sync/peer_connection.py diff --git a/squeaknode/node/peer_task.py b/squeaknode/sync/peer_task.py similarity index 96% rename from squeaknode/node/peer_task.py rename to squeaknode/sync/peer_task.py index f165da86..63081155 100644 --- a/squeaknode/node/peer_task.py +++ b/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 diff --git a/squeaknode/node/squeak_peer_sync_worker.py b/squeaknode/sync/squeak_peer_sync_worker.py similarity index 70% rename from squeaknode/node/squeak_peer_sync_worker.py rename to squeaknode/sync/squeak_peer_sync_worker.py index 8f1d763c..39a230f8 100644 --- a/squeaknode/node/squeak_peer_sync_worker.py +++ b/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: diff --git a/squeaknode/sync/squeak_sync_status.py b/squeaknode/sync/squeak_sync_status.py new file mode 100644 index 00000000..19f7c27f --- /dev/null +++ b/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