Raise custom error when inserting duplicate squaek in db. (#1478)

* Raise custom error when inserting duplicate squaek in db.

* Don't raise error in insert db method, return optional primary key

* Update the docstring for insert squeak

* Insert received offer db method returns optional primary key

* Use optional return type for insert received payment method

* Comment out unused db exceptions for duplicate inserts

itests/docker-compose.yml

@@ -13,8 +13,6 @@ services:
     environment:
       - NETWORK=simnet
       - MINING_ADDRESS
-    ports:
-      - 18556:18556
     entrypoint: ["./start-btcd.sh"]
 
   btcctl:

squeaknode/admin/squeak_admin_server_handler.py

@@ -316,6 +316,10 @@ class SqueakAdminServerHandler(object):
         inserted_squeak_hash = self.squeak_controller.make_squeak(
             profile_id, content_str, replyto_hash
         )
+        if inserted_squeak_hash is None:
+            return squeak_admin_pb2.MakeSqueakReply(
+                squeak_hash=None,
+            )
         return squeak_admin_pb2.MakeSqueakReply(
             squeak_hash=inserted_squeak_hash.hex(),
         )

squeaknode/db/exception.py

@@ -25,9 +25,13 @@ class SqueakDatabaseError(Exception):
     """Base class for other squeak database exceptions"""
 
 
-class DuplicateReceivedPaymentError(SqueakDatabaseError):
-    """Raised when the inserted received payment is a duplicate"""
+# class DuplicateSqueakError(SqueakDatabaseError):
+#     """Raised when the inserted squeak is a duplicate"""
 
 
-class DuplicateReceivedOfferError(SqueakDatabaseError):
-    """Raised when the inserted received offer is a duplicate"""
+# class DuplicateReceivedPaymentError(SqueakDatabaseError):
+#     """Raised when the inserted received payment is a duplicate"""
+
+
+# class DuplicateReceivedOfferError(SqueakDatabaseError):
+#     """Raised when the inserted received offer is a duplicate"""

squeaknode/db/squeak_db.py

@@ -46,8 +46,6 @@ from squeaknode.core.squeak_entry import SqueakEntry
 from squeaknode.core.squeak_peer import SqueakPeer
 from squeaknode.core.squeak_profile import SqueakProfile
 from squeaknode.core.squeaks import get_hash
-from squeaknode.db.exception import DuplicateReceivedOfferError
-from squeaknode.db.exception import DuplicateReceivedPaymentError
 from squeaknode.db.migrations import run_migrations
 from squeaknode.db.models import Models
 
@@ -182,10 +180,11 @@ class SqueakDb:
         )
         return self.timestamp_now_ms / 1000 < expire_time
 
-    def insert_squeak(self, squeak: CSqueak, block_header: CBlockHeader) -> bytes:
+    def insert_squeak(self, squeak: CSqueak, block_header: CBlockHeader) -> Optional[bytes]:
         """ Insert a new squeak.
 
         Return the hash (bytes) of the inserted squeak.
+        Return None if squeak already exists.
         """
         ins = self.squeaks.insert().values(
             created_time_ms=self.timestamp_now_ms,
@@ -201,11 +200,12 @@ class SqueakDb:
         )
         with self.get_connection() as connection:
             try:
-                connection.execute(ins)
-                # inserted_squeak_hash = res.inserted_primary_key[0]
+                res = connection.execute(ins)
+                squeak_hash = res.inserted_primary_key[0]
+                return squeak_hash
             except sqlalchemy.exc.IntegrityError:
-                pass
-            return get_hash(squeak)
+                logger.debug("Failed to insert squeak.", exc_info=True)
+                return None
 
     def get_squeak(self, squeak_hash: bytes) -> Optional[CSqueak]:
         """ Get a squeak. """
@@ -983,8 +983,12 @@ class SqueakDb:
         with self.get_connection() as connection:
             connection.execute(delete_peer_stmt)
 
