diff --git a/node/src/aggregates/graph_jobs.rs b/node/src/aggregates/graph_jobs.rs index 46187387..d8ad2069 100644 --- a/node/src/aggregates/graph_jobs.rs +++ b/node/src/aggregates/graph_jobs.rs @@ -29,7 +29,7 @@ use super::graph_setup::{ use super::jobs::{ JobDeps, Plan, aggregate_uuid, author_peer, committee_pubkeys, decode_inbound, ensure_committee_sender, gate_closed, instance_status, local_member, local_operator, - plan_outcome, signing_nonce_id, + plan_outcome, retry_contract_revert, signing_nonce_id, }; use crate::clock::current_time_secs; use crate::engine_bridge::submit_event; @@ -124,15 +124,28 @@ impl JobRunner for GraphJobRunner { let action = ctx.job.action.clone(); let attempts = ctx.job.attempts; let pegout = PEGOUT_JOBS.contains(&action.as_str()); - match run_step(&self.deps, &action, graph_id, ¶ms, attempts).await { + let result = run_step(&self.deps, &action, graph_id, ¶ms, attempts).await; + if let Err(error) = &result + && let Some(retry) = retry_contract_revert(error, ctx.job.last_error.as_deref()) + { + warn!(event = "graph_job", graph_id = %graph_id, action = %action, error = %format!("{error:#}"), "contract call reverted; re-reading the chain before giving up"); + return retry; + } + match result { Ok(step) => step.into_outcome(), Err(error) => match classify_failure(&error) { FailureClass::Permanent(reason) if pegout => { - // A Goat revert on a peg-out call: retrying cannot help. + // A Goat revert that persisted, on a peg-out call. warn!(event = "graph_job", graph_id = %graph_id, action = %action, reason = %reason, "peg-out action rejected"); self.deps.metrics_state.record_instance_alert("pegout_action_rejected"); plan_outcome(&GraphPlan::noted(format!("rejected: {reason}"))) } + // Goat refused the graph: fail the job, which raises + // post_graph_data_rejected. + FailureClass::Permanent(reason) if action == JOB_POST_GRAPH_DATA => { + warn!(event = "graph_job", graph_id = %graph_id, action = %action, reason = %reason, "postGraphData rejected"); + JobOutcome::Failed(format!("rejected: {reason}")) + } FailureClass::Permanent(reason) => { warn!(event = "graph_job", graph_id = %graph_id, action = %action, reason = %reason, "message rejected"); self.deps.metrics_state.record_graph_validation(false); diff --git a/node/src/aggregates/instance_jobs.rs b/node/src/aggregates/instance_jobs.rs index 5771cdf4..6d47616a 100644 --- a/node/src/aggregates/instance_jobs.rs +++ b/node/src/aggregates/instance_jobs.rs @@ -16,7 +16,8 @@ use super::instance::{ }; use super::jobs::{ JobDeps, Plan, aggregate_uuid, committee_pubkeys, decode_inbound, ensure_committee_sender, - gate_closed, instance_status, local_member, local_operator, plan_outcome, signing_nonce_id, + gate_closed, instance_status, local_member, local_operator, plan_outcome, + retry_contract_revert, signing_nonce_id, }; use crate::clock::current_time_secs; use crate::env::{ @@ -971,7 +972,8 @@ pub enum PostOutcome { Rejected(String), } -/// The gateway reverts these for good; everything else is retried. +/// The client's own checks before the call; these are final. A revert from +/// the gateway is retried a few times first (`retry_contract_revert`). fn is_permanent_post_rejection(error: &anyhow::Error) -> bool { let message = format!("{error:#}"); ["not a pending pegin request", "pegin amount mismatch", "only committee member can call"] @@ -1067,6 +1069,15 @@ impl JobRunner for PostPeginDataRunner { }, Ok(PostOutcome::Rejected(reason)) => JobOutcome::Failed(reason), Err(error) => { + if let Some(retry) = retry_contract_revert(&error, ctx.job.last_error.as_deref()) { + warn!(event = "instance_post", instance_id = %instance_id, error = %format!("{error:#}"), "postPeginData reverted; re-reading the chain before giving up"); + return retry; + } + // A revert that persisted is the gateway's answer: fail the job, + // which raises PostPeginDataRejected. + if crate::error::is_contract_revert(&error) { + return JobOutcome::Failed(format!("{error:#}")); + } JobOutcome::Transient { error: format!("{error:#}"), retry_after_secs: None } } } diff --git a/node/src/aggregates/jobs.rs b/node/src/aggregates/jobs.rs index 6a9d1e87..0f9a0ae2 100644 --- a/node/src/aggregates/jobs.rs +++ b/node/src/aggregates/jobs.rs @@ -85,6 +85,36 @@ pub fn plan_outcome(plan: &impl Serialize) -> JobOutcome { } } +/// Consecutive runs a contract revert gets before it counts as a rejection. +/// A load-balanced or lagging Goat node can answer from an older state. Every +/// Goat call reads the chain state first, so a call that another relayer made +/// meanwhile ends the job on the next run. +pub(super) const CONTRACT_REVERT_RUNS: u32 = 5; +const CONTRACT_REVERT_RETRY_SECS: i64 = 60; +const CONTRACT_REVERT_PREFIX: &str = "contract revert "; + +/// The retry for a contract revert that has runs left, `None` for any other +/// error or once the runs are used up. The count lives in the job's last +/// error, so any other failure in between starts it again. +pub(super) fn retry_contract_revert( + error: &anyhow::Error, + last_error: Option<&str>, +) -> Option { + if !crate::error::is_contract_revert(error) { + return None; + } + let previous = last_error + .and_then(|last| last.strip_prefix(CONTRACT_REVERT_PREFIX)) + .and_then(|rest| rest.split_once(" of ")) + .and_then(|(run, _)| run.parse::().ok()) + .unwrap_or(0); + let run = previous + 1; + (run < CONTRACT_REVERT_RUNS).then(|| JobOutcome::Transient { + error: format!("{CONTRACT_REVERT_PREFIX}{run} of {CONTRACT_REVERT_RUNS}: {error:#}"), + retry_after_secs: Some(CONTRACT_REVERT_RETRY_SECS), + }) +} + /// The aggregate id of a job as a UUID; `what` names the aggregate in the /// failure ("a graph", "an instance"). pub(super) fn aggregate_uuid(ctx: &JobContext, what: &str) -> Result { @@ -171,3 +201,55 @@ pub(super) async fn ensure_committee_sender( let peer = author_peer(inbound)?; validate_committee(&deps.goat_client, &peer, inbound.instance_id, committee_pubkey).await } + +#[cfg(test)] +mod contract_revert_tests { + use super::*; + + fn revert() -> anyhow::Error { + // A Goat node's JSON-RPC answer to a reverting eth_call or estimateGas. + let payload = serde_json::from_value(serde_json::json!({ + "code": 3, + "message": "execution reverted: PeginNotPending", + })) + .unwrap(); + let error: alloy::transports::TransportError = + alloy::transports::RpcError::ErrorResp(payload); + anyhow::Error::new(error).context("postGraphData") + } + + fn retried(outcome: Option) -> Option { + match outcome { + Some(JobOutcome::Transient { error, retry_after_secs }) => { + assert_eq!(retry_after_secs, Some(CONTRACT_REVERT_RETRY_SECS)); + Some(error) + } + None => None, + Some(other) => panic!("unexpected outcome {other:?}"), + } + } + + #[test] + fn a_revert_is_retried_a_few_consecutive_runs_then_left_to_the_rejection_path() { + let mut last_error: Option = None; + for run in 1..CONTRACT_REVERT_RUNS { + let error = retried(retry_contract_revert(&revert(), last_error.as_deref())) + .unwrap_or_else(|| panic!("run {run} is retried")); + assert!(error.starts_with(&format!("contract revert {run} of {CONTRACT_REVERT_RUNS}"))); + assert!(error.contains("PeginNotPending")); + last_error = Some(error); + } + assert!(retry_contract_revert(&revert(), last_error.as_deref()).is_none()); + } + + #[test] + fn another_failure_in_between_starts_the_count_again() { + let error = retried(retry_contract_revert(&revert(), Some("Goat SPV at 10, block 12"))); + assert!(error.unwrap().starts_with("contract revert 1 of")); + } + + #[test] + fn other_errors_are_not_handled() { + assert!(retry_contract_revert(&anyhow::anyhow!("connection refused"), None).is_none()); + } +}