Skip to content

fix(consensus): appeal claim omits contract_snapshot/consensus_history, wiping contract state on successful validator appeal #1724

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
MuncleUscles merged 2 commits into from
Aug 13, 2026
Merged

fix(consensus): appeal claim omits contract_snapshot/consensus_history, wiping contract state on successful validator appeal #1724

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f8a1864
fix(consensus): return contract_snapshot and consensus_history from c…
rasca Aug 11, 2026
5490390
refactor(consensus): generate claim queries from a shared column mani…
MuncleUscles Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 66 additions & 13 deletions backend/consensus/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1197,18 +1197,66 @@ async def process_validator_appeal(
await self.rollback_transactions(context)

# Get the previous state of the contract
previous_contract_state = None
if context.transaction.contract_snapshot:
previous_contact_state = (
previous_contract_state = (
context.transaction.contract_snapshot.states["accepted"]
)
elif (
context.transaction.type == TransactionType.DEPLOY_CONTRACT
):
# Rolling back a deploy: clear the contract state
previous_contract_state = {}
else:
previous_contact_state = {}
# Defense in depth: the in-memory transaction may have
# been built without the stored contract_snapshot.
# Re-fetch it instead of clobbering the contract state
# with {} (which would wipe the code slot).
refetched = (
context.transactions_processor.get_transaction_by_hash(
context.transaction.hash
)
)
refetched_snapshot = ContractSnapshot.from_dict(
(refetched or {}).get("contract_snapshot")
)
if refetched_snapshot:
previous_contract_state = refetched_snapshot.states[
"accepted"
]
else:
from loguru import logger

logger.error(
f"Missing contract_snapshot for appealed "
f"transaction {context.transaction.hash}; "
f"skipping contract state restore"
)
# Surface to monitoring: the appeal succeeded but
# the contract kept the appealed transaction's
# state — recoverable, but needs operator eyes.
context.msg_handler.send_message(
LogEvent(
"consensus_event",
EventType.ERROR,
EventScope.CONSENSUS,
"Missing contract_snapshot on successful "
"validator appeal; contract state restore "
"skipped",
{
"transaction_hash": context.transaction.hash,
"contract_address": context.transaction.to_address,
},
transaction_hash=context.transaction.hash,
)
)

# Restore the contract state
context.contract_processor.update_contract_state(
context.transaction.to_address,
accepted_state=previous_contact_state,
)
if previous_contract_state is not None:
context.contract_processor.update_contract_state(
context.transaction.to_address,
accepted_state=previous_contract_state,
)

# Always clear snapshot on successful appeal (including timeout appeals)
# so re-execution loads fresh state from DB
Expand Down Expand Up @@ -1869,6 +1917,18 @@ class PendingState(TransactionState):
"""

async def handle(self, context):
# Refresh transaction from DB FIRST. The claim-path dict that built
# context.transaction omits columns whose Transaction.from_dict
# defaults are silently wrong for re-processed appeals (e.g.
# appeal_undetermined/appeal_leader_timeout default to False) — the
# same drift class as the PR #1724 state wipe. Nothing in this state
# may read claim-built appeal fields before this refresh.
context.transaction = Transaction.from_dict(
context.transactions_processor.get_transaction_by_hash(
context.transaction.hash
)
)

# Pre-effects: timestamp + reset rotation count
pre_effects = decide_pending_pre(
tx_hash=context.transaction.hash,
Expand All @@ -1877,13 +1937,6 @@ async def handle(self, context):
)
await EffectExecutor(context).execute(pre_effects)

# Refresh transaction from DB
context.transaction = Transaction.from_dict(
context.transactions_processor.get_transaction_by_hash(
context.transaction.hash
)
)

# Log executing message (unless appeal)
if (
not context.transaction.appeal_leader_timeout
Expand Down
193 changes: 89 additions & 104 deletions backend/consensus/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,82 @@
# stop-the-worker circuit breaker (see _transaction_context).
_TRANSIENT_LEADER_FATAL_CAUSES = frozenset({"NO_PROVIDER_FOR_PROMPT"})

# ---------------------------------------------------------------------------
# Claim column manifest
#
# Single source of truth for what a claim query RETURNs and what the rebuilt
# transaction dict contains. Both the RETURNING clause and the row->dict
# conversion are generated from these tuples, so the two can never drift
# apart. Drift is exactly what caused the contract-state wipe fixed in
# PR #1724: claim_next_appeal's hand-written RETURNING list omitted
# contract_snapshot/consensus_history and Transaction.from_dict silently
# defaulted them, so a successful validator appeal "restored" the contract
# state to {}.
#
# Entries are (sql_column, dict_key). dict_key differs from the column name
# only for triggered_by_hash, which Transaction.from_dict consumes as
# "triggered_by".
# ---------------------------------------------------------------------------
_TX_CLAIM_BASE_COLUMNS: tuple[tuple[str, str], ...] = (
("hash", "hash"),
("from_address", "from_address"),
("to_address", "to_address"),
("data", "data"),
("value", "value"),
("type", "type"),
("nonce", "nonce"),
("gaslimit", "gaslimit"),
("r", "r"),
("s", "s"),
("v", "v"),
("leader_only", "leader_only"),
("execution_mode", "execution_mode"),
("sim_config", "sim_config"),
("status", "status"),
("consensus_data", "consensus_data"),
("input_data", "input_data"),
("created_at", "created_at"),
("blocked_at", "blocked_at"),
("triggered_by_hash", "triggered_by"),
)

# Stored per-transaction state. Heavy: contract_snapshot has been observed at
# tens of MB, so only claims whose downstream consumes it include this group.
# Appeals restore the contract's accepted state from it; the pending and
# finalization paths rebuild snapshots via contract_snapshot_factory and
# persist consensus_history through DB-side jsonb merges, so they omit it
# deliberately (see _TX_CLAIM_OMISSIONS in the manifest tests).
_TX_STATE_COLUMNS: tuple[tuple[str, str], ...] = (
("contract_snapshot", "contract_snapshot"),
("consensus_history", "consensus_history"),
)

_TX_APPEAL_COLUMNS: tuple[tuple[str, str], ...] = (
("appealed", "appealed"),
("appeal_failed", "appeal_failed"),
("timestamp_appeal", "timestamp_appeal"),
("appeal_undetermined", "appeal_undetermined"),
("appeal_leader_timeout", "appeal_leader_timeout"),
("appeal_validators_timeout", "appeal_validators_timeout"),
)

_TX_FINALIZATION_COLUMNS: tuple[tuple[str, str], ...] = (
("timestamp_awaiting_finalization", "timestamp_awaiting_finalization"),
("appeal_failed", "appeal_failed"),
)


def _tx_returning_clause(*column_groups: tuple[tuple[str, str], ...]) -> str:
"""Render a RETURNING column list from manifest groups."""
columns = [col for group in column_groups for col, _ in group]
return ", ".join(f"transactions.{col}" for col in columns)


def _tx_row_to_dict(row: Any, *column_groups: tuple[tuple[str, str], ...]) -> dict:
"""Build the claimed-transaction dict from the same manifest groups that
generated the query's RETURNING clause."""
return {key: getattr(row, col) for group in column_groups for col, key in group}


class ConsensusWorker:
"""
Expand Down Expand Up @@ -223,7 +299,7 @@ async def claim_next_finalization(self, session: Session) -> Optional[dict]:
# They must be in ACCEPTED/UNDETERMINED/TIMEOUT states and appeal window must have passed
start_time = time.perf_counter()
query = text(
"""
f"""
WITH locked_finalizations AS (
SELECT t.*
FROM transactions t
Expand Down Expand Up @@ -285,13 +361,7 @@ async def claim_next_finalization(self, session: Session) -> Optional[dict]:
worker_id = :worker_id
FROM single_finalization
WHERE transactions.hash = single_finalization.hash
RETURNING transactions.hash, transactions.from_address, transactions.to_address,
transactions.data, transactions.value, transactions.type, transactions.nonce,
transactions.gaslimit, transactions.r, transactions.s, transactions.v,
transactions.leader_only, transactions.execution_mode, transactions.sim_config,
transactions.status, transactions.consensus_data,
transactions.input_data, transactions.created_at, transactions.timestamp_awaiting_finalization,
transactions.appeal_failed, transactions.blocked_at, transactions.triggered_by_hash;
RETURNING {_tx_returning_clause(_TX_CLAIM_BASE_COLUMNS, _TX_FINALIZATION_COLUMNS)};
"""
)

Expand All @@ -313,31 +383,9 @@ async def claim_next_finalization(self, session: Session) -> Optional[dict]:
f"[Worker {self.worker_id}] Claimed next finalization result {result.hash}"
)
session.commit()
# Convert result to dict
return {
"hash": result.hash,
"from_address": result.from_address,
"to_address": result.to_address,
"data": result.data,
"value": result.value,
"type": result.type,
"nonce": result.nonce,
"gaslimit": result.gaslimit,
"r": result.r,
"s": result.s,
"v": result.v,
"leader_only": result.leader_only,
"execution_mode": result.execution_mode,
"sim_config": result.sim_config,
"status": result.status,
"consensus_data": result.consensus_data,
"input_data": result.input_data,
"created_at": result.created_at,
"timestamp_awaiting_finalization": result.timestamp_awaiting_finalization,
"appeal_failed": result.appeal_failed,
"blocked_at": result.blocked_at,
"triggered_by": result.triggered_by_hash,
}
return _tx_row_to_dict(
result, _TX_CLAIM_BASE_COLUMNS, _TX_FINALIZATION_COLUMNS
)

