Internal: store tx_hash in rejected messages table

deployment/migrations/versions/0019_3bf484f2cc95_tx_hash_in_rejected_messages.py

@@ -0,0 +1,32 @@
+"""tx_hash in rejected messages
+
+Revision ID: 3bf484f2cc95
+Revises: 7bcb8e5fe186
+Create Date: 2023-07-31 00:08:17.990537
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "3bf484f2cc95"
+down_revision = "7bcb8e5fe186"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column("rejected_messages", sa.Column("tx_hash", sa.String(), nullable=True))
+    op.create_foreign_key(None, "rejected_messages", "chain_txs", ["tx_hash"], ["hash"])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(
+        "rejected_messages_tx_hash_fkey", "rejected_messages", type_="foreignkey"
+    )
+    op.drop_column("rejected_messages", "tx_hash")
+    # ### end Alembic commands ###

src/aleph/db/accessors/messages.py

@@ -401,20 +401,23 @@ def make_upsert_rejected_message_statement(
     error_code: int,
     details: Optional[Mapping[str, Any]] = None,
     exc_traceback: Optional[str] = None,
+    tx_hash: Optional[str] = None,
 ) -> Insert:
     insert_rejected_message_stmt = insert(RejectedMessageDb).values(
         item_hash=item_hash,
         message=pending_message_dict,
         error_code=error_code,
         details=details,
         traceback=exc_traceback,
+        tx_hash=tx_hash,
     )
     upsert_rejected_message_stmt = insert_rejected_message_stmt.on_conflict_do_update(
         constraint="rejected_messages_pkey",
         set_={
             "error_code": insert_rejected_message_stmt.excluded.error_code,
             "details": details,
             "traceback": insert_rejected_message_stmt.excluded.traceback,
+            "tx_hash": tx_hash,
         },
     )
     return upsert_rejected_message_stmt
@@ -425,6 +428,7 @@ def mark_pending_message_as_rejected(
     item_hash: str,
     pending_message_dict: Mapping[str, Any],
     exception: BaseException,
+    tx_hash: Optional[str],
 ) -> RejectedMessageDb:
     if isinstance(exception, MessageProcessingException):
         error_code = exception.error_code
@@ -452,6 +456,7 @@ def mark_pending_message_as_rejected(
         details=details,
         error_code=error_code,
         exc_traceback=exc_traceback,
+        tx_hash=tx_hash,
     )
 
     session.execute(upsert_status_stmt)
@@ -463,6 +468,7 @@ def mark_pending_message_as_rejected(
         traceback=exc_traceback,
         error_code=error_code,
         details=details,
+        tx_hash=tx_hash,
     )
 
 
@@ -471,6 +477,7 @@ def reject_new_pending_message(
     session: DbSession,
     pending_message: Mapping[str, Any],
     exception: BaseException,
+        tx_hash: Optional[str],
 ) -> None:
     ...
 
@@ -480,6 +487,7 @@ def reject_new_pending_message(
     session: DbSession,
     pending_message: PendingMessageDb,
     exception: BaseException,
+    tx_hash: Optional[str],
 ) -> None:
     ...
 
@@ -488,6 +496,7 @@ def reject_new_pending_message(
     session: DbSession,
     pending_message: Union[Mapping[str, Any], PendingMessageDb],
     exception: BaseException,
+    tx_hash: Optional[str],
 ) -> Optional[RejectedMessageDb]:
     """
     Reject a pending message that is not yet in the DB.
@@ -529,6 +538,7 @@ def reject_new_pending_message(
         item_hash=item_hash,
         pending_message_dict=pending_message_dict,
         exception=exception,
+        tx_hash=tx_hash,
     )
 
 
@@ -566,6 +576,7 @@ def reject_existing_pending_message(
         item_hash=item_hash,
         pending_message_dict=pending_message_dict,
         exception=exception,
+        tx_hash=pending_message.tx_hash,
     )
     delete_pending_message(session=session, pending_message=pending_message)
     return rejected_message

src/aleph/db/models/messages.py

@@ -208,3 +208,4 @@ class RejectedMessageDb(Base):
     )
     details: Optional[Dict[str, Any]] = Column(JSONB, nullable=True)
     traceback: Optional[str] = Column(String, nullable=True)
+    tx_hash: Optional[str] = Column(ForeignKey("chain_txs.hash"), nullable=True)

src/aleph/handlers/message_handler.py

@@ -183,7 +183,10 @@ async def add_pending_message(
             except InvalidMessageException as e:
                 LOGGER.warning(e)
                 reject_new_pending_message(
-                    session=session, pending_message=message_dict, exception=e
+                    session=session,
+                    pending_message=message_dict,
+                    exception=e,
+                    tx_hash=tx_hash,
                 )
                 session.commit()
                 return None
@@ -202,7 +205,10 @@ async def add_pending_message(
             except InvalidMessageException as e:
                 LOGGER.warning("Invalid message: %s - %s", message.item_hash, str(e))
                 reject_new_pending_message(
-                    session=session, pending_message=message_dict, exception=e
+                    session=session,
+                    pending_message=message_dict,
+                    exception=e,
+                    tx_hash=tx_hash,
                 )
                 session.commit()
                 return None
@@ -234,6 +240,7 @@ async def add_pending_message(
                     session=session,
                     pending_message=message_dict,
                     exception=e,
+                    tx_hash=tx_hash,
                 )
                 session.commit()
                 return None