From 129d87421cfe409d804c0b9bca6c71b8d6fa0f85 Mon Sep 17 00:00:00 2001 From: Aliyu Ibrahim <271224223+ykargeee-bit@users.noreply.github.com> Date: Sat, 17 Oct 2026 07:45:08 +0100 Subject: [PATCH] feat(escrow,sdk,worker): funding event, resumable batches, endpoint health, donation dry-run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Additive escrow funding notification, resumable SDK batch submission, actionable endpoint health/error classification, and a worker donation dry-run. One branch for #978/#979/#980/#981. #978 — Escrow funding event + transition audit - `create_escrow` now also emits the additive topic `escrow`/`funded` carrying `(commission_id, funder, amount, asset, ledger)`. The existing `escrow`/`created` topic and its `(commission_id, amount)` payload are unchanged, so downstream indexers keep working. - `finalize_atomic_commit` now emits `escrow`/`settled`, so a client watching the multi-step atomic-commit flow no longer has to poll `get_atomic_commit` after each of the four steps. - Audited every escrow transition (create/fund, partial, release, auto release, refund, dispute open, dispute resolve, cancel, expire) and documented the events in `docs/EVENTS.md` (with a state-transition audit table) and `docs/event_schemas.md` (`EscrowFunded`, `EscrowSettled`). - New `contracts/escrow/src/funding_event_tests.rs` covers the funded event values, the unchanged `created` payload, and the finalize event. #979 — Resumable batch checkpoint - New `sdk::batch_checkpoint`: `BatchCheckpoint` (serde-serialisable calls + per-call `CallState` + cursor), a caller-supplied `BatchCheckpointStore` trait (with in-memory and atomic file implementations), and `ResumableBatch::submit_remaining`, which persists after every call. - Additive `BatchBuilder::checkpoint` / `from_checkpoint`; no existing `BatchBuilder` method changed. - Documented the guarantee explicitly: submission is at-least-once, a crash loses at most the one in-flight call, and callers must make calls idempotent (deterministic per-call marker) to dedupe. Complements `retry_async` rather than replacing it. #980 — Endpoint health + actionable error taxonomy - New `StellarAidError` variants `EndpointUnreachable`, `RequestTimeout`, `BadRequest`, `TransactionRejected`, plus `is_retryable()` and `from_reqwest`/`from_http_status` classifiers. (Adding variants is a breaking change for exhaustive matches; acceptable for an unreleased SDK.) - `retry_async_classified` (predicate) and `retry_stellar_async`, so retrying can stop on a non-retryable error instead of retrying it fully. - Typed `SorobanRpcClient::health()` and `HorizonClient::health()`, bounded by a short `HEALTH_PROBE_TIMEOUT`; intentionally not a per-operation precondition. - `Config::validate()` checks the URL shapes and that the passphrase matches the named network — the most common integration mistake now fails immediately with a clear message. #981 — Worker donation dry-run - New `POST /api/donations/validate` builds the same XDR as submit and simulates it via the real `SorobanRpcClient::simulate_transaction` RPC (not the placeholder `TransactionSimulator`, which returns success unconditionally). It reports `valid`, `gas_consumed`, the contract's own error, and an extracted `error_code`. - Cheap local validation first (donor, positive amount, non-zero campaign id, memo length) so obvious failures cost no round trip; the route has its own tighter rate-limit budget. `/api/donations/submit` is unchanged. - The success / contract-error / RPC-failure interpretation is a pure function (`interpret_simulation`), so it is covered without a network round trip. - Unit tests for a valid request, an invalid campaign id, a zero amount, an empty donor, an over-long memo, error-code extraction, and the three simulation outcomes. Enabling fixes needed for the SDK/worker to build: - Bump the workspace `stellar-strkey` pin from 0.0.2 to 0.0.8 — the SDK is written against the 0.0.8 API (`stellar_strkey::ed25519::*`, `stellar_strkey::Contract`) and 0.0.2 has neither; 0.0.8 is also what `stellar-xdr` 21 already pulls in. `transaction_builder.rs` updated to the 0.0.8 `Strkey::Contract` variant. - Declare `ed25519-dalek` in `sdk` (already a workspace dep) for key derivation. - Declare `hmac`, `sha2` and `hex` in `worker` for the webhook signature verification already present in `webhooks.rs`. Note: `worker` is still not a workspace member, so its integration cannot be verified in CI; that remains a separate prerequisite (as the issue states). Closes #978 Closes #979 Closes #980 Closes #981 --- Cargo.toml | 7 +- contracts/escrow/src/cross_contract.rs | 8 + contracts/escrow/src/funding_event_tests.rs | 154 +++++ contracts/escrow/src/lib.rs | 17 +- docs/EVENTS.md | 41 ++ docs/SDK_VERSION_COMPATIBILITY.md | 7 + docs/event_schemas.md | 29 +- sdk/Cargo.toml | 1 + sdk/src/batch.rs | 45 +- sdk/src/batch_checkpoint.rs | 654 ++++++++++++++++++++ sdk/src/config.rs | 115 ++++ sdk/src/errors.rs | 114 +++- sdk/src/health.rs | 80 +++ sdk/src/horizon/client.rs | 55 ++ sdk/src/lib.rs | 2 + sdk/src/retry.rs | 158 +++++ sdk/src/soroban/rpc_client.rs | 36 ++ sdk/src/transaction_builder.rs | 6 +- worker/Cargo.toml | 4 + worker/src/main.rs | 275 +++++++- 20 files changed, 1799 insertions(+), 9 deletions(-) create mode 100644 contracts/escrow/src/funding_event_tests.rs create mode 100644 sdk/src/batch_checkpoint.rs create mode 100644 sdk/src/health.rs diff --git a/Cargo.toml b/Cargo.toml index 0985e31..d9c1d47 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,7 +52,12 @@ min-compatible = "0.1.0" [workspace.dependencies] soroban-sdk = { version = "21.0.0" } -stellar-strkey = "0.0.2" +# 0.0.8 is the API the SDK is written against (`stellar_strkey::ed25519::*` +# and `stellar_strkey::Contract`); it is also the version `stellar-xdr` 21 +# already pulls in, so this removes the duplicate 0.0.2 copy. The 0.0.2 pin +# only exposed `Strkey::{PublicKeyEd25519, PrivateKeyEd25519}` and had no +# `ed25519` module, so `sdk` could not compile against it (issue #5). +stellar-strkey = "0.0.8" stellar-xdr = "21.0.0" reqwest = { version = "0.12", features = ["json"] } serde = { version = "1", features = ["derive"] } diff --git a/contracts/escrow/src/cross_contract.rs b/contracts/escrow/src/cross_contract.rs index 076f012..82226ea 100644 --- a/contracts/escrow/src/cross_contract.rs +++ b/contracts/escrow/src/cross_contract.rs @@ -258,6 +258,14 @@ pub fn finalize_atomic_commit( marker.state = AtomicCommitState::Settled; marker.settled_ledger = Some(env.ledger().sequence()); storage::save_atomic_marker(env, &marker); + + // EVENT (#978): the multi-step atomic-commit flow is only observable at the + // end; publishing here means a client watching for the final state does not + // have to poll `get_atomic_commit` after each of the four steps. + env.events().publish( + (symbol_short!("escrow"), symbol_short!("settled")), + (commission_id.clone(), env.ledger().sequence()), + ); Ok(marker) } diff --git a/contracts/escrow/src/funding_event_tests.rs b/contracts/escrow/src/funding_event_tests.rs new file mode 100644 index 0000000..cbf3b8c --- /dev/null +++ b/contracts/escrow/src/funding_event_tests.rs @@ -0,0 +1,154 @@ +//! Tests for the escrow funding notification (#978). +//! +//! These assert the *additive* `escrow`/`funded` topic and the atomic-commit +//! finalization event, and — importantly — that the pre-existing +//! `escrow`/`created` payload is unchanged (it is indexed downstream). + +extern crate std; +use soroban_sdk::{ + contract, contractimpl, symbol_short, testutils::Address as _, testutils::Events, token, + Address, Bytes, Env, Symbol, TryFromVal, Val, +}; + +use crate::cross_contract::AtomicCommitState; +use crate::{EscrowContract, EscrowContractClient}; + +const FEE_BPS: u32 = 500; +const AMOUNT: i128 = 10_000; + +/// Minimal platform-config stand-in so `create_escrow` can resolve fee/token. +#[contract] +pub struct MockConfig; + +#[contractimpl] +impl MockConfig { + pub fn init(env: Env, admin: Address, usdc: Address, platform_wallet: Address) { + env.storage().instance().set(&0u32, &admin); + env.storage().instance().set(&1u32, &usdc); + env.storage().instance().set(&2u32, &platform_wallet); + } + pub fn get_adm(env: Env) -> Address { + env.storage().instance().get(&0u32).unwrap() + } + pub fn get_usdc(env: Env) -> Address { + env.storage().instance().get(&1u32).unwrap() + } + pub fn get_pw(env: Env) -> Address { + env.storage().instance().get(&2u32).unwrap() + } + pub fn get_fee_b(_env: Env) -> u32 { + FEE_BPS + } +} + +struct Fixture<'a> { + env: Env, + escrow: EscrowContractClient<'a>, + config: Address, + usdc: Address, + client: Address, + artist: Address, + commission_id: Bytes, +} + +fn setup<'a>() -> Fixture<'a> { + let env = Env::default(); + env.mock_all_auths(); + + let token_admin = Address::generate(&env); + let usdc = env + .register_stellar_asset_contract_v2(token_admin.clone()) + .address(); + let admin = Address::generate(&env); + let client = Address::generate(&env); + let artist = Address::generate(&env); + let platform_wallet = Address::generate(&env); + + token::StellarAssetClient::new(&env, &usdc).mint(&client, &1_000_000); + + let config = env.register_contract(None, MockConfig); + MockConfigClient::new(&env, &config).init(&admin, &usdc, &platform_wallet); + + let escrow_id = env.register_contract(None, EscrowContract); + let escrow = EscrowContractClient::new(&env, &escrow_id); + + Fixture { + env, + escrow, + config, + usdc, + client, + artist, + commission_id: Bytes::from_slice(&env, b"comm-001"), + } +} + +/// Return the data of the first two-topic event matching `t0`/`t1`. +fn two_topic_data(f: &Fixture, t0: &str, t1: &str) -> Option { + let want0 = Symbol::new(&f.env, t0); + let want1 = Symbol::new(&f.env, t1); + for (_contract, topics, data) in f.env.events().all().iter() { + if topics.len() != 2 { + continue; + } + let a: Symbol = Symbol::try_from_val(&f.env, &topics.get(0).unwrap()).unwrap(); + let b: Symbol = Symbol::try_from_val(&f.env, &topics.get(1).unwrap()).unwrap(); + if a == want0 && b == want1 { + return Some(data); + } + } + None +} + +#[test] +fn create_escrow_emits_funding_event_with_expected_values() { + let f = setup(); + f.escrow + .create_escrow(&f.commission_id, &f.client, &f.artist, &AMOUNT, &f.config); + + let data = two_topic_data(&f, "escrow", "funded") + .expect("expected an (escrow, funded) event"); + let (commission_id, funder, amount, asset, ledger): (Bytes, Address, i128, Address, u32) = + TryFromVal::try_from_val(&f.env, &data).unwrap(); + + assert_eq!(commission_id, f.commission_id); + assert_eq!(funder, f.client); + assert_eq!(amount, AMOUNT); + assert_eq!(asset, f.usdc); + assert_eq!(ledger, f.env.ledger().sequence()); +} + +#[test] +fn created_payload_is_unchanged() { + let f = setup(); + f.escrow + .create_escrow(&f.commission_id, &f.client, &f.artist, &AMOUNT, &f.config); + + let data = + two_topic_data(&f, "escrow", "created").expect("expected an (escrow, created) event"); + let (commission_id, amount): (Bytes, i128) = TryFromVal::try_from_val(&f.env, &data).unwrap(); + assert_eq!(commission_id, f.commission_id); + assert_eq!(amount, AMOUNT); +} + +#[test] +fn finalize_atomic_commit_publishes_settled_event() { + let f = setup(); + f.escrow + .create_escrow(&f.commission_id, &f.client, &f.artist, &AMOUNT, &f.config); + + let marker = f.escrow.begin_atomic_commit(&f.commission_id); + assert_eq!(marker.state, AtomicCommitState::InProgress); + + f.escrow.confirm_atomic_step(&f.commission_id, &f.client); + f.escrow.confirm_atomic_step(&f.commission_id, &f.artist); + + let final_marker = f.escrow.finalize_atomic_commit(&f.commission_id); + assert_eq!(final_marker.state, AtomicCommitState::Settled); + + let data = two_topic_data(&f, "escrow", "settled") + .expect("expected an (escrow, settled) event on finalization"); + let (commission_id, ledger): (Bytes, u32) = TryFromVal::try_from_val(&f.env, &data).unwrap(); + assert_eq!(commission_id, f.commission_id); + assert_eq!(ledger, f.env.ledger().sequence()); +} diff --git a/contracts/escrow/src/lib.rs b/contracts/escrow/src/lib.rs index 525766e..08157d7 100644 --- a/contracts/escrow/src/lib.rs +++ b/contracts/escrow/src/lib.rs @@ -309,11 +309,24 @@ impl EscrowContract { &client, &env.current_contract_address(), &amount, ); - // EVENT + // EVENT — additive funding notification (#978). The existing + // `created` topic and payload are unchanged; `funded` carries the + // extra facts (funder, asset, ledger) a client needs to notify an + // artist without polling `get_escrow`. env.events().publish( (symbol_short!("escrow"), symbol_short!("created")), (commission_id.clone(), amount), ); + env.events().publish( + (symbol_short!("escrow"), symbol_short!("funded")), + ( + commission_id.clone(), + client.clone(), + amount, + usdc_token.clone(), + env.ledger().sequence(), + ), + ); correlation_publish(&env, "escrow", symbol_short!("created"), &commission_id, &commission_id); Ok(()) }) @@ -1032,3 +1045,5 @@ mod atomic_flow_tests; mod correlation_tests; #[cfg(test)] mod invariant_tests; +#[cfg(test)] +mod funding_event_tests; diff --git a/docs/EVENTS.md b/docs/EVENTS.md index d728a5f..948d68a 100644 --- a/docs/EVENTS.md +++ b/docs/EVENTS.md @@ -94,6 +94,19 @@ Emitted by `contracts/escrow` during payment custody lifecycle. | `amount` | `i128` | Total locked token amount (in stroops) | | `fee_bps` | `u32` | Platform fee in basis points (e.g. 500 = 5%) | +#### `escrow` / `funded` +* **Trigger:** `create_escrow` +* **Condition:** Successful lock of client token funds into escrow storage. Emitted alongside `created`, which is **unchanged**; the `created` payload stays `(commission_id, amount)` and `funded` is the additive topic carrying the extra funding facts (#978). +* **Payload:** + +| Field | Type | Description | +|---|---|---| +| `commission_id` | `Bytes` | Unique commission identifier | +| `funder` | `Address` | Address that funded the escrow (the client) | +| `amount` | `i128` | Total locked token amount (in stroops) | +| `asset` | `Address` | USDC token contract that was transferred | +| `ledger` | `u32` | Ledger sequence at which funding was recorded | + #### `escrow` / `released` * **Trigger:** `release_payment` * **Condition:** Client authorizes full release of funds to artist; net payout and fee transfers succeed. @@ -148,6 +161,34 @@ Emitted by `contracts/escrow` during payment custody lifecycle. | `client_amount` | `i128` | Amount returned to client | | `artist_amount` | `i128` | Amount paid to artist | +#### `escrow` / `settled` +* **Trigger:** `finalize_atomic_commit` +* **Condition:** Final step of the multi-transaction atomic-commit flow, once every participant has confirmed. Emitted so a client watching the final state does not have to poll `get_atomic_commit` after each of the four steps (#978). `atomic_escrow_to_commission` keeps its existing `escrow`/`released` event and its `corr`-topic `settled` correlation event; this two-topic `settled` is the additive no-poll marker on the stepwise path. +* **Payload:** + +| Field | Type | Description | +|---|---|---| +| `commission_id` | `Bytes` | Unique commission identifier | +| `ledger` | `u32` | Ledger sequence at which the commit settled | + +#### State-transition audit (#978) + +Every escrow state transition now publishes a topic, with the funding gap closed here: + +| Transition | Function | Event | +|---|---|---| +| create + fund | `create_escrow` | `escrow`/`created` + `escrow`/`funded` | +| partial release | `partial_release` | `escrow`/`partial` | +| full release | `release_payment` | `escrow`/`released` | +| auto release | `auto_release_on_deadline` | `escrow`/`autorls` | +| refund | `refund_client` | `escrow`/`refunded` | +| dispute open | `open_dispute` | `escrow`/`disputed` | +| dispute resolve | `release_payment` / `refund_client` (arbiter decides the winner) | `escrow`/`released` or `escrow`/`refunded` | +| cancel | `cancel_escrow` | `escrow`/`cancelled` | +| expire | `expire_escrow` | `escrow`/`expired` | +| atomic commit finalize | `finalize_atomic_commit` | `escrow`/`settled` (new) | +| atomic commit rollback | `rollback_atomic_commit` | `corr`-topic `rollback` | + #### Correlation Events (Escrow) Cross-contract correlation points emitted by `contracts/escrow` (from diff --git a/docs/SDK_VERSION_COMPATIBILITY.md b/docs/SDK_VERSION_COMPATIBILITY.md index 35c5c4a..1b99dea 100644 --- a/docs/SDK_VERSION_COMPATIBILITY.md +++ b/docs/SDK_VERSION_COMPATIBILITY.md @@ -20,6 +20,13 @@ Every contract crate inherits this via `soroban-sdk.workspace = true`, so there | 20.x | Untested | Predates several APIs this workspace relies on (e.g. `env.deployer().update_current_contract_wasm`, the `#[contracterror]` discriminant conventions used across every contract's error enum). Not verified; likely requires code changes, not just a version bump. | | 22.x+ | Untested | Not yet evaluated. Before upgrading, check the SDK's own changelog for storage-layout or auth-model changes, since this workspace stores non-trivial `contracttype` enums directly as persistent-storage keys (see e.g. `contracts/withdrawal/src/lib.rs`'s `DataKey`) — any change to how the SDK encodes those would be a breaking migration, not just a recompile. | +## Related dependency pins + +| Crate | Pinned | Why | +|---|---|---| +| `stellar-strkey` | `0.0.8` | The `sdk` crate is written against the 0.0.8 API — `stellar_strkey::ed25519::{PublicKey, PrivateKey}` and the `stellar_strkey::Contract` newtype. The previous `0.0.2` pin only exposed `Strkey::{PublicKeyEd25519, PrivateKeyEd25519}` and had no `ed25519` module, so `sdk` could not compile against it. `stellar-xdr` 21 already pulls in 0.0.8, so pinning to it also collapses a duplicate. | +| `ed25519-dalek` | `=2.2.0` | Used by `sdk::utils::keypair` and `sdk::setup::token_setup` to derive a public key from a secret key. | + ## Why this isn't (yet) a CI version matrix The acceptance criteria for #645 also call for adding a version matrix to CI and fuzzing/compatibility test automation. That requires actually building and testing against each candidate SDK version in a CI job — a workflow-level change, not a documentation one. This file exists so that follow-up work (or a maintainer deciding to attempt a 20.x/22.x build) starts from a written record of what's known instead of from scratch. diff --git a/docs/event_schemas.md b/docs/event_schemas.md index 7235499..c05c241 100644 --- a/docs/event_schemas.md +++ b/docs/event_schemas.md @@ -61,4 +61,31 @@ Emitted when a dispute is resolved. | Field | Type | Description | |------------|--------|-------------------------------| | dispute_id | u64 | ID of the resolved dispute | -| winner | String | Address of the winning party | \ No newline at end of file +| winner | String | Address of the winning party | + +--- + +## EscrowFunded + +Emitted when an escrow is funded (`create_escrow`). Additive topic `escrow`/`funded` +(#978); the pre-existing `escrow`/`created` payload is unchanged. + +| Field | Type | Description | +|----------------|--------|-----------------------------------------------| +| commission_id | Bytes | Unique commission identifier | +| funder | String | Address that funded the escrow (the client) | +| amount | i128 | Total locked amount, in stroops | +| asset | String | USDC token contract address | +| ledger | u32 | Ledger sequence at which funding was recorded | + +--- + +## EscrowSettled + +Emitted when the multi-transaction atomic-commit flow finalizes +(`finalize_atomic_commit`), so clients need not poll `get_atomic_commit` (#978). + +| Field | Type | Description | +|----------------|-------|---------------------------------------------| +| commission_id | Bytes | Unique commission identifier | +| ledger | u32 | Ledger sequence at which the commit settled | \ No newline at end of file diff --git a/sdk/Cargo.toml b/sdk/Cargo.toml index c09afce..75a27fb 100644 --- a/sdk/Cargo.toml +++ b/sdk/Cargo.toml @@ -8,6 +8,7 @@ publish = false soroban-sdk = { workspace = true } stellar-strkey = { workspace = true } stellar-xdr = { workspace = true } +ed25519-dalek = { workspace = true } reqwest = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/sdk/src/batch.rs b/sdk/src/batch.rs index 3ad8f57..b023fad 100644 --- a/sdk/src/batch.rs +++ b/sdk/src/batch.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; use std::sync::Arc; +use serde::{Deserialize, Serialize}; use thiserror::Error; use tokio::sync::Mutex; use tracing::{debug, info, warn}; @@ -18,6 +19,9 @@ pub enum BatchError { EmptyBatch, #[error("batch size exceeds maximum ({0})")] TooLarge(usize), + /// A checkpoint store failed to persist or load batch state (#979). + #[error("batch checkpoint error: {0}")] + Checkpoint(String), } // --------------------------------------------------------------------------- @@ -25,7 +29,7 @@ pub enum BatchError { // --------------------------------------------------------------------------- /// A single contract call to be batched. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ContractCall { /// Contract address (C... strkey). pub contract_id: String, @@ -142,6 +146,15 @@ pub enum BatchResult { /// /// If the batch submission fails, each call is retried individually. /// This ensures that partial progress is still possible. +/// +/// # Durability +/// +/// `Submit` handles an in-process failure. A *process* that dies between two +/// submissions is handled by [`crate::batch_checkpoint`], which persists the +/// pending calls and per-call outcomes to a caller-supplied store. The two are +/// complementary: [`crate::retry`] handles transient errors within a run, the +/// checkpoint handles a run that ended. `take_calls()` remains the raw drain; +/// `checkpoint()` is the resumable equivalent. pub struct BatchBuilder { calls: Vec, max_batch_size: usize, @@ -200,6 +213,36 @@ impl BatchBuilder { std::mem::take(&mut self.calls) } + /// Snapshot the accumulated calls into a resumable [`BatchCheckpoint`]. + /// + /// Additive (#979): the checkpoint can be handed to a + /// [`BatchCheckpointStore`] so a crash no longer loses the batch. See + /// [`crate::batch_checkpoint`] for the at-least-once guarantee. + /// + /// [`BatchCheckpoint`]: crate::batch_checkpoint::BatchCheckpoint + /// [`BatchCheckpointStore`]: crate::batch_checkpoint::BatchCheckpointStore + pub fn checkpoint(&self) -> crate::batch_checkpoint::BatchCheckpoint { + crate::batch_checkpoint::BatchCheckpoint::new(self.calls.clone()) + } + + /// Rebuild a builder from a checkpoint's pending calls. + /// + /// Additive (#979). Only calls that have not already succeeded are + /// re-queued, so resuming does not re-submit completed work. + pub fn from_checkpoint(checkpoint: &crate::batch_checkpoint::BatchCheckpoint) -> Self { + let pending: Vec = checkpoint + .retryable_indices() + .into_iter() + .map(|i| checkpoint.calls[i].clone()) + .collect(); + let max_batch_size = pending.len().max(1); + Self { + calls: pending, + max_batch_size, + metrics: Arc::new(Mutex::new(BatchMetrics::default())), + } + } + /// Group calls by contract address for transaction optimization. /// /// Calls to the same contract can be merged into a single transaction diff --git a/sdk/src/batch_checkpoint.rs b/sdk/src/batch_checkpoint.rs new file mode 100644 index 0000000..6b4ae21 --- /dev/null +++ b/sdk/src/batch_checkpoint.rs @@ -0,0 +1,654 @@ +//! Resumable batch submission: a durable checkpoint for [`BatchBuilder`] (#979). +//! +//! [`crate::batch`] is a capable batch builder, but `take_calls()` is +//! one-shot: if the process dies between two submissions, the batch is gone. +//! This module adds the missing *durability* dimension. It is deliberately +//! additive — no existing `BatchBuilder` method changes — and is complementary +//! to [`crate::retry`]: +//! +//! | Concern | Mechanism | +//! |---|---| +//! | A transient error *within* a run | [`crate::retry::retry_async`] with backoff | +//! | A run that *ended* (crash, restart, deploy) | [`BatchCheckpoint`] + [`ResumableBatch`] | +//! +//! # At-least-once, not exactly-once +//! +//! Resuming re-submits every call that is not recorded as succeeded. A crash +//! between *submitting* a call and *persisting* its outcome therefore re-submits +//! that call on the next run. The guarantee is therefore **at-least-once**: +//! every call is submitted one or more times, never zero. +//! +//! Callers who cannot tolerate a duplicate must make their calls idempotent. The +//! recommended mechanism is a deterministic per-call marker — the same value on +//! every attempt, e.g. a hash of `(contract_id, function_name, params_xdr)` +//! embedded in the call's params or in the transaction memo — so the callee can +//! recognise and ignore a replay. The SDK does not do this for you because it +//! cannot know which callee is idempotent. +//! +//! # Maximum-loss window +//! +//! The checkpoint is persisted after **every** call outcome, and a +//! [`BatchCheckpointStore`] must implement `save` atomically (write-temp then +//! rename; see [`FileCheckpointStore`]). The maximum state lost to a crash is +//! therefore the outcome of **one** call — the one in flight — which is exactly +//! the call that will be retried. Batches are never silently truncated. + +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use serde::{Deserialize, Serialize}; + +use crate::batch::{BatchError, ContractCall}; + +/// Outcome of a single call in a checkpointed batch. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum CallState { + /// Not attempted yet in any run. + Pending, + /// Submitted and confirmed; holds the transaction hash. + Succeeded(String), + /// Attempted and failed; holds the reason. Retried on the next run. + Failed(String), +} + +impl Default for CallState { + fn default() -> Self { + CallState::Pending + } +} + +/// A durable snapshot of a batch: the calls, their per-call outcomes, and the +/// cursor within the current run. +/// +/// Serialisable so it can live in a file, a database row, or object storage — +/// the SDK never picks the medium. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct BatchCheckpoint { + /// Schema version of the checkpoint envelope itself. + pub version: u32, + /// The full ordered call list. + pub calls: Vec, + /// Per-call outcome, index-aligned with `calls`. + pub states: Vec, + /// Progress within the current run (not persisted meaning; reset on resume). + #[serde(default)] + pub cursor: usize, +} + +impl BatchCheckpoint { + /// Current checkpoint envelope version. + pub const VERSION: u32 = 1; + + /// Create a fresh checkpoint for `calls`, all pending. + pub fn new(calls: Vec) -> Self { + let states = vec![CallState::Pending; calls.len()]; + Self { + version: Self::VERSION, + calls, + states, + cursor: 0, + } + } + + pub fn len(&self) -> usize { + self.calls.len() + } + + pub fn is_empty(&self) -> bool { + self.calls.is_empty() + } + + /// Fail if this checkpoint was written by an incompatible SDK version. + pub fn validate(&self) -> Result<(), BatchError> { + if self.version != Self::VERSION { + return Err(BatchError::Checkpoint(format!( + "unsupported checkpoint version {} (expected {})", + self.version, + Self::VERSION + ))); + } + if self.states.len() != self.calls.len() { + return Err(BatchError::Checkpoint( + "checkpoint is corrupt: states/calls length mismatch".into(), + )); + } + Ok(()) + } + + /// Calls never attempted. + pub fn pending_indices(&self) -> Vec { + self.indices_where(|s| matches!(s, CallState::Pending)) + } + + /// Calls that were attempted and failed. + pub fn failed_indices(&self) -> Vec { + self.indices_where(|s| matches!(s, CallState::Failed(_))) + } + + /// Calls that still need work: pending or previously failed. This is what a + /// resume pass submits, and what makes "retry only the failures" work. + pub fn retryable_indices(&self) -> Vec { + self.indices_where(|s| !matches!(s, CallState::Succeeded(_))) + } + + /// Calls already confirmed. + pub fn succeeded_indices(&self) -> Vec { + self.indices_where(|s| matches!(s, CallState::Succeeded(_))) + } + + pub fn succeeded_count(&self) -> usize { + self.succeeded_indices().len() + } + + /// True once every call has succeeded. + pub fn is_complete(&self) -> bool { + !self.calls.is_empty() && self.states.iter().all(|s| matches!(s, CallState::Succeeded(_))) + } + + pub fn state(&self, index: usize) -> Option<&CallState> { + self.states.get(index) + } + + /// Record a successful call. `index` must be in range. + pub fn mark_succeeded( + &mut self, + index: usize, + tx_hash: impl Into, + ) -> Result<(), BatchError> { + match self.states.get_mut(index) { + Some(slot) => { + *slot = CallState::Succeeded(tx_hash.into()); + Ok(()) + } + None => Err(BatchError::Checkpoint(format!( + "call index {} out of range ({})", + index, + self.calls.len() + ))), + } + } + + /// Record a failed call. `index` must be in range. + pub fn mark_failed(&mut self, index: usize, reason: impl Into) -> Result<(), BatchError> { + match self.states.get_mut(index) { + Some(slot) => { + *slot = CallState::Failed(reason.into()); + Ok(()) + } + None => Err(BatchError::Checkpoint(format!( + "call index {} out of range ({})", + index, + self.calls.len() + ))), + } + } + + /// Serialise the checkpoint for a store to persist. + pub fn to_json(&self) -> Result { + serde_json::to_string(self) + .map_err(|e| BatchError::Checkpoint(format!("serialise failed: {e}"))) + } + + /// Deserialise and validate a checkpoint. + pub fn from_json(raw: &str) -> Result { + let checkpoint: Self = serde_json::from_str(raw) + .map_err(|e| BatchError::Checkpoint(format!("deserialise failed: {e}")))?; + checkpoint.validate()?; + Ok(checkpoint) + } + + fn indices_where(&self, f: impl Fn(&CallState) -> bool) -> Vec { + self.states + .iter() + .enumerate() + .filter(|(_, s)| f(s)) + .map(|(i, _)| i) + .collect() + } + + fn begin_run(&mut self) { + self.cursor = 0; + } + + fn advance(&mut self, index: usize) { + self.cursor = index + 1; + } + + fn next_unsucceeded(&self) -> Option { + (self.cursor..self.calls.len()) + .find(|&i| !matches!(self.states[i], CallState::Succeeded(_))) + } +} + +/// The caller-supplied medium for a checkpoint. Implement it for a file, a +/// database row, Redis, object storage — whatever the host already uses. +/// +/// # Contract +/// +/// [`BatchCheckpointStore::save`] **must be atomic**: a reader must never see a +/// partial or torn checkpoint. Implementations satisfy this by writing to a +/// temporary location and renaming over the destination. The SDK relies on this +/// to promise that a crash loses at most one call's outcome. +pub trait BatchCheckpointStore: Send + Sync { + fn save(&self, checkpoint: &BatchCheckpoint) -> Result<(), BatchError>; + fn load(&self) -> Result, BatchError>; + fn clear(&self) -> Result<(), BatchError>; +} + +/// In-memory store. Useful for tests and for a caller that already has a +/// durable layer elsewhere. +#[derive(Default)] +pub struct MemoryCheckpointStore { + inner: Mutex>, +} + +impl MemoryCheckpointStore { + pub fn new() -> Self { + Self::default() + } + + /// Peek at the current checkpoint without consuming it. + pub fn snapshot(&self) -> Option { + self.inner.lock().unwrap().clone() + } +} + +impl BatchCheckpointStore for MemoryCheckpointStore { + fn save(&self, checkpoint: &BatchCheckpoint) -> Result<(), BatchError> { + *self.inner.lock().unwrap() = Some(checkpoint.clone()); + Ok(()) + } + + fn load(&self) -> Result, BatchError> { + Ok(self.inner.lock().unwrap().clone()) + } + + fn clear(&self) -> Result<(), BatchError> { + *self.inner.lock().unwrap() = None; + Ok(()) + } +} + +/// File-backed store for scripts. `save` writes `path.tmp` then renames it over +/// `path`, so a concurrent reader sees either the old or the new checkpoint, +/// never a half-written one. +pub struct FileCheckpointStore { + path: PathBuf, +} + +impl FileCheckpointStore { + pub fn new(path: impl Into) -> Self { + Self { path: path.into() } + } + + pub fn path(&self) -> &Path { + &self.path + } + + fn temp_path(&self) -> PathBuf { + let mut tmp = self.path.clone(); + let ext = match self.path.extension().and_then(|e| e.to_str()) { + Some(e) => format!("{e}.tmp"), + None => "tmp".to_string(), + }; + tmp.set_extension(ext); + tmp + } +} + +impl BatchCheckpointStore for FileCheckpointStore { + fn save(&self, checkpoint: &BatchCheckpoint) -> Result<(), BatchError> { + let json = checkpoint.to_json()?; + let tmp = self.temp_path(); + std::fs::write(&tmp, json.as_bytes()) + .map_err(|e| BatchError::Checkpoint(format!("write {}: {e}", tmp.display())))?; + // `rename` replaces the destination atomically on the same filesystem. + std::fs::rename(&tmp, &self.path) + .map_err(|e| BatchError::Checkpoint(format!("rename into {}: {e}", self.path.display()))) + } + + fn load(&self) -> Result, BatchError> { + match std::fs::read_to_string(&self.path) { + Ok(raw) => Ok(Some(BatchCheckpoint::from_json(&raw)?)), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(e) => Err(BatchError::Checkpoint(format!( + "read {}: {e}", + self.path.display() + ))), + } + } + + fn clear(&self) -> Result<(), BatchError> { + match std::fs::remove_file(&self.path) { + Ok(()) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(BatchError::Checkpoint(format!( + "remove {}: {e}", + self.path.display() + ))), + } + } +} + +/// What one resume run did. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct BatchSummary { + pub attempted: usize, + pub succeeded: usize, + pub failed: usize, +} + +/// A checkpoint plus the loop that resumes it. +#[derive(Debug, Clone)] +pub struct ResumableBatch { + pub checkpoint: BatchCheckpoint, +} + +impl ResumableBatch { + pub fn new(checkpoint: BatchCheckpoint) -> Self { + Self { checkpoint } + } + + /// Start a fresh resumable batch from an ordered call list. + pub fn from_calls(calls: Vec) -> Self { + Self::new(BatchCheckpoint::new(calls)) + } + + /// Load the checkpoint from `store` and continue it. + pub fn resume(store: &dyn BatchCheckpointStore) -> Result { + match store.load()? { + Some(checkpoint) => { + checkpoint.validate()?; + Ok(Self::new(checkpoint)) + } + None => Err(BatchError::Checkpoint("no checkpoint to resume".into())), + } + } + + /// Submit every as-yet-unsucceeded call, persisting the checkpoint after + /// each outcome. + /// + /// Already-succeeded calls are skipped, so this retries only the calls that + /// are pending or previously failed. See the module docs for the + /// at-least-once guarantee and the one-call maximum-loss window. + pub async fn submit_remaining( + &mut self, + store: &dyn BatchCheckpointStore, + submit_fn: F, + ) -> Result + where + F: Fn(ContractCall) -> Fut, + Fut: std::future::Future>, + { + self.checkpoint.begin_run(); + let mut summary = BatchSummary::default(); + + while let Some(index) = self.checkpoint.next_unsucceeded() { + let call = self.checkpoint.calls[index].clone(); + summary.attempted += 1; + match submit_fn(call).await { + Ok(hash) => { + self.checkpoint.mark_succeeded(index, hash)?; + summary.succeeded += 1; + } + Err(reason) => { + self.checkpoint.mark_failed(index, reason)?; + summary.failed += 1; + } + } + self.checkpoint.advance(index); + // Persist after every outcome: the maximum window a crash can lose + // is this one call, which is exactly the one that gets retried. + store.save(&self.checkpoint)?; + } + + Ok(summary) + } +} + +/// Warn when a batch is being submitted at the fallback rate. +/// +/// [`crate::batch::BatchMetrics::fallback_rate`] is an estimator that exists to +/// be acted on; this surfaces it as a loggable string when it crosses +/// `threshold_pct`. +pub fn fallback_rate_warning( + metrics: &crate::batch::BatchMetrics, + threshold_pct: f64, +) -> Option { + let rate = metrics.fallback_rate(); + if rate > threshold_pct { + Some(format!( + "batch fallback rate {rate:.1}% exceeds {threshold_pct:.1}% \ + ({} of {} submissions fell back) — inspect batch sizes and contract errors", + metrics.fallback_submissions, + metrics.batch_submissions + metrics.fallback_submissions + )) + } else { + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + fn call(id: &str) -> ContractCall { + ContractCall::new(id, "fn").with_params(vec![1, 2, 3]) + } + + fn three_calls() -> Vec { + vec![call("C1"), call("C2"), call("C3")] + } + + #[test] + fn checkpoint_starts_all_pending() { + let cp = BatchCheckpoint::new(three_calls()); + assert_eq!(cp.len(), 3); + assert_eq!(cp.pending_indices(), vec![0, 1, 2]); + assert!(cp.failed_indices().is_empty()); + assert!(!cp.is_complete()); + } + + #[test] + fn json_roundtrip_preserves_everything() { + let mut cp = BatchCheckpoint::new(three_calls()); + cp.mark_succeeded(0, "hash-0").unwrap(); + cp.mark_failed(1, "boom").unwrap(); + + let raw = cp.to_json().unwrap(); + let back = BatchCheckpoint::from_json(&raw).unwrap(); + assert_eq!(back, cp); + } + + #[test] + fn version_mismatch_is_rejected() { + let mut cp = BatchCheckpoint::new(three_calls()); + cp.version = 999; + assert!(matches!(cp.validate(), Err(BatchError::Checkpoint(_)))); + assert!(BatchCheckpoint::from_json(&cp.to_json().unwrap()).is_err()); + } + + #[test] + fn retryable_excludes_succeeded_only() { + let mut cp = BatchCheckpoint::new(three_calls()); + cp.mark_succeeded(0, "hash-0").unwrap(); + cp.mark_failed(1, "boom").unwrap(); + // 2 is still Pending. + assert_eq!(cp.retryable_indices(), vec![1, 2]); + assert_eq!(cp.failed_indices(), vec![1]); + assert_eq!(cp.pending_indices(), vec![2]); + assert_eq!(cp.succeeded_count(), 1); + } + + #[test] + fn memory_store_roundtrips() { + let store = MemoryCheckpointStore::new(); + assert!(store.load().unwrap().is_none()); + let cp = BatchCheckpoint::new(three_calls()); + store.save(&cp).unwrap(); + assert_eq!(store.load().unwrap().unwrap(), cp); + store.clear().unwrap(); + assert!(store.load().unwrap().is_none()); + } + + #[test] + fn file_store_roundtrips_and_replaces_atomically() { + let dir = std::env::temp_dir(); + let path = dir.join(format!("stellar-aid-checkpoint-{}.json", std::process::id())); + let store = FileCheckpointStore::new(&path); + + assert!(store.load().unwrap().is_none()); + let mut cp = BatchCheckpoint::new(three_calls()); + store.save(&cp).unwrap(); + cp.mark_succeeded(0, "hash-0").unwrap(); + store.save(&cp).unwrap(); + + let loaded = store.load().unwrap().unwrap(); + assert_eq!(loaded.succeeded_count(), 1); + // The temp file must not survive a save. + assert!(!store.temp_path().exists()); + + store.clear().unwrap(); + assert!(store.load().unwrap().is_none()); + } + + #[test] + fn resume_errors_when_no_checkpoint() { + let store = MemoryCheckpointStore::new(); + assert!(ResumableBatch::resume(&store).is_err()); + } + + #[tokio::test] + async fn resume_after_interruption_skips_succeeded_and_retries_the_rest() { + // Simulate a first run that succeeded on call 0, then "crashed" + // before call 1 was recorded. + let store = MemoryCheckpointStore::new(); + let mut cp = BatchCheckpoint::new(three_calls()); + cp.mark_succeeded(0, "hash-0").unwrap(); + store.save(&cp).unwrap(); + + let mut batch = ResumableBatch::resume(&store).unwrap(); + let submitted = Arc::new(Mutex::new(Vec::::new())); + let seen = submitted.clone(); + + let summary = batch + .submit_remaining(&store, move |c| { + let seen = seen.clone(); + async move { + seen.lock().unwrap().push(c.contract_id.clone()); + Ok(format!("hash-{}", c.contract_id)) + } + }) + .await + .unwrap(); + + // Only calls 1 and 2 are re-submitted; call 0 is skipped. + assert_eq!(summary.attempted, 2); + assert_eq!(summary.succeeded, 2); + assert_eq!(summary.failed, 0); + assert_eq!(*submitted.lock().unwrap(), vec!["C2", "C3"]); + assert!(batch.checkpoint.is_complete()); + // The store reflects the completed run. + assert!(store.snapshot().unwrap().is_complete()); + } + + #[tokio::test] + async fn retry_only_failed_on_next_run() { + let store = MemoryCheckpointStore::new(); + store.save(&BatchCheckpoint::new(three_calls())).unwrap(); + + let attempts = Arc::new(AtomicUsize::new(0)); + let seen = attempts.clone(); + let mut batch = ResumableBatch::resume(&store).unwrap(); + + // First run: C1 succeeds, C2 fails, C3 succeeds. + let first = batch + .submit_remaining(&store, move |c| { + let seen = seen.clone(); + async move { + seen.fetch_add(1, Ordering::SeqCst); + if c.contract_id == "C2" { + Err("C2 rejected".into()) + } else { + Ok(format!("hash-{}", c.contract_id)) + } + } + }) + .await + .unwrap(); + assert_eq!(first, BatchSummary { attempted: 3, succeeded: 2, failed: 1 }); + + // Second run over a fresh resume: only the failed call is retried. + let mut batch = ResumableBatch::resume(&store).unwrap(); + let second_calls = Arc::new(Mutex::new(Vec::::new())); + let seen = second_calls.clone(); + let second = batch + .submit_remaining(&store, move |c| { + let seen = seen.clone(); + async move { + seen.lock().unwrap().push(c.contract_id.clone()); + Ok(format!("hash-{}", c.contract_id)) + } + }) + .await + .unwrap(); + + assert_eq!(second, BatchSummary { attempted: 1, succeeded: 1, failed: 0 }); + assert_eq!(*second_calls.lock().unwrap(), vec!["C2"]); + assert!(batch.checkpoint.is_complete()); + } + + #[tokio::test] + async fn checkpoint_is_persisted_after_each_call() { + // The submit_fn reads the store before each call, proving that the + // previous call's outcome was already durable by the time the next call + // starts — the one-call maximum-loss window, observed directly. + let store = Arc::new(MemoryCheckpointStore::new()); + store.save(&BatchCheckpoint::new(three_calls())).unwrap(); + let mut batch = ResumableBatch::resume(&*store).unwrap(); + + let observed = Arc::new(Mutex::new(Vec::::new())); + let seen = observed.clone(); + let store_for_fn = store.clone(); + let _ = batch + .submit_remaining(&*store, move |c| { + let seen = seen.clone(); + let store_for_fn = store_for_fn.clone(); + async move { + // Before submitting call N, everything before it is durable. + let done = store_for_fn.snapshot().map(|c| c.succeeded_count()).unwrap_or(0); + seen.lock().unwrap().push(done); + Ok(format!("hash-{}", c.contract_id)) + } + }) + .await + .unwrap(); + + // First call sees 0 persisted outcomes, then 1, then 2. + assert_eq!(*observed.lock().unwrap(), vec![0, 1, 2]); + } + + #[test] + fn builder_checkpoint_requeues_only_retryable() { + let mut builder = crate::batch::BatchBuilder::new(); + builder.add_call(call("C1")).unwrap(); + builder.add_call(call("C2")).unwrap(); + + let mut cp = builder.checkpoint(); + cp.mark_succeeded(0, "hash-0").unwrap(); + + let rebuilt = crate::batch::BatchBuilder::from_checkpoint(&cp); + assert_eq!(rebuilt.len(), 1); + } + + #[test] + fn fallback_rate_warning_fires_above_threshold() { + let mut metrics = crate::batch::BatchMetrics::default(); + metrics.batch_submissions = 1; + metrics.fallback_submissions = 9; // 90% + assert!(fallback_rate_warning(&metrics, 50.0).is_some()); + assert!(fallback_rate_warning(&metrics, 95.0).is_none()); + } +} diff --git a/sdk/src/config.rs b/sdk/src/config.rs index 090e828..4fe4fad 100644 --- a/sdk/src/config.rs +++ b/sdk/src/config.rs @@ -6,6 +6,18 @@ use thiserror::Error; pub enum ConfigError { #[error("Missing required environment variable: {0}")] MissingVar(String), + #[error("Invalid URL for {0}: {1}")] + InvalidUrl(String, String), + #[error("Unknown network passphrase: {0}")] + UnknownNetwork(String), + #[error( + "Network passphrase does not match network '{network}': expected '{expected}', got '{actual}'" + )] + NetworkMismatch { + network: String, + expected: String, + actual: String, + }, } /// Predefined network configurations for Stellar. @@ -72,6 +84,62 @@ impl Config { soroban_network_passphrase: network.network_passphrase.to_string(), } } + + /// Validate the *contents* of the configuration, not just their presence + /// (#980). + /// + /// Checks that both URLs parse as `http(s)` URLs, and that the network + /// passphrase matches the named network — or, when `stellar_network` is not + /// recognised, is at least one of the two known networks. A wrong + /// passphrase is the single most common integration mistake and produces + /// the most confusing failure, so it is caught here, immediately, with a + /// clear message. + pub fn validate(&self) -> Result<(), ConfigError> { + Self::validate_url("HORIZON_URL", &self.horizon_url)?; + Self::validate_url("SOROBAN_RPC_URL", &self.soroban_rpc_url)?; + + match Self::known_passphrase(&self.stellar_network) { + Some(expected) if self.soroban_network_passphrase != expected => { + Err(ConfigError::NetworkMismatch { + network: self.stellar_network.clone(), + expected: expected.to_string(), + actual: self.soroban_network_passphrase.clone(), + }) + } + Some(_) => Ok(()), + None => { + if matches!( + self.soroban_network_passphrase.as_str(), + TESTNET.network_passphrase | MAINNET.network_passphrase + ) { + Ok(()) + } else { + Err(ConfigError::UnknownNetwork( + self.soroban_network_passphrase.clone(), + )) + } + } + } + } + + fn validate_url(key: &str, value: &str) -> Result<(), ConfigError> { + match reqwest::Url::parse(value) { + Ok(u) + if matches!(u.scheme(), "http" | "https") && u.host_str().is_some() => + { + Ok(()) + } + _ => Err(ConfigError::InvalidUrl(key.to_string(), value.to_string())), + } + } + + fn known_passphrase(network: &str) -> Option<&'static str> { + match network.trim().to_ascii_lowercase().as_str() { + "testnet" => Some(TESTNET.network_passphrase), + "mainnet" | "public" => Some(MAINNET.network_passphrase), + _ => None, + } + } } #[cfg(test)] @@ -87,4 +155,51 @@ mod tests { let result = Config::from_env(); assert!(result.is_err()); } + + #[test] + fn validate_accepts_a_known_network() { + let cfg = Config::from_network(&TESTNET, "SDUMMY"); + assert!(cfg.validate().is_ok()); + let cfg = Config::from_network(&MAINNET, "SDUMMY"); + assert!(cfg.validate().is_ok()); + } + + #[test] + fn validate_rejects_a_wrong_passphrase_immediately() { + // Everything is present and the URL is fine, but the passphrase is for + // the other network — the most common integration mistake. + let mut cfg = Config::from_network(&TESTNET, "SDUMMY"); + cfg.soroban_network_passphrase = MAINNET.network_passphrase.to_string(); + let err = cfg.validate().unwrap_err(); + assert!(matches!(err, ConfigError::NetworkMismatch { .. })); + assert!(err.to_string().contains("does not match")); + } + + #[test] + fn validate_rejects_a_malformed_url() { + let mut cfg = Config::from_network(&TESTNET, "SDUMMY"); + cfg.horizon_url = "not a url".to_string(); + assert!(matches!( + cfg.validate().unwrap_err(), + ConfigError::InvalidUrl(_, _) + )); + + let mut cfg = Config::from_network(&TESTNET, "SDUMMY"); + cfg.soroban_rpc_url = "ftp://rpc.example.com".to_string(); + assert!(matches!( + cfg.validate().unwrap_err(), + ConfigError::InvalidUrl(_, _) + )); + } + + #[test] + fn validate_rejects_an_unknown_passphrase_for_an_unknown_network() { + let mut cfg = Config::from_network(&TESTNET, "SDUMMY"); + cfg.stellar_network = "futurenet".to_string(); + cfg.soroban_network_passphrase = "Future Network ; 2030".to_string(); + assert!(matches!( + cfg.validate().unwrap_err(), + ConfigError::UnknownNetwork(_) + )); + } } \ No newline at end of file diff --git a/sdk/src/errors.rs b/sdk/src/errors.rs index 05361a8..216be7d 100644 --- a/sdk/src/errors.rs +++ b/sdk/src/errors.rs @@ -32,6 +32,24 @@ pub enum StellarAidError { #[error("Batch error: {0}")] Batch(String), + + /// The endpoint could not be reached at all (DNS, refused connection, wrong + /// URL). Retryable (#980). + #[error("endpoint unreachable: {0}")] + EndpointUnreachable(String), + + /// The request was accepted but did not complete in time. Retryable (#980). + #[error("request timed out: {0}")] + RequestTimeout(String), + + /// The endpoint rejected the request as malformed (4xx). Not retryable + /// without changing the request (#980). + #[error("bad request: {0}")] + BadRequest(String), + + /// The chain rejected the transaction. Not retryable as-is (#980). + #[error("transaction rejected: {0}")] + TransactionRejected(String), } impl StellarAidError { @@ -66,6 +84,100 @@ impl StellarAidError { pub fn batch(msg: impl Into) -> Self { Self::Batch(msg.into()) } + + pub fn endpoint_unreachable(msg: impl Into) -> Self { + Self::EndpointUnreachable(msg.into()) + } + + pub fn request_timeout(msg: impl Into) -> Self { + Self::RequestTimeout(msg.into()) + } + + pub fn bad_request(msg: impl Into) -> Self { + Self::BadRequest(msg.into()) + } + + pub fn transaction_rejected(msg: impl Into) -> Self { + Self::TransactionRejected(msg.into()) + } + + /// Classify a transport-level [`reqwest::Error`] into the actionable + /// variants. A timeout and a refused connection are both retryable, but a + /// caller usually wants to tell them apart (#980). + pub fn from_reqwest(err: reqwest::Error) -> Self { + if err.is_timeout() { + Self::RequestTimeout(err.to_string()) + } else if err.is_connect() { + Self::EndpointUnreachable(err.to_string()) + } else if err.is_builder() { + // A malformed URL or request cannot be fixed by retrying. + Self::BadRequest(err.to_string()) + } else if err.is_request() { + Self::EndpointUnreachable(err.to_string()) + } else { + Self::NetworkError(err) + } + } + + /// Classify a non-success HTTP status. 4xx is a malformed request (fix the + /// request); everything else is treated as transient. + pub fn from_http_status(status: u16, body: impl Into) -> Self { + let body = body.into(); + if (400..500).contains(&status) { + Self::BadRequest(format!("HTTP {status}: {body}")) + } else { + Self::RequestTimeout(format!("HTTP {status}: {body}")) + } + } + + /// Whether retrying this error could plausibly succeed. + /// + /// This is the predicate [`crate::retry::retry_async_classified`] takes, and + /// the answer to issue #805: `retry_async` used to retry *every* error, + /// including an unrecoverable validation or rejection failure. + /// + /// Retryable: [`StellarAidError::NetworkError`], + /// [`StellarAidError::EndpointUnreachable`], + /// [`StellarAidError::RequestTimeout`]. + /// + /// Not retryable: validation, bad-request, rejected-transaction, contract, + /// keypair, wallet, compression and batch errors — the server-side + /// [`StellarAidError::HorizonError`]/[`StellarAidError::SorobanError`] + /// variants carry a rejection and are likewise not retried. + pub fn is_retryable(&self) -> bool { + matches!( + self, + Self::NetworkError(_) | Self::EndpointUnreachable(_) | Self::RequestTimeout(_) + ) + } } -pub type Result = std::result::Result; \ No newline at end of file +pub type Result = std::result::Result; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_transport_errors_are_retryable() { + assert!(StellarAidError::endpoint_unreachable("dns").is_retryable()); + assert!(StellarAidError::request_timeout("slow").is_retryable()); + assert!(!StellarAidError::bad_request("HTTP 400").is_retryable()); + assert!(!StellarAidError::transaction_rejected("tx failed").is_retryable()); + assert!(!StellarAidError::validation("amount must be positive").is_retryable()); + } + + #[test] + fn http_status_is_classified() { + assert!(matches!( + StellarAidError::from_http_status(400, "bad"), + StellarAidError::BadRequest(_) + )); + assert!(matches!( + StellarAidError::from_http_status(404, "missing"), + StellarAidError::BadRequest(_) + )); + // 5xx is transient, not a malformed request. + assert!(StellarAidError::from_http_status(503, "unavailable").is_retryable()); + } +} \ No newline at end of file diff --git a/sdk/src/health.rs b/sdk/src/health.rs new file mode 100644 index 0000000..7199f01 --- /dev/null +++ b/sdk/src/health.rs @@ -0,0 +1,80 @@ +//! Typed endpoint health for the SDK's network clients (#980). +//! +//! [`crate::soroban::rpc_client::SorobanRpcClient::health`] and +//! [`crate::horizon::client::HorizonClient::health`] return an [`EndpointHealth`] +//! so a deployment script can tell *which* of the two endpoints is reachable +//! before using it. This is for setup and diagnostics; it is deliberately **not** +//! a precondition on every SDK operation. +//! +//! Distinguishing outcomes: +//! +//! * `Ok(EndpointHealth { healthy: true, .. })` — the endpoint answered. +//! * `Ok(EndpointHealth { healthy: false, .. })` — the endpoint answered but +//! reported itself unhealthy (e.g. a non-2xx status). +//! * `Err(StellarAidError::EndpointUnreachable | RequestTimeout)` — the endpoint +//! did not answer at all. These are the retryable variants. + +use std::time::Duration; + +/// Timeout applied to a single health probe. Short by design, so a probe can +/// never hang the caller. +pub const HEALTH_PROBE_TIMEOUT: Duration = Duration::from_secs(3); + +/// Result of probing a single endpoint. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct EndpointHealth { + /// The URL that was probed. + pub endpoint: String, + /// Whether the endpoint reported itself healthy. + pub healthy: bool, + /// Round-trip time for the probe, in milliseconds. + pub latency_ms: u128, + /// Extra context (server message or status) when `healthy` is false. + pub detail: Option, +} + +impl EndpointHealth { + /// A successful probe. + pub fn up(endpoint: impl Into, latency_ms: u128) -> Self { + Self { + endpoint: endpoint.into(), + healthy: true, + latency_ms, + detail: None, + } + } + + /// The endpoint answered, but is not healthy. + pub fn down( + endpoint: impl Into, + detail: impl Into, + latency_ms: u128, + ) -> Self { + Self { + endpoint: endpoint.into(), + healthy: false, + latency_ms, + detail: Some(detail.into()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn up_has_no_detail() { + let h = EndpointHealth::up("https://rpc", 12); + assert!(h.healthy); + assert_eq!(h.latency_ms, 12); + assert!(h.detail.is_none()); + } + + #[test] + fn down_carries_detail() { + let h = EndpointHealth::down("https://rpc", "HTTP 503", 7); + assert!(!h.healthy); + assert_eq!(h.detail.as_deref(), Some("HTTP 503")); + } +} diff --git a/sdk/src/horizon/client.rs b/sdk/src/horizon/client.rs index 2d48ad8..474729d 100644 --- a/sdk/src/horizon/client.rs +++ b/sdk/src/horizon/client.rs @@ -1,5 +1,8 @@ +use crate::errors::StellarAidError; +use crate::health::{EndpointHealth, HEALTH_PROBE_TIMEOUT}; use reqwest::Client; use serde::Deserialize; +use std::time::Instant; use thiserror::Error; #[derive(Debug, Default)] @@ -138,4 +141,56 @@ impl HorizonClient { } Ok(resp.json().await?) } + + /// Typed liveness probe (#980): hits Horizon's root endpoint. + /// + /// Returns `Ok(EndpointHealth)` when Horizon answers (with `healthy` + /// reflecting the status), `Err(BadRequest)` on a 4xx — almost always a + /// wrong URL — and `Err(EndpointUnreachable | RequestTimeout)` when it does + /// not answer within [`HEALTH_PROBE_TIMEOUT`]. Bounded by that short timeout + /// so a probe can never hang. + pub async fn health(&self) -> Result { + let url = format!("{}/", self.base_url.trim_end_matches('/')); + let started = Instant::now(); + let probe = tokio::time::timeout(HEALTH_PROBE_TIMEOUT, self.client.get(&url).send()).await; + let elapsed = started.elapsed().as_millis(); + + match probe { + Err(_) => Err(StellarAidError::request_timeout(format!( + "Horizon health probe to {url} exceeded {HEALTH_PROBE_TIMEOUT:?}" + ))), + Ok(Err(e)) => Err(StellarAidError::from_reqwest(e)), + Ok(Ok(resp)) => { + let status = resp.status(); + if status.is_success() { + Ok(EndpointHealth::up(url, elapsed)) + } else if status.is_client_error() { + Err(StellarAidError::from_http_status( + status.as_u16(), + status.canonical_reason().unwrap_or("client error"), + )) + } else { + Ok(EndpointHealth::down(url, format!("HTTP {status}"), elapsed)) + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn health_rejects_a_malformed_url_without_retrying() { + // A relative URL fails at request build; this must be a clear, + // non-retryable error rather than a hang. + let client = HorizonClient::new("not-a-valid-url"); + let err = client.health().await.unwrap_err(); + assert!(matches!( + err, + StellarAidError::BadRequest(_) | StellarAidError::EndpointUnreachable(_) + )); + assert!(!matches!(err, StellarAidError::RequestTimeout(_))); + } } diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index 73c5cfe..e540542 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -1,7 +1,9 @@ pub mod batch; +pub mod batch_checkpoint; pub mod compression; pub mod config; pub mod errors; +pub mod health; pub mod horizon; pub mod logging; pub mod retry; diff --git a/sdk/src/retry.rs b/sdk/src/retry.rs index e843120..314c19d 100644 --- a/sdk/src/retry.rs +++ b/sdk/src/retry.rs @@ -1,3 +1,4 @@ +use crate::errors::StellarAidError; use rand::Rng; use tokio::time::{sleep, Duration}; use tracing::{warn, info}; @@ -67,6 +68,80 @@ where } } +/// Like [`retry_async`], but consults a classifier and stops immediately on a +/// non-retryable error (#980, and the mechanism issue #805 needs). +/// +/// `retry_async` retries *every* error, so a malformed request or a rejected +/// transaction is retried the full `max_attempts` before the caller sees it. +/// Pass [`StellarAidError::is_retryable`] here — or use +/// [`retry_stellar_async`] — to retry only transport failures. +/// +/// This is complementary to [`crate::batch_checkpoint`]: the classifier bounds +/// retries *within* a run, the checkpoint survives a run that ended. +pub async fn retry_async_classified( + config: &RetryConfig, + is_retryable: impl Fn(&E) -> bool, + f: F, +) -> Result +where + F: Fn() -> Fut, + Fut: std::future::Future>, + E: std::fmt::Display, +{ + let mut attempt = 0u32; + loop { + attempt += 1; + match f().await { + Ok(val) => { + if attempt > 1 { + info!(attempt = attempt, "operation succeeded after retry"); + } + return Ok(val); + } + Err(e) => { + if !is_retryable(&e) { + warn!( + attempt = attempt, + error = %e, + "non-retryable error; returning without another attempt" + ); + return Err(e); + } + if attempt >= config.max_attempts { + warn!( + attempt = attempt, + max_attempts = config.max_attempts, + error = %e, + "all retry attempts exhausted" + ); + return Err(e); + } + let delay = calculate_delay(config, attempt); + warn!( + attempt = attempt, + max_attempts = config.max_attempts, + delay_ms = delay, + error = %e, + "retryable operation failed, retrying" + ); + sleep(Duration::from_millis(delay)).await; + } + } + } +} + +/// [`retry_async_classified`] specialized to the SDK's own [`StellarAidError`]. +pub async fn retry_stellar_async( + config: &RetryConfig, + f: F, +) -> Result +where + F: Fn() -> Fut, + Fut: std::future::Future>, +{ + retry_async_classified(config, StellarAidError::is_retryable, f).await +} + fn calculate_delay(config: &RetryConfig, attempt: u32) -> u64 { let base = config.base_delay_ms as f64 * config.backoff_factor.powi(attempt as i32 - 1); let clamped = (base as u64).min(config.max_delay_ms); @@ -176,4 +251,87 @@ mod tests { let unique = delays.iter().collect::>().len(); assert!(unique > 1, "jitter should produce varying delays"); } + + // -- classified retry (#980) -- + + #[tokio::test] + async fn classified_stops_immediately_on_non_retryable() { + let config = RetryConfig { + max_attempts: 5, + base_delay_ms: 1, + jitter_ms: 0, + ..Default::default() + }; + let counter = AtomicU32::new(0); + let result = retry_async_classified(&config, |_e: &&str| false, || async { + counter.fetch_add(1, Ordering::SeqCst); + Err::<&str, &str>("permanent") + }) + .await; + + assert_eq!(result.unwrap_err(), "permanent"); + // One attempt only — the classifier said do not retry. + assert_eq!(counter.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn classified_retries_retryable_errors() { + let config = RetryConfig { + max_attempts: 3, + base_delay_ms: 1, + jitter_ms: 0, + ..Default::default() + }; + let counter = AtomicU32::new(0); + let result = retry_async_classified(&config, |_e: &&str| true, || async { + let count = counter.fetch_add(1, Ordering::SeqCst); + if count < 2 { + Err::<&str, &str>("transient") + } else { + Ok("ok") + } + }) + .await; + + assert_eq!(result.unwrap(), "ok"); + assert_eq!(counter.load(Ordering::SeqCst), 3); + } + + #[tokio::test] + async fn stellar_retry_does_not_retry_validation_errors() { + let config = RetryConfig { + max_attempts: 5, + base_delay_ms: 1, + jitter_ms: 0, + ..Default::default() + }; + let counter = AtomicU32::new(0); + let result = retry_stellar_async(&config, || async { + counter.fetch_add(1, Ordering::SeqCst); + Err::<(), StellarAidError>(StellarAidError::validation("amount must be positive")) + }) + .await; + + assert!(result.is_err()); + assert_eq!(counter.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn stellar_retry_retries_unreachable_endpoint() { + let config = RetryConfig { + max_attempts: 4, + base_delay_ms: 1, + jitter_ms: 0, + ..Default::default() + }; + let counter = AtomicU32::new(0); + let result = retry_stellar_async(&config, || async { + counter.fetch_add(1, Ordering::SeqCst); + Err::<(), StellarAidError>(StellarAidError::endpoint_unreachable("refused")) + }) + .await; + + assert!(result.is_err()); + assert_eq!(counter.load(Ordering::SeqCst), 4); + } } diff --git a/sdk/src/soroban/rpc_client.rs b/sdk/src/soroban/rpc_client.rs index a25c735..aa0df74 100644 --- a/sdk/src/soroban/rpc_client.rs +++ b/sdk/src/soroban/rpc_client.rs @@ -1,7 +1,10 @@ use crate::compression::{self, CompressionFormat, CompressionStats}; +use crate::errors::StellarAidError; +use crate::health::{EndpointHealth, HEALTH_PROBE_TIMEOUT}; use reqwest::Client; use serde::{Deserialize, Serialize}; use std::sync::Arc; +use std::time::Instant; use thiserror::Error; use tokio::sync::Mutex; @@ -171,6 +174,39 @@ impl SorobanRpcClient { Ok(()) } + /// Typed liveness probe (#980). + /// + /// Returns `Ok` when the node answers `getHealth` — with `healthy` reflecting + /// that answer — and distinguishes the two failure modes a caller cares + /// about: `Err(EndpointUnreachable)` when the node cannot be reached at all, + /// and `Err(RequestTimeout)` when it does not answer within + /// [`HEALTH_PROBE_TIMEOUT`]. The probe is bounded by that short timeout so it + /// can never hang; it is for setup and diagnostics, not a precondition on + /// every operation. + pub async fn health(&self) -> Result { + let started = Instant::now(); + let probe = tokio::time::timeout( + HEALTH_PROBE_TIMEOUT, + self.call::("getHealth", serde_json::json!({})), + ) + .await; + let elapsed = started.elapsed().as_millis(); + + match probe { + Ok(Ok(_)) => Ok(EndpointHealth::up(self.rpc_url.clone(), elapsed)), + Ok(Err(RpcError::Http(e))) => Err(StellarAidError::from_reqwest(e)), + Ok(Err(other)) => Ok(EndpointHealth::down( + self.rpc_url.clone(), + other.to_string(), + elapsed, + )), + Err(_) => Err(StellarAidError::request_timeout(format!( + "Soroban RPC health probe to {} exceeded {:?}", + self.rpc_url, HEALTH_PROBE_TIMEOUT + ))), + } + } + #[tracing::instrument(skip(self), fields(hash))] pub async fn get_transaction_status(&self, hash: &str) -> Result { let result: TxStatusResult = self diff --git a/sdk/src/transaction_builder.rs b/sdk/src/transaction_builder.rs index e0128bf..fd73860 100644 --- a/sdk/src/transaction_builder.rs +++ b/sdk/src/transaction_builder.rs @@ -83,7 +83,7 @@ pub async fn build_donate_transaction_full( .map_err(|_| StellarAidError::validation("invalid contract id"))?; let contract_addr = match &contract_raw { - stellar_strkey::Strkey::ContractId(h) => ScAddress::Contract(Hash(Uint256(h.0))), + stellar_strkey::Strkey::Contract(c) => ScAddress::Contract(Hash(Uint256(c.0))), _ => return Err(StellarAidError::validation("expected a C... contract id")), }; @@ -104,8 +104,8 @@ pub async fn build_donate_transaction_full( let raw = stellar_strkey::Strkey::from_string(addr) .map_err(|_| StellarAidError::validation("invalid token address"))?; match &raw { - stellar_strkey::Strkey::ContractId(h) => { - ScVal::Address(ScAddress::Contract(Hash(Uint256(h.0)))) + stellar_strkey::Strkey::Contract(c) => { + ScVal::Address(ScAddress::Contract(Hash(Uint256(c.0)))) } _ => return Err(StellarAidError::validation("expected a C... contract id for token")), } diff --git a/worker/Cargo.toml b/worker/Cargo.toml index 0f8c731..f79f396 100644 --- a/worker/Cargo.toml +++ b/worker/Cargo.toml @@ -20,3 +20,7 @@ tracing-subscriber = { workspace = true } axum = { version = "0.7", features = ["macros"] } tower-http = { version = "0.5", features = ["cors", "timeout", "trace"] } sdk = { path = "../sdk" } +# Used by `webhooks::WebhookManager::verify_signature` (HMAC-SHA256). +hmac = "0.12" +sha2 = "0.10" +hex = "0.4" diff --git a/worker/src/main.rs b/worker/src/main.rs index b51d370..9407df5 100644 --- a/worker/src/main.rs +++ b/worker/src/main.rs @@ -22,7 +22,7 @@ use axum::{ use sdk::{ errors::StellarAidError, retry::{retry_async, RetryConfig}, - soroban::rpc_client::SorobanRpcClient, + soroban::rpc_client::{SimulationResult, SorobanRpcClient}, transaction_builder::{build_donate_transaction_full, DonationParams, NetworkConfig}, }; use serde::{Deserialize, Serialize}; @@ -135,6 +135,12 @@ fn build_route_limits() -> HashMap { "/api/donations/submit".to_string(), RateLimit::new(env_u64("SUBMIT_RATE_LIMIT_MAX", 10) as u32, window), ); + // The dry-run validates by simulating, which is an RPC call that amplifies + // load, so it gets its own (tighter) budget (#981). + limits.insert( + "/api/donations/validate".to_string(), + RateLimit::new(env_u64("VALIDATE_RATE_LIMIT_MAX", 5) as u32, window), + ); for route in ["/health", "/ready", "/metrics"] { limits.insert( route.to_string(), @@ -251,6 +257,176 @@ async fn submit_donation( })) } +/// Maximum memo length, mirroring `shared::validation::MAX_MEMO_LEN`. +const MAX_MEMO_LEN: usize = 256; + +/// Returned with every dry-run result: a simulation is a snapshot, not a +/// guarantee. +const SIMULATION_STALE_NOTE: &str = + "the simulation reflects the ledger at probe time; campaign state can change before submission"; + +/// Result of a dry-run validation (#981). +#[derive(Debug, Serialize)] +pub struct ValidateDonationResponse { + pub valid: bool, + pub gas_consumed: Option, + pub contract_error: Option, + /// Contract error code, when one can be extracted, so a client can branch + /// on the number rather than parsing prose. + pub error_code: Option, + pub reason: Option, + pub note: &'static str, +} + +impl ValidateDonationResponse { + fn rejected(reason: impl Into) -> Self { + Self { + valid: false, + gas_consumed: None, + contract_error: None, + error_code: None, + reason: Some(reason.into()), + note: SIMULATION_STALE_NOTE, + } + } +} + +/// Cheap local validation, run before spending a simulation round trip (#981). +/// +/// Returns the reason for the first violation. Chain-side checks — does the +/// campaign exist, is it past its deadline, is it already funded — are the +/// simulation's job, not this function's. +fn validate_donation_request(req: &SubmitDonationRequest) -> Result<(), String> { + if req.donor.trim().is_empty() { + return Err("donor must not be empty".to_string()); + } + if !req.donor.starts_with('G') { + return Err("donor must be a G... stellar address".to_string()); + } + if req.amount <= 0 { + return Err("amount must be positive".to_string()); + } + if req.campaign_id == 0 { + return Err("campaign_id must be non-zero".to_string()); + } + if let Some(memo) = &req.memo { + if memo.len() > MAX_MEMO_LEN { + return Err(format!("memo must be at most {} bytes", MAX_MEMO_LEN)); + } + } + Ok(()) +} + +/// Best-effort extraction of a Soroban contract error code from a simulation +/// error string. Soroban renders contract errors as e.g. +/// `HostError: Error(Contract, #4)`; the first `#` is the code. +fn extract_error_code(message: &str) -> Option { + let hash = message.find('#')?; + let digits: String = message[hash + 1..] + .chars() + .take_while(|c| c.is_ascii_digit()) + .collect(); + digits.parse().ok() +} + +/// `minResourceFee` arrives as a decimal string from some nodes and as a number +/// from others; accept both. +fn min_resource_fee(cost: &serde_json::Value) -> Option { + let v = cost.get("minResourceFee")?; + v.as_u64() + .or_else(|| v.as_str().and_then(|s| s.parse().ok())) +} + +/// Turn a simulation outcome into a dry-run response. +/// +/// Pure — no I/O — so the success, contract-error and RPC-failure paths are +/// testable without a network round trip, and so the handler has exactly one +/// place that decides `valid` (#981). +fn interpret_simulation(outcome: Result) -> ValidateDonationResponse { + match outcome { + Ok(sim) => { + let gas_consumed = sim.cost.as_ref().and_then(min_resource_fee); + match sim.error { + Some(err) => ValidateDonationResponse { + valid: false, + gas_consumed, + error_code: extract_error_code(&err), + contract_error: Some(err), + reason: Some("simulation reported a contract error".to_string()), + note: SIMULATION_STALE_NOTE, + }, + None => ValidateDonationResponse { + valid: true, + gas_consumed, + contract_error: None, + error_code: None, + reason: None, + note: SIMULATION_STALE_NOTE, + }, + } + } + Err(message) => ValidateDonationResponse { + valid: false, + gas_consumed: None, + error_code: extract_error_code(&message), + contract_error: Some(message), + reason: Some("could not simulate; the RPC endpoint may be unavailable".to_string()), + note: SIMULATION_STALE_NOTE, + }, + } +} + +/// Dry-run a donation: validate and simulate, without submitting (#981). +/// +/// This is additive; `/api/donations/submit` is unchanged. Submit returns a +/// signed XDR for the client to broadcast; validate reports whether the call +/// *would* succeed and, if not, the contract's own error, so the failure is +/// free instead of a wasted on-chain transaction. +/// +/// Simulation is a real RPC call, so this route has its own rate-limit budget +/// and is never on the submit path. It builds the same XDR the submit path +/// would, then calls the genuine `simulateTransaction` RPC — deliberately not +/// the placeholder `TransactionSimulator`, which returns success unconditionally +/// (see issue #795). +async fn validate_donation( + State(state): State>, + Json(req): Json, +) -> Result, (StatusCode, Json)> { + // 1. Cheap rejections first — no round trip. + if let Err(reason) = validate_donation_request(&req) { + return Ok(Json(ValidateDonationResponse::rejected(reason))); + } + + let params = DonationParams { + donor: req.donor, + campaign_id: req.campaign_id, + amount: req.amount, + token_address: req.token_address, + anonymous: req.anonymous.unwrap_or(false), + memo: req.memo, + donation_contract_id: state.donation_contract_id.clone(), + }; + let network = state.network_config.clone(); + + // 2. Build the XDR the submit path would build, without returning it. + let xdr = match build_donate_transaction_full(¶ms, &network).await { + Ok(xdr) => xdr, + Err(e) => { + return Ok(Json(ValidateDonationResponse::rejected(format!( + "could not build the transaction: {e}" + )))); + } + }; + + // 3. Simulate against the real RPC path to learn the contract's error. + let rpc = SorobanRpcClient::new(&state.network_config.rpc_url); + let outcome = rpc + .simulate_transaction(&xdr) + .await + .map_err(|e| e.to_string()); + Ok(Json(interpret_simulation(outcome))) +} + async fn get_donation( State(state): State>, Path(tx_hash): Path, @@ -415,6 +591,7 @@ async fn main() { .route("/ready", get(readiness)) .route("/metrics", get(metrics_endpoint)) .route("/api/donations/submit", post(submit_donation)) + .route("/api/donations/validate", post(validate_donation)) .route("/api/donations/{tx_hash}", get(get_donation)) .layer(TimeoutLayer::new(request_timeout)) .layer(middleware::from_fn_with_state(state.clone(), rate_limit)) @@ -440,3 +617,99 @@ async fn main() { shutdown_complete.store(true, Ordering::SeqCst); info!("StellarAid worker stopped"); } + +#[cfg(test)] +mod tests { + use super::*; + + fn req( + donor: &str, + campaign_id: u64, + amount: i128, + memo: Option<&str>, + ) -> SubmitDonationRequest { + SubmitDonationRequest { + donor: donor.to_string(), + campaign_id, + amount, + token_address: None, + anonymous: None, + memo: memo.map(|m| m.to_string()), + } + } + + #[test] + fn local_validation_accepts_a_valid_request() { + assert!(validate_donation_request(&req("GABC123", 1, 100, Some("thanks"))).is_ok()); + } + + #[test] + fn local_validation_rejects_an_invalid_campaign_id() { + let err = validate_donation_request(&req("GABC123", 0, 100, None)).unwrap_err(); + assert!(err.contains("campaign_id")); + } + + #[test] + fn local_validation_rejects_a_zero_amount() { + let err = validate_donation_request(&req("GABC123", 1, 0, None)).unwrap_err(); + assert!(err.contains("amount")); + } + + #[test] + fn local_validation_rejects_an_empty_donor() { + assert!(validate_donation_request(&req("", 1, 100, None)).is_err()); + assert!(validate_donation_request(&req("SABC123", 1, 100, None)).is_err()); + } + + #[test] + fn local_validation_rejects_an_over_long_memo() { + let long = "x".repeat(MAX_MEMO_LEN + 1); + assert!(validate_donation_request(&req("GABC123", 1, 100, Some(long.as_str()))).is_err()); + } + + #[test] + fn contract_error_codes_are_extractable() { + assert_eq!(extract_error_code("HostError: Error(Contract, #4)"), Some(4)); + assert_eq!(extract_error_code("Error(Contract, #1042)"), Some(1042)); + assert_eq!(extract_error_code("no numeric code"), None); + } + + #[test] + fn simulation_success_reports_gas() { + let sim = SimulationResult { + cost: Some(serde_json::json!({ "minResourceFee": "12345" })), + results: None, + error: None, + }; + let resp = interpret_simulation(Ok(sim)); + assert!(resp.valid); + assert_eq!(resp.gas_consumed, Some(12345)); + assert!(resp.contract_error.is_none()); + assert!(resp.note.contains("stale") || resp.note.contains("probe time")); + } + + #[test] + fn simulation_contract_error_reports_message_and_code() { + let sim = SimulationResult { + cost: Some(serde_json::json!({ "minResourceFee": 100 })), + results: None, + error: Some("HostError: Error(Contract, #7)".to_string()), + }; + let resp = interpret_simulation(Ok(sim)); + assert!(!resp.valid); + assert_eq!(resp.error_code, Some(7)); + assert_eq!( + resp.contract_error.as_deref(), + Some("HostError: Error(Contract, #7)") + ); + assert_eq!(resp.gas_consumed, Some(100)); + } + + #[test] + fn simulation_rpc_failure_is_reported_without_network() { + let resp = interpret_simulation(Err("connection refused".to_string())); + assert!(!resp.valid); + assert!(resp.error_code.is_none()); + assert_eq!(resp.contract_error.as_deref(), Some("connection refused")); + } +}