newton-chainio 0.5.2

//! Single definition of the same-nonce cancel + confirm loop.
//!
//! A reserved nonce whose on-chain fate is uncertain (a stuck batch, a post-fill
//! send failure, or a receipt timeout) must be **resolved on-chain** — never
//! abandoned. Abandoning a consumed-but-unmined nonce strands every higher nonce
//! ("blocked from below"), the exact wedge the owned [`NonceAllocator`] exists to
//! prevent. The escape is a 0-value self-send at the same nonce, escalated until
//! it out-prices whatever is stuck there and mines.
//!
//! That loop used to be inlined in the gateway's pipelined receipt tracker. It is
//! now here so both the pipelined path AND the direct `.send()` /
//! `send_with_retries` paths (`send_task`, `batch_respond_to_tasks`,
//! `send_aggregated_response`, `commit_state_root`) share one implementation with
//! identical semantics: unbounded retry with fee escalation, shutdown-aware,
//! blocked-from-below detection + allocator resync, and the same alertable
//! `nonce_cancel_failures_total` metric. Callers layer their own bookkeeping
//! (payload re-queue, item failure) on top of the [`CancelOutcome`] it returns.
//!
//! [`NonceAllocator`]: crate::avs::nonce_allocator::NonceAllocator

use crate::error::ChainIoError;
use std::time::Duration;
use tokio_util::sync::CancellationToken;
use tracing::{error, info, warn};

/// Capabilities the cancel loop needs from whatever owns the signer. Implemented
/// by `AvsWriter` directly (chainio paths) and by `dyn TaskSubmitter` (the
/// gateway pipelined path, so its mock-based tests keep intercepting the loop).
#[async_trait::async_trait]
pub trait NonceCanceller: Sync {
    /// Broadcast a same-nonce 0-value self-send priced to replace what is stuck
    /// at `nonce` (escalated `>= bump_percent` over `prev_*`). Returns the cancel
    /// tx hash + the fees it was actually broadcast with.
    async fn broadcast_cancel(
        &self,
        nonce: u64,
        prev_max_fee: u128,
        prev_priority_fee: u128,
        bump_percent: u32,
    ) -> Result<CancelBroadcast, ChainIoError>;

    /// Poll for a receipt by hash, returning `TransactionTimeout` after `timeout`.
    async fn await_receipt(&self, tx_hash: alloy::primitives::B256, timeout: Duration) -> Result<(), ChainIoError>;

    /// Latest *mined* tx count for the signer (for blocked-from-below detection).
    async fn latest_mined_nonce(&self) -> Result<u64, ChainIoError>;

    /// Re-seed the owned nonce allocator from the chain (gap self-heal). No-op
    /// when no allocator is attached.
    async fn resync_allocator(&self);
}

/// Fees a cancel was broadcast with, threaded into the next escalation.
#[derive(Debug, Clone, Copy)]
pub struct CancelBroadcast {
    /// Hash of the broadcast cancel tx, polled for confirmation.
    pub tx_hash: alloy::primitives::B256,
    /// `max_fee_per_gas` the cancel was broadcast with.
    pub max_fee_per_gas: u128,
    /// `max_priority_fee_per_gas` the cancel was broadcast with.
    pub max_priority_fee_per_gas: u128,
}

/// Terminal outcome of [`cancel_until_resolved`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CancelOutcome {
    /// The nonce slot is confirmed freed on-chain (cancel mined, or the slot was
    /// already consumed under another hash). The caller may re-queue its payload.
    Resolved,
    /// Shutdown was requested mid-resolution. The slot is NOT confirmed freed; the
    /// caller should fail its items definitively (the startup sweep recovers the
    /// slot on the next run). Only ever returned when a `CancellationToken` is
    /// supplied and fires.
    Shutdown,
}

/// Knobs for the cancel loop, so callers can match their existing cadence.
#[derive(Debug, Clone, Copy)]
pub struct CancelParams {
    /// `>=` replacement bump applied to each cancel broadcast.
    pub bump_percent: u32,
    /// How long to wait for a cancel to mine before re-broadcasting (escalated).
    pub confirm_timeout: Duration,
    /// Backoff after an *immediate* (non-timeout) broadcast/poll error, so a
    /// flapping RPC or an unfunded signer can't spin this into a hot loop.
    pub retry_backoff: Duration,
}

impl Default for CancelParams {
    fn default() -> Self {
        Self {
            bump_percent: 20,
            confirm_timeout: Duration::from_secs(60),
            retry_backoff: Duration::from_secs(2),
        }
    }
}