return None

Expand All @@ -352,7 +400,7 @@ async def claim_next_appeal(self, session: Session) -> Optional[dict]:
# Query to atomically claim an appealed transaction
start_time = time.perf_counter()
query = text(
"""
f"""
WITH locked_appeals AS (
SELECT t.hash, t.to_address, t.created_at
FROM transactions t
Expand Down Expand Up @@ -393,16 +441,7 @@ async def claim_next_appeal(self, session: Session) -> Optional[dict]:
worker_id = :worker_id
FROM single_appeal
WHERE transactions.hash = single_appeal.hash
RETURNING transactions.hash, transactions.from_address, transactions.to_address,
transactions.data, transactions.value, transactions.type, transactions.nonce,
transactions.gaslimit, transactions.r, transactions.s, transactions.v,
transactions.leader_only, transactions.execution_mode, transactions.sim_config,
transactions.status, transactions.consensus_data,
transactions.input_data, transactions.created_at, transactions.appealed,
transactions.appeal_failed, transactions.timestamp_appeal,
transactions.appeal_undetermined, transactions.appeal_leader_timeout,
transactions.appeal_validators_timeout, transactions.blocked_at,
transactions.triggered_by_hash;
RETURNING {_tx_returning_clause(_TX_CLAIM_BASE_COLUMNS, _TX_STATE_COLUMNS, _TX_APPEAL_COLUMNS)};
"""
)

Expand All @@ -418,35 +457,9 @@ async def claim_next_appeal(self, session: Session) -> Optional[dict]:

if result:
session.commit()
# Convert result to dict
return {
"hash": result.hash,
"from_address": result.from_address,
"to_address": result.to_address,
"data": result.data,
"value": result.value,
"type": result.type,
"nonce": result.nonce,
"gaslimit": result.gaslimit,
"r": result.r,
"s": result.s,
"v": result.v,
"leader_only": result.leader_only,
"execution_mode": result.execution_mode,
"sim_config": result.sim_config,
"status": result.status,
"consensus_data": result.consensus_data,
"input_data": result.input_data,
"created_at": result.created_at,
"appealed": result.appealed,
"appeal_failed": result.appeal_failed,
"timestamp_appeal": result.timestamp_appeal,
"appeal_undetermined": result.appeal_undetermined,
"appeal_leader_timeout": result.appeal_leader_timeout,
"appeal_validators_timeout": result.appeal_validators_timeout,
"blocked_at": result.blocked_at,
"triggered_by": result.triggered_by_hash,
}
return _tx_row_to_dict(
result, _TX_CLAIM_BASE_COLUMNS, _TX_STATE_COLUMNS, _TX_APPEAL_COLUMNS
)

return None

Expand All @@ -462,7 +475,7 @@ async def claim_next_transaction(self, session: Session) -> Optional[dict]:
# Ensures only one transaction per contract is processed at a time
start_time = time.perf_counter()
query = text(
"""
f"""
WITH candidate_transactions AS (
SELECT t.hash, t.to_address, t.type, t.created_at, t.recovery_count
FROM transactions t
Expand Down Expand Up @@ -542,13 +555,7 @@ async def claim_next_transaction(self, session: Session) -> Optional[dict]:
worker_id = :worker_id
FROM single_transaction
WHERE transactions.hash = single_transaction.hash
RETURNING transactions.hash, transactions.from_address, transactions.to_address,
transactions.data, transactions.value, transactions.type, transactions.nonce,
transactions.gaslimit, transactions.r, transactions.s, transactions.v,
transactions.leader_only, transactions.execution_mode, transactions.sim_config,
transactions.status, transactions.consensus_data,
transactions.input_data, transactions.created_at, transactions.blocked_at,
transactions.triggered_by_hash;
RETURNING {_tx_returning_clause(_TX_CLAIM_BASE_COLUMNS)};
"""
)

Expand All @@ -567,29 +574,7 @@ async def claim_next_transaction(self, session: Session) -> Optional[dict]:
if result:
logger.debug(f"[Worker {self.worker_id}] Claimed transaction {result.hash}")
session.commit()
# Convert result to dict
return {
"hash": result.hash,
"from_address": result.from_address,
"to_address": result.to_address,
"data": result.data,
"value": result.value,
"type": result.type,
"nonce": result.nonce,
"gaslimit": result.gaslimit,
"r": result.r,
"s": result.s,
"v": result.v,
"leader_only": result.leader_only,
"execution_mode": result.execution_mode,
"sim_config": result.sim_config,
"status": result.status,
"consensus_data": result.consensus_data,
"input_data": result.input_data,
"created_at": result.created_at,
"blocked_at": result.blocked_at,
"triggered_by": result.triggered_by_hash,
}
return _tx_row_to_dict(result, _TX_CLAIM_BASE_COLUMNS)

return None

Expand Down
Loading
Loading