From e66dbdeebdecf926298bf7f4b53ccb800c5addbb Mon Sep 17 00:00:00 2001 From: Jonathan Zernik Date: Sat, 31 Oct 2020 21:48:25 -0400 Subject: [PATCH] Create peer sync task class (#350) * Create peer sync task class * Convery peer upload to subclass of peer sync tsk * Use network task class for download task * Use network task class for single squeak download * comment unused network sync methods * Remove old network controller methods * Move all peer task methods into peer task class * Move get offer method into peer sync class * Delete get offer class * Upload timeline squeaks using network task subclass * Fix upload squeaks to only upload unlocked squeaks * Remove unused synctask class --- .../admin/squeak_admin_server_handler.py | 2 +- squeaknode/node/network_task.py | 129 +++++++ squeaknode/node/peer_download.py | 154 --------- squeaknode/node/peer_get_offer.py | 147 -------- squeaknode/node/peer_task.py | 313 +++++++++++++++++ squeaknode/node/peer_upload.py | 123 +++---- squeaknode/node/squeak_node.py | 1 - squeaknode/node/squeak_sync_status.py | 324 ++++++++++-------- 8 files changed, 663 insertions(+), 530 deletions(-) create mode 100644 squeaknode/node/network_task.py delete mode 100644 squeaknode/node/peer_download.py delete mode 100644 squeaknode/node/peer_get_offer.py create mode 100644 squeaknode/node/peer_task.py diff --git a/squeaknode/admin/squeak_admin_server_handler.py b/squeaknode/admin/squeak_admin_server_handler.py index 6b02ed28..0512633e 100644 --- a/squeaknode/admin/squeak_admin_server_handler.py +++ b/squeaknode/admin/squeak_admin_server_handler.py @@ -370,7 +370,7 @@ class SqueakAdminServerHandler(object): ) def handle_sync_squeaks(self, request): - logger.info("Handle get sync squeaks") + logger.info("Handle sync squeaks") self.squeak_node.sync_squeaks() return squeak_admin_pb2.SyncSqueaksReply() diff --git a/squeaknode/node/network_task.py b/squeaknode/node/network_task.py new file mode 100644 index 00000000..4bc84b36 --- /dev/null +++ b/squeaknode/node/network_task.py @@ -0,0 +1,129 @@ +import logging +import threading + +from squeaknode.network.peer_client import PeerClient +from squeaknode.node.peer_task import PeerSyncTask + +logger = logging.getLogger(__name__) + + +LOOKUP_BLOCK_INTERVAL = 1008 # 1 week + + +class NetworkSyncTask: + def __init__( + self, + squeak_store, + postgres_db, + lightning_client, + ): + self.squeak_store = squeak_store + self.postgres_db = postgres_db + self.lightning_client = lightning_client + + def sync(self, peers): + logger.info("Network sync for class {}".format( + self.__class__, + )) + for peer in peers: + sync_peer_thread = threading.Thread( + target=self.sync_peer, + args=(peer,), + ) + sync_peer_thread.start() + + def sync_peer(self, peer): + # peer_upload = PeerUpload( + # peer, + # self.squeak_store, + # self.postgres_db, + # self.lightning_client, + # ) + # try: + # logger.debug("Trying to upload to peer: {}".format(peer)) + # with self.UploadingContextManager( + # peer, peer_upload, self.squeak_sync_status + # ) as uploading_manager: + # peer_upload.upload(block_height) + # except Exception as e: + # logger.error("Upload from peer failed.", exc_info=True) + pass + + +class DownloadTimelineNetworkSyncTask(NetworkSyncTask): + def __init__( + self, + squeak_store, + postgres_db, + lightning_client, + block_height, + ): + super().__init__(squeak_store, postgres_db, lightning_client) + self.block_height = block_height + + def sync_peer(self, peer): + if not peer.downloading: + return + peer_sync_task = PeerSyncTask( + peer, + self.squeak_store, + self.postgres_db, + self.lightning_client, + ) + try: + logger.info("Trying to download with block height: {}".format(self.block_height)) + peer_sync_task.download(self.block_height) + except Exception as e: + logger.error("Download from peer failed.", exc_info=True) + +class UploadTimelineNetworkSyncTask(NetworkSyncTask): + def __init__( + self, + squeak_store, + postgres_db, + lightning_client, + block_height, + ): + super().__init__(squeak_store, postgres_db, lightning_client) + self.block_height = block_height + + def sync_peer(self, peer): + if not peer.uploading: + return + peer_sync_task = PeerSyncTask( + peer, + self.squeak_store, + self.postgres_db, + self.lightning_client, + ) + try: + logger.info("Trying to upload with block height: {}".format(self.block_height)) + peer_sync_task.upload(self.block_height) + except Exception as e: + logger.error("Upload from peer failed.", exc_info=True) + +class SingleSqueakNetworkSyncTask(NetworkSyncTask): + def __init__( + self, + squeak_store, + postgres_db, + lightning_client, + squeak_hash, + ): + super().__init__(squeak_store, postgres_db, lightning_client) + self.squeak_hash = squeak_hash + + def sync_peer(self, peer): + if not peer.downloading: + return + peer_sync_task = PeerSyncTask( + peer, + self.squeak_store, + self.postgres_db, + self.lightning_client, + ) + try: + logger.info("Trying to download single squeak {}".format(self.squeak_hash)) + peer_sync_task.download_single_squeak(self.squeak_hash) + except Exception as e: + logger.error("Download single squeak from peer failed.", exc_info=True) diff --git a/squeaknode/node/peer_download.py b/squeaknode/node/peer_download.py deleted file mode 100644 index 9140173a..00000000 --- a/squeaknode/node/peer_download.py +++ /dev/null @@ -1,154 +0,0 @@ -import logging -import threading - -from squeaknode.network.peer_client import PeerClient -from squeaknode.node.peer_get_offer import PeerGetOffer - -logger = logging.getLogger(__name__) - - -LOOKUP_BLOCK_INTERVAL = 1008 # 1 week - - -class PeerDownload: - def __init__( - self, - peer, - squeak_store, - postgres_db, - lightning_client, - ): - self.peer = peer - self.squeak_store = squeak_store - self.postgres_db = postgres_db - self.lightning_client = lightning_client - - self.peer_client = PeerClient( - self.peer.host, - self.peer.port, - ) - - self._stop_event = threading.Event() - - def download( - self, - block_height, - lookup_block_interval=LOOKUP_BLOCK_INTERVAL, - ): - # Get list of followed addresses. - addresses = self._get_followed_addresses() - logger.debug("Followed addresses: {}".format(addresses)) - min_block = block_height - lookup_block_interval - max_block = block_height - - # Get remote hashes - remote_hashes = self._get_remote_hashes(addresses, min_block, max_block) - logger.debug("Got remote hashes: {}".format(len(remote_hashes))) - for hash in remote_hashes: - logger.debug("remote hash: {}".format(hash.hex())) - - # Get local hashes of downloaded squeaks - local_hashes = self._get_local_hashes(addresses, min_block, max_block) - logger.debug("Got local hashes: {}".format(len(local_hashes))) - for hash in local_hashes: - logger.debug("local hash: {}".format(hash.hex())) - - # Get hashes to download - hashes_to_download = set(remote_hashes) - set(local_hashes) - logger.debug("Hashes to download: {}".format(len(hashes_to_download))) - for hash in hashes_to_download: - logger.debug("hash to download: {}".format(hash.hex())) - - # Download squeaks for the hashes - # TODO: catch exception downloading individual squeak - for hash in hashes_to_download: - if self.stopped(): - return - self._download_squeak(hash) - - # Get local hashes of locked squeaks that don't have an offer from this peer. - locked_hashes = self._get_locked_hashes(addresses, min_block, max_block) - logger.debug("Got locked hashes: {}".format(len(locked_hashes))) - for hash in locked_hashes: - logger.debug("locked hash: {}".format(hash.hex())) - - # Get hashes to get offer - hashes_to_get_offer = set(remote_hashes) & set(locked_hashes) - logger.debug("Hashes to get offer: {}".format(len(hashes_to_get_offer))) - for hash in hashes_to_get_offer: - logger.debug("hash to get offer: {}".format(hash.hex())) - - # Download offers for the hashes - # TODO: catch exception downloading individual squeak - for hash in hashes_to_get_offer: - if self.stopped(): - return - self._download_offer(hash) - - def download_single_squeak(self, squeak_hash): - # Download squeak if not already present. - saved_squeak = self._get_saved_squeak(squeak_hash) - if not saved_squeak: - self._download_squeak(squeak_hash) - - # Download offer from peer if not already present. - saved_offer = self._get_saved_offer(squeak_hash) - if not saved_offer: - self._download_offer(squeak_hash) - - def stop(self): - self._stop_event.set() - - def stopped(self): - return self._stop_event.is_set() - - def _get_local_hashes(self, addresses, min_block, max_block): - return self.squeak_store.lookup_squeaks_include_locked( - addresses, - min_block, - max_block, - ) - - def _get_locked_hashes(self, addresses, min_block, max_block): - return self.squeak_store.lookup_squeaks_needing_offer( - addresses, - min_block, - max_block, - self.peer.peer_id, - ) - - def _get_remote_hashes(self, addresses, min_block, max_block): - return self.peer_client.lookup_squeaks(addresses, min_block, max_block) - - def _save_squeak(self, squeak): - self.squeak_store.save_downloaded_squeak(squeak) - - def _get_saved_squeak(self, squeak_hash): - return self.squeak_store.get_squeak(squeak_hash) - return self.postgres_db.get_offers_with_peer(squeak_hash_str) - - def _get_saved_offer(self, squeak_hash): - offers = self.postgres_db.get_offers_with_peer(squeak_hash) - for offer in offers: - if offer.peer_id == peer_id: - return offer - - def _download_squeak(self, squeak_hash): - logger.info("Downloading squeak: {} from peer: {}".format(squeak_hash.hex(), self.peer.peer_id)) - squeak = self.peer_client.get_squeak(squeak_hash) - self._save_squeak(squeak) - - def _get_followed_addresses(self): - followed_profiles = self.postgres_db.get_following_profiles() - return [profile.address for profile in followed_profiles] - - def _download_offer(self, squeak_hash): - logger.info("Downloading offer for hash: {}".format(squeak_hash.hex())) - peer_get_offer = PeerGetOffer( - self.peer, - squeak_hash, - self.squeak_store, - self.postgres_db, - self.lightning_client, - ) - peer_get_offer.get_offer() diff --git a/squeaknode/node/peer_get_offer.py b/squeaknode/node/peer_get_offer.py deleted file mode 100644 index a35e6225..00000000 --- a/squeaknode/node/peer_get_offer.py +++ /dev/null @@ -1,147 +0,0 @@ -import logging -import threading - -from squeak.core.encryption import generate_data_key - -from squeaknode.core.offer import Offer -from squeaknode.network.peer_client import PeerClient - -logger = logging.getLogger(__name__) - - -class PeerGetOffer: - def __init__(self, peer, squeak_hash, squeak_store, postgres_db, lightning_client): - self.peer = peer - self.squeak_hash = squeak_hash - self.squeak_store = squeak_store - self.postgres_db = postgres_db - self.lightning_client = lightning_client - - self.peer_client = PeerClient( - self.peer.host, - self.peer.port, - ) - - self._stop_event = threading.Event() - - def get_offer(self): - logger.info("Getting offer for squeak hash: {}".format(self.squeak_hash.hex())) - - # Get the squeak from the squeak hash - squeak = self._get_squeak() - - # Get the encryption key - encryption_key = squeak.GetEncryptionKey() - - # Create a new challenge - challenge_proof = self._generate_challenge_proof() - challenge = self._get_challenge(challenge_proof, encryption_key) - - # Download the buy offer - offer = self._download_buy_offer(challenge) - - # Check the proof - proof = offer.proof - logger.info("Proof: {}".format(proof.hex())) - logger.info("Expected proof: {}".format(challenge_proof.hex())) - if proof != challenge_proof: - raise Exception( - "Invalid offer proof: {}, expected: {}".format( - proof.hex(), - challenge_proof.hex(), - ) - ) - - # Get the decoded offer from the payment request string - decoded_offer = self._get_decoded_offer(offer) - - # Save the offer - self._save_offer(decoded_offer) - - def stop(self): - self._stop_event.set() - - def stopped(self): - return self._stop_event.is_set() - - def _get_squeak(self): - squeak_entry = self.squeak_store.get_squeak(self.squeak_hash) - return squeak_entry.squeak - - def _generate_challenge_proof(self): - return generate_data_key() - - def _get_challenge(self, challenge_proof, encryption_key): - return encryption_key.encrypt(challenge_proof) - - def _download_buy_offer(self, challenge): - logger.info( - "Downloading buy offer for squeak hash: {}".format(self.squeak_hash.hex()) - ) - offer_msg = self.peer_client.buy_squeak(self.squeak_hash, challenge) - offer = self._offer_from_msg(offer_msg) - return offer - - def _save_offer(self, offer): - logger.info("Saving offer: {}".format(offer)) - self.postgres_db.insert_offer(offer) - - def _offer_from_msg(self, offer_msg): - if not offer_msg: - return None - return Offer( - offer_id=None, - squeak_hash=offer_msg.squeak_hash, - key_cipher=offer_msg.key_cipher, - iv=offer_msg.iv, - price_msat=None, - payment_hash=offer_msg.preimage_hash, - invoice_timestamp=None, - invoice_expiry=None, - payment_request=offer_msg.payment_request, - destination=None, - node_host=offer_msg.host, - node_port=offer_msg.port, - proof=offer_msg.proof, - peer_id=self.peer.peer_id, - ) - - def _decode_payment_request(self, payment_request): - return self.lightning_client.decode_pay_req(payment_request) - - def _get_decoded_offer(self, offer): - pay_req = self._decode_payment_request(offer.payment_request) - logger.info("Decoded payment request: {}".format(pay_req)) - - price_msat = pay_req.num_msat - destination = pay_req.destination - invoice_timestamp = pay_req.timestamp - invoice_expiry = pay_req.expiry - node_host = offer.node_host or self.peer.host - node_port = offer.node_port - - logger.info("price_msat: {}".format(price_msat)) - logger.info("destination: {}".format(destination)) - logger.info("invoice_timestamp: {}".format(invoice_timestamp)) - logger.info("invoice_expiry: {}".format(invoice_expiry)) - logger.info("node_host: {}".format(node_host)) - logger.info("node_port: {}".format(node_port)) - - decoded_offer = Offer( - offer_id=offer.offer_id, - squeak_hash=offer.squeak_hash, - key_cipher=offer.key_cipher, - iv=offer.iv, - price_msat=price_msat, - payment_hash=offer.payment_hash, - invoice_timestamp=invoice_timestamp, - invoice_expiry=invoice_expiry, - payment_request=offer.payment_request, - destination=destination, - node_host=node_host, - node_port=node_port, - proof=offer.proof, - peer_id=offer.peer_id, - ) - - return decoded_offer diff --git a/squeaknode/node/peer_task.py b/squeaknode/node/peer_task.py new file mode 100644 index 00000000..cffa5689 --- /dev/null +++ b/squeaknode/node/peer_task.py @@ -0,0 +1,313 @@ +import logging +import threading + + +from squeak.core.encryption import generate_data_key + +from squeaknode.core.offer import Offer +from squeaknode.network.peer_client import PeerClient + +logger = logging.getLogger(__name__) + + +LOOKUP_BLOCK_INTERVAL = 1008 # 1 week + + +class PeerSyncTask: + def __init__( + self, + peer, + squeak_store, + postgres_db, + lightning_client, + ): + self.peer = peer + self.squeak_store = squeak_store + self.postgres_db = postgres_db + self.lightning_client = lightning_client + + self.peer_client = PeerClient( + self.peer.host, + self.peer.port, + ) + + self._stop_event = threading.Event() + + def download( + self, + block_height, + lookup_block_interval=LOOKUP_BLOCK_INTERVAL, + ): + # Get list of followed addresses. + addresses = self._get_followed_addresses() + logger.debug("Followed addresses: {}".format(addresses)) + min_block = block_height - lookup_block_interval + max_block = block_height + + # Get remote hashes + remote_hashes = self._get_remote_hashes(addresses, min_block, max_block) + logger.debug("Got remote hashes: {}".format(len(remote_hashes))) + for hash in remote_hashes: + logger.debug("remote hash: {}".format(hash.hex())) + + # Get local hashes of downloaded squeaks + local_hashes = self._get_local_hashes(addresses, min_block, max_block) + logger.debug("Got local hashes: {}".format(len(local_hashes))) + for hash in local_hashes: + logger.debug("local hash: {}".format(hash.hex())) + + # Get hashes to download + hashes_to_download = set(remote_hashes) - set(local_hashes) + logger.debug("Hashes to download: {}".format(len(hashes_to_download))) + for hash in hashes_to_download: + logger.debug("hash to download: {}".format(hash.hex())) + + # Download squeaks for the hashes + # TODO: catch exception downloading individual squeak + for hash in hashes_to_download: + if self.stopped(): + return + self._download_squeak(hash) + + # Get local hashes of locked squeaks that don't have an offer from this peer. + locked_hashes = self._get_locked_hashes(addresses, min_block, max_block) + logger.debug("Got locked hashes: {}".format(len(locked_hashes))) + for hash in locked_hashes: + logger.debug("locked hash: {}".format(hash.hex())) + + # Get hashes to get offer + hashes_to_get_offer = set(remote_hashes) & set(locked_hashes) + logger.debug("Hashes to get offer: {}".format(len(hashes_to_get_offer))) + for hash in hashes_to_get_offer: + logger.debug("hash to get offer: {}".format(hash.hex())) + + # Download offers for the hashes + # TODO: catch exception downloading individual squeak + for hash in hashes_to_get_offer: + if self.stopped(): + return + self._download_offer(hash) + + def upload( + self, + block_height, + lookup_block_interval=LOOKUP_BLOCK_INTERVAL, + ): + # Get list of sharing addresses. + addresses = self._get_sharing_addresses() + logger.debug("Sharing addresses: {}".format(addresses)) + min_block = block_height - lookup_block_interval + max_block = block_height + + # Get remote hashes + remote_hashes = self._get_remote_hashes(addresses, min_block, max_block) + logger.debug("Got remote hashes: {}".format(len(remote_hashes))) + for hash in remote_hashes: + logger.debug("remote hash: {}".format(hash.hex())) + + # Get local hashes + local_hashes = self._get_local_unlocked_hashes(addresses, min_block, max_block) + logger.debug("Got local hashes: {}".format(len(local_hashes))) + for hash in local_hashes: + logger.debug("local hash: {}".format(hash.hex())) + + # Get hashes to upload + hashes_to_upload = set(local_hashes) - set(remote_hashes) + logger.debug("Hashes to upload: {}".format(len(hashes_to_upload))) + for hash in hashes_to_upload: + logger.debug("hash to upload: {}".format(hash.hex())) + + # Upload squeaks for the hashes + # TODO: catch exception uploading individual squeak + for hash in hashes_to_upload: + if self.stopped(): + return + self._upload_squeak(hash) + + def download_single_squeak(self, squeak_hash): + # Download squeak if not already present. + saved_squeak = self._get_saved_squeak(squeak_hash) + if not saved_squeak: + self._download_squeak(squeak_hash) + + # Download offer from peer if not already present. + saved_offer = self._get_saved_offer(squeak_hash) + if not saved_offer: + self._download_offer(squeak_hash) + + def get_offer(self, squeak_hash): + logger.info("Getting offer for squeak hash: {}".format(squeak_hash.hex())) + + # Get the squeak from the squeak hash + squeak = self._get_local_squeak(squeak_hash) + + # Get the encryption key + encryption_key = squeak.GetEncryptionKey() + + # Create a new challenge + challenge_proof = self._generate_challenge_proof() + challenge = self._get_challenge(challenge_proof, encryption_key) + + # Download the buy offer + offer = self._download_buy_offer(squeak_hash, challenge) + + # Check the proof + proof = offer.proof + logger.info("Proof: {}".format(proof.hex())) + logger.info("Expected proof: {}".format(challenge_proof.hex())) + if proof != challenge_proof: + raise Exception( + "Invalid offer proof: {}, expected: {}".format( + proof.hex(), + challenge_proof.hex(), + ) + ) + + # Get the decoded offer from the payment request string + decoded_offer = self._get_decoded_offer(offer) + + # Save the offer + self._save_offer(decoded_offer) + + def stop(self): + self._stop_event.set() + + def stopped(self): + return self._stop_event.is_set() + + def _get_local_hashes(self, addresses, min_block, max_block): + return self.squeak_store.lookup_squeaks_include_locked( + addresses, + min_block, + max_block, + ) + + def _get_local_unlocked_hashes(self, addresses, min_block, max_block): + return self.squeak_store.lookup_squeaks(addresses, min_block, max_block) + + def _get_locked_hashes(self, addresses, min_block, max_block): + return self.squeak_store.lookup_squeaks_needing_offer( + addresses, + min_block, + max_block, + self.peer.peer_id, + ) + + def _get_remote_hashes(self, addresses, min_block, max_block): + return self.peer_client.lookup_squeaks(addresses, min_block, max_block) + + def _save_squeak(self, squeak): + self.squeak_store.save_downloaded_squeak(squeak) + + def _get_saved_squeak(self, squeak_hash): + return self.squeak_store.get_squeak(squeak_hash) + + def _get_saved_offer(self, squeak_hash): + offers = self.postgres_db.get_offers_with_peer(squeak_hash) + for offer in offers: + if offer.peer_id == peer_id: + return offer + + def _download_squeak(self, squeak_hash): + logger.info("Downloading squeak: {} from peer: {}".format(squeak_hash.hex(), self.peer.peer_id)) + squeak = self.peer_client.get_squeak(squeak_hash) + self._save_squeak(squeak) + + def _get_followed_addresses(self): + followed_profiles = self.postgres_db.get_following_profiles() + return [profile.address for profile in followed_profiles] + + def _download_offer(self, squeak_hash): + logger.info("Downloading offer for hash: {}".format(squeak_hash.hex())) + self.get_offer(squeak_hash) + + def _get_local_squeak(self, squeak_hash): + squeak_entry = self.squeak_store.get_squeak(squeak_hash) + return squeak_entry.squeak + + def _upload_squeak(self, squeak_hash): + logger.info("Uploading squeak: {}".format(squeak_hash.hex())) + squeak = self._get_local_squeak(squeak_hash) + self.peer_client.post_squeak(squeak) + + def _get_sharing_addresses(self): + sharing_profiles = self.postgres_db.get_sharing_profiles() + return [profile.address for profile in sharing_profiles] + + def _generate_challenge_proof(self): + return generate_data_key() + + def _get_challenge(self, challenge_proof, encryption_key): + return encryption_key.encrypt(challenge_proof) + + def _download_buy_offer(self, squeak_hash, challenge): + logger.info( + "Downloading buy offer for squeak hash: {}".format(squeak_hash.hex()) + ) + offer_msg = self.peer_client.buy_squeak(squeak_hash, challenge) + offer = self._offer_from_msg(offer_msg) + return offer + + def _save_offer(self, offer): + logger.info("Saving offer: {}".format(offer)) + self.postgres_db.insert_offer(offer) + + def _offer_from_msg(self, offer_msg): + if not offer_msg: + return None + return Offer( + offer_id=None, + squeak_hash=offer_msg.squeak_hash, + key_cipher=offer_msg.key_cipher, + iv=offer_msg.iv, + price_msat=None, + payment_hash=offer_msg.preimage_hash, + invoice_timestamp=None, + invoice_expiry=None, + payment_request=offer_msg.payment_request, + destination=None, + node_host=offer_msg.host, + node_port=offer_msg.port, + proof=offer_msg.proof, + peer_id=self.peer.peer_id, + ) + + def _decode_payment_request(self, payment_request): + return self.lightning_client.decode_pay_req(payment_request) + + def _get_decoded_offer(self, offer): + pay_req = self._decode_payment_request(offer.payment_request) + logger.info("Decoded payment request: {}".format(pay_req)) + + price_msat = pay_req.num_msat + destination = pay_req.destination + invoice_timestamp = pay_req.timestamp + invoice_expiry = pay_req.expiry + node_host = offer.node_host or self.peer.host + node_port = offer.node_port + + logger.info("price_msat: {}".format(price_msat)) + logger.info("destination: {}".format(destination)) + logger.info("invoice_timestamp: {}".format(invoice_timestamp)) + logger.info("invoice_expiry: {}".format(invoice_expiry)) + logger.info("node_host: {}".format(node_host)) + logger.info("node_port: {}".format(node_port)) + + decoded_offer = Offer( + offer_id=offer.offer_id, + squeak_hash=offer.squeak_hash, + key_cipher=offer.key_cipher, + iv=offer.iv, + price_msat=price_msat, + payment_hash=offer.payment_hash, + invoice_timestamp=invoice_timestamp, + invoice_expiry=invoice_expiry, + payment_request=offer.payment_request, + destination=destination, + node_host=node_host, + node_port=node_port, + proof=offer.proof, + peer_id=offer.peer_id, + ) + + return decoded_offer diff --git a/squeaknode/node/peer_upload.py b/squeaknode/node/peer_upload.py index 16026b1f..614f50d9 100644 --- a/squeaknode/node/peer_upload.py +++ b/squeaknode/node/peer_upload.py @@ -2,6 +2,7 @@ import logging import threading from squeaknode.network.peer_client import PeerClient +from squeaknode.node.peer_task import PeerSyncTask logger = logging.getLogger(__name__) @@ -9,90 +10,48 @@ logger = logging.getLogger(__name__) LOOKUP_BLOCK_INTERVAL = 1008 # 1 week -class PeerUpload: - def __init__( - self, - peer, - block_height, - squeak_store, - postgres_db, - lookup_block_interval=LOOKUP_BLOCK_INTERVAL, - ): - self.peer = peer - self.block_height = block_height - self.squeak_store = squeak_store - self.postgres_db = postgres_db - self.lookup_block_interval = lookup_block_interval +# class PeerUpload(PeerSyncTask): +# def __init__( +# self, +# peer, +# squeak_store, +# postgres_db, +# lightning_client, +# ): +# super().__init__(peer, squeak_store, postgres_db, lightning_client) - self.peer_client = PeerClient( - self.peer.host, - self.peer.port, - ) +# def upload( +# self, +# block_height, +# lookup_block_interval=LOOKUP_BLOCK_INTERVAL, +# ): +# # Get list of sharing addresses. +# addresses = self._get_sharing_addresses() +# logger.debug("Sharing addresses: {}".format(addresses)) +# min_block = block_height - lookup_block_interval +# max_block = block_height - self._stop_event = threading.Event() +# # Get remote hashes +# remote_hashes = self._get_remote_hashes(addresses, min_block, max_block) +# logger.debug("Got remote hashes: {}".format(len(remote_hashes))) +# for hash in remote_hashes: +# logger.debug("remote hash: {}".format(hash.hex())) - def upload(self): - # Get list of sharing addresses. - addresses = self._get_sharing_addresses() - logger.debug("Sharing addresses: {}".format(addresses)) - min_block = self.block_height - self.lookup_block_interval - max_block = self.block_height +# # Get local hashes +# local_hashes = self._get_local_hashes(addresses, min_block, max_block) +# logger.debug("Got local hashes: {}".format(len(local_hashes))) +# for hash in local_hashes: +# logger.debug("local hash: {}".format(hash.hex())) - if self.stopped(): - return +# # Get hashes to upload +# hashes_to_upload = set(local_hashes) - set(remote_hashes) +# logger.debug("Hashes to upload: {}".format(len(hashes_to_upload))) +# for hash in hashes_to_upload: +# logger.debug("hash to upload: {}".format(hash.hex())) - # Get remote hashes - remote_hashes = self._get_remote_hashes(addresses, min_block, max_block) - logger.debug("Got remote hashes: {}".format(len(remote_hashes))) - for hash in remote_hashes: - logger.debug("remote hash: {}".format(hash.hex())) - - if self.stopped(): - return - - # Get local hashes - local_hashes = self._get_local_hashes(addresses, min_block, max_block) - logger.debug("Got local hashes: {}".format(len(local_hashes))) - for hash in local_hashes: - logger.debug("local hash: {}".format(hash.hex())) - - if self.stopped(): - return - - # Get hashes to upload - hashes_to_upload = set(local_hashes) - set(remote_hashes) - logger.debug("Hashes to upload: {}".format(len(hashes_to_upload))) - for hash in hashes_to_upload: - logger.debug("hash to upload: {}".format(hash.hex())) - - # Upload squeaks for the hashes - # TODO: catch exception uploading individual squeak - for hash in hashes_to_upload: - if self.stopped(): - return - self._upload_squeak(hash) - - def stop(self): - self._stop_event.set() - - def stopped(self): - return self._stop_event.is_set() - - def _get_local_hashes(self, addresses, min_block, max_block): - return self.squeak_store.lookup_squeaks(addresses, min_block, max_block) - - def _get_remote_hashes(self, addresses, min_block, max_block): - return self.peer_client.lookup_squeaks(addresses, min_block, max_block) - - def _get_local_squeak(self, squeak_hash): - squeak_entry = self.squeak_store.get_squeak(squeak_hash) - return squeak_entry.squeak - - def _upload_squeak(self, squeak_hash): - logger.info("Uploading squeak: {}".format(squeak_hash.hex())) - squeak = self._get_local_squeak(squeak_hash) - self.peer_client.post_squeak(squeak) - - def _get_sharing_addresses(self): - sharing_profiles = self.postgres_db.get_sharing_profiles() - return [profile.address for profile in sharing_profiles] +# # Upload squeaks for the hashes +# # TODO: catch exception uploading individual squeak +# for hash in hashes_to_upload: +# if self.stopped(): +# return +# self._upload_squeak(hash) diff --git a/squeaknode/node/squeak_node.py b/squeaknode/node/squeak_node.py index c788c936..0695302d 100644 --- a/squeaknode/node/squeak_node.py +++ b/squeaknode/node/squeak_node.py @@ -9,7 +9,6 @@ from squeak.core.signing import CSigningKey, CSqueakAddress from squeak.core import CheckSqueak from squeaknode.core.squeak_address_validator import SqueakAddressValidator -from squeaknode.node.peer_download import PeerDownload from squeaknode.node.squeak_block_periodic_worker import SqueakBlockPeriodicWorker from squeaknode.node.squeak_block_queue_worker import SqueakBlockQueueWorker from squeaknode.node.squeak_block_verifier import SqueakBlockVerifier diff --git a/squeaknode/node/squeak_sync_status.py b/squeaknode/node/squeak_sync_status.py index b8c954cc..5294ac7f 100644 --- a/squeaknode/node/squeak_sync_status.py +++ b/squeaknode/node/squeak_sync_status.py @@ -3,8 +3,9 @@ import threading from collections import defaultdict -from squeaknode.node.peer_download import PeerDownload -from squeaknode.node.peer_upload import PeerUpload +from squeaknode.node.network_task import DownloadTimelineNetworkSyncTask +from squeaknode.node.network_task import UploadTimelineNetworkSyncTask +from squeaknode.node.network_task import SingleSqueakNetworkSyncTask logger = logging.getLogger(__name__) @@ -12,55 +13,58 @@ logger = logging.getLogger(__name__) HOUR_IN_SECONDS = 3600 -class SqueakSyncStatus: - def __init__(self): - self.downloads = {} - self.uploads = {} - self.single_squeak_downloads = defaultdict(dict) +# class SqueakSyncStatus: +# def __init__(self): +# self.downloads = {} +# self.uploads = {} +# self.single_squeak_downloads = defaultdict(dict) - def add_download(self, peer, peer_download): - self.downloads[peer.peer_id] = peer_download +# def add_download(self, peer, peer_download): +# self.downloads[peer.peer_id] = peer_download - def add_upload(self, peer, peer_upload): - self.uploads[peer.peer_id] = peer_upload +# def add_upload(self, peer, peer_upload): +# self.uploads[peer.peer_id] = peer_upload - def add_single_squeak_download(self, squeak_hash, peer, peer_download): - self.single_squeak_downloads[squeak_hash][peer.peer_id] = peer_download +# def add_single_squeak_download(self, squeak_hash, peer, peer_download): +# self.single_squeak_downloads[squeak_hash][peer.peer_id] = peer_download - def is_downloading(self, peer): - return peer.peer_id in self.downloads +# def is_downloading(self, peer): +# return peer.peer_id in self.downloads - def is_uploading(self, peer): - return peer.peer_id in self.uploads +# def is_uploading(self, peer): +# return peer.peer_id in self.uploads - def is_downloading_single_squeak(self, squeak_hash, peer): - return peer.peer_id in self.single_squeak_downloads[squeak_hash] +# def is_downloading_single_squeak(self, squeak_hash, peer): +# return peer.peer_id in self.single_squeak_downloads[squeak_hash] - def remove_download(self, peer): - del self.downloads[peer.peer_id] +# def remove_download(self, peer): +# del self.downloads[peer.peer_id] - def remove_upload(self, peer): - del self.uploads[peer.peer_id] +# def remove_upload(self, peer): +# del self.uploads[peer.peer_id] - def remove_single_peer_download(self, squeak_hash, peer): - del self.single_squeak_downloads[squeak_hash][peer.peer_id] +# def remove_single_peer_download(self, squeak_hash, peer): +# del self.single_squeak_downloads[squeak_hash][peer.peer_id] - def get_current_downloads(self): - return self.downloads.values() +# def get_current_downloads(self): +# return self.downloads.values() - def get_current_uploads(self): - return self.uploads.values() +# def get_current_uploads(self): +# return self.uploads.values() class SqueakSyncController: def __init__(self, blockchain_client, squeak_store, postgres_db, lightning_client): - self.squeak_sync_status = SqueakSyncStatus() self.blockchain_client = blockchain_client self.squeak_store = squeak_store self.postgres_db = postgres_db self.lightning_client = lightning_client def sync_peers(self, peers): + self.download_timeline(peers) + self.upload_timeline(peers) + + def download_timeline(self, peers): try: block_info = self.blockchain_client.get_best_block_info() block_height = block_info.block_height @@ -69,138 +73,168 @@ class SqueakSyncController: "Failed to sync because unable to get blockchain info.", exc_info=False ) return - self._download_from_peers(peers, block_height) - self._upload_to_peers(peers, block_height) + dowload_timeline_task = DownloadTimelineNetworkSyncTask( + self.squeak_store, + self.postgres_db, + self.lightning_client, + block_height, + ) + dowload_timeline_task.sync(peers) + + def upload_timeline(self, peers): + try: + block_info = self.blockchain_client.get_best_block_info() + block_height = block_info.block_height + except Exception as e: + logger.error( + "Failed to sync because unable to get blockchain info.", exc_info=False + ) + return + upload_timeline_task = UploadTimelineNetworkSyncTask( + self.squeak_store, + self.postgres_db, + self.lightning_client, + block_height, + ) + upload_timeline_task.sync(peers) def download_single_squeak_from_peers(self, squeak_hash, peers): - for peer in peers: - if peer.downloading: - download_thread = threading.Thread( - target=self._download_single_squeak_from_peer, - args=( - squeak_hash, - peer, - ), - ) - download_thread.start() - - def _download_from_peers(self, peers, block_height): - for peer in peers: - if peer.downloading: - download_thread = threading.Thread( - target=self._download_from_peer, - args=( - peer, - block_height, - ), - ) - download_thread.start() - - def _upload_to_peers(self, peers, block_height): - for peer in peers: - if peer.uploading: - upload_thread = threading.Thread( - target=self._upload_to_peer, - args=( - peer, - block_height, - ), - ) - upload_thread.start() - - def _download_from_peer(self, peer, block_height): - peer_download = PeerDownload( - peer, + # for peer in peers: + # if peer.downloading: + # download_thread = threading.Thread( + # target=self._download_single_squeak_from_peer, + # args=( + # squeak_hash, + # peer, + # ), + # ) + # download_thread.start() + timeline_sync_task = SingleSqueakNetworkSyncTask( self.squeak_store, self.postgres_db, self.lightning_client, + squeak_hash, ) - try: - logger.debug("Trying to download from peer: {}".format(peer)) - with self.DownloadingContextManager( - peer, peer_download, self.squeak_sync_status - ) as downloading_manager: - peer_download.download(block_height) - except Exception as e: - logger.error("Download from peer failed.", exc_info=True) + timeline_sync_task.sync(peers) - def _upload_to_peer(self, peer, block_height): - peer_upload = PeerUpload( - peer, - block_height, - self.squeak_store, - self.postgres_db, - ) - try: - logger.debug("Trying to upload to peer: {}".format(peer)) - with self.UploadingContextManager( - peer, peer_upload, self.squeak_sync_status - ) as uploading_manager: - peer_upload.upload() - except Exception as e: - logger.error("Upload from peer failed.", exc_info=True) - def _download_single_squeak_from_peer(self, squeak_hash, peer): - peer_download = PeerDownload( - peer, - self.squeak_store, - self.postgres_db, - self.lightning_client, - ) - try: - logger.debug("Trying to download single squeak {} from peer: {}".format(squeak_hash, peer)) - with self.SingleSqueakDownloadingContextManager( - squeak_hash, peer, peer_download, self.squeak_sync_status - ) as downloading_manager: - peer_download.download_single_squeak(squeak_hash) - except Exception as e: - logger.error("Download single squeak from peer failed.", exc_info=True) + # def _download_from_peers(self, peers, block_height): + # for peer in peers: + # if peer.downloading: + # download_thread = threading.Thread( + # target=self._download_from_peer, + # args=( + # peer, + # block_height, + # ), + # ) + # download_thread.start() - class DownloadingContextManager: - def __init__(self, peer, peer_download, squeak_sync_status): - self.peer = peer - self.peer_download = peer_download - self.squeak_sync_status = squeak_sync_status + # def _upload_to_peers(self, peers, block_height): + # for peer in peers: + # if peer.uploading: + # upload_thread = threading.Thread( + # target=self._upload_to_peer, + # args=( + # peer, + # block_height, + # ), + # ) + # upload_thread.start() - if self.squeak_sync_status.is_downloading(self.peer): - raise Exception("Peer is already downloading: {}".format(self.peer)) + # def _download_from_peer(self, peer, block_height): + # peer_download = PeerDownload( + # peer, + # self.squeak_store, + # self.postgres_db, + # self.lightning_client, + # ) + # try: + # logger.debug("Trying to download from peer: {}".format(peer)) + # with self.DownloadingContextManager( + # peer, peer_download, self.squeak_sync_status + # ) as downloading_manager: + # peer_download.download(block_height) + # except Exception as e: + # logger.error("Download from peer failed.", exc_info=True) - def __enter__(self): - self.squeak_sync_status.add_download(self.peer, self.peer_download) - return self + # def _upload_to_peer(self, peer, block_height): + # peer_upload = PeerUpload( + # peer, + # self.squeak_store, + # self.postgres_db, + # self.lightning_client, + # ) + # try: + # logger.debug("Trying to upload to peer: {}".format(peer)) + # with self.UploadingContextManager( + # peer, peer_upload, self.squeak_sync_status + # ) as uploading_manager: + # peer_upload.upload(block_height) + # except Exception as e: + # logger.error("Upload from peer failed.", exc_info=True) - def __exit__(self, exc_type, exc_value, exc_traceback): - self.squeak_sync_status.remove_download(self.peer) + # def _download_single_squeak_from_peer(self, squeak_hash, peer): + # peer_download = PeerSingleSqueakDownload( + # peer, + # self.squeak_store, + # self.postgres_db, + # self.lightning_client, + # ) + # try: + # logger.debug("Trying to download single squeak {} from peer: {}".format(squeak_hash, peer)) + # with self.SingleSqueakDownloadingContextManager( + # squeak_hash, peer, peer_download, self.squeak_sync_status + # ) as downloading_manager: + # peer_download.download_single_squeak(squeak_hash) + # except Exception as e: + # logger.error("Download single squeak from peer failed.", exc_info=True) - class UploadingContextManager: - def __init__(self, peer, peer_upload, squeak_sync_status): - self.peer = peer - self.peer_upload = peer_upload - self.squeak_sync_status = squeak_sync_status + # class DownloadingContextManager: + # def __init__(self, peer, peer_download, squeak_sync_status): + # self.peer = peer + # self.peer_download = peer_download + # self.squeak_sync_status = squeak_sync_status - if self.squeak_sync_status.is_uploading(self.peer): - raise Exception("Peer is already uploading: {}".format(self.peer)) + # if self.squeak_sync_status.is_downloading(self.peer): + # raise Exception("Peer is already downloading: {}".format(self.peer)) - def __enter__(self): - self.squeak_sync_status.add_upload(self.peer, self.peer_upload) - return self + # def __enter__(self): + # self.squeak_sync_status.add_download(self.peer, self.peer_download) + # return self - def __exit__(self, exc_type, exc_value, exc_traceback): - self.squeak_sync_status.remove_upload(self.peer) + # def __exit__(self, exc_type, exc_value, exc_traceback): + # self.squeak_sync_status.remove_download(self.peer) - class SingleSqueakDownloadingContextManager: - def __init__(self, squeak_hash, peer, peer_download, squeak_sync_status): - self.squeak_hash = squeak_hash - self.peer = peer - self.peer_download = peer_download - self.squeak_sync_status = squeak_sync_status + # class UploadingContextManager: + # def __init__(self, peer, peer_upload, squeak_sync_status): + # self.peer = peer + # self.peer_upload = peer_upload + # self.squeak_sync_status = squeak_sync_status - if self.squeak_sync_status.is_downloading_single_squeak(self.squeak_hash, self.peer): - raise Exception("Peer {} is already downloading hash: {}".format(self.peer, self.squeak_hash)) + # if self.squeak_sync_status.is_uploading(self.peer): + # raise Exception("Peer is already uploading: {}".format(self.peer)) - def __enter__(self): - self.squeak_sync_status.add_single_squeak_download(self.squeak_hash, self.peer, self.peer_download) - return self + # def __enter__(self): + # self.squeak_sync_status.add_upload(self.peer, self.peer_upload) + # return self - def __exit__(self, exc_type, exc_value, exc_traceback): - self.squeak_sync_status.remove_single_peer_download(self.squeak_hash, self.peer) + # def __exit__(self, exc_type, exc_value, exc_traceback): + # self.squeak_sync_status.remove_upload(self.peer) + + # class SingleSqueakDownloadingContextManager: + # def __init__(self, squeak_hash, peer, peer_download, squeak_sync_status): + # self.squeak_hash = squeak_hash + # self.peer = peer + # self.peer_download = peer_download + # self.squeak_sync_status = squeak_sync_status + + # if self.squeak_sync_status.is_downloading_single_squeak(self.squeak_hash, self.peer): + # raise Exception("Peer {} is already downloading hash: {}".format(self.peer, self.squeak_hash)) + + # def __enter__(self): + # self.squeak_sync_status.add_single_squeak_download(self.squeak_hash, self.peer, self.peer_download) + # return self + + # def __exit__(self, exc_type, exc_value, exc_traceback): + # self.squeak_sync_status.remove_single_peer_download(self.squeak_hash, self.peer)