Fix: drop invalid pending messages

Problem: if an invalid message somehow managed to reach
the pending message collection, the message would be retried
indefinitely logging exceptions on each run.

Solution: drop invalid messages.

Author: odesenfans

src/aleph/handlers/storage.py

@@ -48,7 +48,6 @@ async def handle_new_storage(store_message: ValidatedStoreMessage) -> Optional[b
 
     ipfs_enabled = config.ipfs.enabled.value
     do_standard_lookup = True
-
     if engine == ItemType.ipfs and ipfs_enabled:
         if item_type_from_hash(item_hash) != ItemType.ipfs:
             LOGGER.warning("Invalid IPFS hash: '%s'", item_hash)

src/aleph/jobs/process_pending_messages.py

@@ -19,6 +19,7 @@
 from aleph.model.pending import PendingMessage
 from aleph.services.p2p import singleton
 from .job_utils import prepare_loop, process_job_results
+from ..exceptions import InvalidMessageError
 from ..schemas.pending_messages import parse_message
 
 LOGGER = getLogger("jobs.pending_messages")
@@ -49,7 +50,15 @@ async def handle_pending_message(
     seen_ids: Dict[Tuple, int],
 ) -> List[DbBulkOperation]:
 
-    message = parse_message(pending["message"])
+    delete_pending_message_op = DbBulkOperation(
+        PendingMessage, DeleteOne({"_id": pending["_id"]})
+    )
+
+    try:
+        message = parse_message(pending["message"])
+    except InvalidMessageError:
+        # If an invalid message somehow ended in pending messages, drop it.
+        return [delete_pending_message_op]
 
     async with sem:
         status, operations = await incoming(
@@ -64,9 +73,7 @@ async def handle_pending_message(
         )
 
     if status != IncomingStatus.RETRYING_LATER:
-        operations.append(
-            DbBulkOperation(PendingMessage, DeleteOne({"_id": pending["_id"]}))
-        )
+        operations.append(delete_pending_message_op)
 
     return operations