diff --git a/backend/consensus/base.py b/backend/consensus/base.py index 846dd9279..4137aa6ba 100644 --- a/backend/consensus/base.py +++ b/backend/consensus/base.py @@ -1197,18 +1197,66 @@ async def process_validator_appeal( await self.rollback_transactions(context) # Get the previous state of the contract + previous_contract_state = None if context.transaction.contract_snapshot: - previous_contact_state = ( + previous_contract_state = ( context.transaction.contract_snapshot.states["accepted"] ) + elif ( + context.transaction.type == TransactionType.DEPLOY_CONTRACT + ): + # Rolling back a deploy: clear the contract state + previous_contract_state = {} else: - previous_contact_state = {} + # Defense in depth: the in-memory transaction may have + # been built without the stored contract_snapshot. + # Re-fetch it instead of clobbering the contract state + # with {} (which would wipe the code slot). + refetched = ( + context.transactions_processor.get_transaction_by_hash( + context.transaction.hash + ) + ) + refetched_snapshot = ContractSnapshot.from_dict( + (refetched or {}).get("contract_snapshot") + ) + if refetched_snapshot: + previous_contract_state = refetched_snapshot.states[ + "accepted" + ] + else: + from loguru import logger + + logger.error( + f"Missing contract_snapshot for appealed " + f"transaction {context.transaction.hash}; " + f"skipping contract state restore" + ) + # Surface to monitoring: the appeal succeeded but + # the contract kept the appealed transaction's + # state — recoverable, but needs operator eyes. + context.msg_handler.send_message( + LogEvent( + "consensus_event", + EventType.ERROR, + EventScope.CONSENSUS, + "Missing contract_snapshot on successful " + "validator appeal; contract state restore " + "skipped", + { + "transaction_hash": context.transaction.hash, + "contract_address": context.transaction.to_address, + }, + transaction_hash=context.transaction.hash, + ) + ) # Restore the contract state - context.contract_processor.update_contract_state( - context.transaction.to_address, - accepted_state=previous_contact_state, - ) + if previous_contract_state is not None: + context.contract_processor.update_contract_state( + context.transaction.to_address, + accepted_state=previous_contract_state, + ) # Always clear snapshot on successful appeal (including timeout appeals) # so re-execution loads fresh state from DB @@ -1869,6 +1917,18 @@ class PendingState(TransactionState): """ async def handle(self, context): + # Refresh transaction from DB FIRST. The claim-path dict that built + # context.transaction omits columns whose Transaction.from_dict + # defaults are silently wrong for re-processed appeals (e.g. + # appeal_undetermined/appeal_leader_timeout default to False) — the + # same drift class as the PR #1724 state wipe. Nothing in this state + # may read claim-built appeal fields before this refresh. + context.transaction = Transaction.from_dict( + context.transactions_processor.get_transaction_by_hash( + context.transaction.hash + ) + ) + # Pre-effects: timestamp + reset rotation count pre_effects = decide_pending_pre( tx_hash=context.transaction.hash, @@ -1877,13 +1937,6 @@ async def handle(self, context): ) await EffectExecutor(context).execute(pre_effects) - # Refresh transaction from DB - context.transaction = Transaction.from_dict( - context.transactions_processor.get_transaction_by_hash( - context.transaction.hash - ) - ) - # Log executing message (unless appeal) if ( not context.transaction.appeal_leader_timeout diff --git a/backend/consensus/worker.py b/backend/consensus/worker.py index de8490596..bcd29c2e2 100644 --- a/backend/consensus/worker.py +++ b/backend/consensus/worker.py @@ -33,6 +33,82 @@ # stop-the-worker circuit breaker (see _transaction_context). _TRANSIENT_LEADER_FATAL_CAUSES = frozenset({"NO_PROVIDER_FOR_PROMPT"}) +# --------------------------------------------------------------------------- +# Claim column manifest +# +# Single source of truth for what a claim query RETURNs and what the rebuilt +# transaction dict contains. Both the RETURNING clause and the row->dict +# conversion are generated from these tuples, so the two can never drift +# apart. Drift is exactly what caused the contract-state wipe fixed in +# PR #1724: claim_next_appeal's hand-written RETURNING list omitted +# contract_snapshot/consensus_history and Transaction.from_dict silently +# defaulted them, so a successful validator appeal "restored" the contract +# state to {}. +# +# Entries are (sql_column, dict_key). dict_key differs from the column name +# only for triggered_by_hash, which Transaction.from_dict consumes as +# "triggered_by". +# --------------------------------------------------------------------------- +_TX_CLAIM_BASE_COLUMNS: tuple[tuple[str, str], ...] = ( + ("hash", "hash"), + ("from_address", "from_address"), + ("to_address", "to_address"), + ("data", "data"), + ("value", "value"), + ("type", "type"), + ("nonce", "nonce"), + ("gaslimit", "gaslimit"), + ("r", "r"), + ("s", "s"), + ("v", "v"), + ("leader_only", "leader_only"), + ("execution_mode", "execution_mode"), + ("sim_config", "sim_config"), + ("status", "status"), + ("consensus_data", "consensus_data"), + ("input_data", "input_data"), + ("created_at", "created_at"), + ("blocked_at", "blocked_at"), + ("triggered_by_hash", "triggered_by"), +) + +# Stored per-transaction state. Heavy: contract_snapshot has been observed at +# tens of MB, so only claims whose downstream consumes it include this group. +# Appeals restore the contract's accepted state from it; the pending and +# finalization paths rebuild snapshots via contract_snapshot_factory and +# persist consensus_history through DB-side jsonb merges, so they omit it +# deliberately (see _TX_CLAIM_OMISSIONS in the manifest tests). +_TX_STATE_COLUMNS: tuple[tuple[str, str], ...] = ( + ("contract_snapshot", "contract_snapshot"), + ("consensus_history", "consensus_history"), +) + +_TX_APPEAL_COLUMNS: tuple[tuple[str, str], ...] = ( + ("appealed", "appealed"), + ("appeal_failed", "appeal_failed"), + ("timestamp_appeal", "timestamp_appeal"), + ("appeal_undetermined", "appeal_undetermined"), + ("appeal_leader_timeout", "appeal_leader_timeout"), + ("appeal_validators_timeout", "appeal_validators_timeout"), +) + +_TX_FINALIZATION_COLUMNS: tuple[tuple[str, str], ...] = ( + ("timestamp_awaiting_finalization", "timestamp_awaiting_finalization"), + ("appeal_failed", "appeal_failed"), +) + + +def _tx_returning_clause(*column_groups: tuple[tuple[str, str], ...]) -> str: + """Render a RETURNING column list from manifest groups.""" + columns = [col for group in column_groups for col, _ in group] + return ", ".join(f"transactions.{col}" for col in columns) + + +def _tx_row_to_dict(row: Any, *column_groups: tuple[tuple[str, str], ...]) -> dict: + """Build the claimed-transaction dict from the same manifest groups that + generated the query's RETURNING clause.""" + return {key: getattr(row, col) for group in column_groups for col, key in group} + class ConsensusWorker: """ @@ -223,7 +299,7 @@ async def claim_next_finalization(self, session: Session) -> Optional[dict]: # They must be in ACCEPTED/UNDETERMINED/TIMEOUT states and appeal window must have passed start_time = time.perf_counter() query = text( - """ + f""" WITH locked_finalizations AS ( SELECT t.* FROM transactions t @@ -285,13 +361,7 @@ async def claim_next_finalization(self, session: Session) -> Optional[dict]: worker_id = :worker_id FROM single_finalization WHERE transactions.hash = single_finalization.hash - RETURNING transactions.hash, transactions.from_address, transactions.to_address, - transactions.data, transactions.value, transactions.type, transactions.nonce, - transactions.gaslimit, transactions.r, transactions.s, transactions.v, - transactions.leader_only, transactions.execution_mode, transactions.sim_config, - transactions.status, transactions.consensus_data, - transactions.input_data, transactions.created_at, transactions.timestamp_awaiting_finalization, - transactions.appeal_failed, transactions.blocked_at, transactions.triggered_by_hash; + RETURNING {_tx_returning_clause(_TX_CLAIM_BASE_COLUMNS, _TX_FINALIZATION_COLUMNS)}; """ ) @@ -313,31 +383,9 @@ async def claim_next_finalization(self, session: Session) -> Optional[dict]: f"[Worker {self.worker_id}] Claimed next finalization result {result.hash}" ) session.commit() - # Convert result to dict - return { - "hash": result.hash, - "from_address": result.from_address, - "to_address": result.to_address, - "data": result.data, - "value": result.value, - "type": result.type, - "nonce": result.nonce, - "gaslimit": result.gaslimit, - "r": result.r, - "s": result.s, - "v": result.v, - "leader_only": result.leader_only, - "execution_mode": result.execution_mode, - "sim_config": result.sim_config, - "status": result.status, - "consensus_data": result.consensus_data, - "input_data": result.input_data, - "created_at": result.created_at, - "timestamp_awaiting_finalization": result.timestamp_awaiting_finalization, - "appeal_failed": result.appeal_failed, - "blocked_at": result.blocked_at, - "triggered_by": result.triggered_by_hash, - } + return _tx_row_to_dict( + result, _TX_CLAIM_BASE_COLUMNS, _TX_FINALIZATION_COLUMNS + ) return None @@ -352,7 +400,7 @@ async def claim_next_appeal(self, session: Session) -> Optional[dict]: # Query to atomically claim an appealed transaction start_time = time.perf_counter() query = text( - """ + f""" WITH locked_appeals AS ( SELECT t.hash, t.to_address, t.created_at FROM transactions t @@ -393,16 +441,7 @@ async def claim_next_appeal(self, session: Session) -> Optional[dict]: worker_id = :worker_id FROM single_appeal WHERE transactions.hash = single_appeal.hash - RETURNING transactions.hash, transactions.from_address, transactions.to_address, - transactions.data, transactions.value, transactions.type, transactions.nonce, - transactions.gaslimit, transactions.r, transactions.s, transactions.v, - transactions.leader_only, transactions.execution_mode, transactions.sim_config, - transactions.status, transactions.consensus_data, - transactions.input_data, transactions.created_at, transactions.appealed, - transactions.appeal_failed, transactions.timestamp_appeal, - transactions.appeal_undetermined, transactions.appeal_leader_timeout, - transactions.appeal_validators_timeout, transactions.blocked_at, - transactions.triggered_by_hash; + RETURNING {_tx_returning_clause(_TX_CLAIM_BASE_COLUMNS, _TX_STATE_COLUMNS, _TX_APPEAL_COLUMNS)}; """ ) @@ -418,35 +457,9 @@ async def claim_next_appeal(self, session: Session) -> Optional[dict]: if result: session.commit() - # Convert result to dict - return { - "hash": result.hash, - "from_address": result.from_address, - "to_address": result.to_address, - "data": result.data, - "value": result.value, - "type": result.type, - "nonce": result.nonce, - "gaslimit": result.gaslimit, - "r": result.r, - "s": result.s, - "v": result.v, - "leader_only": result.leader_only, - "execution_mode": result.execution_mode, - "sim_config": result.sim_config, - "status": result.status, - "consensus_data": result.consensus_data, - "input_data": result.input_data, - "created_at": result.created_at, - "appealed": result.appealed, - "appeal_failed": result.appeal_failed, - "timestamp_appeal": result.timestamp_appeal, - "appeal_undetermined": result.appeal_undetermined, - "appeal_leader_timeout": result.appeal_leader_timeout, - "appeal_validators_timeout": result.appeal_validators_timeout, - "blocked_at": result.blocked_at, - "triggered_by": result.triggered_by_hash, - } + return _tx_row_to_dict( + result, _TX_CLAIM_BASE_COLUMNS, _TX_STATE_COLUMNS, _TX_APPEAL_COLUMNS + ) return None @@ -462,7 +475,7 @@ async def claim_next_transaction(self, session: Session) -> Optional[dict]: # Ensures only one transaction per contract is processed at a time start_time = time.perf_counter() query = text( - """ + f""" WITH candidate_transactions AS ( SELECT t.hash, t.to_address, t.type, t.created_at, t.recovery_count FROM transactions t @@ -542,13 +555,7 @@ async def claim_next_transaction(self, session: Session) -> Optional[dict]: worker_id = :worker_id FROM single_transaction WHERE transactions.hash = single_transaction.hash - RETURNING transactions.hash, transactions.from_address, transactions.to_address, - transactions.data, transactions.value, transactions.type, transactions.nonce, - transactions.gaslimit, transactions.r, transactions.s, transactions.v, - transactions.leader_only, transactions.execution_mode, transactions.sim_config, - transactions.status, transactions.consensus_data, - transactions.input_data, transactions.created_at, transactions.blocked_at, - transactions.triggered_by_hash; + RETURNING {_tx_returning_clause(_TX_CLAIM_BASE_COLUMNS)}; """ ) @@ -567,29 +574,7 @@ async def claim_next_transaction(self, session: Session) -> Optional[dict]: if result: logger.debug(f"[Worker {self.worker_id}] Claimed transaction {result.hash}") session.commit() - # Convert result to dict - return { - "hash": result.hash, - "from_address": result.from_address, - "to_address": result.to_address, - "data": result.data, - "value": result.value, - "type": result.type, - "nonce": result.nonce, - "gaslimit": result.gaslimit, - "r": result.r, - "s": result.s, - "v": result.v, - "leader_only": result.leader_only, - "execution_mode": result.execution_mode, - "sim_config": result.sim_config, - "status": result.status, - "consensus_data": result.consensus_data, - "input_data": result.input_data, - "created_at": result.created_at, - "blocked_at": result.blocked_at, - "triggered_by": result.triggered_by_hash, - } + return _tx_row_to_dict(result, _TX_CLAIM_BASE_COLUMNS) return None diff --git a/tests/unit/consensus/test_appeal_state_restore.py b/tests/unit/consensus/test_appeal_state_restore.py new file mode 100644 index 000000000..06d2ed7f5 --- /dev/null +++ b/tests/unit/consensus/test_appeal_state_restore.py @@ -0,0 +1,326 @@ +"""Tests for contract state restoration on a successful validator appeal. + +Regression tests for the bug where `ConsensusWorker.claim_next_appeal` +omitted `transactions.contract_snapshot` and `transactions.consensus_history` +from its RETURNING list, so every appealed transaction was rebuilt with +`contract_snapshot=None` and an empty `consensus_history`. On +VALIDATOR_APPEAL_SUCCESSFUL, `process_validator_appeal` then fell into the +"no snapshot" branch and restored `accepted_state = {}` for non-deploy +transactions — wiping the contract's accepted state, including the code slot. + +Two layers are covered: +1. `claim_next_appeal` must return the two columns (worker.py fix). +2. `process_validator_appeal` must never restore `{}` for a non-deploy + transaction that is missing its snapshot; it re-fetches from the DB and, + failing that, skips the restore entirely (base.py defense in depth). +""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +import backend.consensus.base as consensus_base +from backend.consensus.base import ConsensusAlgorithm +from backend.consensus.types import ConsensusRound +from backend.consensus.worker import ConsensusWorker +from backend.database_handler.contract_snapshot import ContractSnapshot +from backend.database_handler.transactions_processor import TransactionStatus +from backend.domain.types import Transaction, TransactionType + +TX_HASH = "0xtxhash" +CONTRACT_ADDRESS = "0xcontract" + + +def _make_snapshot_dict(accepted=None): + return { + "contract_address": CONTRACT_ADDRESS, + "states": {"accepted": accepted or {}, "finalized": {}}, + "balance": 0, + } + + +def _make_transaction(*, tx_type=TransactionType.RUN_CONTRACT, contract_snapshot=None): + """Build a minimal appealed Transaction for process_validator_appeal.""" + tx = Mock(spec=Transaction) + tx.hash = TX_HASH + tx.type = tx_type + tx.to_address = CONTRACT_ADDRESS + tx.contract_snapshot = contract_snapshot + tx.consensus_data = Mock() + tx.consensus_data.leader_receipt = None + tx.consensus_history = {} + tx.appeal_failed = 0 + tx.appealed = True + tx.status = TransactionStatus.ACCEPTED + return tx + + +def _make_algorithm(): + """ConsensusAlgorithm without __init__ (which requires env vars / live deps).""" + algorithm = ConsensusAlgorithm.__new__(ConsensusAlgorithm) + algorithm.msg_handler = Mock() + algorithm.consensus_service = Mock() + algorithm.genvm_manager = Mock() + algorithm.rollback_transactions = AsyncMock() + return algorithm + + +async def _run_validator_appeal( + algorithm, transaction, transactions_processor, contract_processor, monkeypatch +): + """Drive process_validator_appeal straight into VALIDATOR_APPEAL_SUCCESSFUL.""" + monkeypatch.setattr( + ConsensusAlgorithm, + "get_extra_validators", + staticmethod(Mock(return_value=(None, [{"address": "0xvalidator"}]))), + ) + monkeypatch.setattr( + ConsensusAlgorithm, + "dispatch_transaction_status_update", + staticmethod(AsyncMock()), + ) + committing_state = Mock() + committing_state.handle = AsyncMock( + return_value=ConsensusRound.VALIDATOR_APPEAL_SUCCESSFUL + ) + monkeypatch.setattr( + consensus_base, "CommittingState", Mock(return_value=committing_state) + ) + + validators_snapshot = Mock() + validators_snapshot.nodes = [] + + factory_snapshot = ContractSnapshot.from_dict( + _make_snapshot_dict({"factory_slot": "factory_data"}) + ) + + await algorithm.process_validator_appeal( + transaction=transaction, + transactions_processor=transactions_processor, + chain_snapshot=None, + accounts_manager=Mock(), + contract_snapshot_factory=Mock(return_value=factory_snapshot), + contract_processor=contract_processor, + node_factory=Mock(), + validators_snapshot=validators_snapshot, + ) + + +class TestValidatorAppealStateRestore: + """Successful validator appeal must not wipe contract state.""" + + @pytest.mark.asyncio + async def test_missing_snapshot_non_deploy_refetches_from_db(self, monkeypatch): + """No in-memory snapshot on a RUN_CONTRACT tx: restore from the DB row, + not from `{}`.""" + stored_accepted = {"code_slot": "b64code", "storage_slot": "value"} + transactions_processor = Mock() + transactions_processor.get_transaction_by_hash.return_value = { + "hash": TX_HASH, + "contract_snapshot": _make_snapshot_dict(stored_accepted), + } + contract_processor = Mock() + + transaction = _make_transaction(contract_snapshot=None) + await _run_validator_appeal( + _make_algorithm(), + transaction, + transactions_processor, + contract_processor, + monkeypatch, + ) + + transactions_processor.get_transaction_by_hash.assert_called_once_with(TX_HASH) + contract_processor.update_contract_state.assert_called_once_with( + CONTRACT_ADDRESS, accepted_state=stored_accepted + ) + + @pytest.mark.asyncio + async def test_missing_snapshot_non_deploy_unfetchable_skips_restore( + self, monkeypatch + ): + """No in-memory snapshot and no DB row either: skip the restore instead + of writing `{}` (which would delete the contract's code slot).""" + transactions_processor = Mock() + transactions_processor.get_transaction_by_hash.return_value = None + contract_processor = Mock() + + transaction = _make_transaction(contract_snapshot=None) + await _run_validator_appeal( + _make_algorithm(), + transaction, + transactions_processor, + contract_processor, + monkeypatch, + ) + + transactions_processor.get_transaction_by_hash.assert_called_once_with(TX_HASH) + contract_processor.update_contract_state.assert_not_called() + + @pytest.mark.asyncio + async def test_missing_snapshot_non_deploy_row_without_snapshot_skips_restore( + self, monkeypatch + ): + """DB row exists but carries no snapshot: still skip the restore.""" + transactions_processor = Mock() + transactions_processor.get_transaction_by_hash.return_value = { + "hash": TX_HASH, + "contract_snapshot": None, + } + contract_processor = Mock() + + transaction = _make_transaction(contract_snapshot=None) + await _run_validator_appeal( + _make_algorithm(), + transaction, + transactions_processor, + contract_processor, + monkeypatch, + ) + + contract_processor.update_contract_state.assert_not_called() + + @pytest.mark.asyncio + async def test_missing_snapshot_deploy_still_clears_state(self, monkeypatch): + """DEPLOY_CONTRACT with no snapshot keeps the pre-fix behavior: rolling + back a deploy legitimately clears the contract state.""" + transactions_processor = Mock() + contract_processor = Mock() + + transaction = _make_transaction( + tx_type=TransactionType.DEPLOY_CONTRACT, contract_snapshot=None + ) + await _run_validator_appeal( + _make_algorithm(), + transaction, + transactions_processor, + contract_processor, + monkeypatch, + ) + + contract_processor.update_contract_state.assert_called_once_with( + CONTRACT_ADDRESS, accepted_state={} + ) + transactions_processor.get_transaction_by_hash.assert_not_called() + + @pytest.mark.asyncio + async def test_in_memory_snapshot_restores_saved_state(self, monkeypatch): + """When the transaction carries its snapshot, restore from it directly.""" + saved_accepted = {"code_slot": "b64code", "counter": "0x01"} + snapshot = ContractSnapshot.from_dict(_make_snapshot_dict(saved_accepted)) + transactions_processor = Mock() + contract_processor = Mock() + + transaction = _make_transaction(contract_snapshot=snapshot) + await _run_validator_appeal( + _make_algorithm(), + transaction, + transactions_processor, + contract_processor, + monkeypatch, + ) + + contract_processor.update_contract_state.assert_called_once_with( + CONTRACT_ADDRESS, accepted_state=saved_accepted + ) + transactions_processor.get_transaction_by_hash.assert_not_called() + + +def _make_worker(): + """ConsensusWorker without __init__ (which spins up polling infrastructure).""" + worker = ConsensusWorker.__new__(ConsensusWorker) + worker.worker_id = "worker-test" + worker.transaction_timeout_minutes = 20 + worker._log_query_result = Mock() + return worker + + +def _make_appeal_row(contract_snapshot, consensus_history): + """A row shaped like the RETURNING clause of claim_next_appeal.""" + return SimpleNamespace( + hash=TX_HASH, + from_address="0xsender", + to_address=CONTRACT_ADDRESS, + data=None, + value=0, + type=TransactionType.RUN_CONTRACT.value, + nonce=1, + gaslimit=0, + r=None, + s=None, + v=None, + leader_only=False, + execution_mode="NORMAL", + sim_config=None, + status=TransactionStatus.ACCEPTED.value, + consensus_data=None, + contract_snapshot=contract_snapshot, + consensus_history=consensus_history, + input_data=None, + created_at=None, + appealed=True, + appeal_failed=0, + timestamp_appeal=None, + appeal_undetermined=False, + appeal_leader_timeout=False, + appeal_validators_timeout=False, + blocked_at=None, + triggered_by_hash=None, + ) + + +class TestClaimNextAppealReturnsSnapshot: + """claim_next_appeal must surface contract_snapshot and consensus_history.""" + + @pytest.mark.asyncio + async def test_returned_dict_includes_snapshot_and_history(self): + stored_snapshot = _make_snapshot_dict({"code_slot": "b64code"}) + stored_history = {"consensus_results": [{"consensus_round": "Accepted"}]} + row = _make_appeal_row(stored_snapshot, stored_history) + + session = Mock() + session.execute.return_value.first.return_value = row + + worker = _make_worker() + result = await worker.claim_next_appeal(session) + + assert result is not None + assert result["contract_snapshot"] == stored_snapshot + assert result["consensus_history"] == stored_history + session.commit.assert_called_once() + + @pytest.mark.asyncio + async def test_returning_clause_selects_snapshot_and_history_columns(self): + session = Mock() + session.execute.return_value.first.return_value = None + + worker = _make_worker() + result = await worker.claim_next_appeal(session) + + assert result is None + executed_sql = str(session.execute.call_args[0][0]) + returning_clause = executed_sql.split("RETURNING", 1)[1] + assert "transactions.contract_snapshot" in returning_clause + assert "transactions.consensus_history" in returning_clause + + @pytest.mark.asyncio + async def test_claimed_appeal_hydrates_transaction_snapshot(self): + """End-to-end over the claimed dict: Transaction.from_dict must produce + a non-None contract_snapshot and the stored consensus_history — the + exact properties process_validator_appeal depends on.""" + stored_accepted = {"code_slot": "b64code", "storage_slot": "value"} + stored_snapshot = _make_snapshot_dict(stored_accepted) + stored_history = {"consensus_results": [{"consensus_round": "Accepted"}]} + row = _make_appeal_row(stored_snapshot, stored_history) + + session = Mock() + session.execute.return_value.first.return_value = row + + worker = _make_worker() + appeal_data = await worker.claim_next_appeal(session) + + transaction = Transaction.from_dict(appeal_data) + assert transaction.contract_snapshot is not None + assert transaction.contract_snapshot.states["accepted"] == stored_accepted + assert transaction.consensus_history == stored_history diff --git a/tests/unit/consensus/test_claim_column_manifest.py b/tests/unit/consensus/test_claim_column_manifest.py new file mode 100644 index 000000000..7744df4eb --- /dev/null +++ b/tests/unit/consensus/test_claim_column_manifest.py @@ -0,0 +1,221 @@ +"""Guards against claim-query / transaction-dict drift. + +The bug class: each worker claim query hand-maintained a RETURNING column +list plus a parallel row->dict conversion, and ``Transaction.from_dict`` +silently defaults every missing key. A column omitted from one list was +therefore indistinguishable from a legitimately-absent value — which is how +claim_next_appeal shipped without contract_snapshot/consensus_history and a +successful validator appeal wiped contract state (PR #1724). + +Three layers of defense, in order of strength: + +1. RETURNING clauses and row->dict conversion are now *generated* from the + shared column manifest in ``backend.consensus.worker`` — they cannot + drift from each other. Tests here pin that each claim's executed SQL + really contains its manifest. +2. The incident pin: the appeal claim manifest must include + contract_snapshot and consensus_history. +3. The class-killer: every key ``Transaction.from_dict`` consumes must be + either provided by a claim's manifest or listed in that claim's explicit + omission allowlist below. Adding a new field to ``Transaction.from_dict`` + fails this test until someone makes a *conscious* per-claim decision. +""" + +import inspect +import re + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from backend.consensus.worker import ( + ConsensusWorker, + _TX_APPEAL_COLUMNS, + _TX_CLAIM_BASE_COLUMNS, + _TX_FINALIZATION_COLUMNS, + _TX_STATE_COLUMNS, + _tx_returning_clause, + _tx_row_to_dict, +) +from backend.domain.types import Transaction + +# --------------------------------------------------------------------------- +# Per-claim manifests, mirroring the groups each claim function uses. +# --------------------------------------------------------------------------- +CLAIM_MANIFESTS = { + "claim_next_transaction": (_TX_CLAIM_BASE_COLUMNS,), + "claim_next_finalization": (_TX_CLAIM_BASE_COLUMNS, _TX_FINALIZATION_COLUMNS), + "claim_next_appeal": ( + _TX_CLAIM_BASE_COLUMNS, + _TX_STATE_COLUMNS, + _TX_APPEAL_COLUMNS, + ), +} + +# --------------------------------------------------------------------------- +# Explicit omission allowlists (dict-key names as Transaction.from_dict +# consumes them). Every entry is a conscious decision: the from_dict default +# must be CORRECT for that claim's downstream, not merely tolerated. +# --------------------------------------------------------------------------- + +# Fields recomputed during processing or purely informational, whose +# from_dict defaults are correct for every claim path: +_COMMON_OMISSIONS = { + "appeal_processing_time", # accounting value, only read for finality math from DB + "config_rotation_rounds", # config, re-read from DB where consumed + "num_of_initial_validators", # config, re-read from DB where consumed + "last_vote_timestamp", # monitoring bookkeeping, rewritten during processing + "rotation_count", # reset on PENDING entry, tracked in-context afterwards + "leader_timeout_validators", # rebuilt by the timeout-appeal flow from DB + "origin_address", # informational; not consumed by worker flows +} + +CLAIM_ALLOWED_OMISSIONS = { + # Pending/activated claims: PendingState.handle refreshes the full + # transaction from the DB before any field is consumed, so claim-dict + # staleness cannot leak (see the refresh-first comment in PendingState). + # contract_snapshot is deliberately NOT returned here: it can weigh tens + # of MB and the pending path rebuilds snapshots via + # contract_snapshot_factory instead. + "claim_next_transaction": _COMMON_OMISSIONS + | { + "contract_snapshot", + "consensus_history", + "appealed", + "appeal_failed", + "appeal_undetermined", + "appeal_leader_timeout", + "appeal_validators_timeout", + "timestamp_appeal", + "timestamp_awaiting_finalization", + }, + # Finalization: FinalizingState reads current contract state and merges + # consensus_history via DB-side jsonb updates; the heavy state columns + # are deliberately omitted. + "claim_next_finalization": _COMMON_OMISSIONS + | { + "contract_snapshot", + "consensus_history", + "appealed", + "appeal_undetermined", + "appeal_leader_timeout", + "appeal_validators_timeout", + "timestamp_appeal", + }, + # Appeals restore contract state from the claimed row — nothing + # state-bearing may be omitted here. + "claim_next_appeal": _COMMON_OMISSIONS | {"timestamp_awaiting_finalization"}, +} + + +def _manifest_keys(groups) -> set: + return {key for group in groups for _, key in group} + + +def _manifest_columns(groups) -> list: + return [col for group in groups for col, _ in group] + + +def _from_dict_consumed_keys() -> set: + """Statically extract every dict key Transaction.from_dict reads.""" + source = inspect.getsource(Transaction.from_dict) + keys = set(re.findall(r"input\.get\(\s*[\"'](\w+)[\"']", source)) + keys |= set(re.findall(r"input\[[\"'](\w+)[\"']\]", source)) + assert keys, "failed to extract keys from Transaction.from_dict source" + return keys + + +def _make_worker(): + worker = ConsensusWorker.__new__(ConsensusWorker) + worker.worker_id = "worker-test" + worker.transaction_timeout_minutes = 20 + worker._log_query_result = Mock() + worker.consensus_algorithm = Mock( + finality_window_time=1800, finality_window_appeal_failed_reduction=0.2 + ) + return worker + + +class TestManifestGeneratesQueries: + """Each claim's executed SQL must contain exactly its manifest columns.""" + + @pytest.mark.asyncio + @pytest.mark.parametrize("claim_name", sorted(CLAIM_MANIFESTS)) + async def test_returning_clause_matches_manifest(self, claim_name): + session = Mock() + session.execute.return_value.first.return_value = None + + worker = _make_worker() + result = await getattr(worker, claim_name)(session) + assert result is None + + executed_sql = str(session.execute.call_args[0][0]) + returning_clause = executed_sql.split("RETURNING", 1)[1] + for column in _manifest_columns(CLAIM_MANIFESTS[claim_name]): + assert ( + f"transactions.{column}" in returning_clause + ), f"{claim_name} RETURNING is missing manifest column {column!r}" + + def test_row_to_dict_covers_manifest(self): + for claim_name, groups in CLAIM_MANIFESTS.items(): + columns = _manifest_columns(groups) + row = SimpleNamespace(**{col: f"value-{col}" for col in columns}) + data = _tx_row_to_dict(row, *groups) + assert set(data) == _manifest_keys(groups), claim_name + # Every value must come from its column, aliases included. + for group in groups: + for col, key in group: + assert data[key] == f"value-{col}" + + def test_returning_clause_renders_all_groups(self): + clause = _tx_returning_clause( + _TX_CLAIM_BASE_COLUMNS, _TX_STATE_COLUMNS, _TX_APPEAL_COLUMNS + ) + assert "transactions.contract_snapshot" in clause + assert "transactions.consensus_history" in clause + assert "transactions.triggered_by_hash" in clause + + +class TestIncidentPins: + """Regression pins for the PR #1724 contract-state wipe.""" + + def test_appeal_claim_includes_state_columns(self): + keys = _manifest_keys(CLAIM_MANIFESTS["claim_next_appeal"]) + assert "contract_snapshot" in keys + assert "consensus_history" in keys + + def test_triggered_by_alias_preserved(self): + # Transaction.from_dict consumes "triggered_by", not the SQL column + # name "triggered_by_hash". + keys = _manifest_keys((_TX_CLAIM_BASE_COLUMNS,)) + assert "triggered_by" in keys + assert "triggered_by_hash" not in keys + + +class TestFromDictCoverage: + """The class-killer: silent from_dict defaults require explicit opt-out. + + If this test fails after you added a field to Transaction.from_dict, + decide FOR EACH claim whether the query must return the new column + (add it to a manifest group) or whether the from_dict default is truly + correct for that claim's downstream (add it to the claim's allowlist + with a rationale comment). Do not add blanket entries. + """ + + @pytest.mark.parametrize("claim_name", sorted(CLAIM_MANIFESTS)) + def test_consumed_keys_covered_or_explicitly_omitted(self, claim_name): + consumed = _from_dict_consumed_keys() + provided = _manifest_keys(CLAIM_MANIFESTS[claim_name]) + allowed = CLAIM_ALLOWED_OMISSIONS[claim_name] + + unaccounted = consumed - provided - allowed + assert not unaccounted, ( + f"{claim_name}: Transaction.from_dict consumes {sorted(unaccounted)} " + f"but the claim neither returns them nor allowlists the omission" + ) + + # Allowlists must not rot: entries that are actually provided, or that + # from_dict no longer consumes, must be removed. + stale = (allowed & provided) | (allowed - consumed) + assert not stale, f"{claim_name}: stale allowlist entries {sorted(stale)}"