dbsync: check collateral outputs

cardano_node_tests/tests/test_plutus_v2_spend_build.py

@@ -2575,7 +2575,7 @@ def test_with_total_return_collateral(
         return_collateral_utxos = cluster.get_utxo(tx_raw_output=tx_output_redeem)
 
         # when total collateral amount is specified, it is necessary to specify also return
-        # collateral `TxOut`
+        # collateral `TxOut` to get the change otherwise all collaterals will be collected
         if use_total_collateral and not use_return_collateral:
             assert not return_collateral_utxos, "Return collateral UTxO was unexpectedly created"
             return
@@ -2606,6 +2606,8 @@ def test_with_total_return_collateral(
             utxos=return_collateral_utxos
         )
 
+        collateral_charged = amount_for_collateral - returned_collateral_amount
+
         if use_return_collateral:
             assert (
                 returned_collateral_amount == return_collateral_amount
@@ -2615,7 +2617,6 @@ def test_with_total_return_collateral(
             ), "Return collateral address doesn't match the specified address"
         else:
             # check that the collateral amount charged corresponds to 'collateralPercentage'
-            collateral_charged = amount_for_collateral - returned_collateral_amount
             assert collateral_charged == round(
                 tx_output_redeem.fee * protocol_params["collateralPercentage"] / 100
             ), "The collateral amount charged is not the expected amount"
@@ -2627,6 +2628,12 @@ def test_with_total_return_collateral(
         # check "transaction view"
         tx_view.check_tx_view(cluster_obj=cluster, tx_raw_output=tx_output_redeem)
 
+        dbsync_utils.check_tx_phase_2_failure(
+            cluster_obj=cluster,
+            tx_raw_output=tx_output_redeem,
+            collateral_charged=collateral_charged,
+        )
+
     @allure.link(helpers.get_vcs_link())
     @pytest.mark.dbsync
     def test_collateral_with_tokens(
@@ -2754,6 +2761,12 @@ def test_collateral_with_tokens(
             ), "Token is missing from tx view return collateral"
             assert tx_view_token_rec[tx_view_asset_key] == token_amount, "Incorrect token amount"
 
+        dbsync_utils.check_tx_phase_2_failure(
+            cluster_obj=cluster,
+            tx_raw_output=tx_output_redeem,
+            collateral_charged=amount_for_collateral - return_collateral_amount,
+        )
+
 
 @pytest.mark.testnets
 class TestCompatibility:

cardano_node_tests/utils/dbsync_queries.py

@@ -72,6 +72,7 @@ class TxDBRow(NamedTuple):
     withdrawal_count: int
     collateral_count: int
     reference_input_count: int
+    collateral_out_count: int
     script_count: int
     redeemer_count: int
     ma_tx_out_id: Optional[int]
@@ -147,6 +148,14 @@ class TxInNoMADBRow(NamedTuple):
     tx_hash: memoryview
 
 
+class CollateralTxOutDBRow(NamedTuple):
+    tx_out_id: int
+    utxo_ix: int
+    address: str
+    value: decimal.Decimal
+    tx_hash: memoryview
+
+
 class ScriptDBRow(NamedTuple):
     id: int
     tx_id: int
@@ -286,6 +295,7 @@ def query_tx(txhash: str) -> Generator[TxDBRow, None, None]:
         " (SELECT COUNT(id) FROM withdrawal WHERE tx_id=tx.id) AS withdrawal_count,"
         " (SELECT COUNT(id) FROM collateral_tx_in WHERE tx_in_id=tx.id) AS collateral_count,"
         " (SELECT COUNT(id) FROM reference_tx_in WHERE tx_in_id=tx.id) AS reference_input_count,"
+        " (SELECT COUNT(id) FROM collateral_tx_out WHERE tx_id=tx.id) AS collateral_out_count,"
         " (SELECT COUNT(id) FROM script WHERE tx_id=tx.id) AS script_count,"
         " (SELECT COUNT(id) FROM redeemer WHERE tx_id=tx.id) AS redeemer_count,"
         " ma_tx_out.id, join_ma_out.policy, join_ma_out.name, ma_tx_out.quantity,"
@@ -365,6 +375,23 @@ def query_reference_tx_ins(txhash: str) -> Generator[TxInNoMADBRow, None, None]:
             yield TxInNoMADBRow(*result)
 
 
+def query_collateral_tx_outs(txhash: str) -> Generator[CollateralTxOutDBRow, None, None]:
+    """Query transaction collateral txouts in db-sync."""
+    query = (
+        "SELECT "
+        "collateral_tx_out.id, collateral_tx_out.index, collateral_tx_out.address, "
+        "collateral_tx_out.value, "
+        "(SELECT hash FROM tx WHERE id = collateral_tx_out.tx_id) AS tx_hash "
+        "FROM collateral_tx_out "
+        "LEFT JOIN tx ON tx.id = collateral_tx_out.tx_id "
+        "WHERE tx.hash = %s;"
+    )
+
+    with execute(query=query, vars=(rf"\x{txhash}",)) as cur:
+        while (result := cur.fetchone()) is not None:
+            yield CollateralTxOutDBRow(*result)
+
+
 def query_plutus_scripts(txhash: str) -> Generator[ScriptDBRow, None, None]:
     """Query transaction plutus scripts in db-sync."""
     query = (

cardano_node_tests/utils/dbsync_utils.py

@@ -140,6 +140,7 @@ class TxRecord(NamedTuple):
     txouts: List[UTxORecord]
     mint: List[UTxORecord]
     collaterals: List[clusterlib.UTXOData]
+    collateral_outputs: List[clusterlib.UTXOData]
     reference_inputs: List[clusterlib.UTXOData]
     scripts: List[ScriptRecord]
     redeemers: List[RedeemerRecord]
@@ -165,8 +166,8 @@ class TxPrelimRecord(NamedTuple):
     last_row: dbsync_queries.TxDBRow
 
 
-def utxodata2txout(utxodata: UTxORecord) -> clusterlib.TxOut:
-    """Convert `UTxORecord` to `clusterlib.TxOut`."""
+def utxodata2txout(utxodata: Union[UTxORecord, clusterlib.UTXOData]) -> clusterlib.TxOut:
+    """Convert `UTxORecord` or `UTxOData` to `clusterlib.TxOut`."""
     return clusterlib.TxOut(
         address=utxodata.address,
         amount=utxodata.amount,
@@ -435,6 +436,7 @@ def get_tx_record(txhash: str) -> TxRecord:  # noqa: C901
 
     Compile data from multiple SQL queries to get as much information about the TX as possible.
     """
+    # pylint: disable=too-many-branches
     txdata = get_prelim_tx_record(txhash)
     txins = get_txins(txhash)
 
@@ -507,6 +509,18 @@ def get_tx_record(txhash: str) -> TxRecord:  # noqa: C901
             for r in dbsync_queries.query_collateral_tx_ins(txhash=txhash)
         ]
 
+    collateral_outputs = []
+    if txdata.last_row.collateral_out_count:
+        collateral_outputs = [
+            clusterlib.UTXOData(
+                utxo_hash=r.tx_hash.hex(),
+                utxo_ix=int(r.utxo_ix),
+                amount=int(r.value),
+                address=str(r.address),
+            )
+            for r in dbsync_queries.query_collateral_tx_outs(txhash=txhash)
+        ]
+
     reference_inputs = []
     if txdata.last_row.reference_input_count:
         reference_inputs = [
@@ -562,6 +576,7 @@ def get_tx_record(txhash: str) -> TxRecord:  # noqa: C901
         txouts=[*txdata.utxo_out, *txdata.ma_utxo_out],
         mint=txdata.mint_utxo_out,
         collaterals=collaterals,
+        collateral_outputs=collateral_outputs,
         reference_inputs=reference_inputs,
         scripts=scripts,
         redeemers=redeemers,
@@ -837,6 +852,16 @@ def check_tx(
         tx_collaterals == db_collaterals
     ), f"TX collaterals don't match ({tx_collaterals} != {db_collaterals})"
 
+    if tx_collaterals:
+        tx_collaterals_amount = sum(col.amount for col in tx_collaterals)
+        db_collateral_output_amount = sum(col_out.amount for col_out in response.collateral_outputs)
+        protocol_params = cluster_obj.get_protocol_params()
+
+        assert db_collateral_output_amount == int(
+            tx_collaterals_amount
+            - tx_raw_output.fee * protocol_params["collateralPercentage"] / 100
+        )
+
     db_plutus_scripts = {r for r in response.scripts if r.type == "plutus"}
     # a script is added to `script` table only the first time it is seen, so the record
     # can be empty for the current transaction
@@ -921,6 +946,55 @@ def check_tx(
     return response
 
 
+def check_tx_phase_2_failure(
+    cluster_obj: clusterlib.ClusterLib,
+    tx_raw_output: clusterlib.TxRawOutput,
+    collateral_charged: int,
+    retry_num: int = 3,
+) -> Optional[TxRecord]:
+    """Check a transaction in db-sync when a phase 2 failure happens."""
+    if not configuration.HAS_DBSYNC:
+        return None
+
+    txhash = cluster_obj.get_txid(tx_body_file=tx_raw_output.out_file)
+    response = get_tx_record_retry(txhash=txhash, retry_num=retry_num)
+
+    # In case of a phase 2 failure, the collateral output, becomes the output of the tx.
+
+    assert (
+        not response.collateral_outputs
+    ), "Collaterals outputs present on dbsync when a tx have a phase 2 failure"
+
+    db_txouts = {utxodata2txout(r) for r in response.txouts}
+    tx_out = {utxodata2txout(r) for r in cluster_obj.get_utxo(tx_raw_output=tx_raw_output)}
+
+    assert db_txouts == tx_out, f"The output tx is not the expected ({db_txouts} != {tx_out})"
+
+    if tx_raw_output.return_collateral_txouts:
+        tx_return_collaterals = {
+            _sanitize_txout(cluster_obj=cluster_obj, txout=r)
+            for r in tx_raw_output.return_collateral_txouts
+        }
+
+        assert (
+            tx_out == tx_return_collaterals
+        ), f"The tx output is not the expected ({tx_out} != {tx_return_collaterals})"
+
+    # In case of a phase 2 failure, the fees charged is the collateral amount
+
+    if tx_raw_output.total_collateral_amount:
+        expected_fee = round(tx_raw_output.total_collateral_amount)
+    elif tx_raw_output.return_collateral_txouts and not tx_raw_output.total_collateral_amount:
+        expected_fee = collateral_charged
+    else:
+        protocol_params = cluster_obj.get_protocol_params()
+        expected_fee = round(tx_raw_output.fee * protocol_params["collateralPercentage"] / 100)
+
+    assert response.fee == expected_fee, f"TX fee doesn't match ({response.fee} != {expected_fee})"
+
+    return response
+
+
 def check_pool_deregistration(pool_id: str, retiring_epoch: int) -> Optional[PoolDataRecord]:
     """Check pool retirement in db-sync."""
     if not configuration.HAS_DBSYNC: