thruput-node-supervisor 0.43.0

use std::collections::HashMap;
use std::sync::{
    Arc, Mutex,
    atomic::{AtomicU64, Ordering},
};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use jsonrpsee::core::RpcResult;
use thruput_json_rpc_server::{
    ChainPlane, RpcRequestContext, RpcServerError, ValidatedTransaction,
    native::transaction::NativeTransactionReadPlane, tx::validate_raw_transaction_bytes,
};
use thruput_json_rpc_types::{
    ContractAddress,
    airdrop::{
        AirdropAttemptResponse, AirdropRequest, FundAccountRequest, FundAccountResponse,
        LocalFundingErrorData,
    },
    base58::Base58,
    receipt::ExecStatus,
    runtime_manifest::{BOOTSTRAP_ACCOUNT_ROLE_FAUCET, RuntimeManifest},
    transaction::{SchedulerTransactionError, TransactionStatus, TxHash},
};
use thruput_signing::{
    DEFAULT_TX_GAS_LIMIT, LocalSigner, PubkeyAccount, SignTransactionParams,
    encode_signed_transaction_xdr, sign_transaction, transaction_hash,
};
use thruput_tx_validation::NetworkConfigParams;
use tokio::sync::Semaphore;

use crate::{
    LocalAirdropConfig, NodeSupervisorConfig,
    block_driver::BlockDriverTrigger,
    funding_journal::{FundingJournal, FundingRecord, RetainedFundingTransaction},
    request_deadline::within_request_deadline,
};

const DEFAULT_BLOCK_EXPIRY_OFFSET: i64 = 1_000;
const LOCAL_AIRDROP_RETRY_AFTER_SECONDS: u64 = 1;

/// Admits one locally signed funding transaction to the local scheduler.
#[async_trait]
pub(crate) trait FundingSubmitPlane: Send + Sync + 'static {
    async fn submit_funding(
        &self,
        context: RpcRequestContext,
        tx: ValidatedTransaction,
    ) -> Result<(), RpcServerError>;
}

/// Status the local faucet uses to reconcile a retained funding attempt.
#[async_trait]
pub(crate) trait FundingStatusPlane: Send + Sync + 'static {
    async fn funding_status(
        &self,
        context: RpcRequestContext,
        tx_hash: TxHash,
    ) -> Result<TransactionStatus, RpcServerError>;
}

#[derive(Clone)]
pub(crate) struct LocalAirdropContext {
    network_hash: thruput_xdr::types::NetworkHash,
    signer: Arc<LocalSigner>,
    config: LocalAirdropConfig,
    journal: Arc<Mutex<FundingJournal>>,
    lock_lease: Option<Arc<std::fs::File>>,
}

impl LocalAirdropContext {
    pub(crate) fn from_validated_runtime(
        manifest: &RuntimeManifest,
        config: &NodeSupervisorConfig,
        signer: Arc<LocalSigner>,
        journal: Arc<Mutex<FundingJournal>>,
        lock_lease: Option<Arc<std::fs::File>>,
    ) -> Result<Self, String> {
        let mut faucets = manifest
            .bootstrap_accounts
            .iter()
            .filter(|account| account.role == BOOTSTRAP_ACCOUNT_ROLE_FAUCET);
        let faucet = faucets
            .next()
            .ok_or_else(|| "runtime manifest is missing its local faucet".to_owned())?;
        if faucets.next().is_some() {
            return Err("runtime manifest has multiple local faucets".into());
        }
        if faucet.public_key.as_ref() != signer.verifying_key_bytes() {
            return Err("runtime faucet public key does not match its private credential".into());
        }
        let account = PubkeyAccount::from_public_key_bytes(signer.verifying_key_bytes(), 0);
        if faucet.wallet_address.0 != account.address() {
            return Err("runtime faucet address is not its nonce-zero native account".into());
        }
        Ok(Self {
            network_hash: manifest.network_hash.to_network_hash(),
            signer,
            config: config.local_airdrop,
            journal,
            lock_lease,
        })
    }
}

#[derive(Clone)]
pub(crate) struct LocalAirdropRpc {
    backend: Arc<dyn AirdropBackend>,
    builder: Arc<dyn AirdropTransactionBuilder>,
    attempts: Arc<Mutex<HashMap<AttemptKey, Arc<Attempt>>>>,
    capacity: usize,
    concurrency: Arc<Semaphore>,
    admission: Arc<Semaphore>,
    persistence: Arc<Semaphore>,
    timeout: Duration,
    raw_amount: u128,
    faucet: ContractAddress,
    journal: Arc<Mutex<FundingJournal>>,
    lock_lease: Option<Arc<std::fs::File>>,
    block_driver: BlockDriverTrigger,
}