/// Resolve a stuck nonce on-chain with an **unbounded**, fee-escalating same-nonce
/// cancel + confirm loop. Never gives up except on shutdown — a wedged signer
/// (e.g. unfunded) recovers automatically once unblocked, and abandoning the
/// nonce is exactly the bug this prevents.
///
/// `seed_max_fee`/`seed_priority` seed the first cancel's price (pass the stuck
/// tx's own fees so the first attempt out-prices it; 0 falls back to a fresh
/// market quote inside `broadcast_cancel`). `cancel_token` is optional: when
/// `None` the loop is truly unbounded (used by paths with no shutdown plumbing);
/// when `Some`, the loop exits promptly with [`CancelOutcome::Shutdown`].
pub async fn cancel_until_resolved<C: NonceCanceller + ?Sized>(
    canceller: &C,
    chain_id: u64,
    nonce: u64,
    seed_max_fee: u128,
    seed_priority: u128,
    params: CancelParams,
    cancel_token: Option<&CancellationToken>,
) -> CancelOutcome {
    let is_cancelled = || cancel_token.is_some_and(|t| t.is_cancelled());

    // ── Phase 1: broadcast the cancel, retrying until accepted ──────────────
    let mut prev_max_fee = seed_max_fee;
    let mut prev_priority = seed_priority;
    let broadcast = loop {
        if is_cancelled() {
            return CancelOutcome::Shutdown;
        }
        match canceller
            .broadcast_cancel(nonce, prev_max_fee, prev_priority, params.bump_percent)
            .await
        {
            Ok(b) => break b,
            // "nonce too low" — slot already consumed on-chain (original, bump, or
            // an earlier cancel mined). Nothing to confirm; treat as resolved.
            Err(e) if e.is_nonce_consumed() => {
                info!(
                    nonce,
                    "cancel rejected nonce-too-low — slot already consumed on-chain, resolved"
                );
                return CancelOutcome::Resolved;
            }
            Err(e) => {
                // Loud + alertable: a sustained rate here means the signer is
                // wedged (most likely unfunded) and needs operator attention. We
                // do NOT give up — funding recovers it automatically.
                newton_metric::inc_batch_nonce_cancel_failures(chain_id);
                error!(
                    error = %e,
                    nonce,
                    "NONCE CANCEL BROADCAST FAILED — signer may be out of funds; retrying until it clears (alert on nonce_cancel_failures_total)"
                );
                // Climb only when a higher bid can help. On "insufficient funds"
                // the bid is already unaffordable — escalating deepens the hole
                // (the runaway that priced a 21k cancel at ~786k gwei). Hold flat.
                if !e.is_insufficient_funds() {
                    prev_max_fee = bump(prev_max_fee);
                    prev_priority = bump(prev_priority);
                }
                sleep_or_cancel(params.retry_backoff, cancel_token).await;
            }
        }
    };
    newton_metric::inc_batch_nonce_cancels(chain_id);

    // ── Phase 2: confirm it mines; re-broadcast (escalated) on stall ────────
    let mut tx_hash = broadcast.tx_hash;
    prev_max_fee = broadcast.max_fee_per_gas;
    prev_priority = broadcast.max_priority_fee_per_gas;
    // Set when the last re-broadcast was rejected unaffordable: don't climb (that
    // only deepens the hole); re-broadcast flat until the signer is funded.
    let mut unaffordable = false;
    loop {
        if is_cancelled() {
            return CancelOutcome::Shutdown;
        }
        let poll = match cancel_token {
            Some(token) => tokio::select! {
                biased;
                _ = token.cancelled() => return CancelOutcome::Shutdown,
                r = canceller.await_receipt(tx_hash, params.confirm_timeout) => r,
            },
            None => canceller.await_receipt(tx_hash, params.confirm_timeout).await,
        };
        match poll {
            Ok(()) => return CancelOutcome::Resolved, // cancel mined — slot freed
            Err(e) => {
                let was_timeout = matches!(e, ChainIoError::TransactionTimeout { .. });
                if !was_timeout {
                    // Immediate (RpcError) poll failure — back off so a flapping
                    // RPC can't hot-loop the node.
                    sleep_or_cancel(params.retry_backoff, cancel_token).await;
                } else {
                    // Cancel accepted but unmined for a full timeout. If the
                    // chain's latest-mined nonce is still BELOW ours, a lower nonce
                    // is stuck and no fee on our nonce can help until the gap
                    // drains — surface it and resync the allocator (backstop; the
                    // orphan that created such gaps is now prevented at the source).
                    if let Ok(latest) = canceller.latest_mined_nonce().await {
                        let gap = nonce.saturating_sub(latest);
                        newton_metric::set_batch_signer_nonce_gap(chain_id, gap);
                        if gap > 0 {
                            warn!(
                                nonce,
                                latest, gap, "cancel stalled with chain blocked from below; resyncing allocator"
                            );
                            canceller.resync_allocator().await;
                        }
                    }
                }
                if !unaffordable {
                    prev_max_fee = bump(prev_max_fee);
                    prev_priority = bump(prev_priority);
                }
                match canceller
                    .broadcast_cancel(nonce, prev_max_fee, prev_priority, params.bump_percent)
                    .await
                {
                    Ok(b) => {
                        tx_hash = b.tx_hash;
                        prev_max_fee = b.max_fee_per_gas;
                        prev_priority = b.max_priority_fee_per_gas;
                        unaffordable = false;
                    }
                    // Slot drained while polling a stale hash — confirmed freed.
                    Err(e) if e.is_nonce_consumed() => {
                        info!(
                            nonce,
                            "cancel re-broadcast rejected nonce-too-low — slot consumed on-chain, freed"
                        );
                        return CancelOutcome::Resolved;
                    }
                    Err(e) => {
                        newton_metric::inc_batch_nonce_cancel_failures(chain_id);
                        unaffordable = e.is_insufficient_funds();
                        error!(
                            error = %e,
                            nonce,
                            unaffordable,
                            "cancel re-broadcast failed while awaiting confirmation; retrying (alert on nonce_cancel_failures_total)"
                        );
                    }
                }
            }
        }
    }
}

/// `>= 10%` bump — clears the node's EIP-1559 replacement floor.
fn bump(fee: u128) -> u128 {
    fee.saturating_mul(110) / 100
}

/// Sleep, returning early on shutdown (no-op wait when no token is supplied).
async fn sleep_or_cancel(dur: Duration, cancel_token: Option<&CancellationToken>) {
    match cancel_token {
        Some(token) => {
            tokio::select! {
                biased;
                _ = token.cancelled() => {}
                _ = tokio::time::sleep(dur) => {}
            }
        }
        None => tokio::time::sleep(dur).await,
    }
}