use std::fmt;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use thruput_node_supervisor::{BlockDriverConfig, LogLevel, ManualBlockProducer, ProducedBlock};
use thruput_json_rpc_client::native::{
NativeClient, NativeClientError,
receipt::{ReceiptPollConfig, poll_receipt},
};
use thruput_json_rpc_client::{HttpClient, HttpClientBuilder};
use thruput_json_rpc_types::runtime_manifest::RuntimeManifest;
use thruput_json_rpc_types::transaction::TxHash;
use thruput_json_rpc_types::{
NetworkHash,
requests::{self, EmptyRequest, EventQuery, TransactionQuery},
responses::{
ChainTip, DecodedResponse, EventPage, SimulationResult, StorageResult, TransactionReceipt,
TransactionStatus,
},
};
use thruput_json_rpc_types::{base58::Base58, hex::HexBytes};
use thruput_xdr::contract::ContractEntrypointID;
use thruput_xdr::types::ContractAddress;
use crate::error::{Result, TestkitError};
const RECEIPT_POLL_INTERVAL: Duration = Duration::from_millis(250);
const RECEIPT_TIMEOUT: Duration = Duration::from_secs(10);
const PENDING_POLL_INTERVAL: Duration = Duration::from_millis(25);
const PENDING_POLL_ATTEMPTS: usize = 400;
#[derive(Clone)]
pub struct Network {
inner: Arc<NetworkInner>,
}
#[derive(Debug, Clone)]
#[must_use = "submitted transactions should be awaited or intentionally retained"]
pub struct SubmittedTransaction {
network: Network,
tx_hash: TxHash,
tx_hex: String,
}
impl SubmittedTransaction {
pub(crate) fn new(network: Network, tx_hash: TxHash, tx_hex: String) -> Self {
Self {
network,
tx_hash,
tx_hex,
}
}
pub fn signed_transaction(&self) -> &str {
&self.tx_hex
}
pub fn tx_hash(&self) -> TxHash {
self.tx_hash
}
pub async fn await_pending(&self) -> Result<()> {
self.network.await_pending(self.tx_hash).await
}
pub async fn await_receipt(&self) -> Result<TransactionReceipt> {
self.network.await_receipt(self.tx_hash).await
}
}
#[derive(Debug, PartialEq, Eq)]
struct FundingEvidence {
tx_hash: TxHash,
signed_transaction: HexBytes,
}
#[derive(Debug)]
struct LocalFundingAttempt {
request: requests::FundAccountRequest,
admitted: Option<thruput_json_rpc_types::responses::FundingSubmission>,
evidence: Option<FundingEvidence>,
}
impl LocalFundingAttempt {
fn retained_evidence(&self) -> Option<Box<(TxHash, HexBytes)>> {
self.evidence
.as_ref()
.map(|e| Box::new((e.tx_hash, e.signed_transaction.clone())))
}
fn accept_response(
&mut self,
response: &thruput_json_rpc_types::responses::FundingSubmission,
) -> Result<()> {
validate_funding_response(response, &self.request).map_err(|error| {
TestkitError::InvalidFundingResponse {
response: Box::new(response.clone()),
retained: self.retained_evidence(),
detail: error.to_string(),
}
})?;
let evidence = FundingEvidence {
tx_hash: response.transaction_hash,
signed_transaction: response.signed_transaction.clone(),
};
if self
.evidence
.as_ref()
.is_some_and(|original| original != &evidence)
{
return Err(TestkitError::InvalidFundingResponse {
response: Box::new(response.clone()),
retained: self.retained_evidence(),
detail: "local funding replaced retained signed evidence".into(),
});
}
self.admitted = Some(response.clone());
Ok(())
}
fn retain_error_evidence(&mut self, error: &NativeClientError) -> Result<()> {
use thruput_json_rpc_types::error::native::{DecodedError, ErrorData};
if let NativeClientError::Application { data, .. } = error
&& let DecodedError::Known { data, .. } = data.as_ref()
{
let (tx_hash, signed_transaction) = match data.as_ref() {
ErrorData::LocalFundingUnresolved {
transaction_hash,
signed_transaction,
..
} => (*transaction_hash, signed_transaction.clone()),
ErrorData::FundingAttemptStopped(stopped) => {
let (Some(hash), Some(bytes)) =
(stopped.transaction_hash, &stopped.signed_transaction)
else {
return Ok(());
};
(hash, bytes.clone())
}
_ => return Ok(()),
};
let evidence = FundingEvidence {
tx_hash,
signed_transaction,
};
validate_funding_evidence(&evidence)?;
if self
.evidence
.as_ref()
.is_some_and(|original| original != &evidence)
{
return Err(TestkitError::signing(
"local funding replaced retained signed evidence",
));
}
self.evidence = Some(evidence);
}
Ok(())
}
}
struct NetworkInner {
rpc_url: Option<String>,
client: Option<NativeClient<HttpClient>>,
network_hash: Option<NetworkHash>,
faucet: tokio::sync::Mutex<Option<crate::manifest::TestAccount>>,
local_faucet: Option<ContractAddress>,
local_funding:
tokio::sync::Mutex<std::collections::HashMap<ContractAddress, LocalFundingAttempt>>,
manual_block_producer: Option<ManualBlockProducer>,
}
impl fmt::Debug for Network {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Network")
.field("rpc_url", &self.inner.rpc_url)
.field("online", &self.inner.client.is_some())
.field(
"manual_block_production",
&self.inner.manual_block_producer.is_some(),
)
.finish_non_exhaustive()
}
}
impl Network {
pub(crate) async fn connect(
rpc_url: &str,
manifest: &RuntimeManifest,
manual_block_producer: Option<ManualBlockProducer>,
) -> Result<Self> {
let transport = HttpClientBuilder::default()
.build(rpc_url)
.map_err(|e| TestkitError::rpc("connect", e))?;
let client = NativeClient::new(transport);
thruput_client::check_endpoint(
&client,
&[
"getNetworkInfo",
"sendTransaction",
"getTransactionStatus",
"getTransactionReceipt",
],
)
.await
.map_err(TestkitError::native)?;
thruput_client::validate_network(&client, &manifest.network_hash.to_network_hash())
.await
.map_err(|error| match error {
thruput_client::NativeRpcError::NetworkMismatch { .. } => TestkitError::Manifest {
field: "networkHash",
detail: format!("{error} at {rpc_url}"),
},
error => TestkitError::native(error),
})?;
Ok(Self {
inner: Arc::new(NetworkInner {
rpc_url: Some(rpc_url.to_owned()),
client: Some(client),
network_hash: Some(Base58(manifest.network_hash.to_network_hash().into_inner())),
faucet: tokio::sync::Mutex::new(crate::manifest::derive_faucet(manifest)?),
local_faucet: local_faucet(manifest)?,
local_funding: tokio::sync::Mutex::new(std::collections::HashMap::new()),
manual_block_producer,
}),
})
}
pub(crate) fn offline() -> Self {
Self {
inner: Arc::new(NetworkInner {
rpc_url: None,
client: None,
network_hash: None,
faucet: tokio::sync::Mutex::new(None),
local_faucet: None,
local_funding: tokio::sync::Mutex::new(std::collections::HashMap::new()),
manual_block_producer: None,
}),
}
}
pub fn rpc_url(&self) -> Option<&str> {
self.inner.rpc_url.as_deref()
}
pub fn client(&self) -> Option<&NativeClient<HttpClient>> {
self.inner.client.as_ref()
}
pub async fn chain_tip(&self) -> Result<ChainTip> {
known(
self.require_client("getChainTip")?
.get_chain_tip(EmptyRequest {})
.await,
"getChainTip",
)
}
pub async fn submit_raw(&self, tx_hex: String) -> Result<TxHash> {
let bytes = thruput_json_rpc_types::hex::decode_prefixed_hex(&tx_hex)
.map_err(TestkitError::signing)?;
let signed: thruput_xdr::transaction::SignedTransaction =
xdrs::txs::decode(&bytes).map_err(TestkitError::signing)?;
let hash = TxHash(Base58(
thruput_signing::transaction_hash(&signed.tx).map_err(TestkitError::signing)?,
));
let canonical = thruput_json_rpc_types::hex::prefixed_hex(&bytes);
crate::account::submit_signed(self, hash, &canonical).await?;
Ok(hash)
}
pub async fn receipt(&self, tx_hash: TxHash) -> Result<Option<TransactionReceipt>> {
let receipt: Option<TransactionReceipt> = known(
self.require_client("getTransactionReceipt")?
.get_transaction_receipt(TransactionQuery {
transaction_hash: tx_hash,
})
.await,
"getTransactionReceipt",
)?;
if let Some(observed) = &receipt
&& observed.transaction_hash != tx_hash
{
return Err(TestkitError::ReceiptHashMismatch {
expected: tx_hash,
receipt: Box::new(observed.clone()),
});
}
if receipt
.as_ref()
.is_some_and(|r| r.transaction_hash == tx_hash)
{
self.inner.local_funding.lock().await.retain(|_, attempt| {
!attempt
.admitted
.as_ref()
.is_some_and(|admitted| admitted.transaction_hash == tx_hash)
&& !attempt
.evidence
.as_ref()
.is_some_and(|evidence| evidence.tx_hash == tx_hash)
});
}
Ok(receipt)
}
pub async fn await_receipt(&self, tx_hash: TxHash) -> Result<TransactionReceipt> {
let client = self.require_client("getTransactionReceipt")?;
let network_hash = self.inner.network_hash.ok_or(TestkitError::Offline {
operation: "getTransactionReceipt",
})?;
let receipt = poll_receipt(
client,
network_hash,
tx_hash,
None,
ReceiptPollConfig {
timeout: RECEIPT_TIMEOUT,
interval: RECEIPT_POLL_INTERVAL,
},
std::future::pending(),
)
.await
.map_err(|source| TestkitError::ReceiptWait {
source: Box::new(source),
})?;
self.inner.local_funding.lock().await.retain(|_, attempt| {
!attempt
.admitted
.as_ref()
.is_some_and(|r| r.transaction_hash == tx_hash)
&& !attempt
.evidence
.as_ref()
.is_some_and(|r| r.tx_hash == tx_hash)
});
Ok(receipt)
}
pub async fn transaction_status(&self, tx_hash: TxHash) -> Result<TransactionStatus> {
known(
self.require_client("getTransactionStatus")?
.get_transaction_status(TransactionQuery {
transaction_hash: tx_hash,
})
.await,
"getTransactionStatus",
)
}
pub async fn await_pending(&self, tx_hash: TxHash) -> Result<()> {
self.await_pending_inner(tx_hash).await
}
async fn await_pending_inner(&self, tx_hash: TxHash) -> Result<()> {
let mut last_status = TransactionStatus::Unknown {};
for _ in 0..PENDING_POLL_ATTEMPTS {
let status = self.transaction_status(tx_hash).await?;
match classify_pending_status(&status) {
PendingStatusClassification::Ready => return Ok(()),
PendingStatusClassification::Retry => {
last_status = status;
tokio::time::sleep(PENDING_POLL_INTERVAL).await;
}
PendingStatusClassification::Terminal => {
return Err(TestkitError::TransactionNotPending {
tx_hash: format!("{tx_hash:?}"),
status: Box::new(status),
});
}
}
}
Err(TestkitError::PendingTimeout {
tx_hash: format!("{tx_hash:?}"),
last_status: Box::new(last_status),
})
}
pub async fn produce_block(&self) -> Result<ProducedBlock> {
self.inner
.manual_block_producer
.as_ref()
.ok_or(TestkitError::ManualBlockProductionUnavailable)?
.produce_block()
.await
.map_err(|source| TestkitError::BlockProduction { source })
}
pub async fn get_storage_at(
&self,
address: ContractAddress,
key: &[u8],
) -> Result<StorageResult> {
known(
self.require_client("getStorage")?
.get_storage(requests::StorageQuery {
address: Base58(address),
key: HexBytes(key.to_vec()),
})
.await,
"getStorage",
)
}
pub async fn view_call(
&self,
contract: ContractAddress,
entrypoint: ContractEntrypointID,
calldata: &[u8],
caller: Option<ContractAddress>,
) -> Result<SimulationResult> {
let request = requests::SimulateInvocationRequest {
assumed_auth: None,
min_state_block_height: None,
address: Base58(contract),
entrypoint: HexBytes(entrypoint.into_inner().to_vec()),
calldata: HexBytes(calldata.to_vec()),
caller: caller.map(Base58),
gas_limit: None,
};
known(
self.require_client("simulateInvocation")?
.simulate_invocation(request)
.await,
"simulateInvocation",
)
}
pub async fn get_events(&self, query: EventQuery) -> Result<EventPage> {
known(
self.require_client("getEvents")?.get_events(query).await,
"getEvents",
)
}
pub(crate) async fn submit_fund(
&self,
recipient: ContractAddress,
amount: u128,
) -> Result<SubmittedTransaction> {
self.require_client("fund")?;
if let Some(faucet) = self.inner.local_faucet {
return self.submit_local_fund(recipient, amount, faucet).await;
}
let calldata = thruput_signing::native_token_transfer_calldata(&recipient, amount)
.map_err(TestkitError::signing)?;
let faucet = self.inner.faucet.lock().await;
let faucet = faucet.as_ref().ok_or(TestkitError::MissingFaucet)?;
let mut nonce = [0; 8];
getrandom::fill(&mut nonce).map_err(|error| {
TestkitError::signing(format!("generate faucet transaction nonce: {error}"))
})?;
let (tx_hash, tx_hex) = crate::account::prepare_call(
faucet,
thruput_xdr::genesis_addresses::ADDRESS_NATIVE_TOKEN,
thruput_signing::NATIVE_TOKEN_TRANSFER_ENTRYPOINT,
&calldata,
u64::from_le_bytes(nonce),
)?;
crate::account::submit_signed(self, tx_hash, &tx_hex).await?;
Ok(SubmittedTransaction::new(self.clone(), tx_hash, tx_hex))
}
async fn submit_local_fund(
&self,
recipient: ContractAddress,
amount: u128,
faucet: ContractAddress,
) -> Result<SubmittedTransaction> {
use thruput_json_rpc_types::{airdrop::ClientNonce, requests::FundAccountRequest};
if amount == 0 {
return Err(TestkitError::signing("funding amount must be positive"));
}
let mut pending = self.inner.local_funding.lock().await;
if let Some(request) = pending.get(&recipient) {
if request.request.raw_amount != amount {
return Err(TestkitError::FundingRequestMismatch { recipient });
}
} else {
let mut nonce = [0; 16];
getrandom::fill(&mut nonce).map_err(TestkitError::signing)?;
pending.insert(
recipient,
LocalFundingAttempt {
request: FundAccountRequest {
recipient: Base58(recipient),
raw_amount: amount,
faucet: Base58(faucet),
client_nonce: ClientNonce::new(nonce.to_vec())
.map_err(TestkitError::signing)?,
},
admitted: None,
evidence: None,
},
);
}
let attempt = pending
.get(&recipient)
.expect("bug: retained funding intent");
let request = attempt.request.clone();
let response = if let Some(response) = &attempt.admitted {
response.clone()
} else {
match self
.require_client("fund")?
.fund_account(request.clone())
.await
{
Ok(response) => known(Ok(response), "dev_fundAccount")?,
Err(error) => {
let attempt = pending
.get_mut(&recipient)
.expect("bug: retained funding intent");
if let Err(invalid) = attempt.retain_error_evidence(&error) {
return Err(TestkitError::InvalidFundingEvidence {
detail: invalid.to_string(),
retained: attempt.retained_evidence(),
source: Box::new(error),
});
}
return Err(match &attempt.evidence {
Some(evidence) => TestkitError::Submission {
tx_hash: evidence.tx_hash,
signed_transaction: thruput_json_rpc_types::hex::prefixed_hex(
&evidence.signed_transaction.0,
),
detail: error.to_string(),
source: Box::new(error),
},
None => TestkitError::NativeCall {
method: "dev_fundAccount",
source: Box::new(error),
},
});
}
}
};
pending
.get_mut(&recipient)
.expect("bug: retained funding intent")
.accept_response(&response)?;
Ok(SubmittedTransaction::new(
self.clone(),
response.transaction_hash,
thruput_json_rpc_types::hex::prefixed_hex(&response.signed_transaction.0),
))
}
pub(crate) fn require_client(
&self,
operation: &'static str,
) -> Result<&NativeClient<HttpClient>> {
self.inner
.client
.as_ref()
.ok_or(TestkitError::Offline { operation })
}
}
fn known<T>(
result: std::result::Result<DecodedResponse<T>, NativeClientError>,
method: &'static str,
) -> Result<T> {
match result {
Ok(DecodedResponse::Known(value)) => Ok(value),
Ok(DecodedResponse::Unsupported { raw }) => {
Err(TestkitError::UnsupportedResponse { method, raw })
}
Ok(_) => Err(TestkitError::rpc(
method,
"unsupported response representation",
)),
Err(source) => Err(TestkitError::NativeCall {
method,
source: Box::new(source),
}),
}
}
fn validate_funding_response(
response: &thruput_json_rpc_types::responses::FundingSubmission,
request: &requests::FundAccountRequest,
) -> Result<()> {
let signed: thruput_xdr::transaction::SignedTransaction =
xdrs::txs::decode(&response.signed_transaction.0).map_err(TestkitError::signing)?;
let hash = thruput_signing::transaction_hash(&signed.tx).map_err(TestkitError::signing)?;
if response.recipient != request.recipient
|| response.raw_amount != request.raw_amount
|| response.transaction_hash.0.0 != hash
{
return Err(TestkitError::signing(
"local funding response does not match retained request and signed hash",
));
}
Ok(())
}
fn validate_funding_evidence(evidence: &FundingEvidence) -> Result<()> {
let signed: thruput_xdr::transaction::SignedTransaction =
xdrs::txs::decode(&evidence.signed_transaction.0).map_err(TestkitError::signing)?;
let hash = thruput_signing::transaction_hash(&signed.tx).map_err(TestkitError::signing)?;
if evidence.tx_hash.0.0 != hash {
return Err(TestkitError::signing(
"local funding evidence hash does not match signed bytes",
));
}
Ok(())
}
fn local_faucet(manifest: &RuntimeManifest) -> Result<Option<ContractAddress>> {
use thruput_json_rpc_types::runtime_manifest::{
BOOTSTRAP_ACCOUNT_ROLE_FAUCET, HOGD_DEV_RUNTIME_MANIFEST_KIND,
};
if manifest.kind != HOGD_DEV_RUNTIME_MANIFEST_KIND {
return Ok(None);
}
let mut faucets = manifest
.bootstrap_accounts
.iter()
.filter(|a| a.role == BOOTSTRAP_ACCOUNT_ROLE_FAUCET);
let faucet = faucets.next().ok_or(TestkitError::MissingFaucet)?;
if faucets.next().is_some()
|| thruput_signing::PubkeyAccount::from_public_key_bytes(*faucet.public_key.0, 0).address()
!= faucet.wallet_address.0
{
return Err(TestkitError::Manifest {
field: "bootstrapAccounts",
detail: "local faucet must uniquely match its public nonce-zero account".into(),
});
}
Ok(Some(faucet.wallet_address.0))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PendingStatusClassification {
Ready,
Retry,
Terminal,
}
fn classify_pending_status(status: &TransactionStatus) -> PendingStatusClassification {
match status {
TransactionStatus::Pending { .. } => PendingStatusClassification::Ready,
TransactionStatus::Unknown {} => PendingStatusClassification::Retry,
_ => PendingStatusClassification::Terminal,
}
}
#[cfg(test)]
mod tests {
use thruput_json_rpc_types::responses::TransactionStatus;
use thruput_node_supervisor::BlockDriverMode;
use super::{
LocalNetworkConfig, Network, PendingStatusClassification, classify_pending_status,
};
use crate::TestkitError;
#[test]
fn local_network_keeps_instant_block_production_by_default() {
assert_eq!(
LocalNetworkConfig::default().block_driver.mode,
BlockDriverMode::Instant
);
}
#[test]
fn pending_status_classification_never_resubmits() {
use thruput_json_rpc_types::responses::{SchedulerError, SchedulerErrorSource};
assert_eq!(
classify_pending_status(&TransactionStatus::Pending {
recent_failure: None
}),
PendingStatusClassification::Ready
);
assert_eq!(
classify_pending_status(&TransactionStatus::Unknown {}),
PendingStatusClassification::Retry
);
for status in [
TransactionStatus::Committed {
block_height: 2.into(),
},
TransactionStatus::Retry {
recent_failure: None,
error: SchedulerError {
source: SchedulerErrorSource::Executor,
code: 7,
},
},
TransactionStatus::Dropped {
recent_failure: None,
error: SchedulerError {
source: SchedulerErrorSource::Executor,
code: 7,
},
},
] {
assert_eq!(
classify_pending_status(&status),
PendingStatusClassification::Terminal
);
}
}
#[tokio::test]
async fn offline_network_rejects_manual_block_production() {
let error = Network::offline()
.produce_block()
.await
.expect_err("offline network must not expose manual production");
assert!(matches!(
error,
TestkitError::ManualBlockProductionUnavailable
));
}
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum NetworkConfig {
Existing {
rpc_url: String,
runtime_manifest: PathBuf,
},
Local(LocalNetworkConfig),
Offline {
accounts: usize,
},
}
impl NetworkConfig {
pub fn existing(rpc_url: impl Into<String>, runtime_manifest: impl Into<PathBuf>) -> Self {
Self::Existing {
rpc_url: rpc_url.into(),
runtime_manifest: runtime_manifest.into(),
}
}
pub fn local(config: LocalNetworkConfig) -> Self {
Self::Local(config)
}
pub fn offline(accounts: usize) -> Self {
Self::Offline { accounts }
}
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct LocalNetworkConfig {
pub hog_devnode_binary: Option<PathBuf>,
pub contract_artifact_directories: Vec<PathBuf>,
pub rpc_port: u16,
pub hogd_protocol: Option<String>,
pub log_level: Option<LogLevel>,
pub initial_account_tokens: u64,
pub startup_timeout: Duration,
pub block_driver: BlockDriverConfig,
}
impl Default for LocalNetworkConfig {
fn default() -> Self {
Self {
hog_devnode_binary: None,
contract_artifact_directories: Vec::new(),
rpc_port: 0,
hogd_protocol: None,
log_level: None,
initial_account_tokens: thruput_node_supervisor::config::DEFAULT_LOCAL_ACCOUNT_TOKENS,
startup_timeout: Duration::from_secs(10),
block_driver: BlockDriverConfig::default(),
}
}
}