impl LocalAirdropRpc {
    pub(crate) fn new<S, R>(
        context: LocalAirdropContext,
        submit_chain: S,
        read: R,
        block_driver: BlockDriverTrigger,
    ) -> Self
    where
        S: FundingSubmitPlane + ChainPlane,
        R: FundingStatusPlane + NativeTransactionReadPlane,
    {
        let faucet = Base58(
            PubkeyAccount::from_public_key_bytes(context.signer.verifying_key_bytes(), 0).address(),
        );
        let builder = NativeAirdropBuilder {
            network_hash: context.network_hash,
            signer: context.signer,
            next_nonce: AtomicU64::new(nonce_from_timestamp()),
        };
        let mut rpc = Self::with_parts(
            Arc::new(PlaneAirdropBackend { submit_chain, read }),
            Arc::new(builder),
            context.config,
            block_driver,
        );
        rpc.faucet = faucet;
        rpc.journal = context.journal;
        rpc.lock_lease = context.lock_lease;
        rpc
    }

    fn with_parts(
        backend: Arc<dyn AirdropBackend>,
        builder: Arc<dyn AirdropTransactionBuilder>,
        config: LocalAirdropConfig,
        block_driver: BlockDriverTrigger,
    ) -> Self {
        Self {
            backend,
            builder,
            attempts: Arc::new(Mutex::new(HashMap::new())),
            capacity: config.attempt_capacity,
            concurrency: Arc::new(Semaphore::new(config.max_concurrency)),
            // Allow a finite burst of 16 execution waves across all callers, including
            // same-key observers. Execution concurrency remains independently bounded.
            admission: Arc::new(Semaphore::new(
                config
                    .max_concurrency
                    .saturating_mul(16)
                    .min(Semaphore::MAX_PERMITS),
            )),
            persistence: Arc::new(Semaphore::new(config.max_concurrency)),
            lock_lease: None,
            timeout: config.request_timeout,
            raw_amount: config.raw_amount,
            faucet: Base58(thruput_xdr::types::ContractAddress::from([0; 32])),
            journal: Arc::new(Mutex::new(FundingJournal::memory(
                [0; 32].into(),
                Base58(thruput_xdr::types::ContractAddress::from([0; 32])),
            ))),
            block_driver,
        }
    }

    async fn request(
        &self,
        request: AirdropRequest,
    ) -> Result<AirdropAttemptResponse, RpcServerError> {
        let context = RpcRequestContext::new(Some(self.timeout));
        let _admission = self.admission.try_acquire().map_err(|_| {
            RpcServerError::Overloaded("local airdrop admission capacity reached".into())
        })?;
        within_request_deadline(context, self.request_admitted(request, context)).await
    }

    async fn request_admitted(
        &self,
        request: AirdropRequest,
        context: RpcRequestContext,
    ) -> Result<AirdropAttemptResponse, RpcServerError> {
        let attempt = self
            .find_or_admit(FundAccountRequest {
                recipient: request.recipient,
                raw_amount: self.raw_amount.to_string(),
                client_nonce: request.client_nonce.clone(),
                faucet: self.faucet,
            })
            .await?;
        let observed_revision = attempt.revision()?;
        let _operation = attempt.operation.lock().await;

        if attempt.revision()? != observed_revision {
            return attempt.last_result()?.ok_or_else(|| {
                RpcServerError::BackendUnavailable(
                    "local airdrop attempt completed without an outcome".into(),
                )
            })?;
        }
        if attempt.is_terminal()? {
            return attempt.last_result()?.ok_or_else(|| {
                RpcServerError::BackendUnavailable(
                    "terminal local airdrop attempt has no outcome".into(),
                )
            })?;
        }

        let permit = self.concurrency.acquire().await.map_err(|_| {
            RpcServerError::BackendUnavailable("local airdrop execution is closed".into())
        })?;
        if context
            .remaining()
            .is_some_and(|remaining| remaining.is_zero())
        {
            return Err(RpcServerError::BackendTimeout);
        }
        let result = self
            .advance_attempt(&attempt, request.recipient, context)
            .await;
        drop(permit);
        attempt.record_result(result.clone());
        result
    }

    async fn find_or_admit(
        &self,
        request: FundAccountRequest,
    ) -> Result<Arc<Attempt>, RpcServerError> {
        let permit = self
            .persistence
            .clone()
            .acquire_owned()
            .await
            .map_err(|_| {
                RpcServerError::BackendUnavailable("funding persistence is closed".into())
            })?;
        let rpc = self.clone();
        tokio::task::spawn_blocking(move || {
            let _permit = permit;
            rpc.register_attempt(request)
        })
        .await
        .map_err(|_| {
            RpcServerError::BackendUnavailable("funding persistence worker stopped".into())
        })?
    }

    fn register_attempt(
        &self,
        request: FundAccountRequest,
    ) -> Result<Arc<Attempt>, RpcServerError> {
        let key = AttemptKey::funding(&request);
        let mut attempts = self.attempts.lock().map_err(|_| attempt_state_poisoned())?;
        let mut journal = self.journal.lock().map_err(|_| attempt_state_poisoned())?;
        let retained = journal
            .records()
            .iter()
            .any(|record| AttemptKey::funding(&record.request) == key);
        if !retained && journal.records().len() >= self.capacity {
            return Err(RpcServerError::Overloaded(
                "local funding attempt capacity reached; existing requests can still reconcile"
                    .into(),
            ));
        }
        let record = journal.register(request)?;
        if let Some(attempt) = attempts.get(&key) {
            return Ok(attempt.clone());
        }
        let attempt = Arc::new(Attempt::from_record(record));
        attempts.insert(key, attempt.clone());
        Ok(attempt)
    }

    async fn retain_signed(
        &self,
        attempt: &Attempt,
        signed: &SignedAirdrop,
    ) -> Result<(), RpcServerError> {
        let permit = self
            .persistence
            .clone()
            .acquire_owned()
            .await
            .map_err(|_| {
                RpcServerError::BackendUnavailable("funding persistence is closed".into())
            })?;
        let lock_lease = self.lock_lease.clone();
        let journal = self.journal.clone();
        let request = attempt.request.clone();
        let retained = RetainedFundingTransaction {
            tx_hash: signed.tx_hash,
            signed_transaction: thruput_json_rpc_types::hex::HexBytes(signed.bytes.clone()),
        };
        tokio::task::spawn_blocking(move || {
            let _permit = permit;
            let _lock_lease = lock_lease;
            journal
                .lock()
                .map_err(|_| attempt_state_poisoned())?
                .retain_signed(&request, retained)
        })
        .await
        .map_err(|_| {
            RpcServerError::BackendUnavailable("funding persistence worker stopped".into())
        })?
    }

    async fn prepare_signed(
        &self,
        attempt: &Attempt,
        context: RpcRequestContext,
    ) -> Result<SignedAirdrop, RpcServerError> {
        if attempt.signed()?.is_none() {
            let tip = self.backend.chain_tip(context).await?;
            let signed = self.builder.build(
                attempt.request.recipient,
                (*tip.height).saturating_add(DEFAULT_BLOCK_EXPIRY_OFFSET),
                attempt
                    .request
                    .amount()
                    .map_err(|e| RpcServerError::InvalidParam(e.into()))?,
            )?;
            attempt.store_signed(signed)?;
        }
        let signed = attempt.signed()?.ok_or_else(|| {
            RpcServerError::BackendUnavailable("local funding transaction was not retained".into())
        })?;
        self.retain_signed(attempt, &signed).await?;
        Ok(signed)
    }

    async fn observed_receipt(
        &self,
        context: RpcRequestContext,
        hash: TxHash,
    ) -> Result<Option<thruput_json_rpc_types::receipt::TransactionReceipt>, RpcServerError> {
        let receipt = self.backend.receipt(context, hash).await?;
        if receipt.as_ref().is_some_and(|receipt| {
            receipt.tx_hash != hash
                || *receipt.height < 0
                || !receipt.has_consistent_execution_output()
        }) {
            return Err(RpcServerError::invalid_data(
                thruput_json_rpc_types::error::native::BackendDataInvalidReason::InconsistentResult,
                "funding receipt identity, height or execution output is invalid",
            ));
        }
        Ok(receipt)
    }

    async fn fund(
        &self,
        request: FundAccountRequest,
    ) -> Result<FundAccountResponse, RpcServerError> {
        request
            .amount()
            .map_err(|e| RpcServerError::InvalidParam(e.into()))?;
        if request.faucet != self.faucet {
            return Err(RpcServerError::InvalidParam(
                "expected faucet belongs to a different local chain".into(),
            ));
        }
        let _admission = self.admission.try_acquire().map_err(|_| {
            RpcServerError::Overloaded("local funding admission capacity reached".into())
        })?;
        let context = RpcRequestContext::new(Some(self.timeout));
        within_request_deadline(context, self.fund_admitted(request.clone(), context))
            .await
            .map_err(|error| {
                // Do not extend the request deadline waiting for a persistence worker.
                // A retry can recover evidence if that worker still holds the journal.
                let evidence = self
                    .journal
                    .try_lock()
                    .ok()
                    .and_then(|journal| journal.durable_signed(&request));
                match evidence {
                    Some(retained) => RpcServerError::LocalFundingUnresolved {
                        reason: error.to_string(),
                        cause: Some(Box::new(error)),
                        evidence: LocalFundingErrorData::new(
                            retained.tx_hash,
                            retained.signed_transaction,
                        ),
                    },
                    None => error,
                }
            })
    }

    async fn fund_admitted(
        &self,
        request: FundAccountRequest,
        context: RpcRequestContext,
    ) -> Result<FundAccountResponse, RpcServerError> {
        let attempt = self.find_or_admit(request).await?;
        let _operation = attempt.operation.lock().await;
        let _permit =
            self.concurrency.acquire().await.map_err(|_| {
                RpcServerError::BackendUnavailable("local funding is closed".into())
            })?;
        let replay = attempt.signed()?.is_some();
        let signed = self.prepare_signed(&attempt, context).await?;
        let response = || {
            FundAccountResponse::new(
                attempt.request.recipient,
                attempt.request.raw_amount.clone(),
                signed.tx_hash,
                thruput_json_rpc_types::hex::HexBytes(signed.bytes.clone()),
            )
        };
        if replay {
            if self
                .observed_receipt(context, signed.tx_hash)
                .await?
                .is_some()
            {
                return Ok(response());
            }
            match self.backend.status(context, signed.tx_hash).await? {
                TransactionStatus::Pending | TransactionStatus::Committed { .. } => {
                    return Ok(response());
                }
                TransactionStatus::Invalid { reason } => {
                    return Err(RpcServerError::TxRejected(reason));
                }
                TransactionStatus::Dropped { error } => {
                    return Err(dropped_funding_error(&error));
                }
                TransactionStatus::Unknown
                | TransactionStatus::Retry { .. }
                | TransactionStatus::RecentFailure { .. }
                | TransactionStatus::Other => {}
                _ => {
                    return Err(RpcServerError::BackendUnavailable(
                        "unrecognized retained funding status".into(),
                    ));
                }
            }
            self.check_retained_expiry(&signed, context).await?;
        }
        match self.backend.submit(context, &signed).await {
            Ok(()) => {
                attempt.set_submitted(true)?;
                self.block_driver.trigger_after_submit();
                Ok(response())
            }
            Err(error) => {
                // A request-scoped admission error must not overwrite an older
                // accepted or committed copy of this same retained transaction.
                if self
                    .observed_receipt(context, signed.tx_hash)
                    .await?
                    .is_some()
                {
                    return Ok(response());
                }
                match self.backend.status(context, signed.tx_hash).await? {
                    TransactionStatus::Pending | TransactionStatus::Committed { .. } => {
                        self.block_driver.trigger_after_submit();
                        Ok(response())
                    }
                    _ => Err(classify_backend_error(error)),
                }
            }
        }
    }

    async fn check_retained_expiry(
        &self,
        signed: &SignedAirdrop,
        context: RpcRequestContext,
    ) -> Result<(), RpcServerError> {
        let tx =
            validate_raw_transaction_bytes(signed.bytes.clone(), &NetworkConfigParams::default())
                .map_err(|_| {
                RpcServerError::BackendUnavailable(
                    "retained funding transaction is malformed".into(),
                )
            })?;
        let tip = self.backend.chain_tip(context).await?;
        if *tx.block_expiry_height <= *tip.height {
            return Err(RpcServerError::BackendUnavailable("retained funding expired with no authoritative outcome; reconcile its hash instead of creating another transfer".into()));
        }
        Ok(())
    }

    async fn reconcile_restored_attempt(
        &self,
        attempt: &Attempt,
        signed: &SignedAirdrop,
        recipient: ContractAddress,
        context: RpcRequestContext,
    ) -> Result<Option<AirdropAttemptResponse>, RpcServerError> {
        if let Some(receipt) = self.observed_receipt(context, signed.tx_hash).await? {
            if matches!(receipt.status, ExecStatus::Ok) {
                return Ok(Some(AirdropAttemptResponse::Succeeded {
                    recipient,
                    tx_hash: signed.tx_hash,
                    raw_amount: self.raw_amount.to_string(),
                    height: receipt.height,
                    status: receipt.status,
                    return_data: receipt.return_data,
                }));
            }
            return Err(RpcServerError::TxRejected(format!(
                "local airdrop execution failed: {:?}",
                receipt.status
            )));
        }
        match self.backend.status(context, signed.tx_hash).await? {
            TransactionStatus::Pending | TransactionStatus::Committed { .. } => {
                return Ok(Some(self.pending_response(recipient, signed.tx_hash)));
            }
            TransactionStatus::Invalid { reason } => {
                return Err(RpcServerError::TxRejected(reason));
            }
            TransactionStatus::Dropped { error } => {
                attempt.set_terminal(true)?;
                return Err(dropped_funding_error(&error));
            }
            _ => self.check_retained_expiry(signed, context).await?,
        }
        Ok(None)
    }

    async fn advance_attempt(
        &self,
        attempt: &Attempt,
        recipient: ContractAddress,
        context: RpcRequestContext,
    ) -> Result<AirdropAttemptResponse, RpcServerError> {
        let signed = self.prepare_signed(attempt, context).await?;
        if attempt.restored
            && let Some(response) = self
                .reconcile_restored_attempt(attempt, &signed, recipient, context)
                .await?
        {
            return Ok(response);
        }

        let mut submit_error = if attempt.is_submitted()? {
            None
        } else {
            match self.backend.submit(context, &signed).await {
                Ok(()) => {
                    attempt.set_submitted(true)?;
                    None
                }
                Err(error) => Some(classify_backend_error(error)),
            }
        };
        if let Some(error) = submit_error.take_if(|error| is_deterministic_failure(error)) {
            attempt.set_terminal(true)?;
            return Err(error);
        }

        match self.observed_receipt(context, signed.tx_hash).await {
            Ok(Some(receipt)) if matches!(receipt.status, ExecStatus::Ok) => {
                attempt.set_terminal(true)?;
                Ok(AirdropAttemptResponse::Succeeded {
                    recipient,
                    tx_hash: signed.tx_hash,
                    raw_amount: self.raw_amount.to_string(),
                    height: receipt.height,
                    status: receipt.status,
                    return_data: receipt.return_data,
                })
            }
            Ok(Some(receipt)) => {
                attempt.set_terminal(true)?;
                Err(RpcServerError::TxRejected(format!(
                    "local airdrop execution failed: {:?}",
                    receipt.status
                )))
            }
            Ok(None) => match self.backend.status(context, signed.tx_hash).await {
                Ok(TransactionStatus::Invalid { reason }) => {
                    attempt.set_terminal(true)?;
                    Err(RpcServerError::TxRejected(reason))
                }
                Ok(TransactionStatus::Dropped { error }) => {
                    attempt.set_terminal(true)?;
                    Err(dropped_funding_error(&error))
                }
                Ok(TransactionStatus::Retry { .. }) => {
                    attempt.set_submitted(false)?;
                    Ok(self.pending_response(recipient, signed.tx_hash))
                }
                Ok(TransactionStatus::Pending) => {
                    if submit_error.is_some() {
                        self.block_driver.trigger_after_submit();
                    }
                    attempt.set_submitted(true)?;
                    Ok(self.pending_response(recipient, signed.tx_hash))
                }
                Ok(TransactionStatus::RecentFailure { .. }) => {
                    attempt.set_submitted(false)?;
                    Err(RpcServerError::BackendUnavailable(
                        "local airdrop has a recent nondeterministic failure and may still commit"
                            .into(),
                    ))
                }
                Ok(TransactionStatus::Unknown) => {
                    attempt.set_submitted(false)?;
                    match submit_error {
                        Some(error) => Err(error),
                        None => Ok(self.pending_response(recipient, signed.tx_hash)),
                    }
                }
                Ok(_) => match submit_error {
                    Some(error) => Err(error),
                    None => Ok(self.pending_response(recipient, signed.tx_hash)),
                },
                Err(error) => Err(submit_error.unwrap_or_else(|| classify_backend_error(error))),
            },
            Err(error) => Err(classify_backend_error(error)),
        }
    }

    fn pending_response(
        &self,
        recipient: ContractAddress,
        tx_hash: TxHash,
    ) -> AirdropAttemptResponse {
        AirdropAttemptResponse::Pending {
            recipient,
            raw_amount: self.raw_amount.to_string(),
            tx_hash: Some(tx_hash),
            retry_after_seconds: LOCAL_AIRDROP_RETRY_AFTER_SECONDS,
        }
    }
}

#[derive(Clone, Hash, PartialEq, Eq)]
struct AttemptKey {
    recipient: [u8; 32],
    client_nonce: Vec<u8>,
}

impl AttemptKey {
    fn funding(request: &FundAccountRequest) -> Self {
        Self {
            recipient: request.recipient.0.into_inner(),
            client_nonce: request.client_nonce.as_bytes().to_vec(),
        }
    }
}

struct Attempt {
    request: FundAccountRequest,
    restored: bool,
    operation: tokio::sync::Mutex<()>,
    state: Mutex<AttemptState>,
}

#[derive(Default)]
struct AttemptState {
    signed: Option<SignedAirdrop>,
    last_result: Option<Result<AirdropAttemptResponse, RpcServerError>>,
    revision: u64,
    terminal: bool,
    submitted: bool,
}

impl Attempt {
    fn from_record(record: FundingRecord) -> Self {
        let signed = record.signed.map(|signed| SignedAirdrop {
            bytes: signed.signed_transaction.0,
            tx_hash: signed.tx_hash,
        });
        Self {
            request: record.request,
            restored: signed.is_some(),
            operation: tokio::sync::Mutex::new(()),
            state: Mutex::new(AttemptState {
                signed,
                ..AttemptState::default()
            }),
        }
    }

    fn signed(&self) -> Result<Option<SignedAirdrop>, RpcServerError> {
        self.state
            .lock()
            .map(|state| state.signed.clone())
            .map_err(|_| attempt_state_poisoned())
    }

    fn store_signed(&self, signed: SignedAirdrop) -> Result<(), RpcServerError> {
        let mut state = self.state.lock().map_err(|_| attempt_state_poisoned())?;
        if state.signed.is_some() {
            return Err(RpcServerError::BackendUnavailable(
                "local airdrop transaction was constructed more than once".into(),
            ));
        }
        state.signed = Some(signed);
        Ok(())
    }

    fn revision(&self) -> Result<u64, RpcServerError> {
        self.state
            .lock()
            .map(|state| state.revision)
            .map_err(|_| attempt_state_poisoned())
    }

    fn is_terminal(&self) -> Result<bool, RpcServerError> {
        self.state
            .lock()
            .map(|state| state.terminal)
            .map_err(|_| attempt_state_poisoned())
    }

    fn set_terminal(&self, terminal: bool) -> Result<(), RpcServerError> {
        self.state
            .lock()
            .map(|mut state| state.terminal = terminal)
            .map_err(|_| attempt_state_poisoned())
    }

    fn is_submitted(&self) -> Result<bool, RpcServerError> {
        self.state
            .lock()
            .map(|state| state.submitted)
            .map_err(|_| attempt_state_poisoned())
    }

    fn set_submitted(&self, submitted: bool) -> Result<(), RpcServerError> {
        self.state
            .lock()
            .map(|mut state| state.submitted = submitted)
            .map_err(|_| attempt_state_poisoned())
    }

    fn last_result(
        &self,
    ) -> Result<Option<Result<AirdropAttemptResponse, RpcServerError>>, RpcServerError> {
        self.state
            .lock()
            .map(|state| state.last_result.clone())
            .map_err(|_| attempt_state_poisoned())
    }

    fn record_result(&self, result: Result<AirdropAttemptResponse, RpcServerError>) {
        match self.state.lock() {
            Ok(mut state) => {
                state.last_result = Some(result);
                state.revision = state.revision.wrapping_add(1);
            }
            Err(error) => tracing::error!(%error, "local airdrop outcome store is poisoned"),
        }
    }
}

fn attempt_state_poisoned() -> RpcServerError {
    RpcServerError::BackendUnavailable("local airdrop attempt state is poisoned".into())
}

#[derive(Clone)]
struct SignedAirdrop {
    bytes: Vec<u8>,
    tx_hash: TxHash,
}

trait AirdropTransactionBuilder: Send + Sync + 'static {
    fn build(
        &self,
        recipient: ContractAddress,
        block_expiry_height: i64,
        raw_amount: u128,
    ) -> Result<SignedAirdrop, RpcServerError>;
}

struct NativeAirdropBuilder {
    network_hash: thruput_xdr::types::NetworkHash,
    signer: Arc<LocalSigner>,
    next_nonce: AtomicU64,
}

/// Rebuild retained bytes from their immutable request and saved uniquifier.
/// This checks the transfer intent, network signature, and every transaction field.
pub(crate) fn validate_retained_funding(
    request: &FundAccountRequest,
    retained: &RetainedFundingTransaction,
    network_hash: thruput_xdr::types::NetworkHash,
    signer: Arc<LocalSigner>,
) -> Result<(), String> {
    let retained_tx: thruput_xdr::transaction::SignedTransaction =
        xdrs::txs::decode(&retained.signed_transaction.0)
            .map_err(|_| "retained funding transaction is malformed")?;
    let expected = NativeAirdropBuilder {
        network_hash,
        signer,
        next_nonce: AtomicU64::new(retained_tx.tx.uniquifier.nonce),
    }
    .build(
        request.recipient,
        *retained_tx.tx.uniquifier.block_expiry_height,
        request.amount()?,
    )
    .map_err(|_| "could not validate retained funding intent")?;
    if expected.tx_hash != retained.tx_hash || expected.bytes != retained.signed_transaction.0 {
        return Err(
            "retained funding transaction differs from its request or faucet signature".into(),
        );
    }
    Ok(())
}

impl AirdropTransactionBuilder for NativeAirdropBuilder {
    fn build(
        &self,
        recipient: ContractAddress,
        block_expiry_height: i64,
        raw_amount: u128,
    ) -> Result<SignedAirdrop, RpcServerError> {
        let build = || {
            let account =
                PubkeyAccount::from_public_key_bytes(self.signer.verifying_key_bytes(), 0);
            let calldata =
                thruput_signing::native_token_transfer_calldata(&recipient.0, raw_amount)?;
            let tx = thruput_signing::default_wallet::build_default_wallet_call_transaction(
                thruput_signing::default_wallet::BuildDefaultWalletCallTransaction {
                    account: &account,
                    target_address: thruput_xdr::genesis_addresses::ADDRESS_NATIVE_TOKEN,
                    target_entrypoint: thruput_signing::NATIVE_TOKEN_TRANSFER_ENTRYPOINT,
                    target_calldata: &calldata,
                    gas_limit: DEFAULT_TX_GAS_LIMIT * 2,
                    target_gas_limit: DEFAULT_TX_GAS_LIMIT,
                    block_expiry_height,
                    nonce: self.next_nonce.fetch_add(1, Ordering::Relaxed),
                },
            )?;
            let tx_hash = transaction_hash(&tx)?;
            let signed = sign_transaction(
                tx,
                self.signer.as_ref(),
                SignTransactionParams::default(),
                &self.network_hash,
            )?;
            let bytes = encode_signed_transaction_xdr(&signed)?;
            Ok::<_, thruput_signing::SigningError>(SignedAirdrop {
                bytes,
                tx_hash: TxHash(Base58(tx_hash)),
            })
        };
        build().map_err(|error| {
            RpcServerError::BackendUnavailable(format!(
                "failed to construct local airdrop transaction: {error}"
            ))
        })
    }
}

#[async_trait]
trait AirdropBackend: Send + Sync + 'static {
    async fn chain_tip(
        &self,
        context: RpcRequestContext,
    ) -> Result<thruput_json_rpc_types::block::ChainTip, RpcServerError>;
    async fn submit(
        &self,
        context: RpcRequestContext,
        signed: &SignedAirdrop,
    ) -> Result<(), RpcServerError>;
    async fn receipt(
        &self,
        context: RpcRequestContext,
        tx_hash: TxHash,
    ) -> Result<Option<thruput_json_rpc_types::receipt::TransactionReceipt>, RpcServerError>;
    async fn status(
        &self,
        context: RpcRequestContext,
        tx_hash: TxHash,
    ) -> Result<TransactionStatus, RpcServerError>;
}

/// Read once through the native validator before adapting for the existing funding state machine.
#[tracing::instrument(skip_all, fields(?hash))]
pub(crate) async fn read_funding_receipt<R: NativeTransactionReadPlane>(
    read: &R,
    context: RpcRequestContext,
    hash: TxHash,
) -> Result<Option<thruput_json_rpc_types::receipt::TransactionReceipt>, RpcServerError> {
    use thruput_json_rpc_types::{
        error::native::BackendDataInvalidReason,
        events::EmittedEvent,
        receipt::{DETERMINISTIC_ERROR_KIND_FAULT, TransactionReceipt},
        responses::ReceiptExecution,
    };
    let value = within_request_deadline(context, async {
        read.native_receipt(context, hash)
            .await
            .map_err(|error| error.error)
    })
    .await?;
    value
        .map(|receipt| {
            let (status, return_data, abort_data) = match receipt.execution {
                ReceiptExecution::Return { return_data } => {
                    (ExecStatus::Ok, Some(return_data), None)
                }
                ReceiptExecution::Abort { abort_data } => (
                    ExecStatus::Reverted { message: None },
                    None,
                    Some(abort_data),
                ),
                ReceiptExecution::Fault {} => (
                    ExecStatus::DeterministicError {
                        kind: DETERMINISTIC_ERROR_KIND_FAULT.into(),
                    },
                    None,
                    None,
                ),
                ReceiptExecution::NonDeterministicError {} => {
                    (ExecStatus::NonDeterministicError, None, None)
                }
                ReceiptExecution::FailedGasSponsorship {} => {
                    (ExecStatus::FailedGasSponsorship, None, None)
                }
                ReceiptExecution::Other {} => (ExecStatus::Other, None, None),
                _ => {
                    return Err(RpcServerError::invalid_data(
                        BackendDataInvalidReason::UnsupportedValue,
                        "unsupported funding receipt execution",
                    ));
                }
            };
            Ok(TransactionReceipt {
                tx_hash: receipt.transaction_hash,
                height: receipt.block_height,
                status,
                return_data,
                abort_data,
                gas_used: receipt.gas_used,
                events: receipt
                    .events
                    .into_iter()
                    .map(|event| EmittedEvent {
                        contract_address: event.address,
                        topics: event.topics,
                        data: event.data,
                    })
                    .collect(),
            })
        })
        .transpose()
}

struct PlaneAirdropBackend<S, R> {
    submit_chain: S,
    read: R,
}

