[TXs] Store the full TX context in confirmations

We now store the entire context of the transaction in the confirmations
array of messages. This means that two additional fields are now
preserved: the transaction block timestamp and the publisher.

As we need to re-fetch this data from chain data, a new migration
script resets the chain height to re-process all transactions.
We reset the confirmation status of all messages to unconfirmed
and deleted their confirmations array to let the node automatically
migrate to the new format.

Renamed some fields of the `TxContext` class in order to use
the same format in all DB collections and to avoid a breaking
change in the messages confirmation format.

Author: odesenfans

deployment/migrations/scripts/0003-retrieve-confirmation-time.py

@@ -0,0 +1,48 @@
+"""
+This migration retrieves additional metadata regarding chain confirmation of messages,
+including the block timestamp. We reset the TX height of the node to reprocess
+all the chain data messages and insert additional values
+"""
+
+
+import logging
+import os
+from configmanager import Config
+from aleph.model.chains import Chain
+from aleph.model.pending import PendingMessage, PendingTX
+from aleph.model.messages import Message
+
+logger = logging.getLogger(os.path.basename(__file__))
+
+
+async def upgrade(config: Config, **kwargs):
+    logger.info("Resetting chain height to re-fetch all chaindata...")
+    start_height = config.ethereum.start_height.value
+    await Chain.set_last_height("ETH", start_height)
+
+    logger.info("Dropping all pending transactions...")
+    await PendingTX.collection.delete_many({})
+
+    logger.info(
+        "Dropping all pending confirmation messages "
+        "(they will be reinserted automatically)..."
+    )
+    await PendingMessage.collection.delete_many({"source.chain_name": {"$ne": None}})
+
+    logger.info("Removing confirmation data for all messages...")
+    # Confirmations will be automatically added again by the pending TX processor.
+    # By removing the confirmation entirely, we make sure to avoid intermediate states
+    # if a message was confirmed in an unexpected way.
+    await Message.collection.update_many(
+        {"confirmed": True},
+        {
+            "$set": {
+                "confirmed": False,
+            },
+            "$unset": {"confirmations": 1},
+        },
+    )
+
+
+async def downgrade(**kwargs):
+    raise NotImplementedError("Downgrading this migration is not supported.")

src/aleph/chains/common.py

@@ -51,18 +51,17 @@ async def mark_confirmed_data(chain_name, tx_hash, height):
     }
 
 
-async def delayed_incoming(message, chain_name=None, tx_hash=None, height=None):
+async def delayed_incoming(
+    message, tx_context: Optional[TxContext] = None, check_message: bool = True
+):
     if message is None:
         return
+
     await PendingMessage.collection.insert_one(
         {
             "message": message,
-            "source": dict(
-                chain_name=chain_name,
-                tx_hash=tx_hash,
-                height=height,
-                check_message=True,  # should we store this?
-            ),
+            "tx_context": asdict(tx_context) if tx_context is not None else None,
+            "check_message": check_message,
         }
     )
 
@@ -75,25 +74,13 @@ class IncomingStatus(IntEnum):
 
 async def mark_message_for_retry(
     message: Dict,
-    chain_name: Optional[str],
-    tx_hash: Optional[str],
-    height: Optional[int],
+    tx_context: Optional[TxContext],
     check_message: bool,
     retrying: bool,
     existing_id,
 ):
     if not retrying:
-        await PendingMessage.collection.insert_one(
-            {
-                "message": message,
-                "source": dict(
-                    chain_name=chain_name,
-                    tx_hash=tx_hash,
-                    height=height,
-                    check_message=check_message,  # should we store this?
-                ),
-            }
-        )
+        await delayed_incoming(message, tx_context, check_message)
     else:
         LOGGER.debug(f"Incrementing for {existing_id}")
         result = await PendingMessage.collection.update_one(
@@ -104,9 +91,7 @@ async def mark_message_for_retry(
 
 async def incoming(
     message: Dict,
-    chain_name: Optional[str] = None,
-    tx_hash: Optional[str] = None,
-    height: Optional[int] = None,
+    tx_context: Optional[TxContext] = None,
     seen_ids: Optional[Dict[Tuple, int]] = None,
     check_message: bool = False,
     retrying: bool = False,
@@ -120,11 +105,12 @@ async def incoming(
 
     item_hash = message["item_hash"]
     sender = message["sender"]
+    chain_name = tx_context.chain if tx_context is not None else None
     ids_key = (item_hash, sender, chain_name)
 
-    if chain_name and tx_hash and height and seen_ids is not None:
+    if tx_context is not None and seen_ids is not None:
         if ids_key in seen_ids.keys():
-            if height > seen_ids[ids_key]:
+            if tx_context.height > seen_ids[ids_key]:
                 return IncomingStatus.MESSAGE_HANDLED, []
 
     filters = {
@@ -143,7 +129,7 @@ async def incoming(
             # check/sanitize the message if needed
             try:
                 message = await check_message_fn(
-                    message, from_chain=(chain_name is not None)
+                    message, from_chain=(tx_context is not None)
                 )
             except InvalidMessageError:
                 return IncomingStatus.FAILED_PERMANENTLY, []
@@ -164,16 +150,14 @@ async def incoming(
     #     existing = await Message.collection.find_one(
     #         filters, projection={'confirmed': 1, 'confirmations': 1, 'time': 1})
 
-    if chain_name and tx_hash and height:
+    if tx_context is not None:
         # We are getting a confirmation here
-        new_values = await mark_confirmed_data(chain_name, tx_hash, height)
-
         updates = {
             "$set": {
                 "confirmed": True,
             },
             "$min": {"time": message["time"]},
-            "$addToSet": {"confirmations": new_values["confirmations"][0]},
+            "$addToSet": {"confirmations": asdict(tx_context)},
         }
     else:
         updates = {
@@ -186,14 +170,14 @@ async def incoming(
     # new_values = {'confirmed': False}  # this should be our default.
     should_commit = False
     if existing:
-        if seen_ids is not None and height is not None:
+        if seen_ids is not None and tx_context is not None:
             if ids_key in seen_ids.keys():
-                if height > seen_ids[ids_key]:
+                if tx_context.height > seen_ids[ids_key]:
                     return IncomingStatus.MESSAGE_HANDLED, []
                 else:
-                    seen_ids[ids_key] = height
+                    seen_ids[ids_key] = tx_context.height
             else:
-                seen_ids[ids_key] = height
+                seen_ids[ids_key] = tx_context.height
 
         # THIS CODE SHOULD BE HERE...
         # But, if a race condition appeared, we might have the message twice.
@@ -203,7 +187,7 @@ async def incoming(
 
         LOGGER.debug("Updating %s." % item_hash)
 
-        if chain_name and tx_hash and height:
+        if tx_context is not None:
             # we need to update messages adding the confirmation
             # await Message.collection.update_many(filters, updates)
             should_commit = True
@@ -224,9 +208,7 @@ async def incoming(
                 LOGGER.exception("Can't get content of object %r" % item_hash)
             await mark_message_for_retry(
                 message=message,
-                chain_name=chain_name,
-                tx_hash=tx_hash,
-                height=height,
+                tx_context=tx_context,
                 check_message=check_message,
                 retrying=retrying,
                 existing_id=existing_id,
@@ -266,9 +248,7 @@ async def incoming(
             LOGGER.debug("Message type handler has failed, retrying later.")
             await mark_message_for_retry(
                 message=message,
-                chain_name=chain_name,
-                tx_hash=tx_hash,
-                height=height,
+                tx_context=tx_context,
                 check_message=check_message,
                 retrying=retrying,
                 existing_id=existing_id,
@@ -286,14 +266,14 @@ async def incoming(
             LOGGER.warning("Invalid sender for %s" % item_hash)
             return IncomingStatus.MESSAGE_HANDLED, []
 
-        if seen_ids is not None and height is not None:
+        if seen_ids is not None and tx_context is not None:
             if ids_key in seen_ids.keys():
-                if height > seen_ids[ids_key]:
+                if tx_context.height > seen_ids[ids_key]:
                     return IncomingStatus.MESSAGE_HANDLED, []
                 else:
-                    seen_ids[ids_key] = height
+                    seen_ids[ids_key] = tx_context.height
             else:
-                seen_ids[ids_key] = height
+                seen_ids[ids_key] = tx_context.height
 
         LOGGER.debug("New message to store for %s." % item_hash)
         # message.update(new_values)

src/aleph/chains/ethereum.py

@@ -196,8 +196,8 @@ async def request_transactions(
         try:
             jdata = json.loads(message)
             context = TxContext(
-                chain_name=CHAIN_NAME,
-                tx_hash=event_data.transactionHash.hex(),
+                chain=CHAIN_NAME,
+                hash=event_data.transactionHash.hex(),
                 time=timestamp,
                 height=event_data.blockNumber,
                 publisher=publisher,

src/aleph/chains/nuls2.py

@@ -174,8 +174,8 @@ async def request_transactions(config, session, start_height) -> AsyncIterator[T
             jdata = json.loads(ddata)
 
             context = TxContext(
-                chain_name=CHAIN_NAME,
-                tx_hash=tx["hash"],
+                chain=CHAIN_NAME,
+                hash=tx["hash"],
                 height=tx["height"],
                 time=tx["createTime"],
                 publisher=tx["coinFroms"][0]["address"],

src/aleph/chains/tx_context.py

@@ -3,8 +3,8 @@
 
 @dataclass
 class TxContext:
-    chain_name: str
-    tx_hash: str
+    chain: str
+    hash: str
     height: int
     # Transaction timestamp, in Unix time (number of seconds since epoch).
     time: int

src/aleph/jobs/process_pending_messages.py

@@ -19,6 +19,7 @@
 from aleph.services.p2p import singleton
 from aleph.types import ItemType
 from .job_utils import prepare_loop, gather_and_perform_db_operations
+from ..chains.tx_context import TxContext
 
 LOGGER = getLogger("jobs.pending_messages")
 
@@ -27,13 +28,17 @@ async def handle_pending_message(
     pending: Dict,
     seen_ids: Dict[Tuple, int],
 ) -> List[DbBulkOperation]:
+
+    try:
+        tx_context = TxContext(**pending["tx_context"])
+    except KeyError:
+        tx_context = None
+
     status, operations = await incoming(
         pending["message"],
-        chain_name=pending["source"].get("chain_name"),
-        tx_hash=pending["source"].get("tx_hash"),
-        height=pending["source"].get("height"),
+        tx_context=tx_context,
         seen_ids=seen_ids,
-        check_message=pending["source"].get("check_message", True),
+        check_message=pending.get("check_message", True),
         retrying=True,
         existing_id=pending["_id"],
     )

src/aleph/jobs/process_pending_txs.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import logging
+from dataclasses import asdict
 from typing import List, Dict, Optional
 
 import sentry_sdk
@@ -15,12 +16,12 @@
 from aleph.chains.tx_context import TxContext
 from aleph.exceptions import InvalidMessageError
 from aleph.logging import setup_logging
+from aleph.model.db_bulk_operation import DbBulkOperation
 from aleph.model.pending import PendingMessage, PendingTX
 from aleph.network import check_message
 from aleph.services.p2p import singleton
-from .job_utils import prepare_loop, gather_and_perform_db_operations
-from aleph.model.db_bulk_operation import DbBulkOperation
 from aleph.toolkit.batch import async_batch
+from .job_utils import prepare_loop, gather_and_perform_db_operations
 
 LOGGER = logging.getLogger("jobs.pending_txs")
 
@@ -31,7 +32,7 @@ async def handle_pending_tx(
 
     db_operations: List[DbBulkOperation] = []
     tx_context = TxContext(**pending_tx["context"])
-    LOGGER.info("%s Handling TX in block %s", tx_context.chain_name, tx_context.height)
+    LOGGER.info("%s Handling TX in block %s", tx_context.chain, tx_context.height)
 
     messages = await get_chaindata_messages(
         pending_tx["content"], tx_context, seen_ids=seen_ids
@@ -55,12 +56,8 @@ async def handle_pending_tx(
                     operation=InsertOne(
                         {
                             "message": message,
-                            "source": dict(
-                                chain_name=tx_context.chain_name,
-                                tx_hash=tx_context.tx_hash,
-                                height=tx_context.height,
-                                check_message=True,  # should we store this?
-                            ),
+                            "tx_context": asdict(tx_context),
+                            "check_message": True,
                         }
                     ),
                 )