diff --git a/backend/consensus/base.py b/backend/consensus/base.py index 0a7c0b340..38c9c728d 100644 --- a/backend/consensus/base.py +++ b/backend/consensus/base.py @@ -21,7 +21,7 @@ from sqlalchemy import text from sqlalchemy.orm import Session from backend.consensus.constants import DEFAULT_VALIDATORS_COUNT -from backend.consensus.errors import NoValidatorsAvailableError +from backend.consensus.errors import FinalizationStateError, NoValidatorsAvailableError from backend.consensus.vrf import get_validators_for_transaction from backend.database_handler.chain_snapshot import ChainSnapshot from backend.database_handler.contract_snapshot import ContractSnapshot @@ -52,6 +52,7 @@ Vote, ExecutionResultStatus, PendingTransaction, + _compute_contract_state_hash, ) from backend.protocol_rpc.message_handler.base import MessageHandler from backend.protocol_rpc.message_handler.types import ( @@ -380,6 +381,43 @@ def node_factory( ) +def upgrade_code_slot( + new_code: str, genvm_executor_selector: str | None +) -> tuple[str, str]: + """Storage key and value a code upgrade writes for this contract.""" + code_bytes = new_code.encode("utf-8") + code_len_prefix = len(code_bytes).to_bytes(4, byteorder="little", signed=False) + key = base64.b64encode( + get_code_slot(legacy=uses_legacy_storage(genvm_executor_selector)) + ).decode("ascii") + return key, base64.b64encode(code_len_prefix + code_bytes).decode("ascii") + + +def with_later_upgrade_code( + transactions_processor: TransactionsProcessor, + transaction_hash: str, + state: dict[str, str], +) -> dict[str, str]: + """Keep the code of a FINALIZED upgrade that overtook this older decision.""" + upgrade = transactions_processor.get_latest_finalized_upgrade_after( + transaction_hash + ) + if not upgrade or not upgrade.get("new_code"): + return state + key, value = upgrade_code_slot( + upgrade["new_code"], upgrade.get("genvm_executor_selector") + ) + return {**state, key: value} + + +def _closed_lifecycle(transaction: dict) -> bool: + return ( + transaction.get("status") + in (TransactionStatus.FINALIZED.value, TransactionStatus.CANCELED.value) + or transaction.get("type") == TransactionType.UPGRADE_CONTRACT.value + ) + + def transaction_genvm_executor_selector(transaction: Transaction) -> str | None: """Select and persist the deployment's executor, honoring explicit pins.""" explicit_selector = ( @@ -1410,6 +1448,17 @@ async def process_validator_appeal( ) ) + if ( + previous_contract_state is not None + and context.transaction.type + != TransactionType.DEPLOY_CONTRACT + ): + previous_contract_state = with_later_upgrade_code( + context.transactions_processor, + context.transaction.hash, + previous_contract_state, + ) + # Restore the contract state if previous_contract_state is not None: context.contract_processor.update_contract_state( @@ -1449,11 +1498,14 @@ async def rollback_transactions(self, context: TransactionContext): Rollback newer transactions. In the simplified system, we just need to reset future transactions to PENDING. """ - # Set all transactions with higher created_at to PENDING - future_transactions = context.transactions_processor.get_newer_transactions( - context.transaction.hash + future_transactions = ( + context.transactions_processor.get_rewindable_newer_transactions( + context.transaction.hash + ) ) for future_transaction in future_transactions: + if _closed_lifecycle(future_transaction): + continue future_hash = future_transaction["hash"] future_data = deepcopy(future_transaction.get("data") or {}) if bool(future_transaction.get("appealed")) or ( @@ -3727,10 +3779,8 @@ async def handle(self, context): ), has_contract_snapshot=bool(context.transaction.contract_snapshot), contract_snapshot_dict=( - context.contract_snapshot.to_dict() - if not context.transaction.contract_snapshot - else None - ), + context.transaction.contract_snapshot or context.contract_snapshot + ).to_dict(), execution_result_success=execution_success, tx_type_deploy=is_deploy, accepted_contract_state=accepted_contract_state, @@ -3883,6 +3933,83 @@ async def handle(self, context): return None +FINALIZATION_SUCCESSOR_LOOKUP_LIMIT = 3 + + +def _recovery_pre_states( + transactions_processor, tx_hash: str +) -> Iterable[tuple[str, dict | None]]: + yield "own pre-state", transactions_processor.get_transaction_pre_state(tx_hash) + successors = transactions_processor.get_successor_snapshot_hashes( + tx_hash, FINALIZATION_SUCCESSOR_LOOKUP_LIMIT + ) + for position, successor_hash in enumerate(successors, start=1): + yield ( + f"successor {position} ({successor_hash})", + transactions_processor.get_transaction_pre_state(successor_hash), + ) + + +def resolve_finalization_state( + context: TransactionContext, leader_receipt: Receipt +) -> tuple[dict[str, str], str]: + """Return the transaction's hash-verified post-state and where it came from.""" + saved_snapshot = context.transaction.contract_snapshot + state = saved_snapshot.finalization_state if saved_snapshot is not None else None + source = "saved state" + if state is None: + state = context.transactions_processor.get_transaction_finalization_state( + context.transaction.hash + ) + if state is None and leader_receipt.contract_state: + state = leader_receipt.contract_state + source = "receipt" + + expected_hash = leader_receipt.contract_state_hash + if state is not None: + if expected_hash and _compute_contract_state_hash(state) != expected_hash: + raise FinalizationStateError( + f"Finalization state hash mismatch for transaction {context.transaction.hash}" + ) + return state, source + + # Pre-fix rows did not persist post-state. The moving accepted head is only + # safe to reuse when the original receipt proves it is exactly this result. + if expected_hash: + snapshot = context.contract_snapshot_factory(context.transaction.to_address) + accepted_state = ( + snapshot.states.get("accepted") if snapshot is not None else None + ) + if _compute_contract_state_hash(accepted_state) == expected_hash: + return accepted_state, "head" + + # A later tx's saved pre-state is this tx's post-state unless something intervened. + for source, pre_state in _recovery_pre_states( + context.transactions_processor, context.transaction.hash + ): + if _compute_contract_state_hash(pre_state) == expected_hash: + return pre_state, source + + raise FinalizationStateError( + f"Missing transaction-local finalization state for {context.transaction.hash}; " + "cannot finalize from an unverified contract head" + ) + + +def _transaction_finalization_state( + context: TransactionContext, leader_receipt: Receipt +) -> dict[str, str]: + state, source = resolve_finalization_state(context, leader_receipt) + if source not in ("saved state", "receipt"): + from loguru import logger + + logger.info( + f"Recovered legacy finalization state for {context.transaction.hash} " + f"from {source}" + ) + return state + + class FinalizingState(TransactionState): """ Class representing the finalizing state of a transaction. @@ -3897,12 +4024,6 @@ async def handle(self, context): ) leader_receipt = leader_receipts[0] if leader_receipts else None - # Acceptance is a separate message lifecycle phase. If the helper EVM - # or worker died after the agreed decision was committed, repair that - # phase before allowing finalization to overtake it. - if leader_receipt is not None: - _dispatch_messages_for_phase(context, leader_receipt, "accepted") - pre_effects, post_effects, should_finalize_contract = decide_finalizing( tx_hash=context.transaction.hash, tx_status_accepted=( @@ -3917,6 +4038,19 @@ async def handle(self, context): ), ) + # Verify state before emitting messages, changing status or settling fees. + finalization_state = ( + _transaction_finalization_state(context, leader_receipt) + if should_finalize_contract + else None + ) + + # Acceptance is a separate message lifecycle phase. If the helper EVM + # or worker died after the agreed decision was committed, repair that + # phase before allowing finalization to overtake it. + if leader_receipt is not None: + _dispatch_messages_for_phase(context, leader_receipt, "accepted") + executor = EffectExecutor(context) await executor.execute(pre_effects) @@ -3938,21 +4072,13 @@ async def handle(self, context): # Impure: contract finalization + triggered transactions (needs DB reads) if should_finalize_contract: - snapshot = context.contract_snapshot_factory(context.transaction.to_address) - if snapshot is None: - raise RuntimeError( - "Missing contract snapshot while finalizing a transaction" - ) - - accepted_state = snapshot.states.get("accepted") - if not accepted_state: - raise RuntimeError( - "Missing accepted contract state prior to finalization" - ) - context.contract_processor.update_contract_state( context.transaction.to_address, - finalized_state=accepted_state, + finalized_state=with_later_upgrade_code( + context.transactions_processor, + context.transaction.hash, + finalization_state, + ), ) _dispatch_messages_for_phase( diff --git a/backend/consensus/decisions.py b/backend/consensus/decisions.py index 4cac5c283..4d5a71775 100644 --- a/backend/consensus/decisions.py +++ b/backend/consensus/decisions.py @@ -401,12 +401,14 @@ def decide_accepted( # ── Contract write decision ── if not appealed: - # Save contract snapshot for rollback - if not has_contract_snapshot: + # Keep the rollback pre-state and this decision's post-state together. + # A successful recomputation can follow a saved timeout snapshot. + if not has_contract_snapshot or execution_result_success: + snapshot = dict(contract_snapshot_dict or {}) + if execution_result_success: + snapshot["finalization_state"] = accepted_contract_state pre_effects.append( - SetContractSnapshotEffect( - tx_hash=tx_hash, snapshot_dict=contract_snapshot_dict - ) + SetContractSnapshotEffect(tx_hash=tx_hash, snapshot_dict=snapshot) ) if execution_result_success: diff --git a/backend/consensus/errors.py b/backend/consensus/errors.py index 4597cafc5..5cf9a673c 100644 --- a/backend/consensus/errors.py +++ b/backend/consensus/errors.py @@ -1,2 +1,6 @@ class NoValidatorsAvailableError(Exception): """Raised when no validators are available to process a transaction.""" + + +class FinalizationStateError(RuntimeError): + """An agreed transaction lacks a verifiable post-state; preserve its decision.""" diff --git a/backend/consensus/worker.py b/backend/consensus/worker.py index c4d0232b0..750a62e42 100644 --- a/backend/consensus/worker.py +++ b/backend/consensus/worker.py @@ -2,10 +2,11 @@ import os import asyncio +import json import time import uuid from contextlib import asynccontextmanager -from typing import Callable, Optional, Any +from typing import Callable, Optional, Any, Sequence from sqlalchemy.orm import Session from sqlalchemy import text, bindparam @@ -15,17 +16,27 @@ from backend.database_handler.errors import ContractNotFoundError from backend.domain.types import Transaction from backend.node.genvm.error_codes import GenVMInternalError +from backend.node.types import ExecutionResultStatus from backend.consensus.base import ( ConsensusAlgorithm, InternalMessageEmissionError, NoValidatorsAvailableError, + TransactionContext, + resolve_finalization_state, + upgrade_code_slot, ) from backend.consensus.history import APPEAL_RECOVERY_SNAPSHOT_KEY +from backend.consensus.errors import FinalizationStateError from backend.protocol_rpc.fees import acceptance_dispatch_pending # Alias for use in context manager (avoids circular import issues) _NoValidatorsError = NoValidatorsAvailableError from backend.protocol_rpc.message_handler.base import MessageHandler +from backend.protocol_rpc.message_handler.types import ( + EventScope, + EventType, + LogEvent, +) from backend.rollup.consensus_service import ConsensusService import backend.validators as validators from loguru import logger @@ -40,6 +51,25 @@ _TRANSIENT_LEADER_FATAL_CAUSES = frozenset({"NO_PROVIDER_FOR_PROMPT"}) +UPGRADE_CLOSE_KEY = "upgrade_close" + + +class _UpgradeDeferred(Exception): + """Same-contract work may still be running; try the upgrade again later.""" + + +class _UpgradeRejected(Exception): + """The upgrade can never apply; cancel it with this receipt data.""" + + def __init__(self, consensus_data: dict): + super().__init__(consensus_data.get("error")) + self.consensus_data = consensus_data + + +def _upgrade_failed(error: str, **details: Any) -> dict: + return {"upgrade_result": "failed", "error": error, **details} + + def _acceptance_dispatch_pending(transaction_data: dict[str, Any]) -> bool: data = transaction_data.get("data") if not isinstance(data, dict): @@ -96,8 +126,8 @@ def _acceptance_dispatch_pending(transaction_data: dict[str, Any]) -> bool: # 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 +# Appeals restore the contract's accepted state from it. Pending rebuilds its +# snapshot; finalization separately reads only the saved post-state. Both # 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], ...] = ( @@ -292,10 +322,17 @@ def _parse_max_parallel_txs(self) -> int: return parsed - async def claim_next_finalization(self, session: Session) -> Optional[dict]: + async def claim_next_finalization( + self, session: Session, exclude_hashes: Sequence[str] = () + ) -> Optional[dict]: """ Claim the next transaction that needs finalization (appeal window expired). + Args: + session: Database session + exclude_hashes: Hashes in retry backoff; their contracts are skipped + so another contract's head can be claimed instead. + Returns: Transaction data dict if claimed, None otherwise """ @@ -370,6 +407,15 @@ async def claim_next_finalization(self, session: Session) -> Optional[dict]: AND earlier.status NOT IN ('FINALIZED', 'CANCELED') AND earlier.hash != t.hash ) + AND NOT EXISTS ( + -- A backed-off tx holds back itself and every younger tx on its contract + SELECT 1 FROM transactions backed_off + WHERE backed_off.hash = ANY(CAST(:backoff_hashes AS text[])) + AND backed_off.to_address IS NOT DISTINCT FROM t.to_address + AND backed_off.queue_order <= t.queue_order + -- Stale entries for already-terminal txs must not block + AND backed_off.status NOT IN ('FINALIZED', 'CANCELED') + ) AND NOT EXISTS ( -- Ensure no other transaction for same contract is being processed SELECT 1 FROM transactions t2 @@ -417,6 +463,7 @@ async def claim_next_finalization(self, session: Session) -> Optional[dict]: "finality_window_seconds": self.consensus_algorithm.finality_window_time, "appeal_failed_reduction": self.consensus_algorithm.finality_window_appeal_failed_reduction, "stranded_threshold_seconds": STRANDED_TX_AFTER_SECONDS, + "backoff_hashes": list(exclude_hashes), }, ).first() duration = time.perf_counter() - start_time @@ -972,7 +1019,9 @@ async def _transaction_context( e, cancel_on_exhaustion=( not protected_appeal_work - and not isinstance(e, InternalMessageEmissionError) + and not isinstance( + e, (InternalMessageEmissionError, FinalizationStateError) + ) ), ) finally: @@ -1629,6 +1678,7 @@ async def _handle_generic_error_retry( error: Exception, *, cancel_on_exhaustion: bool = True, + cancel_details: Optional[dict] = None, ): """ Handle retry logic for generic errors during transaction processing. @@ -1665,6 +1715,7 @@ async def _handle_generic_error_retry( "error": "max_generic_retries_exceeded", "last_error": str(error), "retries": retry_info["count"], + **(cancel_details or {}), } # Refund sender for payable tx that was never activated if tx.value and tx.value > 0 and tx.from_address: @@ -1709,7 +1760,7 @@ async def _handle_generic_error_retry( + ( f"/{self.MAX_GENERIC_ERROR_RETRIES}" if cancel_on_exhaustion - else " (appeal preserved)" + else " (decision preserved)" ) + ", " f"next attempt in {backoff}s - error: {error}" @@ -1812,141 +1863,478 @@ async def _process_upgrade_transaction( ): """ Process a contract upgrade transaction (type=3). No consensus needed. - Directly updates contract code in database. + + The upgrade closes the contract's earlier lifecycle: decided + transactions are finalized through the normal finalization path + (pending paid appeals are refunded first), never-started ones stay + queued and run on the new code, and the code is applied last. Args: transaction_data: Transaction data dictionary containing new_code session: Database session """ - import base64 - from backend.node.genvm import get_code_slot - from backend.node.genvm.executor_selection import uses_legacy_storage - from backend.database_handler.models import CurrentState - tx_hash = transaction_data["hash"] + contract_address = transaction_data["to_address"] + logger.info( + f"[Worker {self.worker_id}] Processing upgrade transaction {tx_hash} for contract {contract_address}" + ) try: - contract_address = transaction_data["to_address"] + new_code = self._validate_upgrade(session, transaction_data) + decided = self._plan_upgrade_close(session, tx_hash, contract_address) data = transaction_data.get("data") or {} - new_code = data.get("new_code") - if not new_code: - raise ValueError("Missing new_code in transaction data") + progress = { + "closed_transactions": [], + "refunded_appeals": [], + **(data.get(UPGRADE_CLOSE_KEY) or {}), + } + for predecessor in decided: + await self._force_finalize_for_upgrade(tx_hash, predecessor, progress) + await self._apply_upgrade_code(session, tx_hash, contract_address, new_code) + except _UpgradeDeferred as e: + session.rollback() logger.info( - f"[Worker {self.worker_id}] Processing upgrade transaction {tx_hash} for contract {contract_address}" + f"[Worker {self.worker_id}] Deferring upgrade {tx_hash}: " + f"transaction {e} on {contract_address} may still be running" ) - - # Load contract - contract = ( - session.query(CurrentState).filter_by(id=contract_address).one_or_none() + # Keep the lease briefly so the upgrade is not re-claimed in a hot loop. + await asyncio.sleep(float(os.environ.get("UPGRADE_DEFER_SECONDS", "5"))) + except _UpgradeRejected as e: + logger.error( + f"[Worker {self.worker_id}] Contract upgrade rejected for {tx_hash}: {e}" + ) + session.rollback() + await self._cancel_upgrade(session, tx_hash, e.consensus_data) + except Exception as e: + logger.error( + f"[Worker {self.worker_id}] Contract upgrade failed for {tx_hash}: {e}", + exc_info=True, + ) + session.rollback() + # Finalized predecessors stay final; the code is applied on retry. + await self._handle_generic_error_retry( + tx_hash, e, cancel_details={"upgrade_result": "failed"} ) - if not contract: - raise ValueError(f"Contract {contract_address} not found") - # Validate contract has expected state structure + def _validate_upgrade(self, session: Session, transaction_data: dict) -> str: + from backend.database_handler.models import CurrentState + + contract_address = transaction_data["to_address"] + new_code = (transaction_data.get("data") or {}).get("new_code") + if not new_code: + raise _UpgradeRejected( + _upgrade_failed("Missing new_code in transaction data") + ) + contract = ( + session.query(CurrentState).filter_by(id=contract_address).one_or_none() + ) + if not contract: + raise _UpgradeRejected( + _upgrade_failed(f"Contract {contract_address} not found") + ) + if ( + not contract.data + or "state" not in contract.data + or "accepted" not in contract.data["state"] + or "finalized" not in contract.data["state"] + ): + raise _UpgradeRejected( + _upgrade_failed( + f"Contract {contract_address} has invalid state structure" + ) + ) + try: + compile(new_code, "", "exec") + except SyntaxError as e: + raise _UpgradeRejected( + _upgrade_failed(f"Invalid Python syntax: {e}") + ) from e + session.rollback() + return new_code + + def _plan_upgrade_close( + self, session: Session, upgrade_hash: str, contract_address: str + ) -> list[str]: + """Return the decided transactions to finalize before the upgrade applies.""" + rows = session.execute( + text( + """ + SELECT hash, status, blocked_at + FROM transactions + WHERE to_address = :contract + AND hash != :hash + AND status NOT IN ('FINALIZED', 'CANCELED') + ORDER BY queue_order ASC + FOR UPDATE + """ + ), + {"contract": contract_address, "hash": upgrade_hash}, + ).all() + for row in rows: + # A stale lease or mid-consensus row may belong to a slow live worker. + if row.blocked_at is not None or str(row.status) in ( + "PROPOSING", + "COMMITTING", + "REVEALING", + ): + raise _UpgradeDeferred(row.hash) + decided = [ + row.hash + for row in rows + if str(row.status) in self._FINALIZATION_ELIGIBLE_STATUSES + ] + + # Never touch a predecessor unless every one of them can finalize. + transactions_processor = TransactionsProcessor(session) + for predecessor in decided: + transaction = Transaction.from_dict( + transactions_processor.get_transaction_by_hash(predecessor) + ) + receipts = ( + transaction.consensus_data.leader_receipt + if transaction.consensus_data is not None + else None + ) + receipt = receipts[0] if receipts else None if ( - not contract.data - or "state" not in contract.data - or "accepted" not in contract.data["state"] - or "finalized" not in contract.data["state"] + transaction.status != TransactionStatus.ACCEPTED + or receipt is None + or receipt.execution_result != ExecutionResultStatus.SUCCESS ): - raise ValueError( - f"Contract {contract_address} has invalid state structure" + continue + try: + resolve_finalization_state( + self._finalization_context(transaction, session), receipt ) + except FinalizationStateError as e: + raise _UpgradeRejected( + _upgrade_failed( + "predecessor_finalization_state", transaction=predecessor + ) + ) from e + session.rollback() + return decided + + def _finalization_context( + self, transaction: Transaction, session: Session + ) -> TransactionContext: + from backend.consensus.base import ( + contract_snapshot_factory, + contract_processor_factory, + node_factory, + ) - # Validate Python syntax before proceeding - try: - compile(new_code, "", "exec") - except SyntaxError as e: - raise ValueError(f"Invalid Python syntax: {e}") from e - - # Encode code for slot storage: base64(4-byte-len-prefix + code-bytes) - code_bytes = new_code.encode("utf-8") - code_len_prefix = len(code_bytes).to_bytes( - 4, byteorder="little", signed=False - ) - code_slot_value = base64.b64encode(code_len_prefix + code_bytes).decode( - "ascii" + return TransactionContext( + transaction=transaction, + transactions_processor=TransactionsProcessor(session), + chain_snapshot=None, + accounts_manager=AccountsManager(session), + contract_snapshot_factory=lambda contract_address: contract_snapshot_factory( + contract_address, session, transaction + ), + contract_processor=contract_processor_factory(session), + node_factory=node_factory, + msg_handler=self.msg_handler, + consensus_service=self.consensus_service, + validators_snapshot=None, + genvm_manager=self.consensus_algorithm.genvm_manager, + ) + + def _record_upgrade_progress( + self, session: Session, upgrade_hash: str, progress: dict + ) -> None: + """Persist close progress and renew the lease; fail if the lease was lost.""" + owned = session.execute( + text( + """ + UPDATE transactions + SET blocked_at = NOW(), + data = jsonb_set( + COALESCE(data, '{}'::jsonb), + CAST(:path AS text[]), + CAST(:progress AS jsonb), + true + ) + WHERE hash = :hash + AND worker_id = :worker_id + AND status = 'PENDING' + RETURNING hash + """ + ), + { + "hash": upgrade_hash, + "worker_id": self.worker_id, + "path": [UPGRADE_CLOSE_KEY], + "progress": json.dumps(progress), + }, + ).first() + if owned is None: + raise RuntimeError( + f"Upgrade {upgrade_hash} is no longer leased by worker {self.worker_id}" ) - code_slot_key = base64.b64encode( - get_code_slot( - legacy=uses_legacy_storage(contract.genvm_executor_selector) + + async def _force_finalize_for_upgrade( + self, upgrade_hash: str, predecessor: str, progress: dict + ) -> None: + with self.get_session() as fin_session: + row = fin_session.execute( + text( + "SELECT status, appealed, data ? :snapshot_key AS has_snapshot" + " FROM transactions WHERE hash = :hash FOR UPDATE" + ), + {"hash": predecessor, "snapshot_key": APPEAL_RECOVERY_SNAPSHOT_KEY}, + ).first() + if row is None or str(row.status) not in ( + self._FINALIZATION_ELIGIBLE_STATUSES + ): + # Finalized by an earlier attempt. + fin_session.rollback() + return + + refund = None + if row.appealed or row.has_snapshot: + refund = AccountsManager(fin_session).abort_pending_appeal_admission( + predecessor, "upgrade_closed_lifecycle" + ) + if refund is not None: + progress["refunded_appeals"].append( + {"transaction": predecessor, "refund": refund} + ) + if predecessor not in progress["closed_transactions"]: + progress["closed_transactions"].append(predecessor) + # Same commit as the refund, so a retry never records or refunds twice. + self._record_upgrade_progress(fin_session, upgrade_hash, progress) + fin_session.commit() + if refund is not None: + self.msg_handler.send_message( + log_event=LogEvent( + "transaction_appeal_updated", + EventType.INFO, + EventScope.CONSENSUS, + "Appeal refunded: contract upgraded", + {"hash": predecessor}, + ), + log_to_terminal=False, ) - ).decode("ascii") - - # Update contract data - update BOTH accepted and finalized state - # Since upgrade transactions bypass consensus and go directly to FINALIZED, - # both state trees must be updated for reads to see the new code - contract.data = { - "state": { - "accepted": { - **contract.data["state"]["accepted"], - code_slot_key: code_slot_value, - }, - "finalized": { - **contract.data["state"]["finalized"], - code_slot_key: code_slot_value, - }, + + transaction_data = TransactionsProcessor( + fin_session + ).get_transaction_by_hash(predecessor) + transaction = Transaction.from_dict(transaction_data) + logger.info( + f"[Worker {self.worker_id}] Finalizing {predecessor} before upgrade {upgrade_hash}" + ) + transaction = await self._repair_accepted_delivery( + transaction, transaction_data, fin_session + ) + await self._finalize_transaction(transaction, fin_session) + fin_session.commit() + + await self.usage_metrics_service.send_finalized_transaction_metrics( + transaction, transaction_data + ) + + async def _apply_upgrade_code( + self, + session: Session, + tx_hash: str, + contract_address: str, + new_code: str, + ) -> None: + from backend.database_handler.models import CurrentState + + contract = ( + session.query(CurrentState) + .filter_by(id=contract_address) + .with_for_update() + .populate_existing() + .one_or_none() + ) + upgrade = ( + session.query(Transactions) + .filter_by(hash=tx_hash) + .with_for_update() + .populate_existing() + .one() + ) + if ( + upgrade.status != TransactionStatus.PENDING + or upgrade.worker_id != self.worker_id + ): + session.rollback() + logger.warning( + f"[Worker {self.worker_id}] Upgrade {tx_hash} is no longer owned and pending; not applying" + ) + return + if contract is None or "state" not in (contract.data or {}): + raise _UpgradeRejected( + _upgrade_failed(f"Contract {contract_address} not found") + ) + + still_open = session.execute( + text( + """ + SELECT hash FROM transactions + WHERE to_address = :contract + AND hash != :hash + AND status NOT IN ('FINALIZED', 'CANCELED') + AND ( + blocked_at IS NOT NULL + OR status IN :open_statuses + ) + LIMIT 1 + """ + ).bindparams(bindparam("open_statuses", expanding=True)), + { + "contract": contract_address, + "hash": tx_hash, + "open_statuses": [ + *self._FINALIZATION_ELIGIBLE_STATUSES, + "PROPOSING", + "COMMITTING", + "REVEALING", + ], + }, + ).first() + if still_open is not None: + raise RuntimeError( + f"Transaction {still_open.hash} reopened before upgrade {tx_hash} applied" + ) + + # A recompute must not restore its pre-upgrade appeal decision after a crash. + session.execute( + text( + """ + UPDATE transactions + SET data = data - :snapshot_key + WHERE to_address = :contract + AND hash != :hash + AND status IN ('PENDING', 'ACTIVATED') + AND data ? :snapshot_key + """ + ), + { + "contract": contract_address, + "hash": tx_hash, + "snapshot_key": APPEAL_RECOVERY_SNAPSHOT_KEY, + }, + ) + + code_slot_key, code_slot_value = upgrade_code_slot( + new_code, contract.genvm_executor_selector + ) + contract.data = { + "state": { + "accepted": { + **contract.data["state"]["accepted"], + code_slot_key: code_slot_value, }, - } + "finalized": { + **contract.data["state"]["finalized"], + code_slot_key: code_slot_value, + }, + }, + } + progress = (upgrade.data or {}).get(UPGRADE_CLOSE_KEY) or {} + upgrade.consensus_data = { + "upgrade_result": "success", + "contract_address": contract_address, + "closed_transactions": progress.get("closed_transactions", []), + "refunded_appeals": progress.get("refunded_appeals", []), + "applied_at": int(time.time()), + } + # The status update commits the code, receipt and FINALIZED together. + await ConsensusAlgorithm.dispatch_transaction_status_update( + TransactionsProcessor(session), + tx_hash, + TransactionStatus.FINALIZED, + self.msg_handler, + ) + self._generic_error_retries.pop(tx_hash, None) + logger.info( + f"[Worker {self.worker_id}] Contract {contract_address} upgraded successfully via tx {tx_hash}" + ) - # Store success in consensus_data for receipt + async def _cancel_upgrade( + self, session: Session, tx_hash: str, consensus_data: dict + ) -> None: + try: tx = session.query(Transactions).filter_by(hash=tx_hash).one() - tx.consensus_data = { - "upgrade_result": "success", - "contract_address": contract_address, - } - session.commit() - - # Mark transaction as finalized and send WebSocket notification - transactions_processor = TransactionsProcessor(session) + if tx.status != TransactionStatus.PENDING: + session.rollback() + return + tx.consensus_data = consensus_data await ConsensusAlgorithm.dispatch_transaction_status_update( - transactions_processor, + TransactionsProcessor(session), tx_hash, - TransactionStatus.FINALIZED, + TransactionStatus.CANCELED, self.msg_handler, ) - - logger.info( - f"[Worker {self.worker_id}] Contract {contract_address} upgraded successfully via tx {tx_hash}" - ) - - except Exception as e: + self._generic_error_retries.pop(tx_hash, None) + except Exception as update_error: + session.rollback() logger.error( - f"[Worker {self.worker_id}] Contract upgrade failed for {tx_hash}: {e}", - exc_info=True, + f"[Worker {self.worker_id}] Failed to update failed upgrade tx status: {update_error}" ) - session.rollback() - # Mark transaction as canceled on failure and send WebSocket notification - try: - tx = session.query(Transactions).filter_by(hash=tx_hash).one() - tx.consensus_data = {"upgrade_result": "failed", "error": str(e)} - session.commit() + async def _repair_accepted_delivery( + self, transaction: Transaction, transaction_data: dict, session: Session + ) -> Transaction: + """Replay a missed acceptance-phase delivery; returns the reloaded tx.""" + if not _acceptance_dispatch_pending(transaction_data): + return transaction + from backend.consensus.base import ( + accounts_manager_factory, + contract_snapshot_factory, + contract_processor_factory, + node_factory, + transactions_processor_factory, + ) - transactions_processor = TransactionsProcessor(session) - await ConsensusAlgorithm.dispatch_transaction_status_update( - transactions_processor, - tx_hash, - TransactionStatus.CANCELED, - self.msg_handler, - ) - except Exception as update_error: - logger.error( - f"[Worker {self.worker_id}] Failed to update failed upgrade tx status: {update_error}" - ) + logger.info( + f"[Worker {self.worker_id}] Repairing accepted message delivery " + f"for transaction {transaction.hash}" + ) + transactions_processor = transactions_processor_factory(session) + await self.consensus_algorithm.repair_accepted_message_delivery( + transaction, + transactions_processor, + accounts_manager_factory(session), + lambda contract_address: contract_snapshot_factory( + contract_address, session, transaction + ), + contract_processor_factory(session), + node_factory, + ) + latest_transaction = transactions_processor.get_transaction_by_hash( + transaction.hash + ) + if latest_transaction is None: + return transaction + return Transaction.from_dict(latest_transaction) - finally: - # Clear current transaction tracking - self.current_transactions.pop(tx_hash, None) - # Release the transaction - try: - with self.get_session() as release_session: - self.release_transaction(release_session, tx_hash) - except Exception as release_error: - logger.error( - f"[Worker {self.worker_id}] Failed to release upgrade transaction {tx_hash}: {release_error}", - exc_info=True, - ) + async def _finalize_transaction( + self, transaction: Transaction, session: Session + ) -> None: + from backend.consensus.base import ( + accounts_manager_factory, + contract_snapshot_factory, + contract_processor_factory, + node_factory, + transactions_processor_factory, + ) + + await self.consensus_algorithm.process_finalization( + transaction, + transactions_processor_factory(session), + None, # chain_snapshot not needed for finalization + accounts_manager_factory(session), + lambda contract_address: contract_snapshot_factory( + contract_address, session, transaction + ), + contract_processor_factory(session), + node_factory, + ) async def process_finalization(self, finalization_data: dict, session: Session): """ @@ -1965,41 +2353,15 @@ async def process_finalization(self, finalization_data: dict, session: Session): # Convert to Transaction domain object transaction = Transaction.from_dict(finalization_data) - # Check if the appeal window has passed - from backend.consensus.base import ( - contract_snapshot_factory, - contract_processor_factory, - transactions_processor_factory, - accounts_manager_factory, - node_factory, - ) - - transactions_processor = transactions_processor_factory(session) + from backend.consensus.base import transactions_processor_factory - if _acceptance_dispatch_pending(finalization_data): - logger.info( - f"[Worker {self.worker_id}] Repairing accepted message delivery " - f"for transaction {transaction.hash}" - ) - await self.consensus_algorithm.repair_accepted_message_delivery( - transaction, - transactions_processor, - accounts_manager_factory(session), - lambda contract_address: contract_snapshot_factory( - contract_address, session, transaction - ), - contract_processor_factory(session), - node_factory, - ) - latest_transaction = transactions_processor.get_transaction_by_hash( - transaction.hash - ) - if latest_transaction is not None: - transaction = Transaction.from_dict(latest_transaction) + transaction = await self._repair_accepted_delivery( + transaction, finalization_data, session + ) # Check if can finalize (appeal window expired) can_finalize = self.consensus_algorithm.can_finalize_transaction( - transactions_processor, + transactions_processor_factory(session), transaction, 0, # index in queue (not used in our case) [finalization_data], # mock queue with just this transaction @@ -2010,22 +2372,14 @@ async def process_finalization(self, finalization_data: dict, session: Session): f"[Worker {self.worker_id}] Finalizing transaction {transaction.hash}" ) - await self.consensus_algorithm.process_finalization( - transaction, - transactions_processor, - None, # chain_snapshot not needed for finalization - accounts_manager_factory(session), - lambda contract_address: contract_snapshot_factory( - contract_address, session, transaction - ), - contract_processor_factory(session), - node_factory, - ) + await self._finalize_transaction(transaction, session) session.commit() logger.info( f"[Worker {self.worker_id}] Successfully finalized transaction {transaction.hash}" ) + self._no_validators_retries.pop(transaction.hash, None) + self._generic_error_retries.pop(transaction.hash, None) # Send usage metrics to external API (non-blocking) await self.usage_metrics_service.send_finalized_transaction_metrics( @@ -2223,33 +2577,22 @@ def _log_query_result( state["last_log"] = now_monotonic state["polls"] = 0 - def _is_in_backoff(self, transaction_data: dict) -> bool: + def _backoff_window(self, tx_hash: str) -> Optional[tuple[str, float, float]]: """ - Check if a transaction is in backoff due to no validators or generic errors. - - Args: - transaction_data: Transaction data dictionary - - Returns: - True if transaction is in backoff period, False otherwise + Return (kind, elapsed, backoff) if the transaction is inside a retry + backoff window, None otherwise. """ - tx_hash = transaction_data["hash"] + now = time.time() - # Check no-validators backoff retry_info = self._no_validators_retries.get(tx_hash) if retry_info: backoff = self._no_validators_base_backoff * ( 2 ** (retry_info["count"] - 1) ) - time_since_last = time.time() - retry_info["last_attempt"] - if time_since_last < backoff: - logger.debug( - f"[Worker {self.worker_id}] Transaction {tx_hash} in no-validators backoff " - f"({time_since_last:.1f}s < {backoff}s)" - ) - return True + elapsed = now - retry_info["last_attempt"] + if elapsed < backoff: + return "no-validators", elapsed, backoff - # Check generic error backoff generic_info = self._generic_error_retries.get(tx_hash) if generic_info: backoff = self._generic_error_base_backoff * ( @@ -2259,15 +2602,38 @@ def _is_in_backoff(self, transaction_data: dict) -> bool: self.MAX_GENERIC_ERROR_RETRIES - 1, ) ) - time_since_last = time.time() - generic_info["last_attempt"] - if time_since_last < backoff: - logger.debug( - f"[Worker {self.worker_id}] Transaction {tx_hash} in generic-error backoff " - f"({time_since_last:.1f}s < {backoff}s)" - ) - return True + elapsed = now - generic_info["last_attempt"] + if elapsed < backoff: + return "generic-error", elapsed, backoff - return False + return None + + def _backoff_hashes(self) -> list[str]: + """Hashes of transactions currently inside a retry backoff window.""" + candidates = set(self._no_validators_retries) | set(self._generic_error_retries) + return sorted(h for h in candidates if self._backoff_window(h) is not None) + + def _is_in_backoff(self, transaction_data: dict) -> bool: + """ + Check if a transaction is in backoff due to no validators or generic errors. + + Args: + transaction_data: Transaction data dictionary + + Returns: + True if transaction is in backoff period, False otherwise + """ + tx_hash = transaction_data["hash"] + window = self._backoff_window(tx_hash) + if window is None: + return False + + kind, elapsed, backoff = window + logger.debug( + f"[Worker {self.worker_id}] Transaction {tx_hash} in {kind} backoff " + f"({elapsed:.1f}s < {backoff}s)" + ) + return True async def _process_transaction_task(self, transaction_data: dict): """Task wrapper for processing a transaction with its own session.""" @@ -2308,9 +2674,12 @@ async def _try_claim_work(self, session: Session) -> bool: return True self.release_transaction(session, tx_hash) - finalization_data = await self.claim_next_finalization(session) + finalization_data = await self.claim_next_finalization( + session, exclude_hashes=self._backoff_hashes() + ) if finalization_data: tx_hash = finalization_data["hash"] + # Backoff can start after the exclusion list was built. if not self._is_in_backoff(finalization_data): logger.debug( f"[Worker {self.worker_id}] Claimed finalization for transaction {tx_hash}" diff --git a/backend/database_handler/accounts_manager.py b/backend/database_handler/accounts_manager.py index bd02ef034..0b2598384 100644 --- a/backend/database_handler/accounts_manager.py +++ b/backend/database_handler/accounts_manager.py @@ -9,6 +9,7 @@ FEE_ACCOUNTING_KEY, abort_latest_appeal_admission, cancel_fee_accounting, + latest_appeal_admission_unresolved, settle_fee_accounting, ) from backend.consensus.history import ( @@ -326,6 +327,45 @@ def abort_tx_appeal_admission_once( self.session.expire_all() return refund + def abort_pending_appeal_admission( + self, + tx_hash: str, + reason: str, + ) -> int | None: + """Refund an admitted appeal that has no outcome yet; None if none was pending.""" + transaction = ( + self.session.query(Transactions) + .filter_by(hash=tx_hash) + .populate_existing() + .with_for_update() + .one_or_none() + ) + if transaction is None or not isinstance(transaction.data, dict): + return None + has_snapshot = APPEAL_RECOVERY_SNAPSHOT_KEY in transaction.data + if not transaction.appealed and not has_snapshot: + return None + # Resolved bonds keep admissionRollback, so only abort under this guard. + if latest_appeal_admission_unresolved( + transaction.data.get(FEE_ACCOUNTING_KEY), + transaction.consensus_history, + ): + return self.abort_tx_appeal_admission_once(tx_hash, reason) + if not transaction.appealed: + # The appeal already has an outcome; finalization settles it. + return None + # An appeal flag without a revertible paid admission has nothing to refund. + data = dict(transaction.data) + data.pop(APPEAL_RECOVERY_SNAPSHOT_KEY, None) + transaction.data = data + history = dict(transaction.consensus_history or {}) + history.pop(ACTIVE_APPEAL_BASIS_KEY, None) + transaction.consensus_history = history + transaction.appealed = False + transaction.timestamp_appeal = None + self.session.flush() + return 0 + def settle_tx_fee_accounting_once( self, tx_hash: str, diff --git a/backend/database_handler/contract_snapshot.py b/backend/database_handler/contract_snapshot.py index 230323de2..bae49e3ab 100644 --- a/backend/database_handler/contract_snapshot.py +++ b/backend/database_handler/contract_snapshot.py @@ -22,6 +22,8 @@ class ContractSnapshot: contract_address: str balance: int states: Dict[str, Dict[str, str]] + # Transaction-local post-state, separate from the pre-state used for appeals. + finalization_state: Optional[Dict[str, str]] = None # Executor version or `re:` selector this contract is pinned to, forwarded # to the GenVM manager as `reroute_to`. genvm_executor_selector: Optional[str] = None @@ -53,6 +55,7 @@ def to_dict(self): int(b) if (b := getattr(self, "balance", None)) is not None else None ), "genvm_executor_selector": self.genvm_executor_selector, + "finalization_state": self.finalization_state, } @classmethod @@ -61,6 +64,7 @@ def from_dict(cls, input: dict | None) -> Optional["ContractSnapshot"]: instance = cls.__new__(cls) instance.contract_address = input.get("contract_address", None) instance.states = input.get("states", {"accepted": {}, "finalized": {}}) + instance.finalization_state = input.get("finalization_state") raw_balance = input.get("balance") instance.balance = int(raw_balance) if raw_balance is not None else None instance.genvm_executor_selector = input.get( diff --git a/backend/database_handler/finalization_legacy_report.py b/backend/database_handler/finalization_legacy_report.py new file mode 100644 index 000000000..82ba7a5e3 --- /dev/null +++ b/backend/database_handler/finalization_legacy_report.py @@ -0,0 +1,189 @@ +"""Read-only report of ACCEPTED rows that lack a saved finalization state. + +Run before deploying transaction-local finalization to see which legacy rows +will finalize, and from which verified source, and which will stay ACCEPTED. +It also lists decided rows that a pre-change code upgrade overtook; they now +finalize on top of that upgrade's code instead of reverting it. +""" + +from __future__ import annotations + +import argparse +import logging +from collections import Counter +from types import SimpleNamespace +from typing import Callable + +from eth_utils import to_checksum_address +from sqlalchemy import create_engine, text +from sqlalchemy.orm import Session, sessionmaker + +from backend.consensus.base import resolve_finalization_state +from backend.consensus.errors import FinalizationStateError +from backend.database_handler.contract_snapshot import ContractSnapshot +from backend.database_handler.errors import ContractNotFoundError +from backend.database_handler.prune_terminal_snapshots import get_database_url +from backend.database_handler.transactions_processor import TransactionsProcessor + + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger(__name__) + +UNRECOVERABLE = "UNRECOVERABLE" + +_LEGACY_ROWS = text( + """ + SELECT hash, to_address, queue_order, + consensus_data->'leader_receipt'->0->>'contract_state_hash' + AS contract_state_hash + FROM transactions + WHERE status = 'ACCEPTED' + AND consensus_data->'leader_receipt'->0->>'execution_result' = 'SUCCESS' + AND COALESCE( + jsonb_typeof(contract_snapshot->'finalization_state'), 'null' + ) = 'null' + AND COALESCE( + consensus_data->'leader_receipt'->0->'contract_state', 'null'::jsonb + ) IN ('null'::jsonb, '{}'::jsonb) + ORDER BY queue_order ASC + LIMIT :limit + """ +) + +_UPGRADED_AFTER = text( + """ + SELECT EXISTS ( + SELECT 1 FROM transactions upgrade + WHERE upgrade.type = 3 + AND upgrade.status = 'FINALIZED' + AND upgrade.to_address = :to_address + AND upgrade.queue_order > :queue_order + ) + """ +) + +_INTERLEAVED_ROWS = text( + """ + SELECT tx.hash, tx.to_address, tx.status, upgrade.hash AS upgrade_hash + FROM transactions tx + JOIN LATERAL ( + SELECT hash FROM transactions + WHERE type = 3 + AND status = 'FINALIZED' + AND to_address = tx.to_address + AND queue_order > tx.queue_order + AND NOT COALESCE(consensus_data ? 'applied_at', false) + ORDER BY queue_order DESC + LIMIT 1 + ) upgrade ON true + WHERE tx.status IN ( + 'ACCEPTED', 'UNDETERMINED', 'LEADER_TIMEOUT', 'VALIDATORS_TIMEOUT' + ) + ORDER BY tx.queue_order ASC + LIMIT :limit + """ +) + +INTERLEAVED = "interleaved upgrade" + + +def interleaved_rows(session: Session, limit: int) -> list[dict]: + """Decided rows older than a FINALIZED upgrade that did not close them first.""" + return [ + dict(row) + for row in session.execute(_INTERLEAVED_ROWS, {"limit": limit}).mappings() + ] + + +def classify( + row: dict, + transactions_processor, + contract_snapshot_factory: Callable, + upgraded_after: Callable[[dict], bool], +) -> tuple[str, str]: + """Return (source, reason) the finalizer would use for a legacy row.""" + expected_hash = row.get("contract_state_hash") + if not expected_hash: + return UNRECOVERABLE, "no hash" + + context = SimpleNamespace( + transaction=SimpleNamespace( + hash=row["hash"], to_address=row["to_address"], contract_snapshot=None + ), + transactions_processor=transactions_processor, + contract_snapshot_factory=contract_snapshot_factory, + ) + receipt = SimpleNamespace(contract_state=None, contract_state_hash=expected_hash) + try: + _, source = resolve_finalization_state(context, receipt) + except ContractNotFoundError: + return UNRECOVERABLE, "contract not found" + except FinalizationStateError: + if upgraded_after(row): + return UNRECOVERABLE, "upgrade after" + return UNRECOVERABLE, "no matching snapshot" + return source, "" + + +def report(session: Session, limit: int) -> Counter: + session.execute(text("SET TRANSACTION READ ONLY")) + transactions_processor = TransactionsProcessor(session) + + def upgraded_after(row: dict) -> bool: + return bool( + session.execute( + _UPGRADED_AFTER, + {"to_address": row["to_address"], "queue_order": row["queue_order"]}, + ).scalar_one() + ) + + totals: Counter = Counter() + rows = session.execute(_LEGACY_ROWS, {"limit": limit}).mappings().all() + for row in rows: + source, reason = classify( + dict(row), + transactions_processor, + lambda address: ContractSnapshot(to_checksum_address(address), session), + upgraded_after, + ) + totals[source.split(" (")[0]] += 1 + print(f"{row['hash']}\t{row['to_address']}\t{source}\t{reason}".rstrip()) + for row in interleaved_rows(session, limit): + totals[INTERLEAVED] += 1 + print( + f"{row['hash']}\t{row['to_address']}\t" + f"{INTERLEAVED} ({row['upgrade_hash']})\t{row['status']}" + ) + session.rollback() + return totals + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "List ACCEPTED transactions without a saved finalization state and " + "classify how each would finalize, plus decided transactions that an " + "earlier code upgrade overtook. Read-only." + ) + ) + parser.add_argument( + "--limit", + type=int, + default=10000, + help="Maximum number of rows to classify, oldest first.", + ) + return parser + + +def main() -> int: + args = build_parser().parse_args() + engine = create_engine(get_database_url(), pool_pre_ping=True) + SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) + with SessionLocal() as session: + totals = report(session, args.limit) + logger.info("Legacy finalization report: %s", dict(totals)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/database_handler/transactions_processor.py b/backend/database_handler/transactions_processor.py index 92d3d6b52..1f79083a6 100644 --- a/backend/database_handler/transactions_processor.py +++ b/backend/database_handler/transactions_processor.py @@ -1858,6 +1858,8 @@ def admit_transaction_appeal( raise ValueError("CanNotAppeal") if acceptance_dispatch_pending((row["data"] or {}).get(FEE_ACCOUNTING_KEY)): raise ValueError("CanNotAppeal") + if self.upgrade_closes_appeals(transaction_hash): + raise ValueError("CanNotAppeal") # The RPC may have waited for this row lock. Consensus evaluates the # deadline at transaction execution, not when the caller began # submitting it, so re-sample after lock acquisition and reject an @@ -2107,7 +2109,8 @@ def get_transactions_for_block( return block_details - def get_newer_transactions(self, transaction_hash: str): + def get_rewindable_newer_transactions(self, transaction_hash: str): + """Newer same-contract rows a rewind may reset; terminal rows and upgrades never replay.""" transaction = ( self.session.query(Transactions).filter_by(hash=transaction_hash).one() ) @@ -2117,6 +2120,13 @@ def get_newer_transactions(self, transaction_hash: str): .filter( _transaction_order_tuple() > _transaction_order_value(transaction), Transactions.to_address == transaction.to_address, + Transactions.status.notin_( + [TransactionStatus.FINALIZED, TransactionStatus.CANCELED] + ), + or_( + Transactions.type.is_(None), + Transactions.type != TransactionType.UPGRADE_CONTRACT.value, + ), ) .order_by(Transactions.queue_order) .all() @@ -2308,6 +2318,104 @@ def set_transaction_contract_snapshot( ) self.session.commit() + def get_transaction_finalization_state(self, transaction_hash: str) -> dict | None: + """Read only the accepted result, without transferring rollback state.""" + return self.session.execute( + text( + "SELECT contract_snapshot->'finalization_state' FROM transactions " + "WHERE hash = :hash" + ), + {"hash": transaction_hash}, + ).scalar_one_or_none() + + def get_successor_snapshot_hashes( + self, transaction_hash: str, limit: int + ) -> list[str]: + """Oldest same-contract successors that saved a pre-state snapshot.""" + rows = self.session.execute( + text( + """ + SELECT successor.hash + FROM transactions tx + JOIN transactions successor + ON successor.to_address = tx.to_address + AND successor.queue_order > tx.queue_order + WHERE tx.hash = :hash + AND successor.contract_snapshot IS NOT NULL + ORDER BY successor.queue_order ASC + LIMIT :limit + """ + ), + {"hash": transaction_hash, "limit": limit}, + ).scalars() + return list(rows) + + def get_latest_finalized_upgrade_after(self, transaction_hash: str) -> dict | None: + """The newest FINALIZED code upgrade queued after this transaction.""" + row = ( + self.session.execute( + text( + """ + SELECT upgrade.hash, + upgrade.data->>'new_code' AS new_code, + contract.genvm_executor_selector + FROM transactions tx + JOIN transactions upgrade + ON upgrade.to_address = tx.to_address + AND upgrade.queue_order > tx.queue_order + LEFT JOIN current_state contract ON contract.id = tx.to_address + WHERE tx.hash = :hash + AND upgrade.type = :upgrade_type + AND upgrade.status = 'FINALIZED' + ORDER BY upgrade.queue_order DESC + LIMIT 1 + """ + ), + { + "hash": transaction_hash, + "upgrade_type": TransactionType.UPGRADE_CONTRACT.value, + }, + ) + .mappings() + .first() + ) + return dict(row) if row is not None else None + + def upgrade_closes_appeals(self, transaction_hash: str) -> bool: + """Whether a queued or applied code upgrade closed this decision's lifecycle.""" + return bool( + self.session.execute( + text( + """ + SELECT EXISTS ( + SELECT 1 + FROM transactions tx + JOIN transactions upgrade + ON upgrade.to_address = tx.to_address + AND upgrade.queue_order > tx.queue_order + WHERE tx.hash = :hash + AND upgrade.type = :upgrade_type + AND upgrade.status IN ('PENDING', 'FINALIZED') + ) + """ + ), + { + "hash": transaction_hash, + "upgrade_type": TransactionType.UPGRADE_CONTRACT.value, + }, + ).scalar_one() + ) + + def get_transaction_pre_state(self, transaction_hash: str) -> dict | None: + """Read only the saved pre-execution accepted state.""" + return self.session.execute( + text( + "SELECT contract_snapshot->'states'->'accepted' FROM transactions " + "WHERE hash = :hash" + ), + {"hash": transaction_hash}, + ).scalar_one_or_none() + def transactions_in_process_by_contract(self) -> list[dict]: transactions = ( self.session.query(Transactions) diff --git a/backend/node/types.py b/backend/node/types.py index 93936096a..b3b217fa0 100644 --- a/backend/node/types.py +++ b/backend/node/types.py @@ -308,7 +308,8 @@ def to_dict(self, strip_contract_state: bool = False): Args: strip_contract_state: If True, replaces contract_state with empty dict to save storage. - Contract state is always available from CurrentState table. + Acceptance saves the transaction's post-state in + its contract snapshot for later finalization. """ result = base64.b64encode(self.result).decode("ascii") calldata = str(base64.b64encode(self.calldata), encoding="ascii") diff --git a/backend/protocol_rpc/endpoints.py b/backend/protocol_rpc/endpoints.py index 9977c4768..ca4f6571a 100644 --- a/backend/protocol_rpc/endpoints.py +++ b/backend/protocol_rpc/endpoints.py @@ -988,7 +988,8 @@ def admin_upgrade_contract_code( ) -> dict: """ Queue a contract code upgrade. Returns immediately with tx hash. - Upgrade executes when worker processes it (after any pending txs for this contract). + The worker runs it ahead of this contract's queued txs once in-flight work ends: + earlier decided txs are finalized first, never-started txs run on the new code. Access control: - Local (no env vars): open access @@ -2128,6 +2129,10 @@ def _appeal_admission_context( or has_terminal_validator_appeal(transaction.get("consensus_history")) ): raise InvalidTransactionError("CanNotAppeal") + if session is not None and TransactionsProcessor(session).upgrade_closes_appeals( + str(transaction.get("hash")) + ): + raise InvalidTransactionError("CanNotAppeal") deadline = _transaction_appeal_deadline(transaction) if deadline is not None and time.time() >= deadline: diff --git a/backend/protocol_rpc/fees.py b/backend/protocol_rpc/fees.py index d148db2d1..2b3518b53 100644 --- a/backend/protocol_rpc/fees.py +++ b/backend/protocol_rpc/fees.py @@ -2737,6 +2737,26 @@ def abort_latest_appeal_admission( return updated, recipient, refund +def latest_appeal_admission_unresolved( + accounting: dict[str, Any] | None, + consensus_history: dict[str, Any] | None, +) -> bool: + """Whether the latest admitted appeal bond is still revertible and has no outcome.""" + + bonds = (accounting or {}).get("appeal_bonds") + if not isinstance(bonds, list) or not bonds: + return False + bond = bonds[-1] + if not isinstance(bond, dict) or not isinstance( + bond.get("admissionRollback"), dict + ): + return False + _, outcome = _appeal_outcome_after_round( + _round_outcomes(consensus_history), int(bond.get("round", 0) or 0) + ) + return outcome is None + + def calculate_appeal_charge( fees_distribution: dict[str, Any], *, diff --git a/tests/db-sqlalchemy/accounts_manager_test.py b/tests/db-sqlalchemy/accounts_manager_test.py index 378138505..b12820967 100644 --- a/tests/db-sqlalchemy/accounts_manager_test.py +++ b/tests/db-sqlalchemy/accounts_manager_test.py @@ -291,6 +291,66 @@ def test_abort_tx_appeal_admission_refunds_charge_and_restores_decision( assert restored["fees_distribution"] == accounting["fees_distribution"] +@pytest.mark.parametrize("appealed, expected", [(False, None), (True, 0)]) +def test_abort_pending_appeal_admission_leaves_resolved_bonds_alone( + accounts_manager: AccountsManager, + session: Session, + appealed: bool, + expected: int | None, +): + sender = "0x9F0e84243496AcFB3Cd99D02eA59673c05901501" + appealer = "0x1111111111111111111111111111111111111111" + tx_hash = "0x" + "ae" * 32 + fees = _fees_distribution(appeals=0, rotations=[0]) + accounting = create_fee_accounting( + fees_distribution=fees, + num_of_validators=5, + submitted_value=required_fee_deposit(fees, 5), + user_value=0, + sender=sender, + ) + charge = calculate_appeal_charge( + accounting["fees_distribution"], current_round=0, status="ACCEPTED" + ) + recorded = record_appeal_bond( + accounting, + amount=charge["bond"] + charge["funding"], + appealer=appealer, + current_round=0, + status="ACCEPTED", + ) + _insert_fee_accounted_transaction( + session, + sender=sender, + accounting=recorded, + tx_hash=tx_hash, + consensus_history={ + "consensus_results": [ + {"consensus_round": "Accepted"}, + {"consensus_round": "Validator Appeal Failed"}, + ] + }, + ) + tx_model = session.query(Transactions).filter_by(hash=tx_hash).one() + tx_model.appealed = appealed + tx_model.data = { + **tx_model.data, + APPEAL_RECOVERY_SNAPSHOT_KEY: {"status": "ACCEPTED"}, + } + session.commit() + + assert ( + accounts_manager.abort_pending_appeal_admission(tx_hash, "upgrade") == expected + ) + session.commit() + session.expire_all() + + assert accounts_manager.get_account_balance(appealer) == 0 + tx = session.query(Transactions).filter_by(hash=tx_hash).one() + assert tx.appealed is False + assert tx.data[FEE_ACCOUNTING_KEY]["appeal_bonds"] == recorded["appeal_bonds"] + + def test_abort_zero_value_appeal_admission_still_persists_rollback( accounts_manager: AccountsManager, session: Session, diff --git a/tests/db-sqlalchemy/test_finalization_ordering.py b/tests/db-sqlalchemy/test_finalization_ordering.py index 103eda05d..c2aa56505 100644 --- a/tests/db-sqlalchemy/test_finalization_ordering.py +++ b/tests/db-sqlalchemy/test_finalization_ordering.py @@ -375,3 +375,87 @@ async def test_ordering_is_per_contract(worker: ConsensusWorker, session: Sessio "Ordering must be per-contract; other contracts' state shouldn't matter. " f"Got: {result}" ) + + +def _insert_ready(session: Session, tx_hash: str, nonce: int, to_address: str): + _insert_tx( + session, + tx_hash=tx_hash, + status="ACCEPTED", + nonce=nonce, + to_address=to_address, + timestamp_awaiting_finalization=int(time.time()) - 600, + ) + + +@pytest.mark.asyncio +async def test_backed_off_head_does_not_starve_other_contracts( + worker: ConsensusWorker, session: Session +): + stuck = "0x" + "80" * 32 + other = "0x" + "81" * 32 + _insert_ready(session, stuck, 0, CONTRACT) + _insert_ready(session, other, 0, OTHER_CONTRACT) + + with worker.get_session() as s: + result = await worker.claim_next_finalization(s, exclude_hashes=[stuck]) + + assert result is not None and result["hash"] == other + + +@pytest.mark.asyncio +async def test_backed_off_head_still_blocks_younger_same_contract_tx( + worker: ConsensusWorker, session: Session +): + stuck = "0x" + "82" * 32 + _insert_ready(session, stuck, 0, CONTRACT) + _insert_ready(session, "0x" + "83" * 32, 1, CONTRACT) + + with worker.get_session() as s: + result = await worker.claim_next_finalization(s, exclude_hashes=[stuck]) + + assert result is None + + +@pytest.mark.asyncio +async def test_empty_exclusion_list_claims_normally( + worker: ConsensusWorker, session: Session +): + head = "0x" + "84" * 32 + _insert_ready(session, head, 0, CONTRACT) + + with worker.get_session() as s: + result = await worker.claim_next_finalization(s, exclude_hashes=[]) + + assert result is not None and result["hash"] == head + + +@pytest.mark.asyncio +async def test_head_is_claimed_once_backoff_exclusion_is_lifted( + worker: ConsensusWorker, session: Session +): + stuck = "0x" + "85" * 32 + _insert_ready(session, stuck, 0, CONTRACT) + _insert_ready(session, "0x" + "86" * 32, 1, CONTRACT) + + with worker.get_session() as s: + assert await worker.claim_next_finalization(s, exclude_hashes=[stuck]) is None + with worker.get_session() as s: + result = await worker.claim_next_finalization(s) + + assert result is not None and result["hash"] == stuck + + +@pytest.mark.asyncio +async def test_terminal_backed_off_tx_does_not_block_its_successor( + worker: ConsensusWorker, session: Session +): + done = "0x" + "87" * 32 + successor = "0x" + "88" * 32 + _insert_tx(session, tx_hash=done, status="FINALIZED", nonce=0) + _insert_ready(session, successor, 1, CONTRACT) + + with worker.get_session() as s: + result = await worker.claim_next_finalization(s, exclude_hashes=[done]) + + assert result is not None and result["hash"] == successor diff --git a/tests/db-sqlalchemy/test_finalization_starvation.py b/tests/db-sqlalchemy/test_finalization_starvation.py index 1b147cbb6..0c14b9912 100644 --- a/tests/db-sqlalchemy/test_finalization_starvation.py +++ b/tests/db-sqlalchemy/test_finalization_starvation.py @@ -757,3 +757,45 @@ async def test_safety_net_warns_on_long_stuck_finalization( tx_hash in record.getMessage() and "starvation" in record.getMessage().lower() for record in caplog.records ), f"Expected starvation warning for {tx_hash}; got {[r.getMessage() for r in caplog.records]}" + + +@pytest.mark.asyncio +async def test_upgrade_is_claimed_ahead_of_older_pending_and_open_window( + engine: Engine, worker: ConsensusWorker, session: Session +): + """The upgrade keeps type 3 priority; it finalizes decided rows itself.""" + _setup_contract(engine) + upgrade_hash = "0x" + "c3" * 32 + _insert_tx( + session, + tx_hash="0x" + "c1" * 32, + status="ACCEPTED", + nonce=0, + timestamp_awaiting_finalization=int(time.time()) - 1, + consensus_data={"leader_receipt": [{}], "votes": {}, "validators": []}, + created_at_offset_seconds=-30, + ) + _insert_tx( + session, + tx_hash="0x" + "c2" * 32, + status="PENDING", + nonce=1, + created_at_offset_seconds=-20, + ) + _insert_tx( + session, + tx_hash=upgrade_hash, + status="PENDING", + nonce=2, + data={"new_code": "class C: pass"}, + ) + session.execute( + text("UPDATE transactions SET type = 3 WHERE hash = :hash"), + {"hash": upgrade_hash}, + ) + session.commit() + + claimed = await worker.claim_next_transaction(session) + + assert claimed is not None + assert claimed["hash"] == upgrade_hash diff --git a/tests/db-sqlalchemy/test_transaction_finalized_state.py b/tests/db-sqlalchemy/test_transaction_finalized_state.py new file mode 100644 index 000000000..d5d8a356b --- /dev/null +++ b/tests/db-sqlalchemy/test_transaction_finalized_state.py @@ -0,0 +1,903 @@ +"""Persisted acceptance results, not the moving contract head, become final.""" + +import base64 +import json +import time +from contextlib import contextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from sqlalchemy import text +from sqlalchemy.orm import sessionmaker + +from backend.consensus.base import ( + AcceptedState, + TransactionContext, + contract_snapshot_factory, + node_factory, + upgrade_code_slot, +) +import backend.consensus.base as consensus_base +from backend.consensus.history import APPEAL_RECOVERY_SNAPSHOT_KEY +from backend.consensus.types import ConsensusRound +from backend.consensus.worker import ConsensusWorker +from backend.database_handler.accounts_manager import AccountsManager +from backend.database_handler.contract_processor import ContractProcessor +from backend.database_handler.contract_snapshot import ContractSnapshot +from backend.database_handler.models import CurrentState, Transactions +from backend.database_handler.transactions_processor import TransactionsProcessor +from backend.database_handler.types import ConsensusData +from backend.domain.types import Transaction, TransactionType +from backend.node.base import _SnapshotView +from backend.node.genvm import get_code_slot +from backend.node.types import Address, ExecutionMode, ExecutionResultStatus, Receipt +from backend.protocol_rpc.endpoints import admin_upgrade_contract_code +from backend.protocol_rpc.fees import ( + FEE_ACCOUNTING_KEY, + calculate_appeal_charge, + create_fee_accounting, + record_appeal_bond, + required_fee_deposit, +) + + +SENDER = "0x" + "11" * 20 +CONTRACT = "0x" + "22" * 20 +SLOT = b"\x01" * 32 +KEY = base64.b64encode(SLOT).decode() + + +def state(value): + return {KEY: base64.b64encode(bytes([value])).decode()} + + +@pytest.fixture +def lifecycle(engine, monkeypatch): + monkeypatch.setenv("VITE_FINALITY_WINDOW", "3600") + monkeypatch.setenv("VITE_FINALITY_WINDOW_APPEAL_FAILED_REDUCTION", "0.2") + sessions = sessionmaker(bind=engine, expire_on_commit=False) + + @contextmanager + def get_session(): + with sessions() as session: + yield session + + rollup = Mock() + rollup.emit_transaction_event.return_value = {"txIds": [], "messages": []} + messages = Mock() + messages.send_message_async = AsyncMock() + worker = ConsensusWorker(get_session, messages, rollup, None, None) + worker.usage_metrics_service = SimpleNamespace( + send_finalized_transaction_metrics=AsyncMock() + ) + with sessions() as session: + accounts = AccountsManager(session) + accounts.create_new_account_with_address(SENDER) + accounts.create_new_account_with_address(CONTRACT) + session.query(CurrentState).filter_by(id=CONTRACT).one().data = { + "state": {"accepted": state(0), "finalized": state(0)} + } + session.commit() + return SimpleNamespace(sessions=sessions, worker=worker, rollup=rollup) + + +def context_for(lifecycle, session, tx): + return TransactionContext( + transaction=tx, + transactions_processor=TransactionsProcessor(session), + chain_snapshot=None, + accounts_manager=AccountsManager(session), + contract_snapshot_factory=lambda addr: contract_snapshot_factory( + addr, session, tx + ), + contract_processor=ContractProcessor(session), + node_factory=node_factory, + msg_handler=lifecycle.worker.msg_handler, + consensus_service=lifecycle.rollup, + validators_snapshot=None, + genvm_manager=None, + ) + + +async def accept( + lifecycle, + nonce, + post_state, + *, + execution_result=ExecutionResultStatus.SUCCESS, + recompute=False, + tx_type=TransactionType.RUN_CONTRACT, +): + tx_hash = "0x" + f"{nonce + 1:064x}" + with lifecycle.sessions() as session: + tp = TransactionsProcessor(session) + if not recompute: + tp.insert_transaction( + from_address=SENDER, + to_address=CONTRACT, + data=( + { + "contract_address": CONTRACT, + "contract_code": base64.b64encode( + b"class Contract: pass" + ).decode(), + "calldata": "Bg==", + } + if tx_type == TransactionType.DEPLOY_CONTRACT + else {"calldata": "Bg=="} + ), + value=0, + type=tx_type.value, + nonce=nonce, + leader_only=False, + config_rotation_rounds=3, + transaction_hash=tx_hash, + ) + session.commit() + claimed = await lifecycle.worker.claim_next_transaction(session) + assert claimed is not None and claimed["hash"] == tx_hash + tx = Transaction.from_dict(tp.get_transaction_by_hash(tx_hash)) + context = context_for(lifecycle, session, tx) + # Only VM execution is replaced: scheduling, acceptance, persistence, + # appeal timing, finalization and storage reads use the real code. + receipt = Receipt( + result=b"\x00\x00", + calldata=b"\x06", + gas_used=0, + mode=ExecutionMode.LEADER, + contract_state=post_state, + node_config={"address": SENDER, "stake": 100}, + execution_result=execution_result, + pending_transactions=[], + genvm_result={"stdout": "", "stderr": ""}, + ) + context.consensus_data = ConsensusData( + votes={}, leader_receipt=[receipt], validators=[] + ) + await AcceptedState().handle(context) + session.commit() + lifecycle.worker.release_transaction(session, tx_hash) + assert tp.get_transaction_by_hash(tx_hash)["status"] == "ACCEPTED" + stored = session.execute( + text("SELECT consensus_data FROM transactions WHERE hash=:hash"), + {"hash": tx_hash}, + ).scalar_one() + assert stored["leader_receipt"][0]["contract_state"] == {} + return tx_hash + + +async def finalize(lifecycle, tx_hash, *, expected_status="FINALIZED"): + with lifecycle.sessions() as session: + tp = TransactionsProcessor(session) + row = tp.get_transaction_by_hash(tx_hash) + history = row["consensus_history"] + history["latestDecision"]["appealDeadline"] = int(time.time()) - 1 + session.execute( + text( + "UPDATE transactions SET timestamp_awaiting_finalization=:ts, " + "consensus_history=CAST(:history AS jsonb) WHERE hash=:hash" + ), + { + "ts": int(time.time()) - 3601, + "history": json.dumps(history), + "hash": tx_hash, + }, + ) + session.commit() + claimed = await lifecycle.worker.claim_next_finalization(session) + assert claimed is not None and claimed["hash"] == tx_hash + await lifecycle.worker.process_finalization(claimed, session) + session.expire_all() + assert tp.get_transaction_by_hash(tx_hash)["status"] == expected_status + + +@pytest.mark.asyncio +@pytest.mark.parametrize("overlap", [False, True]) +async def test_finalizes_only_its_own_accepted_result(lifecycle, overlap): + first = await accept(lifecycle, 0, state(1)) + second = await accept(lifecycle, 1, state(2)) if overlap else None + with lifecycle.sessions() as session: + assert await lifecycle.worker.claim_next_finalization(session) is None + await finalize(lifecycle, first) + with lifecycle.sessions() as session: + snapshot = ContractSnapshot(CONTRACT, session) + view = _SnapshotView( + snapshot, lambda addr: ContractSnapshot(addr, session), True, "finalized" + ) + assert view.storage_read(Address(CONTRACT), SLOT, 0, 1) == b"\x01" + assert snapshot.states["accepted"] == state(2 if overlap else 1) + if second: + tp = TransactionsProcessor(session) + row = tp.get_transaction_by_hash(second) + assert row["status"] == "ACCEPTED" + assert not lifecycle.worker.consensus_algorithm.can_finalize_transaction( + tp, Transaction.from_dict(row), 0, [row] + ) + if second: + await finalize(lifecycle, second) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states["finalized"] == state(2) + + +@pytest.mark.asyncio +async def test_deployment_finalizes_constructor_state_not_later_write(lifecycle): + with lifecycle.sessions() as session: + session.query(CurrentState).filter_by(id=CONTRACT).one().data = {} + session.commit() + code_slot = base64.b64encode(get_code_slot()).decode() + deployed = {**state(1), code_slot: base64.b64encode(b"contract code").decode()} + updated = {**deployed, **state(2)} + deployment = await accept( + lifecycle, 0, deployed, tx_type=TransactionType.DEPLOY_CONTRACT + ) + await accept(lifecycle, 1, updated) + await finalize(lifecycle, deployment) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states == { + "accepted": updated, + "finalized": deployed, + } + + +@pytest.mark.asyncio +async def test_failed_appeal_preserves_original_post_state(lifecycle): + first = await accept(lifecycle, 0, state(1)) + await accept(lifecycle, 1, state(2)) + with lifecycle.sessions() as session: + tp = TransactionsProcessor(session) + tp.set_transaction_appeal(first, True) + tx = Transaction.from_dict(tp.get_transaction_by_hash(first)) + assert not tx.consensus_data.leader_receipt[0].contract_state + context = context_for(lifecycle, session, tx) + context.consensus_data = tx.consensus_data + await AcceptedState().handle(context) + row = tp.get_transaction_by_hash(first) + assert row["appeal_failed"] == 1 + assert row["contract_snapshot"]["states"]["accepted"] == state(0) + assert row["contract_snapshot"]["finalization_state"] == state(1) + assert ContractSnapshot(CONTRACT, session).states["accepted"] == state(2) + await finalize(lifecycle, first) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states["finalized"] == state(1) + + +@pytest.mark.asyncio +async def test_successful_appeal_discards_post_states_before_reexecution( + lifecycle, monkeypatch +): + first = await accept(lifecycle, 0, state(1)) + second = await accept(lifecycle, 1, state(2)) + # Supply the jury verdict, but run real appeal restoration and descendant reset. + monkeypatch.setattr( + consensus_base.ConsensusAlgorithm, + "get_extra_validators", + staticmethod(lambda *args, **kwargs: (None, [{"address": SENDER}])), + ) + monkeypatch.setattr( + consensus_base, + "CommittingState", + lambda: SimpleNamespace( + handle=AsyncMock(return_value=ConsensusRound.VALIDATOR_APPEAL_SUCCESSFUL) + ), + ) + with lifecycle.sessions() as session: + tp = TransactionsProcessor(session) + tp.set_transaction_appeal(first, True) + tx = Transaction.from_dict(tp.get_transaction_by_hash(first)) + context = context_for(lifecycle, session, tx) + await lifecycle.worker.consensus_algorithm.process_validator_appeal( + transaction=tx, + transactions_processor=tp, + chain_snapshot=None, + accounts_manager=context.accounts_manager, + contract_snapshot_factory=context.contract_snapshot_factory, + contract_processor=context.contract_processor, + node_factory=node_factory, + validators_snapshot=SimpleNamespace(nodes=[]), + ) + for tx_hash in (first, second): + row = tp.get_transaction_by_hash(tx_hash) + assert row["status"] == "PENDING" + assert row["contract_snapshot"] is None + assert ContractSnapshot(CONTRACT, session).states == { + "accepted": state(0), + "finalized": state(0), + } + await accept(lifecycle, 0, state(3), recompute=True) + await accept(lifecycle, 1, state(4), recompute=True) + await finalize(lifecycle, first) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states == { + "accepted": state(4), + "finalized": state(3), + } + await finalize(lifecycle, second) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states["finalized"] == state(4) + + +@pytest.mark.asyncio +async def test_empty_post_state_is_distinct_from_missing_state(lifecycle): + tx_hash = await accept(lifecycle, 0, {}) + await finalize(lifecycle, tx_hash) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states["finalized"] == {} + + +@pytest.mark.asyncio +async def test_accepted_execution_error_does_not_promote_contract_state(lifecycle): + tx_hash = await accept( + lifecycle, 0, state(9), execution_result=ExecutionResultStatus.ERROR + ) + await finalize(lifecycle, tx_hash) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states == { + "accepted": state(0), + "finalized": state(0), + } + + +def strip_finalization_state(session, tx_hash): + session.execute( + text( + "UPDATE transactions SET contract_snapshot = contract_snapshot - 'finalization_state' WHERE hash=:hash" + ), + {"hash": tx_hash}, + ) + session.commit() + + +@pytest.mark.asyncio +async def test_legacy_row_without_successor_uses_matching_head(lifecycle): + first = await accept(lifecycle, 0, state(1)) + with lifecycle.sessions() as session: + strip_finalization_state(session, first) + await finalize(lifecycle, first) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states["finalized"] == state(1) + + +@pytest.mark.asyncio +async def test_legacy_row_recovers_post_state_from_successor_pre_state(lifecycle): + first = await accept(lifecycle, 0, state(1)) + second = await accept(lifecycle, 1, state(2)) + with lifecycle.sessions() as session: + strip_finalization_state(session, first) + await finalize(lifecycle, first) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states == { + "accepted": state(2), + "finalized": state(1), + } + await finalize(lifecycle, second) + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states["finalized"] == state(2) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("successor_snapshot", ["tampered", "missing"]) +async def test_legacy_row_without_verifiable_state_stays_accepted( + lifecycle, successor_snapshot +): + first = await accept(lifecycle, 0, state(1)) + second = await accept(lifecycle, 1, state(2)) + with lifecycle.sessions() as session: + strip_finalization_state(session, first) + session.execute( + text( + "UPDATE transactions SET contract_snapshot = " + + ( + "jsonb_set(contract_snapshot, '{states,accepted}', " + "CAST(:tampered AS jsonb))" + if successor_snapshot == "tampered" + else "NULL" + ) + + " WHERE hash=:hash" + ), + {"hash": second, "tampered": json.dumps(state(7))}, + ) + session.commit() + lifecycle.rollup.reset_mock() + # Exhaust the generic retry budget: an unverifiable agreed decision + # must neither be finalized nor canceled to let younger state escape. + for _ in range(lifecycle.worker.MAX_GENERIC_ERROR_RETRIES + 1): + await finalize(lifecycle, first, expected_status="ACCEPTED") + lifecycle.rollup.emit_transaction_event.assert_not_called() + with lifecycle.sessions() as session: + assert ContractSnapshot(CONTRACT, session).states["finalized"] == state(0) + + +@pytest.mark.asyncio +async def test_legacy_report_classifies_without_writing(lifecycle, capsys): + from backend.database_handler.finalization_legacy_report import report + + first = await accept(lifecycle, 0, state(1)) + second = await accept(lifecycle, 1, state(2)) + with lifecycle.sessions() as session: + strip_finalization_state(session, first) + capsys.readouterr() + with lifecycle.sessions() as session: + totals = report(session, limit=10) + lines = capsys.readouterr().out.strip().splitlines() + assert lines == [f"{first}\t{CONTRACT}\tsuccessor 1 ({second})"] + assert dict(totals) == {"successor 1": 1} + with lifecycle.sessions() as session: + tp = TransactionsProcessor(session) + assert tp.get_transaction_by_hash(first)["status"] == "ACCEPTED" + assert ContractSnapshot(CONTRACT, session).states["finalized"] == state(0) + + +UPGRADE_CODE = "class Contract: pass # v2" +APPEALER = "0x" + "33" * 20 +FEES = { + "leaderTimeunitsAllocation": 100, + "validatorTimeunitsAllocation": 200, + "appealRounds": 0, + "executionBudgetPerRound": 0, + "executionConsumed": 0, + "totalMessageFees": 0, + "rotations": [0], + "maxPriceGenPerTimeUnit": 0, + "storageFeeMaxGasPrice": 0, + "receiptFeeMaxGasPrice": 0, +} + + +def code(new_code=UPGRADE_CODE): + return dict([upgrade_code_slot(new_code, None)]) + + +CODE_KEY = next(iter(code())) + + +@pytest.fixture +def upgrades(lifecycle, monkeypatch): + monkeypatch.setenv("UPGRADE_DEFER_SECONDS", "0") + monkeypatch.delenv("VITE_IS_HOSTED", raising=False) + monkeypatch.delenv("ADMIN_API_KEY", raising=False) + return lifecycle + + +def queue_upgrade(lifecycle, *, legacy_finalized=False): + with lifecycle.sessions() as session: + upgrade_hash = admin_upgrade_contract_code(session, CONTRACT, UPGRADE_CODE)[ + "transaction_hash" + ] + if legacy_finalized: + # What the pre-change worker left: code applied, no predecessor close. + session.query(Transactions).filter_by(hash=upgrade_hash).one().status = ( + "FINALIZED" + ) + session.query(Transactions).filter_by( + hash=upgrade_hash + ).one().consensus_data = {"upgrade_result": "success"} + contract = session.query(CurrentState).filter_by(id=CONTRACT).one() + contract.data = { + "state": { + name: {**tree, **code()} + for name, tree in contract.data["state"].items() + } + } + session.commit() + return upgrade_hash + + +async def run_upgrade(lifecycle, upgrade_hash): + with lifecycle.sessions() as session: + claimed = await lifecycle.worker.claim_next_transaction(session) + assert claimed is not None and claimed["hash"] == upgrade_hash + await lifecycle.worker.process_transaction(claimed, session) + return row_of(lifecycle, upgrade_hash) + + +def row_of(lifecycle, tx_hash): + with lifecycle.sessions() as session: + return TransactionsProcessor(session).get_transaction_by_hash(tx_hash) + + +def contract_states(lifecycle): + with lifecycle.sessions() as session: + return ContractSnapshot(CONTRACT, session).states + + +def balance(lifecycle, address): + with lifecycle.sessions() as session: + return AccountsManager(session).get_account_balance(address) + + +def add_fee_accounting(lifecycle, tx_hash, *, appeal=False): + """Give a decided row fee custody, optionally with an admitted paid appeal.""" + accounting = create_fee_accounting( + fees_distribution=dict(FEES), + num_of_validators=5, + submitted_value=required_fee_deposit(FEES, 5), + user_value=0, + sender=SENDER, + ) + charge = 0 + if appeal: + quote = calculate_appeal_charge( + accounting["fees_distribution"], current_round=0, status="ACCEPTED" + ) + charge = quote["bond"] + quote["funding"] + accounting = record_appeal_bond( + accounting, + amount=charge, + appealer=APPEALER, + current_round=0, + status="ACCEPTED", + ) + with lifecycle.sessions() as session: + row = session.query(Transactions).filter_by(hash=tx_hash).one() + row.num_of_initial_validators = 5 + data = {**(row.data or {}), FEE_ACCOUNTING_KEY: accounting} + if appeal: + row.appealed = True + row.timestamp_appeal = int(time.time()) + data[APPEAL_RECOVERY_SNAPSHOT_KEY] = {"status": "ACCEPTED"} + row.data = data + session.commit() + return charge + + +def insert_pending(lifecycle, nonce): + tx_hash = "0x" + f"{nonce + 1:064x}" + with lifecycle.sessions() as session: + TransactionsProcessor(session).insert_transaction( + from_address=SENDER, + to_address=CONTRACT, + data={"calldata": "Bg=="}, + value=0, + type=TransactionType.RUN_CONTRACT.value, + nonce=nonce, + leader_only=False, + config_rotation_rounds=3, + transaction_hash=tx_hash, + ) + session.commit() + return tx_hash + + +async def succeed_validator_appeal(lifecycle, monkeypatch, tx_hash): + monkeypatch.setattr( + consensus_base.ConsensusAlgorithm, + "get_extra_validators", + staticmethod(lambda *args, **kwargs: (None, [{"address": SENDER}])), + ) + monkeypatch.setattr( + consensus_base, + "CommittingState", + lambda: SimpleNamespace( + handle=AsyncMock(return_value=ConsensusRound.VALIDATOR_APPEAL_SUCCESSFUL) + ), + ) + with lifecycle.sessions() as session: + tp = TransactionsProcessor(session) + tp.set_transaction_appeal(tx_hash, True) + tx = Transaction.from_dict(tp.get_transaction_by_hash(tx_hash)) + context = context_for(lifecycle, session, tx) + await lifecycle.worker.consensus_algorithm.process_validator_appeal( + transaction=tx, + transactions_processor=tp, + chain_snapshot=None, + accounts_manager=context.accounts_manager, + contract_snapshot_factory=context.contract_snapshot_factory, + contract_processor=context.contract_processor, + node_factory=node_factory, + validators_snapshot=SimpleNamespace(nodes=[]), + ) + + +@pytest.mark.asyncio +async def test_upgrade_finalizes_decided_predecessor_before_applying_code(upgrades): + first = await accept(upgrades, 0, state(1)) + add_fee_accounting(upgrades, first) + upgrade_hash = queue_upgrade(upgrades) + + upgrade = await run_upgrade(upgrades, upgrade_hash) + + assert upgrade["status"] == "FINALIZED" + assert upgrade["consensus_data"]["upgrade_result"] == "success" + assert upgrade["consensus_data"]["closed_transactions"] == [first] + assert upgrade["consensus_data"]["refunded_appeals"] == [] + assert "applied_at" in upgrade["consensus_data"] + finalized = row_of(upgrades, first) + assert finalized["status"] == "FINALIZED" + assert finalized["data"][FEE_ACCOUNTING_KEY]["status"] == "settled" + assert contract_states(upgrades) == { + "accepted": {**state(1), **code()}, + "finalized": {**state(1), **code()}, + } + + +@pytest.mark.asyncio +async def test_never_started_transaction_stays_queued_and_runs_on_new_code( + upgrades, +): + first = await accept(upgrades, 0, state(1)) + pending = insert_pending(upgrades, 1) + upgrade_hash = queue_upgrade(upgrades) + + assert (await run_upgrade(upgrades, upgrade_hash))["status"] == "FINALIZED" + assert row_of(upgrades, first)["status"] == "FINALIZED" + assert row_of(upgrades, pending)["status"] == "PENDING" + + await accept(upgrades, 1, {**state(2), **code()}, recompute=True) + pre_state = row_of(upgrades, pending)["contract_snapshot"]["states"]["accepted"] + assert pre_state == {**state(1), **code()} + + +@pytest.mark.asyncio +async def test_upgrade_refunds_pending_paid_appeal_before_finalizing(upgrades): + first = await accept(upgrades, 0, state(1)) + charge = add_fee_accounting(upgrades, first, appeal=True) + upgrade_hash = queue_upgrade(upgrades) + + upgrade = await run_upgrade(upgrades, upgrade_hash) + + assert upgrade["status"] == "FINALIZED" + assert upgrade["consensus_data"]["refunded_appeals"] == [ + {"transaction": first, "refund": charge} + ] + assert balance(upgrades, APPEALER) == charge + finalized = row_of(upgrades, first) + assert finalized["status"] == "FINALIZED" + assert finalized["appealed"] is False + assert finalized["data"][FEE_ACCOUNTING_KEY]["appeal_bonds"] == [] + assert APPEAL_RECOVERY_SNAPSHOT_KEY not in finalized["data"] + assert contract_states(upgrades)["finalized"] == {**state(1), **code()} + + +@pytest.mark.asyncio +async def test_queued_or_applied_upgrade_closes_appeal_admission(upgrades): + first = await accept(upgrades, 0, state(1)) + with upgrades.sessions() as session: + assert not TransactionsProcessor(session).upgrade_closes_appeals(first) + upgrade_hash = queue_upgrade(upgrades) + with upgrades.sessions() as session: + tp = TransactionsProcessor(session) + assert tp.upgrade_closes_appeals(first) + with pytest.raises(ValueError, match="CanNotAppeal"): + tp.admit_transaction_appeal( + first, + expected_decision_id=1, + submitted_at=int(time.time()), + appeal_deadline=int(time.time()) + 600, + retention_bps=0, + prepare_fee_accounting=lambda accounting: (accounting, 0), + ) + session.rollback() + for status, closes in (("FINALIZED", True), ("CANCELED", False)): + session.execute( + text( + "UPDATE transactions SET status = CAST(:status AS" + " transaction_status) WHERE hash = :hash" + ), + {"status": status, "hash": upgrade_hash}, + ) + assert tp.upgrade_closes_appeals(first) is closes + session.rollback() + + +@pytest.mark.asyncio +async def test_successful_appeal_after_upgrade_keeps_new_code(upgrades, monkeypatch): + upgrade_hash = queue_upgrade(upgrades) + assert (await run_upgrade(upgrades, upgrade_hash))["status"] == "FINALIZED" + later = await accept(upgrades, 0, {**state(1), **code()}) + + await succeed_validator_appeal(upgrades, monkeypatch, later) + + assert row_of(upgrades, upgrade_hash)["status"] == "FINALIZED" + assert row_of(upgrades, later)["status"] == "PENDING" + assert contract_states(upgrades) == { + "accepted": {**state(0), **code()}, + "finalized": {**state(0), **code()}, + } + + +@pytest.mark.asyncio +async def test_rewind_never_resurrects_canceled_rows_or_upgrades(upgrades, monkeypatch): + first = await accept(upgrades, 0, state(1)) + canceled = insert_pending(upgrades, 1) + with upgrades.sessions() as session: + session.query(Transactions).filter_by(hash=canceled).one().status = "CANCELED" + session.commit() + upgrade_hash = queue_upgrade(upgrades, legacy_finalized=True) + + await succeed_validator_appeal(upgrades, monkeypatch, first) + + assert row_of(upgrades, first)["status"] == "PENDING" + assert row_of(upgrades, canceled)["status"] == "CANCELED" + assert row_of(upgrades, upgrade_hash)["status"] == "FINALIZED" + # The restored pre-state must not drop the code of the upgrade that overtook it. + assert contract_states(upgrades)["accepted"] == {**state(0), **code()} + + +@pytest.mark.asyncio +async def test_unverifiable_predecessor_cancels_upgrade_without_side_effects( + upgrades, +): + first = await accept(upgrades, 0, state(1)) + second = await accept(upgrades, 1, state(2)) + with upgrades.sessions() as session: + strip_finalization_state(session, first) + session.execute( + text( + "UPDATE transactions SET contract_snapshot = jsonb_set(" + "contract_snapshot, '{states,accepted}', CAST(:tampered AS jsonb))" + " WHERE hash=:hash" + ), + {"hash": second, "tampered": json.dumps(state(7))}, + ) + session.commit() + before = contract_states(upgrades) + upgrades.rollup.reset_mock() + upgrade_hash = queue_upgrade(upgrades) + + upgrade = await run_upgrade(upgrades, upgrade_hash) + + assert upgrade["status"] == "CANCELED" + assert upgrade["consensus_data"] == { + "upgrade_result": "failed", + "error": "predecessor_finalization_state", + "transaction": first, + } + assert row_of(upgrades, first)["status"] == "ACCEPTED" + assert row_of(upgrades, second)["status"] == "ACCEPTED" + assert contract_states(upgrades) == before + upgrades.rollup.emit_transaction_event.assert_not_called() + + +@pytest.mark.asyncio +async def test_upgrade_retry_after_crash_never_refunds_or_settles_twice( + upgrades, monkeypatch +): + first = await accept(upgrades, 0, state(1)) + second = await accept(upgrades, 1, state(2)) + add_fee_accounting(upgrades, first) + charge = add_fee_accounting(upgrades, second, appeal=True) + upgrade_hash = queue_upgrade(upgrades) + worker = upgrades.worker + finalize_transaction = worker._finalize_transaction + crashed = [] + + async def crash_once(transaction, session): + if transaction.hash == second and not crashed: + crashed.append(transaction.hash) + raise RuntimeError("worker died mid-close") + await finalize_transaction(transaction, session) + + monkeypatch.setattr(worker, "_finalize_transaction", crash_once) + + interrupted = await run_upgrade(upgrades, upgrade_hash) + + assert crashed == [second] + assert interrupted["status"] == "PENDING" + assert interrupted["data"]["upgrade_close"] == { + "closed_transactions": [first, second], + "refunded_appeals": [{"transaction": second, "refund": charge}], + } + assert row_of(upgrades, first)["status"] == "FINALIZED" + assert row_of(upgrades, second)["status"] == "ACCEPTED" + assert row_of(upgrades, second)["appealed"] is False + assert balance(upgrades, APPEALER) == charge + settled_first = row_of(upgrades, first)["data"][FEE_ACCOUNTING_KEY] + sender_balance = balance(upgrades, SENDER) + assert CODE_KEY not in contract_states(upgrades)["accepted"] + + upgrade = await run_upgrade(upgrades, upgrade_hash) + + assert upgrade["status"] == "FINALIZED" + assert upgrade["consensus_data"]["closed_transactions"] == [first, second] + assert upgrade["consensus_data"]["refunded_appeals"] == [ + {"transaction": second, "refund": charge} + ] + assert balance(upgrades, APPEALER) == charge + assert row_of(upgrades, first)["data"][FEE_ACCOUNTING_KEY] == settled_first + second_row = row_of(upgrades, second) + assert second_row["status"] == "FINALIZED" + second_accounting = second_row["data"][FEE_ACCOUNTING_KEY] + assert [a["refund"] for a in second_accounting["aborted_appeals"]] == [charge] + # total_refunded also counts the appeal refund paid to the appealer. + assert ( + balance(upgrades, SENDER) + == sender_balance + second_accounting["total_refunded"] - charge + ) + assert contract_states(upgrades)["finalized"] == {**state(2), **code()} + with upgrades.sessions() as session: + assert await worker.claim_next_transaction(session) is None + + +@pytest.mark.asyncio +async def test_upgrade_defers_while_stale_leased_appeal_may_still_run(upgrades): + first = await accept(upgrades, 0, state(1)) + with upgrades.sessions() as session: + session.execute( + text( + "UPDATE transactions SET status = 'COMMITTING'," + " blocked_at = NOW() - INTERVAL '1 hour', worker_id = 'dead'," + " data = data || CAST(:snapshot AS jsonb) WHERE hash = :hash" + ), + { + "hash": first, + "snapshot": json.dumps( + {APPEAL_RECOVERY_SNAPSHOT_KEY: {"status": "ACCEPTED"}} + ), + }, + ) + session.commit() + before_row = row_of(upgrades, first) + before_states = contract_states(upgrades) + upgrade_hash = queue_upgrade(upgrades) + + upgrade = await run_upgrade(upgrades, upgrade_hash) + + assert upgrade["status"] == "PENDING" + assert "upgrade_close" not in upgrade["data"] + with upgrades.sessions() as session: + assert ( + session.execute( + text("SELECT blocked_at FROM transactions WHERE hash = :hash"), + {"hash": upgrade_hash}, + ).scalar_one() + is None + ) + after_row = row_of(upgrades, first) + assert after_row["status"] == "COMMITTING" + assert after_row["data"] == before_row["data"] + assert contract_states(upgrades) == before_states + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["UNDETERMINED", "LEADER_TIMEOUT"]) +async def test_upgrade_finalizes_undecided_predecessor_without_promotion( + upgrades, status +): + first = await accept(upgrades, 0, state(1)) + with upgrades.sessions() as session: + session.execute( + text( + "UPDATE transactions SET status = CAST(:status AS" + " transaction_status), value = 50, value_credited = false" + " WHERE hash = :hash" + ), + {"status": status, "hash": first}, + ) + session.commit() + sender_balance = balance(upgrades, SENDER) + upgrade_hash = queue_upgrade(upgrades) + + upgrade = await run_upgrade(upgrades, upgrade_hash) + + assert upgrade["status"] == "FINALIZED" + assert upgrade["consensus_data"]["closed_transactions"] == [first] + assert row_of(upgrades, first)["status"] == "FINALIZED" + assert balance(upgrades, SENDER) == sender_balance + 50 + assert contract_states(upgrades)["finalized"] == {**state(0), **code()} + + +@pytest.mark.asyncio +async def test_legacy_interleaved_upgrade_keeps_code_when_older_row_finalizes( + upgrades, capsys +): + from backend.database_handler.finalization_legacy_report import report + + first = await accept(upgrades, 0, state(1)) + upgrade_hash = queue_upgrade(upgrades, legacy_finalized=True) + + capsys.readouterr() + with upgrades.sessions() as session: + totals = report(session, limit=10) + lines = capsys.readouterr().out.strip().splitlines() + assert lines == [ + f"{first}\t{CONTRACT}\tinterleaved upgrade ({upgrade_hash})\tACCEPTED" + ] + assert dict(totals) == {"interleaved upgrade": 1} + + await finalize(upgrades, first) + + assert contract_states(upgrades)["finalized"] == {**state(1), **code()} + with upgrades.sessions() as session: + totals = report(session, limit=10) + assert dict(totals) == {} diff --git a/tests/unit/consensus/test_appeal_state_restore.py b/tests/unit/consensus/test_appeal_state_restore.py index fb46e1e36..4a6686f7c 100644 --- a/tests/unit/consensus/test_appeal_state_restore.py +++ b/tests/unit/consensus/test_appeal_state_restore.py @@ -118,6 +118,7 @@ async def test_missing_snapshot_non_deploy_refetches_from_db(self, monkeypatch): not from `{}`.""" stored_accepted = {"code_slot": "b64code", "storage_slot": "value"} transactions_processor = Mock() + transactions_processor.get_latest_finalized_upgrade_after.return_value = None transactions_processor.get_transaction_by_hash.return_value = { "hash": TX_HASH, "contract_snapshot": _make_snapshot_dict(stored_accepted), @@ -213,6 +214,7 @@ async def test_in_memory_snapshot_restores_saved_state(self, monkeypatch): saved_accepted = {"code_slot": "b64code", "counter": "0x01"} snapshot = ContractSnapshot.from_dict(_make_snapshot_dict(saved_accepted)) transactions_processor = Mock() + transactions_processor.get_latest_finalized_upgrade_after.return_value = None contract_processor = Mock() transaction = _make_transaction(contract_snapshot=snapshot) @@ -229,6 +231,33 @@ async def test_in_memory_snapshot_restores_saved_state(self, monkeypatch): ) transactions_processor.get_transaction_by_hash.assert_not_called() + @pytest.mark.asyncio + async def test_restore_keeps_code_of_later_finalized_upgrade(self, monkeypatch): + """A legacy upgrade that finalized after this decision keeps its code.""" + saved_accepted = {"counter": "0x01"} + snapshot = ContractSnapshot.from_dict(_make_snapshot_dict(saved_accepted)) + transactions_processor = Mock() + transactions_processor.get_latest_finalized_upgrade_after.return_value = { + "hash": "0xupgrade", + "new_code": "class C: pass", + "genvm_executor_selector": None, + } + contract_processor = Mock() + + await _run_validator_appeal( + _make_algorithm(), + _make_transaction(contract_snapshot=snapshot), + transactions_processor, + contract_processor, + monkeypatch, + ) + + code_key, code_value = consensus_base.upgrade_code_slot("class C: pass", None) + contract_processor.update_contract_state.assert_called_once_with( + CONTRACT_ADDRESS, + accepted_state={"counter": "0x01", code_key: code_value}, + ) + @pytest.mark.asyncio async def test_validator_appeal_filters_live_registry_through_frozen_pool( self, monkeypatch @@ -435,7 +464,7 @@ async def test_ancestor_rewind_refunds_active_descendant_appeal_before_reset( } } transactions_processor = Mock() - transactions_processor.get_newer_transactions.return_value = [ + transactions_processor.get_rewindable_newer_transactions.return_value = [ {"hash": future_hash, "appealed": True, "data": stale_data} ] transactions_processor.get_transaction_by_hash.return_value = { @@ -476,3 +505,42 @@ async def test_ancestor_rewind_refunds_active_descendant_appeal_before_reset( TransactionStatus.PENDING, context.msg_handler, ) + + +@pytest.mark.asyncio +async def test_rewind_never_resets_terminal_or_upgrade_rows(monkeypatch): + transactions_processor = Mock() + transactions_processor.get_rewindable_newer_transactions.return_value = [ + {"hash": "0xfinal", "status": TransactionStatus.FINALIZED.value, "type": 2}, + {"hash": "0xcanceled", "status": TransactionStatus.CANCELED.value, "type": 2}, + { + "hash": "0xupgrade", + "status": TransactionStatus.PENDING.value, + "type": TransactionType.UPGRADE_CONTRACT.value, + }, + {"hash": "0xopen", "status": TransactionStatus.ACCEPTED.value, "type": 2}, + ] + context = SimpleNamespace( + transaction=SimpleNamespace(hash=TX_HASH), + transactions_processor=transactions_processor, + accounts_manager=Mock(), + msg_handler=Mock(), + ) + status_update = AsyncMock() + monkeypatch.setattr( + ConsensusAlgorithm, + "dispatch_transaction_status_update", + staticmethod(status_update), + ) + + await ConsensusAlgorithm.__new__(ConsensusAlgorithm).rollback_transactions(context) + + transactions_processor.reset_transaction_for_recomputation.assert_called_once_with( + "0xopen", {} + ) + status_update.assert_awaited_once_with( + transactions_processor, + "0xopen", + TransactionStatus.PENDING, + context.msg_handler, + ) diff --git a/tests/unit/consensus/test_claim_column_manifest.py b/tests/unit/consensus/test_claim_column_manifest.py index f96b713cc..08e93f84d 100644 --- a/tests/unit/consensus/test_claim_column_manifest.py +++ b/tests/unit/consensus/test_claim_column_manifest.py @@ -93,9 +93,9 @@ "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. + # Finalization: FinalizingState separately reads the transaction's saved + # post-state and merges history via DB-side jsonb updates; the full + # rollback snapshot is deliberately omitted from the claim. "claim_next_finalization": _COMMON_OMISSIONS | { "contract_snapshot", diff --git a/tests/unit/consensus/test_decisions_accepted.py b/tests/unit/consensus/test_decisions_accepted.py index 04c930136..03198d520 100644 --- a/tests/unit/consensus/test_decisions_accepted.py +++ b/tests/unit/consensus/test_decisions_accepted.py @@ -144,13 +144,19 @@ def test_saves_contract_snapshot_when_missing(self): ) e = _find_effect(pre, SetContractSnapshotEffect) assert e is not None - assert e.snapshot_dict == {"s": 1} + assert e.snapshot_dict == {"s": 1, "finalization_state": {"slot1": b"data"}} - def test_no_contract_snapshot_when_already_saved(self): + def test_adds_post_state_to_existing_rollback_snapshot(self): + snapshot = {"states": {"accepted": {"before": "value"}}} pre, _, _, _ = decide_accepted( - **_base_accepted_kwargs(has_contract_snapshot=True) + **_base_accepted_kwargs( + has_contract_snapshot=True, contract_snapshot_dict=snapshot + ) ) - assert _find_effect(pre, SetContractSnapshotEffect) is None + effect = _find_effect(pre, SetContractSnapshotEffect) + assert effect.snapshot_dict["states"] == snapshot["states"] + assert effect.snapshot_dict["finalization_state"] == {"slot1": b"data"} + assert "finalization_state" not in snapshot def test_post_effects_contain_status_update(self): _, post, _, _ = decide_accepted(**_base_accepted_kwargs()) diff --git a/tests/unit/consensus/test_helpers.py b/tests/unit/consensus/test_helpers.py index a4c129154..0159566d1 100644 --- a/tests/unit/consensus/test_helpers.py +++ b/tests/unit/consensus/test_helpers.py @@ -183,17 +183,27 @@ def get_pending_transactions(self): result.append(transaction) return sorted(result, key=lambda x: x["created_at"]) - def get_newer_transactions(self, transaction_hash: str): + def get_rewindable_newer_transactions(self, transaction_hash: str): current_transaction = self.get_transaction_by_hash(transaction_hash) result = [] for transaction in self.transactions.values(): - if (transaction["created_at"] > current_transaction["created_at"]) and ( - transaction["to_address"] == current_transaction["to_address"] + if ( + transaction["created_at"] > current_transaction["created_at"] + and transaction["to_address"] == current_transaction["to_address"] + and transaction["status"] + not in ( + TransactionStatus.FINALIZED.value, + TransactionStatus.CANCELED.value, + ) + and transaction.get("type") != TransactionType.UPGRADE_CONTRACT.value ): result.append(transaction) return sorted(result, key=lambda x: x["created_at"]) + def get_latest_finalized_upgrade_after(self, transaction_hash: str): + return None + def update_consensus_history( self, transaction_hash: str, diff --git a/tests/unit/consensus/test_transaction_finalized_state.py b/tests/unit/consensus/test_transaction_finalized_state.py new file mode 100644 index 000000000..6de1455e9 --- /dev/null +++ b/tests/unit/consensus/test_transaction_finalized_state.py @@ -0,0 +1,190 @@ +"""Finalization state provenance and backward-compatible snapshot decoding.""" + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from backend.consensus.base import ( + _transaction_finalization_state, + resolve_finalization_state, +) +from backend.consensus.errors import FinalizationStateError +from backend.database_handler.contract_snapshot import ContractSnapshot +from backend.node.types import _compute_contract_state_hash + + +BEFORE = {"slot": "before"} +AFTER = {"slot": "after"} +YOUNGER = {"slot": "younger"} + + +def context(*, saved=None, stored=None, live=YOUNGER, pre_states=None, successors=()): + pre_states = pre_states or {} + tp = Mock() + tp.get_transaction_finalization_state.return_value = stored + tp.get_successor_snapshot_hashes.return_value = list(successors) + tp.get_transaction_pre_state.side_effect = pre_states.get + factory = Mock(return_value=SimpleNamespace(states={"accepted": live})) + return SimpleNamespace( + transaction=SimpleNamespace( + hash="0xtx", to_address="0xcontract", contract_snapshot=saved + ), + transactions_processor=tp, + contract_snapshot_factory=factory, + ) + + +def receipt(*, state=None, expected=AFTER): + return SimpleNamespace( + contract_state=state, contract_state_hash=_compute_contract_state_hash(expected) + ) + + +def test_snapshot_roundtrip_keeps_pre_and_post_state_separate(): + saved = ContractSnapshot.from_dict( + { + "contract_address": "0xcontract", + "states": {"accepted": BEFORE, "finalized": {}}, + "finalization_state": AFTER, + } + ) + restored = ContractSnapshot.from_dict(saved.to_dict()) + assert restored.states["accepted"] == BEFORE + assert restored.finalization_state == AFTER + + +def test_old_snapshot_has_no_finalization_state(): + assert ( + ContractSnapshot.from_dict({"states": {"accepted": BEFORE}}).finalization_state + is None + ) + + +def test_uses_transaction_snapshot_without_reading_live_contract(): + ctx = context(saved=SimpleNamespace(finalization_state=AFTER)) + assert _transaction_finalization_state(ctx, receipt()) == AFTER + ctx.transactions_processor.get_transaction_finalization_state.assert_not_called() + ctx.contract_snapshot_factory.assert_not_called() + + +def test_narrow_worker_claim_reloads_only_post_state(): + ctx = context(stored=AFTER) + assert _transaction_finalization_state(ctx, receipt()) == AFTER + ctx.transactions_processor.get_transaction_finalization_state.assert_called_once_with( + "0xtx" + ) + ctx.contract_snapshot_factory.assert_not_called() + + +def test_saved_empty_state_is_valid_without_hash(): + ctx = context(stored={}) + assert _transaction_finalization_state(ctx, receipt(expected={})) == {} + ctx.contract_snapshot_factory.assert_not_called() + + +@pytest.mark.parametrize("source", ["snapshot", "database", "receipt"]) +def test_mismatched_post_state_is_rejected(source): + ctx = context( + saved=( + SimpleNamespace(finalization_state=YOUNGER) + if source == "snapshot" + else None + ), + stored=YOUNGER if source == "database" else None, + ) + with pytest.raises(FinalizationStateError, match="hash mismatch"): + _transaction_finalization_state( + ctx, receipt(state=YOUNGER if source == "receipt" else None) + ) + ctx.contract_snapshot_factory.assert_not_called() + + +def test_legacy_unstripped_receipt_remains_usable(): + ctx = context() + assert _transaction_finalization_state(ctx, receipt(state=AFTER)) == AFTER + ctx.contract_snapshot_factory.assert_not_called() + + +def test_legacy_live_head_requires_exact_hash_match(): + assert _transaction_finalization_state(context(live=AFTER), receipt()) == AFTER + with pytest.raises(FinalizationStateError, match="unverified contract head"): + _transaction_finalization_state(context(), receipt()) + + +def test_legacy_stripped_receipt_without_hash_cannot_authorize_live_state(): + ctx = context(live=AFTER) + with pytest.raises(FinalizationStateError, match="Missing transaction-local"): + _transaction_finalization_state(ctx, receipt(state={}, expected=None)) + ctx.contract_snapshot_factory.assert_not_called() + + +def test_legacy_row_recovers_from_first_matching_successor(): + ctx = context( + pre_states={"0xtx": BEFORE, "0xs1": AFTER, "0xs2": YOUNGER}, + successors=["0xs1", "0xs2"], + ) + assert _transaction_finalization_state(ctx, receipt()) == AFTER + tp = ctx.transactions_processor + tp.get_successor_snapshot_hashes.assert_called_once_with("0xtx", 3) + assert [c.args for c in tp.get_transaction_pre_state.call_args_list] == [ + ("0xtx",), + ("0xs1",), + ] + + +def test_mismatching_successors_are_skipped(): + ctx = context( + pre_states={"0xtx": BEFORE, "0xs1": YOUNGER, "0xs2": AFTER}, + successors=["0xs1", "0xs2"], + ) + assert resolve_finalization_state(ctx, receipt()) == ( + AFTER, + "successor 2 (0xs2)", + ) + + +def test_no_matching_pre_state_preserves_the_decision(): + ctx = context( + pre_states={"0xtx": BEFORE, "0xs1": YOUNGER, "0xs2": None}, + successors=["0xs1", "0xs2", "0xs3"], + ) + with pytest.raises(FinalizationStateError, match="unverified contract head"): + _transaction_finalization_state(ctx, receipt()) + assert ctx.transactions_processor.get_transaction_pre_state.call_count == 4 + + +def test_unchanged_state_write_recovers_from_own_pre_state(): + ctx = context(pre_states={"0xtx": AFTER}, successors=["0xs1"]) + assert resolve_finalization_state(ctx, receipt()) == (AFTER, "own pre-state") + ctx.transactions_processor.get_successor_snapshot_hashes.assert_not_called() + + +def test_without_expected_hash_no_pre_state_is_consulted(): + ctx = context(pre_states={"0xtx": AFTER}, successors=["0xs1"]) + with pytest.raises(FinalizationStateError, match="Missing transaction-local"): + _transaction_finalization_state(ctx, receipt(expected=None)) + ctx.transactions_processor.get_transaction_pre_state.assert_not_called() + ctx.transactions_processor.get_successor_snapshot_hashes.assert_not_called() + + +@pytest.mark.parametrize("source", ["database", "receipt"]) +def test_primary_source_skips_pre_state_lookup(source): + ctx = context( + stored=AFTER if source == "database" else None, + pre_states={"0xs1": AFTER}, + successors=["0xs1"], + ) + state = _transaction_finalization_state( + ctx, receipt(state=AFTER if source == "receipt" else None) + ) + assert state == AFTER + ctx.transactions_processor.get_transaction_pre_state.assert_not_called() + ctx.transactions_processor.get_successor_snapshot_hashes.assert_not_called() + + +def test_primary_source_mismatch_is_not_rescued_by_successors(): + ctx = context(stored=YOUNGER, pre_states={"0xs1": AFTER}, successors=["0xs1"]) + with pytest.raises(FinalizationStateError, match="hash mismatch"): + _transaction_finalization_state(ctx, receipt()) + ctx.transactions_processor.get_transaction_pre_state.assert_not_called() diff --git a/tests/unit/test_finalization_legacy_report.py b/tests/unit/test_finalization_legacy_report.py new file mode 100644 index 000000000..81f3dadd4 --- /dev/null +++ b/tests/unit/test_finalization_legacy_report.py @@ -0,0 +1,79 @@ +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from backend.database_handler.errors import ContractNotFoundError +from backend.database_handler.finalization_legacy_report import ( + UNRECOVERABLE, + classify, +) +from backend.node.types import _compute_contract_state_hash + + +AFTER = {"slot": "after"} +OTHER = {"slot": "other"} + + +def row(expected=AFTER): + return { + "hash": "0xtx", + "to_address": "0xcontract", + "queue_order": 1, + "contract_state_hash": _compute_contract_state_hash(expected), + } + + +def processor(pre_states=None, successors=()): + tp = Mock() + tp.get_transaction_finalization_state.return_value = None + tp.get_successor_snapshot_hashes.return_value = list(successors) + tp.get_transaction_pre_state.side_effect = (pre_states or {}).get + return tp + + +def head(state): + return Mock(return_value=SimpleNamespace(states={"accepted": state})) + + +@pytest.mark.parametrize( + "tp, live, expected", + [ + (processor(), AFTER, ("head", "")), + (processor({"0xtx": AFTER}), OTHER, ("own pre-state", "")), + ( + processor({"0xs1": OTHER, "0xs2": AFTER}, ["0xs1", "0xs2"]), + OTHER, + ("successor 2 (0xs2)", ""), + ), + ], +) +def test_classifies_recoverable_sources(tp, live, expected): + assert classify(row(), tp, head(live), Mock(return_value=False)) == expected + + +def test_missing_hash_is_unrecoverable_without_lookups(): + tp = processor() + factory = head(AFTER) + legacy = {**row(), "contract_state_hash": None} + assert classify(legacy, tp, factory, Mock()) == (UNRECOVERABLE, "no hash") + factory.assert_not_called() + tp.get_transaction_pre_state.assert_not_called() + + +@pytest.mark.parametrize( + "upgraded, reason", [(True, "upgrade after"), (False, "no matching snapshot")] +) +def test_unmatched_rows_report_why(upgraded, reason): + tp = processor({"0xs1": OTHER}, ["0xs1"]) + upgraded_after = Mock(return_value=upgraded) + assert classify(row(), tp, head(OTHER), upgraded_after) == (UNRECOVERABLE, reason) + upgraded_after.assert_called_once_with(row()) + + +def test_missing_contract_is_unrecoverable(): + factory = Mock(side_effect=ContractNotFoundError("0xcontract")) + assert classify(row(), processor(), factory, Mock()) == ( + UNRECOVERABLE, + "contract not found", + ) diff --git a/tests/unit/test_legacy_executor_selection.py b/tests/unit/test_legacy_executor_selection.py index 5a2130fff..7bc6db089 100644 --- a/tests/unit/test_legacy_executor_selection.py +++ b/tests/unit/test_legacy_executor_selection.py @@ -285,10 +285,15 @@ async def test_code_upgrade_preserves_executor_and_other_storage(monkeypatch, se }, ) session = MagicMock() - session.query.return_value.filter_by.return_value.one_or_none.return_value = ( - contract + query = ( + session.query.return_value.filter_by.return_value.with_for_update.return_value.populate_existing.return_value ) - tx = session.query.return_value.filter_by.return_value.one.return_value + query.one_or_none.return_value = contract + tx = query.one.return_value + tx.status = TransactionStatus.PENDING + tx.worker_id = "test" + tx.data = {} + session.execute.return_value.first.return_value = None dispatch = AsyncMock() monkeypatch.setattr( consensus_worker.ConsensusAlgorithm, @@ -299,14 +304,10 @@ async def test_code_upgrade_preserves_executor_and_other_storage(monkeypatch, se worker = consensus_worker.ConsensusWorker.__new__(consensus_worker.ConsensusWorker) worker.worker_id = "test" worker.msg_handler = Mock() - worker.current_transactions = {} - worker.get_session = MagicMock() - worker.release_transaction = Mock() + worker._generic_error_retries = {} code = LEGACY_CODE.decode() - await worker._process_upgrade_transaction( - {"hash": "0xabc", "to_address": ADDRESS, "data": {"new_code": code}}, session - ) + await worker._apply_upgrade_code(session, "0xabc", ADDRESS, code) assert tx.consensus_data["upgrade_result"] == "success" assert contract.genvm_executor_selector == selector diff --git a/tests/unit/test_studio_fees.py b/tests/unit/test_studio_fees.py index e82658c88..697996b74 100644 --- a/tests/unit/test_studio_fees.py +++ b/tests/unit/test_studio_fees.py @@ -14180,3 +14180,42 @@ def test_transaction_payload_exposes_only_consensus_child_transactions(): "0x" + "02" * 32, "0x" + "03" * 32, ] + + +@pytest.mark.parametrize( + "rounds, unresolved", + [ + (["Accepted"], True), + (["Accepted", "Validator Appeal Failed"], False), + (["Accepted", "Validator Appeal Successful"], False), + ], +) +def test_latest_appeal_admission_unresolved_only_before_an_outcome(rounds, unresolved): + from backend.protocol_rpc.fees import latest_appeal_admission_unresolved + + fees = { + "leaderTimeunitsAllocation": 100, + "validatorTimeunitsAllocation": 200, + "appealRounds": 0, + "executionBudgetPerRound": 0, + "rotations": [0], + } + accounting = create_fee_accounting( + fees_distribution=fees, + num_of_validators=5, + submitted_value=required_fee_deposit(fees, 5), + user_value=0, + sender="0x" + "11" * 20, + ) + charge = calculate_appeal_charge(fees, current_round=0, status="ACCEPTED") + appealed = record_appeal_bond( + accounting, + amount=charge["bond"] + charge["funding"], + appealer="0x" + "22" * 20, + current_round=0, + status="ACCEPTED", + ) + history = {"consensus_results": [{"consensus_round": r} for r in rounds]} + + assert latest_appeal_admission_unresolved(appealed, history) is unresolved + assert latest_appeal_admission_unresolved(accounting, history) is False diff --git a/tests/unit/test_worker_transient_fatal.py b/tests/unit/test_worker_transient_fatal.py index 78ea59d09..6e264c868 100644 --- a/tests/unit/test_worker_transient_fatal.py +++ b/tests/unit/test_worker_transient_fatal.py @@ -319,6 +319,90 @@ async def test_acceptance_repair_finalization_honors_retry_backoff(): create_task.assert_not_called() +@pytest.mark.asyncio +async def test_finalization_claim_excludes_backed_off_heads(): + worker = _make_worker() + session = MagicMock(spec=Session) + worker.claim_next_appeal = AsyncMock(return_value=None) + worker.claim_next_finalization = AsyncMock(return_value={"hash": "0xother"}) + worker.claim_next_transaction = AsyncMock(return_value=None) + worker._generic_error_retries["0xstuck"] = { + "count": 1, + "last_attempt": 100.0, + "last_error": "missing finalization state", + } + + with ( + patch("backend.consensus.worker.time.time", return_value=100.0), + patch.object(worker, "release_transaction") as release_transaction, + patch("backend.consensus.worker.asyncio.create_task") as create_task, + patch.object(worker, "_process_finalization_task", new=MagicMock()), + ): + assert await worker._try_claim_work(session) is True + + worker.claim_next_finalization.assert_awaited_once_with( + session, exclude_hashes=["0xstuck"] + ) + release_transaction.assert_not_called() + create_task.assert_called_once() + + +def test_backoff_hashes_lists_only_in_window_entries(): + worker = _make_worker() + worker._no_validators_base_backoff = 30.0 + worker._generic_error_base_backoff = 10.0 + worker._no_validators_retries = { + "0xnv_active": {"count": 1, "last_attempt": 90.0}, + "0xnv_expired": {"count": 1, "last_attempt": 50.0}, + } + worker._generic_error_retries = { + "0xge_active": {"count": 2, "last_attempt": 85.0, "last_error": "x"}, + "0xge_expired": {"count": 1, "last_attempt": 80.0, "last_error": "x"}, + } + + with patch("backend.consensus.worker.time.time", return_value=100.0): + assert worker._backoff_hashes() == ["0xge_active", "0xnv_active"] + assert worker._is_in_backoff({"hash": "0xge_active"}) is True + assert worker._is_in_backoff({"hash": "0xge_expired"}) is False + + +@pytest.mark.asyncio +async def test_successful_finalization_clears_retry_tracking(): + worker = _make_worker() + session = MagicMock(spec=Session) + worker._no_validators_retries["0xtx"] = {"count": 1, "last_attempt": 1.0} + worker._generic_error_retries["0xtx"] = { + "count": 2, + "last_attempt": 1.0, + "last_error": "boom", + } + worker._generic_error_retries["0xother"] = { + "count": 1, + "last_attempt": 1.0, + "last_error": "boom", + } + worker.consensus_algorithm = MagicMock() + worker.consensus_algorithm.can_finalize_transaction.return_value = True + worker.consensus_algorithm.process_finalization = AsyncMock() + worker.usage_metrics_service = MagicMock() + worker.usage_metrics_service.send_finalized_transaction_metrics = AsyncMock() + + with ( + patch( + "backend.consensus.worker.Transaction.from_dict", + return_value=MagicMock(hash="0xtx"), + ), + patch("backend.consensus.base.transactions_processor_factory"), + patch.object(worker, "release_transaction"), + ): + await worker.process_finalization({"hash": "0xtx"}, session) + + worker.consensus_algorithm.process_finalization.assert_awaited_once() + assert "0xtx" not in worker._no_validators_retries + assert "0xtx" not in worker._generic_error_retries + assert "0xother" in worker._generic_error_retries + + @pytest.mark.asyncio async def test_generic_appeal_error_is_retryable_without_cancellation(): worker = _make_worker()