#[async_trait]
impl<S, R> AirdropBackend for PlaneAirdropBackend<S, R>
where
    S: FundingSubmitPlane + ChainPlane,
    R: FundingStatusPlane + NativeTransactionReadPlane,
{
    async fn chain_tip(
        &self,
        context: RpcRequestContext,
    ) -> Result<thruput_json_rpc_types::block::ChainTip, RpcServerError> {
        within_request_deadline(context, async {
            Ok(self.submit_chain.chain_tip(context).await?.tip)
        })
        .await
    }

    async fn submit(
        &self,
        context: RpcRequestContext,
        signed: &SignedAirdrop,
    ) -> Result<(), RpcServerError> {
        let tx =
            validate_raw_transaction_bytes(signed.bytes.clone(), &NetworkConfigParams::default())
                .map_err(|error| {
                RpcServerError::BackendUnavailable(format!(
                    "locally constructed airdrop transaction failed validation: {error}"
                ))
            })?;
        if tx.hash != signed.tx_hash {
            return Err(RpcServerError::BackendUnavailable(
                "retained local airdrop transaction hash does not match its signed bytes".into(),
            ));
        }
        within_request_deadline(context, self.submit_chain.submit_funding(context, tx)).await
    }

    async fn receipt(
        &self,
        context: RpcRequestContext,
        tx_hash: TxHash,
    ) -> Result<Option<thruput_json_rpc_types::receipt::TransactionReceipt>, RpcServerError> {
        read_funding_receipt(&self.read, context, tx_hash).await
    }

    async fn status(
        &self,
        context: RpcRequestContext,
        tx_hash: TxHash,
    ) -> Result<TransactionStatus, RpcServerError> {
        within_request_deadline(context, self.read.funding_status(context, tx_hash)).await
    }
}

fn classify_backend_error(error: RpcServerError) -> RpcServerError {
    match error {
        RpcServerError::TxInvalid(message) => RpcServerError::TxRejected(message),
        RpcServerError::NotImplemented(method) => RpcServerError::BackendUnavailable(format!(
            "local airdrop requires backend operation `{method}`"
        )),
        other => other,
    }
}

fn is_deterministic_failure(error: &RpcServerError) -> bool {
    matches!(error, RpcServerError::TxRejected(_))
}

fn dropped_funding_error(error: &SchedulerTransactionError) -> RpcServerError {
    RpcServerError::TxRejected(format!(
        "local funding transaction dropped: {:?} code {}",
        error.source, error.code
    ))
}

fn nonce_from_timestamp() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .ok()
        .and_then(|duration| u64::try_from(duration.as_nanos()).ok())
        .unwrap_or(1)
}

#[cfg(test)]
mod tests;

#[async_trait]
impl thruput_json_rpc_server::native::funding::NativeFundingPlane for LocalAirdropRpc {
    fn capabilities(&self) -> thruput_json_rpc_server::native::funding::FundingCapabilities {
        thruput_json_rpc_server::native::funding::FundingCapabilities::Local
    }

    #[tracing::instrument(skip_all)]
    async fn request_airdrop(
        &self,
        request: thruput_json_rpc_types::requests::AirdropRequest,
    ) -> RpcResult<thruput_json_rpc_types::responses::AirdropAttempt> {
        use thruput_json_rpc_server::native::funding;
        let recipient = request.recipient;
        let retained_request = FundAccountRequest {
            recipient,
            raw_amount: self.raw_amount.to_string(),
            client_nonce: request.client_nonce.clone(),
            faucet: self.faucet,
        };
        match self
            .request(AirdropRequest {
                recipient,
                client_nonce: request.client_nonce,
            })
            .await
        {
            Ok(result) => funding::airdrop_result(recipient, result)
                .map_err(|error| funding::rpc_error(&error)),
            Err(error) => {
                let retained = self
                    .journal
                    .try_lock()
                    .ok()
                    .and_then(|journal| journal.durable_signed(&retained_request));
                let evidence =
                    retained.map(|v| LocalFundingErrorData::new(v.tx_hash, v.signed_transaction));
                Err(funding::rpc_error_with_evidence(&error, evidence.as_ref()))
            }
        }
    }

    #[tracing::instrument(skip_all)]
    async fn fund_account(
        &self,
        request: thruput_json_rpc_types::requests::FundAccountRequest,
    ) -> RpcResult<thruput_json_rpc_types::responses::FundingSubmission> {
        use thruput_json_rpc_server::native::funding;
        let legacy = FundAccountRequest {
            recipient: request.recipient,
            raw_amount: request.raw_amount.to_string(),
            client_nonce: request.client_nonce.clone(),
            faucet: request.faucet,
        };
        let result = self
            .fund(legacy)
            .await
            .map_err(|error| funding::rpc_error(&error))?;
        funding::submission_result(&request, result).map_err(|error| funding::rpc_error(&error))
    }
}