From 343da897eb1e2457be20ebd27e5b0d3584383d26 Mon Sep 17 00:00:00 2001 From: Edgars Date: Fri, 16 Oct 2026 16:26:25 +0100 Subject: [PATCH 1/6] fix(consensus): finalize transaction-local accepted state Persist each successful acceptance's post-state alongside its rollback snapshot, then reload only that post-state during finalization. Verify receipt hashes and preserve agreed decisions when legacy state cannot be recovered safely, rather than promoting a younger accepted head or canceling the older transaction. Cover overlap, deployments, failed and successful appeals, recomputation, empty state, execution errors, and legacy rows through the real PostgreSQL worker path. No schema migration is required. Rollout must drain or explicitly recover pre-fix in-flight decisions that lack verifiable state. Validation: 1690 unit tests passed (repository-recommended RPC manager exclusion); 244 database tests passed, 1 existing xfail; changed-file pre-commit checks passed. Full stack E2E not run. Local branch only pending release-plan confirmation. --- backend/consensus/base.py | 78 ++-- backend/consensus/decisions.py | 12 +- backend/consensus/errors.py | 4 + backend/consensus/worker.py | 11 +- backend/database_handler/contract_snapshot.py | 4 + .../transactions_processor.py | 10 + backend/node/types.py | 3 +- .../test_transaction_finalized_state.py | 357 ++++++++++++++++++ .../consensus/test_claim_column_manifest.py | 6 +- .../unit/consensus/test_decisions_accepted.py | 14 +- .../test_transaction_finalized_state.py | 113 ++++++ 11 files changed, 571 insertions(+), 41 deletions(-) create mode 100644 tests/db-sqlalchemy/test_transaction_finalized_state.py create mode 100644 tests/unit/consensus/test_transaction_finalized_state.py diff --git a/backend/consensus/base.py b/backend/consensus/base.py index 0a7c0b340..8353ce4d4 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 ( @@ -3727,10 +3728,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 +3882,42 @@ async def handle(self, context): return None +def _transaction_finalization_state( + context: TransactionContext, leader_receipt: Receipt +) -> dict[str, str]: + saved_snapshot = context.transaction.contract_snapshot + state = saved_snapshot.finalization_state if saved_snapshot is not None else None + 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 + + 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 + + # 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 + + raise FinalizationStateError( + f"Missing transaction-local finalization state for {context.transaction.hash}; " + "cannot finalize from an unverified contract head" + ) + + class FinalizingState(TransactionState): """ Class representing the finalizing state of a transaction. @@ -3897,12 +3932,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 +3946,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 +3980,9 @@ 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=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..3f65d06dc 100644 --- a/backend/consensus/worker.py +++ b/backend/consensus/worker.py @@ -21,6 +21,7 @@ NoValidatorsAvailableError, ) 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) @@ -96,8 +97,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], ...] = ( @@ -972,7 +973,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: @@ -1709,7 +1712,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}" 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/transactions_processor.py b/backend/database_handler/transactions_processor.py index 92d3d6b52..98a76a888 100644 --- a/backend/database_handler/transactions_processor.py +++ b/backend/database_handler/transactions_processor.py @@ -2308,6 +2308,16 @@ 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 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/tests/db-sqlalchemy/test_transaction_finalized_state.py b/tests/db-sqlalchemy/test_transaction_finalized_state.py new file mode 100644 index 000000000..669691564 --- /dev/null +++ b/tests/db-sqlalchemy/test_transaction_finalized_state.py @@ -0,0 +1,357 @@ +"""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, +) +import backend.consensus.base as consensus_base +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 +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 + + +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), + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize("overlap", [False, True]) +async def test_legacy_row_requires_matching_receipt_hash(lifecycle, overlap): + first = await accept(lifecycle, 0, state(1)) + if overlap: + await accept(lifecycle, 1, state(2)) + with lifecycle.sessions() as session: + session.execute( + text( + "UPDATE transactions SET contract_snapshot = contract_snapshot - 'finalization_state' WHERE hash=:hash" + ), + {"hash": first}, + ) + session.commit() + if not overlap: + await finalize(lifecycle, first) + else: + 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 if overlap else 1 + ) 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_transaction_finalized_state.py b/tests/unit/consensus/test_transaction_finalized_state.py new file mode 100644 index 000000000..48d5d940c --- /dev/null +++ b/tests/unit/consensus/test_transaction_finalized_state.py @@ -0,0 +1,113 @@ +"""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 +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): + tp = Mock() + tp.get_transaction_finalization_state.return_value = stored + 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() From 84dea930279c6ec9144241947de0f331a4dbaeb2 Mon Sep 17 00:00:00 2001 From: Daniel Rojas Date: Thu, 22 Oct 2026 11:57:16 -0500 Subject: [PATCH 2/6] fix(consensus): skip backed-off finalization heads without reordering contracts A finalization head in retry backoff was claimed and released every poll, so the oldest stuck head starved every other contract's finalization. claim_next_finalization now takes the worker's backed-off hashes and skips contracts whose head (or an older non-terminal tx) is backing off, while the existing ordering guard still blocks younger same-contract txs. Retry tracking is cleared once a finalization succeeds. --- backend/consensus/worker.py | 92 +++++++++++++------ .../test_finalization_ordering.py | 84 +++++++++++++++++ tests/unit/test_worker_transient_fatal.py | 84 +++++++++++++++++ 3 files changed, 231 insertions(+), 29 deletions(-) diff --git a/backend/consensus/worker.py b/backend/consensus/worker.py index 3f65d06dc..553e02450 100644 --- a/backend/consensus/worker.py +++ b/backend/consensus/worker.py @@ -5,7 +5,7 @@ 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 @@ -293,10 +293,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 """ @@ -371,6 +378,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 @@ -418,6 +434,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 @@ -2029,6 +2046,8 @@ async def process_finalization(self, finalization_data: dict, session: Session): 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( @@ -2226,33 +2245,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 * ( @@ -2262,15 +2270,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.""" @@ -2311,9 +2342,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/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/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() From 7a394045c0210a08d5000c4cad421db7ba2f9791 Mon Sep 17 00:00:00 2001 From: Daniel Rojas Date: Thu, 22 Oct 2026 12:02:14 -0500 Subject: [PATCH 3/6] fix(consensus): recover legacy finalization state from successor pre-states Rows accepted before transaction-local finalization state was persisted can only finalize when the moving contract head still hashes to their receipt. Once a successor is accepted the head moves and the row is stuck. A successor's saved pre-state is exactly the predecessor's post-state when nothing intervened, so after the existing sources fail the resolver now tries the transaction's own pre-state (unchanged-state writes) and up to three successor pre-states, accepting only one whose hash matches the receipt's contract_state_hash. Anything else still raises FinalizationStateError without canceling. Adds a read-only finalization_legacy_report CLI that classifies pending legacy rows with the same resolver before deploy. --- backend/consensus/base.py | 49 +++++- .../finalization_legacy_report.py | 148 ++++++++++++++++++ .../transactions_processor.py | 32 ++++ .../test_transaction_finalized_state.py | 97 +++++++++--- .../test_transaction_finalized_state.py | 81 +++++++++- tests/unit/test_finalization_legacy_report.py | 79 ++++++++++ 6 files changed, 462 insertions(+), 24 deletions(-) create mode 100644 backend/database_handler/finalization_legacy_report.py create mode 100644 tests/unit/test_finalization_legacy_report.py diff --git a/backend/consensus/base.py b/backend/consensus/base.py index 8353ce4d4..61a3475de 100644 --- a/backend/consensus/base.py +++ b/backend/consensus/base.py @@ -3882,17 +3882,37 @@ async def handle(self, context): return None -def _transaction_finalization_state( +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 -) -> dict[str, str]: +) -> 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: @@ -3900,7 +3920,7 @@ def _transaction_finalization_state( raise FinalizationStateError( f"Finalization state hash mismatch for transaction {context.transaction.hash}" ) - return state + 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. @@ -3910,7 +3930,14 @@ def _transaction_finalization_state( snapshot.states.get("accepted") if snapshot is not None else None ) if _compute_contract_state_hash(accepted_state) == expected_hash: - return accepted_state + 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}; " @@ -3918,6 +3945,20 @@ def _transaction_finalization_state( ) +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. diff --git a/backend/database_handler/finalization_legacy_report.py b/backend/database_handler/finalization_legacy_report.py new file mode 100644 index 000000000..20bc5ad43 --- /dev/null +++ b/backend/database_handler/finalization_legacy_report.py @@ -0,0 +1,148 @@ +"""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. +""" + +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 + ) + """ +) + + +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()) + 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. 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 98a76a888..d4dd78ad1 100644 --- a/backend/database_handler/transactions_processor.py +++ b/backend/database_handler/transactions_processor.py @@ -2318,6 +2318,38 @@ def get_transaction_finalization_state(self, transaction_hash: str) -> dict | No {"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_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/tests/db-sqlalchemy/test_transaction_finalized_state.py b/tests/db-sqlalchemy/test_transaction_finalized_state.py index 669691564..4ff086840 100644 --- a/tests/db-sqlalchemy/test_transaction_finalized_state.py +++ b/tests/db-sqlalchemy/test_transaction_finalized_state.py @@ -328,30 +328,91 @@ async def test_accepted_execution_error_does_not_promote_contract_state(lifecycl } +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 -@pytest.mark.parametrize("overlap", [False, True]) -async def test_legacy_row_requires_matching_receipt_hash(lifecycle, overlap): +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)) - if overlap: - await accept(lifecycle, 1, state(2)) + 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 = contract_snapshot - 'finalization_state' WHERE hash=:hash" + "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": first}, + {"hash": second, "tampered": json.dumps(state(7))}, ) session.commit() - if not overlap: - await finalize(lifecycle, first) - else: - 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() + 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 if overlap else 1 - ) + 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) diff --git a/tests/unit/consensus/test_transaction_finalized_state.py b/tests/unit/consensus/test_transaction_finalized_state.py index 48d5d940c..6de1455e9 100644 --- a/tests/unit/consensus/test_transaction_finalized_state.py +++ b/tests/unit/consensus/test_transaction_finalized_state.py @@ -5,7 +5,10 @@ import pytest -from backend.consensus.base import _transaction_finalization_state +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 @@ -16,9 +19,12 @@ YOUNGER = {"slot": "younger"} -def context(*, saved=None, stored=None, live=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( @@ -111,3 +117,74 @@ def test_legacy_stripped_receipt_without_hash_cannot_authorize_live_state(): 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", + ) From 4713ec3b52722b678dcd25af7ef509a2c38d985f Mon Sep 17 00:00:00 2001 From: Daniel Rojas Date: Thu, 22 Oct 2026 12:32:24 -0500 Subject: [PATCH 4/6] fix(consensus): finalize decided predecessors before a code upgrade applies A type 3 upgrade used to swap the code slot while earlier decided transactions on the same contract were still inside their appeal window. Those rows then finalized, or were re-executed by an appeal, against code they were never decided on, and their paid appeals stayed open across the code change. The upgrade now closes the contract's earlier lifecycle before touching code: - It defers while any other row on the contract holds a lease or is mid-consensus, so it never races live work (UPGRADE_DEFER_SECONDS). - Every ACCEPTED+SUCCESS predecessor must resolve a verified finalization state up front. If one cannot, the upgrade is canceled with predecessor_finalization_state and nothing is changed. - Each decided predecessor (ACCEPTED, UNDETERMINED, *_TIMEOUT) is force-finalized through the normal path: acceptance delivery repair, messages, fee settlement and value refunds. A pending paid appeal is refunded first via abort_pending_appeal_admission, which only aborts an admission with no recorded outcome. - Progress (closed rows, refunds) is persisted on the upgrade row in the same commit as each refund, and that write also renews the lease. A retry skips rows that are already final and can never refund twice. - The code is applied last, in one commit with the FINALIZED status, after rechecking ownership and that no predecessor reopened. - Never-started PENDING rows stay queued and run on the new code. Validation and contract errors cancel the upgrade as before. Transient failures go through the generic retry path, and rows that are already finalized stay final. Co-Authored-By: Claude --- backend/consensus/base.py | 12 + backend/consensus/worker.py | 624 ++++++++++++++---- backend/database_handler/accounts_manager.py | 40 ++ backend/protocol_rpc/endpoints.py | 3 +- backend/protocol_rpc/fees.py | 20 + tests/db-sqlalchemy/accounts_manager_test.py | 60 ++ .../test_finalization_starvation.py | 42 ++ .../test_transaction_finalized_state.py | 349 +++++++++- tests/unit/test_legacy_executor_selection.py | 19 +- tests/unit/test_studio_fees.py | 39 ++ 10 files changed, 1051 insertions(+), 157 deletions(-) diff --git a/backend/consensus/base.py b/backend/consensus/base.py index 61a3475de..7969de1f3 100644 --- a/backend/consensus/base.py +++ b/backend/consensus/base.py @@ -381,6 +381,18 @@ 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 transaction_genvm_executor_selector(transaction: Transaction) -> str | None: """Select and persist the deployment's executor, honoring explicit pins.""" explicit_selector = ( diff --git a/backend/consensus/worker.py b/backend/consensus/worker.py index 553e02450..750a62e42 100644 --- a/backend/consensus/worker.py +++ b/backend/consensus/worker.py @@ -2,6 +2,7 @@ import os import asyncio +import json import time import uuid from contextlib import asynccontextmanager @@ -15,10 +16,14 @@ 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 @@ -27,6 +32,11 @@ # 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 @@ -41,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): @@ -1649,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. @@ -1685,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: @@ -1832,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" ) - ).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, - }, + 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, + ) + + 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): """ @@ -1985,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 @@ -2030,17 +2372,7 @@ 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( 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/protocol_rpc/endpoints.py b/backend/protocol_rpc/endpoints.py index 9977c4768..15f689921 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 diff --git a/backend/protocol_rpc/fees.py b/backend/protocol_rpc/fees.py index 108b2623a..5d3b93b2f 100644 --- a/backend/protocol_rpc/fees.py +++ b/backend/protocol_rpc/fees.py @@ -2735,6 +2735,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_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 index 4ff086840..53a8de9b5 100644 --- a/tests/db-sqlalchemy/test_transaction_finalized_state.py +++ b/tests/db-sqlalchemy/test_transaction_finalized_state.py @@ -16,20 +16,30 @@ 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 +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 @@ -416,3 +426,340 @@ async def test_legacy_report_classifies_without_writing(lifecycle, capsys): 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): + with lifecycle.sessions() as session: + return admin_upgrade_contract_code(session, CONTRACT, UPGRADE_CODE)[ + "transaction_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 + + +@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_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()} 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 4c6fd14d4..9fd2c8579 100644 --- a/tests/unit/test_studio_fees.py +++ b/tests/unit/test_studio_fees.py @@ -13944,3 +13944,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 From b2fc9eea1a44f0035d10cbcd064529ff26e93048 Mon Sep 17 00:00:00 2001 From: Daniel Rojas Date: Thu, 22 Oct 2026 12:32:39 -0500 Subject: [PATCH 5/6] fix(consensus): upgrades permanently close the appeal and rewind lifecycle Force-finalizing predecessors only holds if nothing can reopen them afterwards. Before this change, an appeal could still be admitted while the upgrade was queued. A successful appeal's rewind also reset every newer row on the contract to PENDING, including FINALIZED and CANCELED rows and the upgrade itself. That replayed an applied upgrade and resurrected terminal transactions. - Appeal admission is rejected with CanNotAppeal when a PENDING or FINALIZED upgrade is queued after the decision. The check runs in the RPC precheck and again under the row lock in admit_transaction_appeal. - Rewinds only reset rows that can still be replayed (get_rewindable_newer_transactions). Terminal rows and upgrades are never reset, and rollback_transactions guards this again in memory. Upgrades applied before this change could overtake decided rows that are still open. When such an older row finalizes, or a validator appeal on it restores its pre-state, the code of the latest FINALIZED upgrade after it is kept (with_later_upgrade_code). Without this, the old code would come back. The legacy finalization report now also lists these rows as "interleaved upgrade ()" so operators can see them before deploying. Co-Authored-By: Claude --- backend/consensus/base.py | 51 ++++++- .../finalization_legacy_report.py | 43 +++++- .../transactions_processor.py | 65 +++++++- backend/protocol_rpc/endpoints.py | 4 + .../test_transaction_finalized_state.py | 142 +++++++++++++++++- .../consensus/test_appeal_state_restore.py | 70 ++++++++- tests/unit/consensus/test_helpers.py | 16 +- 7 files changed, 379 insertions(+), 12 deletions(-) diff --git a/backend/consensus/base.py b/backend/consensus/base.py index 7969de1f3..38c9c728d 100644 --- a/backend/consensus/base.py +++ b/backend/consensus/base.py @@ -393,6 +393,31 @@ def upgrade_code_slot( 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 = ( @@ -1423,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( @@ -1462,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 ( @@ -4035,7 +4074,11 @@ async def handle(self, context): if should_finalize_contract: context.contract_processor.update_contract_state( context.transaction.to_address, - finalized_state=finalization_state, + finalized_state=with_later_upgrade_code( + context.transactions_processor, + context.transaction.hash, + finalization_state, + ), ) _dispatch_messages_for_phase( diff --git a/backend/database_handler/finalization_legacy_report.py b/backend/database_handler/finalization_legacy_report.py index 20bc5ad43..82ba7a5e3 100644 --- a/backend/database_handler/finalization_legacy_report.py +++ b/backend/database_handler/finalization_legacy_report.py @@ -2,6 +2,8 @@ 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 @@ -60,6 +62,38 @@ """ ) +_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, @@ -114,6 +148,12 @@ def upgraded_after(row: dict) -> bool: ) 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 @@ -122,7 +162,8 @@ def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=( "List ACCEPTED transactions without a saved finalization state and " - "classify how each would finalize. Read-only." + "classify how each would finalize, plus decided transactions that an " + "earlier code upgrade overtook. Read-only." ) ) parser.add_argument( diff --git a/backend/database_handler/transactions_processor.py b/backend/database_handler/transactions_processor.py index d4dd78ad1..204905203 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,10 @@ 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] + ), + Transactions.type != TransactionType.UPGRADE_CONTRACT.value, ) .order_by(Transactions.queue_order) .all() @@ -2340,6 +2347,62 @@ def get_successor_snapshot_hashes( ).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( diff --git a/backend/protocol_rpc/endpoints.py b/backend/protocol_rpc/endpoints.py index 15f689921..ca4f6571a 100644 --- a/backend/protocol_rpc/endpoints.py +++ b/backend/protocol_rpc/endpoints.py @@ -2129,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/tests/db-sqlalchemy/test_transaction_finalized_state.py b/tests/db-sqlalchemy/test_transaction_finalized_state.py index 53a8de9b5..d5d8a356b 100644 --- a/tests/db-sqlalchemy/test_transaction_finalized_state.py +++ b/tests/db-sqlalchemy/test_transaction_finalized_state.py @@ -459,11 +459,28 @@ def upgrades(lifecycle, monkeypatch): return lifecycle -def queue_upgrade(lifecycle): +def queue_upgrade(lifecycle, *, legacy_finalized=False): with lifecycle.sessions() as session: - return admin_upgrade_contract_code(session, CONTRACT, UPGRADE_CODE)[ + 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): @@ -542,6 +559,36 @@ def insert_pending(lifecycle, nonce): 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)) @@ -602,6 +649,71 @@ async def test_upgrade_refunds_pending_paid_appeal_before_finalizing(upgrades): 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, @@ -763,3 +875,29 @@ async def test_upgrade_finalizes_undecided_predecessor_without_promotion( 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_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, From 5a53b7f617b1427236f9e7e50b1b40c2e8ba9e74 Mon Sep 17 00:00:00 2001 From: Daniel Rojas Date: Thu, 22 Oct 2026 12:40:17 -0500 Subject: [PATCH 6/6] fix(consensus): keep untyped legacy rows rewindable `type != UPGRADE_CONTRACT` evaluates to NULL for rows with no type, so the rewind query silently dropped them. Treat a NULL type as rewindable. Validation: unit 1731 passed (known env-only test_studio_fees failure), PostgreSQL suite 268 passed, 1 xfailed. Co-Authored-By: Claude --- backend/database_handler/transactions_processor.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/backend/database_handler/transactions_processor.py b/backend/database_handler/transactions_processor.py index 204905203..1f79083a6 100644 --- a/backend/database_handler/transactions_processor.py +++ b/backend/database_handler/transactions_processor.py @@ -2123,7 +2123,10 @@ def get_rewindable_newer_transactions(self, transaction_hash: str): Transactions.status.notin_( [TransactionStatus.FINALIZED, TransactionStatus.CANCELED] ), - Transactions.type != TransactionType.UPGRADE_CONTRACT.value, + or_( + Transactions.type.is_(None), + Transactions.type != TransactionType.UPGRADE_CONTRACT.value, + ), ) .order_by(Transactions.queue_order) .all()