From 95b00f4f41de4ab3c4628e20826f7517b484f3eb Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Wed, 26 Aug 2026 22:15:16 +0200 Subject: [PATCH 01/11] fix(deps): bump h2 to 0.4.19 RUSTSEC-2026-0258: an h2 peer can hold a connection open with unbounded empty DATA frames. Reached transitively through every HTTP client here. Pinned to the version alone: letting cargo re-resolve moved a dozen unrelated windows-sys selections that h2 does not need. --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2ef5244..a866448 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6761,9 +6761,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.14" +version = "0.4.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "171fefbc92fe4a4de27e0698d6a5b392d6a0e333506bc49133760b3bcf948733" +checksum = "ef8e5e5a340588f4452631496976cf8636d4a7ecf600239fdc27615d2530bc16" dependencies = [ "atomic-waker", "bytes", From 1c490c9c7e3ac79efad871cf2f4a2ca67a2e27b5 Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Wed, 26 Aug 2026 22:16:14 +0200 Subject: [PATCH 02/11] fix(broadcast): retry an unavailable endpoint safely An endpoint can briefly answer `assertions are unavailable` while a credible layer's assertion state catches up with the chain, which failed the command. PCL now fills and signs once, then resubmits those exact bytes: same hash and nonce, so a node already holding the transaction deduplicates the retry instead of accepting a second one. An "already known" answer counts as submitted, since a send whose response was lost would otherwise fail spuriously. Gas estimation is retried too, before anything is signed; assertion rejections are reverts and stay terminal. Exhausting the retries says whether anything is in flight: before signing the nonce is untouched and the command can be re-run, after signing the error carries the hash to check first. --- CHANGELOG.md | 4 + Cargo.lock | 17 +- crates/pcl/core/Cargo.toml | 4 +- crates/pcl/core/src/api/broadcast.rs | 43 +- crates/pcl/core/src/api/error.rs | 76 ++- crates/pcl/core/src/api/tests.rs | 56 +++ crates/pcl/core/src/onchain.rs | 680 +++++++++++++++++++++++++-- 7 files changed, 793 insertions(+), 87 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e1858f6..19ea49e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All notable user-facing changes should be recorded here. ## Unreleased +### Fixed + +- Broadcasting no longer fails on a brief `assertions are unavailable` refusal. The transaction is signed once and resubmitted byte for byte, so a retry is deduplicated by hash rather than becoming a second transaction on the same nonce. If the retries run out, `onchain.assertions_unavailable` means nothing was submitted, while `onchain.tx_submission_unconfirmed` carries the signed hash to check first. + ### Added - `pcl deploy` warns when the assertions it is about to release use the V2 spec but the target does not support it (the `app.phylax.systems` platform, or a Linea chain). The warning names the files and the V2 triggers/precompiles found in them, prints before the protocol-manager step and again at the end, and appears in `--json` output as `data.warnings`. It never blocks a deploy. diff --git a/Cargo.lock b/Cargo.lock index a866448..58ef3ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8697,6 +8697,17 @@ dependencies = [ "objc2-encode", ] +[[package]] +name = "objc2-app-kit" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d49e936b501e5c5bf01fda3a9452ff86dc3ea98ad5f283e1455153142d97518c" +dependencies = [ + "bitflags 2.11.1", + "objc2", + "objc2-foundation", +] + [[package]] name = "objc2-core-foundation" version = "0.3.2" @@ -15114,15 +15125,15 @@ dependencies = [ [[package]] name = "webbrowser" -version = "1.2.1" +version = "1.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fc95580916af1e68ff6a7be07446fc5db73ebf71cf092de939bbf5f7e189f72" +checksum = "ef62a3d5f7b2411119a11b6f62570dbff91d7105e011a20fb83fbf8f5761c40f" dependencies = [ - "core-foundation 0.10.1", "jni", "log", "ndk-context", "objc2", + "objc2-app-kit", "objc2-foundation", "url", "web-sys", diff --git a/crates/pcl/core/Cargo.toml b/crates/pcl/core/Cargo.toml index 9920c04..6f38de1 100644 --- a/crates/pcl/core/Cargo.toml +++ b/crates/pcl/core/Cargo.toml @@ -62,6 +62,6 @@ credible = ["dep:assertion-executor", "dep:assertion-verification"] [dev-dependencies] mockito = "1.2" rand = "0.8" -# `test-util` lets the config-lock timeout be asserted on a paused clock instead -# of waiting out the real 30-second budget. +# `test-util` lets the config-lock timeout and the submission backoff be +# asserted on a paused clock instead of waiting out real time. tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/pcl/core/src/api/broadcast.rs b/crates/pcl/core/src/api/broadcast.rs index 2b77a78..93031e4 100644 --- a/crates/pcl/core/src/api/broadcast.rs +++ b/crates/pcl/core/src/api/broadcast.rs @@ -36,6 +36,7 @@ use alloy_primitives::{ Address, Bytes, }; +use alloy_rpc_types_eth::TransactionRequest; use alloy_signer_local::PrivateKeySigner; use colored::Colorize; use pcl_common::args::CliArgs; @@ -240,24 +241,28 @@ fn progress(cli_args: &CliArgs, message: &str) { } } -/// `platform_required` is the confirmation requirement stated by the platform -/// in the calldata response, when it carries one; it overrides the static -/// per-chain fallback (the platform's chain profiles differ per deployment). async fn send_tx( config: &CliConfig, + cli_args: &CliArgs, tx_args: &TxArgs, signer: PrivateKeySigner, chain_id: u64, - to: Address, - data: Bytes, + request: TransactionRequest, + // The platform's requirement when the calldata response carries one; it + // overrides the static per-chain fallback. platform_required: Option, ) -> Result { let rpc = tx_args.resolve_rpc(config, chain_id)?; let confirmations = tx_args.resolve_confirmations(config, chain_id, platform_required)?; let sender = TxSender::connect(rpc, signer, chain_id).await?; - Ok(sender - .send_and_confirm(to, data, confirmations, tx_args.timeout()) - .await?) + // Boxed to keep the fill/submit/confirm state machine off every caller's stack. + Ok(Box::pin(sender.send_and_confirm( + request, + confirmations, + tx_args.timeout(), + &|message: &str| progress(cli_args, message), + )) + .await?) } fn tx_value(outcome: &TxOutcome) -> Value { @@ -561,11 +566,13 @@ impl ApiArgs { progress(cli_args, "Broadcasting StateOracle.batch transaction"); let outcome = send_tx( config, + cli_args, &args.broadcast.tx, signer, calldata.chain_id, - calldata.state_oracle_address, - encode_batch(calldata.calldata.clone()), + TransactionRequest::default() + .to(calldata.state_oracle_address) + .input(encode_batch(calldata.calldata.clone()).into()), calldata.required_confirmations, ) .await?; @@ -660,11 +667,13 @@ impl ApiArgs { progress(cli_args, "Broadcasting removal transaction"); let outcome = send_tx( config, + cli_args, &args.broadcast.tx, signer, chain_id, - calldata.to, - calldata.data.clone(), + TransactionRequest::default() + .to(calldata.to) + .input(calldata.data.clone().into()), calldata.required_confirmations, ) .await?; @@ -913,11 +922,11 @@ impl ApiArgs { progress(cli_args, "Broadcasting manager transfer transaction"); let outcome = send_tx( config, + cli_args, &args.broadcast.tx, signer, chain_id, - to, - data, + TransactionRequest::default().to(to).input(data.into()), calldata.required_confirmations, ) .await?; @@ -985,11 +994,13 @@ impl ApiArgs { let signer_address = signer.address(); let outcome = send_tx( config, + cli_args, &args.broadcast.tx, signer, calldata.chain_id, - calldata.to, - calldata.calldata.clone(), + TransactionRequest::default() + .to(calldata.to) + .input(calldata.calldata.clone().into()), calldata.required_confirmations, ) .await?; diff --git a/crates/pcl/core/src/api/error.rs b/crates/pcl/core/src/api/error.rs index 3da67d5..c09ac98 100644 --- a/crates/pcl/core/src/api/error.rs +++ b/crates/pcl/core/src/api/error.rs @@ -2,6 +2,7 @@ use crate::{ error::AuthError, output::with_envelope_metadata, }; +use alloy_primitives::B256; use serde_json::{ Map, Value, @@ -132,6 +133,22 @@ pub enum ApiCommandError { } impl ApiCommandError { + // A signed transaction whose outcome is unobserved: the hash exists, the + // result does not, so a retry could double-broadcast. + fn ambiguous_submission(&self) -> Option<(&B256, u64)> { + match self { + Self::Onchain( + crate::onchain::OnchainError::ConfirmationUnknown { + tx_hash, chain_id, .. + } + | crate::onchain::OnchainError::SubmissionUnconfirmed { + tx_hash, chain_id, .. + }, + ) => Some((tx_hash, *chain_id)), + _ => None, + } + } + pub fn code(&self) -> &'static str { match self { Self::NoAuthToken => "auth.no_token", @@ -187,6 +204,12 @@ impl ApiCommandError { crate::onchain::OnchainError::ConfirmationUnknown { .. } => { "onchain.tx_submitted_confirmation_unknown" } + crate::onchain::OnchainError::AssertionsUnavailable { .. } => { + "onchain.assertions_unavailable" + } + crate::onchain::OnchainError::SubmissionUnconfirmed { .. } => { + "onchain.tx_submission_unconfirmed" + } _ => "onchain.failed", } } @@ -201,6 +224,18 @@ impl ApiCommandError { } pub fn next_actions(&self) -> Vec { + // The transaction is signed and may be in flight; re-running now could + // broadcast a duplicate while the first is still pending. + if let Some((tx_hash, chain_id)) = self.ambiguous_submission() { + return vec![ + format!( + "Check transaction {tx_hash} on chain {chain_id} first (e.g. cast receipt {tx_hash} --rpc-url , or a block explorer)" + ), + "If it confirmed, re-run the command — landed state reconciles without a new transaction; if it is still pending, wait — do not re-broadcast" + .to_string(), + ]; + } + match self { Self::NoAuthToken | Self::ExpiredAuthToken(_) | Self::AuthRefresh(_) => { vec![ @@ -376,19 +411,13 @@ impl ApiCommandError { "Or pass --rpc-url / set PCL_RPC_URL".to_string(), ] } - Self::Onchain(crate::onchain::OnchainError::ConfirmationUnknown { - tx_hash, - chain_id, - .. - }) => { - // The transaction was accepted by the RPC; only the receipt - // poll failed. Re-running immediately could broadcast a - // duplicate mutation while the first is still pending. + Self::Onchain(crate::onchain::OnchainError::AssertionsUnavailable { .. }) => { + // Nothing was signed or submitted, so the same command is the + // recovery: the refusal is transient by construction. vec![ - format!( - "Check transaction {tx_hash} on chain {chain_id} first (e.g. cast receipt {tx_hash} --rpc-url , or a block explorer)" - ), - "If it confirmed, re-run the command — landed state reconciles without a new transaction; if it is still pending, wait — do not re-broadcast" + "Re-run the same command; nothing was submitted and the account nonce is unchanged" + .to_string(), + "If it keeps failing, check the Credible RPC endpoint's assertion alignment" .to_string(), ] } @@ -466,13 +495,16 @@ impl ApiCommandError { } Self::HttpStatus { .. } => vec!["inspect_response_body", "retry"], Self::Wallet(_) => vec!["fix_wallet", "retry"], - Self::Onchain(crate::onchain::OnchainError::ConfirmationUnknown { .. }) => { + _ if self.ambiguous_submission().is_some() => { vec![ "check_tx_status", "wait_for_confirmation", "reconcile_mutation", ] } + Self::Onchain(crate::onchain::OnchainError::AssertionsUnavailable { .. }) => { + vec!["retry_later", "retry"] + } Self::Onchain(_) => vec!["check_chain_config", "retry"], Self::BroadcastCancelled => vec!["rerun_with_yes"], Self::ConfirmAfterTx { .. } => vec!["confirm_platform_only", "reconcile_mutation"], @@ -547,16 +579,7 @@ impl ApiCommandError { error.insert("source".to_string(), source.json_envelope()); } - // A transaction that was submitted but whose receipt could not be - // observed (timeout, transport failure while polling) is an ambiguous - // mutation: the hash exists, the outcome does not — a retry could - // double-broadcast. - if let Self::Onchain(crate::onchain::OnchainError::ConfirmationUnknown { - tx_hash, - chain_id, - .. - }) = self - { + if let Some((tx_hash, chain_id)) = self.ambiguous_submission() { error.insert("tx_hash".to_string(), json!(tx_hash)); error.insert("chain_id".to_string(), json!(chain_id)); error.insert( @@ -595,8 +618,11 @@ impl ApiCommandError { ); } - if let Self::ConfirmAfterTx { tx_hash, .. } - | Self::Onchain(crate::onchain::OnchainError::ConfirmationUnknown { tx_hash, .. }) = self + let tx_hash = match self { + Self::ConfirmAfterTx { tx_hash, .. } => Some(tx_hash), + _ => self.ambiguous_submission().map(|(tx_hash, _)| tx_hash), + }; + if let Some(tx_hash) = tx_hash && let Some(object) = envelope.as_object_mut() { object.insert("tx_hash".to_string(), json!(tx_hash)); diff --git a/crates/pcl/core/src/api/tests.rs b/crates/pcl/core/src/api/tests.rs index 872f6f2..9a87cfa 100644 --- a/crates/pcl/core/src/api/tests.rs +++ b/crates/pcl/core/src/api/tests.rs @@ -1254,6 +1254,62 @@ fn confirmation_unknown_envelope_marks_the_mutation_ambiguous() { ); } +#[test] +fn an_unconfirmed_submission_is_ambiguous_but_a_refused_one_is_not() { + let tx_hash = alloy_primitives::B256::repeat_byte(0xef); + let unconfirmed = + ApiCommandError::Onchain(crate::onchain::OnchainError::SubmissionUnconfirmed { + tx_hash, + chain_id: 84532, + redacted_url: "https://rpc.example.com".to_string(), + message: "credible layer: assertions are unavailable".to_string(), + attempts: 6, + }) + .json_envelope(); + + // The transaction may already be upstream, so the envelope must carry the + // hash and refuse to advertise a retry. + assert_eq!( + unconfirmed["error"]["code"], + "onchain.tx_submission_unconfirmed" + ); + assert_eq!(unconfirmed["tx_hash"], serde_json::json!(tx_hash)); + assert_eq!( + unconfirmed["error"]["mutation"]["onchain_landed"], + "unknown" + ); + let actions = unconfirmed["next_actions"].as_array().unwrap(); + assert!( + actions + .iter() + .any(|action| action.as_str().unwrap().contains(&tx_hash.to_string())), + "{unconfirmed}" + ); + + let refused = ApiCommandError::Onchain(crate::onchain::OnchainError::AssertionsUnavailable { + chain_id: 84532, + redacted_url: "https://rpc.example.com".to_string(), + attempts: 6, + waited_ms: 5750, + message: "credible layer: assertions are unavailable".to_string(), + }) + .json_envelope(); + + // Nothing was signed, so there is no hash to reconcile and re-running is + // the recovery rather than a risk. + assert_eq!(refused["error"]["code"], "onchain.assertions_unavailable"); + assert_eq!(refused["error"]["recoverable"], true); + assert!(refused["tx_hash"].is_null(), "{refused}"); + assert!(refused["error"]["mutation"].is_null(), "{refused}"); + assert!( + refused["next_actions"][0] + .as_str() + .unwrap() + .contains("Re-run the same command"), + "{refused}" + ); +} + #[tokio::test] async fn authenticated_workflow_refuses_a_foreign_platform_with_a_valid_token() { let mut server = mockito::Server::new_async().await; diff --git a/crates/pcl/core/src/onchain.rs b/crates/pcl/core/src/onchain.rs index f918b0a..fd150d9 100644 --- a/crates/pcl/core/src/onchain.rs +++ b/crates/pcl/core/src/onchain.rs @@ -3,21 +3,43 @@ //! The backend computes all calldata (see the `*-calldata` API endpoints); //! this module only signs and submits it — either as `StateOracle.batch(bytes[])` //! or as a raw `{to, data}` transaction — and waits for confirmations. +//! +//! Submission signs once and resubmits those exact bytes when an endpoint says +//! it is unavailable, so a node deduplicates the retry rather than accepting a +//! second transaction. use crate::config::CliConfig; -use alloy_network::EthereumWallet; +use alloy_network::{ + Ethereum, + EthereumWallet, + eip2718::Encodable2718, +}; use alloy_primitives::{ - Address, B256, Bytes, }; use alloy_provider::{ - DynProvider, + PendingTransactionBuilder, PendingTransactionError, Provider, ProviderBuilder, + RootProvider, + SendableTx, + fillers::{ + FillProvider, + JoinFill, + WalletFiller, + }, + transport::{ + RpcError, + TransportError, + }, + utils::JoinedRecommendedFillers, +}; +use alloy_rpc_types_eth::{ + TransactionReceipt, + TransactionRequest, }; -use alloy_rpc_types_eth::TransactionRequest; use alloy_signer_local::PrivateKeySigner; use alloy_sol_types::{ SolCall, @@ -26,8 +48,15 @@ use alloy_sol_types::{ use serde::Serialize; use std::time::Duration; use thiserror::Error; +use tokio::time::Instant; use url::Url; +// Attempts a credible-layer-gated step gets before it is treated as terminal. +// The alignment window is 0.2-1.2s in practice; the backoff below spans ~6s. +const ALIGNMENT_ATTEMPTS: u32 = 6; +const ALIGNMENT_FIRST_DELAY: Duration = Duration::from_millis(250); +const ALIGNMENT_MAX_DELAY: Duration = Duration::from_secs(2); + sol! { /// `StateOracle`'s batch entrypoint; the only function pcl encodes locally. function batch(bytes[] calldata data) external; @@ -109,6 +138,28 @@ pub enum OnchainError { message: String, }, + #[error( + "{redacted_url} could not judge the transaction after {attempts} attempt(s) over {waited_ms}ms ({message}). Nothing was signed or submitted and the nonce is unchanged: re-run the command." + )] + AssertionsUnavailable { + chain_id: u64, + redacted_url: String, + attempts: u32, + waited_ms: u128, + message: String, + }, + + #[error( + "Transaction {tx_hash} was submitted to chain {chain_id} {attempts} time(s) without confirmed acceptance ({message}). It may be in flight: check the hash before re-running, which would reuse the same nonce." + )] + SubmissionUnconfirmed { + tx_hash: B256, + chain_id: u64, + redacted_url: String, + attempts: u32, + message: String, + }, + #[error("Transaction {tx_hash} reverted on-chain (block {block})")] Reverted { tx_hash: B256, block: u64 }, } @@ -210,9 +261,22 @@ fn scrub_rpc_error(text: &str, rpc: &Url) -> String { .replace(rpc.as_str(), &redacted) } +// Concrete rather than erased, so `fill` can build and sign before submitting. +type WalletProvider = FillProvider< + JoinFill>, + RootProvider, +>; + +// Built once, so the hash is known before the first submission and identical on +// every retry. +struct PreparedTx { + tx_hash: B256, + raw: Vec, +} + /// A connected, chain-checked transaction sender. pub struct TxSender { - provider: DynProvider, + provider: WalletProvider, chain_id: u64, rpc: Url, } @@ -242,7 +306,7 @@ impl TxSender { }); } Ok(Self { - provider: provider.erased(), + provider, chain_id: expected_chain_id, rpc, }) @@ -258,43 +322,17 @@ impl TxSender { /// provider (transactions are strictly sequential in pcl). pub async fn send_and_confirm( &self, - to: Address, - data: Bytes, + request: TransactionRequest, confirmations: u64, timeout: Duration, + notify: &dyn Fn(&str), ) -> Result { - let request = TransactionRequest::default().to(to).input(data.into()); - - let pending = self - .provider - .send_transaction(request) - .await - .map_err(|error| { - OnchainError::Send { - redacted_url: crate::config::redacted_rpc_host(self.rpc.as_str()), - message: scrub_rpc_error(&error.to_string(), &self.rpc), - } - })?; - let tx_hash = *pending.tx_hash(); - - // A failure while waiting for the receipt does NOT mean the - // transaction failed — it was already accepted by the RPC. Surface - // that ambiguity explicitly (submitted, confirmation unknown) and - // scrub the provider error, which can echo the credential-bearing - // RPC URL. - let receipt = pending - .with_required_confirmations(confirmations) - .with_timeout(Some(timeout)) - .get_receipt() - .await - .map_err(|source: PendingTransactionError| { - OnchainError::ConfirmationUnknown { - tx_hash, - chain_id: self.chain_id, - redacted_url: crate::config::redacted_rpc_host(self.rpc.as_str()), - message: scrub_rpc_error(&source.to_string(), &self.rpc), - } - })?; + let prepared = self.prepare(request, notify).await?; + self.submit(&prepared, notify).await?; + notify("Transaction submitted; waiting for confirmation"); + let receipt = self + .await_receipt(prepared.tx_hash, confirmations, timeout) + .await?; if !receipt.status() { return Err(OnchainError::Reverted { @@ -312,13 +350,209 @@ impl TxSender { confirmations_waited: confirmations, }) } + + // `eth_estimateGas` is gated too, so filling hits the same window; retrying + // costs nothing while nothing is signed. + async fn prepare( + &self, + request: TransactionRequest, + notify: &dyn Fn(&str), + ) -> Result { + let started = Instant::now(); + let attempts = ALIGNMENT_ATTEMPTS; + let mut backoff = Backoff::new(); + let mut message = String::new(); + for attempt in 1..=attempts { + match self.provider.fill(request.clone()).await { + Ok(filled) => return self.signed(filled), + Err(error) if assertions_unavailable(&error) => { + message = self.scrub(&error); + backoff + .wait( + attempt, + attempts, + "Credible layer is realigning while preparing the transaction", + notify, + ) + .await; + } + Err(error) => { + return Err(OnchainError::Send { + redacted_url: self.redacted(), + message: self.scrub(&error), + }); + } + } + } + Err(OnchainError::AssertionsUnavailable { + chain_id: self.chain_id, + redacted_url: self.redacted(), + attempts, + waited_ms: started.elapsed().as_millis(), + message, + }) + } + + fn signed(&self, filled: SendableTx) -> Result { + let envelope = filled.try_into_envelope().map_err(|unsigned| { + OnchainError::Send { + redacted_url: self.redacted(), + message: format!("wallet did not sign the filled transaction: {unsigned}"), + } + })?; + Ok(PreparedTx { + tx_hash: *envelope.tx_hash(), + raw: envelope.encoded_2718(), + }) + } + + // Every attempt sends the same hash and nonce, so a node already holding the + // transaction deduplicates the retry instead of accepting a second one. + async fn submit( + &self, + prepared: &PreparedTx, + notify: &dyn Fn(&str), + ) -> Result<(), OnchainError> { + let attempts = ALIGNMENT_ATTEMPTS; + let mut backoff = Backoff::new(); + let mut message = String::new(); + for attempt in 1..=attempts { + match self.provider.send_raw_transaction(&prepared.raw).await { + Ok(_) => return Ok(()), + // The node already holds these exact bytes, so an earlier + // attempt reached it even if its answer did not reach us. + Err(error) if already_submitted(&error) => return Ok(()), + Err(error) if assertions_unavailable(&error) => { + message = self.scrub(&error); + backoff + .wait( + attempt, + attempts, + "Credible layer is realigning while submitting the transaction", + notify, + ) + .await; + } + Err(error) => { + return Err(OnchainError::Send { + redacted_url: self.redacted(), + message: self.scrub(&error), + }); + } + } + } + Err(OnchainError::SubmissionUnconfirmed { + tx_hash: prepared.tx_hash, + chain_id: self.chain_id, + redacted_url: self.redacted(), + attempts, + message, + }) + } + + // Waits for `tx_hash` to reach `confirmations`. + async fn await_receipt( + &self, + tx_hash: B256, + confirmations: u64, + timeout: Duration, + ) -> Result { + PendingTransactionBuilder::new(self.provider.root().clone(), tx_hash) + .with_required_confirmations(confirmations) + .with_timeout(Some(timeout)) + .get_receipt() + .await + .map_err(|source: PendingTransactionError| { + OnchainError::ConfirmationUnknown { + tx_hash, + chain_id: self.chain_id, + redacted_url: self.redacted(), + message: scrub_rpc_error(&source.to_string(), &self.rpc), + } + }) + } + + fn redacted(&self) -> String { + crate::config::redacted_rpc_host(self.rpc.as_str()) + } + + fn scrub(&self, error: &TransportError) -> String { + scrub_rpc_error(&error.to_string(), &self.rpc) + } +} + +/// Doubling delay between attempts at a credible-layer-gated step. +struct Backoff { + delay: Duration, +} + +impl Backoff { + fn new() -> Self { + Self { + delay: ALIGNMENT_FIRST_DELAY, + } + } + + /// The final attempt has nothing left to wait for. + async fn wait(&mut self, attempt: u32, attempts: u32, status: &str, notify: &dyn Fn(&str)) { + if attempt >= attempts { + return; + } + notify(&format!( + "{status}; retrying in {}ms ({attempt}/{attempts})", + self.delay.as_millis() + )); + tokio::time::sleep(self.delay).await; + self.delay = (self.delay * 2).min(ALIGNMENT_MAX_DELAY); + } +} + +/// Whether the credible layer refused to judge the call because it has no +/// assertion state aligned with the block it would judge against — transient by +/// construction. +/// +/// Matched on the message, not the code: the refusal reuses the generic +/// internal-error code, which would sweep in unrelated failures. An assertion +/// rejection is a revert (code 3) and is never transient, so it cannot match. +fn assertions_unavailable(error: &TransportError) -> bool { + const REVERT: i64 = 3; + let RpcError::ErrorResp(payload) = error else { + return false; + }; + let message = payload.message.to_ascii_lowercase(); + payload.code != REVERT && message.contains("credible layer") && message.contains("unavailable") +} + +/// Whether the node is telling us it already holds these exact bytes. reth and +/// geth both answer a resubmission this way, which is the success case when a +/// refusal was raised after the transaction had already been forwarded. +fn already_submitted(error: &TransportError) -> bool { + let RpcError::ErrorResp(payload) = error else { + return false; + }; + let message = payload.message.to_ascii_lowercase(); + ["already known", "already imported", "already exists"] + .iter() + .any(|known| message.contains(known)) } #[cfg(test)] mod tests { use super::*; use crate::config::RpcEndpoint; - use alloy_primitives::hex; + use alloy_primitives::{ + address, + hex, + }; + use mockito::{ + Matcher, + Mock, + ServerGuard, + }; + use serde_json::{ + Value, + json, + }; fn config_with_rpc(chain_id: u64, url: &str, confirmations: Option) -> CliConfig { let mut config = CliConfig::default(); @@ -556,4 +790,368 @@ mod tests { assert!(rendered.contains(&B256::repeat_byte(0xab).to_string())); assert!(rendered.contains("Do not re-broadcast")); } + + #[tokio::test] + async fn a_receipt_that_never_arrives_stops_at_the_configured_timeout() { + let mut server = mockito::Server::new_async().await; + ok(&mut server, "eth_chainId", json!(format!("{CHAIN_ID:#x}"))).await; + ok(&mut server, "eth_getTransactionReceipt", Value::Null).await; + ok(&mut server, "eth_blockNumber", json!("0x1")).await; + let sender = TxSender::connect(server.url().parse().unwrap(), signer(), CHAIN_ID) + .await + .unwrap(); + let started = Instant::now(); + let timeout = Duration::from_millis(50); + + let error = sender + .await_receipt(RECEIPT_HASH, 1, timeout) + .await + .unwrap_err(); + + assert!(matches!(error, OnchainError::ConfirmationUnknown { .. })); + assert!(started.elapsed() >= timeout); + assert!(started.elapsed() < Duration::from_secs(1)); + } + + const CHAIN_ID: u64 = 31337; + /// Verbatim from the Credible RPC gate (`src/simulation/error.rs`). + const UNAVAILABLE: &str = "credible layer: assertions are unavailable, try again shortly"; + const INTERNAL: i64 = -32603; + const MINED_BLOCK: u64 = 0x10; + const RECEIPT_HASH: B256 = B256::repeat_byte(0xcd); + + fn signer() -> PrivateKeySigner { + PrivateKeySigner::from_bytes(&B256::repeat_byte(0x11)).unwrap() + } + + /// Mocks are matched in creation order, and one that has met its `expect` + /// count is skipped, so two mocks for the same method answer in sequence. + async fn rpc(server: &mut ServerGuard, method: &str, response: Value) -> Mock { + server + .mock("POST", "/") + .match_body(Matcher::PartialJson(json!({ "method": method }))) + .with_status(200) + .with_header("content-type", "application/json") + .with_body(response.to_string()) + .create_async() + .await + } + + async fn ok(server: &mut ServerGuard, method: &str, result: Value) -> Mock { + rpc( + server, + method, + json!({ "jsonrpc": "2.0", "id": 1, "result": result }), + ) + .await + } + + async fn fails(server: &mut ServerGuard, method: &str, code: i64, message: &str) -> Mock { + rpc( + server, + method, + json!({ + "jsonrpc": "2.0", + "id": 1, + "error": { "code": code, "message": message }, + }), + ) + .await + } + + /// The ungated reads a fill and a receipt wait perform. + async fn chain(server: &mut ServerGuard) { + ok(server, "eth_chainId", json!("0x7a69")).await; + ok( + server, + "eth_blockNumber", + json!(format!("{MINED_BLOCK:#x}")), + ) + .await; + ok( + server, + "eth_feeHistory", + json!({ + "oldestBlock": "0xe", + "baseFeePerGas": ["0x3b9aca00", "0x3b9aca00", "0x3b9aca00"], + "gasUsedRatio": [0.5, 0.5], + "reward": [["0x1"], ["0x1"]], + }), + ) + .await; + ok(server, "eth_getTransactionReceipt", receipt()).await; + } + + fn receipt() -> Value { + json!({ + "type": "0x2", + "status": "0x1", + "cumulativeGasUsed": "0x5208", + "logs": [], + "logsBloom": format!("0x{}", "0".repeat(512)), + "transactionHash": RECEIPT_HASH, + "transactionIndex": "0x0", + "blockHash": B256::repeat_byte(0xbb), + "blockNumber": format!("{MINED_BLOCK:#x}"), + "gasUsed": "0x5208", + "effectiveGasPrice": "0x3b9aca00", + "from": signer().address(), + "to": address!("0202020202020202020202020202020202020202"), + "contractAddress": Value::Null, + }) + } + + async fn send(server: &ServerGuard) -> Result { + let sender = TxSender::connect(server.url().parse().unwrap(), signer(), CHAIN_ID).await?; + sender + .send_and_confirm( + TransactionRequest::default() + .to(address!("0202020202020202020202020202020202020202")) + .input(Bytes::from_static(&[0x12, 0x34]).into()), + 1, + Duration::from_secs(2), + &|message: &str| { + if message == "Transaction submitted; waiting for confirmation" { + tokio::time::resume(); + } + }, + ) + .await + } + + #[tokio::test(start_paused = true)] + async fn a_refused_submission_resends_the_same_signed_transaction() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + // Exactly one fill: a resubmission must reuse the signed envelope, not + // rebuild it, or it could take a second nonce. + let nonce = ok(&mut server, "eth_getTransactionCount", json!("0x8")) + .await + .expect(1); + let gas = ok(&mut server, "eth_estimateGas", json!("0x5208")) + .await + .expect(1); + let refused = fails(&mut server, "eth_sendRawTransaction", INTERNAL, UNAVAILABLE) + .await + .expect(1); + let accepted = ok(&mut server, "eth_sendRawTransaction", json!(RECEIPT_HASH)) + .await + .expect(1); + let outcome = send(&server).await.unwrap(); + + assert_eq!(outcome.tx_hash, RECEIPT_HASH); + assert_eq!(outcome.block_number, Some(MINED_BLOCK)); + nonce.assert_async().await; + gas.assert_async().await; + refused.assert_async().await; + accepted.assert_async().await; + } + + #[tokio::test(start_paused = true)] + async fn a_deduplicated_resend_counts_as_submitted() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + ok(&mut server, "eth_estimateGas", json!("0x5208")).await; + fails(&mut server, "eth_sendRawTransaction", INTERNAL, UNAVAILABLE) + .await + .expect(1); + ok(&mut server, "eth_getTransactionByHash", Value::Null).await; + // The resend arrives after the pool took the first copy: same bytes, + // same hash, so the node deduplicates instead of queueing a second one. + let dedup = fails( + &mut server, + "eth_sendRawTransaction", + -32000, + "already known", + ) + .await + .expect(1); + + let outcome = send(&server).await.unwrap(); + + assert_eq!(outcome.tx_hash, RECEIPT_HASH); + dedup.assert_async().await; + } + + #[tokio::test(start_paused = true)] + async fn refused_gas_estimation_is_retried_before_anything_is_signed() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + let refused = fails(&mut server, "eth_estimateGas", INTERNAL, UNAVAILABLE) + .await + .expect(1); + let estimated = ok(&mut server, "eth_estimateGas", json!("0x5208")) + .await + .expect(1); + let submitted = ok(&mut server, "eth_sendRawTransaction", json!(RECEIPT_HASH)) + .await + .expect(1); + + assert_eq!(send(&server).await.unwrap().tx_hash, RECEIPT_HASH); + refused.assert_async().await; + estimated.assert_async().await; + submitted.assert_async().await; + } + + #[tokio::test(start_paused = true)] + async fn a_lasting_refusal_before_signing_reports_that_nothing_was_submitted() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + fails(&mut server, "eth_estimateGas", INTERNAL, UNAVAILABLE).await; + let never = ok(&mut server, "eth_sendRawTransaction", json!(RECEIPT_HASH)) + .await + .expect(0); + + let error = send(&server).await.unwrap_err(); + + assert!(matches!( + error, + OnchainError::AssertionsUnavailable { + chain_id: CHAIN_ID, + attempts: ALIGNMENT_ATTEMPTS, + .. + } + )); + assert!( + error + .to_string() + .contains("Nothing was signed or submitted") + ); + never.assert_async().await; + } + + #[tokio::test(start_paused = true)] + async fn a_lasting_refusal_after_signing_reports_the_signed_hash() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + ok(&mut server, "eth_estimateGas", json!("0x5208")).await; + fails(&mut server, "eth_sendRawTransaction", INTERNAL, UNAVAILABLE).await; + ok(&mut server, "eth_getTransactionByHash", Value::Null).await; + + let error = send(&server).await.unwrap_err(); + + // The hash is what makes this recoverable: the transaction may be + // upstream, so the operator needs it before deciding anything. + let OnchainError::SubmissionUnconfirmed { tx_hash, .. } = &error else { + panic!("expected an unconfirmed submission, got {error:?}"); + }; + assert!(error.to_string().contains(&tx_hash.to_string())); + assert!(error.to_string().contains("may be in flight")); + } + + #[tokio::test(start_paused = true)] + async fn an_assertion_rejection_is_never_retried() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + ok(&mut server, "eth_estimateGas", json!("0x5208")).await; + // A rejection is a verdict about the transaction, not about the node: + // it names the credible layer but resending can only fail again. + let rejected = fails( + &mut server, + "eth_sendRawTransaction", + 3, + "execution reverted: credible layer: transaction rejected by an assertion", + ) + .await + .expect(1); + + let error = send(&server).await.unwrap_err(); + + assert!(matches!(error, OnchainError::Send { .. })); + rejected.assert_async().await; + } + + #[tokio::test(start_paused = true)] + async fn an_ordinary_rejection_is_not_retried() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + ok(&mut server, "eth_estimateGas", json!("0x5208")).await; + let rejected = fails( + &mut server, + "eth_sendRawTransaction", + -32000, + "insufficient funds for gas * price + value", + ) + .await + .expect(1); + // No hash probe either: nothing was signed into flight. + let probe = ok(&mut server, "eth_getTransactionByHash", Value::Null) + .await + .expect(0); + + assert!(matches!( + send(&server).await.unwrap_err(), + OnchainError::Send { .. } + )); + rejected.assert_async().await; + probe.assert_async().await; + } + + #[tokio::test(start_paused = true)] + async fn a_stale_nonce_on_the_first_attempt_is_not_retried() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + ok(&mut server, "eth_estimateGas", json!("0x5208")).await; + // Another transaction took the nonce before this one was submitted, so + // this envelope can never land; only a resubmission could have spent + // its own nonce. + let rejected = fails( + &mut server, + "eth_sendRawTransaction", + -32000, + "nonce too low", + ) + .await + .expect(1); + + assert!(matches!( + send(&server).await.unwrap_err(), + OnchainError::Send { .. } + )); + rejected.assert_async().await; + } + + #[tokio::test(start_paused = true)] + async fn a_reverted_transaction_still_reports_its_receipt() { + let mut server = mockito::Server::new_async().await; + ok(&mut server, "eth_chainId", json!("0x7a69")).await; + ok( + &mut server, + "eth_blockNumber", + json!(format!("{MINED_BLOCK:#x}")), + ) + .await; + ok( + &mut server, + "eth_feeHistory", + json!({ + "oldestBlock": "0xe", + "baseFeePerGas": ["0x3b9aca00", "0x3b9aca00", "0x3b9aca00"], + "gasUsedRatio": [0.5, 0.5], + "reward": [["0x1"], ["0x1"]], + }), + ) + .await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + ok(&mut server, "eth_estimateGas", json!("0x5208")).await; + ok(&mut server, "eth_sendRawTransaction", json!(RECEIPT_HASH)).await; + let mut reverted = receipt(); + reverted["status"] = json!("0x0"); + ok(&mut server, "eth_getTransactionReceipt", reverted).await; + + assert!(matches!( + send(&server).await.unwrap_err(), + OnchainError::Reverted { + tx_hash: RECEIPT_HASH, + block: MINED_BLOCK, + } + )); + } } From 1ba64828ba8e5db53db83631bb53d440e0dfb067 Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Wed, 26 Aug 2026 22:16:14 +0200 Subject: [PATCH 03/11] feat(broadcast): scale the retry to the endpoint An `assertions are unavailable` refusal is worth retrying wherever it comes from, but only a Credible RPC has an alignment window long enough to wait out. `--with-credible-rpc` says which kind of endpoint is being broadcast to: set, a refusal gets the full window (6 attempts over ~6s per step); unset, it gets a short retry (3 attempts) and the wording drops the credible-layer reference. The two are indistinguishable from the URL alone, so the kind is stated rather than probed. --- CHANGELOG.md | 2 +- README.md | 4 ++ crates/pcl/core/src/api/broadcast.rs | 2 +- crates/pcl/core/src/onchain.rs | 89 ++++++++++++++++++++++------ 4 files changed, 78 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 19ea49e..2f48e47 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,7 @@ All notable user-facing changes should be recorded here. ### Fixed -- Broadcasting no longer fails on a brief `assertions are unavailable` refusal. The transaction is signed once and resubmitted byte for byte, so a retry is deduplicated by hash rather than becoming a second transaction on the same nonce. If the retries run out, `onchain.assertions_unavailable` means nothing was submitted, while `onchain.tx_submission_unconfirmed` carries the signed hash to check first. +- Broadcasting no longer fails on a brief `assertions are unavailable` refusal. The transaction is signed once and resubmitted byte for byte, so a retry is deduplicated by hash rather than becoming a second transaction on the same nonce. `--with-credible-rpc` waits out a Credible RPC's full alignment window; without it an unavailable endpoint still gets a short retry. If the retries run out, `onchain.assertions_unavailable` means nothing was submitted, while `onchain.tx_submission_unconfirmed` carries the signed hash to check first. ### Added diff --git a/README.md b/README.md index 473e820..1e817b2 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,10 @@ checks, broadcast `StateOracle.batch`, and confirm — use `pcl deploy`: # one-time: store an RPC endpoint for the chain pcl config set-rpc [--confirmations N] +# a Credible RPC endpoint can refuse briefly while its assertion state catches +# up with the chain; this waits that window out instead of failing +pcl deploy --with-credible-rpc + pcl deploy --dry-run pcl deploy --private-key $PCL_PRIVATE_KEY # or --account pcl deploy --project-name my-protocol --chain-id --private-key ... --yes --json diff --git a/crates/pcl/core/src/api/broadcast.rs b/crates/pcl/core/src/api/broadcast.rs index 93031e4..3aa0545 100644 --- a/crates/pcl/core/src/api/broadcast.rs +++ b/crates/pcl/core/src/api/broadcast.rs @@ -254,7 +254,7 @@ async fn send_tx( ) -> Result { let rpc = tx_args.resolve_rpc(config, chain_id)?; let confirmations = tx_args.resolve_confirmations(config, chain_id, platform_required)?; - let sender = TxSender::connect(rpc, signer, chain_id).await?; + let sender = TxSender::connect(rpc, signer, chain_id, tx_args.with_credible_rpc).await?; // Boxed to keep the fill/submit/confirm state machine off every caller's stack. Ok(Box::pin(sender.send_and_confirm( request, diff --git a/crates/pcl/core/src/onchain.rs b/crates/pcl/core/src/onchain.rs index fd150d9..54bee38 100644 --- a/crates/pcl/core/src/onchain.rs +++ b/crates/pcl/core/src/onchain.rs @@ -6,7 +6,7 @@ //! //! Submission signs once and resubmits those exact bytes when an endpoint says //! it is unavailable, so a node deduplicates the retry rather than accepting a -//! second transaction. +//! second transaction. How long it waits: [`TxArgs::with_credible_rpc`]. use crate::config::CliConfig; use alloy_network::{ @@ -54,6 +54,8 @@ use url::Url; // Attempts a credible-layer-gated step gets before it is treated as terminal. // The alignment window is 0.2-1.2s in practice; the backoff below spans ~6s. const ALIGNMENT_ATTEMPTS: u32 = 6; +// A plain endpoint has no alignment window to wait out, only a blip. +const PLAIN_ATTEMPTS: u32 = 3; const ALIGNMENT_FIRST_DELAY: Duration = Duration::from_millis(250); const ALIGNMENT_MAX_DELAY: Duration = Duration::from_secs(2); @@ -178,6 +180,12 @@ pub struct TxArgs { /// Seconds to wait for the transaction to confirm before giving up #[arg(long, default_value_t = 300)] pub tx_timeout_secs: u64, + + /// Broadcast through a Credible RPC endpoint, which can refuse briefly + /// while its assertion state catches up with the chain. Set, an unavailable + /// endpoint is retried through that window; unset, only past a blip. + #[arg(long = "with-credible-rpc", env = "PCL_WITH_CREDIBLE_RPC")] + pub with_credible_rpc: bool, } impl TxArgs { @@ -279,6 +287,7 @@ pub struct TxSender { provider: WalletProvider, chain_id: u64, rpc: Url, + credible: bool, } impl TxSender { @@ -288,6 +297,7 @@ impl TxSender { rpc: Url, signer: PrivateKeySigner, expected_chain_id: u64, + credible: bool, ) -> Result { let provider = ProviderBuilder::new() .wallet(EthereumWallet::from(signer)) @@ -309,6 +319,7 @@ impl TxSender { provider, chain_id: expected_chain_id, rpc, + credible, }) } @@ -359,7 +370,7 @@ impl TxSender { notify: &dyn Fn(&str), ) -> Result { let started = Instant::now(); - let attempts = ALIGNMENT_ATTEMPTS; + let attempts = self.attempts(); let mut backoff = Backoff::new(); let mut message = String::new(); for attempt in 1..=attempts { @@ -368,12 +379,7 @@ impl TxSender { Err(error) if assertions_unavailable(&error) => { message = self.scrub(&error); backoff - .wait( - attempt, - attempts, - "Credible layer is realigning while preparing the transaction", - notify, - ) + .wait(attempt, attempts, &self.waiting_for("preparing"), notify) .await; } Err(error) => { @@ -413,7 +419,7 @@ impl TxSender { prepared: &PreparedTx, notify: &dyn Fn(&str), ) -> Result<(), OnchainError> { - let attempts = ALIGNMENT_ATTEMPTS; + let attempts = self.attempts(); let mut backoff = Backoff::new(); let mut message = String::new(); for attempt in 1..=attempts { @@ -425,12 +431,7 @@ impl TxSender { Err(error) if assertions_unavailable(&error) => { message = self.scrub(&error); backoff - .wait( - attempt, - attempts, - "Credible layer is realigning while submitting the transaction", - notify, - ) + .wait(attempt, attempts, &self.waiting_for("submitting"), notify) .await; } Err(error) => { @@ -472,6 +473,25 @@ impl TxSender { }) } + // A Credible RPC has an alignment window to wait out; a plain endpoint only + // has to get past a blip. + fn attempts(&self) -> u32 { + if self.credible { + ALIGNMENT_ATTEMPTS + } else { + PLAIN_ATTEMPTS + } + } + + // What the wait is for, in the terms of the endpoint being waited on. + fn waiting_for(&self, step: &str) -> String { + if self.credible { + format!("Credible layer is realigning while {step} the transaction") + } else { + format!("Endpoint is unavailable while {step} the transaction") + } + } + fn redacted(&self) -> String { crate::config::redacted_rpc_host(self.rpc.as_str()) } @@ -797,7 +817,7 @@ mod tests { ok(&mut server, "eth_chainId", json!(format!("{CHAIN_ID:#x}"))).await; ok(&mut server, "eth_getTransactionReceipt", Value::Null).await; ok(&mut server, "eth_blockNumber", json!("0x1")).await; - let sender = TxSender::connect(server.url().parse().unwrap(), signer(), CHAIN_ID) + let sender = TxSender::connect(server.url().parse().unwrap(), signer(), CHAIN_ID, true) .await .unwrap(); let started = Instant::now(); @@ -902,7 +922,8 @@ mod tests { } async fn send(server: &ServerGuard) -> Result { - let sender = TxSender::connect(server.url().parse().unwrap(), signer(), CHAIN_ID).await?; + let sender = + TxSender::connect(server.url().parse().unwrap(), signer(), CHAIN_ID, true).await?; sender .send_and_confirm( TransactionRequest::default() @@ -947,6 +968,40 @@ mod tests { accepted.assert_async().await; } + // An unavailable endpoint is worth a few attempts either way, but only a + // Credible RPC has an alignment window long enough to justify waiting it out. + #[tokio::test(start_paused = true)] + async fn a_plain_endpoint_stops_short_of_the_alignment_window() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + ok(&mut server, "eth_estimateGas", json!("0x5208")).await; + let refused = fails(&mut server, "eth_sendRawTransaction", INTERNAL, UNAVAILABLE) + .await + .expect(PLAIN_ATTEMPTS as usize); + + let sender = TxSender::connect(server.url().parse().unwrap(), signer(), CHAIN_ID, false) + .await + .unwrap(); + let error = sender + .send_and_confirm( + TransactionRequest::default() + .to(address!("0202020202020202020202020202020202020202")) + .input(Bytes::from_static(&[0x12, 0x34]).into()), + 1, + Duration::from_secs(2), + &|_: &str| {}, + ) + .await + .unwrap_err(); + + assert!( + matches!(error, OnchainError::SubmissionUnconfirmed { attempts, .. } if attempts == PLAIN_ATTEMPTS), + "{error:?}" + ); + refused.assert_async().await; + } + #[tokio::test(start_paused = true)] async fn a_deduplicated_resend_counts_as_submitted() { let mut server = mockito::Server::new_async().await; From b91a9373f8e2c973e7cea04fd18a733cbcf288bb Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Wed, 26 Aug 2026 22:27:10 +0200 Subject: [PATCH 04/11] docs(broadcast): say what the backoff delays are chosen for One line each for the attempt budgets and the two delays: what the value is for, not how the loop uses it. --- crates/pcl/core/src/onchain.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/pcl/core/src/onchain.rs b/crates/pcl/core/src/onchain.rs index 54bee38..b6b0d17 100644 --- a/crates/pcl/core/src/onchain.rs +++ b/crates/pcl/core/src/onchain.rs @@ -56,7 +56,10 @@ use url::Url; const ALIGNMENT_ATTEMPTS: u32 = 6; // A plain endpoint has no alignment window to wait out, only a blip. const PLAIN_ATTEMPTS: u32 = 3; +// First delay, doubling from here: short enough that the common sub-second +// window costs one wait. const ALIGNMENT_FIRST_DELAY: Duration = Duration::from_millis(250); +// Cap, so the last attempts stay useful instead of sleeping the budget away. const ALIGNMENT_MAX_DELAY: Duration = Duration::from_secs(2); sol! { From 2ec35ccdbd1bcfeaf4bac767708bae977b1ffced Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Wed, 26 Aug 2026 22:28:49 +0200 Subject: [PATCH 05/11] refactor(broadcast): carry the signed transaction as alloy produced it `PreparedTx` cached the hash and the encoded bytes; both come off the envelope the wallet returns, so the envelope is the one thing to carry. --- crates/pcl/core/src/api/broadcast.rs | 1 - crates/pcl/core/src/onchain.rs | 53 ++++++++++++---------------- 2 files changed, 23 insertions(+), 31 deletions(-) diff --git a/crates/pcl/core/src/api/broadcast.rs b/crates/pcl/core/src/api/broadcast.rs index 3aa0545..22de043 100644 --- a/crates/pcl/core/src/api/broadcast.rs +++ b/crates/pcl/core/src/api/broadcast.rs @@ -255,7 +255,6 @@ async fn send_tx( let rpc = tx_args.resolve_rpc(config, chain_id)?; let confirmations = tx_args.resolve_confirmations(config, chain_id, platform_required)?; let sender = TxSender::connect(rpc, signer, chain_id, tx_args.with_credible_rpc).await?; - // Boxed to keep the fill/submit/confirm state machine off every caller's stack. Ok(Box::pin(sender.send_and_confirm( request, confirmations, diff --git a/crates/pcl/core/src/onchain.rs b/crates/pcl/core/src/onchain.rs index b6b0d17..4acc0b3 100644 --- a/crates/pcl/core/src/onchain.rs +++ b/crates/pcl/core/src/onchain.rs @@ -12,6 +12,7 @@ use crate::config::CliConfig; use alloy_network::{ Ethereum, EthereumWallet, + Network, eip2718::Encodable2718, }; use alloy_primitives::{ @@ -62,6 +63,15 @@ const ALIGNMENT_FIRST_DELAY: Duration = Duration::from_millis(250); // Cap, so the last attempts stay useful instead of sleeping the budget away. const ALIGNMENT_MAX_DELAY: Duration = Duration::from_secs(2); +// The signed transaction the wallet produced: hash and raw bytes both come off it. +type SignedTx = ::TxEnvelope; + +// Concrete rather than erased, so `fill` can build and sign before submitting. +type WalletProvider = FillProvider< + JoinFill>, + RootProvider, +>; + sol! { /// `StateOracle`'s batch entrypoint; the only function pcl encodes locally. function batch(bytes[] calldata data) external; @@ -272,19 +282,6 @@ fn scrub_rpc_error(text: &str, rpc: &Url) -> String { .replace(rpc.as_str(), &redacted) } -// Concrete rather than erased, so `fill` can build and sign before submitting. -type WalletProvider = FillProvider< - JoinFill>, - RootProvider, ->; - -// Built once, so the hash is known before the first submission and identical on -// every retry. -struct PreparedTx { - tx_hash: B256, - raw: Vec, -} - /// A connected, chain-checked transaction sender. pub struct TxSender { provider: WalletProvider, @@ -341,11 +338,11 @@ impl TxSender { timeout: Duration, notify: &dyn Fn(&str), ) -> Result { - let prepared = self.prepare(request, notify).await?; - self.submit(&prepared, notify).await?; + let signed = self.prepare(request, notify).await?; + self.submit(&signed, notify).await?; notify("Transaction submitted; waiting for confirmation"); let receipt = self - .await_receipt(prepared.tx_hash, confirmations, timeout) + .await_receipt(*signed.tx_hash(), confirmations, timeout) .await?; if !receipt.status() { @@ -371,7 +368,7 @@ impl TxSender { &self, request: TransactionRequest, notify: &dyn Fn(&str), - ) -> Result { + ) -> Result { let started = Instant::now(); let attempts = self.attempts(); let mut backoff = Backoff::new(); @@ -402,31 +399,27 @@ impl TxSender { }) } - fn signed(&self, filled: SendableTx) -> Result { - let envelope = filled.try_into_envelope().map_err(|unsigned| { + fn signed(&self, filled: SendableTx) -> Result { + filled.try_into_envelope().map_err(|unsigned| { OnchainError::Send { redacted_url: self.redacted(), message: format!("wallet did not sign the filled transaction: {unsigned}"), } - })?; - Ok(PreparedTx { - tx_hash: *envelope.tx_hash(), - raw: envelope.encoded_2718(), }) } // Every attempt sends the same hash and nonce, so a node already holding the // transaction deduplicates the retry instead of accepting a second one. - async fn submit( - &self, - prepared: &PreparedTx, - notify: &dyn Fn(&str), - ) -> Result<(), OnchainError> { + async fn submit(&self, signed: &SignedTx, notify: &dyn Fn(&str)) -> Result<(), OnchainError> { let attempts = self.attempts(); let mut backoff = Backoff::new(); let mut message = String::new(); for attempt in 1..=attempts { - match self.provider.send_raw_transaction(&prepared.raw).await { + match self + .provider + .send_raw_transaction(&signed.encoded_2718()) + .await + { Ok(_) => return Ok(()), // The node already holds these exact bytes, so an earlier // attempt reached it even if its answer did not reach us. @@ -446,7 +439,7 @@ impl TxSender { } } Err(OnchainError::SubmissionUnconfirmed { - tx_hash: prepared.tx_hash, + tx_hash: *signed.tx_hash(), chain_id: self.chain_id, redacted_url: self.redacted(), attempts, From 3e845ff532f6c5e2025962e886a55972b1155d92 Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Wed, 26 Aug 2026 22:32:31 +0200 Subject: [PATCH 06/11] fix(broadcast): keep the nonce across retried gas estimation The recommended fillers use a caching nonce manager, which increments its cached value on every fill after the first. A gas estimate refused inside the alignment window therefore signed the retry with the next nonce, and the transaction sat pending behind the gap it had skipped. The nonce is now resolved once and set on the request, which the filler treats as finished and never looks up. pcl broadcasts sequentially, so one reading covers the whole window. --- crates/pcl/core/src/onchain.rs | 63 ++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/crates/pcl/core/src/onchain.rs b/crates/pcl/core/src/onchain.rs index 4acc0b3..1bd3070 100644 --- a/crates/pcl/core/src/onchain.rs +++ b/crates/pcl/core/src/onchain.rs @@ -26,6 +26,7 @@ use alloy_provider::{ ProviderBuilder, RootProvider, SendableTx, + WalletProvider as _, fillers::{ FillProvider, JoinFill, @@ -362,6 +363,27 @@ impl TxSender { }) } + // Resolves the nonce once and sets it, so the filler skips its own lookup. + // Left to the filler, every retried fill would increment its cached nonce + // and the signed transaction would skip one, stranding it as pending. + async fn with_pinned_nonce( + &self, + request: TransactionRequest, + ) -> Result { + let from = self.provider.default_signer_address(); + let nonce = self + .provider + .get_transaction_count(from) + .await + .map_err(|error| { + OnchainError::Send { + redacted_url: self.redacted(), + message: self.scrub(&error), + } + })?; + Ok(request.from(from).nonce(nonce)) + } + // `eth_estimateGas` is gated too, so filling hits the same window; retrying // costs nothing while nothing is signed. async fn prepare( @@ -369,6 +391,7 @@ impl TxSender { request: TransactionRequest, notify: &dyn Fn(&str), ) -> Result { + let request = self.with_pinned_nonce(request).await?; let started = Instant::now(); let attempts = self.attempts(); let mut backoff = Backoff::new(); @@ -936,6 +959,46 @@ mod tests { .await } + // A retried fill must not advance the nonce: the signed transaction would + // skip one and sit pending behind the gap until something else filled it. + #[tokio::test(start_paused = true)] + async fn a_retried_estimate_keeps_the_first_nonce() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + let count = ok(&mut server, "eth_getTransactionCount", json!("0x8")) + .await + .expect(1); + let refused = fails(&mut server, "eth_estimateGas", INTERNAL, UNAVAILABLE) + .await + .expect(1); + let estimated = ok(&mut server, "eth_estimateGas", json!("0x5208")) + .await + .expect(1); + // In the signed envelope the chain id is followed by the nonce, so + // `827a69` then `08` is nonce 8; a nonce advanced by the retry reads + // `827a6909`. + let sent = server + .mock("POST", "/") + .match_body(Matcher::AllOf(vec![ + Matcher::Regex("eth_sendRawTransaction".into()), + Matcher::Regex("827a6908".into()), + ])) + .with_header("content-type", "application/json") + .with_body(json!({"jsonrpc": "2.0", "id": 1, "result": RECEIPT_HASH}).to_string()) + .create_async() + .await + .expect(1); + + send(&server) + .await + .expect("the retried estimate keeps nonce 8"); + + count.assert_async().await; + refused.assert_async().await; + estimated.assert_async().await; + sent.assert_async().await; + } + #[tokio::test(start_paused = true)] async fn a_refused_submission_resends_the_same_signed_transaction() { let mut server = mockito::Server::new_async().await; From d374205a737f981c4b2e4414a351c3963e5899c2 Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Thu, 27 Aug 2026 11:18:53 +0200 Subject: [PATCH 07/11] refactor(onchain): use transaction domain aliases --- crates/pcl/core/src/api/error.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/crates/pcl/core/src/api/error.rs b/crates/pcl/core/src/api/error.rs index c09ac98..734d970 100644 --- a/crates/pcl/core/src/api/error.rs +++ b/crates/pcl/core/src/api/error.rs @@ -2,7 +2,10 @@ use crate::{ error::AuthError, output::with_envelope_metadata, }; -use alloy_primitives::B256; +use alloy_primitives::{ + ChainId, + TxHash, +}; use serde_json::{ Map, Value, @@ -135,7 +138,7 @@ pub enum ApiCommandError { impl ApiCommandError { // A signed transaction whose outcome is unobserved: the hash exists, the // result does not, so a retry could double-broadcast. - fn ambiguous_submission(&self) -> Option<(&B256, u64)> { + fn ambiguous_submission(&self) -> Option<(&TxHash, ChainId)> { match self { Self::Onchain( crate::onchain::OnchainError::ConfirmationUnknown { From c4c9af39397955d55f99ef2baebe517f7ef50649 Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Thu, 27 Aug 2026 11:38:50 +0200 Subject: [PATCH 08/11] fix(onchain): use pending nonce for broadcasts --- crates/pcl/core/src/onchain.rs | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/crates/pcl/core/src/onchain.rs b/crates/pcl/core/src/onchain.rs index 1bd3070..04ded54 100644 --- a/crates/pcl/core/src/onchain.rs +++ b/crates/pcl/core/src/onchain.rs @@ -374,6 +374,7 @@ impl TxSender { let nonce = self .provider .get_transaction_count(from) + .pending() .await .map_err(|error| { OnchainError::Send { @@ -959,6 +960,35 @@ mod tests { .await } + #[tokio::test] + async fn nonce_lookup_uses_the_pending_block_tag() { + let mut server = mockito::Server::new_async().await; + ok(&mut server, "eth_chainId", json!(format!("{CHAIN_ID:#x}"))).await; + let count = server + .mock("POST", "/") + .match_body(Matcher::PartialJson(json!({ + "method": "eth_getTransactionCount", + "params": [signer().address(), "pending"], + }))) + .with_status(200) + .with_header("content-type", "application/json") + .with_body(json!({ "jsonrpc": "2.0", "id": 1, "result": "0x8" }).to_string()) + .create_async() + .await + .expect(1); + let sender = TxSender::connect(server.url().parse().unwrap(), signer(), CHAIN_ID, true) + .await + .unwrap(); + + let request = sender + .with_pinned_nonce(TransactionRequest::default()) + .await + .unwrap(); + + assert_eq!(request.nonce, Some(8)); + count.assert_async().await; + } + // A retried fill must not advance the nonce: the signed transaction would // skip one and sit pending behind the gap until something else filled it. #[tokio::test(start_paused = true)] From fb851ba7c476346ff38474434e981ff4337a6ac1 Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Thu, 27 Aug 2026 11:53:08 +0200 Subject: [PATCH 09/11] fix(onchain): preserve ambiguous submission errors --- crates/pcl/core/src/error.rs | 154 +++++++++++++++++++++++++++++++++ crates/pcl/core/src/onchain.rs | 137 ++++++++++------------------- 2 files changed, 200 insertions(+), 91 deletions(-) diff --git a/crates/pcl/core/src/error.rs b/crates/pcl/core/src/error.rs index 6385453..2e0b563 100644 --- a/crates/pcl/core/src/error.rs +++ b/crates/pcl/core/src/error.rs @@ -4,6 +4,15 @@ use crate::{ abi::ConstructorAbiError, credible_config::CredibleConfigError, }; +use alloy_primitives::{ + BlockNumber, + ChainId, + TxHash, +}; +use alloy_provider::transport::{ + RpcError, + TransportError, +}; use chrono::{ DateTime, Utc, @@ -15,6 +24,151 @@ use dapp_api_client::generated::client::{ use pcl_phoundry::error::PhoundryError; use thiserror::Error; +/// Errors that can occur while broadcasting transactions. +/// +/// RPC endpoints in these messages are redacted to their scheme/host/port +/// origin: stored provider URLs commonly embed API keys, and error envelopes +/// end up in logs and transcripts. +#[derive(Error, Debug)] +pub enum OnchainError { + #[error( + "No RPC endpoint for chain {chain_id}. Pass --rpc-url (or set PCL_RPC_URL), or store one with `pcl config set-rpc {chain_id} `." + )] + RpcUrlMissing { chain_id: ChainId }, + + #[error("Invalid RPC URL ({redacted_url}) configured for chain {chain_id}: {reason}")] + InvalidRpcUrl { + chain_id: ChainId, + redacted_url: String, + reason: String, + }, + + #[error( + "{requested} confirmation(s) is below the {required} required on chain {chain_id}; the platform would reject the confirmation with INSUFFICIENT_CONFIRMATIONS. Raise --confirmations (or the stored per-chain value) to at least {required}." + )] + InsufficientConfirmations { + chain_id: ChainId, + requested: u64, + required: u64, + }, + + #[error( + "RPC endpoint {redacted_url} serves chain {actual}, but this transaction targets chain {expected}. Refusing to broadcast." + )] + ChainIdMismatch { + redacted_url: String, + expected: ChainId, + actual: ChainId, + }, + + #[error("RPC transport error communicating with {redacted_url}: {message}")] + Transport { + redacted_url: String, + message: String, + }, + + #[error("Failed to send transaction via {redacted_url}: {message}")] + Send { + redacted_url: String, + message: String, + }, + + #[error( + "Transaction {tx_hash} was submitted to chain {chain_id} but its confirmation state is unknown ({message}). Do not re-broadcast: the transaction may still confirm. Check it by hash first, and re-run the command only once its status is known." + )] + ConfirmationUnknown { + tx_hash: TxHash, + chain_id: ChainId, + redacted_url: String, + message: String, + }, + + #[error( + "{redacted_url} could not judge the transaction after {attempts} attempt(s) over {waited_ms}ms ({message}). Nothing was signed or submitted and the nonce is unchanged: re-run the command." + )] + AssertionsUnavailable { + chain_id: ChainId, + redacted_url: String, + attempts: u32, + waited_ms: u128, + message: String, + }, + + #[error( + "Transaction {tx_hash} was submitted to chain {chain_id} {attempts} time(s) without confirmed acceptance ({message}). It may be in flight: check the hash before re-running, which would reuse the same nonce." + )] + SubmissionUnconfirmed { + tx_hash: TxHash, + chain_id: ChainId, + redacted_url: String, + attempts: u32, + message: String, + }, + + #[error("Transaction {tx_hash} reverted on-chain (block {block})")] + Reverted { tx_hash: TxHash, block: BlockNumber }, +} + +// Context required to classify a failed send without losing the signed hash. +pub(crate) struct SubmissionAttemptError { + // Transaction hash. + tx_hash: TxHash, + // Chain identifier. + chain_id: ChainId, + // Redacted RPC endpoint. + redacted_url: String, + // Submission attempt. + attempts: u32, + // Sanitized error message. + message: String, + // Transport error. + error: TransportError, +} + +impl SubmissionAttemptError { + pub(crate) fn new( + tx_hash: TxHash, + chain_id: ChainId, + redacted_url: String, + attempts: u32, + message: String, + error: TransportError, + ) -> Self { + Self { + tx_hash, + chain_id, + redacted_url, + attempts, + message, + error, + } + } +} + +impl From for OnchainError { + fn from(failure: SubmissionAttemptError) -> Self { + match failure.error { + // The request may have reached the node before its response was lost. + RpcError::Transport(_) | RpcError::NullResp | RpcError::DeserError { .. } => { + Self::SubmissionUnconfirmed { + tx_hash: failure.tx_hash, + chain_id: failure.chain_id, + redacted_url: failure.redacted_url, + attempts: failure.attempts, + message: failure.message, + } + } + // Other failures happen locally or are definite RPC rejections. + _ => { + Self::Send { + redacted_url: failure.redacted_url, + message: failure.message, + } + } + } + } +} + /// Errors that can occur during declarative apply. #[derive(Error, Debug)] pub enum ApplyError { diff --git a/crates/pcl/core/src/onchain.rs b/crates/pcl/core/src/onchain.rs index 04ded54..d6fa1c3 100644 --- a/crates/pcl/core/src/onchain.rs +++ b/crates/pcl/core/src/onchain.rs @@ -8,7 +8,11 @@ //! it is unavailable, so a node deduplicates the retry rather than accepting a //! second transaction. How long it waits: [`TxArgs::with_credible_rpc`]. -use crate::config::CliConfig; +pub use crate::error::OnchainError; +use crate::{ + config::CliConfig, + error::SubmissionAttemptError, +}; use alloy_network::{ Ethereum, EthereumWallet, @@ -49,7 +53,6 @@ use alloy_sol_types::{ }; use serde::Serialize; use std::time::Duration; -use thiserror::Error; use tokio::time::Instant; use url::Url; @@ -95,91 +98,6 @@ pub fn fallback_confirmations(chain_id: u64) -> u64 { } } -/// Errors that can occur while broadcasting transactions. -/// -/// RPC endpoints in these messages are redacted to their scheme/host/port -/// origin: stored provider URLs commonly embed API keys, and error envelopes -/// end up in logs and transcripts. -#[derive(Error, Debug)] -pub enum OnchainError { - #[error( - "No RPC endpoint for chain {chain_id}. Pass --rpc-url (or set PCL_RPC_URL), or store one with `pcl config set-rpc {chain_id} `." - )] - RpcUrlMissing { chain_id: u64 }, - - #[error("Invalid RPC URL ({redacted_url}) configured for chain {chain_id}: {reason}")] - InvalidRpcUrl { - chain_id: u64, - redacted_url: String, - reason: String, - }, - - #[error( - "{requested} confirmation(s) is below the {required} required on chain {chain_id}; the platform would reject the confirmation with INSUFFICIENT_CONFIRMATIONS. Raise --confirmations (or the stored per-chain value) to at least {required}." - )] - InsufficientConfirmations { - chain_id: u64, - requested: u64, - required: u64, - }, - - #[error( - "RPC endpoint {redacted_url} serves chain {actual}, but this transaction targets chain {expected}. Refusing to broadcast." - )] - ChainIdMismatch { - redacted_url: String, - expected: u64, - actual: u64, - }, - - #[error("RPC transport error communicating with {redacted_url}: {message}")] - Transport { - redacted_url: String, - message: String, - }, - - #[error("Failed to send transaction via {redacted_url}: {message}")] - Send { - redacted_url: String, - message: String, - }, - - #[error( - "Transaction {tx_hash} was submitted to chain {chain_id} but its confirmation state is unknown ({message}). Do not re-broadcast: the transaction may still confirm. Check it by hash first, and re-run the command only once its status is known." - )] - ConfirmationUnknown { - tx_hash: B256, - chain_id: u64, - redacted_url: String, - message: String, - }, - - #[error( - "{redacted_url} could not judge the transaction after {attempts} attempt(s) over {waited_ms}ms ({message}). Nothing was signed or submitted and the nonce is unchanged: re-run the command." - )] - AssertionsUnavailable { - chain_id: u64, - redacted_url: String, - attempts: u32, - waited_ms: u128, - message: String, - }, - - #[error( - "Transaction {tx_hash} was submitted to chain {chain_id} {attempts} time(s) without confirmed acceptance ({message}). It may be in flight: check the hash before re-running, which would reuse the same nonce." - )] - SubmissionUnconfirmed { - tx_hash: B256, - chain_id: u64, - redacted_url: String, - attempts: u32, - message: String, - }, - - #[error("Transaction {tx_hash} reverted on-chain (block {block})")] - Reverted { tx_hash: B256, block: u64 }, -} - /// Transaction arguments shared by every command that broadcasts. #[derive(clap::Args, Clone, Debug, Default)] pub struct TxArgs { @@ -455,10 +373,16 @@ impl TxSender { .await; } Err(error) => { - return Err(OnchainError::Send { - redacted_url: self.redacted(), - message: self.scrub(&error), - }); + let message = self.scrub(&error); + return Err(SubmissionAttemptError::new( + *signed.tx_hash(), + self.chain_id, + self.redacted(), + attempt, + message, + error, + ) + .into()); } } } @@ -1187,6 +1111,37 @@ mod tests { assert!(error.to_string().contains("may be in flight")); } + #[tokio::test] + async fn an_aborted_submission_response_preserves_the_signed_hash() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + ok(&mut server, "eth_estimateGas", json!("0x5208")).await; + let aborted = server + .mock("POST", "/") + .match_body(Matcher::PartialJson(json!({ + "method": "eth_sendRawTransaction", + }))) + .with_status(200) + .with_header("content-type", "application/json") + .with_chunked_body(|_| Err(std::io::Error::other("response aborted"))) + .create_async() + .await + .expect(1); + + let error = send(&server).await.unwrap_err(); + + let OnchainError::SubmissionUnconfirmed { + tx_hash, attempts, .. + } = &error + else { + panic!("expected an unconfirmed submission, got {error:?}"); + }; + assert_eq!(*attempts, 1); + assert!(error.to_string().contains(&tx_hash.to_string())); + aborted.assert_async().await; + } + #[tokio::test(start_paused = true)] async fn an_assertion_rejection_is_never_retried() { let mut server = mockito::Server::new_async().await; From eda1be46a332c837ee4ac8037523d648bbae9a60 Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Thu, 27 Aug 2026 12:51:26 +0200 Subject: [PATCH 10/11] fix(onchain): pin and track signed transactions --- crates/pcl/core/src/onchain.rs | 70 ++++++++++++++++++++++------------ 1 file changed, 45 insertions(+), 25 deletions(-) diff --git a/crates/pcl/core/src/onchain.rs b/crates/pcl/core/src/onchain.rs index d6fa1c3..3dd0107 100644 --- a/crates/pcl/core/src/onchain.rs +++ b/crates/pcl/core/src/onchain.rs @@ -17,7 +17,7 @@ use alloy_network::{ Ethereum, EthereumWallet, Network, - eip2718::Encodable2718, + TransactionBuilder, }; use alloy_primitives::{ B256, @@ -258,11 +258,9 @@ impl TxSender { notify: &dyn Fn(&str), ) -> Result { let signed = self.prepare(request, notify).await?; - self.submit(&signed, notify).await?; + let pending = self.submit(&signed, notify).await?; notify("Transaction submitted; waiting for confirmation"); - let receipt = self - .await_receipt(*signed.tx_hash(), confirmations, timeout) - .await?; + let receipt = self.await_receipt(pending, confirmations, timeout).await?; if !receipt.status() { return Err(OnchainError::Reverted { @@ -281,9 +279,9 @@ impl TxSender { }) } - // Resolves the nonce once and sets it, so the filler skips its own lookup. - // Left to the filler, every retried fill would increment its cached nonce - // and the signed transaction would skip one, stranding it as pending. + // Pins the pending nonce and chain validated at connect, skipping filler lookups. + // Retried fills could otherwise advance the cached nonce and skip one, + // stranding the signed transaction as pending. async fn with_pinned_nonce( &self, request: TransactionRequest, @@ -300,7 +298,7 @@ impl TxSender { message: self.scrub(&error), } })?; - Ok(request.from(from).nonce(nonce)) + Ok(request.from(from).nonce(nonce).with_chain_id(self.chain_id)) } // `eth_estimateGas` is gated too, so filling hits the same window; retrying @@ -352,20 +350,26 @@ impl TxSender { // Every attempt sends the same hash and nonce, so a node already holding the // transaction deduplicates the retry instead of accepting a second one. - async fn submit(&self, signed: &SignedTx, notify: &dyn Fn(&str)) -> Result<(), OnchainError> { + async fn submit( + &self, + signed: &SignedTx, + notify: &dyn Fn(&str), + ) -> Result, OnchainError> { let attempts = self.attempts(); let mut backoff = Backoff::new(); let mut message = String::new(); for attempt in 1..=attempts { - match self - .provider - .send_raw_transaction(&signed.encoded_2718()) - .await - { - Ok(_) => return Ok(()), + match self.provider.send_tx_envelope(signed.clone()).await { + Ok(pending) => return Ok(pending), // The node already holds these exact bytes, so an earlier - // attempt reached it even if its answer did not reach us. - Err(error) if already_submitted(&error) => return Ok(()), + // attempt reached it even if its answer did not reach us. It + // returned no builder, so resume tracking from the known hash. + Err(error) if already_submitted(&error) => { + return Ok(PendingTransactionBuilder::new( + self.provider.root().clone(), + *signed.tx_hash(), + )); + } Err(error) if assertions_unavailable(&error) => { message = self.scrub(&error); backoff @@ -395,14 +399,15 @@ impl TxSender { }) } - // Waits for `tx_hash` to reach `confirmations`. + // Waits for the submitted transaction to reach `confirmations`. async fn await_receipt( &self, - tx_hash: B256, + pending: PendingTransactionBuilder, confirmations: u64, timeout: Duration, ) -> Result { - PendingTransactionBuilder::new(self.provider.root().clone(), tx_hash) + let tx_hash = *pending.tx_hash(); + pending .with_required_confirmations(confirmations) .with_timeout(Some(timeout)) .get_receipt() @@ -766,11 +771,9 @@ mod tests { .unwrap(); let started = Instant::now(); let timeout = Duration::from_millis(50); + let pending = PendingTransactionBuilder::new(sender.provider.root().clone(), RECEIPT_HASH); - let error = sender - .await_receipt(RECEIPT_HASH, 1, timeout) - .await - .unwrap_err(); + let error = sender.await_receipt(pending, 1, timeout).await.unwrap_err(); assert!(matches!(error, OnchainError::ConfirmationUnknown { .. })); assert!(started.elapsed() >= timeout); @@ -913,6 +916,23 @@ mod tests { count.assert_async().await; } + #[tokio::test] + async fn the_validated_chain_id_is_pinned_before_signing() { + let mut server = mockito::Server::new_async().await; + ok(&mut server, "eth_chainId", json!(format!("{CHAIN_ID:#x}"))).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + let sender = TxSender::connect(server.url().parse().unwrap(), signer(), CHAIN_ID, true) + .await + .unwrap(); + + let request = sender + .with_pinned_nonce(TransactionRequest::default()) + .await + .unwrap(); + + assert_eq!(request.chain_id, Some(CHAIN_ID)); + } + // A retried fill must not advance the nonce: the signed transaction would // skip one and sit pending behind the gap until something else filled it. #[tokio::test(start_paused = true)] From 7ee2d22861055c022b0f6fc718dbbd09975af16f Mon Sep 17 00:00:00 2001 From: Lea Na <78718413+lean-apple@users.noreply.github.com> Date: Thu, 27 Aug 2026 13:08:55 +0200 Subject: [PATCH 11/11] fix(onchain): preserve credible RPC retry contract --- crates/pcl/core/src/api/definitions.rs | 22 ++++++++++++ .../src/api/workflows/protocol_manager.rs | 4 +-- crates/pcl/core/src/api/workflows/releases.rs | 4 +-- crates/pcl/core/src/error.rs | 36 ++++++++++--------- crates/pcl/core/src/onchain.rs | 35 ++++++++++++++++++ 5 files changed, 81 insertions(+), 20 deletions(-) diff --git a/crates/pcl/core/src/api/definitions.rs b/crates/pcl/core/src/api/definitions.rs index fddf6d2..9e80daf 100644 --- a/crates/pcl/core/src/api/definitions.rs +++ b/crates/pcl/core/src/api/definitions.rs @@ -326,6 +326,28 @@ mod tests { assert_eq!(broadcast_actions, 4); } + #[test] + fn broadcast_actions_expose_the_credible_rpc_retry_flag() { + for definition in workflow_definitions() { + for action in definition + .actions + .iter() + .filter(|action| action.name.ends_with("_broadcast")) + { + let manifest = action.manifest_value(); + let optional_flags = manifest["optional_flags"].as_array().unwrap(); + assert!( + optional_flags + .iter() + .any(|flag| flag == "--with-credible-rpc"), + "{}.{} omits --with-credible-rpc", + definition.name, + action.name + ); + } + } + } + /// The regression the metadata exists for: an action anchored to a /// read-only calldata GET (`transfer_broadcast`) is still mutating, while /// the plain calldata action on the same endpoint stays read-only. diff --git a/crates/pcl/core/src/api/workflows/protocol_manager.rs b/crates/pcl/core/src/api/workflows/protocol_manager.rs index a5e8df8..17f9f4f 100644 --- a/crates/pcl/core/src/api/workflows/protocol_manager.rs +++ b/crates/pcl/core/src/api/workflows/protocol_manager.rs @@ -186,9 +186,9 @@ workflow_definition!( action!("set", true, "post_projects_project_id_protocol_manager", "pcl protocol-manager --project --set --body-template", required: ["--project"], body_template: "protocol_manager_set"), action!("clear", true, "delete_projects_project_id_protocol_manager", "pcl protocol-manager --project --clear", required: ["--project"], body_template: "empty_object"), action!("transfer_calldata", true, "get_projects_project_id_protocol_manager_transfer_calldata", "pcl protocol-manager --project --transfer-calldata --new-manager 0x...", required: ["--project", "--new-manager"], query: {"new_manager" => "
"}), - action!("transfer_broadcast", true, "get_projects_project_id_protocol_manager_transfer_calldata", "pcl protocol-manager --project --transfer-calldata --new-manager 0x... --broadcast --private-key --rpc-url --yes", required: ["--project", "--new-manager", "--broadcast"], optional: ["--private-key", "--account", "--keystore-password-file", "--rpc-url", "--confirmations", "--yes"], query: {"new_manager" => "
"}, side_effect: "onchain_transaction"), + action!("transfer_broadcast", true, "get_projects_project_id_protocol_manager_transfer_calldata", "pcl protocol-manager --project --transfer-calldata --new-manager 0x... --broadcast --private-key --rpc-url --yes", required: ["--project", "--new-manager", "--broadcast"], optional: ["--private-key", "--account", "--keystore-password-file", "--rpc-url", "--confirmations", "--with-credible-rpc", "--yes"], query: {"new_manager" => "
"}, side_effect: "onchain_transaction"), action!("accept_calldata", true, "get_projects_project_id_protocol_manager_accept_calldata", "pcl protocol-manager --project --accept-calldata", required: ["--project"]), - action!("accept_broadcast", true, "post_projects_project_id_protocol_manager_confirm_transfer", "pcl protocol-manager --project --accept-calldata --broadcast --private-key --rpc-url --yes", required: ["--project", "--broadcast"], optional: ["--private-key", "--account", "--keystore-password-file", "--rpc-url", "--confirmations", "--yes"], side_effect: "onchain_transaction"), + action!("accept_broadcast", true, "post_projects_project_id_protocol_manager_confirm_transfer", "pcl protocol-manager --project --accept-calldata --broadcast --private-key --rpc-url --yes", required: ["--project", "--broadcast"], optional: ["--private-key", "--account", "--keystore-password-file", "--rpc-url", "--confirmations", "--with-credible-rpc", "--yes"], side_effect: "onchain_transaction"), action!("confirm_transfer", true, "post_projects_project_id_protocol_manager_confirm_transfer", "pcl protocol-manager --project --confirm-transfer --body-template", required: ["--project"], body_template: "protocol_manager_confirm"), ], ); diff --git a/crates/pcl/core/src/api/workflows/releases.rs b/crates/pcl/core/src/api/workflows/releases.rs index 2e6dd0e..90671c8 100644 --- a/crates/pcl/core/src/api/workflows/releases.rs +++ b/crates/pcl/core/src/api/workflows/releases.rs @@ -214,10 +214,10 @@ workflow_definition!( action!("backtest_progress", true, "get_projects_project_id_releases_release_id_backtest_progress", "pcl releases backtest-progress ", required: ["", ""]), action!("retry_check", true, "post_projects_project_id_releases_release_id_checks_check_id_retry", "pcl releases retry-check ", required: ["", "", ""], body_template: "empty_object"), action!("deploy_calldata", true, "get_projects_project_id_releases_release_id_deploy_calldata", "pcl releases calldata deploy --signer-address ", required: ["", "", "--signer-address"], query: {"signerAddress" => ""}), - action!("deploy_broadcast", true, "post_projects_project_id_releases_release_id_deploy", "pcl releases calldata deploy --broadcast --private-key --rpc-url --yes", required: ["", "", "--broadcast"], optional: ["--private-key", "--account", "--keystore-password-file", "--rpc-url", "--confirmations", "--yes", "--signer-address"], side_effect: "onchain_transaction"), + action!("deploy_broadcast", true, "post_projects_project_id_releases_release_id_deploy", "pcl releases calldata deploy --broadcast --private-key --rpc-url --yes", required: ["", "", "--broadcast"], optional: ["--private-key", "--account", "--keystore-password-file", "--rpc-url", "--confirmations", "--with-credible-rpc", "--yes", "--signer-address"], side_effect: "onchain_transaction"), action!("deploy", true, "post_projects_project_id_releases_release_id_deploy", "pcl releases deploy --body-template", required: ["", ""], body_template: "release_deploy"), action!("remove_calldata", true, "get_projects_project_id_releases_release_id_remove_calldata", "pcl releases calldata remove ", required: ["", ""]), - action!("remove_broadcast", true, "post_projects_project_id_releases_release_id_remove", "pcl releases calldata remove --broadcast --private-key --rpc-url --yes", required: ["", "", "--broadcast"], optional: ["--private-key", "--account", "--keystore-password-file", "--rpc-url", "--confirmations", "--yes"], side_effect: "onchain_transaction"), + action!("remove_broadcast", true, "post_projects_project_id_releases_release_id_remove", "pcl releases calldata remove --broadcast --private-key --rpc-url --yes", required: ["", "", "--broadcast"], optional: ["--private-key", "--account", "--keystore-password-file", "--rpc-url", "--confirmations", "--with-credible-rpc", "--yes"], side_effect: "onchain_transaction"), action!("remove", true, "post_projects_project_id_releases_release_id_remove", "pcl releases remove --body-template", required: ["", ""], body_template: "release_remove"), ], ); diff --git a/crates/pcl/core/src/error.rs b/crates/pcl/core/src/error.rs index 2e0b563..5cc9472 100644 --- a/crates/pcl/core/src/error.rs +++ b/crates/pcl/core/src/error.rs @@ -119,6 +119,8 @@ pub(crate) struct SubmissionAttemptError { redacted_url: String, // Submission attempt. attempts: u32, + // Whether an earlier response made submission ambiguous. + previously_ambiguous: bool, // Sanitized error message. message: String, // Transport error. @@ -131,6 +133,7 @@ impl SubmissionAttemptError { chain_id: ChainId, redacted_url: String, attempts: u32, + previously_ambiguous: bool, message: String, error: TransportError, ) -> Self { @@ -139,6 +142,7 @@ impl SubmissionAttemptError { chain_id, redacted_url, attempts, + previously_ambiguous, message, error, } @@ -147,23 +151,23 @@ impl SubmissionAttemptError { impl From for OnchainError { fn from(failure: SubmissionAttemptError) -> Self { - match failure.error { - // The request may have reached the node before its response was lost. - RpcError::Transport(_) | RpcError::NullResp | RpcError::DeserError { .. } => { - Self::SubmissionUnconfirmed { - tx_hash: failure.tx_hash, - chain_id: failure.chain_id, - redacted_url: failure.redacted_url, - attempts: failure.attempts, - message: failure.message, - } + let ambiguous = failure.previously_ambiguous + || matches!( + failure.error, + RpcError::Transport(_) | RpcError::NullResp | RpcError::DeserError { .. } + ); + if ambiguous { + Self::SubmissionUnconfirmed { + tx_hash: failure.tx_hash, + chain_id: failure.chain_id, + redacted_url: failure.redacted_url, + attempts: failure.attempts, + message: failure.message, } - // Other failures happen locally or are definite RPC rejections. - _ => { - Self::Send { - redacted_url: failure.redacted_url, - message: failure.message, - } + } else { + Self::Send { + redacted_url: failure.redacted_url, + message: failure.message, } } } diff --git a/crates/pcl/core/src/onchain.rs b/crates/pcl/core/src/onchain.rs index 3dd0107..6368f8a 100644 --- a/crates/pcl/core/src/onchain.rs +++ b/crates/pcl/core/src/onchain.rs @@ -358,6 +358,7 @@ impl TxSender { let attempts = self.attempts(); let mut backoff = Backoff::new(); let mut message = String::new(); + let mut previously_ambiguous = false; for attempt in 1..=attempts { match self.provider.send_tx_envelope(signed.clone()).await { Ok(pending) => return Ok(pending), @@ -371,6 +372,7 @@ impl TxSender { )); } Err(error) if assertions_unavailable(&error) => { + previously_ambiguous = true; message = self.scrub(&error); backoff .wait(attempt, attempts, &self.waiting_for("submitting"), notify) @@ -383,6 +385,7 @@ impl TxSender { self.chain_id, self.redacted(), attempt, + previously_ambiguous, message, error, ) @@ -1062,6 +1065,38 @@ mod tests { dedup.assert_async().await; } + #[tokio::test(start_paused = true)] + async fn a_rejection_after_an_ambiguous_refusal_preserves_the_signed_hash() { + let mut server = mockito::Server::new_async().await; + chain(&mut server).await; + ok(&mut server, "eth_getTransactionCount", json!("0x8")).await; + ok(&mut server, "eth_estimateGas", json!("0x5208")).await; + fails(&mut server, "eth_sendRawTransaction", INTERNAL, UNAVAILABLE) + .await + .expect(1); + let rejected = fails( + &mut server, + "eth_sendRawTransaction", + -32000, + "nonce too low", + ) + .await + .expect(1); + ok(&mut server, "eth_getTransactionByHash", Value::Null).await; + + let error = send(&server).await.unwrap_err(); + + let OnchainError::SubmissionUnconfirmed { + tx_hash, attempts, .. + } = &error + else { + panic!("expected an ambiguous submission, got {error:?}"); + }; + assert_eq!(*attempts, 2); + assert!(error.to_string().contains(&tx_hash.to_string())); + rejected.assert_async().await; + } + #[tokio::test(start_paused = true)] async fn refused_gas_estimation_is_retried_before_anything_is_signed() { let mut server = mockito::Server::new_async().await;