diff --git a/Cargo.lock b/Cargo.lock index cd709200bb9..224c403b437 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5497,6 +5497,7 @@ dependencies = [ "platform-wallet", "rs-sdk-ffi", "serde_json", + "tempfile", "tokio", "tokio-metrics", "tracing", diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/errors/DashSdkError.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/errors/DashSdkError.kt index 3227c3d606b..3c2bde003ab 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/errors/DashSdkError.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/errors/DashSdkError.kt @@ -84,6 +84,13 @@ sealed class DashSdkError( class DriveInternalError(message: String, cause: Throwable? = null) : DashSdkError(message, cause) + /** The transition may have executed; reconcile it before submitting again. */ + class StateTransitionResultUnconfirmed(message: String, cause: Throwable? = null) : + DashSdkError(message, cause) { + override val userMessage: String get() = + "Submission result is unconfirmed. Check history and reconcile the original operation before submitting again." + } + class InternalError(message: String, cause: Throwable? = null) : DashSdkError(message, cause) @@ -322,16 +329,15 @@ sealed class DashSdkError( ) /** - * `ErrorTransactionBroadcastUnconfirmed` (native code 20). A core - * transaction broadcast had an AMBIGUOUS outcome — it may already be - * on the network. The wallet keeps the spent inputs reserved so a - * retry can't double-spend; the reservation TTL or a later sync - * reconciles the outcome. Do **NOT** auto-retry. + * `ErrorTransactionBroadcastUnconfirmed` (native code 20). A Core + * transaction or Platform state transition may already have executed, + * but its result is unconfirmed. Reconcile the original operation + * before submitting again. Do **NOT** auto-retry. */ class TransactionBroadcastUnconfirmed(message: String, cause: Throwable? = null) : PlatformWallet( - "$message (do NOT retry: the transaction may already be on the network; " + - "the wallet keeps its inputs reserved until a sync reconciles the outcome)", + "$message (do NOT retry: the submission may already have executed; " + + "reconcile the original operation before submitting again)", cause, ) @@ -659,7 +665,7 @@ sealed class DashSdkError( /** * Offset the JNI bridge adds to a `PlatformWalletFFIResultCode` * before throwing it as a `DashSDKException`, keeping it clear of the - * rs-sdk-ffi `DashSDKErrorCode` range (1–10). Must stay in lockstep + * rs-sdk-ffi `DashSDKErrorCode` range (1–11, 99). Must stay in lockstep * with `support::PWFFI_CODE_OFFSET` in `rs-unified-sdk-jni`. */ const val PLATFORM_WALLET_CODE_OFFSET = 1000 @@ -689,6 +695,7 @@ sealed class DashSdkError( 8 -> Timeout(message, e) 9 -> NotImplemented(message, e) 10 -> DriveInternalError(message, e) + 11 -> StateTransitionResultUnconfirmed(message, e) else -> InternalError(message, e) } } diff --git a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/DashSDKException.kt b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/DashSDKException.kt index ff35b52834f..dfab0175dd2 100644 --- a/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/DashSDKException.kt +++ b/packages/kotlin-sdk/sdk/src/main/kotlin/org/dashfoundation/dashsdk/ffi/DashSDKException.kt @@ -13,7 +13,8 @@ import org.dashfoundation.dashsdk.errors.PlatformConsensusError * [code] values mirror `DashSDKErrorCode` in `rs-sdk-ffi/src/error.rs`: * 0=Success, 1=InvalidParameter, 2=InvalidState, 3=NetworkError, * 4=SerializationError, 5=ProtocolError, 6=CryptoError, 7=NotFound, - * 8=Timeout, 9=NotImplemented, 10=DriveInternalError, 99=InternalError. + * 8=Timeout, 9=NotImplemented, 10=DriveInternalError, + * 11=StateTransitionResultUnconfirmed, 99=InternalError. * * [consensusError] is the consensus rejection behind the failure when the * native result or `DashSDKError` carried one, `null` otherwise. diff --git a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/errors/DashSdkErrorTest.kt b/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/errors/DashSdkErrorTest.kt index fd09ae4051a..9693c023771 100644 --- a/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/errors/DashSdkErrorTest.kt +++ b/packages/kotlin-sdk/sdk/src/test/kotlin/org/dashfoundation/dashsdk/errors/DashSdkErrorTest.kt @@ -33,6 +33,7 @@ class DashSdkErrorTest { 8 to DashSdkError.Timeout::class, 9 to DashSdkError.NotImplemented::class, 10 to DashSdkError.DriveInternalError::class, + 11 to DashSdkError.StateTransitionResultUnconfirmed::class, 99 to DashSdkError.InternalError::class, ) cases.forEach { (code, expected) -> @@ -42,6 +43,19 @@ class DashSdkErrorTest { } } + @Test + fun shouldPreserveUnconfirmedSubmissionWithoutRetrying() { + val detail = "Identity top-up may have executed: wait timeout after nonce rejection" + val native = DashSDKException(11, detail) + val mapped = DashSdkError.fromNative(native) + assertTrue(mapped is DashSdkError.StateTransitionResultUnconfirmed) + assertEquals(detail, mapped.message) + assertEquals(native, mapped.cause) + assertNull(mapped.consensusError) + assertFalse(mapped.isRetryable) + assertTrue(mapped.userMessage.contains("before submitting again")) + } + @Test fun shouldPreserveShieldedIdentityDebitPendingWithoutClaimingSubmission() { val message = "Identity has an unresolved shielded debit; " + @@ -181,6 +195,8 @@ class DashSdkErrorTest { ) // The message must warn against retrying (distinct from the anchor case). assertTrue(broadcastUnconfirmed.message!!.contains("do NOT retry")) + assertTrue(broadcastUnconfirmed.message!!.contains("reconcile the original operation")) + assertFalse(broadcastUnconfirmed.message!!.contains("inputs reserved")) // Definitive broadcast rejection (26) must reach callers as its own type, // NOT as Generic: it is the definitive counterpart to the ambiguous diff --git a/packages/rs-dapi-client/src/dapi_client.rs b/packages/rs-dapi-client/src/dapi_client.rs index 36685a14ce0..94775ab4a1d 100644 --- a/packages/rs-dapi-client/src/dapi_client.rs +++ b/packages/rs-dapi-client/src/dapi_client.rs @@ -1101,6 +1101,32 @@ impl DapiRequestExecutor for DapiClient { request: R, settings: RequestSettings, ) -> ExecutionResult + where + R: TransportRequest + Mockable, + R::Response: Mockable, + TransportError: Mockable, + { + self.execute_with_transport_error_observer(request, settings, &|_| {}) + .await + } +} + +impl DapiClient { + /// Execute with the ordinary retry policy, observing each failed dispatched + /// attempt before its error can be replaced by a later attempt's result. + /// + /// The observer also sees client-side attempt deadlines. It is not called + /// for local request refusal, address exhaustion or client construction + /// errors, where the request has not been dispatched. It cannot change the + /// returned result, retry budget, address selection or banning policy. + // Keep the established execute() result and error layout unchanged. + #[allow(clippy::result_large_err)] + pub async fn execute_with_transport_error_observer( + &self, + request: R, + settings: RequestSettings, + observer: &(dyn Fn(&TransportError) + Send + Sync), + ) -> ExecutionResult where R: TransportRequest + Mockable, R::Response: Mockable, @@ -1275,6 +1301,7 @@ impl DapiRequestExecutor for DapiClient { }) } Err(transport_error) => { + observer(&transport_error); tracing::debug!(error = ?transport_error, "received error: {transport_error}"); Err(ExecutionError { inner: DapiClientError::Transport(transport_error), diff --git a/packages/rs-dapi-client/src/transport/tonic_channel.rs b/packages/rs-dapi-client/src/transport/tonic_channel.rs index 7d311adb33c..ddcf54ad538 100644 --- a/packages/rs-dapi-client/src/transport/tonic_channel.rs +++ b/packages/rs-dapi-client/src/transport/tonic_channel.rs @@ -124,6 +124,47 @@ mod tests { use super::*; use crate::transport::{ProxyEndpoint, Socks5Auth, Socks5Proxy}; use crate::RequestSettings; + use dapi_grpc::platform::v0::BroadcastStateTransitionRequest; + use dapi_grpc::tonic::{Code, ConnectError}; + use hyper_util::rt::TokioIo; + use std::io; + use tokio::io::DuplexStream; + use tower::service_fn; + + #[tokio::test] + async fn should_preserve_typed_connector_failure_in_real_grpc_status() { + for kind in [ + io::ErrorKind::ConnectionRefused, + io::ErrorKind::NotFound, + io::ErrorKind::InvalidData, + ] { + // Exercise tonic's actual connector -> channel -> generated unary + // request path. The failing connector never opens a socket. + let channel = Channel::builder("http://unreachable.invalid".parse().unwrap()) + .connect_with_connector_lazy(service_fn(move |_| async move { + Err::, _>(io::Error::new(kind, "establishment failed")) + })); + let mut client = PlatformGrpcClient::new(channel); + let status = client + .broadcast_state_transition(BroadcastStateTransitionRequest::default()) + .await + .unwrap_err(); + assert_eq!(status.code(), Code::Unavailable); + let mut source = status.source(); + let mut found = false; + while let Some(error) = source { + if error.is::() { + found = true; + break; + } + source = error.source(); + } + assert!( + found, + "the SDK can distinguish establishment from a lost ACK: {status:?}" + ); + } + } #[test] fn should_strip_only_ipv6_brackets() { diff --git a/packages/rs-platform-wallet-ffi/Cargo.toml b/packages/rs-platform-wallet-ffi/Cargo.toml index fd8a735f80a..549a9a75577 100644 --- a/packages/rs-platform-wallet-ffi/Cargo.toml +++ b/packages/rs-platform-wallet-ffi/Cargo.toml @@ -65,6 +65,7 @@ bs58 = "0.5" zeroize = { version = "1", features = ["derive"] } [dev-dependencies] +tempfile = "3.27.0" dpp = { path = "../rs-dpp", features = ["fixtures-and-mocks"] } platform-wallet = { path = "../rs-platform-wallet", features = ["test-utils"] } diff --git a/packages/rs-platform-wallet-ffi/src/error.rs b/packages/rs-platform-wallet-ffi/src/error.rs index 0bc67f532c9..26a49873944 100644 --- a/packages/rs-platform-wallet-ffi/src/error.rs +++ b/packages/rs-platform-wallet-ffi/src/error.rs @@ -171,15 +171,17 @@ pub enum PlatformWalletFFIResultCode { /// boundary) and try again. Distinct from `ErrorShieldedSpendUnconfirmed`, /// where a spend WAS broadcast and must NOT be retried. ErrorShieldedNoRecordedAnchor = 19, - /// Maps `PlatformWalletError::TransactionBroadcastUnconfirmed`. A core + /// Maps `PlatformWalletError::TransactionBroadcastUnconfirmed`. A Core /// transaction broadcast (send-to-addresses, DashPay payment, or /// asset-lock funding) failed with an AMBIGUOUS outcome — the transaction /// may already be on the network (transport timeout after delivery, /// partial peer send, or an internal multi-node retry whose earlier /// attempt may have delivered). The wallet intentionally KEEPS the spent /// inputs' UTXO reservation, so an immediate retry fails at input - /// selection instead of double-spending; the reservation TTL or a sync - /// observing the transaction reconciles the outcome. The host must NOT + /// selection instead of double-spending; observing the spend reconciles + /// the outcome. Also covers address-funded identity top-ups whose result + /// could not be verified after possible acceptance, without Core UTXO + /// reservations. The host must reconcile the original operation and NOT /// auto-retry. Shielded sibling: [`Self::ErrorShieldedSpendUnconfirmed`]. ErrorTransactionBroadcastUnconfirmed = 20, /// Maps `PlatformWalletError::AddressNonceMismatch`. Platform rejected an diff --git a/packages/rs-platform-wallet-ffi/src/identity_top_up.rs b/packages/rs-platform-wallet-ffi/src/identity_top_up.rs index 7762a0f7382..49c51337af8 100644 --- a/packages/rs-platform-wallet-ffi/src/identity_top_up.rs +++ b/packages/rs-platform-wallet-ffi/src/identity_top_up.rs @@ -43,6 +43,9 @@ use crate::identity_registration::IdentityFundingInputFFI; use crate::runtime::block_on_worker; use crate::{unwrap_option_or_return, unwrap_result_or_return}; +#[cfg(test)] +mod submission_result_tests; + /// Top up an existing identity's credit balance from one or more /// Platform addresses, using an external `SignerHandle` for the /// per-address funding signatures. diff --git a/packages/rs-platform-wallet-ffi/src/identity_top_up/submission_result_tests.rs b/packages/rs-platform-wallet-ffi/src/identity_top_up/submission_result_tests.rs new file mode 100644 index 00000000000..b5015cbd37c --- /dev/null +++ b/packages/rs-platform-wallet-ffi/src/identity_top_up/submission_result_tests.rs @@ -0,0 +1,395 @@ +//! Exercise the real SDK -> identity wallet -> FFI error conversion without network I/O. + +use crate::error::{PlatformWalletFFIResult, PlatformWalletFFIResultCode}; +use dash_sdk::dapi_client::mock::MockResult; +use dash_sdk::dapi_client::transport::{TransportError, TransportRequest}; +use dash_sdk::dapi_client::{DapiClientError, DumpData, ExecutionError, ExecutionResponse}; +use dash_sdk::dapi_grpc::platform::v0::wait_for_state_transition_result_response::{ + wait_for_state_transition_result_response_v0::Result as WaitResult, Version as WaitVersion, + WaitForStateTransitionResultResponseV0, +}; +use dash_sdk::dapi_grpc::platform::v0::{ + BroadcastStateTransitionRequest, BroadcastStateTransitionResponse, Proof, ResponseMetadata, + StateTransitionBroadcastError, WaitForStateTransitionResultRequest, + WaitForStateTransitionResultResponse, +}; +use dash_sdk::dapi_grpc::tonic::metadata::MetadataValue; +use dash_sdk::dapi_grpc::tonic::Status; +use dash_sdk::platform::transition::broadcast_request::BroadcastRequestForStateTransition; +use dash_sdk::platform::transition::put_settings::PutSettings; +use dash_sdk::query_types::{AddressInfo, AddressInfos}; +use dash_sdk::{Error as SdkError, RequestSettings, Sdk, SdkBuilder}; +use dpp::address_funds::{AddressWitness, PlatformAddress}; +use dpp::consensus::state::address_funds::AddressInvalidNonceError; +use dpp::consensus::ConsensusError; +use dpp::identity::accessors::IdentityGettersV0; +use dpp::identity::signer::Signer; +use dpp::identity::IdentityV0; +use dpp::platform_value::BinaryData; +use dpp::prelude::{Identifier, Identity}; +use dpp::serialization::PlatformSerializableWithPlatformVersion; +use dpp::state_transition::identity_topup_from_addresses_transition::methods::IdentityTopUpFromAddressesTransitionMethodsV0; +use dpp::state_transition::identity_topup_from_addresses_transition::IdentityTopUpFromAddressesTransition; +use dpp::version::PlatformVersion; +use dpp::ProtocolError; +use key_wallet::wallet::initialization::WalletAccountCreationOptions; +use platform_wallet::error::promote_address_nonce_error; +use platform_wallet::events::{EventHandler, PlatformEventHandler}; +use platform_wallet::wallet::persister::NoPlatformPersistence; +use platform_wallet::{PlatformWallet, PlatformWalletManager}; +use std::collections::{BTreeMap, BTreeSet}; +use std::ffi::CStr; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +const ADDRESS: PlatformAddress = PlatformAddress::P2pkh([7; 20]); +const AMOUNT: u64 = 100_000_000; +const OLD_BALANCE: u64 = 123_000; + +#[derive(Debug, Default)] +struct AddressSigner(AtomicUsize); + +#[async_trait::async_trait] +impl Signer for AddressSigner { + async fn sign(&self, _: &PlatformAddress, _: &[u8]) -> Result { + self.0.fetch_add(1, Ordering::SeqCst); + Ok(vec![7; 65].into()) + } + + async fn sign_create_witness( + &self, + key: &PlatformAddress, + data: &[u8], + ) -> Result { + Ok(AddressWitness::P2pkh { + signature: self.sign(key, data).await?, + }) + } + + fn can_sign_with(&self, _: &PlatformAddress) -> bool { + true + } +} + +struct NoopEvents; +impl EventHandler for NoopEvents {} +impl PlatformEventHandler for NoopEvents {} + +fn nonce_error_bytes() -> Vec { + ConsensusError::from(AddressInvalidNonceError::new(ADDRESS, 1, 2)) + .serialize_to_bytes_with_platform_version(PlatformVersion::latest()) + .unwrap() +} + +fn wait_response(result: WaitResult) -> WaitForStateTransitionResultResponse { + WaitForStateTransitionResultResponse { + version: Some(WaitVersion::V0(WaitForStateTransitionResultResponseV0 { + result: Some(result), + metadata: Some(ResponseMetadata { + protocol_version: PlatformVersion::latest().protocol_version, + ..Default::default() + }), + })), + } +} + +fn save(directory: &tempfile::TempDir, request: &R, response: &MockResult) { + let dump = DumpData::new(request, response); + dump.save(&directory.path().join(dump.filename().unwrap())) + .unwrap(); +} + +fn response(inner: R) -> ExecutionResponse { + ExecutionResponse { + inner, + retries: 0, + address: "http://127.0.0.1".parse().unwrap(), + } +} + +fn transport_error(status: Status) -> ExecutionError { + ExecutionError { + inner: DapiClientError::Transport(TransportError::Grpc(status)), + address: None, + retries: 0, + } +} + +struct Fixture { + wallet: Arc, + sdk: Sdk, + identity_id: Identifier, + broadcast: BroadcastStateTransitionRequest, + wait: WaitForStateTransitionResultRequest, + signer: AddressSigner, +} + +impl Fixture { + async fn new( + broadcast: Result, + wait: Result, + funded: bool, + ) -> Self { + let mut sdk = SdkBuilder::new_mock() + .with_version(PlatformVersion::latest()) + .with_time_tolerance(None) + .with_settings(RequestSettings { + retries: Some(0), + ..Default::default() + }) + .build() + .unwrap(); + let identity = Identity::V0(IdentityV0 { + id: Identifier::from([9; 32]), + public_keys: Default::default(), + balance: OLD_BALANCE, + revision: 0, + }); + let transition = IdentityTopUpFromAddressesTransition::try_from_inputs_with_signer( + &identity, + BTreeMap::from([(ADDRESS, (1, AMOUNT))]), + &AddressSigner::default(), + 0, + sdk.version(), + None, + ) + .await + .unwrap(); + let broadcast_request = transition.broadcast_request_for_state_transition().unwrap(); + let wait_request = transition + .wait_for_state_transition_result_request() + .unwrap(); + let directory = tempfile::tempdir().unwrap(); + save( + &directory, + &broadcast_request, + &broadcast.map(response).map_err(transport_error), + ); + save( + &directory, + &wait_request, + &wait.map(response).map_err(transport_error), + ); + sdk.mock().load_expectations_sync(directory.path()).unwrap(); + sdk.mock() + .expect_fetch_many::( + BTreeSet::from([ADDRESS]), + Some(AddressInfos::from_iter([( + ADDRESS, + Some(AddressInfo { + address: ADDRESS, + nonce: 0, + balance: if funded { AMOUNT } else { 0 }, + }), + )])), + ) + .await + .unwrap(); + let manager = PlatformWalletManager::new( + Arc::new(sdk.clone()), + Arc::new(NoPlatformPersistence), + Arc::new(NoopEvents), + ); + let wallet = manager + .create_wallet_from_seed_bytes( + key_wallet::Network::Testnet, + &[42; 64], + WalletAccountCreationOptions::None, + Some(0), + ) + .await + .unwrap(); + wallet + .wallet_manager() + .write() + .await + .get_wallet_info_mut(&wallet.wallet_id()) + .unwrap() + .identity_manager + .add_identity(identity.clone(), 0, wallet.wallet_id(), wallet.persister()) + .unwrap(); + Self { + wallet, + sdk, + identity_id: identity.id(), + broadcast: broadcast_request, + wait: wait_request, + signer: AddressSigner::default(), + } + } + + async fn check( + mut self, + code: PlatformWalletFFIResultCode, + broadcasts: usize, + waits: usize, + ) -> String { + let error = self + .wallet + .top_up_from_addresses( + &self.identity_id, + BTreeMap::from([(ADDRESS, AMOUNT)]), + &self.signer, + Some(PutSettings { + request_settings: RequestSettings { + retries: Some(0), + ..Default::default() + }, + ..Default::default() + }), + ) + .await + .expect_err("a refused or unverified top-up must never succeed"); + // This is the same blanket conversion used by the production FFI entry point. + let result = PlatformWalletFFIResult::from(error); + let message = unsafe { CStr::from_ptr(result.message) } + .to_string_lossy() + .into_owned(); + assert_eq!(result.code, code, "{message}"); + if code == PlatformWalletFFIResultCode::ErrorTransactionBroadcastUnconfirmed { + assert_eq!( + result.consensus_code, 0, + "a later refusal is not a safe-retry verdict" + ); + } + assert_eq!( + self.sdk.mock().request_count(&self.broadcast).await, + broadcasts + ); + assert_eq!(self.sdk.mock().request_count(&self.wait).await, waits); + assert_eq!( + self.signer.0.load(Ordering::SeqCst), + broadcasts, + "never rebuild/re-sign after ambiguous submission" + ); + assert_eq!( + self.wallet + .wallet_manager() + .read() + .await + .get_wallet_info(&self.wallet.wallet_id()) + .unwrap() + .identity_manager + .identity(&self.identity_id) + .unwrap() + .identity + .balance(), + OLD_BALANCE + ); + message + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn should_keep_acknowledged_and_lost_ack_top_ups_unconfirmed_after_invalid_proof() { + for broadcast in [ + Ok(BroadcastStateTransitionResponse {}), + Err(Status::unavailable("lost acknowledgement")), + ] { + let fixture = Fixture::new( + broadcast, + Ok(wait_response(WaitResult::Proof(Proof::default()))), + true, + ) + .await; + let message = fixture + .check( + PlatformWalletFFIResultCode::ErrorTransactionBroadcastUnconfirmed, + 1, + 1, + ) + .await; + assert!(message.contains("unconfirmed")); + assert!(message.contains("Proof verification error"), "{message}"); + assert!(!message.contains("inputs stay unspendable")); + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn should_keep_top_up_unconfirmed_after_wait_transport_failure() { + for broadcast in [ + Ok(BroadcastStateTransitionResponse {}), + Err(Status::deadline_exceeded("lost acknowledgement")), + ] { + Fixture::new( + broadcast, + Err(Status::deadline_exceeded("result wait timed out")), + true, + ) + .await + .check( + PlatformWalletFFIResultCode::ErrorTransactionBroadcastUnconfirmed, + 1, + 1, + ) + .await; + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn should_not_promote_later_nonce_rejection_after_possible_acceptance() { + Fixture::new( + Err(Status::unavailable("lost acknowledgement")), + Ok(wait_response(WaitResult::Error( + StateTransitionBroadcastError { + code: 40603, + message: "address nonce mismatch".into(), + data: nonce_error_bytes(), + }, + ))), + true, + ) + .await + .check( + PlatformWalletFFIResultCode::ErrorTransactionBroadcastUnconfirmed, + 1, + 1, + ) + .await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn should_preserve_definite_broadcast_refusal_without_waiting() { + Fixture::new( + Err(Status::invalid_argument("invalid transition")), + Ok(WaitForStateTransitionResultResponse::default()), + true, + ) + .await + .check(PlatformWalletFFIResultCode::ErrorUnknown, 1, 0) + .await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn should_preserve_preflight_failure_without_dispatching() { + Fixture::new( + Ok(BroadcastStateTransitionResponse {}), + Ok(WaitForStateTransitionResultResponse::default()), + false, + ) + .await + .check(PlatformWalletFFIResultCode::ErrorUnknown, 0, 0) + .await; +} + +#[test] +fn should_preserve_definite_nonce_refusal_through_sdk_wallet_and_ffi_conversion() { + // MockableStatus drops metadata when serializing transport fixtures. Feed + // the real status directly through the same conversion/promotion functions. + let mut status = Status::invalid_argument("address nonce mismatch"); + status.metadata_mut().insert_bin( + "dash-serialized-consensus-error-bin", + MetadataValue::from_bytes(&nonce_error_bytes()), + ); + let error = SdkError::from(DapiClientError::Transport(TransportError::Grpc(status))); + let wallet_error = promote_address_nonce_error(&error) + .expect("a definite nonce refusal retains its typed wallet error"); + let result = PlatformWalletFFIResult::from(wallet_error); + assert_eq!( + result.code, + PlatformWalletFFIResultCode::ErrorAddressNonceMismatch + ); + let message = unsafe { CStr::from_ptr(result.message) }.to_string_lossy(); + assert!(message.contains("submitted nonce 1")); + assert!(message.contains("Platform expected 2")); +} diff --git a/packages/rs-platform-wallet/src/error.rs b/packages/rs-platform-wallet/src/error.rs index 5696d681d2b..df180a0f391 100644 --- a/packages/rs-platform-wallet/src/error.rs +++ b/packages/rs-platform-wallet/src/error.rs @@ -164,14 +164,21 @@ pub enum PlatformWalletError { #[error("Transaction broadcast failed: {0}")] TransactionBroadcast(String), - /// A core transaction broadcast failed with an **ambiguous** outcome — the + /// A transaction broadcast or address-funded identity top-up has an + /// **ambiguous** outcome. Reconcile it before submitting a replacement. + /// + /// For Core transactions, the /// transaction may already have reached the network (transport timeout /// after delivery, partial peer send, or an internal multi-node retry - /// whose earlier attempt may have succeeded). The spent inputs are + /// whose earlier attempt may have succeeded). The spent Core inputs are /// intentionally kept out of the selectable set, so an immediate retry /// fails at input selection instead of double-spending. /// - /// # What actually reconciles this, and what does not + /// An address-funded identity top-up may instead have been accepted but its + /// result proof could not be verified. Its balances must be re-read before + /// retrying; the Core input protections below do not apply to Platform addresses. + /// + /// # Core input reconciliation /// /// The inputs are held by TWO independent things, and only one of them /// expires. Key-wallet's `ReservationSet` entry is swept by its own TTL — @@ -204,9 +211,8 @@ pub enum PlatformWalletError { /// /// The shielded sibling is [`Self::ShieldedSpendUnconfirmed`]. #[error( - "Transaction broadcast outcome unknown — it may already be on the \ - network; its inputs stay unspendable until this wallet observes them \ - spent: {0}" + "Transaction outcome unknown — it may already have executed; \ + sync and reconcile the original operation before submitting another: {0}" )] TransactionBroadcastUnconfirmed(String), diff --git a/packages/rs-platform-wallet/src/wallet/identity/network/top_up_from_addresses.rs b/packages/rs-platform-wallet/src/wallet/identity/network/top_up_from_addresses.rs index 38096472926..03954f61147 100644 --- a/packages/rs-platform-wallet/src/wallet/identity/network/top_up_from_addresses.rs +++ b/packages/rs-platform-wallet/src/wallet/identity/network/top_up_from_addresses.rs @@ -12,6 +12,7 @@ use dpp::address_funds::PlatformAddress; use dpp::fee::Credits; use dash_sdk::query_types::AddressInfos; +use dash_sdk::Error as SdkError; use crate::error::PlatformWalletError; use crate::BlockTime; @@ -76,6 +77,9 @@ impl IdentityWallet { .top_up_from_addresses_with_metadata(&self.sdk, inputs, address_signer, settings) .await .map_err(|e| { + if matches!(&e, SdkError::StateTransitionResultUnconfirmed(_)) { + return PlatformWalletError::TransactionBroadcastUnconfirmed(e.to_string()); + } crate::error::promote_address_nonce_error(&e).unwrap_or_else(|| { PlatformWalletError::InvalidIdentityData(format!( "Failed to top up identity from addresses: {}", diff --git a/packages/rs-sdk-ffi/src/error.rs b/packages/rs-sdk-ffi/src/error.rs index 530c50f4c63..32654c83c9f 100644 --- a/packages/rs-sdk-ffi/src/error.rs +++ b/packages/rs-sdk-ffi/src/error.rs @@ -32,6 +32,9 @@ pub enum DashSDKErrorCode { NotImplemented = 9, /// Drive returned an internal error (e.g., storage-level constraint violation) DriveInternalError = 10, + /// The transition may have executed, but its result could not be verified. + /// Reconcile the original operation before submitting another transition. + StateTransitionResultUnconfirmed = 11, /// Internal error InternalError = 99, } @@ -216,12 +219,26 @@ impl FFIError { } } +/// Preserve the SDK's submission phase through retry envelopes; the inner +/// diagnostic is not a definitive refusal of the original operation. +fn is_state_transition_result_unconfirmed(error: &dash_sdk::Error) -> bool { + match error { + dash_sdk::Error::StateTransitionResultUnconfirmed(_) => true, + dash_sdk::Error::NoAvailableAddressesToRetry(source) => { + is_state_transition_result_unconfirmed(source) + } + _ => false, + } +} + impl From for DashSDKError { fn from(err: FFIError) -> Self { let (code, message) = match &err { FFIError::InvalidParameter(_) => (DashSDKErrorCode::InvalidParameter, err.to_string()), FFIError::SDKCallFailed { source, .. } => { - let code = if source.is_quorum_source_unavailable() { + let code = if is_state_transition_result_unconfirmed(source) { + DashSDKErrorCode::StateTransitionResultUnconfirmed + } else if source.is_quorum_source_unavailable() { DashSDKErrorCode::NetworkError } else { DashSDKErrorCode::InternalError @@ -236,9 +253,12 @@ impl From for DashSDKError { // substrings inside Drive messages (e.g., "data contract not found" // emitted as a DriveInternalError would otherwise be misclassified // as NotFound). - let (code, detailed_msg) = if let dash_sdk::Error::DriveInternalError(inner) = - sdk_err - { + let (code, detailed_msg) = if is_state_transition_result_unconfirmed(sdk_err) { + ( + DashSDKErrorCode::StateTransitionResultUnconfirmed, + error_str, + ) + } else if let dash_sdk::Error::DriveInternalError(inner) = sdk_err { // The DriveInternalError code already conveys the classification; // emit only the inner Drive message so downstream FFI consumers // don't double-render the "Drive internal error: " prefix. @@ -382,6 +402,125 @@ mod tests { use super::*; use std::ffi::CStr; + mod unconfirmed_result { + use super::*; + use crate::types::{dash_sdk_result_free, DashSDKResult}; + use dash_sdk::dpp::address_funds::PlatformAddress; + use dash_sdk::dpp::consensus::state::address_funds::AddressInvalidNonceError; + use dash_sdk::dpp::consensus::ConsensusError; + use std::time::Duration; + + fn nonce_refusal() -> dash_sdk::Error { + let consensus: ConsensusError = + AddressInvalidNonceError::new(PlatformAddress::P2pkh([7; 20]), 1, 2).into(); + consensus.into() + } + + /// Both public error ownership paths must retain the phase and copyable + /// diagnostics, even if retry exhaustion surrounds the SDK marker. + #[test] + fn should_preserve_unconfirmed_results_through_both_ffi_wrappers_and_free_paths() { + for contextual in [false, true] { + for retry_envelopes in 0..=2 { + for source in [ + dash_sdk::Error::TimeoutReached(Duration::from_secs(1), "wait".into()), + dash_sdk::Error::InvalidProvedResponse( + "unexpected identity in proof".into(), + ), + nonce_refusal(), + ] { + let diagnostic = source.to_string(); + let mut source = + dash_sdk::Error::StateTransitionResultUnconfirmed(Box::new(source)); + for _ in 0..retry_envelopes { + source = dash_sdk::Error::NoAvailableAddressesToRetry(Box::new(source)); + } + let direct_message = source.to_string(); + let error = if contextual { + FFIError::sdk_call_failed("Failed to top up identity and wait", source) + } else { + FFIError::SDKError(source) + }; + let expected_message = if contextual { + error.to_string() + } else { + direct_message + }; + let mut result = DashSDKResult::error(error.into()); + assert!(result.data.is_null()); + let message = unsafe { + let converted = &*result.error; + assert_eq!( + converted.code, + DashSDKErrorCode::StateTransitionResultUnconfirmed + ); + assert_eq!(converted.code as u32, 11); + assert_eq!(converted.consensus_code, 0); + assert_eq!( + converted.consensus_kind, + DashSDKConsensusErrorKind::ConsensusErrorKindNone + ); + let message = CStr::from_ptr(converted.message) + .to_str() + .unwrap() + .to_owned(); + if contextual { + dash_sdk_result_free(&mut result); + assert!(result.error.is_null()); + dash_sdk_result_free(&mut result); + } else { + // The direct Swift top-up wrapper uses this free function. + dash_sdk_error_free(result.error); + } + message + }; + assert_eq!(message, expected_message); + assert!(message.contains(&diagnostic)); + } + } + } + } + + #[test] + fn should_keep_definite_nonce_refusals_and_unmarked_failures_unchanged() { + for contextual in [false, true] { + for (source, expected_code, expected_consensus) in [ + (nonce_refusal(), DashSDKErrorCode::ProtocolError, 40603), + ( + dash_sdk::Error::TimeoutReached(Duration::from_secs(1), "wait".into()), + DashSDKErrorCode::Timeout, + 0, + ), + ( + dash_sdk::Error::InvalidProvedResponse("unexpected identity".into()), + DashSDKErrorCode::InternalError, + 0, + ), + ] { + let error = if contextual { + FFIError::sdk_call_failed("Failed before submission", source) + } else { + FFIError::SDKError(source) + }; + let mut result = DashSDKResult::error(error.into()); + unsafe { + assert_eq!( + (*result.error).code, + if contextual { + DashSDKErrorCode::InternalError + } else { + expected_code + } + ); + assert_eq!((*result.error).consensus_code, expected_consensus); + dash_sdk_result_free(&mut result); + } + assert!(result.error.is_null()); + } + } + } + } + fn classify(err: dash_sdk::Error) -> DashSDKErrorCode { let dash_sdk_error: DashSDKError = FFIError::SDKError(err).into(); let code = dash_sdk_error.code; diff --git a/packages/rs-sdk/src/error.rs b/packages/rs-sdk/src/error.rs index d070903ac46..0a216e93403 100644 --- a/packages/rs-sdk/src/error.rs +++ b/packages/rs-sdk/src/error.rs @@ -48,6 +48,12 @@ pub enum Error { /// accept snapshot outcomes explicitly. #[error("proof authenticates the transition's affected state only, not its execution: {0}")] ExecutionNotProved(String), + /// An address-funded identity top-up was accepted or may have been accepted, + /// but its result could not be verified. The source is diagnostic only: even + /// a later nonce rejection cannot make resubmission safe after possible acceptance. + /// Reconcile the original operation before creating another top-up. + #[error("identity top-up may have executed but its result is unconfirmed; do not re-submit before reconciling the original operation: {0}")] + StateTransitionResultUnconfirmed(#[source] Box), /// DAPI client error, for example, connection error #[error("Dapi client error: {0}")] DapiClientError(rs_dapi_client::DapiClientError), @@ -240,6 +246,9 @@ impl Error { Self::StateTransitionBroadcastError(broadcast_err) => broadcast_err.cause.as_ref(), Self::Protocol(ProtocolError::ConsensusError(ce)) => Some(ce.as_ref()), Self::NoAvailableAddressesToRetry(inner) => inner.consensus_error(), + // An unconfirmed submission's later diagnostic verdict does not + // establish that an earlier node never accepted the operation. + Self::StateTransitionResultUnconfirmed(_) => None, _ => None, } } @@ -924,3 +933,28 @@ mod tests { } } } + +#[cfg(test)] +mod unconfirmed_top_up_tests { + use super::*; + use dpp::address_funds::PlatformAddress; + use dpp::consensus::state::address_funds::AddressInvalidNonceError; + use std::error::Error as StdError; + + #[test] + fn should_preserve_unconfirmed_source_without_authorizing_a_retry() { + for source in [ + Error::TimeoutReached(Duration::from_secs(1), "wait timed out".into()), + Error::InvalidProvedResponse("unexpected identity in proof".into()), + Error::Protocol(ProtocolError::ConsensusError(Box::new( + AddressInvalidNonceError::new(PlatformAddress::P2pkh([7; 20]), 1, 2).into(), + ))), + ] { + let expected = source.to_string(); + let error = Error::StateTransitionResultUnconfirmed(Box::new(source)); + assert_eq!(StdError::source(&error).unwrap().to_string(), expected); + assert!(!error.can_retry()); + assert!(error.consensus_error().is_none()); + } + } +} diff --git a/packages/rs-sdk/src/platform/transition/broadcast.rs b/packages/rs-sdk/src/platform/transition/broadcast.rs index f01ef2ab82c..5504d156fd1 100644 --- a/packages/rs-sdk/src/platform/transition/broadcast.rs +++ b/packages/rs-sdk/src/platform/transition/broadcast.rs @@ -10,6 +10,7 @@ use dapi_grpc::platform::v0::{ wait_for_state_transition_result_response, BroadcastStateTransitionRequest, ResponseMetadata, WaitForStateTransitionResultResponse, }; +use dapi_grpc::tonic::{Code, ConnectError}; use dash_context_provider::ContextProviderError; use dpp::consensus::signature::SignatureError; use dpp::consensus::state::state_error::StateError; @@ -25,17 +26,28 @@ use dpp::state_transition::StateTransition; use dpp::system_data_contracts::SystemDataContract; use dpp::util::hash::hash_single; use dpp::ProtocolError; +use rs_dapi_client::transport::TransportError; use rs_dapi_client::AddressList; use rs_dapi_client::CanRetry; +use rs_dapi_client::DapiClientError; use rs_dapi_client::ExecutionResult; use rs_dapi_client::WrapToExecutionResult; use rs_dapi_client::{DapiRequest, ExecutionError, InnerInto, IntoInner, RequestSettings}; +use std::error::Error as StdError; use std::future::Future; -use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; +use std::sync::atomic::{AtomicBool, AtomicUsize}; +use std::sync::Arc; use tracing::{debug, trace, warn}; const DPNS_REGISTRATION_BROADCAST_RETRIES: usize = 2; +// rs-dapi emits this only after finding the transition in neither the local +// mempool nor chain and successfully re-running CheckTx. It refuses this attempt; +// it cannot rule out acceptance by another node during an earlier attempt. +const DAPI_PROCESSING_REJECTION: &str = "State Transition processing error. Please report faulty state transition and try to create a new state transition with different hash as a workaround."; + +#[cfg(test)] +pub(super) mod retry_tests; #[async_trait::async_trait] pub trait BroadcastStateTransition { @@ -123,52 +135,11 @@ pub trait BroadcastStateTransition { #[async_trait::async_trait] impl BroadcastStateTransition for StateTransition { async fn broadcast(&self, sdk: &Sdk, settings: Option) -> Result<(), Error> { - trace!( - state_transition = %self.name(), - transaction_id = %self - .transaction_id() - .map(hex::encode) - .unwrap_or("UNKNOWN".to_string()), - "broadcast: start" - ); - - let retry_settings = match settings { - Some(s) => sdk.dapi_client_settings.override_by(s.request_settings), - None => sdk.dapi_client_settings, - }; - - let execute = |request: BroadcastStateTransitionRequest, request_settings| async move { - trace!("broadcast: executing request"); - let result = request - .execute(sdk, request_settings) - .await - .map_err(|e| e.inner_into()); - - match &result { - Ok(_) => trace!("broadcast: request succeeded"), - Err(e) => warn!(error = ?e, "broadcast: request failed"), - } - result - }; - - // response is empty for a broadcast, result comes from the stream wait for state transition result - trace!("broadcast: starting retry mechanism"); - let result = broadcast_with_retries(self, sdk.address_list(), retry_settings, execute) + broadcast_with_observation(self, sdk, settings, BroadcastPolicy::Standard) .await - .into_inner() - .map(|_| ()); - - match &result { - Ok(_) => trace!("broadcast: completed successfully"), - Err(e) => { - warn!(error = ?e, "broadcast: failed after retries"); - if let Some(owner_id) = self.owner_id() { - sdk.refresh_identity_nonce(&owner_id).await; - } - } - } - result + .into_result() } + async fn wait_for_response + Send>( &self, sdk: &Sdk, @@ -233,10 +204,11 @@ impl BroadcastStateTransition for StateTransition { settings: Option, ) -> Result<(T, ResponseMetadata), Error> { trace!(state_transition = %self.name(), "broadcast_and_wait_for_affected_state: start"); - self.broadcast(sdk, settings).await?; - let result = self - .wait_for_affected_state_with_metadata::(sdk, settings) - .await; + let result = wait_after_broadcast( + broadcast_with_observation(self, sdk, settings, BroadcastPolicy::Standard).await, + self.wait_for_affected_state_with_metadata::(sdk, settings), + ) + .await; match &result { Ok(_) => trace!("broadcast_and_wait_for_affected_state: complete success"), Err(e) => warn!(error = ?e, "broadcast_and_wait_for_affected_state: failed"), @@ -261,11 +233,11 @@ impl BroadcastStateTransition for StateTransition { ) -> Result<(T, ResponseMetadata), Error> { trace!(state_transition = %self.name(), "broadcast_and_wait: start"); trace!("broadcast_and_wait: step 1 - broadcasting"); - self.broadcast(sdk, settings).await?; - trace!("broadcast_and_wait: step 2 - waiting for response"); - let result = self - .wait_for_response_with_metadata::(sdk, settings) - .await; + let result = wait_after_broadcast( + broadcast_with_observation(self, sdk, settings, BroadcastPolicy::Standard).await, + self.wait_for_response_with_metadata::(sdk, settings), + ) + .await; match &result { Ok(_) => trace!("broadcast_and_wait: complete success"), Err(e) => warn!(error = ?e, "broadcast_and_wait: failed"), @@ -274,6 +246,224 @@ impl BroadcastStateTransition for StateTransition { } } +/// The ordinary broadcast result plus evidence from every transport attempt. +/// Fields remain private so observation metadata can grow without changing callers. +/// Broadcast-only APIs explicitly project this into their existing final result. +#[derive(Debug)] +pub(super) struct BroadcastOutcome { + result: Result<(), Error>, + possibly_accepted: bool, +} + +impl BroadcastOutcome { + /// Preserve the final relay result for broadcast-only callers. + pub(super) fn into_result(self) -> Result<(), Error> { + self.result + } + + /// The final relay error, retained for legacy combined APIs with a specific fallback. + pub(super) fn error(&self) -> Option<&Error> { + self.result.as_ref().err() + } +} + +impl From> for BroadcastOutcome { + fn from(result: Result<(), Error>) -> Self { + let possibly_accepted = result + .as_ref() + .err() + .is_some_and(broadcast_may_have_been_accepted); + Self { + result, + possibly_accepted, + } + } +} + +/// Existing caller policies, not additional retry mechanisms. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum BroadcastPolicy { + /// SDK retries/settings overrides and nonce refresh on final broadcast failure. + Standard, + /// One executor invocation with transport retries and no nonce-cache refresh. + TransportOnly, +} + +/// Single send entry point for signed transitions, including broadcast-only APIs. +/// A future interceptor belongs around this call, not around individual retry +/// attempts. Recovery observes every attempt independently of any optional hook. +pub(super) async fn broadcast_with_observation( + transition: &StateTransition, + sdk: &Sdk, + settings: Option, + policy: BroadcastPolicy, +) -> BroadcastOutcome { + trace!( + state_transition = %transition.name(), + transaction_id = %transition.transaction_id().map(hex::encode) + .unwrap_or("UNKNOWN".to_string()), + "broadcast: start" + ); + let retry_settings = match policy { + BroadcastPolicy::Standard => match settings { + Some(s) => sdk.dapi_client_settings.override_by(s.request_settings), + None => sdk.dapi_client_settings, + }, + BroadcastPolicy::TransportOnly => settings.unwrap_or_default().request_settings, + }; + let outcome = broadcast_and_observe( + transition, + sdk.address_list(), + retry_settings, + policy, + |request, request_settings, possibly_accepted| async move { + sdk.execute_with_transport_error_observer(request, request_settings, &|error| { + if transport_broadcast_may_have_been_accepted(error) { + possibly_accepted.store(true, Ordering::Relaxed); + } + }) + .await + .map_err(|error| error.inner_into()) + }, + ) + .await; + match &outcome.result { + Ok(_) => trace!("broadcast: completed successfully"), + Err(error) => { + warn!(?error, "broadcast: failed after retries"); + if policy == BroadcastPolicy::Standard { + if let Some(owner_id) = transition.owner_id() { + sdk.refresh_identity_nonce(&owner_id).await; + } + } + } + } + outcome +} + +/// Executor seam for the common observation logic; production dispatch is above. +async fn broadcast_and_observe( + transition: &StateTransition, + addresses: &AddressList, + settings: RequestSettings, + policy: BroadcastPolicy, + mut execute: Execute, +) -> BroadcastOutcome +where + Fut: Future> + Send, + Execute: FnMut(BroadcastStateTransitionRequest, RequestSettings, Arc) -> Fut + Send, +{ + // Invocation-local: never shared with another transition or another call. + let possibly_accepted = Arc::new(AtomicBool::new(false)); + let mut observed_execute = |request, settings| { + let observation = Arc::clone(&possibly_accepted); + let response = execute(request, settings, Arc::clone(&observation)); + async move { + let result = response.await; + // Also covers SDK retries and mocks that return only a final error. + if result + .as_ref() + .err() + .is_some_and(|error| broadcast_may_have_been_accepted(&error.inner)) + { + observation.store(true, Ordering::Relaxed); + } + result + } + }; + let result = match policy { + BroadcastPolicy::Standard => { + broadcast_with_retries(transition, addresses, settings, observed_execute).await + } + BroadcastPolicy::TransportOnly => { + let request = match transition.broadcast_request_for_state_transition() { + Ok(request) => request, + Err(error) => return Err(error).into(), + }; + observed_execute(request, settings).await + } + } + .into_inner() + .map(|_| ()); + BroadcastOutcome { + result, + possibly_accepted: possibly_accepted.load(Ordering::Relaxed), + } +} + +/// A lost relay acknowledgement or cache hit says nothing about execution. +/// Keep waiting on the SAME signed transition through the caller's existing +/// proof-verification path. Neither an ambiguous broadcast nor a failed proof +/// is promoted to success, and this helper never builds or broadcasts a retry. +pub(super) async fn wait_after_broadcast( + broadcast: impl Into, + wait: impl Future>, +) -> Result { + let broadcast = broadcast.into(); + if let Err(error) = broadcast.result { + if !broadcast.possibly_accepted { + return Err(error); + } + debug!(error = ?error, "broadcast acknowledgement unavailable; awaiting verified result"); + } + wait.await +} + +fn broadcast_may_have_been_accepted(error: &Error) -> bool { + let status = match error { + Error::AlreadyExists(_) | Error::TimeoutReached(..) => return true, + Error::NoAvailableAddressesToRetry(inner) => { + return broadcast_may_have_been_accepted(inner); + } + Error::DapiClientError(DapiClientError::Transport(transport)) => transport, + Error::DapiClientError(DapiClientError::NoAvailableAddressesToRetry(transport)) => { + transport + } + _ => return false, + }; + transport_broadcast_may_have_been_accepted(status) +} + +fn transport_broadcast_may_have_been_accepted(error: &TransportError) -> bool { + let TransportError::Grpc(status) = error; + // tonic preserves its typed connector failure through Status's source chain. + // DNS, TCP and TLS establishment errors happen before this request is sent. + // A bare Unavailable (including one received from a node) does not carry + // that guarantee: it can also mean a lost acknowledgement after dispatch. + let mut source = status.source(); + while let Some(error) = source { + if error.is::() { + return false; + } + source = error.source(); + } + if status.code() == Code::Internal + && status + .message() + .strip_prefix("Internal error: ") + .unwrap_or(status.message()) + == DAPI_PROCESSING_REJECTION + { + return false; + } + // DAPI maps Tenderdash's confirmation timeout to this exact status. + // Other ResourceExhausted responses are mempool-full or rate-limit + // refusals and must not be treated as a possibly accepted submission. + (status.code() == Code::ResourceExhausted + && status.message() == "broadcasting state transition is timed out") + || matches!( + status.code(), + Code::AlreadyExists + | Code::DeadlineExceeded + | Code::Cancelled + | Code::Unknown + | Code::Internal + | Code::Aborted + | Code::DataLoss + | Code::Unavailable + ) +} + /// DPNS registration has two separate creates. Fail over each signed create, /// never restart the registration (which would generate a new preorder salt). /// This applies only to the immediate CheckTx rejection, not wait outcomes. @@ -596,21 +786,131 @@ mod address_nonce_tests; #[cfg(test)] mod tests { + use super::retry_tests::{retrying_broadcast, Attempt}; use super::*; + use dapi_grpc::tonic::metadata::MetadataValue; use dapi_grpc::tonic::Status; use dpp::consensus::signature::IdentityNotFoundError; use dpp::prelude::Identifier; + use dpp::serialization::PlatformSerializableWithPlatformVersion; use dpp::state_transition::batch_transition::document_base_transition::v0::DocumentBaseTransitionV0; use dpp::state_transition::batch_transition::document_create_transition::DocumentCreateTransitionV0; use dpp::state_transition::batch_transition::document_replace_transition::DocumentReplaceTransitionV0; use dpp::state_transition::batch_transition::BatchTransition; use dpp::state_transition::batch_transition::BatchTransitionV0; + use dpp::version::PlatformVersion; use rs_dapi_client::transport::TransportError; use rs_dapi_client::DapiClientError; use rs_dapi_client::ExecutionResponse; use std::future::ready; use std::time::Duration; + #[tokio::test] + async fn should_wait_for_verified_result_after_ambiguous_broadcast() { + let errors = [ + Error::AlreadyExists("tx already exists in cache".into()), + Error::TimeoutReached(Duration::from_secs(10), "lost acknowledgement".into()), + Error::DapiClientError(DapiClientError::Transport(TransportError::Grpc( + Status::deadline_exceeded("relay timed out"), + ))), + Error::NoAvailableAddressesToRetry(Box::new(Error::AlreadyExists("cached".into()))), + ]; + for error in errors { + let result = wait_after_broadcast(Err(error), async { Ok(42) }).await; + assert_eq!(result.expect("verified wait result"), 42); + } + } + + #[tokio::test] + async fn should_wait_after_dapi_confirmation_timeout_including_retry_wrappers() { + let timeout = || { + TransportError::Grpc(Status::resource_exhausted( + "broadcasting state transition is timed out", + )) + }; + for error in [ + Error::DapiClientError(DapiClientError::Transport(timeout())), + Error::DapiClientError(DapiClientError::NoAvailableAddressesToRetry(Box::new( + timeout(), + ))), + Error::NoAvailableAddressesToRetry(Box::new(Error::DapiClientError( + DapiClientError::Transport(timeout()), + ))), + Error::NoAvailableAddressesToRetry(Box::new(Error::DapiClientError( + DapiClientError::NoAvailableAddressesToRetry(Box::new(timeout())), + ))), + ] { + assert_eq!( + wait_after_broadcast(Err(error), async { Ok(42) }) + .await + .unwrap(), + 42 + ); + } + } + + #[tokio::test] + async fn should_not_wait_after_other_resource_exhausted_refusals() { + for message in [ + "mempool is full", + "tenderdash is not responding: too many requests", + "rate limited", + ] { + let error = Error::NoAvailableAddressesToRetry(Box::new(Error::DapiClientError( + DapiClientError::NoAvailableAddressesToRetry(Box::new(TransportError::Grpc( + Status::resource_exhausted(message), + ))), + ))); + let expected = error.to_string(); + let result: Result<(), Error> = wait_after_broadcast(Err(error), async { + panic!("a definite capacity refusal must not poll the result wait") + }) + .await; + assert_eq!(result.unwrap_err().to_string(), expected); + } + } + + #[tokio::test] + async fn should_never_substitute_cache_hit_for_a_valid_proof() { + for wait_error in [ + Error::InvalidProvedResponse("bad signature".into()), + Error::ExecutionNotProved("affected-state snapshot only".into()), + Error::TimeoutReached(Duration::from_secs(10), "no verified result".into()), + ] { + let expected = wait_error.to_string(); + let result: Result<(), Error> = wait_after_broadcast( + Err(Error::AlreadyExists("tx already exists in cache".into())), + async { Err(wait_error) }, + ) + .await; + assert_eq!( + result.expect_err("proof is still mandatory").to_string(), + expected + ); + } + } + + #[tokio::test] + async fn should_not_poll_result_wait_after_definite_broadcast_rejection() { + for error in [ + missing_owner(owner_id()), + Error::DapiClientError(DapiClientError::NoAvailableAddresses), + Error::DapiClientError(DapiClientError::Transport(TransportError::Grpc( + Status::invalid_argument("invalid transition"), + ))), + ] { + let expected = error.to_string(); + let result: Result<(), Error> = wait_after_broadcast(Err(error), async { + panic!("a definitive rejection must not poll the wait future") + }) + .await; + assert_eq!( + result.expect_err("original rejection").to_string(), + expected + ); + } + } + fn owner_id() -> Identifier { Identifier::from([1; 32]) } @@ -652,6 +952,48 @@ mod tests { .expect("valid addresses") } + #[tokio::test(start_paused = true)] + async fn should_preserve_transport_ambiguity_across_an_outer_dpns_retry() { + let transition = document_create(SystemDataContract::DPNS.id(), "domain"); + let consensus: ConsensusError = IdentityNotFoundError::new(owner_id()).into(); + let bytes = consensus + .serialize_to_bytes_with_platform_version(PlatformVersion::latest()) + .unwrap(); + let mut missing_owner = Status::invalid_argument("owner unavailable on this node"); + missing_owner.metadata_mut().insert_bin( + "dash-serialized-consensus-error-bin", + MetadataValue::from_bytes(&bytes), + ); + let (outcome, sent, executions) = retrying_broadcast( + &transition, + vec![ + Attempt::Error(Status::unavailable("acknowledgement lost")), + Attempt::Error(missing_owner), + Attempt::Error(Status::resource_exhausted("mempool is full")), + ], + 2, + ) + .await; + assert_eq!( + sent.len(), + 3, + "the original total dispatch budget is unchanged" + ); + assert_eq!(executions, 2, "the owner rejection triggers the SDK retry"); + assert!(outcome + .result + .as_ref() + .unwrap_err() + .to_string() + .contains("mempool is full")); + assert_eq!( + wait_after_broadcast(outcome, async { Ok(42) }) + .await + .unwrap(), + 42 + ); + } + #[test_case::test_case("preorder")] #[test_case::test_case("domain")] #[tokio::test] diff --git a/packages/rs-sdk/src/platform/transition/broadcast/retry_tests.rs b/packages/rs-sdk/src/platform/transition/broadcast/retry_tests.rs new file mode 100644 index 00000000000..5bc3c741965 --- /dev/null +++ b/packages/rs-sdk/src/platform/transition/broadcast/retry_tests.rs @@ -0,0 +1,382 @@ +//! Real DAPI and SDK retry loops over an in-memory transport, without sockets. + +use super::*; +use dapi_grpc::mock::Mockable; +use dapi_grpc::tonic::metadata::MetadataValue; +use dapi_grpc::tonic::Status; +use dpp::consensus::signature::IdentityNotFoundError; +use dpp::prelude::Identifier; +use dpp::serialization::PlatformSerializableWithPlatformVersion; +use dpp::state_transition::identity_create_transition::v0::IdentityCreateTransitionV0; +use dpp::version::PlatformVersion; +use rs_dapi_client::transport::{ + AppliedRequestSettings, BoxFuture, TransportClient, TransportRequest, +}; +use rs_dapi_client::{ConnectionPool, DapiClient, Uri}; +use std::collections::VecDeque; +use std::io; +use std::sync::Mutex; +use std::time::Duration; + +#[derive(Debug)] +pub(in crate::platform::transition) enum Attempt { + Error(Status), + Success, + Stall, +} + +struct MemoryClient; + +impl TransportClient for MemoryClient { + fn with_uri(_uri: Uri, _pool: &ConnectionPool) -> Result { + Ok(Self) + } + + fn with_uri_and_settings( + uri: Uri, + _settings: &AppliedRequestSettings, + pool: &ConnectionPool, + ) -> Result { + Self::with_uri(uri, pool) + } +} + +#[derive(Clone, Debug)] +struct MemoryBroadcast { + request: BroadcastStateTransitionRequest, + attempts: Arc>>, + sent: Arc>>>, +} + +impl Mockable for MemoryBroadcast {} + +impl TransportRequest for MemoryBroadcast { + type Client = MemoryClient; + type Response = BroadcastStateTransitionResponse; + + const SETTINGS_OVERRIDES: RequestSettings = RequestSettings::default(); + + fn method_name(&self) -> &'static str { + "broadcastStateTransition" + } + + fn execute_transport<'c>( + self, + _client: &'c mut Self::Client, + _settings: &AppliedRequestSettings, + ) -> BoxFuture<'c, Result> { + self.sent + .lock() + .unwrap() + .push(self.request.state_transition); + match self.attempts.lock().unwrap().pop_front().expect("attempt") { + Attempt::Error(status) => Box::pin(async { Err(TransportError::Grpc(status)) }), + Attempt::Success => Box::pin(async { Ok(BroadcastStateTransitionResponse {}) }), + Attempt::Stall => Box::pin(futures::future::pending()), + } + } +} + +/// Uses the same observed broadcast helper as production, including both retry +/// layers. Only the transport itself is replaced with deterministic responses. +pub(in crate::platform::transition) async fn retrying_broadcast( + transition: &StateTransition, + attempts: Vec, + retries: usize, +) -> (BroadcastOutcome, Vec>, usize) { + retrying_broadcast_with_policy(transition, attempts, retries, BroadcastPolicy::Standard).await +} + +async fn retrying_broadcast_with_policy( + transition: &StateTransition, + attempts: Vec, + retries: usize, + policy: BroadcastPolicy, +) -> (BroadcastOutcome, Vec>, usize) { + let addresses = + "http://127.0.0.1:3001,http://127.0.0.1:3002,http://127.0.0.1:3003,http://127.0.0.1:3004" + .parse::() + .unwrap(); + let client = DapiClient::new(addresses.clone(), RequestSettings::default()); + let attempts = Arc::new(Mutex::new(VecDeque::from(attempts))); + let sent = Arc::new(Mutex::new(Vec::new())); + let mut executions = 0; + let outcome = broadcast_and_observe( + transition, + &addresses, + RequestSettings { + retries: Some(retries), + timeout: Some(Duration::from_millis(1)), + ..Default::default() + }, + policy, + |request, settings, observation| { + executions += 1; + let request = MemoryBroadcast { + request, + attempts: Arc::clone(&attempts), + sent: Arc::clone(&sent), + }; + let client = &client; + async move { + client + .execute_with_transport_error_observer(request, settings, &|error| { + if transport_broadcast_may_have_been_accepted(error) { + observation.store(true, Ordering::Relaxed); + } + }) + .await + .map_err(|error| error.inner_into()) + } + }, + ) + .await; + assert!( + attempts.lock().unwrap().is_empty(), + "all expected sends happened" + ); + let sent = sent.lock().unwrap().clone(); + let expected = transition.broadcast_request_for_state_transition().unwrap(); + assert!(sent.iter().all(|bytes| bytes == &expected.state_transition)); + (outcome, sent, executions) +} + +fn signed_transition() -> StateTransition { + StateTransition::IdentityCreate( + IdentityCreateTransitionV0 { + signature: vec![7; 65].into(), + ..Default::default() + } + .into(), + ) +} + +fn refusal() -> Attempt { + Attempt::Error(Status::resource_exhausted("mempool is full")) +} + +#[test_case::test_case(BroadcastPolicy::Standard; "standard")] +#[test_case::test_case(BroadcastPolicy::TransportOnly; "transport_only")] +#[tokio::test(start_paused = true)] +async fn should_preserve_earlier_ambiguity_through_a_final_transport_refusal( + policy: BroadcastPolicy, +) { + for ambiguous in [ + Attempt::Error(Status::unavailable("acknowledgement lost")), + Attempt::Error(Status::resource_exhausted( + "broadcasting state transition is timed out", + )), + Attempt::Stall, + ] { + let (outcome, sent, executions) = retrying_broadcast_with_policy( + &signed_transition(), + vec![ambiguous, refusal()], + 1, + policy, + ) + .await; + // Keep the existing final broadcast-only error, but wait on the exact + // signed transition because an earlier node may have accepted it. + assert!(outcome + .result + .as_ref() + .unwrap_err() + .to_string() + .contains("mempool is full")); + assert_eq!(sent.len(), 2); + assert_eq!(executions, 1, "one inner retrying DAPI execution"); + assert_eq!( + wait_after_broadcast(outcome, async { Ok(42) }) + .await + .unwrap(), + 42 + ); + } +} + +#[test_case::test_case(BroadcastPolicy::Standard; "standard")] +#[test_case::test_case(BroadcastPolicy::TransportOnly; "transport_only")] +#[tokio::test(start_paused = true)] +async fn should_not_wait_when_every_transport_attempt_definitely_refused(policy: BroadcastPolicy) { + let (outcome, sent, executions) = + retrying_broadcast_with_policy(&signed_transition(), vec![refusal(), refusal()], 1, policy) + .await; + assert_eq!(sent.len(), 2); + assert_eq!(executions, 1); + let final_error = outcome.result.as_ref().unwrap_err().to_string(); + let result: Result<(), Error> = wait_after_broadcast(outcome, async { + panic!("definite refusals must not poll a wait") + }) + .await; + assert_eq!(result.unwrap_err().to_string(), final_error); +} + +#[test_case::test_case(BroadcastPolicy::Standard; "standard")] +#[test_case::test_case(BroadcastPolicy::TransportOnly; "transport_only")] +#[tokio::test(start_paused = true)] +async fn should_keep_eventual_success_and_verified_wait_failures_unchanged( + policy: BroadcastPolicy, +) { + let (success, sent, _) = retrying_broadcast_with_policy( + &signed_transition(), + vec![ + Attempt::Error(Status::unavailable("lost")), + Attempt::Success, + ], + 1, + policy, + ) + .await; + assert!(success.result.is_ok()); + assert_eq!(sent.len(), 2); + assert_eq!( + wait_after_broadcast(success, async { Ok(42) }) + .await + .unwrap(), + 42 + ); + + let (ambiguous, _, _) = retrying_broadcast_with_policy( + &signed_transition(), + vec![Attempt::Error(Status::unavailable("lost")), refusal()], + 1, + policy, + ) + .await; + let result: Result<(), Error> = wait_after_broadcast(ambiguous, async { + Err(Error::InvalidProvedResponse("proof rejected".into())) + }) + .await; + assert!(matches!(result, Err(Error::InvalidProvedResponse(_)))); +} + +#[tokio::test(start_paused = true)] +async fn should_keep_concurrent_broadcast_observations_separate() { + let transition = signed_transition(); + let (ambiguous, refused) = futures::join!( + retrying_broadcast( + &transition, + vec![Attempt::Error(Status::unavailable("lost")), refusal()], + 1 + ), + retrying_broadcast(&transition, vec![refusal(), refusal()], 1), + ); + assert!(ambiguous.0.possibly_accepted); + assert!(!refused.0.possibly_accepted); +} + +fn consensus_refusal() -> Status { + let consensus: ConsensusError = IdentityNotFoundError::new(Identifier::from([1; 32])).into(); + let bytes = consensus + .serialize_to_bytes_with_platform_version(PlatformVersion::latest()) + .unwrap(); + let mut status = Status::invalid_argument("owner unavailable on this node"); + status.metadata_mut().insert_bin( + "dash-serialized-consensus-error-bin", + MetadataValue::from_bytes(&bytes), + ); + status +} + +#[tokio::test(start_paused = true)] +async fn should_return_consensus_rejection_after_connection_establishment_fails() { + // tonic wraps failures from its connector before mapping them to Unavailable. + // Preserve that source through real inner retries and SDK error conversion. + let connect_error = Status::from_error(Box::new(ConnectError(Box::new(io::Error::new( + io::ErrorKind::ConnectionRefused, + "connection refused", + ))))); + assert_eq!(connect_error.code(), Code::Unavailable); + let (outcome, sent, executions) = retrying_broadcast( + &signed_transition(), + vec![ + Attempt::Error(connect_error), + Attempt::Error(consensus_refusal()), + ], + 1, + ) + .await; + assert_eq!(sent.len(), 2); + assert_eq!(executions, 1); + let result: Result<(), Error> = wait_after_broadcast(outcome, async { + panic!("a connection failure never submitted the first request") + }) + .await; + assert!(matches!( + result, + Err(Error::Protocol(ProtocolError::ConsensusError(error))) + if matches!(*error, ConsensusError::SignatureError(SignatureError::IdentityNotFoundError(_))) + )); +} + +#[tokio::test(start_paused = true)] +async fn should_keep_real_ambiguity_despite_a_later_state_dependent_consensus_rejection() { + let (outcome, sent, executions) = retrying_broadcast( + &signed_transition(), + vec![ + Attempt::Error(Status::unavailable("acknowledgement lost")), + Attempt::Error(consensus_refusal()), + ], + 1, + ) + .await; + assert_eq!(sent.len(), 2); + assert_eq!(executions, 1); + assert_eq!( + wait_after_broadcast(outcome, async { Ok(42) }) + .await + .unwrap(), + 42 + ); +} + +#[tokio::test(start_paused = true)] +async fn should_treat_exact_dapi_processing_rejection_as_local_to_its_attempt() { + for rejection in [ + DAPI_PROCESSING_REJECTION.to_owned(), + format!("Internal error: {DAPI_PROCESSING_REJECTION}"), + ] { + let (refused, sent, executions) = retrying_broadcast( + &signed_transition(), + vec![ + Attempt::Error(Status::internal(&rejection)), + Attempt::Error(Status::internal(&rejection)), + ], + 1, + ) + .await; + assert_eq!(sent.len(), 2); + assert_eq!(executions, 1); + let result: Result<(), Error> = wait_after_broadcast(refused, async { + panic!("both attempts returned the exact local processing refusal") + }) + .await; + assert!(result + .unwrap_err() + .to_string() + .contains(DAPI_PROCESSING_REJECTION)); + + for ambiguous in [ + Status::unavailable("acknowledgement lost"), + Status::internal("Failed checking state transition: connection reset"), + ] { + let (outcome, sent, executions) = retrying_broadcast( + &signed_transition(), + vec![ + Attempt::Error(ambiguous), + Attempt::Error(Status::internal(&rejection)), + ], + 1, + ) + .await; + assert_eq!(sent.len(), 2); + assert_eq!(executions, 1); + assert_eq!( + wait_after_broadcast(outcome, async { Ok(42) }) + .await + .unwrap(), + 42 + ); + } + } +} diff --git a/packages/rs-sdk/src/platform/transition/purchase_document.rs b/packages/rs-sdk/src/platform/transition/purchase_document.rs index 0dec88d6fe6..1e541638f98 100644 --- a/packages/rs-sdk/src/platform/transition/purchase_document.rs +++ b/packages/rs-sdk/src/platform/transition/purchase_document.rs @@ -1,4 +1,6 @@ -use super::broadcast::BroadcastStateTransition; +use super::broadcast::{ + broadcast_with_observation, wait_after_broadcast, BroadcastPolicy, BroadcastStateTransition, +}; use super::validation::ensure_valid_state_transition_structure; use super::waitable::Waitable; use crate::platform::transition::put_settings::PutSettings; @@ -61,42 +63,19 @@ impl> PurchaseDocument for Document { signer: &S, settings: Option, ) -> Result { - // A local failure after the nonce is reserved would leave the cached nonce ahead of - // Platform's, so what can be refused without it is refused first. - if let Some(creation_options) = - settings.and_then(|settings| settings.state_transition_creation_options) - { - creation_options.validate_base_carries_action_fee_agreement(sdk.version())?; - } - - let new_identity_contract_nonce = sdk - .get_identity_contract_nonce( - purchaser_id, - document_type.data_contract_id(), - true, - settings, - ) - .await?; - - let settings = settings.unwrap_or_default(); - - let transition = BatchTransition::new_document_purchase_transition_from_document( - self.clone(), - document_type.as_ref(), - purchaser_id, + let transition = prepare_purchase_state_transition( + self, price, - &identity_public_key, - new_identity_contract_nonce, - settings.user_fee_increase.unwrap_or_default(), + sdk, + document_type, + purchaser_id, + identity_public_key, token_payment_info, signer, - sdk.version(), - settings.state_transition_creation_options, + settings, ) .await?; - ensure_valid_state_transition_structure(&transition, sdk.version())?; - - transition.broadcast(sdk, Some(settings)).await?; + transition.broadcast(sdk, settings).await?; // response is empty for a broadcast, result comes from the stream wait for state transition result Ok(transition) } @@ -112,19 +91,165 @@ impl> PurchaseDocument for Document { signer: &S, settings: Option, ) -> Result { - let state_transition = self - .purchase_document( - price, - sdk, - document_type, - purchaser_id, - identity_public_key, - token_payment_info, - signer, - settings, + let state_transition = prepare_purchase_state_transition( + self, + price, + sdk, + document_type, + purchaser_id, + identity_public_key, + token_payment_info, + signer, + settings, + ) + .await?; + + let broadcast = + broadcast_with_observation(&state_transition, sdk, settings, BroadcastPolicy::Standard) + .await; + wait_after_broadcast(broadcast, async move { + Self::wait_for_response(sdk, state_transition, settings).await + }) + .await + } +} + +/// Construct and sign once; dispatch is chosen by the calling API. +#[allow(clippy::too_many_arguments)] +async fn prepare_purchase_state_transition>( + document: &Document, + price: Credits, + sdk: &Sdk, + document_type: DocumentType, + purchaser_id: Identifier, + identity_public_key: IdentityPublicKey, + token_payment_info: Option, + signer: &S, + settings: Option, +) -> Result { + // A local failure after the nonce is reserved would leave the cached nonce ahead of + // Platform's, so what can be refused without it is refused first. + if let Some(creation_options) = + settings.and_then(|settings| settings.state_transition_creation_options) + { + creation_options.validate_base_carries_action_fee_agreement(sdk.version())?; + } + + let new_identity_contract_nonce = sdk + .get_identity_contract_nonce( + purchaser_id, + document_type.data_contract_id(), + true, + settings, + ) + .await?; + + let settings = settings.unwrap_or_default(); + + let transition = BatchTransition::new_document_purchase_transition_from_document( + document.clone(), + document_type.as_ref(), + purchaser_id, + price, + &identity_public_key, + new_identity_contract_nonce, + settings.user_fee_increase.unwrap_or_default(), + token_payment_info, + signer, + sdk.version(), + settings.state_transition_creation_options, + ) + .await?; + ensure_valid_state_transition_structure(&transition, sdk.version())?; + + Ok(transition) +} + +#[cfg(all(test, feature = "mocks"))] +mod submission_tests { + use super::*; + use crate::sdk::put_test_support::PutFixture; + use dapi_grpc::tonic::Status; + + #[tokio::test] + async fn should_wait_for_original_signed_transition_after_ambiguous_submission() { + for (status, expected_waits) in [ + (Status::unavailable("acknowledgement lost"), 1), + (Status::already_exists("tx already exists in cache"), 1), + ( + Status::resource_exhausted("broadcasting state transition is timed out"), + 1, + ), + (Status::resource_exhausted("mempool is full"), 0), + ] { + let mut fixture = PutFixture::new().await; + let prepared = prepare_purchase_state_transition( + &fixture.document, + 10, + &fixture.sdk, + fixture.document_type.clone(), + Identifier::new([9; 32]), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), ) - .await?; + .await + .unwrap(); + let requests = fixture.expect_submission(&prepared, status).await; + let result = fixture + .document + .purchase_document_and_wait_for_response( + 10, + &fixture.sdk, + fixture.document_type.clone(), + Identifier::new([9; 32]), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()); + fixture + .assert_calls(&requests, expected_waits, result) + .await; + } + } - Self::wait_for_response(sdk, state_transition, settings).await + #[tokio::test] + async fn should_preserve_plain_broadcast_error_without_waiting() { + let mut fixture = PutFixture::new().await; + let prepared = prepare_purchase_state_transition( + &fixture.document, + 10, + &fixture.sdk, + fixture.document_type.clone(), + Identifier::new([9; 32]), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .unwrap(); + let requests = fixture + .expect_submission(&prepared, Status::unavailable("acknowledgement lost")) + .await; + let result = fixture + .document + .purchase_document( + 10, + &fixture.sdk, + fixture.document_type.clone(), + Identifier::new([9; 32]), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()); + fixture.assert_calls(&requests, 0, result).await; } } diff --git a/packages/rs-sdk/src/platform/transition/put_contract.rs b/packages/rs-sdk/src/platform/transition/put_contract.rs index 19e0ec51182..236d7b4a4e6 100644 --- a/packages/rs-sdk/src/platform/transition/put_contract.rs +++ b/packages/rs-sdk/src/platform/transition/put_contract.rs @@ -12,7 +12,9 @@ use dpp::state_transition::data_contract_create_transition::methods::DataContrac use dpp::state_transition::data_contract_create_transition::DataContractCreateTransition; use dpp::state_transition::StateTransition; -use super::broadcast::BroadcastStateTransition; +use super::broadcast::{ + broadcast_with_observation, wait_after_broadcast, BroadcastPolicy, BroadcastStateTransition, +}; use super::validation::ensure_valid_state_transition_structure; use super::waitable::Waitable; @@ -48,31 +50,9 @@ impl> PutContract for DataContract { signer: &S, settings: Option, ) -> Result { - let new_identity_nonce = sdk - .get_identity_nonce(self.owner_id(), true, settings) - .await?; - - let key_id = identity_public_key.id(); - - let partial_identity = PartialIdentity { - id: self.owner_id(), - loaded_public_keys: BTreeMap::from([(key_id, identity_public_key)]), - balance: None, - revision: None, - not_found_public_keys: Default::default(), - }; - let transition = DataContractCreateTransition::new_from_data_contract( - self.clone(), - new_identity_nonce, - &partial_identity, - key_id, - signer, - sdk.version(), - None, - ) - .await?; - ensure_valid_state_transition_structure(&transition, sdk.version())?; - + let transition = + prepare_contract_state_transition(self, sdk, identity_public_key, signer, settings) + .await?; transition.broadcast(sdk, settings).await?; // response is empty for a broadcast, result comes from the stream wait for state transition result @@ -86,10 +66,125 @@ impl> PutContract for DataContract { signer: &S, settings: Option, ) -> Result { - let state_transition = self - .put_to_platform(sdk, identity_public_key, signer, settings) - .await?; + let state_transition = + prepare_contract_state_transition(self, sdk, identity_public_key, signer, settings) + .await?; + + let broadcast = + broadcast_with_observation(&state_transition, sdk, settings, BroadcastPolicy::Standard) + .await; + wait_after_broadcast(broadcast, async move { + Self::wait_for_response(sdk, state_transition, settings).await + }) + .await + } +} + +/// Construct and sign once; dispatch is chosen by the calling API. +async fn prepare_contract_state_transition>( + contract: &DataContract, + sdk: &Sdk, + identity_public_key: IdentityPublicKey, + signer: &S, + settings: Option, +) -> Result { + let new_identity_nonce = sdk + .get_identity_nonce(contract.owner_id(), true, settings) + .await?; - Self::wait_for_response(sdk, state_transition, settings).await + let key_id = identity_public_key.id(); + + let partial_identity = PartialIdentity { + id: contract.owner_id(), + loaded_public_keys: BTreeMap::from([(key_id, identity_public_key)]), + balance: None, + revision: None, + not_found_public_keys: Default::default(), + }; + let transition = DataContractCreateTransition::new_from_data_contract( + contract.clone(), + new_identity_nonce, + &partial_identity, + key_id, + signer, + sdk.version(), + None, + ) + .await?; + ensure_valid_state_transition_structure(&transition, sdk.version())?; + + Ok(transition) +} + +#[cfg(all(test, feature = "mocks"))] +mod submission_tests { + use super::*; + use crate::sdk::put_test_support::PutFixture; + use dapi_grpc::tonic::Status; + + #[tokio::test] + async fn should_wait_for_original_signed_transition_after_ambiguous_submission() { + for (status, expected_waits) in [ + (Status::unavailable("acknowledgement lost"), 1), + (Status::already_exists("tx already exists in cache"), 1), + ( + Status::resource_exhausted("broadcasting state transition is timed out"), + 1, + ), + (Status::resource_exhausted("mempool is full"), 0), + ] { + let mut fixture = PutFixture::new().await; + let prepared = prepare_contract_state_transition( + &fixture.contract, + &fixture.sdk, + fixture.key.clone(), + &fixture.signer, + PutFixture::settings(), + ) + .await + .unwrap(); + let requests = fixture.expect_submission(&prepared, status).await; + let result = fixture + .contract + .put_to_platform_and_wait_for_response( + &fixture.sdk, + fixture.key.clone(), + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()); + fixture + .assert_calls(&requests, expected_waits, result) + .await; + } + } + + #[tokio::test] + async fn should_preserve_plain_broadcast_error_without_waiting() { + let mut fixture = PutFixture::new().await; + let prepared = prepare_contract_state_transition( + &fixture.contract, + &fixture.sdk, + fixture.key.clone(), + &fixture.signer, + PutFixture::settings(), + ) + .await + .unwrap(); + let requests = fixture + .expect_submission(&prepared, Status::unavailable("acknowledgement lost")) + .await; + let result = fixture + .contract + .put_to_platform( + &fixture.sdk, + fixture.key.clone(), + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()); + fixture.assert_calls(&requests, 0, result).await; } } diff --git a/packages/rs-sdk/src/platform/transition/put_document.rs b/packages/rs-sdk/src/platform/transition/put_document.rs index 6b35a67ee32..900eb24ef8d 100644 --- a/packages/rs-sdk/src/platform/transition/put_document.rs +++ b/packages/rs-sdk/src/platform/transition/put_document.rs @@ -1,4 +1,6 @@ -use super::broadcast::BroadcastStateTransition; +use super::broadcast::{ + broadcast_with_observation, wait_after_broadcast, BroadcastPolicy, BroadcastStateTransition, +}; use super::validation::ensure_valid_state_transition_structure; use super::waitable::{ wait_for_document_and_owner_balance, wait_for_document_and_owner_balance_with_metadata, @@ -146,104 +148,19 @@ impl> PutDocument for Document { signer: &S, settings: Option, ) -> Result { - // A local failure after the nonce is reserved would leave the cached nonce ahead of - // Platform's, so what can be refused without it is refused first. - let creation_options = - settings.and_then(|settings| settings.state_transition_creation_options); - if let Some(creation_options) = creation_options { - creation_options.validate_base_carries_action_fee_agreement(sdk.version())?; - } - - let document = prepare_document_for_transition(self, &document_type); - let is_replacement = - self.revision().is_some() && self.revision().unwrap() != INITIAL_REVISION; - // A contested create states the most it pays to join its contest, read before the - // nonce is reserved - let creation_options = if is_replacement { - creation_options - } else { - with_contest_fund_to_join(sdk, document_type.as_ref(), &document, creation_options) - .await? - }; - - let new_identity_contract_nonce = sdk - .get_identity_contract_nonce( - self.owner_id(), - document_type.data_contract_id(), - true, - settings, - ) - .await?; - - let settings = settings.unwrap_or_default(); - let transition = if is_replacement { - BatchTransition::new_document_replacement_transition_from_document( - document, - document_type.as_ref(), - &identity_public_key, - new_identity_contract_nonce, - settings.user_fee_increase.unwrap_or_default(), - token_payment_info, - signer, - sdk.version(), - creation_options, - ) - .await? - } else { - let (document, document_state_transition_entropy) = - match document_state_transition_entropy { - Some(entropy) => { - // While the id derives from the entropy alone, a caller-supplied - // entropy must derive the document's own id: consensus recomputes - // it and rejects the create with InvalidDocumentTransitionIdError - // on mismatch, so guard here before broadcasting to fail locally - // (no wasted nonce/fee). Once the id also commits to the identity - // contract nonce, the id the caller set is only a placeholder and - // the transition is built with the id derived below. - if !Document::document_id_depends_on_nonce(sdk.version())? { - ensure_entropy_matches_document_id( - &document_type.data_contract_id(), - &document.owner_id(), - document_type.name(), - &entropy, - document.id(), - )?; - } - (document, entropy) - } - None => { - let mut rng = StdRng::from_os_rng(); - let mut document = document; - let entropy = rng.random::<[u8; 32]>(); - document.set_id(Document::generate_document_id( - &document_type.data_contract_id(), - &document.owner_id(), - document_type.name(), - entropy.as_slice(), - new_identity_contract_nonce, - sdk.version(), - )?); - (document, entropy) - } - }; - BatchTransition::new_document_creation_transition_from_document( - document, - document_type.as_ref(), - document_state_transition_entropy, - &identity_public_key, - new_identity_contract_nonce, - settings.user_fee_increase.unwrap_or_default(), - token_payment_info, - signer, - sdk.version(), - creation_options, - ) - .await? - }; - ensure_valid_state_transition_structure(&transition, sdk.version())?; - + let transition = prepare_document_state_transition( + self, + sdk, + document_type, + document_state_transition_entropy, + identity_public_key, + token_payment_info, + signer, + settings, + ) + .await?; // response is empty for a broadcast, result comes from the stream wait for state transition result - transition.broadcast(sdk, Some(settings)).await?; + transition.broadcast(sdk, settings).await?; Ok(transition) } @@ -258,26 +175,103 @@ impl> PutDocument for Document { settings: Option, ) -> Result { let index_only = document_type.index_only(); - let state_transition = self - .put_to_platform( + let state_transition = prepare_document_state_transition( + self, + sdk, + document_type, + document_state_transition_entropy, + identity_public_key, + token_payment_info, + signer, + settings, + ) + .await?; + + let broadcast = + broadcast_with_observation(&state_transition, sdk, settings, BroadcastPolicy::Standard) + .await; + wait_after_broadcast(broadcast, async move { + // An indexOnly document keeps no row: its proof shows the entry the create leaves, not + // that this create executed, and no stronger proof exists for it. + if index_only { + return wait_for_document_and_owner_balance(sdk, state_transition, settings) + .await + .map(|(document, _owner_balance)| document); + } + Self::wait_for_response(sdk, state_transition, settings).await + }) + .await + } + + async fn put_to_platform_and_wait_for_response_with_owner_balance( + &self, + sdk: &Sdk, + document_type: DocumentType, + document_state_transition_entropy: Option<[u8; 32]>, + identity_public_key: IdentityPublicKey, + token_payment_info: Option, + signer: &S, + settings: Option, + ) -> Result<(Document, Option), Error> { + let state_transition = prepare_document_state_transition( + self, + sdk, + document_type, + document_state_transition_entropy, + identity_public_key, + token_payment_info, + signer, + settings, + ) + .await?; + + let broadcast = + broadcast_with_observation(&state_transition, sdk, settings, BroadcastPolicy::Standard) + .await; + wait_after_broadcast(broadcast, async move { + wait_for_document_and_owner_balance(sdk, state_transition, settings).await + }) + .await + } + + async fn put_to_platform_and_wait_for_response_with_owner_balance_and_metadata( + &self, + sdk: &Sdk, + document_type: DocumentType, + document_state_transition_entropy: Option<[u8; 32]>, + identity_public_key: IdentityPublicKey, + token_payment_info: Option, + signer: &S, + settings: Option, + ) -> Result<(Document, Option, ResponseMetadata), Error> { + // An indexOnly document keeps no row: its proof shows the entry the + // create leaves, not that this create executed. + let execution_proof_required = !document_type.index_only(); + let state_transition = prepare_document_state_transition( + self, + sdk, + document_type, + document_state_transition_entropy, + identity_public_key, + token_payment_info, + signer, + settings, + ) + .await?; + + let broadcast = + broadcast_with_observation(&state_transition, sdk, settings, BroadcastPolicy::Standard) + .await; + wait_after_broadcast(broadcast, async move { + wait_for_document_and_owner_balance_with_metadata( sdk, - document_type, - document_state_transition_entropy, - identity_public_key, - token_payment_info, - signer, + state_transition, settings, + execution_proof_required, ) - .await?; - - // An indexOnly document keeps no row: its proof shows the entry the create leaves, not - // that this create executed, and no stronger proof exists for it. - if index_only { - return wait_for_document_and_owner_balance(sdk, state_transition, settings) - .await - .map(|(document, _owner_balance)| document); - } - Self::wait_for_response(sdk, state_transition, settings).await + .await + }) + .await } } @@ -322,6 +316,115 @@ fn ensure_entropy_matches_document_id( Ok(()) } +/// Construct and sign once; dispatch is chosen by the calling API. +#[allow(clippy::too_many_arguments)] +async fn prepare_document_state_transition>( + document: &Document, + sdk: &Sdk, + document_type: DocumentType, + document_state_transition_entropy: Option<[u8; 32]>, + identity_public_key: IdentityPublicKey, + token_payment_info: Option, + signer: &S, + settings: Option, +) -> Result { + // A local failure after the nonce is reserved would leave the cached nonce ahead of + // Platform's, so what can be refused without it is refused first. + let creation_options = settings.and_then(|settings| settings.state_transition_creation_options); + if let Some(creation_options) = creation_options { + creation_options.validate_base_carries_action_fee_agreement(sdk.version())?; + } + + let document = prepare_document_for_transition(document, &document_type); + let is_replacement = + document.revision().is_some() && document.revision().unwrap() != INITIAL_REVISION; + // A contested create states the most it pays to join its contest, read before the + // nonce is reserved + let creation_options = if is_replacement { + creation_options + } else { + with_contest_fund_to_join(sdk, document_type.as_ref(), &document, creation_options).await? + }; + + let new_identity_contract_nonce = sdk + .get_identity_contract_nonce( + document.owner_id(), + document_type.data_contract_id(), + true, + settings, + ) + .await?; + + let settings = settings.unwrap_or_default(); + let transition = if is_replacement { + BatchTransition::new_document_replacement_transition_from_document( + document, + document_type.as_ref(), + &identity_public_key, + new_identity_contract_nonce, + settings.user_fee_increase.unwrap_or_default(), + token_payment_info, + signer, + sdk.version(), + creation_options, + ) + .await? + } else { + let (document, document_state_transition_entropy) = match document_state_transition_entropy + { + Some(entropy) => { + // While the id derives from the entropy alone, a caller-supplied + // entropy must derive the document's own id: consensus recomputes + // it and rejects the create with InvalidDocumentTransitionIdError + // on mismatch, so guard here before broadcasting to fail locally + // (no wasted nonce/fee). Once the id also commits to the identity + // contract nonce, the id the caller set is only a placeholder and + // the transition is built with the id derived below. + if !Document::document_id_depends_on_nonce(sdk.version())? { + ensure_entropy_matches_document_id( + &document_type.data_contract_id(), + &document.owner_id(), + document_type.name(), + &entropy, + document.id(), + )?; + } + (document, entropy) + } + None => { + let mut rng = StdRng::from_os_rng(); + let mut document = document; + let entropy = rng.random::<[u8; 32]>(); + document.set_id(Document::generate_document_id( + &document_type.data_contract_id(), + &document.owner_id(), + document_type.name(), + entropy.as_slice(), + new_identity_contract_nonce, + sdk.version(), + )?); + (document, entropy) + } + }; + BatchTransition::new_document_creation_transition_from_document( + document, + document_type.as_ref(), + document_state_transition_entropy, + &identity_public_key, + new_identity_contract_nonce, + settings.user_fee_increase.unwrap_or_default(), + token_payment_info, + signer, + sdk.version(), + creation_options, + ) + .await? + }; + ensure_valid_state_transition_structure(&transition, sdk.version())?; + + Ok(transition) +} + #[cfg(test)] mod tests { use super::*; @@ -439,3 +542,98 @@ mod tests { ); } } + +#[cfg(all(test, feature = "mocks"))] +mod submission_tests { + use super::*; + use crate::sdk::put_test_support::PutFixture; + use dapi_grpc::tonic::Status; + + #[tokio::test] + async fn should_observe_each_combined_document_api_without_changing_plain_broadcast() { + for api in 0..4 { + for (status, ambiguous) in [ + (Status::unavailable("acknowledgement lost"), true), + (Status::already_exists("tx already exists in cache"), true), + ( + Status::resource_exhausted("broadcasting state transition is timed out"), + true, + ), + (Status::resource_exhausted("mempool is full"), false), + ] { + let mut fixture = PutFixture::new().await; + let prepared = prepare_document_state_transition( + &fixture.document, + &fixture.sdk, + fixture.document_type.clone(), + None, + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .unwrap(); + let requests = fixture.expect_submission(&prepared, status).await; + let result = match api { + 0 => fixture + .document + .put_to_platform( + &fixture.sdk, + fixture.document_type.clone(), + None, + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()), + 1 => fixture + .document + .put_to_platform_and_wait_for_response( + &fixture.sdk, + fixture.document_type.clone(), + None, + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()), + 2 => fixture + .document + .put_to_platform_and_wait_for_response_with_owner_balance( + &fixture.sdk, + fixture.document_type.clone(), + None, + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()), + 3 => fixture + .document + .put_to_platform_and_wait_for_response_with_owner_balance_and_metadata( + &fixture.sdk, + fixture.document_type.clone(), + None, + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()), + _ => unreachable!(), + }; + fixture + .assert_calls(&requests, usize::from(api != 0 && ambiguous), result) + .await; + } + } + } +} diff --git a/packages/rs-sdk/src/platform/transition/put_identity.rs b/packages/rs-sdk/src/platform/transition/put_identity.rs index 0a27297e9d6..e67c3467212 100644 --- a/packages/rs-sdk/src/platform/transition/put_identity.rs +++ b/packages/rs-sdk/src/platform/transition/put_identity.rs @@ -1,4 +1,7 @@ use crate::platform::transition::address_inputs::{fetch_inputs_with_nonce, nonce_inc}; +use crate::platform::transition::broadcast::{ + broadcast_with_observation, wait_after_broadcast, BroadcastOutcome, BroadcastPolicy, +}; use crate::platform::transition::broadcast_identity::BroadcastRequestForNewIdentity; use crate::platform::transition::{ address_inputs::collect_address_infos_from_proof, broadcast::BroadcastStateTransition, @@ -22,7 +25,6 @@ use dpp::state_transition::proof_result::StateTransitionProofResult; use dpp::state_transition::StateTransition; use drive_proof_verifier::types::AddressInfos; use std::collections::{BTreeMap, BTreeSet}; -use tracing::debug; /// Trait for creating identities on the platform. #[async_trait::async_trait] @@ -46,9 +48,8 @@ pub trait PutIdentity>: Waitable { /// Creates an identity using an asset lock and waits for confirmation. /// - /// A broadcast answered with `AlreadyExists` (this exact transition is - /// already in the mempool or on chain) is not an error: the result wait - /// that follows resolves it. + /// An ambiguous broadcast acknowledgement (including `AlreadyExists`) + /// is resolved by waiting for the same signed transition's verified result. /// /// In-process private-key counterpart to /// [`Self::put_to_platform_and_wait_for_response_with_signer`]. @@ -87,8 +88,9 @@ pub trait PutIdentity>: Waitable { AS: dpp::key_wallet::signer::Signer + Send + Sync; /// Creates an identity using an asset-lock signer and waits for - /// confirmation. `AlreadyExists` from the broadcast is resolved by the - /// wait, as in the private-key variant. + /// confirmation. An ambiguous broadcast acknowledgement (including + /// `AlreadyExists`) is resolved by waiting for the same signed transition's + /// verified result, as in the private-key variant. /// /// Signer-driven counterpart to /// [`Self::put_to_platform_and_wait_for_response_with_private_key`]. @@ -187,7 +189,9 @@ impl> PutIdentity for Identity { settings, ) .await?; - let broadcast = state_transition.broadcast(sdk, settings).await; + let broadcast = + broadcast_with_observation(&state_transition, sdk, settings, BroadcastPolicy::Standard) + .await; broadcast_then_wait(broadcast, state_transition, |st| { Self::wait_for_response(sdk, st, settings) }) @@ -244,7 +248,9 @@ impl> PutIdentity for Identity { settings, ) .await?; - let broadcast = state_transition.broadcast(sdk, settings).await; + let broadcast = + broadcast_with_observation(&state_transition, sdk, settings, BroadcastPolicy::Standard) + .await; broadcast_then_wait(broadcast, state_transition, |st| { Self::wait_for_response(sdk, st, settings) }) @@ -298,29 +304,13 @@ impl> PutIdentity for Identity { } } -/// A broadcast whose result is waited for next: `AlreadyExists` counts as -/// delivered. DAPI returns it only when this exact transition is already in -/// the mempool or on chain (a broadcast that timed out after the node took it, -/// retried on another node), so the wait resolves it; failing here would -/// report an identity create that lands as a failure. -fn already_known_is_delivered(broadcast: Result<(), Error>) -> Result<(), Error> { - match broadcast { - Err(Error::AlreadyExists(message)) => { - debug!(%message, "identity create is already known to Platform; waiting for its result"); - Ok(()) - } - other => other, - } -} - /// After an identity create's broadcast, wait for its result — the step both /// `put_to_platform_and_wait_for_response_*` variants share, with the wait /// injected so it can be tested without a Platform: the same signed -/// transition is waited for after a delivered broadcast, including one DAPI -/// reports as already known, and nothing is waited for after any other -/// broadcast failure. +/// transition is waited for after a successful or ambiguous acknowledgement. +/// Definite broadcast refusals are returned without invoking the wait. async fn broadcast_then_wait( - broadcast: Result<(), Error>, + broadcast: impl Into, state_transition: StateTransition, wait: W, ) -> Result @@ -328,8 +318,7 @@ where W: FnOnce(StateTransition) -> WFut, WFut: std::future::Future>, { - already_known_is_delivered(broadcast)?; - wait(state_transition).await + wait_after_broadcast(broadcast, async move { wait(state_transition).await }).await } /// Build and structurally validate the identity create, without broadcasting. @@ -472,14 +461,8 @@ async fn put_identity_with_address_funding< #[cfg(test)] mod tests { use super::*; - - #[test] - fn should_treat_an_already_known_broadcast_as_delivered() { - let broadcast = Err(Error::AlreadyExists( - "state transition already in mempool".to_string(), - )); - assert!(already_known_is_delivered(broadcast).is_ok()); - } + use crate::platform::transition::broadcast::retry_tests::{retrying_broadcast, Attempt}; + use dapi_grpc::tonic::Status; fn signed_create() -> StateTransition { use dpp::state_transition::identity_create_transition::v0::IdentityCreateTransitionV0; @@ -502,6 +485,30 @@ mod tests { }) } + #[tokio::test(start_paused = true)] + async fn should_verify_the_same_identity_after_mixed_transport_retry_outcomes() { + let submitted = signed_create(); + let (broadcast, sent, executions) = retrying_broadcast( + &submitted, + vec![ + Attempt::Error(Status::unavailable("acknowledgement lost")), + Attempt::Error(Status::resource_exhausted("mempool is full")), + ], + 1, + ) + .await; + assert_eq!(sent.len(), 2); + assert_eq!(executions, 1); + let expected = submitted.clone(); + let identity = broadcast_then_wait(broadcast, submitted, |transition| async move { + assert_eq!(transition, expected); + Ok(created()) + }) + .await + .expect("identity from the existing verified-result wait"); + assert_eq!(identity, created()); + } + #[tokio::test] async fn should_wait_for_the_same_transition_after_an_already_known_broadcast() { let submitted = signed_create(); @@ -545,7 +552,7 @@ mod tests { } #[tokio::test] - async fn should_not_wait_after_any_other_broadcast_failure() { + async fn should_not_wait_after_definite_broadcast_failure() { let waited = std::sync::atomic::AtomicBool::new(false); let result = broadcast_then_wait( Err(Error::Generic("broadcast rejected".to_string())), @@ -560,10 +567,69 @@ mod tests { assert!(!waited.load(std::sync::atomic::Ordering::SeqCst)); } - #[test] - fn should_keep_every_other_broadcast_outcome() { - assert!(already_known_is_delivered(Ok(())).is_ok()); - let rejected = already_known_is_delivered(Err(Error::Generic("rejected".to_string()))); - assert!(matches!(rejected, Err(Error::Generic(_)))); + #[tokio::test] + async fn should_return_the_verified_identity_after_ambiguous_acknowledgements() { + use dapi_grpc::tonic::Status; + use rs_dapi_client::transport::TransportError; + use rs_dapi_client::DapiClientError; + use std::time::Duration; + + for error in [ + Error::TimeoutReached(Duration::from_secs(10), "lost acknowledgement".into()), + Error::DapiClientError(DapiClientError::Transport(TransportError::Grpc( + Status::deadline_exceeded("relay timed out"), + ))), + Error::DapiClientError(DapiClientError::Transport(TransportError::Grpc( + Status::unavailable("relay disconnected"), + ))), + Error::NoAvailableAddressesToRetry(Box::new(Error::DapiClientError( + DapiClientError::NoAvailableAddressesToRetry(Box::new(TransportError::Grpc( + Status::resource_exhausted("broadcasting state transition is timed out"), + ))), + ))), + ] { + let submitted = signed_create(); + let expected = submitted.clone(); + let identity = broadcast_then_wait(Err(error), submitted, |transition| async move { + assert_eq!(transition, expected, "wait for the original signed bytes"); + // The production closure is Identity::wait_for_response, whose + // proof gate owns this identity result; this seam tests that the + // result is awaited and propagated without another broadcast. + Ok(created()) + }) + .await + .expect("verified identity result"); + assert_eq!(identity, created()); + } + } + + #[tokio::test] + async fn should_preserve_failed_verification_after_ambiguous_acknowledgement() { + let result = broadcast_then_wait( + Err(Error::TimeoutReached( + std::time::Duration::from_secs(10), + "relay timed out".into(), + )), + signed_create(), + |_| async { + Err(Error::InvalidProvedResponse( + "invalid identity proof".into(), + )) + }, + ) + .await; + assert!( + matches!(result, Err(Error::InvalidProvedResponse(message)) if message == "invalid identity proof") + ); + } + + #[tokio::test] + async fn should_wait_after_successful_identity_broadcast() { + assert_eq!( + broadcast_then_wait(Ok(()), signed_create(), |_| async { Ok(created()) }) + .await + .unwrap(), + created() + ); } } diff --git a/packages/rs-sdk/src/platform/transition/top_up_identity_from_addresses.rs b/packages/rs-sdk/src/platform/transition/top_up_identity_from_addresses.rs index 01526d11fae..7d850cb28a4 100644 --- a/packages/rs-sdk/src/platform/transition/top_up_identity_from_addresses.rs +++ b/packages/rs-sdk/src/platform/transition/top_up_identity_from_addresses.rs @@ -6,14 +6,16 @@ use super::put_settings::PutSettings; use super::validation::ensure_valid_state_transition_structure; use super::waitable::Waitable; use crate::platform::transition::address_inputs::{collect_address_infos_from_proof, nonce_inc}; -use crate::platform::transition::broadcast::BroadcastStateTransition; +use crate::platform::transition::broadcast::{ + broadcast_with_observation, wait_after_broadcast, BroadcastPolicy, BroadcastStateTransition, +}; use crate::{Error, Sdk}; use dpp::address_funds::PlatformAddress; use dpp::fee::Credits; use dpp::identity::accessors::IdentityGettersV0; use dpp::identity::signer::Signer; use dpp::identity::Identity; -use dpp::prelude::AddressNonce; +use dpp::prelude::{AddressNonce, Identifier}; use dpp::state_transition::identity_topup_from_addresses_transition::methods::IdentityTopUpFromAddressesTransitionMethodsV0; use dpp::state_transition::identity_topup_from_addresses_transition::IdentityTopUpFromAddressesTransition; use dpp::state_transition::proof_result::StateTransitionProofResult; @@ -137,41 +139,61 @@ impl> TopUpIdentityFromAddressesWithMetadata for I .await?; ensure_valid_state_transition_structure(&state_transition, sdk.version())?; - // `metadata.height` is the proof's committed block — the height - // pin for these absolutes (`AddressFunds::as_of_height`). - let (st_result, metadata) = state_transition - .broadcast_and_wait_for_affected_state_with_metadata::( - sdk, settings, - ) - .await?; - match st_result { - StateTransitionProofResult::VerifiedIdentityWithAddressInfos( - identity, - address_infos_map, - ) => { - if identity.id != self.id() { - return Err(Error::InvalidProvedResponse(format!( - "proof returned identity {} but {} was topped up", - identity.id, - self.id() - ))); + // Only the wait future runs after accepted/possibly accepted submission. + // Keep preparation and definite broadcast refusals outside the wrapper. + wait_after_broadcast( + broadcast_with_observation(&state_transition, sdk, settings, BroadcastPolicy::Standard) + .await, + async { + let result = async { + // `metadata.height` pins these proved balance absolutes. + let (st_result, metadata) = state_transition + .wait_for_affected_state_with_metadata::( + sdk, settings, + ) + .await?; + top_up_result(self.id(), st_result, &expected_addresses) + .map(|(infos, balance)| (infos, balance, metadata)) } + .await; + result.map_err(|error| Error::StateTransitionResultUnconfirmed(Box::new(error))) + }, + ) + .await + } +} - let address_infos = - collect_address_infos_from_proof(address_infos_map, &expected_addresses)?; +fn top_up_result( + identity_id: Identifier, + st_result: StateTransitionProofResult, + expected_addresses: &BTreeSet, +) -> Result<(AddressInfos, Credits), Error> { + match st_result { + StateTransitionProofResult::VerifiedIdentityWithAddressInfos( + identity, + address_infos_map, + ) => { + if identity.id != identity_id { + return Err(Error::InvalidProvedResponse(format!( + "proof returned identity {} but {} was topped up", + identity.id, identity_id + ))); + } - let balance = identity.balance.ok_or_else(|| { - Error::InvalidProvedResponse( - "identity proof did not include updated balance".to_string(), - ) - })?; + let address_infos = + collect_address_infos_from_proof(address_infos_map, expected_addresses)?; - Ok((address_infos, balance, metadata)) - } - other => Err(Error::InvalidProvedResponse(format!( - "identity proof was expected for {:?}, but received {:?}", - state_transition, other - ))), + let balance = identity.balance.ok_or_else(|| { + Error::InvalidProvedResponse( + "identity proof did not include updated balance".to_string(), + ) + })?; + + Ok((address_infos, balance)) } + other => Err(Error::InvalidProvedResponse(format!( + "identity top-up proof was expected for {}, but received {:?}", + identity_id, other + ))), } } diff --git a/packages/rs-sdk/src/platform/transition/transfer_document.rs b/packages/rs-sdk/src/platform/transition/transfer_document.rs index bf8dedc6f04..f1d71c28ce8 100644 --- a/packages/rs-sdk/src/platform/transition/transfer_document.rs +++ b/packages/rs-sdk/src/platform/transition/transfer_document.rs @@ -1,6 +1,6 @@ +use super::broadcast::{broadcast_with_observation, wait_after_broadcast, BroadcastPolicy}; use super::validation::ensure_valid_state_transition_structure; use super::waitable::Waitable; -use crate::platform::transition::broadcast_request::BroadcastRequestForStateTransition; use crate::platform::transition::put_settings::PutSettings; use crate::platform::Identifier; use crate::{Error, Sdk}; @@ -13,7 +13,6 @@ use dpp::state_transition::batch_transition::methods::v0::DocumentsBatchTransiti use dpp::state_transition::batch_transition::BatchTransition; use dpp::state_transition::StateTransition; use dpp::tokens::token_payment_info::TokenPaymentInfo; -use rs_dapi_client::{DapiRequest, IntoInner}; #[async_trait::async_trait] /// A trait for transferring a document on Platform @@ -58,49 +57,20 @@ impl> TransferDocument for Document { signer: &S, settings: Option, ) -> Result { - // A local failure after the nonce is reserved would leave the cached nonce ahead of - // Platform's, so what can be refused without it is refused first. - if let Some(creation_options) = - settings.and_then(|settings| settings.state_transition_creation_options) - { - creation_options.validate_base_carries_action_fee_agreement(sdk.version())?; - } - - let new_identity_contract_nonce = sdk - .get_identity_contract_nonce( - self.owner_id(), - document_type.data_contract_id(), - true, - settings, - ) - .await?; - - let settings = settings.unwrap_or_default(); - - let transition = BatchTransition::new_document_transfer_transition_from_document( - self.clone(), - document_type.as_ref(), + let transition = prepare_transfer_state_transition( + self, recipient_id, - &identity_public_key, - new_identity_contract_nonce, - settings.user_fee_increase.unwrap_or_default(), + sdk, + document_type, + identity_public_key, token_payment_info, signer, - sdk.version(), - settings.state_transition_creation_options, + settings, ) .await?; - ensure_valid_state_transition_structure(&transition, sdk.version())?; - - let request = transition.broadcast_request_for_state_transition()?; - - request - .clone() - .execute(sdk, settings.request_settings) - .await // TODO: We need better way to handle execution errors - .into_inner()?; - - // response is empty for a broadcast, result comes from the stream wait for state transition result + broadcast_with_observation(&transition, sdk, settings, BroadcastPolicy::TransportOnly) + .await + .into_result()?; Ok(transition) } @@ -115,18 +85,184 @@ impl> TransferDocument for Document { signer: &S, settings: Option, ) -> Result { - let state_transition = self - .transfer_document_to_identity( - recipient_id, - sdk, - document_type, - identity_public_key, - token_payment_info, - signer, - settings, + let state_transition = prepare_transfer_state_transition( + self, + recipient_id, + sdk, + document_type, + identity_public_key, + token_payment_info, + signer, + settings, + ) + .await?; + + let broadcast = broadcast_with_observation( + &state_transition, + sdk, + settings, + BroadcastPolicy::TransportOnly, + ) + .await; + wait_after_broadcast(broadcast, async move { + Self::wait_for_response(sdk, state_transition, settings).await + }) + .await + } +} + +/// Construct and sign once; dispatch is chosen by the calling API. +#[allow(clippy::too_many_arguments)] +async fn prepare_transfer_state_transition>( + document: &Document, + recipient_id: Identifier, + sdk: &Sdk, + document_type: DocumentType, + identity_public_key: IdentityPublicKey, + token_payment_info: Option, + signer: &S, + settings: Option, +) -> Result { + // A local failure after the nonce is reserved would leave the cached nonce ahead of + // Platform's, so what can be refused without it is refused first. + if let Some(creation_options) = + settings.and_then(|settings| settings.state_transition_creation_options) + { + creation_options.validate_base_carries_action_fee_agreement(sdk.version())?; + } + + let new_identity_contract_nonce = sdk + .get_identity_contract_nonce( + document.owner_id(), + document_type.data_contract_id(), + true, + settings, + ) + .await?; + + let settings = settings.unwrap_or_default(); + + let transition = BatchTransition::new_document_transfer_transition_from_document( + document.clone(), + document_type.as_ref(), + recipient_id, + &identity_public_key, + new_identity_contract_nonce, + settings.user_fee_increase.unwrap_or_default(), + token_payment_info, + signer, + sdk.version(), + settings.state_transition_creation_options, + ) + .await?; + ensure_valid_state_transition_structure(&transition, sdk.version())?; + + Ok(transition) +} + +#[cfg(all(test, feature = "mocks"))] +mod submission_tests { + use super::*; + use crate::sdk::put_test_support::PutFixture; + use dapi_grpc::tonic::Status; + use drive_proof_verifier::types::IdentityContractNonceFetcher; + + #[tokio::test] + async fn should_wait_for_original_signed_transition_after_ambiguous_submission() { + for (status, expected_waits) in [ + (Status::unavailable("acknowledgement lost"), 1), + (Status::already_exists("tx already exists in cache"), 1), + ( + Status::resource_exhausted("broadcasting state transition is timed out"), + 1, + ), + (Status::resource_exhausted("mempool is full"), 0), + ] { + let mut fixture = PutFixture::new().await; + let prepared = prepare_transfer_state_transition( + &fixture.document, + Identifier::new([9; 32]), + &fixture.sdk, + fixture.document_type.clone(), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), ) - .await?; + .await + .unwrap(); + let requests = fixture.expect_submission(&prepared, status).await; + let result = fixture + .document + .transfer_document_to_identity_and_wait_for_response( + Identifier::new([9; 32]), + &fixture.sdk, + fixture.document_type.clone(), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()); + fixture + .assert_calls(&requests, expected_waits, result) + .await; + fixture + .sdk + .mock() + .remove_fetch_expectation::(( + fixture.document.owner_id(), + fixture.document_type.data_contract_id(), + )) + .await; + assert_eq!( + fixture + .sdk + .get_identity_contract_nonce( + fixture.document.owner_id(), + fixture.document_type.data_contract_id(), + false, + PutFixture::settings() + ) + .await + .unwrap(), + 1 + ); + } + } - Self::wait_for_response(sdk, state_transition, settings).await + #[tokio::test] + async fn should_preserve_plain_broadcast_error_without_waiting() { + let mut fixture = PutFixture::new().await; + let prepared = prepare_transfer_state_transition( + &fixture.document, + Identifier::new([9; 32]), + &fixture.sdk, + fixture.document_type.clone(), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .unwrap(); + let requests = fixture + .expect_submission(&prepared, Status::unavailable("acknowledgement lost")) + .await; + let result = fixture + .document + .transfer_document_to_identity( + Identifier::new([9; 32]), + &fixture.sdk, + fixture.document_type.clone(), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()); + fixture.assert_calls(&requests, 0, result).await; } } diff --git a/packages/rs-sdk/src/platform/transition/update_price_of_document.rs b/packages/rs-sdk/src/platform/transition/update_price_of_document.rs index fb013102fc2..1d3921e1904 100644 --- a/packages/rs-sdk/src/platform/transition/update_price_of_document.rs +++ b/packages/rs-sdk/src/platform/transition/update_price_of_document.rs @@ -1,6 +1,8 @@ use crate::{Error, Sdk}; -use super::broadcast::BroadcastStateTransition; +use super::broadcast::{ + broadcast_with_observation, wait_after_broadcast, BroadcastPolicy, BroadcastStateTransition, +}; use super::validation::ensure_valid_state_transition_structure; use super::waitable::Waitable; use crate::platform::transition::put_settings::PutSettings; @@ -58,42 +60,19 @@ impl> UpdatePriceOfDocument for Document { signer: &S, settings: Option, ) -> Result { - // A local failure after the nonce is reserved would leave the cached nonce ahead of - // Platform's, so what can be refused without it is refused first. - if let Some(creation_options) = - settings.and_then(|settings| settings.state_transition_creation_options) - { - creation_options.validate_base_carries_action_fee_agreement(sdk.version())?; - } - - let new_identity_contract_nonce = sdk - .get_identity_contract_nonce( - self.owner_id(), - document_type.data_contract_id(), - true, - settings, - ) - .await?; - - let settings = settings.unwrap_or_default(); - - let transition = BatchTransition::new_document_update_price_transition_from_document( - self.clone(), - document_type.as_ref(), + let transition = prepare_price_update_state_transition( + self, price, - &identity_public_key, - new_identity_contract_nonce, - settings.user_fee_increase.unwrap_or_default(), + sdk, + document_type, + identity_public_key, token_payment_info, signer, - sdk.version(), - settings.state_transition_creation_options, + settings, ) .await?; - ensure_valid_state_transition_structure(&transition, sdk.version())?; - // response is empty for a broadcast, result comes from the stream wait for state transition result - transition.broadcast(sdk, Some(settings)).await?; + transition.broadcast(sdk, settings).await?; Ok(transition) } @@ -107,22 +86,77 @@ impl> UpdatePriceOfDocument for Document { signer: &S, settings: Option, ) -> Result { - let state_transition = self - .update_price_of_document( - price, - sdk, - document_type, - identity_public_key, - token_payment_info, - signer, - settings, - ) - .await?; + let state_transition = prepare_price_update_state_transition( + self, + price, + sdk, + document_type, + identity_public_key, + token_payment_info, + signer, + settings, + ) + .await?; - Self::wait_for_response(sdk, state_transition, settings).await + let broadcast = + broadcast_with_observation(&state_transition, sdk, settings, BroadcastPolicy::Standard) + .await; + wait_after_broadcast(broadcast, async move { + Self::wait_for_response(sdk, state_transition, settings).await + }) + .await } } +/// Construct and sign once; dispatch is chosen by the calling API. +#[allow(clippy::too_many_arguments)] +async fn prepare_price_update_state_transition>( + document: &Document, + price: Credits, + sdk: &Sdk, + document_type: DocumentType, + identity_public_key: IdentityPublicKey, + token_payment_info: Option, + signer: &S, + settings: Option, +) -> Result { + // A local failure after the nonce is reserved would leave the cached nonce ahead of + // Platform's, so what can be refused without it is refused first. + if let Some(creation_options) = + settings.and_then(|settings| settings.state_transition_creation_options) + { + creation_options.validate_base_carries_action_fee_agreement(sdk.version())?; + } + + let new_identity_contract_nonce = sdk + .get_identity_contract_nonce( + document.owner_id(), + document_type.data_contract_id(), + true, + settings, + ) + .await?; + + let settings = settings.unwrap_or_default(); + + let transition = BatchTransition::new_document_update_price_transition_from_document( + document.clone(), + document_type.as_ref(), + price, + &identity_public_key, + new_identity_contract_nonce, + settings.user_fee_increase.unwrap_or_default(), + token_payment_info, + signer, + sdk.version(), + settings.state_transition_creation_options, + ) + .await?; + ensure_valid_state_transition_structure(&transition, sdk.version())?; + + Ok(transition) +} + #[cfg(test)] mod tests { use super::*; @@ -236,3 +270,88 @@ mod tests { ); } } + +#[cfg(all(test, feature = "mocks"))] +mod submission_tests { + use super::*; + use crate::sdk::put_test_support::PutFixture; + use dapi_grpc::tonic::Status; + + #[tokio::test] + async fn should_wait_for_original_signed_transition_after_ambiguous_submission() { + for (status, expected_waits) in [ + (Status::unavailable("acknowledgement lost"), 1), + (Status::already_exists("tx already exists in cache"), 1), + ( + Status::resource_exhausted("broadcasting state transition is timed out"), + 1, + ), + (Status::resource_exhausted("mempool is full"), 0), + ] { + let mut fixture = PutFixture::new().await; + let prepared = prepare_price_update_state_transition( + &fixture.document, + 10, + &fixture.sdk, + fixture.document_type.clone(), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .unwrap(); + let requests = fixture.expect_submission(&prepared, status).await; + let result = fixture + .document + .update_price_of_document_and_wait_for_response( + 10, + &fixture.sdk, + fixture.document_type.clone(), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()); + fixture + .assert_calls(&requests, expected_waits, result) + .await; + } + } + + #[tokio::test] + async fn should_preserve_plain_broadcast_error_without_waiting() { + let mut fixture = PutFixture::new().await; + let prepared = prepare_price_update_state_transition( + &fixture.document, + 10, + &fixture.sdk, + fixture.document_type.clone(), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .unwrap(); + let requests = fixture + .expect_submission(&prepared, Status::unavailable("acknowledgement lost")) + .await; + let result = fixture + .document + .update_price_of_document( + 10, + &fixture.sdk, + fixture.document_type.clone(), + fixture.key.clone(), + None, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()); + fixture.assert_calls(&requests, 0, result).await; + } +} diff --git a/packages/rs-sdk/src/platform/transition/vote.rs b/packages/rs-sdk/src/platform/transition/vote.rs index 8046e38e9f9..d050975eee7 100644 --- a/packages/rs-sdk/src/platform/transition/vote.rs +++ b/packages/rs-sdk/src/platform/transition/vote.rs @@ -1,5 +1,5 @@ +use super::broadcast::{broadcast_with_observation, wait_after_broadcast, BroadcastPolicy}; use crate::platform::query::VoteQuery; -use crate::platform::transition::broadcast_request::BroadcastRequestForStateTransition; use crate::platform::transition::put_settings::PutSettings; use crate::platform::transition::validation::ensure_valid_state_transition_structure; use crate::platform::Fetch; @@ -12,7 +12,6 @@ use dpp::prelude::Identifier; use dpp::state_transition::masternode_vote_transition::methods::MasternodeVoteTransitionMethodsV0; use dpp::state_transition::masternode_vote_transition::MasternodeVoteTransition; use dpp::voting::votes::Vote; -use rs_dapi_client::{DapiRequest, IntoInner}; use super::waitable::Waitable; @@ -89,12 +88,14 @@ impl> PutVote for Vote { ) .await?; ensure_valid_state_transition_structure(&masternode_vote_transition, sdk.version())?; - let request = masternode_vote_transition.broadcast_request_for_state_transition()?; - - request - .execute(sdk, settings.request_settings) - .await // TODO: We need better way to handle execution errors - .into_inner()?; + broadcast_with_observation( + &masternode_vote_transition, + sdk, + Some(settings), + BroadcastPolicy::TransportOnly, + ) + .await + .into_result()?; Ok(()) } @@ -128,37 +129,36 @@ impl> PutVote for Vote { ) .await?; ensure_valid_state_transition_structure(&masternode_vote_transition, sdk.version())?; - let request = masternode_vote_transition.broadcast_request_for_state_transition()?; - // TODO: Implement retry logic - let response_result = request - .execute(sdk, settings.request_settings) - .await - .into_inner(); - - match response_result { - Ok(_) => {} - //todo make this more reliable - Err(e) => { - if !e.to_string().contains("already exists") { - return Err(e.into()); - } - match self { - Vote::ResourceVote(_) => { - let vote = - Vote::fetch(sdk, VoteQuery::new(voter_pro_tx_hash, vote_poll_id)) - .await?; - return vote.ok_or(Error::Generic( - "vote was proved to not exist but was said to exist".to_string(), - )); - } - // `VoteQuery` reads the contested resource votes index only, so no query - // fetches a yes/no vote yet. The duplicate is this very transition, already - // delivered: wait for it as for one just broadcast, which proves it executed. - Vote::YesNoVote(_) => {} + let broadcast = broadcast_with_observation( + &masternode_vote_transition, + sdk, + Some(settings), + BroadcastPolicy::TransportOnly, + ) + .await; + // Preserve the resource-vote lookup; other ambiguous acknowledgements + // still reach the signed transition's verified wait. + if broadcast + .error() + .is_some_and(|error| error.to_string().contains("already exists")) + { + match self { + Vote::ResourceVote(_) => { + let vote = + Vote::fetch(sdk, VoteQuery::new(voter_pro_tx_hash, vote_poll_id)).await?; + return vote.ok_or(Error::Generic( + "vote was proved to not exist but was said to exist".to_string(), + )); } + // VoteQuery only reads resource votes. Yes/no duplicates must + // reach the existing proof wait for the same signed transition. + Vote::YesNoVote(_) => {} } } - Self::wait_for_response(sdk, masternode_vote_transition, Some(settings)).await + wait_after_broadcast(broadcast, async move { + Self::wait_for_response(sdk, masternode_vote_transition, Some(settings)).await + }) + .await } } @@ -179,3 +179,175 @@ fn get_voting_identity_id( &pub_key_hash, )) } + +#[cfg(all(test, feature = "mocks"))] +mod submission_tests { + use super::*; + use crate::platform::transition::masternode_vote_keys::voter_identity_voting_key; + use crate::sdk::put_test_support::PutFixture; + use dapi_grpc::tonic::Status; + use dpp::platform_value::{BinaryData, Value}; + use dpp::voting::vote_choices::resource_vote_choice::ResourceVoteChoice; + use dpp::voting::vote_choices::yes_no_abstain_vote_choice::YesNoAbstainVoteChoice; + use dpp::voting::vote_polls::contested_document_resource_vote_poll::ContestedDocumentResourceVotePoll; + use dpp::voting::vote_polls::yes_no_vote_poll::{YesNoMinimumVotingPower, YesNoVotePoll}; + use dpp::voting::vote_polls::VotePoll; + use dpp::voting::votes::resource_vote::v0::ResourceVoteV0; + use dpp::voting::votes::resource_vote::ResourceVote; + use dpp::voting::votes::yes_no_vote::v0::YesNoVoteV0; + use dpp::voting::votes::yes_no_vote::YesNoVote; + use drive_proof_verifier::types::IdentityNonceFetcher; + + #[tokio::test] + async fn should_wait_for_duplicate_yes_no_vote_without_resource_lookup_or_resubmission() { + let mut fixture = PutFixture::new().await; + let key = voter_identity_voting_key(&[3; 20]); + let pro_tx_hash = Identifier::new([4; 32]); + let voting_identity_id = get_voting_identity_id(pro_tx_hash, &key).unwrap(); + let vote = Vote::YesNoVote(YesNoVote::V0(YesNoVoteV0 { + vote_poll: YesNoVotePoll { + resource_path: vec![BinaryData::new(vec![0xc1; 32])], + supermajority_numerator: 2, + supermajority_denominator: 3, + minimum_voting_power: YesNoMinimumVotingPower::Absolute { voting_power: 400 }, + }, + vote_choice: YesNoAbstainVoteChoice::Yes, + })); + let prepared = MasternodeVoteTransition::try_from_vote_with_signer( + vote.clone(), + &fixture.signer, + pro_tx_hash, + &key, + 1, + fixture.sdk.version(), + None, + ) + .await + .unwrap(); + let requests = fixture + .expect_submission(&prepared, Status::already_exists("vote already exists")) + .await; + fixture + .sdk + .mock() + .expect_fetch(voting_identity_id, Some(IdentityNonceFetcher(0))) + .await + .unwrap(); + // No resource VoteQuery expectation: that path cannot retrieve yes/no votes. + let result = vote + .put_to_platform_and_wait_for_response( + pro_tx_hash, + &key, + &fixture.sdk, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()); + fixture.assert_calls(&requests, 1, result).await; + } + + #[tokio::test] + async fn should_observe_votes_without_extra_dispatches_or_nonce_invalidation() { + for combined in [false, true] { + for (status, expected_waits, existing) in [ + (Status::unavailable("acknowledgement lost"), 1, false), + ( + Status::resource_exhausted("broadcasting state transition is timed out"), + 1, + false, + ), + (Status::resource_exhausted("mempool is full"), 0, false), + (Status::already_exists("vote already exists"), 0, true), + ] { + let mut fixture = PutFixture::new().await; + let key = voter_identity_voting_key(&[3; 20]); + let pro_tx_hash = Identifier::new([4; 32]); + let voting_identity_id = get_voting_identity_id(pro_tx_hash, &key).unwrap(); + let poll = VotePoll::ContestedDocumentResourceVotePoll( + ContestedDocumentResourceVotePoll { + contract_id: Identifier::new([5; 32]), + document_type_name: "domain".into(), + index_name: "parentNameAndLabel".into(), + index_values: vec![Value::Text("dash".into()), Value::Text("test".into())], + }, + ); + let vote = Vote::ResourceVote(ResourceVote::V0(ResourceVoteV0 { + vote_poll: poll.clone(), + resource_vote_choice: ResourceVoteChoice::Abstain, + })); + let prepared = MasternodeVoteTransition::try_from_vote_with_signer( + vote.clone(), + &fixture.signer, + pro_tx_hash, + &key, + 1, + fixture.sdk.version(), + None, + ) + .await + .unwrap(); + let requests = fixture.expect_submission(&prepared, status).await; + fixture + .sdk + .mock() + .expect_fetch(voting_identity_id, Some(IdentityNonceFetcher(0))) + .await + .unwrap(); + if existing && combined { + let poll_id = poll.unique_id(fixture.sdk.version()).unwrap(); + fixture + .sdk + .mock() + .expect_fetch(VoteQuery::new(pro_tx_hash, poll_id), None::) + .await + .unwrap(); + } + let result = if combined { + vote.put_to_platform_and_wait_for_response( + pro_tx_hash, + &key, + &fixture.sdk, + &fixture.signer, + PutFixture::settings(), + ) + .await + .map(|_| ()) + } else { + vote.put_to_platform( + pro_tx_hash, + &key, + &fixture.sdk, + &fixture.signer, + PutFixture::settings(), + ) + .await + }; + if existing && combined { + assert!( + matches!(&result, Err(Error::Generic(message)) if message.contains("vote was proved to not exist")), + "preserve the legacy fetch fallback: {result:?}" + ); + } + fixture + .assert_calls(&requests, if combined { expected_waits } else { 0 }, result) + .await; + // No new invalidation: removing transport's nonce expectation still + // allows the next nonce read to use the original cached reservation. + fixture + .sdk + .mock() + .remove_fetch_expectation::(voting_identity_id) + .await; + assert_eq!( + fixture + .sdk + .get_identity_nonce(voting_identity_id, false, PutFixture::settings()) + .await + .unwrap(), + 1 + ); + } + } + } +} diff --git a/packages/rs-sdk/src/sdk.rs b/packages/rs-sdk/src/sdk.rs index 8ec06f889c5..5b9e74312a9 100644 --- a/packages/rs-sdk/src/sdk.rs +++ b/packages/rs-sdk/src/sdk.rs @@ -36,7 +36,8 @@ pub use rs_dapi_client::AddressBanInfo; pub use rs_dapi_client::AddressList; pub use rs_dapi_client::RequestSettings; use rs_dapi_client::{ - transport::TransportRequest, DapiClient, DapiClientError, DapiRequestExecutor, ExecutionResult, + transport::{TransportError, TransportRequest}, + DapiClient, DapiClientError, DapiRequestExecutor, ExecutionResult, }; use std::fmt::Debug; #[cfg(feature = "mocks")] @@ -58,6 +59,8 @@ mod identity_discovery_tests; mod protocol_gate; #[cfg(all(test, feature = "mocks"))] mod protocol_gate_tests; +#[cfg(all(test, feature = "mocks"))] +pub(crate) mod put_test_support; mod quorum_key; #[cfg(all(test, feature = "mocks"))] mod signed_version_tests; @@ -1097,9 +1100,24 @@ impl DapiRequestExecutor for Sdk { &self, request: R, settings: RequestSettings, + ) -> ExecutionResult { + self.execute_with_transport_error_observer(request, settings, &|_| {}) + .await + } +} + +impl Sdk { + pub(crate) async fn execute_with_transport_error_observer( + &self, + request: R, + settings: RequestSettings, + observer: &(dyn Fn(&TransportError) + Send + Sync), ) -> ExecutionResult { match self.inner { - SdkInstance::Dapi { ref dapi, .. } => dapi.execute(request, settings).await, + SdkInstance::Dapi { ref dapi, .. } => { + dapi.execute_with_transport_error_observer(request, settings, observer) + .await + } #[cfg(feature = "mocks")] SdkInstance::Mock { ref dapi, .. } => { let dapi_guard = dapi.lock().await; diff --git a/packages/rs-sdk/src/sdk/put_test_support.rs b/packages/rs-sdk/src/sdk/put_test_support.rs new file mode 100644 index 00000000000..8de876c4d5c --- /dev/null +++ b/packages/rs-sdk/src/sdk/put_test_support.rs @@ -0,0 +1,192 @@ +//! In-memory transport fixtures for the real public combined put methods. +use super::{SdkBuilder, SdkInstance}; +use crate::platform::transition::broadcast_request::BroadcastRequestForStateTransition; +use crate::platform::transition::put_settings::PutSettings; +use crate::{Error, Sdk}; +use dapi_grpc::platform::v0::{ + BroadcastStateTransitionRequest, WaitForStateTransitionResultRequest, + WaitForStateTransitionResultResponse, +}; +use dapi_grpc::tonic::Status; +use dpp::address_funds::AddressWitness; +use dpp::data_contract::accessors::v0::DataContractV0Getters; +use dpp::data_contract::document_type::random_document::CreateRandomDocument; +use dpp::data_contract::document_type::DocumentType; +use dpp::document::{Document, DocumentV0Getters, DocumentV0Setters}; +use dpp::identity::signer::Signer; +use dpp::identity::IdentityPublicKey; +use dpp::platform_value::BinaryData; +use dpp::prelude::{DataContract, Identifier}; +use dpp::state_transition::StateTransition; +use dpp::system_data_contracts::{load_system_data_contract, SystemDataContract}; +use dpp::version::PlatformVersion; +use dpp::ProtocolError; +use drive_proof_verifier::types::{IdentityContractNonceFetcher, IdentityNonceFetcher}; +use rs_dapi_client::transport::TransportError; +use rs_dapi_client::{DapiClientError, ExecutionError, ExecutionResponse, RequestSettings}; +use std::sync::atomic::{AtomicUsize, Ordering}; + +#[derive(Debug, Default)] +pub(crate) struct CountingSigner(pub AtomicUsize); +#[async_trait::async_trait] +impl Signer for CountingSigner { + async fn sign(&self, _: &IdentityPublicKey, _: &[u8]) -> Result { + self.0.fetch_add(1, Ordering::SeqCst); + Ok(vec![7; 65].into()) + } + async fn sign_create_witness( + &self, + _: &IdentityPublicKey, + _: &[u8], + ) -> Result { + panic!("identity-key operation never requests an address witness") + } + fn can_sign_with(&self, _: &IdentityPublicKey) -> bool { + true + } +} + +pub(crate) struct PutFixture { + pub sdk: Sdk, + pub document: Document, + pub document_type: DocumentType, + pub contract: DataContract, + pub key: IdentityPublicKey, + pub signer: CountingSigner, +} + +impl PutFixture { + pub async fn new() -> Self { + let version = PlatformVersion::latest(); + let mut sdk = SdkBuilder::new_mock() + .with_version(version) + .build() + .unwrap(); + let contract = load_system_data_contract(SystemDataContract::DPNS, version).unwrap(); + let document_type = contract.document_type_cloned_for_name("preorder").unwrap(); + let mut document = document_type.random_document(Some(1), version).unwrap(); + document.set_revision(Some(2)); + let key = + IdentityPublicKey::random_ecdsa_critical_level_authentication_key(1, Some(1), version) + .unwrap() + .0; + for owner in [ + document.owner_id(), + contract.owner_id(), + Identifier::new([9; 32]), + ] { + sdk.mock() + .expect_fetch(owner, Some(IdentityNonceFetcher(0))) + .await + .unwrap(); + sdk.mock() + .expect_fetch( + (owner, contract.id()), + Some(IdentityContractNonceFetcher(0)), + ) + .await + .unwrap(); + } + Self { + sdk, + document, + document_type, + contract, + key, + signer: CountingSigner::default(), + } + } + + pub fn settings() -> Option { + Some(PutSettings { + request_settings: RequestSettings { + retries: Some(0), + ..Default::default() + }, + ..Default::default() + }) + } + + /// Preparation is local; use a fresh nonce cache before invoking the public method so + /// the independently repeated preparation must produce exactly the same bytes. + pub async fn expect_submission( + &mut self, + transition: &StateTransition, + status: Status, + ) -> ( + BroadcastStateTransitionRequest, + WaitForStateTransitionResultRequest, + ) { + self.sdk = Self::new().await.sdk; + self.signer.0.store(0, Ordering::SeqCst); + let broadcast = transition.broadcast_request_for_state_transition().unwrap(); + let wait = transition + .wait_for_state_transition_result_request() + .unwrap(); + let SdkInstance::Mock { dapi, .. } = &self.sdk.inner else { + unreachable!() + }; + let mut dapi = dapi.lock().await; + dapi.expect( + &broadcast, + &Err(ExecutionError { + inner: DapiClientError::Transport(TransportError::Grpc(status)), + address: None, + retries: 0, + }), + ) + .unwrap(); + // An empty response is never valid proof. Reaching it must fail closed, + // while its request hash proves we waited for the original signed bytes. + dapi.expect( + &wait, + &Ok(ExecutionResponse { + inner: WaitForStateTransitionResultResponse::default(), + retries: 0, + address: "http://127.0.0.1".parse().unwrap(), + }), + ) + .unwrap(); + (broadcast, wait) + } + + pub async fn assert_calls( + &self, + requests: &( + BroadcastStateTransitionRequest, + WaitForStateTransitionResultRequest, + ), + expected_waits: usize, + result: Result<(), Error>, + ) { + assert!( + result.is_err(), + "an acknowledgement or invalid proof cannot become success" + ); + let SdkInstance::Mock { dapi, .. } = &self.sdk.inner else { + unreachable!() + }; + let dapi = dapi.lock().await; + assert_eq!( + dapi.request_count(&requests.0), + 1, + "submit the originally signed bytes exactly once: {result:?}" + ); + assert_eq!( + dapi.request_count(&requests.1), + expected_waits, + "poll the original transition hash only when appropriate" + ); + assert_eq!( + self.signer.0.load(Ordering::SeqCst), + 1, + "the public operation must sign once" + ); + if expected_waits == 1 { + assert!( + !matches!(result, Err(Error::AlreadyExists(_))), + "return the failed proof, not the ambiguous relay acknowledgement" + ); + } + } +} diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletResult.swift b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletResult.swift index e7e746105e6..d6e8088b1fa 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletResult.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/PlatformWallet/PlatformWalletResult.swift @@ -46,12 +46,10 @@ public enum PlatformWalletResultCode: Int32, Sendable { /// shielded sync reach a confirmed state and try again. Distinct from /// `errorShieldedSpendUnconfirmed`, which must NOT be retried. case errorShieldedNoRecordedAnchor = 19 - /// A core transaction broadcast (send, DashPay payment, or asset-lock - /// funding) failed with an ambiguous outcome — the transaction may - /// already be on the network. The wallet keeps the spent inputs' UTXO - /// reservation, so an immediate retry fails at input selection instead - /// of double-spending; the reservation TTL or a sync reconciles the - /// outcome. Do NOT auto-retry. + /// A Core transaction or address-funded identity top-up may have executed + /// but its outcome could not be confirmed. Reconcile the original operation + /// before another submission. Core input protections do not apply to + /// Platform address top-ups. Do NOT auto-retry. case errorTransactionBroadcastUnconfirmed = 20 /// A masternode (evonode) identity credit withdrawal was broadcast and /// accepted, but its execution result could not be confirmed — it may @@ -611,11 +609,11 @@ public enum PlatformWalletError: LocalizedError { /// retained public Swift/FFI name, this is not a shielded-pool balance /// failure. case shieldedInsufficientBalance(String) - /// A core transaction broadcast was submitted but its outcome is - /// unknown — the transaction may already be on the network. The wallet - /// keeps the spent inputs reserved so a retry cannot double-spend; the - /// reservation TTL or a later sync reconciles the outcome. Do NOT - /// auto-retry. Core sibling of `shieldedSpendUnconfirmed`. + /// A Core transaction or address-funded identity top-up may have executed, + /// but its outcome is unknown. Sync and reconcile the original operation + /// before another submission. Core transactions retain their input + /// protections; Platform address top-ups have no Core UTXO reservation. + /// Do NOT auto-retry. Sibling of `shieldedSpendUnconfirmed`. case transactionBroadcastUnconfirmed(String) /// A masternode (evonode) credit withdrawal was broadcast and accepted /// but its result could not be confirmed. It may already have executed diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/SDK.swift b/packages/swift-sdk/Sources/SwiftDashSDK/SDK.swift index ed62fecd35e..6620a5fe407 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/SDK.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/SDK.swift @@ -611,6 +611,9 @@ public enum SDKError: Error { case timeout(String) case notImplemented(String) case internalError(String) + /// The transition may have executed. Reconcile the original operation + /// before submitting again; the associated diagnostic is not a safe retry verdict. + case stateTransitionResultUnconfirmed(String) /// Platform refused the state transition. Carries the rs-dpp consensus /// code and family, so a host can branch on the exact rule that refused it /// (`error.consensusError?.code == 10422` is a violated propertyConstraints @@ -629,6 +632,10 @@ public enum SDKError: Error { public static func fromDashSDKError(_ error: DashSDKError) -> SDKError { let message = error.message != nil ? String(cString: error.message!) : "Unknown error" + if error.code == StateTransitionResultUnconfirmed { + return .stateTransitionResultUnconfirmed(message) + } + // Platform's verdict wins over the coarse code, which rs-sdk-ffi picks // from the rendered text: the same refusal can arrive as ProtocolError or // InternalError depending on its wording. @@ -663,8 +670,9 @@ public enum SDKError: Error { } /// The error to throw under `message` when an FFI call that signs, - /// broadcasts or waits for a state transition fails: `.consensusRejection` - /// when `ffiError` carries Platform's verdict, otherwise + /// broadcasts or waits for a state transition fails: an unconfirmed result + /// retains its submission status; `.consensusRejection` when `ffiError` + /// carries Platform's definitive verdict, otherwise /// `fallback(message)`, the case the call site threw before the verdict /// crossed the FFI. Read `ffiError` before freeing it. static func stateTransitionFailure( @@ -672,6 +680,9 @@ public enum SDKError: Error { ffiError: DashSDKError?, otherwise fallback: (String) -> SDKError = SDKError.internalError ) -> SDKError { + if let ffiError, ffiError.code == StateTransitionResultUnconfirmed { + return .stateTransitionResultUnconfirmed(message) + } if let ffiError, let consensus = PlatformConsensusError(ffi: ffiError) { return .consensusRejection(consensus, message) } @@ -702,6 +713,8 @@ extension SDKError: LocalizedError { return "Feature Not Implemented: \(message)" case .internalError(let message): return "Internal Error: \(message)" + case .stateTransitionResultUnconfirmed(let message): + return "Submission result is unconfirmed. Check history and reconcile the original operation before submitting again. \(message)" case .consensusRejection(_, let message): return "Rejected by Platform: \(message)" case .unknown(let message): diff --git a/packages/swift-sdk/Sources/SwiftDashSDK/Utils/ErrorHandling.swift b/packages/swift-sdk/Sources/SwiftDashSDK/Utils/ErrorHandling.swift index 7c17fd2d9e3..963756ff702 100644 --- a/packages/swift-sdk/Sources/SwiftDashSDK/Utils/ErrorHandling.swift +++ b/packages/swift-sdk/Sources/SwiftDashSDK/Utils/ErrorHandling.swift @@ -124,6 +124,12 @@ public enum ErrorFormatter { /// Extract a user-friendly message from any error public static func userFriendlyMessage(from error: Error) -> String { + if let sdkError = error as? SDKError, case .stateTransitionResultUnconfirmed = sdkError { + return sdkError.localizedDescription + } + if let walletError = error as? PlatformWalletError, case .transactionBroadcastUnconfirmed = walletError { + return "Submission result is unconfirmed. Check history and reconcile the original operation before submitting again. \(walletError.localizedDescription)" + } // Handle specific SDK error types let errorString = String(describing: error) @@ -323,6 +329,25 @@ public enum ErrorCategorizer { return userFacing } + if let sdkError = error as? SDKError, case .stateTransitionResultUnconfirmed = sdkError { + return UserFacingError( + title: "Submission unconfirmed", + message: sdkError.localizedDescription, + category: .unknown, + underlyingError: String(describing: sdkError), + isRetryable: false + ) + } + if let walletError = error as? PlatformWalletError, case .transactionBroadcastUnconfirmed = walletError { + return UserFacingError( + title: "Submission unconfirmed", + message: ErrorFormatter.userFriendlyMessage(from: walletError), + category: .unknown, + underlyingError: String(describing: walletError), + isRetryable: false + ) + } + let category = categorize(error) let message = ErrorFormatter.userFriendlyMessage(from: error) let suggestion = ErrorRecovery.suggestion(for: category) diff --git a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DiagnosticsView.swift b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DiagnosticsView.swift index ebcac4b725c..e206f22553e 100644 --- a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DiagnosticsView.swift +++ b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/DiagnosticsView.swift @@ -621,6 +621,8 @@ struct DiagnosticsView: View { return "Not Implemented: \(msg)" case .internalError(let msg): return "Internal Error: \(msg)" + case .stateTransitionResultUnconfirmed: + return sdkError.localizedDescription case .consensusRejection(let consensus, let msg): return "Rejected by Platform (code \(consensus.code)): \(msg)" case .unknown(let msg): diff --git a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/QueryDetailView.swift b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/QueryDetailView.swift index a368658e44b..4a39d778f78 100644 --- a/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/QueryDetailView.swift +++ b/packages/swift-sdk/SwiftExampleApp/SwiftExampleApp/Views/QueryDetailView.swift @@ -201,6 +201,8 @@ struct QueryDetailView: View { self.error = "Not Implemented: \(message)" case .internalError(let message): self.error = "Internal Error: \(message)" + case .stateTransitionResultUnconfirmed: + self.error = sdkError.localizedDescription case .consensusRejection(let consensus, let message): self.error = "Rejected by Platform (code \(consensus.code)): \(message)" case .unknown(let message): diff --git a/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/SDKUnconfirmedErrorTests.swift b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/SDKUnconfirmedErrorTests.swift new file mode 100644 index 00000000000..3965f6e2cd1 --- /dev/null +++ b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/SDKUnconfirmedErrorTests.swift @@ -0,0 +1,89 @@ +import DashSDKFFI +import Foundation +import XCTest +@testable import SwiftDashSDK + +final class SDKUnconfirmedErrorTests: XCTestCase { + func testDirectSDKDecoderPreservesUnconfirmedResultAfterMessageIsFreed() { + XCTAssertEqual(StateTransitionResultUnconfirmed.rawValue, 11) + for detail in ["wait timed out", "invalid proof", "address nonce mismatch"] { + let rendered = "Identity top-up may have executed: \(detail)" + let message = strdup(rendered) + let decoded = SDKError.fromDashSDKError(DashSDKError( + code: StateTransitionResultUnconfirmed, + message: message, + consensus_code: 0, + consensus_kind: ConsensusErrorKindNone + )) + free(message) + + guard case .stateTransitionResultUnconfirmed(let preserved) = decoded else { + return XCTFail("Expected unconfirmed result, got \(decoded)") + } + XCTAssertEqual(preserved, rendered) + XCTAssertNil(decoded.consensusError) + XCTAssertTrue(decoded.localizedDescription.contains("before submitting again")) + XCTAssertTrue(decoded.localizedDescription.contains(rendered)) + } + } + + func testUnconfirmedCodeTakesPrecedenceOverInnerConsensusForBothHelpers() { + let ffi = DashSDKError( + code: StateTransitionResultUnconfirmed, + message: nil, + consensus_code: 40603, + consensus_kind: ConsensusErrorKindState + ) + // Rust emits no consensus fields for this code. Even contradictory + // input must not turn a possibly completed write into a safe refusal. + let contextual = SDKError.stateTransitionFailure( + "Top-up result unconfirmed: address nonce mismatch", ffiError: ffi, + otherwise: SDKError.timeout + ) + for error in [SDKError.fromDashSDKError(ffi), contextual] { + guard case .stateTransitionResultUnconfirmed = error else { + return XCTFail("Expected unconfirmed result, got \(error)") + } + XCTAssertNil(error.consensusError) + } + guard case .stateTransitionResultUnconfirmed(let message) = contextual else { return } + XCTAssertEqual(message, "Top-up result unconfirmed: address nonce mismatch") + } + + func testDefiniteRefusalAndOrdinaryTimeoutKeepExistingCases() { + let refusal = DashSDKError( + code: DashSDKErrorCode(rawValue: 5), message: nil, + consensus_code: 40603, consensus_kind: ConsensusErrorKindState + ) + for error in [ + SDKError.fromDashSDKError(refusal), + SDKError.stateTransitionFailure("Refused", ffiError: refusal) + ] { + XCTAssertEqual(error.consensusError?.code, 40603) + } + let timeout = DashSDKError( + code: DashSDKErrorCode(rawValue: 8), message: nil, + consensus_code: 0, consensus_kind: ConsensusErrorKindNone + ) + guard case .timeout = SDKError.fromDashSDKError(timeout) else { + return XCTFail("An ordinary timeout must keep its existing case") + } + guard case .internalError = SDKError.stateTransitionFailure("Failed", ffiError: timeout) else { + return XCTFail("An unmarked contextual failure must keep its existing fallback") + } + } + + func testUserFacingHelpersPreserveUnconfirmedWarningWithoutSuggestingRetry() { + for detail in ["timeout", "networkError connection lost", "invalidParameter nonce mismatch"] { + let error = SDKError.stateTransitionResultUnconfirmed(detail) + let displayed = ErrorCategorizer.toUserFacingError(error) + XCTAssertEqual(ErrorFormatter.userFriendlyMessage(from: error), error.localizedDescription) + XCTAssertEqual(displayed.message, error.localizedDescription) + XCTAssertTrue(displayed.message.contains(detail)) + XCTAssertTrue(displayed.message.contains("before submitting again")) + XCTAssertFalse(displayed.isRetryable) + XCTAssertNil(displayed.recoverySuggestion) + } + XCTAssertTrue(ErrorCategorizer.toUserFacingError(SDKError.timeout("timeout")).isRetryable) + } +} diff --git a/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/TopUpUnconfirmedErrorTests.swift b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/TopUpUnconfirmedErrorTests.swift new file mode 100644 index 00000000000..a55c09962f2 --- /dev/null +++ b/packages/swift-sdk/SwiftTests/SwiftDashSDKTests/TopUpUnconfirmedErrorTests.swift @@ -0,0 +1,35 @@ +import DashSDKFFI +import Foundation +import XCTest +@testable import SwiftDashSDK + +final class TopUpUnconfirmedErrorTests: XCTestCase { + func testTopUpUnconfirmedResultPreservesExistingTypedError() throws { + XCTAssertEqual(PlatformWalletResultCode.errorTransactionBroadcastUnconfirmed.rawValue, 20) + for detail in [ + "Identity top-up may have executed; reconcile before submitting again", + "wait timeout", + "networkError: connection lost" + ] { + let ffi = PlatformWalletFFIResult( + code: PLATFORM_WALLET_FFI_RESULT_CODE_ERROR_TRANSACTION_BROADCAST_UNCONFIRMED, + message: strdup(detail) + ) + XCTAssertThrowsError(try ffi.check()) { error in + guard case PlatformWalletError.transactionBroadcastUnconfirmed(let message) = error else { + return XCTFail("Expected typed unconfirmed result, got \(error)") + } + XCTAssertEqual(message, detail) + XCTAssertEqual(error.localizedDescription, detail) + let formatted = ErrorFormatter.userFriendlyMessage(from: error) + let displayed = ErrorCategorizer.toUserFacingError(error) + XCTAssertTrue(formatted.contains("before submitting again")) + XCTAssertTrue(formatted.contains(detail)) + XCTAssertEqual(displayed.message, formatted) + XCTAssertFalse(displayed.isRetryable) + XCTAssertNil(displayed.recoverySuggestion) + } + } + XCTAssertTrue(ErrorCategorizer.toUserFacingError(SDKError.timeout("timeout")).isRetryable) + } +} diff --git a/packages/wasm-sdk/src/error.rs b/packages/wasm-sdk/src/error.rs index 387b7750f7c..3808e5be22a 100644 --- a/packages/wasm-sdk/src/error.rs +++ b/packages/wasm-sdk/src/error.rs @@ -186,6 +186,12 @@ impl From for WasmSdkError { ExecutionNotProved(msg) => { Self::new(WasmSdkErrorKind::ExecutionNotProved, msg, None, false) } + StateTransitionResultUnconfirmed(source) => Self::new( + WasmSdkErrorKind::Generic, + format!("identity top-up result is unconfirmed; reconcile before re-submitting: {source}"), + None, + false, + ), InvalidProvedResponse(msg) => Self::new( WasmSdkErrorKind::InvalidProvedResponse, msg,