newton-tx-executor 0.7.3

//! Durable signer-lane executor.
//!
//! Each [`ManagedSigner`] is one independent nonce lane with at most one durable
//! task assignment. A new assignment always reads its nonce from the chain's
//! mined transaction count; no local or provider nonce cache advances it.
//! Execution then follows a strict persist-before-broadcast sequence:
//!
//! 1. reserve the job and signer together;
//! 2. prepare deterministic signed bytes;
//! 3. commit those bytes and their nonce;
//! 4. broadcast exactly the committed bytes;
//! 5. watch through finality, replacing or cancelling at the same nonce;
//! 6. classify every batch item from authoritative on-chain hashes.
//!
//! Recoverable chain uncertainty makes only the affected signer lane wait and
//! re-reconcile. Hard durable-state invariant failures quarantine that lane;
//! the executor keeps checking it so repairing the underlying deployment or
//! durable state does not require an application admin API. Other signers
//! continue.

mod outcomes;
mod tracking;

use self::{
    outcomes::{classify_and_project, classify_preflight_failure, finalize_preflight_failure},
    tracking::{maximum_transaction_cost, track_attempt},
};
use crate::{BackendError, ChainBackend, ObservationScope, TransactionObservation};
use alloy::primitives::{Address, U256};
use newton_submission_protocol::SignerId;
use newton_submission_service::{
    ActiveAssignment, AttemptKind, AttemptRecord, AttemptState, ExecutableIntent, JobRecord, PreparedAttempt,
    RuntimeHealth, RuntimeTask, SignerLifecycle, SignerRecord, SignerRole, Store, StoreError,
};
use newton_task_submission::{submission_ids_field, task_ids_field};
use std::{
    sync::Arc,
    time::{Duration, SystemTime, UNIX_EPOCH},
};
use tokio::{task::JoinSet, time::MissedTickBehavior};
use tokio_util::sync::CancellationToken;
use tracing::{error, info, info_span, warn, Instrument, Span};

/// One physical nonce lane and its backend.
#[derive(Clone)]
pub struct ManagedSigner {
    /// Stable operational label for this nonce lane.
    pub signer_id: SignerId,
    /// EVM chain whose nonce space this lane owns.
    pub chain_id: u64,
    /// Transaction class this lane may reserve.
    pub role: SignerRole,
    /// Expected destination-chain block interval used to pace receipt checks.
    pub block_time: Duration,
    /// Signing and ordered-provider boundary for the lane.
    pub backend: Arc<dyn ChainBackend>,
}

impl std::fmt::Debug for ManagedSigner {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ManagedSigner")
            .field("signer_id", &self.signer_id)
            .field("chain_id", &self.chain_id)
            .field("role", &self.role)
            .field("block_time", &self.block_time)
            .field("address", &self.backend.address())
            .finish()
    }
}

impl ManagedSigner {
    fn address(&self) -> Address {
        self.backend.address()
    }

    /// Snapshots the configured lane identity for durable registration and reservation.
    ///
    /// The backend address and chain identify the durable EVM nonce lane. The
    /// signer ID remains an operational label that startup reconciliation may
    /// safely rebind when the loaded key still owns the same address.
    fn record(&self) -> SignerRecord {
        SignerRecord {
            signer_id: self.signer_id.clone(),
            chain_id: self.chain_id,
            role: self.role,
            address: self.address(),
        }
    }
}

/// Worker timing policy.
#[derive(Debug, Clone)]
pub struct ExecutorConfig {
    /// Fallback delay for idle work polling and lane reconciliation.
    pub poll_interval: Duration,
    /// Delay between receipt and finality observations.
    pub receipt_poll_interval: Duration,
    /// Upper bound for adaptive receipt polling after repeated misses/errors.
    pub receipt_poll_max_interval: Duration,
    /// Pending duration before preparing a same-nonce replacement.
    pub watchdog_timeout: Duration,
    /// Maximum task replacements before switching to cancellation.
    pub cancel_after_bumps: u32,
    /// Balance recheck interval for an unsigned underfunded signer lane.
    pub underfunded_poll_interval: Duration,
}

/// Supervises one independent worker per configured signer lane.
#[derive(Debug)]
pub struct Executor {
    store: Arc<Store>,
    signers: Vec<ManagedSigner>,
    config: ExecutorConfig,
    health: Arc<RuntimeHealth>,
}

impl Executor {
    /// Creates an executor without starting workers or touching durable state.
    ///
    /// Signer registration and chain reconciliation are deliberately deferred to
    /// [`Self::run`], so construction cannot make a lane eligible for work.
    pub fn new(
        store: Arc<Store>,
        signers: Vec<ManagedSigner>,
        config: ExecutorConfig,
        health: Arc<RuntimeHealth>,
    ) -> Self {
        Self {
            store,
            signers,
            config,
            health,
        }
    }

