[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

@@ -54,18 +54,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,
         }
     )
 
@@ -78,25 +77,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(
@@ -124,9 +111,7 @@ def update_message_item_type(message_dict: Dict) -> Dict:
 
 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,
@@ -144,11 +129,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 = {
@@ -167,7 +153,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, []
@@ -188,16 +174,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 = {
@@ -210,14 +194,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.
@@ -227,7 +211,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
@@ -248,9 +232,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,
@@ -290,9 +272,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,
@@ -310,14 +290,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

@@ -197,8 +197,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

@@ -175,8 +175,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.model.pending import PendingMessage
 from aleph.services.p2p import singleton
 from .job_utils import prepare_loop, process_job_results
+from ..chains.tx_context import TxContext
 
 LOGGER = getLogger("jobs.pending_messages")
 
@@ -48,12 +49,15 @@ async def handle_pending_message(
     seen_ids: Dict[Tuple, int],
 ) -> List[DbBulkOperation]:
 
+    try:
+        tx_context = TxContext(**pending["tx_context"])
+    except KeyError:
+        tx_context = None
+
     async with sem:
         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),
             retrying=True,

src/aleph/jobs/process_pending_txs.py

@@ -4,7 +4,9 @@
 
 import asyncio
 import logging
-from typing import List, Dict, Optional, Set
+from dataclasses import asdict
+from typing import List, Dict, Optional
+from typing import Set
 
 import sentry_sdk
 from configmanager import Config
@@ -31,7 +33,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 +57,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,
                         }
                     ),
                 )

tests/chains/test_confirmation.py

@@ -1,12 +1,13 @@
 import json
+from dataclasses import asdict
 from typing import Dict
 
 import pytest
 
 from aleph.chains.common import process_one_message
+from aleph.chains.tx_context import TxContext
 from aleph.model.messages import CappedMessage, Message
 
-
 MESSAGE = {
     "chain": "ETH",
     "sender": "0x971300C78A38e0F85E60A3b04ae3fA70b4276B64",
@@ -56,18 +57,21 @@ async def test_confirm_message(test_db):
     assert remove_id_key(message_in_db) == remove_id_key(capped_message_in_db)
 
     # Now, confirm the message
-    chain_name, tx_hash, height = "ETH", "123", 8000
-    await process_one_message(
-        MESSAGE, chain_name=chain_name, tx_hash=tx_hash, height=height
+    tx_context = TxContext(
+        chain="ETH",
+        hash="123",
+        height=8000,
+        time=120000,
+        publisher="0xdeadbeef",
     )
+    await process_one_message(MESSAGE, tx_context)
 
     message_in_db = await Message.collection.find_one({"item_hash": item_hash})
 
     assert message_in_db is not None
     assert message_in_db["confirmed"]
-    assert {"chain": chain_name, "hash": tx_hash, "height": height} in message_in_db[
-        "confirmations"
-    ]
+    expected_confirmations = [asdict(tx_context)]
+    assert message_in_db["confirmations"] == expected_confirmations
 
     capped_message_after_confirmation = await CappedMessage.collection.find_one(
         {"item_hash": item_hash}
@@ -89,17 +93,21 @@ async def test_process_confirmed_message(test_db):
     item_hash = MESSAGE["item_hash"]
 
     # Now, confirm the message
-    chain_name, tx_hash, height = "ETH", "123", 8000
-    await process_one_message(
-        MESSAGE, chain_name=chain_name, tx_hash=tx_hash, height=height
+    tx_context = TxContext(
+        chain="ETH",
+        hash="123",
+        height=8000,
+        time=120000,
+        publisher="0xdeadbeef",
     )
+    await process_one_message(MESSAGE, tx_context=tx_context)
 
     message_in_db = await Message.collection.find_one({"item_hash": item_hash})
 
     assert message_in_db is not None
     assert message_in_db["confirmed"]
 
-    expected_confirmations = [{"chain": chain_name, "hash": tx_hash, "height": height}]
+    expected_confirmations = [asdict(tx_context)]
     assert message_in_db["confirmations"] == expected_confirmations
 
     capped_message_in_db = await CappedMessage.collection.find_one(
@@ -108,4 +116,4 @@ async def test_process_confirmed_message(test_db):
 
     assert remove_id_key(message_in_db) == remove_id_key(capped_message_in_db)
     assert capped_message_in_db["confirmed"]
-    assert capped_message_in_db["confirmations"] == expected_confirmations
+    assert capped_message_in_db["confirmations"] == [asdict(tx_context)]