From 7d1272b16f4ab1787ab8a868dcbdd9d635351363 Mon Sep 17 00:00:00 2001 From: Aayush Giri Date: Mon, 10 Aug 2026 21:56:28 +0530 Subject: [PATCH] faucet: block until the send lands and return the real txid requestfaucetdonation queued the address and returned a fixed amount the instant it was called, before any transaction was built. Callers could not tell a paid request from a dropped one: the reply was success-shaped whether or not money moved, carried no txid to verify, and a failed build/broadcast was silently dropped from the queue with no retry. This makes it behave like wallet_staking_action, which already does the right thing a few hundred lines away: stage the request with a one-shot reply channel, and have the wallet loop answer only once the transaction actually reaches SENT, with the real txid, or on a definite failure, with an error. - FAUCET_STAGE mirrors STAKING_STAGE: (address, oneshot::Sender>). - The service handler blocks on the channel instead of calling a closure that returned immediately; a second concurrent request gets a busy error rather than silently queueing. - The wallet loop proposes the send from the miner wallet, drives it through build and broadcast, and replies with (amount, txid) on SENT or a real error on failure. A send that cannot select notes now reports that instead of reporting success. - FaucetResponse carries the txid; the zaino gRPC conversion reads .amount and is unaffected. Removes the now-unreachable FAUCET_REQUEST closure, FAUCET_Q queue, the idle-tick drainer, and the TEST_FAUCET fake-read, so there is one faucet path rather than two half-wired ones. The fixed amount and memo are unchanged. --- zebra-crosslink/wallet/src/lib.rs | 120 +++++++------------ zebra-crosslink/zebra-crosslink/src/lib.rs | 24 +++- zebra-crosslink/zebra-rpc/src/methods.rs | 3 +- zebra-crosslink/zebra-state/src/crosslink.rs | 2 +- 4 files changed, 67 insertions(+), 82 deletions(-) diff --git a/zebra-crosslink/wallet/src/lib.rs b/zebra-crosslink/wallet/src/lib.rs index 9405867a..2a8e8f7f 100644 --- a/zebra-crosslink/wallet/src/lib.rs +++ b/zebra-crosslink/wallet/src/lib.rs @@ -107,14 +107,15 @@ pub mod scanner; use zcash_protocol::consensus::{NetworkType, Parameters, MAIN_NETWORK, TEST_NETWORK}; -#[derive(Clone)] -pub struct FaucetRequestClosure(pub Arc Result + Sync + Send + 'static>); -pub static FAUCET_REQUEST: Mutex> = Mutex::new(None); pub static USER_UFVK_STRING: Mutex> = Mutex::new(None); pub static GUI_ENABLE_MINE: Mutex = Mutex::new(true); pub static STAKING_STAGE: Mutex>)>> = Mutex::new(None); +// Faucet RPC requests, staged for the wallet loop the same way staking is: an address +// plus a reply channel the loop fills with the real (amount, txid) once the send lands. +pub static FAUCET_STAGE: Mutex>)>> = Mutex::new(None); + #[derive(Clone)] pub struct RecencyRequestClosure(pub Arc Option + Sync + Send + 'static>); pub static RECENCY_REQUEST: Mutex> = Mutex::new(None); @@ -2711,23 +2712,6 @@ fn shard_tree_root(tree: &OrchardShardTree) -> orchard::tree::MerkleHashOrchard .unwrap() } -const FAUCET_Q_LEN: usize = 16; -struct FaucetQ { - pub read_o: u8, - pub write_o: u8, - pub data: [Option; FAUCET_Q_LEN], -} -impl FaucetQ { - fn len(&self) -> usize { - self.write_o.wrapping_sub(self.read_o).into() - } -} -// TODO: atomic -static FAUCET_Q: Mutex = Mutex::new(FaucetQ { - read_o: 0, - write_o: 0, - data: [None; FAUCET_Q_LEN], -}); const TEST_FAUCET: bool = false; const FAUCET_VALUE: u64 = 50_000_000; @@ -3183,41 +3167,6 @@ pub async fn wallet_main(wallet_state: Arc>) { let network = &TEST_NETWORK; - *FAUCET_REQUEST.lock().unwrap() = Some(FaucetRequestClosure(Arc::new(|ua_str: String| -> Result { - // let ua = zcash_address::unified::Address::decode(&ua_str).map_err(|err| - // Err(format!("invalid address: \"{ua_str}\" failed: {err}")) - // )?.1; - let ua = match zcash_keys::address::Address::decode(network, &ua_str) { - Some(zcash_keys::address::Address::Unified(ua)) => ua, - Some(_) => return Err(format!("must be an orchard-containing UA")), - None => return Err(format!("couldn't decode address")), - }; - let Some(orchard_addr) = ua.orchard() else { - return Err(format!("must contain an orchard receiver")); - }; - - let mut q = FAUCET_Q.lock().unwrap(); - if q.len() == q.data.len() { - return Err(format!("faucet too busy, come back later")); - } - - for idx in 0..q.len() { - let i = (q.read_o as usize + idx) % q.data.len(); - if let Some(existing_addr) = &q.data[i] { - if orchard_addr == existing_addr { - return Err(format!("the last request for this address is still pending, come back later")); - } - } else { - println!("Faucet Q error: got None result where there should be valid data"); - } - } - - let i = q.write_o as usize % q.data.len(); - q.write_o += 1; - q.data[i] = Some(orchard_addr.clone()); - - Ok(FAUCET_VALUE) - }))); let ( @@ -3333,7 +3282,6 @@ pub async fn wallet_main(wallet_state: Arc>) { println!("faucet: {:?}", client.request_faucet_donation(FaucetRequest{ address: user_ua_str.clone() }).await); println!("faucet: {:?}", client.request_faucet_donation(FaucetRequest{ address: "arosienarsoienaroisetn".to_owned() }).await); println!("faucet: {:?}", client.request_faucet_donation(FaucetRequest{ address: user_ua_str.clone() }).await); - FAUCET_Q.lock().unwrap().read_o += 1; // fake read println!("faucet: {:?}", client.request_faucet_donation(FaucetRequest{ address: user_ua_str.clone() }).await); } @@ -3425,6 +3373,7 @@ pub async fn wallet_main(wallet_state: Arc>) { let mut proposed_send = ProposedTx::EMPTY; let mut rpc_stake: Option<(ProposedTx, tokio::sync::oneshot::Sender>)> = None; + let mut rpc_faucet: Option<(ProposedTx, tokio::sync::oneshot::Sender>)> = None; let mut wallet_state_push_time = Instant::now(); @@ -4435,6 +4384,33 @@ pub async fn wallet_main(wallet_state: Arc>) { } } + // Faucet: take a staged RPC request, decode the recipient, and PROPOSE the send + // from the miner wallet. The reply channel travels with it to the drive step + // below, which answers only once the tx actually reaches SENT. + if rpc_faucet.is_none() { + let staged = FAUCET_STAGE.lock().unwrap().take(); + if let Some((ua_str, sender)) = staged { + let orchard_addr = match zcash_keys::address::Address::decode(network, &ua_str) { + Some(zcash_keys::address::Address::Unified(ua)) => ua.orchard().cloned(), + _ => None, + }; + match orchard_addr { + None => { let _ = sender.send(Err(format!("address must be an orchard-containing unified address: \"{ua_str}\""))); } + Some(orchard_addr) => { + let mut tx = ProposedTx::EMPTY; + let memo = MemoBytes::from_bytes("With love from your favourite faucet... Don't spend it all at once!".as_bytes()).unwrap(); + let ok = miner_wallet.send_orchard_to_orchard_zats(network, &mut tx, &mut client, &miner_usk, FAUCET_VALUE, &orchard_tree, orchard_addr, memo).is_some(); + if ok { + just_init_new_tx = true; + rpc_faucet = Some((tx, sender)); + } else { + let _ = sender.send(Err("faucet has no spendable orchard notes right now".to_string())); + } + } + } + } + } + // @todo(judah): I'm thinking the weird frame hitch we get in the UI is caused by this loop, // since it's probably waiting for the wallet_state mutex to unlock. let mut retries_this_round = 6; @@ -4456,25 +4432,6 @@ pub async fn wallet_main(wallet_state: Arc>) { if DUMP_ACTIONS { println!("*** wallet has {:?} actions in flight", wallet_state.actions_in_flight.len()); } let Some(action) = wallet_state.actions_in_flight.front() else { - // if we're not doing anything else, process a faucet RPC request - if ! proposed_faucet.is_in_progress() { - let mut q = FAUCET_Q.lock().unwrap(); - if DUMP_FAUCET { println!("faucet Q read_o: {}, write_o: {}", q.read_o, q.write_o); } - if q.len() > 0 { - let i = q.read_o as usize % q.data.len(); - if DUMP_FAUCET { println!("faucet Q new element at {i}: {:?}", q.data[i]); } - if let Some(orchard_addr) = q.data[i] { - let memo = MemoBytes::from_bytes("With love from your favourite faucet... Don't spend it all at once!".as_bytes()).unwrap(); - let ok = miner_wallet.send_orchard_to_orchard_zats(network, &mut proposed_faucet, &mut client, &miner_usk, FAUCET_VALUE, &orchard_tree, orchard_addr, memo).is_some(); - proposed_faucet.is_user_faucet = false; - if DUMP_ACTIONS { println!("Try RPC faucet send: {ok:?}"); } - just_init_new_tx |= ok; - q.read_o += 1; - } else { - println!("Faucet Q error: got None result where there should be valid data"); - } - } - } break; }; @@ -4645,6 +4602,19 @@ pub async fn wallet_main(wallet_state: Arc>) { } } } + + // Faucet: drive the proposed send through build -> broadcast. Answer the RPC + // caller ONLY on SENT (with the real txid) or a definite failure; otherwise + // keep it in flight for the next tick. This is what makes a cTAZ "sent" mean + // the same thing a TAZ "sent" does. + if let Some((mut tx, sender)) = rpc_faucet.take() { + let tx_progress = continue_proposed_tx(&mut miner_wallet, network, &mut tx, &mut client, "rpc faucet", DUMP_TX_SEND).await; + match tx_progress.h { + BlockHeight::INVALID => { let _ = sender.send(Err("faucet transaction failed to build".to_string())); } + BlockHeight::SENT => { let _ = sender.send(Ok((FAUCET_VALUE, format!("{}", tx_progress.txid)))); } + _ => { rpc_faucet = Some((tx, sender)); } + } + } } } } diff --git a/zebra-crosslink/zebra-crosslink/src/lib.rs b/zebra-crosslink/zebra-crosslink/src/lib.rs index 6fe95a10..1f08ecd3 100644 --- a/zebra-crosslink/zebra-crosslink/src/lib.rs +++ b/zebra-crosslink/zebra-crosslink/src/lib.rs @@ -1843,12 +1843,26 @@ async fn tfl_service_incoming_request( // wallet TFLServiceRequest::Faucet(request) => { + // Stage the request and BLOCK until the wallet loop has actually built and + // broadcast the transaction, mirroring WalletStakingAction. The old path + // returned a fixed amount the instant the address was queued, before any + // send, so a caller could not tell a paid request from a dropped one. Now the + // reply carries the real (amount, txid) or a real error. Ok(TFLServiceResponse::Faucet({ - let closure = wallet::FAUCET_REQUEST.lock().unwrap(); - if let Some(closure) = closure.as_ref() { - (closure.0)(request) - } else { - Err("No faucet available".to_owned()) + let rx = { + let mut lock = wallet::FAUCET_STAGE.lock().unwrap(); + match *lock { + None => { + let (tx, rx) = tokio::sync::oneshot::channel(); + *lock = Some((request, tx)); + rx + } + Some(_) => return Err(zebra_state::crosslink::TFLServiceError::Misc("faucet is busy with another request, come back soon".to_string())), + } + }; + match rx.await { + Ok(result) => result, + Err(err) => return Err(zebra_state::crosslink::TFLServiceError::Misc(format!("{err}"))), } })) } diff --git a/zebra-crosslink/zebra-rpc/src/methods.rs b/zebra-crosslink/zebra-rpc/src/methods.rs index 6200c1d4..c9985422 100644 --- a/zebra-crosslink/zebra-rpc/src/methods.rs +++ b/zebra-crosslink/zebra-rpc/src/methods.rs @@ -1585,7 +1585,7 @@ where .await; match ret { - Ok(TFLServiceResponse::Faucet(Ok(amount))) => Ok(FaucetResponse{ amount }), + Ok(TFLServiceResponse::Faucet(Ok((amount, txid)))) => Ok(FaucetResponse{ amount, txid }), Ok(TFLServiceResponse::Faucet(Err(err))) => Err(ErrorObject::owned( server::error::LegacyCode::Verify.into(), format!("Faucet request for \"{ua_str}\" failed: {err}"), @@ -4261,6 +4261,7 @@ pub struct FaucetRequest { )] pub struct FaucetResponse { pub amount: u64, + pub txid: String, } /// A hex-encoded [`ConsensusBranchId`] string. diff --git a/zebra-crosslink/zebra-state/src/crosslink.rs b/zebra-crosslink/zebra-state/src/crosslink.rs index 251e4a36..cb3919d3 100644 --- a/zebra-crosslink/zebra-state/src/crosslink.rs +++ b/zebra-crosslink/zebra-state/src/crosslink.rs @@ -89,7 +89,7 @@ pub enum TFLServiceResponse { /// Send a staking command transaction StakingCmd, /// Faucet - Faucet(Result), + Faucet(Result<(u64, String), String>), /// Response to [`ReadRequest::TotalIssuanceFromKey`] TotalIssuanceFromKey(Result, String>), /// Finalizer recency status + reference UTC