    /// Registers, reconciles, and supervises every independent nonce lane.
    ///
    /// All configured signer identities are validated against the durable store
    /// before any workers start. Each worker then reconciles its lane with the chain
    /// before entering the reservation loop. Permanent provider/configuration
    /// failures quarantine only their lane; shared-store failures and worker panics
    /// cancel the siblings so the service-level supervisor can restart the executor
    /// coherently. Normal cancellation also joins every lane before returning.
    pub async fn run(&self, cancellation: CancellationToken) -> Result<(), ExecutorError> {
        self.health.heartbeat(RuntimeTask::Executor);
        let configured_signers = self.signers.iter().map(ManagedSigner::record).collect::<Vec<_>>();
        self.store.register_signers(&configured_signers).await?;
        let lane_cancellation = cancellation.child_token();
        let mut lanes = JoinSet::new();
        for signer in self.signers.clone() {
            let store = self.store.clone();
            let config = self.config.clone();
            let cancellation = lane_cancellation.clone();
            let health = self.health.clone();
            lanes.spawn(async move {
                if recover_lane(&store, &signer, &config, &cancellation, &health).await? == LaneRecovery::Stopped {
                    return Ok(());
                }
                worker_loop(store, signer, config, cancellation, health).await
            });
        }
        let mut heartbeat = tokio::time::interval(self.config.poll_interval.max(Duration::from_secs(1)));
        heartbeat.set_missed_tick_behavior(MissedTickBehavior::Skip);
        let failure = loop {
            tokio::select! {
                _ = cancellation.cancelled() => break None,
                _ = heartbeat.tick() => self.health.heartbeat(RuntimeTask::Executor),
                joined = lanes.join_next() => match joined {
                    Some(Ok(Ok(()))) if cancellation.is_cancelled() => break None,
                    Some(Ok(Ok(()))) => break Some(ExecutorError::WorkerExited),
                    Some(Ok(Err(error))) => break Some(error),
                    Some(Err(error)) => break Some(ExecutorError::WorkerPanicked(error.to_string())),
                    None => break Some(ExecutorError::WorkerExited),
                },
            }
        };
        lane_cancellation.cancel();
        while let Some(joined) = lanes.join_next().await {
            if let Err(error) = joined {
                return Err(ExecutorError::WorkerPanicked(error.to_string()));
            }
        }
        if let Some(error) = failure {
            return Err(error);
        }
        Ok(())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Result of bringing one signer lane back to a runnable boundary.
enum LaneRecovery {
    /// Durable state and authoritative chain state agree, so work may be reserved.
    Ready,
    /// Cancellation interrupted recovery before the lane became eligible.
    Stopped,
}

/// Reconciles one lane until it is safe to reserve work or cancellation wins.
///
/// Chain uncertainty and transient provider failures keep only this lane out of the
/// ready pool and retry on the normal poll interval. An unsigned underfunded lane
/// uses the slower balance-recovery interval. Durable journal contradictions move
/// the lane to `quarantined`, but reconciliation continues so a repaired invariant
/// can recover automatically without an administrative endpoint.
async fn recover_lane(
    store: &Store,
    signer: &ManagedSigner,
    config: &ExecutorConfig,
    cancellation: &CancellationToken,
    health: &RuntimeHealth,
) -> Result<LaneRecovery, ExecutorError> {
    loop {
        match reconcile(store, signer, config, cancellation.clone()).await {
            Ok(()) => {
                health.heartbeat(RuntimeTask::Executor);
                health.record_rpc(signer.chain_id, Ok(()));
                newton_metric::inc_signer_recoveries_total(signer.chain_id, "ready");
                newton_metric::set_signer_ready(signer.chain_id, signer.signer_id.as_str(), true);
                return Ok(LaneRecovery::Ready);
            }
            Err(error) if error.waits_for_chain() => {
                health.heartbeat(RuntimeTask::Executor);
                if matches!(error, ExecutorError::UnownedPendingTransaction { .. }) {
                    store
                        .mark_signer_waiting(&signer.signer_id, signer.chain_id, &error.to_string())
                        .await?;
                }
                newton_metric::set_signer_ready(signer.chain_id, signer.signer_id.as_str(), false);
                warn!(
                    signer_id = %signer.signer_id,
                    signer_address = %signer.address(),
                    %error,
                    "signer lane is waiting for authoritative chain state"
                );
            }
            Err(error) if error.is_underfunded() => {
                health.heartbeat(RuntimeTask::Executor);
                record_backend_error(health, signer.chain_id, &error);
                newton_metric::set_signer_ready(signer.chain_id, signer.signer_id.as_str(), false);
                warn!(
                    signer_id = %signer.signer_id,
                    signer_address = %signer.address(),
                    %error,
                    retry_after = ?config.underfunded_poll_interval,
                    "signer lane is underfunded; waiting before balance recheck"
                );
                if sleep_or_cancel(config.underfunded_poll_interval, cancellation).await {
                    return Ok(LaneRecovery::Stopped);
                }
                continue;
            }
            Err(error) if error.is_transient() => {
                health.heartbeat(RuntimeTask::Executor);
                let message = error.to_string();
                // A contended store is not a sick RPC provider; attributing it
                // to the provider would corrupt submission_rpc_provider_healthy.
                if error.is_provider_failure() {
                    health.record_rpc(signer.chain_id, Err(&message));
                }
                warn!(
                    signer_id = %signer.signer_id,
                    signer_address = %signer.address(),
                    %error,
                    provider_failure = error.is_provider_failure(),
                    "transient signer recovery failure; retrying"
                );
            }
            Err(error) if error.requires_quarantine() => {
                if store.signer_lifecycle(&signer.signer_id, signer.chain_id).await? != SignerLifecycle::Quarantined {
                    store
                        .quarantine_signer(&signer.signer_id, signer.chain_id, &error.to_string())
                        .await?;
                    newton_metric::inc_signer_recoveries_total(signer.chain_id, "quarantined");
                    error!(
                        signer_id = %signer.signer_id,
                        signer_address = %signer.address(),
                        %error,
                        "signer safety invariant failed; quarantining nonce lane"
                    );
                }
                newton_metric::set_signer_ready(signer.chain_id, signer.signer_id.as_str(), false);
            }
            Err(error) => return Err(error),
        }
        if sleep_or_cancel(config.poll_interval, cancellation).await {
            return Ok(LaneRecovery::Stopped);
        }
    }
}

/// Reconciles one signer's durable assignment with authoritative chain state.
///
/// The workflow is intentionally based only on `latest` and `pending` transaction
/// counts; it never increments or caches a nonce. An idle lane becomes ready only
/// when those counts agree. An unsigned reservation can be released safely, while a
/// reservation with persisted signed bytes remains owned and is recovered through
/// [`recover_prepared_assignment`]. Underfunded lanes first prove that their balance
/// meets the previously persisted threshold.
async fn reconcile(
    store: &Store,
    signer: &ManagedSigner,
    config: &ExecutorConfig,
    cancellation: CancellationToken,
) -> Result<(), ExecutorError> {
    if store.signer_lifecycle(&signer.signer_id, signer.chain_id).await? == SignerLifecycle::Underfunded {
        let required = store
            .signer_required_balance(&signer.signer_id, signer.chain_id)
            .await?
            .ok_or(ExecutorError::MissingRequiredBalance)?;
        let balance = signer
            .backend
            .balance()
            .await
            .map_err(ExecutorError::UnderfundedBalanceRead)?;
        if balance < required {
            return Err(ExecutorError::InsufficientSignerBalance { balance, required });
        }
        info!(
            signer_id = %signer.signer_id,
            signer_address = %signer.address(),
            %balance,
            %required,
            "signer balance recovered"
        );
        store.mark_signer_ready(&signer.signer_id, signer.chain_id).await?;
    }
    let latest = signer.backend.latest_transaction_count().await?;
    let pending = signer.backend.pending_transaction_count().await?;
    newton_metric::set_signer_nonce_gap(signer.chain_id, signer.signer_id.as_str(), latest, pending);
    let assignment = store.active_assignment(signer.address(), signer.chain_id).await?;
    if assignment.is_some() {
        store.restore_signer_busy(&signer.signer_id, signer.chain_id).await?;
    }
    match assignment {
        None => {
            if latest != pending {
                return Err(ExecutorError::UnownedPendingTransaction {
                    signer: signer.backend.address(),
                    latest,
                    pending,
                });
            }
            if latest > 0 && !store.signer_has_journal(signer.address(), signer.chain_id).await? {
                info!(
                    signer_id = %signer.signer_id,
                    signer_address = %signer.address(),
                    nonce = latest,
                    "derived signer enrollment baseline from RPC"
                );
            }
            store.mark_signer_ready(&signer.signer_id, signer.chain_id).await?;
        }
        Some(assignment) if assignment.attempts.is_empty() => {
            if latest != pending {
                store
                    .release_safe_reservation(&signer.signer_id, signer.chain_id, assignment.job.job_id)
                    .await?;
                return Err(ExecutorError::UnownedPendingTransaction {
                    signer: signer.backend.address(),
                    latest,
                    pending,
                });
            }
            store
                .release_safe_reservation(&signer.signer_id, signer.chain_id, assignment.job.job_id)
                .await?;
        }
        Some(assignment) => {
            let span = submission_job_span(signer, &assignment.job);
            async {
                info!(
                    attempt_count = assignment.attempts.len(),
                    "resuming durable submission job"
                );
                recover_prepared_assignment(store, signer, config, assignment, latest, pending, cancellation).await
            }
            .instrument(span)
            .await?;
        }
    }
    info!(
        signer_id = %signer.signer_id,
        signer_address = %signer.address(),
        latest,
        "signer reconciled"
    );
    Ok(())
}

/// Reserves and executes jobs serially for one signer-and-chain nonce lane.
///
/// Reservation atomically moves both the selected job and signer to `busy`. With no
/// matching work, the worker waits for the store notification with a polling fallback
/// so missed wakeups cannot strand jobs. Retryable execution errors return to lane
/// reconciliation; local safety-invariant errors quarantine and reconcile the lane;
/// permanent provider or lane-local errors quarantine and reconcile only that lane;
/// shared-store failures still stop the executor rather than abandoning ownership.
async fn worker_loop(
    store: Arc<Store>,
    signer: ManagedSigner,
    config: ExecutorConfig,
    cancellation: CancellationToken,
    health: Arc<RuntimeHealth>,
) -> Result<(), ExecutorError> {
    const MIN_IDLE_FALLBACK: Duration = Duration::from_secs(5);
    let fallback = config.poll_interval.max(MIN_IDLE_FALLBACK);
    loop {
        health.heartbeat(RuntimeTask::Executor);
        match store.reserve_job(&signer.record()).await {
            Ok(Some(job)) => {
                let job_id = job.job_id;
                newton_metric::set_signer_ready(signer.chain_id, signer.signer_id.as_str(), false);
                let span = submission_job_span(&signer, &job);
                let execution = async {
                    log_job_reserved(&job);
                    execute_job(&store, &signer, &config, job, cancellation.clone()).await
                }
                .instrument(span)
                .await;
                match execution {
                    Ok(()) => {
                        health.record_rpc(signer.chain_id, Ok(()));
                        newton_metric::set_signer_ready(signer.chain_id, signer.signer_id.as_str(), true);
                    }
                    Err(error) if error.is_retryable() => {
                        record_backend_error(&health, signer.chain_id, &error);
                        warn!(
                            signer_id = %signer.signer_id,
                            signer_address = %signer.address(),
                            signer_role = signer.role.as_str(),
                            chain_id = signer.chain_id,
                            %job_id,
                            %error,
                            "retryable job execution failure"
                        );
                        // Unsigned transient failures release their reservation.
                        // Pace the next reservation so a healthy nonce read does
                        // not turn one flaky preflight/RPC into a hot retry loop.
                        if error.is_transient() && sleep_or_cancel(config.poll_interval, &cancellation).await {
                            return Ok(());
                        }
                        if recover_lane(&store, &signer, &config, &cancellation, &health).await?
                            == LaneRecovery::Stopped
                        {
                            return Ok(());
                        }
                    }
                    Err(error) if error.requires_quarantine() => {
                        record_backend_error(&health, signer.chain_id, &error);
                        error!(
                            signer_id = %signer.signer_id,
                            signer_address = %signer.address(),
                            signer_role = signer.role.as_str(),
                            chain_id = signer.chain_id,
                            %job_id,
                            %error,
                            "permanent job failure; quarantining only this signer lane"
                        );
                        if !error.assignment_already_finalized() {
                            match store
                                .release_safe_reservation(&signer.signer_id, signer.chain_id, job_id)
                                .await
                            {
                                Ok(()) | Err(StoreError::UnsafeRelease) => {}
                                Err(store_error) => return Err(store_error.into()),
                            }
                        }
                        store
                            .quarantine_signer(&signer.signer_id, signer.chain_id, &error.to_string())
                            .await?;
                        newton_metric::set_signer_ready(signer.chain_id, signer.signer_id.as_str(), false);
                        if recover_lane(&store, &signer, &config, &cancellation, &health).await?
                            == LaneRecovery::Stopped
                        {
                            return Ok(());
                        }
                    }
                    Err(error) => {
                        record_backend_error(&health, signer.chain_id, &error);
                        return Err(error);
                    }
                }
            }
            Ok(None) => {
                tokio::select! {
                    _ = cancellation.cancelled() => return Ok(()),
                    _ = store.wait_for_job(fallback) => {}
                }
            }
            Err(error) => {
                return Err(error.into());
            }
        }
    }
}

/// Creates the correlation boundary inherited by preparation, broadcast,
/// receipt tracking, replacement, and final outcome projection logs.
///
/// `submission_ids` and `task_ids` are stamped here rather than on individual
/// lines so every event from reservation through finality carries the producer
/// identities. One search for a task ID therefore returns the admission line
/// (which logs the same identity) and the whole on-chain trace, without first
/// resolving the task to a job. Transporter jobs carry no task batch and render
/// both fields empty.
fn submission_job_span(signer: &ManagedSigner, job: &JobRecord) -> Span {
    let items = job.intent.task_items();
    info_span!(
        "submission_job",
        job_id = %job.job_id,
        chain_id = job.chain_id,
        signer_id = %signer.signer_id,
        signer_address = %signer.address(),
        signer_role = job.signer_role.as_str(),
        intent_kind = job.intent.kind(),
        item_count = job.intent.item_count(),
        contract_role = %job.contract_role,
        submission_ids = %submission_ids_field(items),
        task_ids = %task_ids_field(items),
    )
}

/// Emits intent-specific identifiers once, when a signer lane owns the job.
///
/// Task batch membership already rides on the enclosing span, so this only
/// marks the point at which the lane took ownership.
fn log_job_reserved(job: &JobRecord) {
    match &job.intent {
        ExecutableIntent::BatchCreateAndRespond { items, .. } | ExecutableIntent::BatchRespond { items, .. } => {
            info!(submission_count = items.len(), "task batch reserved for submission");
        }
        ExecutableIntent::ConfirmGlobalTableRoot {
            root,
            reference_timestamp,
            reference_block_number,
            ..
        } => {
            info!(
                %root,
                reference_timestamp,
                reference_block_number,
                "transporter root confirmation reserved for submission"
            );
        }
        ExecutableIntent::UpdateOperatorTable {
            root,
            reference_timestamp,
            operator_set_index,
            expected_leaf,
            ..
        } => {
            info!(
                %root,
                reference_timestamp,
                operator_set_index,
                %expected_leaf,
                "transporter table update reserved for submission"
            );
        }
    }
}

/// Records only provider-facing failures against runtime RPC health.
///
/// Store, classification, and local-invariant failures describe executor state and
/// must not make a healthy chain provider appear unavailable.
fn record_backend_error(health: &RuntimeHealth, chain_id: u64, error: &ExecutorError) {
    if matches!(
        error,
        ExecutorError::Backend(_) | ExecutorError::UnderfundedBalanceRead(_)
    ) {
        let message = error.to_string();
        health.record_rpc(chain_id, Err(&message));
    }
}

/// Moves one unsigned reservation into durable transaction tracking.
///
/// The lane first reads `latest` and `pending`; disagreement releases the still-safe
/// unsigned reservation. It then prepares the transaction, classifies simulation
/// failures per batch item, and checks the maximum gas cost. Insufficient balance
/// releases the job for another signer and marks this lane underfunded.
///
/// The ownership boundary is [`Store::record_prepared`]: signed bytes, their hash,
/// and nonce are committed before any provider sees them. After that point the job
/// cannot be reassigned merely because broadcast fails. The exact committed bytes
/// remain recoverable and [`track_attempt`] owns receipt, replacement, cancellation,
/// finality, and outcome projection.
async fn execute_job(
    store: &Store,
    signer: &ManagedSigner,
    config: &ExecutorConfig,
    job: newton_submission_service::JobRecord,
    cancellation: CancellationToken,
) -> Result<(), ExecutorError> {
    let latest = match signer.backend.latest_transaction_count().await {
        Ok(latest) => latest,
        Err(error) if error.is_transient() => {
            store
                .release_safe_reservation(&signer.signer_id, signer.chain_id, job.job_id)
                .await?;
            return Err(error.into());
        }
        Err(error) => return Err(error.into()),
    };
    let pending = match signer.backend.pending_transaction_count().await {
        Ok(pending) => pending,
        Err(error) if error.is_transient() => {
            store
                .release_safe_reservation(&signer.signer_id, signer.chain_id, job.job_id)
                .await?;
            return Err(error.into());
        }
        Err(error) => return Err(error.into()),
    };
    newton_metric::set_signer_nonce_gap(signer.chain_id, signer.signer_id.as_str(), latest, pending);
    if latest != pending {
        store
            .release_safe_reservation(&signer.signer_id, signer.chain_id, job.job_id)
            .await?;
        return Err(ExecutorError::UnownedPendingTransaction {
            signer: signer.backend.address(),
            latest,
            pending,
        });
    }
    // The signer has no owned transaction in flight, so the next task nonce is
    // exactly the mined transaction count. We never increment or cache it.
    let signed = match signer.backend.prepare(&job.intent, latest, None).await {
        Ok(signed) => signed,
        Err(error) if error.is_transient() => {
            store
                .release_safe_reservation(&signer.signer_id, signer.chain_id, job.job_id)
                .await?;
            return Err(error.into());
        }
        Err(error @ BackendError::Simulation(_)) => {
            classify_preflight_failure(store, signer, &job, latest, error).await?;
            return Ok(());
        }
        Err(error) => {
            finalize_preflight_failure(store, &job, &error).await?;
            return Err(ExecutorError::FinalizedPreflight(error));
        }
    };
    if signed.nonce != latest {
        return Err(ExecutorError::AttemptNonceMismatch);
    }
    let required = maximum_transaction_cost(&signed);
    let balance = match signer.backend.balance().await {
        Ok(balance) => balance,
        Err(error) if error.is_transient() => {
            store
                .release_safe_reservation(&signer.signer_id, signer.chain_id, job.job_id)
                .await?;
            return Err(error.into());
        }
        Err(error) => return Err(error.into()),
    };
    if balance < required {
        store
            .release_underfunded_reservation(&signer.signer_id, signer.chain_id, job.job_id, balance, required)
            .await?;
        return Err(ExecutorError::InsufficientSignerBalance { balance, required });
    }
    let attempt_number = store
        .record_prepared(&PreparedAttempt {
            job_id: job.job_id,
            signer_id: signer.signer_id.clone(),
            signer_address: signer.backend.address(),
            chain_id: signer.chain_id,
            nonce: signed.nonce,
            gas_limit: signed.gas_limit,
            max_fee_per_gas: signed.max_fee_per_gas,
            max_priority_fee_per_gas: signed.max_priority_fee_per_gas,
            raw_transaction: signed.raw_transaction.clone(),
            transaction_hash: signed.transaction_hash,
            kind: AttemptKind::Canonical,
            replaces_attempt_number: None,
        })
        .await?;
    info!(
        attempt_number,
        attempt_kind = "canonical",
        nonce = signed.nonce,
        tx_hash = %signed.transaction_hash,
        gas_limit = signed.gas_limit,
        max_fee_per_gas = signed.max_fee_per_gas,
        max_priority_fee_per_gas = signed.max_priority_fee_per_gas,
        "transaction prepared and durably recorded"
    );
    let results = signer.backend.broadcast(&signed.raw_transaction).await;
    let any_success = results.iter().any(|result| result.result.is_ok());
    let accepted_provider = results
        .iter()
        .find(|result| result.result.is_ok())
        .map(|result| result.provider.as_str());
    let encoded_results = rmp_serde::to_vec(&results)?;
    store
        .record_broadcast(job.job_id, attempt_number, &encoded_results)
        .await?;
    let broadcast_at_ms = unix_time_ms();
    if any_success {
        info!(
            attempt_number,
            attempt_kind = "canonical",
            nonce = signed.nonce,
            tx_hash = %signed.transaction_hash,
            accepted_provider = accepted_provider.unwrap_or("unknown"),
            provider_attempts = results.len(),
            "transaction broadcast accepted"
        );
    } else {
        warn!(
            job_id = %job.job_id,
            tx_hash = %signed.transaction_hash,
            "all providers rejected broadcast; exact bytes remain recoverable"
        );
    }

    track_attempt(
        store,
        signer,
        config,
        &job,
        vec![AttemptRecord {
            attempt_number,
            nonce: signed.nonce,
            gas_limit: signed.gas_limit,
            max_fee_per_gas: signed.max_fee_per_gas,
            max_priority_fee_per_gas: signed.max_priority_fee_per_gas,
            raw_transaction: signed.raw_transaction,
            transaction_hash: signed.transaction_hash,
            kind: AttemptKind::Canonical,
            state: AttemptState::Broadcast,
            receipt_provider: None,
            updated_at_ms: broadcast_at_ms,
        }],
        cancellation,
    )
    .await
}

/// Resumes an assignment that already owns one or more signed attempts.
///
/// Every attempt must use the same nonce. If `latest` has advanced beyond that nonce,
/// a known receipt resumes finality tracking; without a known receipt, authoritative
/// task-manager effects determine the item outcomes. If the provider's `pending`
/// count is behind the journaled nonce, the lane waits for a consistent provider.
/// Otherwise a merely prepared attempt is broadcast once; an attempt already
/// journaled as broadcast resumes observation without creating a restart burst.
async fn recover_prepared_assignment(
    store: &Store,
    signer: &ManagedSigner,
    config: &ExecutorConfig,
    mut assignment: ActiveAssignment,
    latest: u64,
    pending: u64,
    cancellation: CancellationToken,
) -> Result<(), ExecutorError> {
    let attempt = assignment
        .attempts
        .last()
        .cloned()
        .ok_or(ExecutorError::MissingAttempt)?;
    if assignment
        .attempts
        .iter()
        .any(|candidate| candidate.nonce != attempt.nonce)
    {
        return Err(ExecutorError::AttemptNonceMismatch);
    }
    if latest > attempt.nonce {
        let mut known_receipt = false;
        for candidate in assignment.attempts.iter().rev() {
            match signer
                .backend
                .observe(
                    candidate.transaction_hash,
                    candidate.receipt_provider.as_deref(),
                    ObservationScope::Exhaustive,
                )
                .await?
            {
                TransactionObservation::Pending | TransactionObservation::PendingInconclusive { .. } => {}
                TransactionObservation::Mined { .. } | TransactionObservation::Reverted { .. } => {
                    known_receipt = true;
                    break;
                }
            }
        }
        if known_receipt {
            return track_attempt(
                store,
                signer,
                config,
                &assignment.job,
                assignment.attempts,
                cancellation,
            )
            .await;
        }
        warn!(
            signer_id = %signer.signer_id,
            job_id = %assignment.job.job_id,
            nonce = attempt.nonce,
            latest,
            "chain consumed the persisted nonce without a known receipt; reconciling task effects"
        );
        return classify_and_project(store, signer, &assignment.job, None).await;
    }
    if pending < attempt.nonce {
        return Err(ExecutorError::NonceJournalAhead {
            journal: attempt.nonce,
            pending,
        });
    }
    // Only a prepared attempt is known not to have crossed the broadcast
    // boundary. A previously broadcast attempt keeps its durable watchdog age;
    // restarting the process must not create an immediate rebroadcast burst or
    // postpone replacement forever.
    if attempt.state == AttemptState::Prepared {
        let results = signer.backend.broadcast(&attempt.raw_transaction).await;
        store
            .record_broadcast(
                assignment.job.job_id,
                attempt.attempt_number,
                &rmp_serde::to_vec(&results)?,
            )
            .await?;
        let accepted_provider = results
            .iter()
            .find(|result| result.result.is_ok())
            .map(|result| result.provider.as_str());
        if let Some(accepted_provider) = accepted_provider {
            info!(
                attempt_number = attempt.attempt_number,
                attempt_kind = attempt.kind.as_str(),
                nonce = attempt.nonce,
                tx_hash = %attempt.transaction_hash,
                accepted_provider,
                provider_attempts = results.len(),
                "recovered prepared transaction broadcast accepted"
            );
        } else {
            warn!(
                attempt_number = attempt.attempt_number,
                attempt_kind = attempt.kind.as_str(),
                nonce = attempt.nonce,
                tx_hash = %attempt.transaction_hash,
                provider_attempts = results.len(),
                "all providers rejected recovered transaction broadcast; exact bytes remain durable"
            );
        }
        if let Some(latest_attempt) = assignment.attempts.last_mut() {
            latest_attempt.state = AttemptState::Broadcast;
            latest_attempt.updated_at_ms = unix_time_ms();
        }
    }
    track_attempt(
        store,
        signer,
        config,
        &assignment.job,
        assignment.attempts,
        cancellation,
    )
    .await
}

/// Waits for a retry delay while remaining responsive to shutdown.
///
/// Returns `true` when cancellation won the race and `false` when the delay elapsed.
async fn sleep_or_cancel(duration: Duration, cancellation: &CancellationToken) -> bool {
    tokio::select! {
        _ = cancellation.cancelled() => true,
        _ = tokio::time::sleep(duration) => false,
    }
}

fn unix_time_ms() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
        .try_into()
        .unwrap_or(i64::MAX)
}

/// Executor startup or worker error.
#[derive(Debug, thiserror::Error)]
pub enum ExecutorError {
    /// Durable store.
    #[error(transparent)]
    Store(#[from] newton_submission_service::StoreError),
    /// Chain backend.
    #[error(transparent)]
    Backend(#[from] BackendError),
    /// Attempt result encoding.
    #[error(transparent)]
    Encode(#[from] rmp_serde::encode::Error),
    /// Pending activity exists without a durable assignment owned by this service.
    #[error("signer {signer} has unowned pending activity: latest={latest}, pending={pending}")]
    UnownedPendingTransaction {
        /// Sender.
        signer: Address,
        /// Mined count.
        latest: u64,
        /// Pending count.
        pending: u64,
    },
    /// A durable assignment unexpectedly had no attempt.
    #[error("prepared signer assignment has no durable transaction attempt")]
    MissingAttempt,
    /// Attempts for one canonical job disagreed on nonce.
    #[error("durable attempts for one job use different nonces")]
    AttemptNonceMismatch,
    /// The durable journal is ahead of the provider's pending transaction count.
    #[error("journaled nonce {journal} is ahead of pending transaction count {pending}")]
    NonceJournalAhead {
        /// Journaled nonce.
        journal: u64,
        /// Provider count.
        pending: u64,
    },
    /// Chain evidence omitted a batch member.
    #[error("on-chain effect classification did not cover every batch item")]
    EffectCoverage,
    /// An unsigned signer cannot currently cover the transaction's maximum cost.
    #[error("signer native balance {balance} is below required transaction cost {required}")]
    InsufficientSignerBalance {
        /// Current native balance.
        balance: U256,
        /// Gas limit multiplied by max fee per gas.
        required: U256,
    },
    /// A periodic underfunded-lane balance read failed.
    #[error("failed to recheck underfunded signer balance: {0}")]
    UnderfundedBalanceRead(BackendError),
    /// An underfunded state was missing its persisted recovery threshold.
    #[error("underfunded signer is missing its required balance threshold")]
    MissingRequiredBalance,
    /// A signer worker exited while the executor was still running.
    #[error("signer worker exited unexpectedly")]
    WorkerExited,
    /// A signer worker panicked.
    #[error("signer worker panicked: {0}")]
    WorkerPanicked(String),
    /// A permanent preflight error was projected to the job before lane quarantine.
    #[error("job finalized after permanent preflight failure: {0}")]
    FinalizedPreflight(BackendError),
}

impl ExecutorError {
    /// Whether the complete executor must stop rather than isolating one lane.
    fn is_process_fatal(&self) -> bool {
        matches!(self, Self::Store(_) | Self::WorkerExited | Self::WorkerPanicked(_))
    }

    /// Whether retrying may succeed without changing durable or on-chain state.
    /// Whether this failure is worth retrying rather than failing the lane.
    ///
    /// Covers both directions the executor can be blocked by weather rather
    /// than by a defect: a flaky RPC provider (`Backend`) and a contended
    /// SQLite store (`Store`). The store arm matters because the submitter's
    /// supervisor does not restart a failed required task -- it exits the
    /// process -- so a moment of WAL write contention used to take the whole
    /// service down. The retry sites already pace themselves with
    /// `sleep_or_cancel`, so widening the classification is enough.
    fn is_transient(&self) -> bool {
        match self {
            Self::Backend(error) => error.is_transient(),
            Self::Store(error) => error.is_transient(),
            _ => false,
        }
    }

    /// Whether a transient failure came from the chain provider rather than the
    /// durable store.
    ///
    /// The recovery path records provider health on every transient failure. A
    /// store failure says nothing about the RPC provider, and reporting one as
    /// an RPC failure would corrupt `submission_rpc_provider_healthy`, so the
    /// caller uses this to decide whether the RPC health signal applies.
    fn is_provider_failure(&self) -> bool {
        matches!(self, Self::Backend(_))
    }

    /// Whether authoritative chain state must converge before this lane proceeds.
    fn waits_for_chain(&self) -> bool {
        matches!(
            self,
            Self::UnownedPendingTransaction { .. } | Self::NonceJournalAhead { .. } | Self::EffectCoverage
        )
    }

    /// Whether this lane should wait for its persisted funding threshold.
    fn is_underfunded(&self) -> bool {
        matches!(
            self,
            Self::InsufficientSignerBalance { .. } | Self::UnderfundedBalanceRead(_)
        )
    }

    /// Whether the lane should enter reconciliation instead of failing its worker.
    fn is_retryable(&self) -> bool {
        self.is_transient() || self.waits_for_chain() || self.is_underfunded()
    }

    /// Whether local durable-state invariants require isolating this nonce lane.
    fn requires_quarantine(&self) -> bool {
        !self.is_process_fatal() && !self.is_retryable()
    }

    /// Whether job ownership was already released by terminal outcome projection.
    fn assignment_already_finalized(&self) -> bool {
        matches!(self, Self::FinalizedPreflight(_))
    }
}

#[cfg(test)]
mod tests;