-    def insert_received_offer(self, received_offer: ReceivedOffer):
-        """ Insert a new received offer. """
+    def insert_received_offer(self, received_offer: ReceivedOffer) -> Optional[int]:
+        """ Insert a new received offer.
+
+        Return the received offer id of the inserted received offer.
+        Return None if received offer already exists.
+        """
         ins = self.received_offers.insert().values(
             created_time_ms=self.timestamp_now_ms,
             squeak_hash=received_offer.squeak_hash,
@@ -1007,7 +1011,8 @@ class SqueakDb:
                 received_offer_id = res.inserted_primary_key[0]
                 return received_offer_id
             except sqlalchemy.exc.IntegrityError:
-                raise DuplicateReceivedOfferError()
+                logger.debug("Failed to insert received offer.", exc_info=True)
+                return None
 
     def get_received_offers(self, squeak_hash: bytes) -> List[ReceivedOffer]:
         """ Get offers with peer for a squeak hash. """
@@ -1277,8 +1282,12 @@ class SqueakDb:
             latest_index = row[0]
             return latest_index
 
-    def insert_received_payment(self, received_payment: ReceivedPayment):
-        """ Insert a new received payment. """
+    def insert_received_payment(self, received_payment: ReceivedPayment) -> Optional[int]:
+        """ Insert a new received payment.
+
+        Return the received payment id of the inserted received payment.
+        Return None if received payment already exists.
+        """
         ins = self.received_payments.insert().values(
             created_time_ms=self.timestamp_now_ms,
             squeak_hash=received_payment.squeak_hash,
@@ -1294,7 +1303,9 @@ class SqueakDb:
                 received_payment_id = res.inserted_primary_key[0]
                 return received_payment_id
             except sqlalchemy.exc.IntegrityError:
-                raise DuplicateReceivedPaymentError()
+                logger.debug(
+                    "Failed to insert received payment.", exc_info=True)
+                return None
 
     def get_received_payments(
             self,

squeaknode/node/payment_processor.py

@@ -25,7 +25,6 @@ import threading
 from squeaknode.core.exception import InvoiceSubscriptionError
 from squeaknode.core.received_payment import ReceivedPayment
 from squeaknode.core.sent_offer import SentOffer
-from squeaknode.db.exception import DuplicateReceivedPaymentError
 
 
 logger = logging.getLogger(__name__)
@@ -122,12 +121,12 @@ class PaymentProcessorTask:
     def handle_received_payment(self, received_payment: ReceivedPayment):
         logger.info(
             "Got received payment: {}".format(received_payment))
-        try:
-            self.squeak_db.insert_received_payment(
-                received_payment,
-            )
-        except DuplicateReceivedPaymentError:
-            pass
+        received_payment_id = self.squeak_db.insert_received_payment(
+            received_payment,
+        )
+        if received_payment_id is not None:
+            logger.debug(
+                "Saved received payment: {}".format(received_payment))
         # # TODO: Should not be deleting sent offer.
         # self.squeak_db.delete_sent_offer(
         #     received_payment.payment_hash,

squeaknode/node/squeak_controller.py

@@ -56,7 +56,6 @@ from squeaknode.core.squeak_entry import SqueakEntry
 from squeaknode.core.squeak_peer import SqueakPeer
 from squeaknode.core.squeak_profile import SqueakProfile
 from squeaknode.core.squeaks import get_hash
-from squeaknode.db.exception import DuplicateReceivedOfferError
 from squeaknode.node.listener_subscription_client import EventListener
 from squeaknode.node.received_payments_subscription_client import ReceivedPaymentsSubscriptionClient
 from squeaknode.node.temporary_interest_manager import TemporaryInterest
@@ -86,7 +85,7 @@ class SqueakController:
         self.temporary_interest_manager = TemporaryInterestManager()
         self.config = config
 
-    def save_squeak(self, squeak: CSqueak) -> bytes:
+    def save_squeak(self, squeak: CSqueak) -> Optional[bytes]:
         # Check if the squeak is valid
         self.squeak_core.check_squeak(squeak)
         # Get the block header for the squeak.
@@ -99,6 +98,8 @@ class SqueakController:
             squeak,
             block_header,
         )
+        if inserted_squeak_hash is None:
+            return None
         logger.info("Saved squeak: {}".format(
             inserted_squeak_hash.hex(),
         ))
@@ -123,11 +124,13 @@ class SqueakController:
         # Notify the listener
         self.new_secret_key_listener.handle_new_item(squeak)
 
-    def make_squeak(self, profile_id: int, content_str: str, replyto_hash: bytes) -> bytes:
+    def make_squeak(self, profile_id: int, content_str: str, replyto_hash: bytes) -> Optional[bytes]:
         squeak_profile = self.squeak_db.get_profile(profile_id)
         squeak, decryption_key = self.squeak_core.make_squeak(
             squeak_profile, content_str, replyto_hash)
         inserted_squeak_hash = self.save_squeak(squeak)
+        if inserted_squeak_hash is None:
+            return None
         self.unlock_squeak(
             inserted_squeak_hash,
             decryption_key,
@@ -151,7 +154,8 @@ class SqueakController:
         counter = self.get_temporary_interest_counter(squeak)
         if counter:
             saved_squeak_hash = self.save_squeak(squeak)
-            counter.increment()
+            if saved_squeak_hash:
+                counter.increment()
         elif self.squeak_matches_interest(squeak):
             saved_squeak_hash = self.save_squeak(squeak)
         # Download offers for the new squeak
@@ -523,7 +527,6 @@ class SqueakController:
         return self.squeak_db.get_number_of_squeaks()
 
     def save_received_offer(self, offer: Offer, peer_address: PeerAddress) -> None:
-        logger.info("Saving received offer: {}".format(offer))
         squeak = self.get_squeak(offer.squeak_hash)
         secret_key = self.get_squeak_secret_key(offer.squeak_hash)
         if squeak is None or secret_key is not None:
@@ -537,13 +540,14 @@ class SqueakController:
         except Exception:
             logger.exception("Failed to save received offer.")
             return
-        try:
-            offer_id = self.squeak_db.insert_received_offer(received_offer)
-            received_offer = received_offer._replace(
-                received_offer_id=offer_id)
-            self.new_received_offer_listener.handle_new_item(received_offer)
-        except DuplicateReceivedOfferError:
-            logger.debug("Failed to save duplicate offer.")
+        received_offer_id = self.squeak_db.insert_received_offer(
+            received_offer)
+        if received_offer_id is None:
+            return
+        logger.info("Saved received offer: {}".format(received_offer))
+        received_offer = received_offer._replace(
+            received_offer_id=received_offer_id)
+        self.new_received_offer_listener.handle_new_item(received_offer)
 
     def get_followed_addresses(self) -> List[str]:
         followed_profiles = self.squeak_db.get_following_profiles()