diff --git a/Cargo.lock b/Cargo.lock index 59aeb5d0..26a717a7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1726,6 +1726,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "fs_extra" version = "1.3.0" @@ -3708,9 +3718,11 @@ dependencies = [ "dircmp", "ed25519-dalek", "file-format", + "fs2", "hex", "hkdf 0.12.4", "mockito", + "nonasync", "once_cell", "p256", "rand 0.10.1", diff --git a/Cargo.toml b/Cargo.toml index e5d41639..f794aab5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,6 +36,7 @@ chacha20poly1305 = { version = "0.11.0", default-features = false, features = [ ] } tokio = { version = "1", default-features = false, features = ["rt"] } hex = { version = "0.4.3", default-features = false } +nonasync = { version = "0.1.3", default-features = false } rand = { version = "0.10.1", default-features = false, features = [ "thread_rng", ] } @@ -71,6 +72,7 @@ time = { version = "0.3.52", default-features = false } file-format = { version = "0.29.0", default-features = false, features = [ "reader", ] } +fs2 = { version = "0.4.3", default-features = false } url = { version = "2.5", default-features = false } walkdir = { version = "2.5.0", default-features = false } zip = { version = "8.6.0", default-features = false, features = [ diff --git a/bindings/c-ffi/Cargo.lock b/bindings/c-ffi/Cargo.lock index 072d6ad0..63dd3920 100644 --- a/bindings/c-ffi/Cargo.lock +++ b/bindings/c-ffi/Cargo.lock @@ -1461,6 +1461,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "fs_extra" version = "1.3.0" @@ -3290,8 +3300,10 @@ dependencies = [ "bdk_wallet", "chacha20poly1305", "file-format", + "fs2", "hex", "hkdf 0.12.4", + "nonasync", "rand 0.10.1", "reqwest", "rgb-invoicing", diff --git a/bindings/uniffi/Cargo.lock b/bindings/uniffi/Cargo.lock index 32f21cb4..5038316e 100644 --- a/bindings/uniffi/Cargo.lock +++ b/bindings/uniffi/Cargo.lock @@ -1577,6 +1577,16 @@ dependencies = [ "autocfg", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "fs_extra" version = "1.3.0" @@ -3458,8 +3468,10 @@ dependencies = [ "bdk_wallet", "chacha20poly1305", "file-format", + "fs2", "hex", "hkdf 0.12.4", + "nonasync", "rand 0.10.1", "reqwest", "rgb-invoicing", diff --git a/bindings/uniffi/src/rgb-lib.udl b/bindings/uniffi/src/rgb-lib.udl index 69d747f2..c3db7c9f 100644 --- a/bindings/uniffi/src/rgb-lib.udl +++ b/bindings/uniffi/src/rgb-lib.udl @@ -59,6 +59,7 @@ interface RgbLibError { InsufficientAssignments(string asset_id, AssignmentsCollection available); InsufficientBitcoins(u64 needed, u64 available); Internal(string details); + RgbOperationInProgress(string operation_id); InvalidAddress(string details); InvalidAmountZero(); InvalidAssignment(); diff --git a/src/error.rs b/src/error.rs index cb94df26..e7f02e7e 100644 --- a/src/error.rs +++ b/src/error.rs @@ -200,6 +200,13 @@ pub enum Error { details: String, }, + /// An RGB stock transition is waiting for its surrounding protocol to commit or roll back + #[error("RGB operation {operation_id} is in progress")] + RgbOperationInProgress { + /// Stable identifier of the operation that owns the RGB stock + operation_id: String, + }, + /// An invalid bitcoin address has been provided #[error("Address error: {details}")] InvalidAddress { diff --git a/src/utils.rs b/src/utils.rs index 5a0fe733..9cc53e89 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -3,6 +3,9 @@ //! This module defines some utility methods and structures. use super::*; +use fs2::FileExt; +#[cfg(any(feature = "electrum", feature = "esplora"))] +use nonasync::persistence::CloneNoPersistence; #[cfg(any(feature = "electrum", feature = "esplora"))] use rgbstd::contract::LinkableIssuerWrapper; @@ -32,7 +35,6 @@ pub(crate) const INDEXER_RETRIES: u8 = 3; pub(crate) const INDEXER_BATCH_SIZE: usize = 5; #[cfg(feature = "esplora")] pub(crate) const INDEXER_PARALLEL_REQUESTS: usize = 5; - #[cfg(any(feature = "electrum", feature = "esplora"))] const PROXY_PROTOCOL_VERSION: &str = "0.2"; @@ -749,15 +751,21 @@ impl ResolveWitness for DumbResolver { } } +#[derive(Debug)] +pub(crate) struct RgbRuntimeLock { + _file: std::fs::File, +} + /// Wrapper for the RGB stock and its lockfile. #[doc(hidden)] #[derive(Debug)] pub struct RgbRuntime { /// The RGB stock stock: Stock, - /// The wallet directory, where the lockfile for the runtime is to be held - wallet_dir: PathBuf, - /// Whether dropping the runtime should persist the in-memory stock. + /// Process-scoped ownership of the RGB stock. The operating system releases this lock on exit, + /// including an unclean exit, so a stale path cannot permanently block wallet recovery. + _lock: RgbRuntimeLock, + /// Whether dropping this runtime should persist its stock to the live provider. persist_on_drop: bool, } @@ -787,6 +795,10 @@ impl RgbRuntime { .map_err(InternalError::from) } + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub(crate) fn lock(&self) -> &RgbRuntimeLock { + &self._lock + } #[cfg(any(feature = "electrum", feature = "esplora"))] pub(crate) fn accept_transfer( &mut self, @@ -833,6 +845,29 @@ impl RgbRuntime { .collect()) } + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub(crate) fn contains_transfer_witness( + &self, + contract_id: ContractId, + witness_id: RgbTxid, + ) -> Result { + if !self + .stock + .contracts() + .map_err(InternalError::from)? + .any(|contract| contract.id == contract_id) + { + return Ok(false); + } + + Ok(self + .stock + .contract_data(contract_id) + .map_err(InternalError::from)? + .witness_info(witness_id) + .is_some()) + } + pub(crate) fn contract_wrapper( &self, contract_id: ContractId, @@ -920,6 +955,52 @@ impl RgbRuntime { .map_err(InternalError::from) } + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub(crate) fn stage_transfer( + &self, + seal: GraphSeal, + transfer: ValidTransfer, + resolver: &R, + ) -> Result { + let mut stock = self.stock.clone_no_persistence(); + stock.store_secret_seal(seal)?; + stock.import_contract(transfer.clone().into_valid_contract(), resolver)?; + stock.accept_transfer(transfer, resolver)?; + Ok(stock) + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub(crate) fn stage_fascia( + &self, + fascia: Fascia, + witness_ord: Option, + ) -> Result { + struct FasciaResolver { + witness_id: RgbTxid, + witness_ord: WitnessOrd, + } + + impl WitnessOrdProvider for FasciaResolver { + fn witness_ord(&self, witness_id: RgbTxid) -> Result { + debug_assert_eq!(witness_id, self.witness_id); + Ok(self.witness_ord) + } + } + + let resolver = FasciaResolver { + witness_id: fascia.witness_id(), + witness_ord: witness_ord.unwrap_or(WitnessOrd::Tentative), + }; + let mut stock = self.stock.clone_no_persistence(); + stock.consume_fascia(fascia, resolver)?; + Ok(stock) + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub(crate) fn suppress_persistence(&mut self) { + self.persist_on_drop = false; + } + pub(crate) fn transfer( &self, contract_id: ContractId, @@ -1017,22 +1098,22 @@ impl Drop for RgbRuntime { if self.persist_on_drop { self.stock.store().expect("unable to save stock"); } - fs::remove_file(self.wallet_dir.join(RGB_RUNTIME_LOCK_FILE)) - .expect("should be able to drop lockfile") } } -fn write_rgb_runtime_lockfile(wallet_dir: &Path) -> Result<(), Error> { +pub(crate) fn acquire_rgb_runtime_lock(wallet_dir: &Path) -> Result { let lock_file_path = wallet_dir.join(RGB_RUNTIME_LOCK_FILE); let t_0 = OffsetDateTime::now_utc(); + let file = fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(lock_file_path)?; loop { - match fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(lock_file_path.clone()) - { - Ok(_) => return Ok(()), - Err(e) if e.kind() == io::ErrorKind::AlreadyExists => { + match file.try_lock_exclusive() { + Ok(()) => return Ok(RgbRuntimeLock { _file: file }), + Err(e) if e.kind() == io::ErrorKind::WouldBlock => { if (OffsetDateTime::now_utc() - t_0).as_seconds_f32() > LOCK_FILE_TIMEOUT_SECS { return Err(Error::Internal { details: s!("unreleased lock file"), @@ -1050,8 +1131,12 @@ fn write_rgb_runtime_lockfile(wallet_dir: &Path) -> Result<(), Error> { } } -pub(crate) fn load_rgb_runtime>(wallet_dir: P) -> Result { - write_rgb_runtime_lockfile(wallet_dir.as_ref())?; +fn load_rgb_runtime_with_operation>( + wallet_dir: P, + operation_id: Option<&str>, +) -> Result { + let lock = acquire_rgb_runtime_lock(wallet_dir.as_ref())?; + crate::wallet::rust_only::validate_rgb_runtime_access(wallet_dir.as_ref(), operation_id)?; let rgb_dir = wallet_dir.as_ref().join(RGB_RUNTIME_DIR); if !rgb_dir.exists() { @@ -1074,11 +1159,23 @@ pub(crate) fn load_rgb_runtime>(wallet_dir: P) -> Result>(wallet_dir: P) -> Result { + load_rgb_runtime_with_operation(wallet_dir, None) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +pub(crate) fn load_rgb_runtime_for_operation>( + wallet_dir: P, + operation_id: &str, +) -> Result { + load_rgb_runtime_with_operation(wallet_dir, Some(operation_id)) +} + #[cfg(any(feature = "electrum", feature = "esplora"))] pub(crate) struct OffchainResolver<'a, 'cons, const TRANSFER: bool> { pub(crate) witness_id: RgbTxid, @@ -1109,6 +1206,8 @@ impl ResolveWitness for OffchainResolver<'_, '_, TRANSFER> #[cfg(test)] mod tests { use super::*; + use std::io::Write as _; + use std::process::{Command, Stdio}; #[derive(Debug, Deserialize)] struct MandatoryField { @@ -1240,16 +1339,74 @@ mod tests { } #[test] - fn test_write_rgb_runtime_lockfile_timeout() { + fn test_rgb_runtime_lock_allows_preexisting_file() { let dir = tempfile::tempdir().unwrap(); let lock_path = dir.path().join(RGB_RUNTIME_LOCK_FILE); - // pre-create the lock file so every open attempt sees AlreadyExists fs::File::create(&lock_path).unwrap(); - // with a lower LOCK_FILE_TIMEOUT_SECS in test builds the error is returned immediately - let result = write_rgb_runtime_lockfile(dir.path()); + let lock = acquire_rgb_runtime_lock(dir.path()).unwrap(); + drop(lock); + acquire_rgb_runtime_lock(dir.path()).unwrap(); + } + + #[test] + fn test_rgb_runtime_lock_times_out_while_held() { + let dir = tempfile::tempdir().unwrap(); + let _lock = acquire_rgb_runtime_lock(dir.path()).unwrap(); + let result = acquire_rgb_runtime_lock(dir.path()); assert_matches!(result, Err(Error::Internal { details }) if details == "unreleased lock file"); } + #[test] + #[ignore = "subprocess used by test_rgb_runtime_lock_released_after_process_kill"] + fn rgb_runtime_lock_crash_child() { + let wallet_dir = PathBuf::from(std::env::var("RGB_LOCK_CHILD_WALLET_DIR").unwrap()); + let ready_path = PathBuf::from(std::env::var("RGB_LOCK_CHILD_READY_PATH").unwrap()); + let _runtime = load_rgb_runtime(wallet_dir).unwrap(); + let mut ready = fs::File::create(ready_path).unwrap(); + ready.write_all(b"ready").unwrap(); + ready.sync_all().unwrap(); + loop { + std::thread::park(); + } + } + + #[test] + fn test_rgb_runtime_lock_released_after_process_kill() { + let directory = tempfile::tempdir().unwrap(); + let ready_path = directory.path().join("ready"); + let mut child = Command::new(std::env::current_exe().unwrap()) + .arg("--ignored") + .arg("--exact") + .arg("utils::tests::rgb_runtime_lock_crash_child") + .env("RGB_LOCK_CHILD_WALLET_DIR", directory.path()) + .env("RGB_LOCK_CHILD_READY_PATH", &ready_path) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20); + loop { + if ready_path.exists() { + break; + } + if let Some(status) = child.try_wait().unwrap() { + panic!("RGB runtime lock child exited early with status {status}"); + } + assert!( + std::time::Instant::now() < deadline, + "RGB runtime lock child did not acquire the lock" + ); + std::thread::sleep(std::time::Duration::from_millis(10)); + } + child.kill().unwrap(); + let status = child.wait().unwrap(); + assert!(!status.success()); + + let runtime = load_rgb_runtime(directory.path()).unwrap(); + drop(runtime); + assert!(directory.path().join(RGB_RUNTIME_LOCK_FILE).exists()); + } + // The None return from build_indexer is only reachable when electrum is enabled but esplora // is not, and the URL is not a valid electrum server. With esplora enabled the builder is // infallible so it would always return Some(Indexer::Esplora) instead. diff --git a/src/wallet/core.rs b/src/wallet/core.rs index 3f6f818d..bf21c305 100644 --- a/src/wallet/core.rs +++ b/src/wallet/core.rs @@ -243,6 +243,12 @@ pub(crate) fn setup_rgb>( if bitcoin_network == BitcoinNetwork::Mainnet && supported_schemas.contains(&AssetSchema::Ifa) { return Err(Error::CannotUseIfaOnMainnet); } + // A crash after stock promotion must not make the wallet impossible to reopen. Recovery only + // needs the wallet database and journal; opening or mutating RGB stock remains forbidden until + // the caller explicitly finalizes or rolls back the pending acceptance. + if crate::wallet::rust_only::pending_rgb_acceptance_operation(wallet_dir.as_ref())?.is_some() { + return Ok(()); + } let mut runtime = load_rgb_runtime(wallet_dir)?; let known_schemas = runtime.schemata()?; if known_schemas.len() < NUM_KNOWN_SCHEMAS { diff --git a/src/wallet/mpc.rs b/src/wallet/mpc.rs index 52843ed8..b1a2c91f 100644 --- a/src/wallet/mpc.rs +++ b/src/wallet/mpc.rs @@ -970,7 +970,7 @@ impl MpcWallet { None, )?; begin_op_data.psbt = self.mpc_sign_psbt(begin_op_data.psbt)?; - let res = self.send_end_impl(&txn, &begin_op_data.psbt)?; + let res = self.send_end_impl(&txn, &begin_op_data.psbt, true, None)?; self.update_backup_info(&txn, false)?; txn.commit()?; self.trigger_auto_backup(); @@ -1036,7 +1036,7 @@ impl MpcWallet { self.check_online(online)?; let psbt = Psbt::from_str(&signed_psbt)?; let txn = self.database().begin_transaction()?; - let res = self.send_end_impl(&txn, &psbt)?; + let res = self.send_end_impl(&txn, &psbt, true, None)?; self.update_backup_info(&txn, false)?; txn.commit()?; self.trigger_auto_backup(); diff --git a/src/wallet/multisig.rs b/src/wallet/multisig.rs index 1dd75b8a..b4350560 100644 --- a/src/wallet/multisig.rs +++ b/src/wallet/multisig.rs @@ -803,7 +803,7 @@ impl OperationHandler for SendRgbHandler { wallet: &mut MultisigWallet, combined_psbt: &Psbt, ) -> Result { - let res = wallet.send_end_impl(txn, combined_psbt)?; + let res = wallet.send_end_impl(txn, combined_psbt, true, None)?; Ok(res.txid) } diff --git a/src/wallet/offline.rs b/src/wallet/offline.rs index 74335c9c..ad02a8cf 100644 --- a/src/wallet/offline.rs +++ b/src/wallet/offline.rs @@ -1632,6 +1632,15 @@ pub trait WalletOffline: WalletBackup { valid_contract: ValidContract, valid_transfer: Option, ) -> Result { + #[cfg(any(feature = "electrum", feature = "esplora"))] + if let Some(valid_transfer) = valid_transfer.as_ref() { + return self.extract_asset_data_from_valid_transfer( + contract_id, + asset_schema, + &valid_contract, + valid_transfer, + ); + } let timestamp = valid_contract.genesis.timestamp; let added_at = now().unix_timestamp(); let media_dir = self.media_dir(); @@ -1763,6 +1772,115 @@ pub trait WalletOffline: WalletBackup { }) } + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn extract_asset_data_from_valid_transfer( + &self, + contract_id: ContractId, + asset_schema: AssetSchema, + valid_contract: &ValidContract, + valid_transfer: &ValidTransfer, + ) -> Result { + let timestamp = valid_transfer.genesis.timestamp; + let added_at = now().unix_timestamp(); + let media_dir = self.media_dir(); + Ok(match asset_schema { + AssetSchema::Nia => { + let contract = NiaWrapper::with(valid_contract.contract_data()); + let spec = contract.spec(); + LocalAssetData { + asset_id: contract_id.to_string(), + name: spec.name().to_string(), + asset_schema, + precision: spec.precision.into(), + ticker: Some(spec.ticker().to_string()), + details: spec.details().map(|details| details.to_string()), + media: contract + .contract_terms() + .media + .map(|attachment| Media::from_attachment(&attachment, media_dir)), + initial_supply: contract.total_issued_supply().into(), + max_supply: None, + known_circulating_supply: None, + reject_list_url: None, + token: None, + timestamp, + added_at, + } + } + AssetSchema::Uda => { + let contract = UdaWrapper::with(valid_contract.contract_data()); + let spec = contract.spec(); + LocalAssetData { + asset_id: contract_id.to_string(), + name: spec.name().to_string(), + asset_schema, + precision: spec.precision.into(), + ticker: Some(spec.ticker().to_string()), + details: spec.details().map(|details| details.to_string()), + media: contract + .contract_terms() + .media + .map(|attachment| Media::from_attachment(&attachment, media_dir)), + initial_supply: 1, + max_supply: None, + known_circulating_supply: None, + reject_list_url: None, + token: Some(Token::from_token_data( + &contract.token_data(), + self.media_dir(), + )), + timestamp, + added_at, + } + } + AssetSchema::Cfa => { + let contract = CfaWrapper::with(valid_contract.contract_data()); + LocalAssetData { + asset_id: contract_id.to_string(), + name: contract.name().to_string(), + asset_schema, + precision: contract.precision().into(), + ticker: None, + details: contract.details().map(|details| details.to_string()), + media: contract + .contract_terms() + .media + .map(|attachment| Media::from_attachment(&attachment, media_dir)), + initial_supply: contract.total_issued_supply().into(), + max_supply: None, + known_circulating_supply: None, + reject_list_url: None, + token: None, + timestamp, + added_at, + } + } + AssetSchema::Ifa => { + let contract = IfaWrapper::with(valid_contract.contract_data()); + let transfer_contract = IfaWrapper::with(valid_transfer.contract_data()); + LocalAssetData { + asset_id: contract_id.to_string(), + name: contract.spec().name().to_string(), + asset_schema, + precision: contract.spec().precision.into(), + ticker: Some(contract.spec().ticker().to_string()), + details: contract.spec().details().map(|details| details.to_string()), + media: contract + .contract_terms() + .media + .map(|attachment| Media::from_attachment(&attachment, media_dir)), + initial_supply: contract.total_issued_supply().into(), + max_supply: Some(contract.max_supply().into()), + known_circulating_supply: Some(transfer_contract.total_issued_supply().into()), + reject_list_url: contract.reject_list_url().map(|url| url.to_string()), + token: None, + timestamp, + added_at, + } + } + }) + } + fn save_new_asset_internal( &self, txn: &DbTxn, @@ -2788,6 +2906,16 @@ pub trait WalletOffline: WalletBackup { transfer_dir: &PathBuf, ) -> Result<(), Error> { let runtime = self.rgb_runtime()?; + self.gen_consignments_with_runtime(&runtime, fascia, transfer_info_map, transfer_dir) + } + + fn gen_consignments_with_runtime( + &self, + runtime: &RgbRuntime, + fascia: &Fascia, + transfer_info_map: &BTreeMap, + transfer_dir: &PathBuf, + ) -> Result<(), Error> { for (asset_id, transfer_info) in transfer_info_map { let consignment = runtime.transfer_from_fascia( transfer_info.asset_info.contract_id, diff --git a/src/wallet/online.rs b/src/wallet/online.rs index 496b1c47..ba57c40f 100644 --- a/src/wallet/online.rs +++ b/src/wallet/online.rs @@ -197,9 +197,12 @@ pub trait WalletOnline: WalletOffline { runtime: &mut RgbRuntime, signed_psbt: &Psbt, fascia: Fascia, + consume_fascia: bool, ) -> Result { let tx = self.broadcast_psbt(txn, signed_psbt)?; - runtime.consume_fascia(fascia, None)?; + if consume_fascia { + runtime.consume_fascia(fascia, None)?; + } Ok(tx) } @@ -1955,7 +1958,7 @@ pub trait WalletOnline: WalletOffline { let fascia_path = transfer_dir.join(FASCIA_FILE); let fascia_str = fs::read_to_string(fascia_path)?; let fascia: Fascia = serde_json::from_str(&fascia_str).map_err(InternalError::from)?; - self.broadcast_and_update_rgb(txn, &mut runtime, &signed_psbt, fascia)?; + self.broadcast_and_update_rgb(txn, &mut runtime, &signed_psbt, fascia, true)?; let mut updated_batch_transfer: DbBatchTransferActMod = batch_transfer.clone().into(); updated_batch_transfer.status = ActiveValue::Set(TransferStatus::WaitingConfirmations); @@ -3719,9 +3722,36 @@ pub trait WalletOnline: WalletOffline { status: TransferStatus, fascia: Fascia, sync_tte_used: bool, + consume_fascia: bool, ) -> Result { let mut runtime = self.rgb_runtime()?; - self.broadcast_and_update_rgb(txn, &mut runtime, psbt, fascia)?; + self.finalize_transfer_end_with_runtime( + txn, + txid, + psbt, + info_contents, + status, + fascia, + sync_tte_used, + consume_fascia, + &mut runtime, + ) + } + + #[allow(clippy::too_many_arguments)] + fn finalize_transfer_end_with_runtime( + &mut self, + txn: &DbTxn, + txid: String, + psbt: &Psbt, + info_contents: &InfoBatchTransfer, + status: TransferStatus, + fascia: Fascia, + sync_tte_used: bool, + consume_fascia: bool, + runtime: &mut RgbRuntime, + ) -> Result { + self.broadcast_and_update_rgb(txn, runtime, psbt, fascia, consume_fascia)?; self.update_or_save_transfers(txn, txid, info_contents, status, sync_tte_used) } @@ -3942,7 +3972,13 @@ pub trait WalletOnline: WalletOffline { }) } - fn send_end_impl(&mut self, txn: &DbTxn, signed_psbt: &Psbt) -> Result { + fn send_end_impl( + &mut self, + txn: &DbTxn, + signed_psbt: &Psbt, + consume_fascia: bool, + operation_id: Option<&str>, + ) -> Result { let (txid, transfer_dir, mut info_contents, mut fascia) = self.get_transfer_end_data(signed_psbt)?; @@ -3956,7 +3992,21 @@ pub trait WalletOnline: WalletOffline { fascia.update_pub_witness(PubWitness::with(tx)); } - self.gen_consignments(&fascia, &info_contents.transfers, &transfer_dir)?; + let mut operation_runtime = operation_id + .map(|operation_id| { + crate::utils::load_rgb_runtime_for_operation(self.wallet_dir(), operation_id) + }) + .transpose()?; + if let Some(runtime) = operation_runtime.as_ref() { + self.gen_consignments_with_runtime( + runtime, + &fascia, + &info_contents.transfers, + &transfer_dir, + )?; + } else { + self.gen_consignments(&fascia, &info_contents.transfers, &transfer_dir)?; + } let psbt_out = transfer_dir.join(SIGNED_PSBT_FILE); fs::write(psbt_out, signed_psbt.to_string())?; @@ -3995,15 +4045,30 @@ pub trait WalletOnline: WalletOffline { let sync_tte_used = true; let batch_transfer_idx = if info_contents.donation { - self.finalize_transfer_end( - txn, - txid.clone(), - signed_psbt, - &info_contents, - TransferStatus::WaitingConfirmations, - fascia, - sync_tte_used, - )? + if let Some(runtime) = operation_runtime.as_mut() { + self.finalize_transfer_end_with_runtime( + txn, + txid.clone(), + signed_psbt, + &info_contents, + TransferStatus::WaitingConfirmations, + fascia, + sync_tte_used, + consume_fascia, + runtime, + )? + } else { + self.finalize_transfer_end( + txn, + txid.clone(), + signed_psbt, + &info_contents, + TransferStatus::WaitingConfirmations, + fascia, + sync_tte_used, + consume_fascia, + )? + } } else { self.update_or_save_transfers( txn, @@ -4258,6 +4323,7 @@ pub trait WalletOnline: WalletOffline { TransferStatus::WaitingConfirmations, fascia, false, + true, )?; let (asset_id, transfer_info) = info_contents.transfers.into_iter().next().unwrap(); @@ -4408,6 +4474,7 @@ pub trait WalletOnline: WalletOffline { TransferStatus::WaitingConfirmations, fascia, false, + true, )?; Ok(OperationResult { @@ -4718,7 +4785,7 @@ pub trait WalletOnline: WalletOffline { } } - self.broadcast_and_update_rgb(txn, &mut runtime, signed_psbt, fascia)?; + self.broadcast_and_update_rgb(txn, &mut runtime, signed_psbt, fascia, true)?; let batch_transfer_idx = self.update_or_save_transfers( txn, diff --git a/src/wallet/rust_only.rs b/src/wallet/rust_only.rs index f108b902..bfba09ca 100644 --- a/src/wallet/rust_only.rs +++ b/src/wallet/rust_only.rs @@ -3,7 +3,745 @@ //! This module defines additional utility methods that are not exposed via FFI. use super::*; +#[cfg(any(feature = "electrum", feature = "esplora"))] +use crate::utils::{ + RgbRuntime, RgbRuntimeLock, acquire_rgb_runtime_lock, hash_bytes_hex, + load_rgb_runtime_for_operation, +}; +#[cfg(any(feature = "electrum", feature = "esplora"))] +use amplify::confinement::U32 as U32MAX; +#[cfg(any(feature = "electrum", feature = "esplora"))] +use nonasync::persistence::{PersistenceError, PersistenceProvider}; use rgbstd::Operation as _; +#[cfg(any(feature = "electrum", feature = "esplora"))] +use rgbstd::persistence::{MemIndex, MemStash, MemState}; +#[cfg(any(feature = "electrum", feature = "esplora"))] +use strict_encoding::StrictSerialize; + +#[cfg(all(test, any(feature = "electrum", feature = "esplora")))] +static RGB_PERSISTENCE_STEP: std::sync::atomic::AtomicUsize = + std::sync::atomic::AtomicUsize::new(0); + +#[cfg(all(test, any(feature = "electrum", feature = "esplora")))] +fn rgb_persistence_checkpoint(name: &str) { + use std::io::Write as _; + use std::sync::atomic::Ordering; + + let step = RGB_PERSISTENCE_STEP.fetch_add(1, Ordering::SeqCst) + 1; + if let Ok(path) = std::env::var("RGB_ACCEPTANCE_TRACE_PATH") { + let mut file = fs::OpenOptions::new() + .create(true) + .append(true) + .open(path) + .expect("open RGB acceptance trace"); + writeln!(file, "{step}:{name}").expect("write RGB acceptance trace"); + file.sync_all().expect("sync RGB acceptance trace"); + } + let target = std::env::var("RGB_ACCEPTANCE_CRASH_AFTER_STEP") + .ok() + .and_then(|value| value.parse::().ok()); + if target == Some(step) { + let path = std::env::var("RGB_ACCEPTANCE_CRASH_READY_PATH") + .expect("RGB acceptance crash-ready path"); + let mut file = fs::File::create(path).expect("create RGB acceptance crash-ready file"); + writeln!(file, "{step}:{name}").expect("write RGB acceptance crash-ready file"); + file.sync_all() + .expect("sync RGB acceptance crash-ready file"); + loop { + std::thread::park(); + } + } +} + +#[cfg(all(not(test), any(feature = "electrum", feature = "esplora")))] +#[inline] +fn rgb_persistence_checkpoint(_name: &str) {} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +#[derive(Clone, Debug)] +struct AcceptanceFsBinStore(FsBinStore); + +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl AcceptanceFsBinStore { + fn new(path: PathBuf) -> Result { + rgb_persistence_checkpoint("before-stage-directory-create"); + fs::create_dir(&path)?; + rgb_persistence_checkpoint("after-stage-directory-create"); + Ok(Self(FsBinStore { + stash: path.join("stash.dat"), + state: path.join("state.dat"), + index: path.join("index.dat"), + })) + } +} + +#[cfg(all(test, any(feature = "electrum", feature = "esplora")))] +fn write_all_checkpointed( + file: &mut fs::File, + bytes: &[u8], + checkpoint: &str, +) -> std::io::Result<()> { + const WRITE_CHUNK_SIZE: usize = 4096; + + for (chunk_index, chunk) in bytes.chunks(WRITE_CHUNK_SIZE).enumerate() { + let mut written = 0; + while written < chunk.len() { + rgb_persistence_checkpoint(&format!( + "before-{checkpoint}-write-{chunk_index}-{written}" + )); + let count = file.write(&chunk[written..])?; + rgb_persistence_checkpoint(&format!( + "after-{checkpoint}-write-{chunk_index}-{written}" + )); + if count == 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::WriteZero, + "failed to persist RGB acceptance data", + )); + } + written += count; + } + } + Ok(()) +} + +#[cfg(all(not(test), any(feature = "electrum", feature = "esplora")))] +fn write_all_checkpointed( + file: &mut fs::File, + bytes: &[u8], + _checkpoint: &str, +) -> std::io::Result<()> { + file.write_all(bytes) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn write_file_durably(path: &Path, bytes: &[u8], checkpoint: &str) -> std::io::Result<()> { + rgb_persistence_checkpoint(&format!("before-{checkpoint}-create")); + let mut file = fs::File::create(path)?; + rgb_persistence_checkpoint(&format!("after-{checkpoint}-create")); + write_all_checkpointed(&mut file, bytes, checkpoint)?; + rgb_persistence_checkpoint(&format!("before-{checkpoint}-sync")); + file.sync_all()?; + rgb_persistence_checkpoint(&format!("after-{checkpoint}-sync")); + Ok(()) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +macro_rules! impl_acceptance_store { + ($object:ty, $field:ident, $name:literal) => { + impl PersistenceProvider<$object> for AcceptanceFsBinStore { + fn load(&self) -> Result<$object, PersistenceError> { + >::load(&self.0) + } + + fn store(&self, object: &$object) -> Result<(), PersistenceError> { + let bytes = object + .to_strict_serialized::() + .map_err(PersistenceError::with)?; + write_file_durably( + &self.0.$field, + bytes.as_slice(), + concat!("stage-store-", $name), + ) + .map_err(PersistenceError::with) + } + } + }; +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl_acceptance_store!(MemStash, stash, "stash"); +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl_acceptance_store!(MemState, state, "state"); +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl_acceptance_store!(MemIndex, index, "index"); + +#[cfg(any(feature = "electrum", feature = "esplora"))] +const RGB_ACCEPTANCE_JOURNAL_FILE: &str = "rgb_acceptance_journal.json"; + +#[cfg(any(feature = "electrum", feature = "esplora"))] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)] +enum RgbAcceptanceJournalPhase { + Prepared, + Promoting, + Promoted, +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] +struct RgbAcceptanceJournal { + version: u8, + operation_id: String, + phase: RgbAcceptanceJournalPhase, + stage_dir_name: String, + backup_dir_name: String, + #[serde(default)] + asset_metadata: Option, +} + +/// Resolution to apply to a previously promoted RGB transfer acceptance. +#[cfg(any(feature = "electrum", feature = "esplora"))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RgbAcceptanceResolution { + /// Keep the promoted RGB stock after the associated protocol state is known to be durable. + Finalize, + /// Restore the exact RGB stock that preceded the interrupted protocol operation. + Rollback, +} + +/// Durable metadata for an RGB transfer acceptance awaiting protocol reconciliation. +#[cfg(any(feature = "electrum", feature = "esplora"))] +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PendingRgbAcceptance { + operation_id: String, + promoted: bool, +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl PendingRgbAcceptance { + /// Stable identifier supplied when the acceptance was prepared. + pub fn operation_id(&self) -> &str { + &self.operation_id + } + + /// Whether the staged RGB stock has replaced the previous live stock. + pub fn promoted(&self) -> bool { + self.promoted + } +} + +/// A validated RGB transfer whose resulting stock is staged but not visible to the wallet. +#[cfg(any(feature = "electrum", feature = "esplora"))] +pub struct PreparedRgbTransferAcceptance { + journal: RgbAcceptanceJournal, + wallet_dir: PathBuf, + live_runtime: Option, + consignment: Option, + assignments: Option>, + media_digests: HashSet, + finished: bool, +} + +/// A promoted RGB transfer acceptance awaiting a durable protocol decision. +#[cfg(any(feature = "electrum", feature = "esplora"))] +pub struct PromotedRgbTransferAcceptance { + journal: RgbAcceptanceJournal, + wallet_dir: PathBuf, + consignment: RgbTransfer, + assignments: Vec, +} + +/// A staged RGB fascia whose resulting stock is not yet visible to the wallet. +#[cfg(any(feature = "electrum", feature = "esplora"))] +pub struct PreparedRgbFasciaTransition { + journal: RgbAcceptanceJournal, + wallet_dir: PathBuf, + live_runtime: Option, + finished: bool, +} + +/// A promoted RGB fascia awaiting a durable protocol decision. +#[cfg(any(feature = "electrum", feature = "esplora"))] +pub struct PromotedRgbFasciaTransition { + journal: RgbAcceptanceJournal, + wallet_dir: PathBuf, +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl PromotedRgbFasciaTransition { + /// Stable identifier supplied when the fascia was prepared. + pub fn operation_id(&self) -> &str { + &self.journal.operation_id + } + + /// Reconciles the promoted stock with the associated durable protocol state. + pub fn resolve(self, resolution: RgbAcceptanceResolution) -> Result<(), Error> { + resolve_rgb_acceptance(&self.wallet_dir, &self.journal.operation_id, resolution) + } +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl PromotedRgbTransferAcceptance { + /// Stable identifier supplied when the acceptance was prepared. + pub fn operation_id(&self) -> &str { + &self.journal.operation_id + } + + /// Transfer consignment accepted into the promoted RGB stock. + pub fn consignment(&self) -> &RgbTransfer { + &self.consignment + } + + /// Assignments addressed to the receiving wallet. + pub fn assignments(&self) -> &[Assignment] { + &self.assignments + } + + /// Reconciles the promoted stock with the associated durable protocol state. + pub fn resolve(self, resolution: RgbAcceptanceResolution) -> Result<(), Error> { + resolve_rgb_acceptance(&self.wallet_dir, &self.journal.operation_id, resolution) + } +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl PreparedRgbTransferAcceptance { + /// Stable identifier supplied when the acceptance was prepared. + pub fn operation_id(&self) -> &str { + &self.journal.operation_id + } + + /// Transfer consignment accepted into the staged RGB stock. + pub fn consignment(&self) -> &RgbTransfer { + self.consignment + .as_ref() + .expect("prepared consignment must be present") + } + + /// Assignments addressed to the receiving wallet. + pub fn assignments(&self) -> &[Assignment] { + self.assignments + .as_deref() + .expect("prepared assignments must be present") + } + + /// Hex-encoded digests of media attachments declared by the validated transfer. + pub fn media_digests(&self) -> &HashSet { + &self.media_digests + } + + /// Atomically promotes the staged RGB stock while retaining a rollback snapshot. + pub fn promote(mut self) -> Result { + promote_staged_rgb_stock(&self.wallet_dir, &mut self.journal, &mut self.live_runtime)?; + + let consignment = self.consignment.take().ok_or_else(|| Error::Internal { + details: s!("prepared RGB acceptance is missing its consignment"), + })?; + let assignments = self.assignments.take().ok_or_else(|| Error::Internal { + details: s!("prepared RGB acceptance is missing its assignments"), + })?; + self.finished = true; + Ok(PromotedRgbTransferAcceptance { + journal: self.journal.clone(), + wallet_dir: self.wallet_dir.clone(), + consignment, + assignments, + }) + } + + /// Discards the staged result without changing the live RGB stock. + pub fn abort(mut self) -> Result<(), Error> { + let runtime = self.live_runtime.as_ref().ok_or_else(|| Error::Internal { + details: s!("prepared RGB acceptance is missing its live runtime"), + })?; + rollback_rgb_acceptance_locked(&self.wallet_dir, &self.journal, runtime.lock())?; + self.finished = true; + Ok(()) + } +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl PreparedRgbFasciaTransition { + /// Stable identifier supplied when the fascia was prepared. + pub fn operation_id(&self) -> &str { + &self.journal.operation_id + } + + /// Atomically promotes the staged RGB stock while retaining a rollback snapshot. + pub fn promote(mut self) -> Result { + promote_staged_rgb_stock(&self.wallet_dir, &mut self.journal, &mut self.live_runtime)?; + self.finished = true; + Ok(PromotedRgbFasciaTransition { + journal: self.journal.clone(), + wallet_dir: self.wallet_dir.clone(), + }) + } + + /// Discards the staged result without changing the live RGB stock. + pub fn abort(mut self) -> Result<(), Error> { + let runtime = self.live_runtime.as_ref().ok_or_else(|| Error::Internal { + details: s!("prepared RGB acceptance is missing its live runtime"), + })?; + rollback_rgb_acceptance_locked(&self.wallet_dir, &self.journal, runtime.lock())?; + self.finished = true; + Ok(()) + } +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl Drop for PreparedRgbFasciaTransition { + fn drop(&mut self) { + if !self.finished + && let Some(runtime) = self.live_runtime.as_ref() + { + let _ = rollback_rgb_acceptance_locked(&self.wallet_dir, &self.journal, runtime.lock()); + } + } +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +impl Drop for PreparedRgbTransferAcceptance { + fn drop(&mut self) { + if !self.finished + && let Some(runtime) = self.live_runtime.as_ref() + { + let _ = rollback_rgb_acceptance_locked(&self.wallet_dir, &self.journal, runtime.lock()); + } + } +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn rgb_acceptance_journal_path(wallet_dir: &Path) -> PathBuf { + wallet_dir.join(RGB_ACCEPTANCE_JOURNAL_FILE) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn sync_directory(path: &Path, checkpoint: &str) -> Result<(), Error> { + rgb_persistence_checkpoint(&format!("before-{checkpoint}-open")); + let directory = fs::File::open(path)?; + rgb_persistence_checkpoint(&format!("after-{checkpoint}-open")); + rgb_persistence_checkpoint(&format!("before-{checkpoint}-sync")); + directory.sync_all()?; + rgb_persistence_checkpoint(&format!("after-{checkpoint}-sync")); + Ok(()) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn rename_path(from: &Path, to: &Path, checkpoint: &str) -> Result<(), Error> { + rgb_persistence_checkpoint(&format!("before-{checkpoint}")); + fs::rename(from, to)?; + rgb_persistence_checkpoint(&format!("after-{checkpoint}")); + Ok(()) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn promote_staged_rgb_stock_filesystem( + wallet_dir: &Path, + journal: &mut RgbAcceptanceJournal, +) -> Result<(), Error> { + journal.phase = RgbAcceptanceJournalPhase::Promoting; + write_rgb_acceptance_journal(wallet_dir, journal)?; + + let live_dir = wallet_dir.join(crate::utils::RGB_RUNTIME_DIR); + let stage_dir = wallet_dir.join(&journal.stage_dir_name); + let backup_dir = wallet_dir.join(&journal.backup_dir_name); + if !live_dir.is_dir() || !stage_dir.is_dir() || backup_dir.exists() { + return Err(Error::Internal { + details: s!("invalid RGB stock promotion filesystem state"), + }); + } + + rename_path(&live_dir, &backup_dir, "promote-live-to-backup")?; + // The rollback copy must be durable before the staged stock can replace the live stock. + sync_directory(wallet_dir, "promote-sync-backup-installed")?; + if let Err(error) = rename_path(&stage_dir, &live_dir, "promote-stage-to-live") { + rename_path(&backup_dir, &live_dir, "promote-restore-live-after-error")?; + sync_directory(wallet_dir, "promote-sync-error-restore")?; + return Err(error); + } + sync_directory(wallet_dir, "promote-sync-installed-live")?; + + journal.phase = RgbAcceptanceJournalPhase::Promoted; + write_rgb_acceptance_journal(wallet_dir, journal) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn promote_staged_rgb_stock( + wallet_dir: &Path, + journal: &mut RgbAcceptanceJournal, + live_runtime: &mut Option, +) -> Result<(), Error> { + live_runtime + .as_mut() + .ok_or_else(|| Error::Internal { + details: s!("prepared RGB acceptance is missing its live runtime"), + })? + .suppress_persistence(); + promote_staged_rgb_stock_filesystem(wallet_dir, journal)?; + drop(live_runtime.take()); + Ok(()) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn persist_staged_rgb_stock( + wallet_dir: &Path, + operation_id: String, + mut staged_stock: Stock, + asset_metadata: Option, +) -> Result { + let digest = hash_bytes_hex(operation_id.as_bytes()); + let journal = RgbAcceptanceJournal { + version: 1, + operation_id, + phase: RgbAcceptanceJournalPhase::Prepared, + stage_dir_name: format!(".rgb-acceptance-{digest}-stage"), + backup_dir_name: format!(".rgb-acceptance-{digest}-backup"), + asset_metadata, + }; + validate_rgb_acceptance_journal(&journal)?; + cleanup_orphaned_rgb_acceptance_artifacts(wallet_dir)?; + let stage_dir = wallet_dir.join(&journal.stage_dir_name); + let backup_dir = wallet_dir.join(&journal.backup_dir_name); + remove_path_if_present(&stage_dir, "stale-stage-removed")?; + remove_path_if_present(&backup_dir, "stale-backup-removed")?; + let provider = AcceptanceFsBinStore::new(stage_dir.clone())?; + rgb_persistence_checkpoint("before-stage-provider-attach"); + staged_stock + .make_persistent(provider, false) + .map_err(|error| Error::IO { + details: error.to_string(), + })?; + rgb_persistence_checkpoint("after-stage-provider-attach"); + rgb_persistence_checkpoint("before-stage-stock-store"); + staged_stock.store().map_err(|error| Error::IO { + details: error.to_string(), + })?; + rgb_persistence_checkpoint("after-stage-stock-store"); + sync_directory(&stage_dir, "stage-sync-directory")?; + write_rgb_acceptance_journal(wallet_dir, &journal)?; + Ok(journal) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn write_rgb_acceptance_journal( + wallet_dir: &Path, + journal: &RgbAcceptanceJournal, +) -> Result<(), Error> { + let path = rgb_acceptance_journal_path(wallet_dir); + let temporary_path = path.with_extension("json.tmp"); + let bytes = serde_json::to_vec(journal).map_err(|error| Error::IO { + details: error.to_string(), + })?; + write_file_durably(&temporary_path, &bytes, "journal-temp")?; + rename_path(&temporary_path, &path, "journal-installed")?; + sync_directory(wallet_dir, "journal-parent-synced") +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn read_rgb_acceptance_journal(wallet_dir: &Path) -> Result, Error> { + let path = rgb_acceptance_journal_path(wallet_dir); + if !path.exists() { + return Ok(None); + } + let bytes = fs::read(path)?; + let journal = serde_json::from_slice(&bytes).map_err(|error| Error::IO { + details: error.to_string(), + })?; + Ok(Some(journal)) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +pub(crate) fn pending_rgb_acceptance_operation(wallet_dir: &Path) -> Result, Error> { + let Some(journal) = read_rgb_acceptance_journal(wallet_dir)? else { + return Ok(None); + }; + validate_rgb_acceptance_journal(&journal)?; + Ok(Some(journal.operation_id)) +} + +#[cfg(not(any(feature = "electrum", feature = "esplora")))] +pub(crate) fn pending_rgb_acceptance_operation( + _wallet_dir: &Path, +) -> Result, Error> { + Ok(None) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +pub(crate) fn validate_rgb_runtime_access( + wallet_dir: &Path, + operation_id: Option<&str>, +) -> Result<(), Error> { + let Some(journal) = read_rgb_acceptance_journal(wallet_dir)? else { + return Ok(()); + }; + validate_rgb_acceptance_journal(&journal)?; + if operation_id == Some(journal.operation_id.as_str()) + && journal.phase == RgbAcceptanceJournalPhase::Promoted + { + return Ok(()); + } + Err(Error::RgbOperationInProgress { + operation_id: journal.operation_id, + }) +} + +#[cfg(not(any(feature = "electrum", feature = "esplora")))] +pub(crate) fn validate_rgb_runtime_access( + _wallet_dir: &Path, + _operation_id: Option<&str>, +) -> Result<(), Error> { + Ok(()) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn validate_rgb_acceptance_journal(journal: &RgbAcceptanceJournal) -> Result<(), Error> { + let digest = hash_bytes_hex(journal.operation_id.as_bytes()); + let expected_stage = format!(".rgb-acceptance-{digest}-stage"); + let expected_backup = format!(".rgb-acceptance-{digest}-backup"); + if journal.version != 1 + || journal.stage_dir_name != expected_stage + || journal.backup_dir_name != expected_backup + { + return Err(Error::Internal { + details: s!("invalid RGB acceptance journal"), + }); + } + Ok(()) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn remove_path_if_present(path: &Path, checkpoint: &str) -> Result<(), Error> { + let metadata = match fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error.into()), + }; + if metadata.is_dir() && !metadata.file_type().is_symlink() { + let mut entries = fs::read_dir(path)?.collect::, _>>()?; + entries.sort_by_key(|entry| entry.file_name()); + for entry in entries { + remove_path_if_present(&entry.path(), checkpoint)?; + } + rgb_persistence_checkpoint(&format!("before-{checkpoint}-remove-directory")); + fs::remove_dir(path)?; + rgb_persistence_checkpoint(&format!("after-{checkpoint}-remove-directory")); + } else { + rgb_persistence_checkpoint(&format!("before-{checkpoint}-remove-file")); + fs::remove_file(path)?; + rgb_persistence_checkpoint(&format!("after-{checkpoint}-remove-file")); + } + Ok(()) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn cleanup_orphaned_rgb_acceptance_artifacts(wallet_dir: &Path) -> Result<(), Error> { + if rgb_acceptance_journal_path(wallet_dir).exists() { + return Ok(()); + } + let mut removed = false; + for entry in fs::read_dir(wallet_dir)? { + let entry = entry?; + let name = entry.file_name(); + let Some(name) = name.to_str() else { + continue; + }; + let is_acceptance_artifact = name.starts_with(".rgb-acceptance-") + && (name.ends_with("-stage") + || name.ends_with("-backup") + || name.ends_with("-stage.discarded")); + if is_acceptance_artifact { + remove_path_if_present(&entry.path(), "orphan-artifact-removed")?; + removed = true; + } + } + let temporary_journal = rgb_acceptance_journal_path(wallet_dir).with_extension("json.tmp"); + if temporary_journal.exists() { + remove_path_if_present(&temporary_journal, "orphan-journal-temp-removed")?; + removed = true; + } + if removed { + sync_directory(wallet_dir, "orphan-cleanup-synced")?; + } + Ok(()) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn rollback_rgb_acceptance_locked( + wallet_dir: &Path, + journal: &RgbAcceptanceJournal, + _lock: &RgbRuntimeLock, +) -> Result<(), Error> { + validate_rgb_acceptance_journal(journal)?; + let live_dir = wallet_dir.join(crate::utils::RGB_RUNTIME_DIR); + let stage_dir = wallet_dir.join(&journal.stage_dir_name); + let backup_dir = wallet_dir.join(&journal.backup_dir_name); + + if backup_dir.exists() { + let discarded_dir = wallet_dir.join(format!("{}.discarded", journal.stage_dir_name)); + remove_path_if_present(&discarded_dir, "rollback-old-discarded-removed")?; + if live_dir.exists() { + rename_path(&live_dir, &discarded_dir, "rollback-live-to-discarded")?; + } + rename_path(&backup_dir, &live_dir, "rollback-backup-to-live")?; + sync_directory(wallet_dir, "rollback-restored-live-synced")?; + remove_path_if_present(&discarded_dir, "rollback-discarded-removed")?; + } else if !live_dir.is_dir() { + return Err(Error::Internal { + details: s!("RGB rollback has neither live stock nor backup stock"), + }); + } else { + let discarded_dir = wallet_dir.join(format!("{}.discarded", journal.stage_dir_name)); + remove_path_if_present(&discarded_dir, "rollback-orphan-discarded-removed")?; + } + remove_path_if_present(&stage_dir, "rollback-stage-removed")?; + let journal_path = rgb_acceptance_journal_path(wallet_dir); + if journal_path.exists() { + rgb_persistence_checkpoint("before-rollback-journal-remove"); + fs::remove_file(journal_path)?; + rgb_persistence_checkpoint("after-rollback-journal-remove"); + } + sync_directory(wallet_dir, "rollback-cleanup-synced") +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn rollback_rgb_acceptance(wallet_dir: &Path, journal: &RgbAcceptanceJournal) -> Result<(), Error> { + let lock = acquire_rgb_runtime_lock(wallet_dir)?; + rollback_rgb_acceptance_locked(wallet_dir, journal, &lock) +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn finalize_rgb_acceptance(wallet_dir: &Path, journal: &RgbAcceptanceJournal) -> Result<(), Error> { + let _lock = acquire_rgb_runtime_lock(wallet_dir)?; + validate_rgb_acceptance_journal(journal)?; + if journal.phase != RgbAcceptanceJournalPhase::Promoted + || !wallet_dir.join(crate::utils::RGB_RUNTIME_DIR).exists() + { + return Err(Error::Internal { + details: s!("RGB acceptance has not been promoted"), + }); + } + remove_path_if_present( + &wallet_dir.join(&journal.stage_dir_name), + "finalize-stage-removed", + )?; + remove_path_if_present( + &wallet_dir.join(&journal.backup_dir_name), + "finalize-backup-removed", + )?; + sync_directory(wallet_dir, "finalize-stock-cleanup-synced")?; + let journal_path = rgb_acceptance_journal_path(wallet_dir); + if journal_path.exists() { + rgb_persistence_checkpoint("before-finalize-journal-remove"); + fs::remove_file(journal_path)?; + rgb_persistence_checkpoint("after-finalize-journal-remove"); + } + sync_directory(wallet_dir, "finalize-journal-removal-synced") +} + +#[cfg(any(feature = "electrum", feature = "esplora"))] +fn resolve_rgb_acceptance( + wallet_dir: &Path, + operation_id: &str, + resolution: RgbAcceptanceResolution, +) -> Result<(), Error> { + let journal = read_rgb_acceptance_journal(wallet_dir)?.ok_or_else(|| Error::Internal { + details: s!("RGB acceptance journal not found"), + })?; + validate_rgb_acceptance_journal(&journal)?; + if journal.operation_id != operation_id { + return Err(Error::Internal { + details: s!("RGB acceptance operation ID mismatch"), + }); + } + match resolution { + RgbAcceptanceResolution::Finalize if journal.asset_metadata.is_some() => { + Err(Error::Internal { + details: s!("RGB transfer acceptance must be finalized through the wallet"), + }) + } + RgbAcceptanceResolution::Finalize => finalize_rgb_acceptance(wallet_dir, &journal), + RgbAcceptanceResolution::Rollback => rollback_rgb_acceptance(wallet_dir, &journal), + } +} /// RGB asset-specific information to color a transaction #[derive(Debug, Clone)] @@ -369,6 +1107,18 @@ impl Wallet { coloring_info: ColoringInfo, ) -> Result<(Fascia, AssetBeneficiariesMap), Error> { info!(self.logger(), "Coloring PSBT..."); + let runtime = self.rgb_runtime()?; + let result = self.color_psbt_with_runtime(psbt, coloring_info, &runtime)?; + info!(self.logger(), "Color PSBT completed"); + Ok(result) + } + + fn color_psbt_with_runtime( + &self, + psbt: &mut Psbt, + coloring_info: ColoringInfo, + runtime: &RgbRuntime, + ) -> Result<(Fascia, AssetBeneficiariesMap), Error> { let mut transaction = match psbt.clone().extract_tx() { Ok(tx) => tx, Err(ExtractTxError::MissingInputValue { tx }) => tx, // required for non-standard TXs @@ -396,8 +1146,6 @@ impl Wallet { *psbt = Psbt::from_unsigned_tx(transaction).unwrap(); } - let runtime = self.rgb_runtime()?; - let prev_outputs = psbt .unsigned_tx .input @@ -410,12 +1158,11 @@ impl Wallet { let assignment_name = FieldName::from(RGB_STATE_ASSET_OWNER); for (contract_id, asset_coloring_info) in coloring_info.asset_info_map.clone() { - let schema = - AssetSchema::get_from_contract_id(contract_id, &runtime).map_err(|_| { - Error::AssetNotFound { - asset_id: contract_id.to_string(), - } - })?; + let schema = AssetSchema::get_from_contract_id(contract_id, runtime).map_err(|_| { + Error::AssetNotFound { + asset_id: contract_id.to_string(), + } + })?; let mut asset_transition_builder = runtime.transition_builder(contract_id, "transfer")?; @@ -534,7 +1281,6 @@ impl Wallet { psbt.set_as_unmodifiable(); let fascia = psbt.rgb_commit().map_err(InternalError::from)?; - info!(self.logger(), "Color PSBT completed"); Ok((fascia, asset_beneficiaries)) } @@ -548,11 +1294,12 @@ impl Wallet { coloring_info: ColoringInfo, ) -> Result, Error> { info!(self.logger(), "Coloring PSBT and consuming..."); - let (fascia, asset_beneficiaries) = self.color_psbt(psbt, coloring_info.clone())?; + let mut runtime = self.rgb_runtime()?; + let (fascia, asset_beneficiaries) = + self.color_psbt_with_runtime(psbt, coloring_info, &runtime)?; let witness_txid = psbt.get_txid(); - let mut runtime = self.rgb_runtime()?; runtime.consume_fascia(fascia, None)?; let mut transfers = vec![]; @@ -582,6 +1329,35 @@ impl Wallet { Ok(transfers) } + /// Color and consume a PSBT as the owner of a promoted protocol operation. + /// + /// Normal wallet access remains blocked while a transactional acceptance is pending. This + /// method grants access only when `operation_id` matches the promoted journal owner, keeping + /// commitment construction inside the same rollback boundary as the surrounding protocol. + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn color_psbt_and_consume_for_operation( + &self, + operation_id: &str, + psbt: &mut Psbt, + coloring_info: ColoringInfo, + witness_ord: Option, + ) -> Result { + if operation_id.is_empty() { + return Err(Error::Internal { + details: s!("RGB protocol operation ID cannot be empty"), + }); + } + + let mut runtime = load_rgb_runtime_for_operation(self.wallet_dir(), operation_id)?; + let (fascia, _) = self.color_psbt_with_runtime(psbt, coloring_info, &runtime)?; + match psbt.clone().extract_tx() { + Ok(_) | Err(ExtractTxError::MissingInputValue { .. }) => {} + Err(error) => return Err(InternalError::from(error).into()), + } + runtime.consume_fascia(fascia.clone(), witness_ord)?; + Ok(fascia) + } + /// Create consignments for a PSBT created with the [`send_begin`](Wallet::send_begin) method. /// ///
This method is meant for special usage and is normally not needed, use @@ -667,26 +1443,28 @@ impl Wallet { "Got consignment for asset with {} schema", asset_schema ); - let mut runtime = self.rgb_runtime()?; - - let graph_seal = GraphSeal::with_blinded_vout(vout, blinding); - runtime.store_secret_seal(graph_seal)?; - - let resolver = OffchainResolver { + let fallback_resolver = OffchainResolver { witness_id, consignment: &consignment, fallback: self.blockchain_resolver(), }; + let mut runtime = self.rgb_runtime()?; - debug!(self.logger(), "Validating consignment..."); - let asset_schema: AssetSchema = consignment.schema_id().try_into()?; + let graph_seal = GraphSeal::with_blinded_vout(vout, blinding); + runtime.store_secret_seal(graph_seal)?; + + debug!(self.logger(), "Validating consignment..."); + let asset_schema: AssetSchema = consignment.schema_id().try_into()?; let trusted_typesystem = asset_schema.types(); let validation_config = ValidationConfig { chain_net: self.chain_net(), trusted_typesystem, ..Default::default() }; - let valid_consignment = match consignment.clone().validate(&resolver, &validation_config) { + let valid_consignment = match consignment + .clone() + .validate(&fallback_resolver, &validation_config) + { Ok(consignment) => consignment, Err(ValidationError::InvalidConsignment(e)) => { error!(self.logger(), "Consignment is invalid: {}", e); @@ -708,12 +1486,12 @@ impl Wallet { .iter() .map(|a| hex::encode(a.digest)) .collect::>(); - runtime.import_contract(valid_contract, self.blockchain_resolver())?; + runtime.import_contract(valid_contract, &fallback_resolver)?; let received_rgb_assignments = self.extract_received_assignments(&consignment, witness_id, Some(vout), None); - runtime.accept_transfer(valid_consignment, &resolver)?; + runtime.accept_transfer(valid_consignment, &fallback_resolver)?; info!(self.logger(), "Accept transfer completed"); Ok(( @@ -723,6 +1501,303 @@ impl Wallet { )) } + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn prepare_accept_transfer_consignment( + &mut self, + operation_id: String, + txid: String, + vout: u32, + consignment: RgbTransfer, + blinding: u64, + ) -> Result { + info!(self.logger(), "Preparing transfer acceptance..."); + if operation_id.is_empty() { + return Err(Error::Internal { + details: s!("RGB acceptance operation ID cannot be empty"), + }); + } + let wallet_dir = self.wallet_dir().clone(); + if let Some(existing) = read_rgb_acceptance_journal(&wallet_dir)? { + return Err(Error::Internal { + details: format!( + "RGB acceptance operation '{}' requires reconciliation", + existing.operation_id + ), + }); + } + + let witness_id = RgbTxid::from_str(&txid).map_err(|_| Error::InvalidTxid)?; + let asset_schema: AssetSchema = consignment.schema_id().to_string().try_into()?; + self.check_schema_support(&asset_schema)?; + let fallback_resolver = OffchainResolver { + witness_id, + consignment: &consignment, + fallback: self.blockchain_resolver(), + }; + let validation_config = ValidationConfig { + chain_net: self.chain_net(), + trusted_typesystem: asset_schema.types(), + ..Default::default() + }; + let valid_consignment = match consignment + .clone() + .validate(&fallback_resolver, &validation_config) + { + Ok(consignment) => consignment, + Err(ValidationError::InvalidConsignment(error)) => { + error!(self.logger(), "Consignment is invalid: {}", error); + return Err(Error::InvalidConsignment); + } + Err(ValidationError::ResolverError(error)) => { + return Err(Error::Network { + details: error.to_string(), + }); + } + }; + let assignments = self + .extract_received_assignments(&consignment, witness_id, Some(vout), None) + .into_values() + .collect(); + let graph_seal = GraphSeal::with_blinded_vout(vout, blinding); + let runtime = self.rgb_runtime()?; + let contract_id = valid_consignment.contract_id(); + let valid_contract = valid_consignment.clone().into_valid_contract(); + let media_digests = self + .extract_attachments(&valid_contract, asset_schema) + .iter() + .map(|attachment| hex::encode(attachment.digest)) + .collect(); + let asset_metadata = { + let txn = self.database().begin_transaction()?; + let metadata = if txn.get_asset(contract_id.to_string())?.is_none() { + Some(self.extract_asset_data( + &runtime, + contract_id, + asset_schema, + valid_contract, + Some(valid_consignment.clone()), + )?) + } else { + None + }; + txn.commit()?; + metadata + }; + let staged_stock = + runtime.stage_transfer(graph_seal, valid_consignment, &fallback_resolver)?; + + let journal = + persist_staged_rgb_stock(&wallet_dir, operation_id, staged_stock, asset_metadata)?; + + info!(self.logger(), "Prepare transfer acceptance completed"); + Ok(PreparedRgbTransferAcceptance { + journal, + wallet_dir, + live_runtime: Some(runtime), + consignment: Some(consignment), + assignments: Some(assignments), + media_digests, + finished: false, + }) + } + + /// Validates an RGB transfer and persists its resulting stock in an isolated staging area. + /// + /// The live RGB stock remains byte-for-byte unchanged until + /// [`PreparedRgbTransferAcceptance::promote`] is called. The operation ID must be stable for + /// the surrounding protocol operation and is used to reconcile interrupted promotion. + /// + ///
This method is meant for protocol integrations that provide their + /// own durable commit decision.
+ #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn prepare_accept_transfer( + &mut self, + operation_id: String, + txid: String, + vout: u32, + consignment_endpoint: RgbTransport, + blinding: u64, + ) -> Result { + if operation_id.is_empty() { + return Err(Error::Internal { + details: s!("RGB acceptance operation ID cannot be empty"), + }); + } + if let Some(existing) = read_rgb_acceptance_journal(self.wallet_dir())? { + return Err(Error::Internal { + details: format!( + "RGB acceptance operation '{}' requires reconciliation", + existing.operation_id + ), + }); + } + + let proxy_url = TransportEndpoint::try_from(consignment_endpoint)?.endpoint; + let consignment_res = self.get_consignment(&proxy_url, txid.clone())?; + let consignment_bytes = general_purpose::STANDARD + .decode(consignment_res.consignment) + .map_err(InternalError::from)?; + let consignment = RgbTransfer::load(&consignment_bytes[..]).map_err(InternalError::from)?; + self.prepare_accept_transfer_consignment(operation_id, txid, vout, consignment, blinding) + } + + /// Validates persisted RGB transfer bytes into the same isolated staging area used by live + /// transfer acceptance. + /// + /// This is intended for deterministic protocol recovery when the surrounding journal already + /// contains the exact transfer consignment. It does not fetch or trust replacement data from a + /// transport endpoint. + /// + ///
This method is meant for protocol integrations that provide their + /// own durable commit decision.
+ #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn prepare_accept_transfer_from_consignment( + &mut self, + operation_id: String, + txid: String, + vout: u32, + consignment_bytes: Vec, + blinding: u64, + ) -> Result { + let consignment = RgbTransfer::load(&consignment_bytes[..]).map_err(InternalError::from)?; + self.prepare_accept_transfer_consignment(operation_id, txid, vout, consignment, blinding) + } + + /// Returns whether the live RGB stock already contains the given contract witness. + /// + /// A missing contract is reported as `false`; malformed identifiers and stock corruption are + /// returned as errors. + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn has_accepted_transfer(&self, asset_id: String, txid: String) -> Result { + let contract_id = ContractId::from_str(&asset_id).map_err(|error| Error::Internal { + details: format!("invalid asset ID: {error}"), + })?; + let witness_id = RgbTxid::from_str(&txid).map_err(|_| Error::InvalidTxid)?; + self.rgb_runtime()? + .contains_transfer_witness(contract_id, witness_id) + .map_err(Error::from) + } + + /// Stages a fascia in an isolated RGB stock for a protocol-controlled commit. + /// + /// The operation must be finalized or rolled back after the surrounding protocol has made a + /// durable decision. Until promotion, the live wallet stock is unchanged. + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn prepare_consume_fascia( + &self, + operation_id: String, + fascia: Fascia, + witness_ord: Option, + ) -> Result { + if operation_id.is_empty() { + return Err(Error::Internal { + details: s!("RGB fascia operation ID cannot be empty"), + }); + } + let wallet_dir = self.wallet_dir().clone(); + if let Some(existing) = read_rgb_acceptance_journal(&wallet_dir)? { + return Err(Error::Internal { + details: format!( + "RGB operation '{}' requires reconciliation", + existing.operation_id + ), + }); + } + + let runtime = self.rgb_runtime()?; + let staged_stock = runtime.stage_fascia(fascia, witness_ord)?; + let journal = persist_staged_rgb_stock(&wallet_dir, operation_id, staged_stock, None)?; + Ok(PreparedRgbFasciaTransition { + journal, + wallet_dir, + live_runtime: Some(runtime), + finished: false, + }) + } + + /// Returns a durable RGB acceptance operation awaiting protocol reconciliation, if any. + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn pending_rgb_acceptance(&self) -> Result, Error> { + let Some(journal) = read_rgb_acceptance_journal(self.wallet_dir())? else { + return Ok(None); + }; + validate_rgb_acceptance_journal(&journal)?; + Ok(Some(PendingRgbAcceptance { + operation_id: journal.operation_id, + promoted: journal.phase == RgbAcceptanceJournalPhase::Promoted, + })) + } + + /// Resolves a durable RGB acceptance after inspecting the associated protocol state. + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn resolve_pending_rgb_acceptance( + &self, + operation_id: &str, + resolution: RgbAcceptanceResolution, + ) -> Result<(), Error> { + let journal = + read_rgb_acceptance_journal(self.wallet_dir())?.ok_or_else(|| Error::Internal { + details: s!("RGB acceptance journal not found"), + })?; + validate_rgb_acceptance_journal(&journal)?; + if journal.operation_id != operation_id { + return Err(Error::Internal { + details: s!("RGB acceptance operation ID mismatch"), + }); + } + + match resolution { + RgbAcceptanceResolution::Finalize => { + if let Some(asset_metadata) = journal.asset_metadata.as_ref() { + let txn = self.database().begin_transaction()?; + rgb_persistence_checkpoint("metadata-transaction-begun"); + match txn.get_asset(asset_metadata.asset_id.clone())? { + Some(existing) => { + if existing.schema != asset_metadata.asset_schema + || existing.name != asset_metadata.name + || existing.precision != asset_metadata.precision + || existing.ticker != asset_metadata.ticker + || existing.details != asset_metadata.details + || existing.initial_supply + != asset_metadata.initial_supply.to_string() + || existing.max_supply + != asset_metadata.max_supply.map(|value| value.to_string()) + || existing.timestamp != asset_metadata.timestamp + || existing.reject_list_url != asset_metadata.reject_list_url + { + return Err(Error::Internal { + details: format!( + "stored metadata for RGB asset '{}' does not match the validated transfer", + asset_metadata.asset_id + ), + }); + } + rgb_persistence_checkpoint("metadata-existing-asset-verified"); + } + None => { + self.add_asset_to_db(&txn, asset_metadata)?; + rgb_persistence_checkpoint("metadata-asset-inserted"); + self.update_backup_info(&txn, false)?; + rgb_persistence_checkpoint("metadata-backup-info-updated"); + } + } + rgb_persistence_checkpoint("metadata-transaction-commit-starting"); + txn.commit()?; + rgb_persistence_checkpoint("metadata-transaction-committed"); + // Re-notifying is intentional. A process may have died after the database + // commit but before the original notification reached the backup worker. + rgb_persistence_checkpoint("metadata-backup-notification-starting"); + self.trigger_auto_backup(); + rgb_persistence_checkpoint("metadata-backup-notification-completed"); + } + finalize_rgb_acceptance(self.wallet_dir(), &journal) + } + RgbAcceptanceResolution::Rollback => { + rollback_rgb_acceptance(self.wallet_dir(), &journal) + } + } + } + /// Consume an RGB fascia. /// ///
This method is meant for special usage and is normally not needed, use @@ -752,6 +1827,20 @@ impl Wallet { Ok(height) } + /// Return whether a Bitcoin transaction is visible to the configured indexer. + /// + /// Unlike [`Wallet::get_tx_height`], this also returns `true` for a transaction in the + /// mempool. Protocol recovery code must use this distinction before deciding whether a + /// staged RGB state transition can be rolled back. + #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn is_tx_known(&self, txid: String) -> Result { + let txid = RgbTxid::from_str(&txid).map_err(|_| Error::InvalidTxid)?; + Ok(self + .indexer() + .get_tx_confirmations(&txid.to_string())? + .is_some()) + } + /// Update RGB witnesses. /// ///
This method is meant for special usage and is normally not needed, use @@ -788,6 +1877,32 @@ impl Wallet { Ok(()) } + /// Manually set the [`WitnessOrd`] of a witness TX while owning a matching promoted protocol + /// operation. + /// + /// Normal RGB runtime access remains blocked while a transactional acceptance is pending. This + /// method grants access only when `operation_id` matches the promoted journal owner, keeping + /// the witness mutation inside the same rollback boundary as the surrounding protocol. + /// + ///
This method is intended only for protocol integrations that provide + /// their own durable commit decision.
+ #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn upsert_witness_for_operation( + &self, + operation_id: &str, + witness_id: RgbTxid, + witness_ord: WitnessOrd, + ) -> Result<(), Error> { + if operation_id.is_empty() { + return Err(Error::Internal { + details: s!("RGB protocol operation ID cannot be empty"), + }); + } + let mut runtime = load_rgb_runtime_for_operation(self.wallet_dir(), operation_id)?; + runtime.upsert_witness(witness_id, witness_ord)?; + Ok(()) + } + /// Extract the metadata of a new RGB asset and save the asset into the DB. /// ///
This method is meant for special usage and is normally not needed, use @@ -803,7 +1918,7 @@ impl Wallet { self.check_online(online)?; let contract_id = consignment.contract_id(); let witness_id = RgbTxid::from_str(&offchain_txid).map_err(|_| Error::InvalidTxid)?; - let resolver = OffchainResolver { + let fallback_resolver = OffchainResolver { witness_id, consignment: &consignment, fallback: self.blockchain_resolver(), @@ -815,7 +1930,10 @@ impl Wallet { trusted_typesystem, ..Default::default() }; - let valid_transfer = match consignment.clone().validate(&resolver, &validation_config) { + let valid_transfer = match consignment + .clone() + .validate(&fallback_resolver, &validation_config) + { Ok(consignment) => consignment, Err(ValidationError::InvalidConsignment(error)) => { error!(self.logger(), "Consignment is invalid: {}", error); @@ -959,20 +2077,65 @@ impl Wallet { &mut self, online: Online, signed_psbt: String, + ) -> Result { + self.send_end_db_update_only_impl(online, signed_psbt, None) + } + + /// Complete the donation send operation by updating the DB only while owning a matching + /// durable RGB stock operation. + /// + /// The signed transaction ID must match `operation_id`. Normal RGB runtime access remains + /// blocked while the operation is promoted; this entry point grants access only to the exact + /// journal owner and retains the stock lock through broadcast and database commit. + /// + ///
This method is intended only for protocol integrations that provide + /// their own durable commit decision and exchange consignments out of band.
+ #[cfg(any(feature = "electrum", feature = "esplora"))] + pub fn send_end_db_update_only_for_operation( + &mut self, + online: Online, + operation_id: &str, + signed_psbt: String, + ) -> Result { + if operation_id.is_empty() { + return Err(Error::Internal { + details: s!("RGB protocol operation ID cannot be empty"), + }); + } + self.send_end_db_update_only_impl(online, signed_psbt, Some(operation_id)) + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn send_end_db_update_only_impl( + &mut self, + online: Online, + signed_psbt: String, + operation_id: Option<&str>, ) -> Result { info!(self.logger(), "Sending (end) db update only..."); self.check_online(online)?; let psbt = Psbt::from_str(&signed_psbt)?; + let txid = psbt.unsigned_tx.compute_txid().to_string(); + if operation_id.is_some_and(|operation_id| operation_id != txid) { + return Err(Error::Internal { + details: s!("RGB protocol operation ID does not match the send transaction"), + }); + } + let _runtime_guard = match operation_id { + Some(operation_id) => load_rgb_runtime_for_operation(self.wallet_dir(), operation_id)?, + None => self.rgb_runtime()?, + }; let txn = self.database().begin_transaction()?; // this will also update the DB with the new UTXOs and BDK self.broadcast_psbt(&txn, &psbt)?; - let (txid, _, info_contents, _) = self.get_transfer_end_data(&psbt)?; + let (transfer_txid, _, info_contents, _) = self.get_transfer_end_data(&psbt)?; + debug_assert_eq!(transfer_txid, txid); let batch_transfer_idx = self.update_or_save_transfers( &txn, - txid.clone(), + transfer_txid.clone(), &info_contents, TransferStatus::WaitingConfirmations, true, @@ -984,7 +2147,7 @@ impl Wallet { info!(self.logger(), "Send (end) db update only completed"); Ok(OperationResult { - txid, + txid: transfer_txid, batch_transfer_idx, entropy: info_contents.entropy, }) @@ -995,6 +2158,549 @@ impl Wallet { #[cfg(any(feature = "electrum", feature = "esplora"))] mod tests { use super::*; + use std::process::{Command, Stdio}; + use std::time::{Duration, Instant}; + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn acceptance_journal( + operation_id: &str, + phase: RgbAcceptanceJournalPhase, + ) -> RgbAcceptanceJournal { + let digest = hash_bytes_hex(operation_id.as_bytes()); + RgbAcceptanceJournal { + version: 1, + operation_id: operation_id.to_owned(), + phase, + stage_dir_name: format!(".rgb-acceptance-{digest}-stage"), + backup_dir_name: format!(".rgb-acceptance-{digest}-backup"), + asset_metadata: None, + } + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn write_marker(directory: &Path, value: &str) { + fs::create_dir_all(directory).unwrap(); + fs::write(directory.join("marker"), value).unwrap(); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn read_marker(directory: &Path) -> String { + fs::read_to_string(directory.join("marker")).unwrap() + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn run_crash_child( + mode: &str, + wallet_dir: &Path, + crash_after_step: Option, + trace_path: Option<&Path>, + ready_path: Option<&Path>, + ) -> std::process::ExitStatus { + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .arg("--ignored") + .arg("--exact") + .arg("wallet::rust_only::tests::rgb_acceptance_crash_child") + .arg("--nocapture") + .env("RGB_ACCEPTANCE_CHILD_MODE", mode) + .env("RGB_ACCEPTANCE_CHILD_WALLET_DIR", wallet_dir) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()); + if let Some(step) = crash_after_step { + command.env("RGB_ACCEPTANCE_CRASH_AFTER_STEP", step.to_string()); + } + if let Some(path) = trace_path { + command.env("RGB_ACCEPTANCE_TRACE_PATH", path); + } + if let Some(path) = ready_path { + command.env("RGB_ACCEPTANCE_CRASH_READY_PATH", path); + } + + let mut child = command.spawn().unwrap(); + if let Some(ready_path) = ready_path { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if ready_path.exists() { + child.kill().expect("kill RGB acceptance child"); + return child.wait().expect("wait for killed RGB acceptance child"); + } + if let Some(status) = child.try_wait().expect("poll RGB acceptance child") { + panic!( + "RGB acceptance child exited before crash checkpoint with status {status}" + ); + } + assert!( + Instant::now() < deadline, + "RGB acceptance child did not reach crash checkpoint" + ); + std::thread::sleep(Duration::from_millis(10)); + } + } + child.wait().expect("wait for RGB acceptance child") + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn checkpoint_trace(mode: &str, setup: impl Fn(&Path)) -> Vec { + let directory = tempfile::tempdir().unwrap(); + setup(directory.path()); + let trace_path = directory.path().join("trace"); + let status = run_crash_child(mode, directory.path(), None, Some(&trace_path), None); + assert!( + status.success(), + "RGB acceptance trace child failed in mode '{mode}'" + ); + fs::read_to_string(trace_path) + .unwrap() + .lines() + .map(|line| { + line.split_once(':') + .expect("numbered RGB persistence checkpoint") + .1 + .to_owned() + }) + .collect() + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn checkpoint_count(mode: &str, setup: impl Fn(&Path)) -> usize { + checkpoint_trace(mode, setup).len() + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn kill_at_each_checkpoint( + mode: &str, + setup: impl Fn(&Path), + recover_and_assert: impl Fn(&Path), + ) { + let checkpoints = checkpoint_count(mode, &setup); + assert!( + checkpoints > 0, + "crash scenario has no persistence checkpoints" + ); + for step in 1..=checkpoints { + let directory = tempfile::tempdir().unwrap(); + setup(directory.path()); + let ready_path = directory.path().join("crash-ready"); + let status = + run_crash_child(mode, directory.path(), Some(step), None, Some(&ready_path)); + assert!(!status.success(), "crash child unexpectedly succeeded"); + recover_and_assert(directory.path()); + } + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn stock_fixture(wallet_dir: &Path, operation_id: &str, phase: RgbAcceptanceJournalPhase) { + let journal = acceptance_journal(operation_id, phase); + write_marker(&wallet_dir.join(crate::utils::RGB_RUNTIME_DIR), "old"); + write_marker(&wallet_dir.join(&journal.stage_dir_name), "new"); + write_rgb_acceptance_journal(wallet_dir, &journal).unwrap(); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn metadata_test_wallet(data_dir: &Path) -> Wallet { + const MNEMONIC: &str = "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"; + let keys = crate::keys::restore_keys( + BitcoinNetwork::Regtest, + MNEMONIC.to_owned(), + crate::keys::WitnessVersion::Taproot, + ) + .unwrap(); + Wallet::new( + WalletData { + data_dir: data_dir.to_string_lossy().into_owned(), + bitcoin_network: BitcoinNetwork::Regtest, + database_type: DatabaseType::Sqlite, + max_allocations_per_utxo: 5, + supported_schemas: AssetSchema::VALUES.to_vec(), + reuse_addresses: false, + }, + SinglesigKeys::from_keys(&keys, None), + ) + .unwrap() + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn acceptance_asset_metadata() -> LocalAssetData { + LocalAssetData { + asset_id: "rgb:Ar4ouaLv-b7f7Dc_-z5EMvtu-FA5KNh1-nlae~jk-8xMBo7E".to_owned(), + name: "Crash consistency asset".to_owned(), + asset_schema: AssetSchema::Nia, + precision: 2, + ticker: Some("CRSH".to_owned()), + details: None, + media: None, + initial_supply: 10_000, + max_supply: None, + known_circulating_supply: Some(10_000), + reject_list_url: None, + token: None, + timestamp: 1_700_000_000, + added_at: 1_700_000_000, + } + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn metadata_finalize_fixture(data_dir: &Path) { + let wallet = metadata_test_wallet(data_dir); + let wallet_dir = wallet.wallet_dir(); + let mut journal = acceptance_journal( + "metadata-finalize-operation", + RgbAcceptanceJournalPhase::Promoted, + ); + journal.asset_metadata = Some(acceptance_asset_metadata()); + fs::create_dir_all(wallet_dir.join(&journal.backup_dir_name)).unwrap(); + write_rgb_acceptance_journal(wallet_dir, &journal).unwrap(); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + fn recover_and_assert_metadata_finalize(data_dir: &Path) { + let wallet = metadata_test_wallet(data_dir); + if read_rgb_acceptance_journal(wallet.wallet_dir()) + .unwrap() + .is_some() + { + wallet + .resolve_pending_rgb_acceptance( + "metadata-finalize-operation", + RgbAcceptanceResolution::Finalize, + ) + .unwrap(); + } + let txn = wallet.database().begin_transaction().unwrap(); + let metadata = acceptance_asset_metadata(); + let stored = txn.get_asset(metadata.asset_id.clone()).unwrap().unwrap(); + assert_eq!(stored.id, metadata.asset_id); + assert_eq!(stored.schema, metadata.asset_schema); + assert_eq!(stored.initial_supply, metadata.initial_supply.to_string()); + txn.commit().unwrap(); + assert!(!rgb_acceptance_journal_path(wallet.wallet_dir()).exists()); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + #[ignore = "subprocess used by rgb_acceptance_os_kill_matrix"] + fn rgb_acceptance_crash_child() { + RGB_PERSISTENCE_STEP.store(0, std::sync::atomic::Ordering::SeqCst); + let mode = std::env::var("RGB_ACCEPTANCE_CHILD_MODE").unwrap(); + let wallet_dir = PathBuf::from(std::env::var("RGB_ACCEPTANCE_CHILD_WALLET_DIR").unwrap()); + match mode.as_str() { + "persist" => { + persist_staged_rgb_stock( + &wallet_dir, + "persist-operation".to_owned(), + Stock::in_memory(), + None, + ) + .unwrap(); + } + "promote" => { + let mut journal = read_rgb_acceptance_journal(&wallet_dir).unwrap().unwrap(); + promote_staged_rgb_stock_filesystem(&wallet_dir, &mut journal).unwrap(); + } + "rollback" => { + let journal = read_rgb_acceptance_journal(&wallet_dir).unwrap().unwrap(); + rollback_rgb_acceptance(&wallet_dir, &journal).unwrap(); + } + "finalize" => { + let journal = read_rgb_acceptance_journal(&wallet_dir).unwrap().unwrap(); + finalize_rgb_acceptance(&wallet_dir, &journal).unwrap(); + } + "finalize-metadata" => { + let wallet = metadata_test_wallet(&wallet_dir); + wallet + .resolve_pending_rgb_acceptance( + "metadata-finalize-operation", + RgbAcceptanceResolution::Finalize, + ) + .unwrap(); + } + _ => panic!("unknown RGB acceptance crash mode"), + } + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn rgb_acceptance_os_kill_matrix() { + kill_at_each_checkpoint( + "persist", + |wallet_dir| write_marker(&wallet_dir.join(crate::utils::RGB_RUNTIME_DIR), "old"), + |wallet_dir| { + if let Some(journal) = read_rgb_acceptance_journal(wallet_dir).unwrap() { + rollback_rgb_acceptance(wallet_dir, &journal).unwrap(); + } else { + cleanup_orphaned_rgb_acceptance_artifacts(wallet_dir).unwrap(); + } + assert_eq!( + read_marker(&wallet_dir.join(crate::utils::RGB_RUNTIME_DIR)), + "old" + ); + }, + ); + kill_at_each_checkpoint( + "promote", + |wallet_dir| { + stock_fixture( + wallet_dir, + "promote-operation", + RgbAcceptanceJournalPhase::Prepared, + ) + }, + |wallet_dir| { + let journal = + acceptance_journal("promote-operation", RgbAcceptanceJournalPhase::Promoting); + rollback_rgb_acceptance(wallet_dir, &journal).unwrap(); + assert_eq!( + read_marker(&wallet_dir.join(crate::utils::RGB_RUNTIME_DIR)), + "old" + ); + }, + ); + kill_at_each_checkpoint( + "rollback", + |wallet_dir| { + let journal = + acceptance_journal("rollback-operation", RgbAcceptanceJournalPhase::Promoted); + write_marker(&wallet_dir.join(crate::utils::RGB_RUNTIME_DIR), "new"); + write_marker(&wallet_dir.join(&journal.backup_dir_name), "old"); + write_rgb_acceptance_journal(wallet_dir, &journal).unwrap(); + }, + |wallet_dir| { + let journal = + acceptance_journal("rollback-operation", RgbAcceptanceJournalPhase::Promoted); + rollback_rgb_acceptance(wallet_dir, &journal).unwrap(); + assert_eq!( + read_marker(&wallet_dir.join(crate::utils::RGB_RUNTIME_DIR)), + "old" + ); + }, + ); + kill_at_each_checkpoint( + "finalize", + |wallet_dir| { + let journal = + acceptance_journal("finalize-operation", RgbAcceptanceJournalPhase::Promoted); + write_marker(&wallet_dir.join(crate::utils::RGB_RUNTIME_DIR), "new"); + write_marker(&wallet_dir.join(&journal.backup_dir_name), "old"); + write_rgb_acceptance_journal(wallet_dir, &journal).unwrap(); + }, + |wallet_dir| { + let journal = + acceptance_journal("finalize-operation", RgbAcceptanceJournalPhase::Promoted); + finalize_rgb_acceptance(wallet_dir, &journal).unwrap(); + assert_eq!( + read_marker(&wallet_dir.join(crate::utils::RGB_RUNTIME_DIR)), + "new" + ); + }, + ); + kill_at_each_checkpoint( + "finalize-metadata", + metadata_finalize_fixture, + recover_and_assert_metadata_finalize, + ); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn rgb_acceptance_trace_covers_owned_persistence_syscalls() { + let persist_trace = checkpoint_trace("persist", |wallet_dir| { + write_marker(&wallet_dir.join(crate::utils::RGB_RUNTIME_DIR), "old") + }); + for checkpoint in persist_trace + .iter() + .filter_map(|checkpoint| checkpoint.strip_prefix("before-")) + { + assert!( + persist_trace.contains(&format!("after-{checkpoint}")), + "persistence checkpoint '{checkpoint}' has no post-syscall boundary" + ); + } + for object in ["stash", "state", "index"] { + for operation in ["create", "sync"] { + assert!( + persist_trace.contains(&format!("before-stage-store-{object}-{operation}")) + ); + assert!(persist_trace.contains(&format!("after-stage-store-{object}-{operation}"))); + } + assert!(persist_trace.iter().any(|checkpoint| { + checkpoint.starts_with(&format!("before-stage-store-{object}-write-")) + })); + } + for operation in ["create", "sync"] { + assert!(persist_trace.contains(&format!("before-journal-temp-{operation}"))); + assert!(persist_trace.contains(&format!("after-journal-temp-{operation}"))); + } + assert!( + persist_trace + .iter() + .any(|checkpoint| checkpoint.starts_with("before-journal-temp-write-")) + ); + + let promotion_trace = checkpoint_trace("promote", |wallet_dir| { + stock_fixture( + wallet_dir, + "promote-operation", + RgbAcceptanceJournalPhase::Prepared, + ) + }); + for checkpoint in [ + "before-promote-live-to-backup", + "after-promote-live-to-backup", + "before-promote-sync-backup-installed-sync", + "after-promote-sync-backup-installed-sync", + "before-promote-stage-to-live", + "after-promote-stage-to-live", + "before-promote-sync-installed-live-sync", + "after-promote-sync-installed-live-sync", + ] { + assert!( + promotion_trace.iter().any(|entry| entry == checkpoint), + "promotion trace is missing '{checkpoint}'" + ); + } + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn acceptance_store_is_upstream_fs_bin_store_compatible() { + let directory = tempfile::tempdir().unwrap(); + write_marker(&directory.path().join(crate::utils::RGB_RUNTIME_DIR), "old"); + let journal = persist_staged_rgb_stock( + directory.path(), + "compatibility-operation".to_owned(), + Stock::in_memory(), + None, + ) + .unwrap(); + let provider = FsBinStore::new(directory.path().join(&journal.stage_dir_name)).unwrap(); + let _: Stock = + Stock::load(provider, false).expect("checkpointed stock bytes load through FsBinStore"); + rollback_rgb_acceptance(directory.path(), &journal).unwrap(); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn live_runtime_is_gated_until_acceptance_is_reconciled() { + let directory = tempfile::tempdir().unwrap(); + let journal = + acceptance_journal("exclusive-operation", RgbAcceptanceJournalPhase::Promoted); + write_rgb_acceptance_journal(directory.path(), &journal).unwrap(); + + assert_matches!( + crate::utils::load_rgb_runtime(directory.path()), + Err(Error::RgbOperationInProgress { operation_id }) + if operation_id == "exclusive-operation" + ); + + fs::remove_file(rgb_acceptance_journal_path(directory.path())).unwrap(); + assert!(crate::utils::load_rgb_runtime(directory.path()).is_ok()); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn protocol_operation_runtime_access_requires_exact_promoted_owner() { + let directory = tempfile::tempdir().unwrap(); + + drop( + crate::utils::load_rgb_runtime_for_operation(directory.path(), "normal-operation") + .unwrap(), + ); + + let promoted = acceptance_journal("funding-operation", RgbAcceptanceJournalPhase::Promoted); + write_rgb_acceptance_journal(directory.path(), &promoted).unwrap(); + + assert_matches!( + crate::utils::load_rgb_runtime_for_operation(directory.path(), "foreign-operation"), + Err(Error::RgbOperationInProgress { operation_id }) + if operation_id == "funding-operation" + ); + assert_matches!( + crate::utils::load_rgb_runtime(directory.path()), + Err(Error::RgbOperationInProgress { operation_id }) + if operation_id == "funding-operation" + ); + drop( + crate::utils::load_rgb_runtime_for_operation(directory.path(), "funding-operation") + .unwrap(), + ); + + let prepared = acceptance_journal("funding-operation", RgbAcceptanceJournalPhase::Prepared); + write_rgb_acceptance_journal(directory.path(), &prepared).unwrap(); + assert_matches!( + crate::utils::load_rgb_runtime_for_operation(directory.path(), "funding-operation"), + Err(Error::RgbOperationInProgress { operation_id }) + if operation_id == "funding-operation" + ); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn protocol_operation_waiter_rechecks_owner_after_acquiring_lock() { + let directory = tempfile::tempdir().unwrap(); + let wallet_dir = directory.path().to_path_buf(); + let mut active_runtime = crate::utils::load_rgb_runtime(&wallet_dir).unwrap(); + active_runtime.suppress_persistence(); + let journal = acceptance_journal( + "contended-funding-operation", + RgbAcceptanceJournalPhase::Promoted, + ); + write_rgb_acceptance_journal(&wallet_dir, &journal).unwrap(); + + let (started_tx, started_rx) = std::sync::mpsc::channel(); + let waiting_wallet_dir = wallet_dir.clone(); + let waiter = std::thread::spawn(move || { + started_tx.send(()).unwrap(); + crate::utils::load_rgb_runtime_for_operation( + waiting_wallet_dir, + "contended-funding-operation", + ) + }); + started_rx.recv().unwrap(); + std::thread::sleep(Duration::from_millis(50)); + assert!( + !waiter.is_finished(), + "protocol operation did not wait for the stock lock" + ); + + drop(active_runtime); + drop(waiter.join().unwrap().unwrap()); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn runtime_waiter_rechecks_acceptance_journal_after_acquiring_lock() { + let directory = tempfile::tempdir().unwrap(); + let wallet_dir = directory.path().to_path_buf(); + let mut active_runtime = crate::utils::load_rgb_runtime(&wallet_dir).unwrap(); + active_runtime.suppress_persistence(); + let journal = + acceptance_journal("contended-operation", RgbAcceptanceJournalPhase::Promoted); + write_rgb_acceptance_journal(&wallet_dir, &journal).unwrap(); + + let (started_tx, started_rx) = std::sync::mpsc::channel(); + let waiting_wallet_dir = wallet_dir.clone(); + let waiter = std::thread::spawn(move || { + started_tx.send(()).unwrap(); + crate::utils::load_rgb_runtime(waiting_wallet_dir) + }); + started_rx.recv().unwrap(); + std::thread::sleep(Duration::from_millis(50)); + assert!( + !waiter.is_finished(), + "second runtime did not wait for the lock" + ); + + drop(active_runtime); + let waiter_result = waiter.join().unwrap(); + assert_matches!( + waiter_result, + Err(Error::RgbOperationInProgress { operation_id }) + if operation_id == "contended-operation" + ); + + finalize_rgb_acceptance(&wallet_dir, &journal).unwrap(); + assert!(crate::utils::load_rgb_runtime(wallet_dir).is_ok()); + } #[cfg(any(feature = "electrum", feature = "esplora"))] #[test] @@ -1002,4 +2708,80 @@ mod tests { assert_eq!(IndexerProtocol::Electrum.to_string(), "Electrum"); assert_eq!(IndexerProtocol::Esplora.to_string(), "Esplora"); } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn prepared_acceptance_rollback_leaves_live_stock_unchanged() { + let directory = tempfile::tempdir().unwrap(); + let wallet_dir = directory.path(); + let journal = acceptance_journal("prepared", RgbAcceptanceJournalPhase::Prepared); + let live_dir = wallet_dir.join(crate::utils::RGB_RUNTIME_DIR); + let stage_dir = wallet_dir.join(&journal.stage_dir_name); + write_marker(&live_dir, "old"); + write_marker(&stage_dir, "new"); + write_rgb_acceptance_journal(wallet_dir, &journal).unwrap(); + + rollback_rgb_acceptance(wallet_dir, &journal).unwrap(); + + assert_eq!(read_marker(&live_dir), "old"); + assert!(!stage_dir.exists()); + assert!(!rgb_acceptance_journal_path(wallet_dir).exists()); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn interrupted_promotion_before_live_install_restores_backup() { + let directory = tempfile::tempdir().unwrap(); + let wallet_dir = directory.path(); + let journal = acceptance_journal("before-install", RgbAcceptanceJournalPhase::Promoting); + let live_dir = wallet_dir.join(crate::utils::RGB_RUNTIME_DIR); + let stage_dir = wallet_dir.join(&journal.stage_dir_name); + let backup_dir = wallet_dir.join(&journal.backup_dir_name); + write_marker(&backup_dir, "old"); + write_marker(&stage_dir, "new"); + write_rgb_acceptance_journal(wallet_dir, &journal).unwrap(); + + rollback_rgb_acceptance(wallet_dir, &journal).unwrap(); + + assert_eq!(read_marker(&live_dir), "old"); + assert!(!stage_dir.exists()); + assert!(!backup_dir.exists()); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn interrupted_promotion_after_live_install_restores_backup() { + let directory = tempfile::tempdir().unwrap(); + let wallet_dir = directory.path(); + let journal = acceptance_journal("after-install", RgbAcceptanceJournalPhase::Promoting); + let live_dir = wallet_dir.join(crate::utils::RGB_RUNTIME_DIR); + let backup_dir = wallet_dir.join(&journal.backup_dir_name); + write_marker(&backup_dir, "old"); + write_marker(&live_dir, "new"); + write_rgb_acceptance_journal(wallet_dir, &journal).unwrap(); + + rollback_rgb_acceptance(wallet_dir, &journal).unwrap(); + + assert_eq!(read_marker(&live_dir), "old"); + assert!(!backup_dir.exists()); + } + + #[cfg(any(feature = "electrum", feature = "esplora"))] + #[test] + fn finalized_promotion_keeps_new_stock_and_removes_rollback_state() { + let directory = tempfile::tempdir().unwrap(); + let wallet_dir = directory.path(); + let journal = acceptance_journal("finalize", RgbAcceptanceJournalPhase::Promoted); + let live_dir = wallet_dir.join(crate::utils::RGB_RUNTIME_DIR); + let backup_dir = wallet_dir.join(&journal.backup_dir_name); + write_marker(&backup_dir, "old"); + write_marker(&live_dir, "new"); + write_rgb_acceptance_journal(wallet_dir, &journal).unwrap(); + + finalize_rgb_acceptance(wallet_dir, &journal).unwrap(); + + assert_eq!(read_marker(&live_dir), "new"); + assert!(!backup_dir.exists()); + assert!(!rgb_acceptance_journal_path(wallet_dir).exists()); + } } diff --git a/src/wallet/singlesig.rs b/src/wallet/singlesig.rs index ebc44711..d940e80f 100644 --- a/src/wallet/singlesig.rs +++ b/src/wallet/singlesig.rs @@ -897,7 +897,7 @@ impl Wallet { lock_time, )?; self.sign_psbt_impl(&mut begin_op_data.psbt, None)?; - let res = self.send_end_impl(&txn, &begin_op_data.psbt)?; + let res = self.send_end_impl(&txn, &begin_op_data.psbt, true, None)?; self.update_backup_info(&txn, false)?; txn.commit()?; self.trigger_auto_backup(); @@ -1012,7 +1012,7 @@ impl Wallet { self.check_online(online)?; let psbt = Psbt::from_str(&signed_psbt)?; let txn = self.database().begin_transaction()?; - let res = self.send_end_impl(&txn, &psbt)?; + let res = self.send_end_impl(&txn, &psbt, true, None)?; self.update_backup_info(&txn, false)?; txn.commit()?; self.trigger_auto_backup(); @@ -1091,6 +1091,70 @@ impl Wallet { Ok(res) } + /// Completes a send whose fascia was already promoted by a protocol-controlled transaction. + /// + ///
The caller must finalize or roll back the matching RGB stock journal. + /// This method is intended only for protocols that require the proposed allocation to be + /// visible before the Bitcoin transaction can be broadcast.
+ pub fn send_end_preconsumed( + &mut self, + online: Online, + signed_psbt: String, + ) -> Result { + info!(self.logger(), "Sending preconsumed transfer (end)..."); + self.check_online(online)?; + let psbt = Psbt::from_str(&signed_psbt)?; + let txn = self.database().begin_transaction()?; + let res = self.send_end_impl(&txn, &psbt, false, None)?; + self.update_backup_info(&txn, false)?; + txn.commit()?; + self.trigger_auto_backup(); + info!(self.logger(), "Send preconsumed transfer (end) completed"); + Ok(res) + } + + /// Completes a preconsumed send while owning a matching durable RGB stock operation. + /// + /// If a promoted RGB acceptance is pending, `operation_id` must exactly match its journal. The + /// RGB stock lock is retained while consignments are generated and the transfer is committed, + /// preventing recovery from finalizing or rolling back the stock concurrently. + /// + ///
This method is intended only for protocol integrations that provide + /// their own durable commit decision.
+ pub fn send_end_preconsumed_for_operation( + &mut self, + online: Online, + operation_id: &str, + signed_psbt: String, + ) -> Result { + if operation_id.is_empty() { + return Err(Error::Internal { + details: s!("RGB protocol operation ID cannot be empty"), + }); + } + info!( + self.logger(), + "Sending preconsumed transfer for RGB protocol operation (end)..." + ); + self.check_online(online)?; + let psbt = Psbt::from_str(&signed_psbt)?; + if psbt.unsigned_tx.compute_txid().to_string() != operation_id { + return Err(Error::Internal { + details: s!("RGB protocol operation ID does not match the send transaction"), + }); + } + let txn = self.database().begin_transaction()?; + let res = self.send_end_impl(&txn, &psbt, false, Some(operation_id))?; + self.update_backup_info(&txn, false)?; + txn.commit()?; + self.trigger_auto_backup(); + info!( + self.logger(), + "Send preconsumed transfer for RGB protocol operation completed" + ); + Ok(res) + } + /// Send bitcoins using the vanilla wallet. /// /// This calls [`send_btc_begin`](Wallet::send_btc_begin), signs the resulting PSBT and finally diff --git a/src/wallet/test/new.rs b/src/wallet/test/new.rs index e691e969..c0345aa6 100644 --- a/src/wallet/test/new.rs +++ b/src/wallet/test/new.rs @@ -180,9 +180,9 @@ fn mainnet_esplora_success() { ); check_wallet(&party, bitcoin_network, None); - // UTEXO Mainnet Esplora (Hetzner) - let indexer_url = "https://esplora-mainnet.utexo.com"; - party.go_online(false, Some(indexer_url)); + let indexer_url = std::env::var("RGB_LIB_TEST_MAINNET_ESPLORA_URL") + .unwrap_or_else(|_| s!("https://blockstream.info/api")); + party.go_online(false, Some(&indexer_url)); assert!(!party.wallet.watch_only()); assert_eq!(party.get_wallet_data().bitcoin_network, bitcoin_network); } @@ -218,38 +218,6 @@ fn mainnet_success_electrum() { assert_eq!(party.get_wallet_data().bitcoin_network, bitcoin_network); } -#[cfg(feature = "esplora")] -#[test] -#[ignore = "frequently fails due to timeout"] -#[parallel] -fn mainnet_success_esplora() { - create_test_data_dir(); - - let bitcoin_network = BitcoinNetwork::Mainnet; - let keys = generate_keys(bitcoin_network, WitnessVersion::Taproot); - let mut party = offline_party!( - Wallet::new( - WalletData { - data_dir: get_test_data_dir_string(), - bitcoin_network, - database_type: DatabaseType::Sqlite, - max_allocations_per_utxo: MAX_ALLOCATIONS_PER_UTXO, - // IFA not supported on mainnet - supported_schemas: vec![AssetSchema::Cfa, AssetSchema::Nia, AssetSchema::Uda], - reuse_addresses: false, - }, - SinglesigKeys::from_keys(&keys, None), - ) - .unwrap() - ); - - check_wallet(&party, bitcoin_network, None); - let indexer_url = "https://blockstream.info/api"; - party.go_online(false, Some(indexer_url)); - assert!(!party.wallet.watch_only()); - assert_eq!(party.get_wallet_data().bitcoin_network, bitcoin_network); -} - #[test] #[parallel] fn fail() { diff --git a/src/wallet/test/rust_only.rs b/src/wallet/test/rust_only.rs index 62bc45fd..60c9deba 100644 --- a/src/wallet/test/rust_only.rs +++ b/src/wallet/test/rust_only.rs @@ -132,18 +132,49 @@ fn success() { .unwrap() .save_file(&consignment_path) .unwrap(); - - // accept transfer - recv_party + // recover and accept from the exact persisted consignment without another proxy fetch + let consignment_bytes = std::fs::read(&consignment_path).unwrap(); + assert!( + !recv_party + .wallet + .has_accepted_transfer(asset.asset_id.clone(), txid.clone()) + .unwrap() + ); + let prepared = recv_party .wallet - .accept_transfer_consignment( - recv_party.online, - consignment_path, + .prepare_accept_transfer_from_consignment( + txid.clone(), txid.clone(), vout, + consignment_bytes, blinding, ) .unwrap(); + assert_eq!( + prepared.consignment().contract_id().to_string(), + asset.asset_id + ); + assert!(prepared.media_digests().is_empty()); + let pending = recv_party.wallet.pending_rgb_acceptance().unwrap().unwrap(); + assert_eq!(pending.operation_id(), txid); + assert!(!pending.promoted()); + prepared.promote().unwrap(); + assert_matches!( + recv_party + .wallet + .has_accepted_transfer(asset.asset_id.clone(), txid.clone()), + Err(Error::RgbOperationInProgress { operation_id }) if operation_id == txid + ); + recv_party + .wallet + .resolve_pending_rgb_acceptance(&txid, RgbAcceptanceResolution::Finalize) + .unwrap(); + assert!( + recv_party + .wallet + .has_accepted_transfer(asset.asset_id, txid) + .unwrap() + ); // consume fascia party_send.wallet.consume_fascia(fascia, None).unwrap(); @@ -1067,6 +1098,106 @@ fn send_end_db_update_only_success() { ); } +#[cfg(feature = "electrum")] +#[test] +#[parallel] +fn send_end_db_update_only_for_operation_success() { + initialize(); + + let mut party = get_funded_party!(); + let mut rcv_party = get_funded_party!(); + + let asset = party.issue_asset_nia(None); + let receive_data = rcv_party.blind_receive_asset_expiry(None, None); + let recipient_map = HashMap::from([( + asset.asset_id, + vec![Recipient { + assignment: Assignment::Fungible(10), + recipient_id: receive_data.recipient_id, + witness_data: None, + transport_endpoints: TRANSPORT_ENDPOINTS.clone(), + }], + )]); + let begin = party + .wallet + .send_begin( + party.online, + recipient_map, + true, + FEE_RATE, + MIN_CONFIRMATIONS, + default_send_expiration(), + false, + None, + ) + .unwrap(); + let signed_psbt = party.wallet.sign_psbt(begin.psbt.clone(), None).unwrap(); + party.wallet.create_consignments(begin.psbt).unwrap(); + let operation_id = Psbt::from_str(&signed_psbt) + .unwrap() + .unsigned_tx + .compute_txid() + .to_string(); + let fascia: Fascia = + serde_json::from_str(&fs::read_to_string(&begin.details.fascia_path).unwrap()).unwrap(); + party + .wallet + .prepare_consume_fascia(operation_id.clone(), fascia, None) + .unwrap() + .promote() + .unwrap(); + let witness_id = RgbTxid::from_str(&operation_id).unwrap(); + + assert_matches!( + party + .wallet + .send_end_db_update_only(party.online, signed_psbt.clone()), + Err(Error::RgbOperationInProgress { operation_id: owner }) if owner == operation_id + ); + assert_matches!( + party + .wallet + .upsert_witness(witness_id, WitnessOrd::Tentative), + Err(Error::RgbOperationInProgress { operation_id: owner }) if owner == operation_id + ); + assert_matches!( + party.wallet.upsert_witness_for_operation( + "different-operation", + witness_id, + WitnessOrd::Tentative, + ), + Err(Error::RgbOperationInProgress { operation_id: owner }) if owner == operation_id + ); + party + .wallet + .upsert_witness_for_operation(&operation_id, witness_id, WitnessOrd::Tentative) + .unwrap(); + assert_matches!( + party.wallet.send_end_db_update_only_for_operation( + party.online, + "different-operation", + signed_psbt.clone(), + ), + Err(Error::Internal { details }) + if details == "RGB protocol operation ID does not match the send transaction" + ); + + let result = party + .wallet + .send_end_db_update_only_for_operation(party.online, &operation_id, signed_psbt) + .unwrap(); + assert_eq!(result.txid, operation_id); + assert_eq!(result.batch_transfer_idx, begin.batch_transfer_idx.unwrap()); + party + .wallet + .resolve_pending_rgb_acceptance(&operation_id, RgbAcceptanceResolution::Finalize) + .unwrap(); + assert!( + party + .check_test_transfer_status_sender(&operation_id, TransferStatus::WaitingConfirmations,) + ); +} + #[cfg(feature = "electrum")] #[test] #[parallel] diff --git a/src/wallet/test/send.rs b/src/wallet/test/send.rs index 5c07a271..82f25a3f 100644 --- a/src/wallet/test/send.rs +++ b/src/wallet/test/send.rs @@ -7855,7 +7855,6 @@ fn offline_receiver_blind_restart_waiting_counterparty() { TransferStatus::WaitingConfirmations ) ); - drop(mining_guard); mine(false); rcv_party.wait_for_refresh_raw(None, None); @@ -8083,7 +8082,12 @@ fn offline_receiver_witness_restart_donation_true() { rcv_party.wait_for_asset_balance(&asset.asset_id, &waiting_balance); party.refresh_all(); - assert!(party.check_test_transfer_status_sender(&txid, TransferStatus::WaitingConfirmations)); + let waiting_for_confirmations = + party.check_test_transfer_status_sender(&txid, TransferStatus::WaitingConfirmations); + assert!( + waiting_for_confirmations + || party.check_test_transfer_status_sender(&txid, TransferStatus::Settled) + ); drop(mining_guard); mine(false); @@ -8625,7 +8629,6 @@ fn offline_receiver_mixed_blind_witness_batch_donation_false() { TransferStatus::WaitingConfirmations )); witness_party.wait_for_asset_balance(&asset.asset_id, &witness_waiting_balance); - drop(mining_guard); mine(false); blind_party.wait_for_refresh_raw(None, None);