diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 8771c06..087a649 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -67,6 +67,9 @@ jobs: timeout-minutes: 120 steps: - uses: actions/checkout@v4 + with: + # The upgrade generator compiles the adapter from a pinned old commit. + fetch-depth: 0 - uses: dtolnay/rust-toolchain@stable - name: Install Protoc @@ -76,6 +79,11 @@ jobs: repo-token: ${{ secrets.GITHUB_TOKEN }} - uses: Swatinem/rust-cache@v2.7.8 + with: + # Include the standalone old-version generator's dependency tree. + workspaces: | + . -> target + orange-sdk/tests/fixtures/cdk016-generator -> target - name: Cache package-enabled Esplora id: cache-esplora @@ -94,3 +102,8 @@ jobs: - name: Run cashu tests run: cargo test --features _cashu-tests -- --test-threads=2 --nocapture + + - name: Test funded Cashu upgrades from CDK 0.16 + env: + TMPDIR: ${{ runner.temp }} + run: bash ci/test-cashu-upgrade.sh diff --git a/README.md b/README.md index 9973476..1cd6eaa 100644 --- a/README.md +++ b/README.md @@ -141,6 +141,38 @@ The wallet's behavior is controlled by `Tunables`: - **Spark** - Custodial Lightning wallet (default) - **Cashu** - Ecash-based wallet +### Upgrading Cashu wallets from CDK 0.16 to 0.18.3 + +Orange stores Cashu wallet records in its own KV namespaces within `orange.sqlite` +(or the configured remote store), rather than CDK's SQL wallet tables. On startup, +it automatically rekeys existing saga transactions to CDK's new transaction IDs +before wallet recovery runs. The migration copies before deleting, resumes after +interruption, and rejects conflicting records without overwriting them. Orange's +public payment IDs remain based on quote IDs. + +Existing proofs, quotes, keysets, and in-flight sagas remain readable. CDK supplies +serde defaults for the new fields and accepts the old `payment_preimage` field. +Existing transactions default to completed; new transactions preserve their +pending/completed/failed status. A separate persistent derivation counter is +created as needed, leaving existing keyset counters and quote signing keys intact. +Legacy proof reservation indexes remain readable during recovery. + +Startup runs saga recovery before starting payment tasks, on every open (including +wallets already restored from seed). Database initialization errors or an unreadable +recovery queue stop initialization. Individual failed recoveries are logged and +startup continues, so a mint outage does not block access to the rest of the wallet. +Unresolved operations retain their reservations and are retried on the next startup. + +Back up the complete store before upgrading and open it with only one active +wallet instance. Downgrading an upgraded store is not supported; restore the +pre-upgrade backup if a rollback is needed. The test mint uses an ephemeral CDK +SQLite database; Orange does not manage persistent mint databases or `cdk-mintd` +configuration. Operators of a separate mint must follow the upstream +[CDK migration guide](https://github.com/cashubtc/cdk/blob/v0.18.2/docs/migrations/v0.18.md). + +See the [CDK 0.18.2 release notes](https://github.com/cashubtc/cdk/releases/tag/v0.18.2) +and the upstream [transaction rekey migration](https://github.com/cashubtc/cdk/blob/v0.18.2/crates/cdk-sql-common/src/wallet/migrations/sqlite/20260726000000_rekey_saga_transactions.sql). + ### Chain Sources - Electrum servers - Esplora servers (with optional Basic auth) diff --git a/ci/test-cashu-upgrade.sh b/ci/test-cashu-upgrade.sh new file mode 100644 index 0000000..5815127 --- /dev/null +++ b/ci/test-cashu-upgrade.sh @@ -0,0 +1,43 @@ +#!/usr/bin/env bash +set -euo pipefail + +cd "$(dirname "$0")/.." +scratch=$(mktemp -d "${TMPDIR:-/tmp}/orange-cdk016.XXXXXXXX") +trap 'rm -rf "$scratch"' EXIT + +# Always compile the old adapter, never the adapter under test. CI must fetch +# history containing this revision (actions/checkout with fetch-depth: 0). +old_revision=a3aa1c592625c45314004987430498307b9a1819 +for source in cashu_store dyn_store; do + if [[ "$source" == cashu_store ]]; then + source_path=orange-sdk/src/trusted_wallet/cashu/cashu_store.rs + else + source_path=orange-sdk/src/dyn_store.rs + fi + git show "$old_revision:$source_path" > "$scratch/$source.rs" +done +# include! does not accept inner module documentation comments. +sed -i '/^\/\/!/d' "$scratch/dyn_store.rs" +export ORANGE_OLD_CASHU_STORE="$scratch/cashu_store.rs" +export ORANGE_OLD_DYN_STORE="$scratch/dyn_store.rs" + +cargo build --manifest-path orange-sdk/tests/fixtures/cdk016-generator/Cargo.toml \ + --message-format=json > "$scratch/build.json" +# Cargo reports the actual executable, honoring configured target directories. +ORANGE_CDK016_GENERATOR=$(python3 - "$scratch/build.json" <<'PY' +import json +import sys + +with open(sys.argv[1]) as output: + binaries = [message['executable'] for line in output + if (message := json.loads(line)).get('reason') == 'compiler-artifact' + and message.get('executable') + and message['target']['name'] == 'orange-cdk016-fixture'] +assert len(binaries) == 1, f'Expected one fixture generator, got {binaries}' +print(binaries[0]) +PY +) +export ORANGE_CDK016_GENERATOR +test -x "$ORANGE_CDK016_GENERATOR" +cargo test -p orange-sdk --no-default-features --features _cashu-tests \ + --lib cdk016_funded_wallet -- --ignored --nocapture --test-threads=2 diff --git a/orange-sdk/Cargo.toml b/orange-sdk/Cargo.toml index 7996cac..82901cc 100644 --- a/orange-sdk/Cargo.toml +++ b/orange-sdk/Cargo.toml @@ -16,9 +16,9 @@ name = "orange_sdk" default = ["spark"] uniffi = ["dep:uniffi", "spark", "cashu", "rand", "pin-project-lite"] spark = ["breez-sdk-spark", "uuid", "serde_json"] -cashu = ["cdk", "cdk/npubcash", "serde", "serde_json"] +cashu = ["cdk", "cdk/npubcash", "serde", "serde_json", "uuid"] _test-utils = ["corepc-node", 'electrsd', "cashu", "uuid/v7", "rand"] -_cashu-tests = ["_test-utils", "cdk-ldk-node", "cdk/mint", "cdk-sqlite", "cdk-axum", "axum"] +_cashu-tests = ["_test-utils", "cdk-ldk-node", "cdk/mint", "cdk-sqlite", "cdk-axum", "cdk-fake-wallet", "axum"] [dependencies] graduated-rebalancer = { path = "../graduated-rebalancer", version = "0.1.0" } @@ -31,7 +31,7 @@ rand = { version = "0.8.5", optional = true } breez-sdk-spark = { git = "https://github.com/breez/spark-sdk.git", rev = "512453108958a7e31c870b30c8cc67d44fa3607b", default-features = false, optional = true } tokio = { version = "1.0", default-features = false, features = ["rt-multi-thread", "sync", "macros"] } uuid = { version = "1.0", default-features = false, optional = true } -cdk = { version = "0.16.0", default-features = false, features = ["wallet"], optional = true } +cdk = { version = "0.18.3", default-features = false, features = ["wallet"], optional = true } serde = { version = "1.0", features = ["derive"], optional = true } serde_json = { version = "1.0", optional = true } async-trait = "0.1" @@ -39,13 +39,16 @@ log = "0.4.28" corepc-node = { version = "0.10.1", features = ["29_0", "download"], optional = true } electrsd = { version = "0.36.1", default-features = false, features = ["legacy", "corepc-node_29_0"], optional = true } -cdk-ldk-node = { version = "0.16.0", optional = true } -cdk-sqlite = { version = "0.16.0", optional = true } -cdk-axum = { version = "0.16.0", optional = true } +cdk-ldk-node = { version = "0.18.3", optional = true } +cdk-sqlite = { version = "0.18.3", optional = true } +cdk-fake-wallet = { version = "0.18.3", optional = true } +cdk-axum = { version = "0.18.3", optional = true } axum = { version = "0.8.1", optional = true } uniffi = { version = "0.29", default-features = false, features = ["cli"], optional = true } pin-project-lite = { version = "0.2", default-features = false, optional = true } [dev-dependencies] +tempfile = "3" +cdk-common = { version = "=0.18.3", default-features = false, features = ["wallet", "test"] } test-log = "0.2.18" diff --git a/orange-sdk/src/ffi/cashu.rs b/orange-sdk/src/ffi/cashu.rs index 68c195b..ead7558 100644 --- a/orange-sdk/src/ffi/cashu.rs +++ b/orange-sdk/src/ffi/cashu.rs @@ -19,7 +19,7 @@ impl From for CurrencyUnit { CdkCurrencyUnit::Usd => CurrencyUnit::Usd, CdkCurrencyUnit::Eur => CurrencyUnit::Eur, CdkCurrencyUnit::Auth => CurrencyUnit::Auth, - CdkCurrencyUnit::Custom(unit) => CurrencyUnit::Custom(unit), + CdkCurrencyUnit::Custom(unit) => CurrencyUnit::Custom(unit.to_string()), _ => CurrencyUnit::Sat, } } @@ -33,7 +33,7 @@ impl From for CdkCurrencyUnit { CurrencyUnit::Usd => CdkCurrencyUnit::Usd, CurrencyUnit::Eur => CdkCurrencyUnit::Eur, CurrencyUnit::Auth => CdkCurrencyUnit::Auth, - CurrencyUnit::Custom(unit) => CdkCurrencyUnit::Custom(unit), + CurrencyUnit::Custom(unit) => CdkCurrencyUnit::Custom(unit.into()), } } } diff --git a/orange-sdk/src/rebalancer.rs b/orange-sdk/src/rebalancer.rs index 0db8cfc..8b9a3b9 100644 --- a/orange-sdk/src/rebalancer.rs +++ b/orange-sdk/src/rebalancer.rs @@ -5,7 +5,7 @@ use crate::lightning_wallet::LightningWallet; use crate::logging::Logger; use crate::store::{PaymentId, TxMetadata, TxMetadataStore, TxType}; use crate::trusted_wallet::DynTrustedWalletInterface; -use crate::{Event, EventQueue, PaymentType, Tunables}; +use crate::{Event, EventQueue, PaymentType, Tunables, TxStatus}; use bitcoin_payment_instructions::amount::Amount; use graduated_rebalancer::{RebalanceTrigger, RebalancerEvent, TriggerParams}; use ldk_node::lightning::util::logger::Logger as _; @@ -74,6 +74,10 @@ impl RebalanceTrigger for OrangeTrigger { // TODO: Maybe use this to backfill stuff we lost on crash? continue; } + // Only settled receipts can be recorded or selected as rebalance triggers. + if payment.status != TxStatus::Completed { + continue; + } let payment_id = PaymentId::Trusted(payment.id); let have_metadata = if let Some(metadata) = self.tx_metadata.read().get(&payment_id) diff --git a/orange-sdk/src/trusted_wallet/cashu/cashu_store.rs b/orange-sdk/src/trusted_wallet/cashu/cashu_store.rs index 18ad8b8..2f2dbd3 100644 --- a/orange-sdk/src/trusted_wallet/cashu/cashu_store.rs +++ b/orange-sdk/src/trusted_wallet/cashu/cashu_store.rs @@ -1,6 +1,7 @@ use std::collections::{HashMap, HashSet}; use std::fmt::Debug; use std::str::FromStr; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, RwLock}; use crate::dyn_store::{DynStore, read_keys_bounded}; @@ -37,6 +38,7 @@ const KEYS_KEY: &str = "keys"; const PROOFS_KEY: &str = "proofs"; const PROOFS_BLOB_KEY: &str = "proof_set"; const PROOFS_SCHEMA_VERSION: u8 = 1; +const DERIVATION_COUNTERS_KEY: &str = "derivation_counters"; const KEYSET_COUNTERS_KEY: &str = "keyset_counters"; const TRANSACTIONS_KEY: &str = "transactions"; const KEYSETS_TABLE_KEY: &str = "keysets_table"; @@ -130,9 +132,11 @@ impl From for cdk::cdk_database::Error { pub struct CashuKvDatabase { store: Arc, keyset_counter_lock: Mutex<()>, + records_mutation_lock: Mutex<()>, // This only serializes writers in this process. The whole-snapshot storage // format requires a single active wallet instance for a given store. proofs_mutation_lock: Mutex<()>, + proofs_cache_needs_reload: AtomicBool, // In-memory caches for frequently accessed data mints_cache: Arc>>>, proofs_cache: Arc>>, @@ -168,17 +172,100 @@ impl CashuKvDatabase { let database = Self { store, keyset_counter_lock: Mutex::new(()), + records_mutation_lock: Mutex::new(()), proofs_mutation_lock: Mutex::new(()), + proofs_cache_needs_reload: AtomicBool::new(false), mints_cache: Arc::new(RwLock::new(HashMap::new())), proofs_cache: Arc::new(RwLock::new(Vec::new())), }; + database.migrate_transaction_ids().await?; + // Initialize caches from persistent storage database.load_caches().await?; Ok(database) } + // CDK 0.18 identifies saga transactions by saga ID instead of proof Ys. + // Run before exposing the database to recovery. Copy before deleting so an + // interrupted migration can resume; never overwrite a conflicting record. + async fn migrate_transaction_ids(&self) -> Result<(), DatabaseError> { + const MIGRATION_KEY: &str = "cdk_018_transaction_ids"; + match KVStore::read(self.store.as_ref(), CASHU_PRIMARY_KEY, "", MIGRATION_KEY).await { + Ok(data) if data == [1] => return Ok(()), + Ok(_) => return Err(DatabaseError::InvalidFormat), + Err(e) if e.kind() == io::ErrorKind::NotFound => {}, + Err(e) => return Err(e.into()), + } + let keys = KVStore::list(self.store.as_ref(), CASHU_PRIMARY_KEY, TRANSACTIONS_KEY).await?; + if keys.is_empty() { + return Ok(()); + } + for old_key in keys { + let data = + KVStore::read(self.store.as_ref(), CASHU_PRIMARY_KEY, TRANSACTIONS_KEY, &old_key) + .await?; + if data.is_empty() { + continue; + } + let transaction: Transaction = serde_json::from_slice(&data) + .map_err(|e| DatabaseError::Serialization(e.to_string()))?; + let new_key = transaction.id().to_string(); + if old_key == new_key { + continue; + } + if old_key != TransactionId::new(transaction.ys.clone()).to_string() { + return Err(DatabaseError::InvalidFormat); + } + match KVStore::read(self.store.as_ref(), CASHU_PRIMARY_KEY, TRANSACTIONS_KEY, &new_key) + .await + { + Ok(existing) => { + let existing: Transaction = serde_json::from_slice(&existing) + .map_err(|e| DatabaseError::Serialization(e.to_string()))?; + if existing != transaction { + return Err(DatabaseError::Duplicate); + } + }, + Err(e) if e.kind() == io::ErrorKind::NotFound => { + KVStore::write( + self.store.as_ref(), + CASHU_PRIMARY_KEY, + TRANSACTIONS_KEY, + &new_key, + data, + ) + .await?; + }, + Err(e) => return Err(e.into()), + } + KVStore::remove( + self.store.as_ref(), + CASHU_PRIMARY_KEY, + TRANSACTIONS_KEY, + &old_key, + false, + ) + .await?; + } + // Mark complete only after every old key was removed. Remote stores then + // avoid downloading the whole transaction history on each startup. + KVStore::write(self.store.as_ref(), CASHU_PRIMARY_KEY, "", MIGRATION_KEY, vec![1]).await?; + Ok(()) + } + + async fn write_record( + &self, namespace: &str, key: &str, value: &T, + ) -> Result<(), cdk::cdk_database::Error> { + let data = + serde_json::to_vec(value).map_err(|e| DatabaseError::Serialization(e.to_string()))?; + KVStore::write(self.store.as_ref(), CASHU_PRIMARY_KEY, namespace, key, data) + .await + .map_err(DatabaseError::Io)?; + Ok(()) + } + async fn load_caches(&self) -> Result<(), DatabaseError> { // These use independent namespaces and can be restored concurrently. This // lets remote stores pipeline the requests while synchronous stores retain @@ -186,10 +273,7 @@ impl CashuKvDatabase { let (mints, proofs) = tokio::join!(self.load_mints_from_store(), self.load_proofs_from_store()); - if let Ok(mints) = mints { - let mut cache = self.mints_cache.write().unwrap(); - *cache = mints; - } + *self.mints_cache.write().unwrap() = mints?; // Proof errors must not be hidden: presenting an empty cache for an unreadable // proof set could make the wallet report an incorrect balance. @@ -209,6 +293,10 @@ impl CashuKvDatabase { let mut mints = HashMap::with_capacity(keys.len()); for key in keys { + // Mint metadata shares this namespace under info_* keys. + if !key.starts_with("mint_") { + continue; + } let data = KVStore::read(self.store.as_ref(), CASHU_PRIMARY_KEY, MINTS_KEY, &key) .await .map_err(DatabaseError::Io)?; @@ -218,7 +306,7 @@ impl CashuKvDatabase { .map_err(|e| DatabaseError::Serialization(e.to_string()))?; // Try to load mint info - let mint_info = self.load_mint_info(&mint_url).await.ok().flatten(); + let mint_info = self.load_mint_info(&mint_url).await?; mints.insert(mint_url, mint_info); } } @@ -251,12 +339,71 @@ impl CashuKvDatabase { Ok(blob.proofs) } + // Caller holds proofs_mutation_lock. A successful read establishes the + // authoritative snapshot after an error or cancellation during persistence. + async fn reload_uncertain_proofs(&self) -> Result<(), DatabaseError> { + if self.proofs_cache_needs_reload.load(Ordering::Relaxed) { + let proofs = self.load_proofs_from_store().await?; + *self.proofs_cache.write().unwrap() = proofs; + self.proofs_cache_needs_reload.store(false, Ordering::Release); + } + Ok(()) + } + + // Reads can use a committed cache without waiting for the mutation mutex. + // An in-flight or uncertain write still requires synchronization and reload. + async fn read_proofs( + &self, read: impl FnOnce(&[ProofInfo]) -> T, + ) -> Result { + { + let cache = self.proofs_cache.read().unwrap(); + if !self.proofs_cache_needs_reload.load(Ordering::Acquire) { + return Ok(read(&cache)); + } + } + let _lock = self.proofs_mutation_lock.lock().await; + self.reload_uncertain_proofs().await?; + Ok(read(&self.proofs_cache.read().unwrap())) + } + + async fn reserved_proofs( + &self, operation_id: &uuid::Uuid, + ) -> Result, cdk::cdk_database::Error> { + // 0.16's adapter kept ownership in a separate index. Retain that fallback + // for in-flight operations, but explicit ownership always takes precedence. + let legacy: Vec = match KVStore::read( + self.store.as_ref(), + CASHU_PRIMARY_KEY, + PROOF_RESERVATIONS_KEY, + &operation_id.to_string(), + ) + .await + { + Ok(data) => serde_json::from_slice(&data) + .map_err(|e| DatabaseError::Serialization(e.to_string()))?, + Err(e) if e.kind() == io::ErrorKind::NotFound => Vec::new(), + Err(e) => return Err(DatabaseError::Io(e).into()), + }; + let proofs = self.proofs_cache.read().unwrap(); + Ok(proofs + .iter() + .filter(|proof| match proof.used_by_operation { + Some(owner) => owner == *operation_id, + None => legacy.contains(&hex::encode(proof.y.serialize())), + }) + .cloned() + .collect()) + } + async fn persist_proofs(&self, proofs: &[ProofInfo]) -> Result<(), DatabaseError> { Self::ensure_unique_proof_ys(proofs)?; let blob = ProofsBlobRef { schema_version: PROOFS_SCHEMA_VERSION, proofs }; let data = serde_json::to_vec(&blob).map_err(|e| DatabaseError::Serialization(e.to_string()))?; + // An interrupted write or lost acknowledgement leaves the commit uncertain. + // Keep readers/writers from trusting the old cache until it is reloaded. + self.proofs_cache_needs_reload.store(true, Ordering::Release); KVStore::write(self.store.as_ref(), CASHU_PRIMARY_KEY, PROOFS_KEY, PROOFS_BLOB_KEY, data) .await .map_err(DatabaseError::Io) @@ -302,6 +449,21 @@ impl CashuKvDatabase { fn update_proofs_cache( cache: &mut Vec, added: Vec, removed_ys: Vec, ) { + // Reconstructed/legacy records may omit the NUT-13 index. Match CDK's + // upsert behavior by preserving an index already committed for this proof. + let mut indexes: HashMap<_, _> = + cache.iter().filter_map(|p| p.derivation_index.map(|index| (p.y, index))).collect(); + let added: Vec<_> = added + .into_iter() + .map(|mut proof| { + if let Some(index) = proof.derivation_index { + indexes.insert(proof.y, index); + } else { + proof.derivation_index = indexes.get(&proof.y).copied(); + } + proof + }) + .collect(); let mut added_ys = HashSet::new(); let mut added_deduped = Vec::with_capacity(added.len()); for proof in added.into_iter().rev() { @@ -611,16 +773,15 @@ impl WalletDatabase for CashuKvDatabase { } } - async fn add_mint_quote(&self, quote: MintQuote) -> Result<(), cdk::cdk_database::Error> { - let key = quote.id.clone(); - let data = - serde_json::to_vec("e).map_err(|e| DatabaseError::Serialization(e.to_string()))?; - - KVStore::write(self.store.as_ref(), CASHU_PRIMARY_KEY, MINT_QUOTES_KEY, &key, data) - .await - .map_err(DatabaseError::Io)?; - - Ok(()) + async fn add_mint_quote(&self, mut quote: MintQuote) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; + if let Some(existing) = self.get_mint_quote("e.id).await? { + if existing.version != quote.version { + return Err(cdk::cdk_database::Error::ConcurrentUpdate); + } + quote.version = quote.version.wrapping_add(1); + } + self.write_record(MINT_QUOTES_KEY, "e.id, "e).await } async fn get_mint_quote( @@ -663,6 +824,7 @@ impl WalletDatabase for CashuKvDatabase { } async fn remove_mint_quote(&self, quote_id: &str) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; // Mark as removed by writing empty data KVStore::remove(self.store.as_ref(), CASHU_PRIMARY_KEY, MINT_QUOTES_KEY, quote_id, false) .await @@ -671,16 +833,16 @@ impl WalletDatabase for CashuKvDatabase { Ok(()) } - async fn add_melt_quote(&self, quote: MeltQuote) -> Result<(), cdk::cdk_database::Error> { - let key = quote.id.clone(); - let data = - serde_json::to_vec("e).map_err(|e| DatabaseError::Serialization(e.to_string()))?; - - KVStore::write(self.store.as_ref(), CASHU_PRIMARY_KEY, MELT_QUOTES_KEY, &key, data) - .await - .map_err(DatabaseError::Io)?; - - Ok(()) + async fn add_melt_quote(&self, mut quote: MeltQuote) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; + if let Some(existing) = self.get_melt_quote("e.id).await? { + if existing.version != quote.version { + return Err(cdk::cdk_database::Error::ConcurrentUpdate); + } + quote.version = quote.version.wrapping_add(1); + quote.payment_proof = quote.payment_proof.or(existing.payment_proof); + } + self.write_record(MELT_QUOTES_KEY, "e.id, "e).await } async fn get_melt_quote( @@ -723,6 +885,7 @@ impl WalletDatabase for CashuKvDatabase { } async fn remove_melt_quote(&self, quote_id: &str) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; KVStore::remove(self.store.as_ref(), CASHU_PRIMARY_KEY, MELT_QUOTES_KEY, quote_id, false) .await .map_err(DatabaseError::Io)?; @@ -779,6 +942,7 @@ impl WalletDatabase for CashuKvDatabase { &self, added: Vec, removed_ys: Vec, ) -> Result<(), cdk::cdk_database::Error> { let _guard = self.proofs_mutation_lock.lock().await; + self.reload_uncertain_proofs().await?; let (updated_proofs, changed) = { let committed_proofs = self.proofs_cache.read().unwrap(); let mut updated_proofs = committed_proofs.clone(); @@ -790,6 +954,7 @@ impl WalletDatabase for CashuKvDatabase { if changed { self.persist_proofs(&updated_proofs).await?; *self.proofs_cache.write().unwrap() = updated_proofs; + self.proofs_cache_needs_reload.store(false, Ordering::Release); } Ok(()) @@ -799,8 +964,7 @@ impl WalletDatabase for CashuKvDatabase { &self, mint_url: Option, unit: Option, state: Option>, spending_conditions: Option>, ) -> Result, cdk::cdk_database::Error> { - let cache = self.proofs_cache.read().unwrap(); - let mut filtered_proofs = cache.clone(); + let mut filtered_proofs = self.read_proofs(|cache| cache.to_vec()).await?; // Apply filters if let Some(mint_url) = mint_url { @@ -839,6 +1003,7 @@ impl WalletDatabase for CashuKvDatabase { &self, ys: Vec, state: State, ) -> Result<(), cdk::cdk_database::Error> { let _guard = self.proofs_mutation_lock.lock().await; + self.reload_uncertain_proofs().await?; let ys: HashSet<_> = ys.into_iter().collect(); let mut updated_proofs = self.proofs_cache.read().unwrap().clone(); let mut changed = false; @@ -852,6 +1017,7 @@ impl WalletDatabase for CashuKvDatabase { if changed { self.persist_proofs(&updated_proofs).await?; *self.proofs_cache.write().unwrap() = updated_proofs; + self.proofs_cache_needs_reload.store(false, Ordering::Release); } Ok(()) @@ -868,12 +1034,13 @@ impl WalletDatabase for CashuKvDatabase { match KVStore::read(self.store.as_ref(), CASHU_PRIMARY_KEY, KEYSET_COUNTERS_KEY, &key) .await { - Ok(data) if !data.is_empty() => serde_json::from_slice::(&data) + Ok(data) => serde_json::from_slice::(&data) .map_err(|e| DatabaseError::Serialization(e.to_string()))?, - _ => 0, // Default to 0 if not found + Err(e) if e.kind() == io::ErrorKind::NotFound => 0, + Err(e) => return Err(DatabaseError::Io(e).into()), }; - let new_count = current_count + count; + let new_count = current_count.checked_add(count).ok_or(DatabaseError::InvalidFormat)?; // Write back updated counter let data = serde_json::to_vec(&new_count) @@ -886,6 +1053,34 @@ impl WalletDatabase for CashuKvDatabase { Ok(new_count) } + async fn increment_derivation_counter( + &self, namespace: &str, count: u32, + ) -> Result { + let _lock = self.keyset_counter_lock.lock().await; + // Encode arbitrary CDK namespaces as valid KVStore keys. + let key = hex::encode(namespace.as_bytes()); + let current = match KVStore::read( + self.store.as_ref(), + CASHU_PRIMARY_KEY, + DERIVATION_COUNTERS_KEY, + &key, + ) + .await + { + Ok(data) => serde_json::from_slice::(&data) + .map_err(|e| DatabaseError::Serialization(e.to_string()))?, + Err(e) if e.kind() == io::ErrorKind::NotFound => 0, + Err(e) => return Err(DatabaseError::Io(e).into()), + }; + let next = current.checked_add(count).ok_or(DatabaseError::InvalidFormat)?; + let data = + serde_json::to_vec(&next).map_err(|e| DatabaseError::Serialization(e.to_string()))?; + KVStore::write(self.store.as_ref(), CASHU_PRIMARY_KEY, DERIVATION_COUNTERS_KEY, &key, data) + .await + .map_err(DatabaseError::Io)?; + Ok(next) + } + async fn add_transaction( &self, transaction: Transaction, ) -> Result<(), cdk::cdk_database::Error> { @@ -991,11 +1186,13 @@ impl WalletDatabase for CashuKvDatabase { async fn get_proofs_by_ys( &self, ys: Vec, ) -> Result, cdk::cdk_database::Error> { - let cache = self.proofs_cache.read().unwrap(); - Ok(cache.iter().filter(|p| ys.contains(&p.y)).cloned().collect()) + Ok(self + .read_proofs(|cache| cache.iter().filter(|p| ys.contains(&p.y)).cloned().collect()) + .await?) } async fn add_saga(&self, saga: WalletSaga) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; let key = saga.id.to_string(); let data = serde_json::to_vec(&saga).map_err(|e| DatabaseError::Serialization(e.to_string()))?; @@ -1022,16 +1219,20 @@ impl WalletDatabase for CashuKvDatabase { } async fn update_saga(&self, saga: WalletSaga) -> Result { - let key = saga.id.to_string(); - let data = - serde_json::to_vec(&saga).map_err(|e| DatabaseError::Serialization(e.to_string()))?; - KVStore::write(self.store.as_ref(), CASHU_PRIMARY_KEY, SAGAS_KEY, &key, data) - .await - .map_err(DatabaseError::Io)?; + let _lock = self.records_mutation_lock.lock().await; + let Some(existing) = self.get_saga(&saga.id).await? else { + return Ok(false); + }; + // Unlike quote updates, CDK increments a saga's version before calling us. + if existing.version.checked_add(1) != Some(saga.version) { + return Ok(false); + } + self.write_record(SAGAS_KEY, &saga.id.to_string(), &saga).await?; Ok(true) } async fn delete_saga(&self, id: &uuid::Uuid) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; let key = id.to_string(); KVStore::remove(self.store.as_ref(), CASHU_PRIMARY_KEY, SAGAS_KEY, &key, false) .await @@ -1049,9 +1250,9 @@ impl WalletDatabase for CashuKvDatabase { .await .map_err(DatabaseError::Io)?; if !data.is_empty() { - if let Ok(saga) = serde_json::from_slice::(&data) { - sagas.push(saga); - } + let saga = serde_json::from_slice::(&data) + .map_err(|e| DatabaseError::Serialization(e.to_string()))?; + sagas.push(saga); } } Ok(sagas) @@ -1060,87 +1261,92 @@ impl WalletDatabase for CashuKvDatabase { async fn reserve_proofs( &self, ys: Vec, operation_id: &uuid::Uuid, ) -> Result<(), cdk::cdk_database::Error> { - // Store which Y values are reserved by this operation - let key = operation_id.to_string(); - let ys_data: Vec = ys.iter().map(|y| hex::encode(y.serialize())).collect(); - let data = serde_json::to_vec(&ys_data) - .map_err(|e| DatabaseError::Serialization(e.to_string()))?; - KVStore::write(self.store.as_ref(), CASHU_PRIMARY_KEY, PROOF_RESERVATIONS_KEY, &key, data) - .await - .map_err(DatabaseError::Io)?; - - self.update_proofs_state(ys, State::Reserved).await + if ys.is_empty() { + return Ok(()); + } + let _lock = self.proofs_mutation_lock.lock().await; + self.reload_uncertain_proofs().await?; + let mut proofs = self.proofs_cache.read().unwrap().clone(); + // Selection happens before this lock is acquired. Reject a stale selection + // in its entirety if another operation has removed any requested proof. + let requested: HashSet<_> = ys.iter().copied().collect(); + if requested.len() != ys.len() + || proofs.iter().filter(|p| requested.contains(&p.y)).count() != ys.len() + { + return Err(cdk::cdk_database::Error::ProofNotUnspent); + } + for proof in proofs.iter_mut().filter(|p| requested.contains(&p.y)) { + if proof.used_by_operation.is_some_and(|owner| owner != *operation_id) + || !matches!(proof.state, State::Unspent | State::Reserved) + { + return Err(cdk::cdk_database::Error::ProofNotUnspent); + } + if proof.state == State::Reserved && proof.used_by_operation.is_none() { + // Old adapters kept ownership only in proof_reservations. Do not + // steal a reservation just because its inline owner is absent. + if !self.reserved_proofs(operation_id).await?.iter().any(|p| p.y == proof.y) { + return Err(cdk::cdk_database::Error::ProofNotUnspent); + } + } + proof.state = State::Reserved; + proof.used_by_operation = Some(*operation_id); + } + self.persist_proofs(&proofs).await?; + *self.proofs_cache.write().unwrap() = proofs; + self.proofs_cache_needs_reload.store(false, Ordering::Release); + Ok(()) } async fn release_proofs( &self, operation_id: &uuid::Uuid, ) -> Result<(), cdk::cdk_database::Error> { - let key = operation_id.to_string(); - match KVStore::read(self.store.as_ref(), CASHU_PRIMARY_KEY, PROOF_RESERVATIONS_KEY, &key) - .await - { - Ok(data) if !data.is_empty() => { - let ys_hex: Vec = serde_json::from_slice(&data) - .map_err(|e| DatabaseError::Serialization(e.to_string()))?; - let ys: Vec = ys_hex - .iter() - .filter_map(|h| { - let bytes = hex::decode(h).ok()?; - PublicKey::from_slice(&bytes).ok() - }) - .collect(); - if !ys.is_empty() { - self.update_proofs_state(ys, State::Unspent).await?; - } - KVStore::remove( - self.store.as_ref(), - CASHU_PRIMARY_KEY, - PROOF_RESERVATIONS_KEY, - &key, - false, - ) - .await - .map_err(DatabaseError::Io)?; - }, - _ => {}, + let _lock = self.proofs_mutation_lock.lock().await; + self.reload_uncertain_proofs().await?; + let reserved = self.reserved_proofs(operation_id).await?; + let mut proofs = self.proofs_cache.read().unwrap().clone(); + for proof in &mut proofs { + if reserved.iter().any(|p| p.y == proof.y) + && matches!(proof.state, State::Reserved | State::Pending) + { + proof.state = State::Unspent; + proof.used_by_operation = None; + } } + self.persist_proofs(&proofs).await?; + *self.proofs_cache.write().unwrap() = proofs; + self.proofs_cache_needs_reload.store(false, Ordering::Release); + KVStore::remove( + self.store.as_ref(), + CASHU_PRIMARY_KEY, + PROOF_RESERVATIONS_KEY, + &operation_id.to_string(), + false, + ) + .await + .map_err(DatabaseError::Io)?; Ok(()) } async fn get_reserved_proofs( &self, operation_id: &uuid::Uuid, ) -> Result, cdk::cdk_database::Error> { - let key = operation_id.to_string(); - match KVStore::read(self.store.as_ref(), CASHU_PRIMARY_KEY, PROOF_RESERVATIONS_KEY, &key) - .await - { - Ok(data) if !data.is_empty() => { - let ys_hex: Vec = serde_json::from_slice(&data) - .map_err(|e| DatabaseError::Serialization(e.to_string()))?; - let ys: Vec = ys_hex - .iter() - .filter_map(|h| { - let bytes = hex::decode(h).ok()?; - PublicKey::from_slice(&bytes).ok() - }) - .collect(); - self.get_proofs_by_ys(ys).await - }, - _ => Ok(Vec::new()), - } + let _lock = self.proofs_mutation_lock.lock().await; + self.reload_uncertain_proofs().await?; + self.reserved_proofs(operation_id).await } async fn reserve_melt_quote( &self, quote_id: &str, operation_id: &uuid::Uuid, ) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; if let Some(mut quote) = self.get_melt_quote(quote_id).await? { if quote.used_by_operation.is_some() { - return Err(cdk::cdk_database::Error::Database( - "Quote already in use by another operation".to_string().into(), - )); + return Err(cdk::cdk_database::Error::QuoteAlreadyInUse); } quote.used_by_operation = Some(operation_id.to_string()); - self.add_melt_quote(quote).await?; + self.write_record(MELT_QUOTES_KEY, "e.id, "e).await?; + } else { + return Err(cdk::cdk_database::Error::UnknownQuote); } Ok(()) } @@ -1148,12 +1354,13 @@ impl WalletDatabase for CashuKvDatabase { async fn release_melt_quote( &self, operation_id: &uuid::Uuid, ) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; let quotes = self.get_melt_quotes().await?; let op_str = operation_id.to_string(); for mut quote in quotes { if quote.used_by_operation.as_deref() == Some(&op_str) { quote.used_by_operation = None; - self.add_melt_quote(quote).await?; + self.write_record(MELT_QUOTES_KEY, "e.id, "e).await?; } } Ok(()) @@ -1162,14 +1369,15 @@ impl WalletDatabase for CashuKvDatabase { async fn reserve_mint_quote( &self, quote_id: &str, operation_id: &uuid::Uuid, ) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; if let Some(mut quote) = self.get_mint_quote(quote_id).await? { if quote.used_by_operation.is_some() { - return Err(cdk::cdk_database::Error::Database( - "Quote already in use by another operation".to_string().into(), - )); + return Err(cdk::cdk_database::Error::QuoteAlreadyInUse); } quote.used_by_operation = Some(operation_id.to_string()); - self.add_mint_quote(quote).await?; + self.write_record(MINT_QUOTES_KEY, "e.id, "e).await?; + } else { + return Err(cdk::cdk_database::Error::UnknownQuote); } Ok(()) } @@ -1177,12 +1385,13 @@ impl WalletDatabase for CashuKvDatabase { async fn release_mint_quote( &self, operation_id: &uuid::Uuid, ) -> Result<(), cdk::cdk_database::Error> { + let _lock = self.records_mutation_lock.lock().await; let quotes = self.get_mint_quotes().await?; let op_str = operation_id.to_string(); for mut quote in quotes { if quote.used_by_operation.as_deref() == Some(&op_str) { quote.used_by_operation = None; - self.add_mint_quote(quote).await?; + self.write_record(MINT_QUOTES_KEY, "e.id, "e).await?; } } Ok(()) @@ -1375,7 +1584,7 @@ mod tests { } } - fn proof_info(secret: &str, amount: u64) -> ProofInfo { + pub(super) fn proof_info(secret: &str, amount: u64) -> ProofInfo { let proof = Proof::new( Amount::from(amount), Id::from_str("009a1f293253e41e").unwrap(), @@ -1399,6 +1608,225 @@ mod tests { database.get_proofs(None, None, None, None).await.unwrap() } + pub(super) fn migration_store() -> (std::path::PathBuf, Arc) { + let path = std::env::temp_dir() + // PIDs can repeat across sandbox namespaces sharing this directory. + .join(format!("orange-cdk-migration-{}", uuid::Uuid::new_v4())); + let store = ldk_node::io::sqlite_store::SqliteStore::new( + path.clone(), + Some("orange.sqlite".into()), + None, + ) + .unwrap(); + (path, Arc::new(store)) + } + + // This is the JSON shape written by CDK 0.16, deliberately without status. + fn legacy_transaction() -> serde_json::Value { + serde_json::json!({ + "mint_url": "https://mint.example.com", "direction": "Outgoing", + "amount": 10, "fee": 1, "unit": "sat", "ys": [], "timestamp": 42, + "memo": null, "metadata": {}, "quote_id": "quote-1", + "payment_request": "invoice", "payment_proof": "preimage", + "payment_method": "bolt11", "saga_id": "019fa338-b72f-7f21-9bb2-a504cdd5927b" + }) + } + + #[test] + fn transaction_status_is_preserved_in_orange_payments() { + for (status, expected) in [ + (cdk::wallet::types::TransactionStatus::Pending, crate::store::TxStatus::Pending), + (cdk::wallet::types::TransactionStatus::Completed, crate::store::TxStatus::Completed), + (cdk::wallet::types::TransactionStatus::Failed, crate::store::TxStatus::Failed), + ] { + let mut tx: Transaction = serde_json::from_value(legacy_transaction()).unwrap(); + tx.status = status; + let payment = + super::super::Cashu::convert_transaction_to_payment(tx, &CurrencyUnit::Sat) + .unwrap(); + assert_eq!(payment.status, expected); + } + } + + #[test] + fn outgoing_retries_surface_one_payment_in_either_storage_order() { + use super::super::Cashu; + use cdk::wallet::types::TransactionStatus::{Completed, Failed, Pending}; + for (old_status, new_status, settled_wins) in [ + (Failed, Pending, false), + (Failed, Completed, false), + (Completed, Failed, true), + (Pending, Failed, false), + ] { + for same_second in [false, true] { + let mut old: Transaction = serde_json::from_value(legacy_transaction()).unwrap(); + old.status = old_status; + old.saga_id = Some("019fa338-b72f-7000-8000-000000000001".parse().unwrap()); + let mut retry = old.clone(); + retry.status = new_status; + retry.saga_id = Some("019fa338-b730-7000-8000-000000000001".parse().unwrap()); + retry.timestamp += u64::from(!same_second); + retry.fee = Amount::from(2); + let expected = if settled_wins { &old } else { &retry }; + let expected = + Cashu::convert_transaction_to_payment(expected.clone(), &CurrencyUnit::Sat) + .unwrap(); + for attempts in [vec![old.clone(), retry.clone()], vec![retry.clone(), old.clone()]] + { + let payments = + Cashu::convert_transactions_to_payments(attempts, &CurrencyUnit::Sat) + .unwrap(); + assert_eq!(payments.len(), 1); + assert_eq!(payments[0].id, expected.id); + assert_eq!(payments[0].status, expected.status); + assert_eq!(payments[0].fee, expected.fee); + assert_eq!(payments[0].time_since_epoch, expected.time_since_epoch); + } + } + } + } + + #[test] + fn reconciliation_keeps_distinct_quotes_and_repeated_incoming_receipts() { + use super::super::Cashu; + let outgoing: Transaction = serde_json::from_value(legacy_transaction()).unwrap(); + let mut another_quote = outgoing.clone(); + another_quote.quote_id = Some("quote-2".into()); + let mut incoming = outgoing.clone(); + incoming.direction = cdk::wallet::types::TransactionDirection::Incoming; + let mut second_receipt = incoming.clone(); + second_receipt.saga_id = Some(uuid::Uuid::from_u128(2)); + second_receipt.amount = Amount::from(20); + let payments = Cashu::convert_transactions_to_payments( + vec![outgoing, another_quote, incoming, second_receipt], + &CurrencyUnit::Sat, + ) + .unwrap(); + assert_eq!(payments.len(), 4); + assert_ne!(payments[0].id, payments[1].id); + assert!(!payments[2].outbound && !payments[3].outbound); + assert_ne!(payments[2].amount, payments[3].amount); + } + + #[tokio::test] + async fn transaction_migration_preserves_conflicting_records() { + let (path, store) = migration_store(); + let data = serde_json::to_vec(&legacy_transaction()).unwrap(); + let mut tx: Transaction = serde_json::from_slice(&data).unwrap(); + let old_key = TransactionId::new(tx.ys.clone()).to_string(); + KVStore::write(store.as_ref(), CASHU_PRIMARY_KEY, TRANSACTIONS_KEY, &old_key, data.clone()) + .await + .unwrap(); + tx.amount = Amount::from(99); + let conflicting = serde_json::to_vec(&tx).unwrap(); + KVStore::write( + store.as_ref(), + CASHU_PRIMARY_KEY, + TRANSACTIONS_KEY, + &tx.id().to_string(), + conflicting.clone(), + ) + .await + .unwrap(); + assert!(matches!(CashuKvDatabase::new(store.clone()).await, Err(DatabaseError::Duplicate))); + assert_eq!( + KVStore::read(store.as_ref(), CASHU_PRIMARY_KEY, TRANSACTIONS_KEY, &old_key) + .await + .unwrap(), + data + ); + assert_eq!( + KVStore::read( + store.as_ref(), + CASHU_PRIMARY_KEY, + TRANSACTIONS_KEY, + &tx.id().to_string() + ) + .await + .unwrap(), + conflicting + ); + drop(store); + std::fs::remove_dir_all(path).unwrap(); + } + + #[tokio::test] + async fn derivation_counters_are_namespaced_persistent_and_checked() { + let (path, store) = migration_store(); + let db = Arc::new(CashuKvDatabase::new(store.clone()).await.unwrap()); + let (a, b) = tokio::join!( + db.increment_derivation_counter("nut20_quote", 1), + db.increment_derivation_counter("nut20_quote", 1) + ); + let mut counts = [a.unwrap(), b.unwrap()]; + counts.sort(); + assert_eq!(counts, [1, 2]); + assert_eq!(db.increment_derivation_counter("other", 0).await.unwrap(), 0); + drop(db); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + assert_eq!(db.increment_derivation_counter("nut20_quote", 0).await.unwrap(), 2); + assert!(db.increment_derivation_counter("nut20_quote", u32::MAX).await.is_err()); + assert_eq!(db.increment_derivation_counter("nut20_quote", 0).await.unwrap(), 2); + drop(db); + drop(store); + std::fs::remove_dir_all(path).unwrap(); + } + + #[tokio::test] + async fn release_proofs_preserves_spent_and_foreign_ownership_after_restart() { + let (path, store) = migration_store(); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let operation = uuid::Uuid::from_u128(1); + let other = uuid::Uuid::from_u128(2); + let reserved = proof_info("reserved", 1); + let mut pending = proof_info("pending", 2); + pending.state = State::Pending; + pending.used_by_operation = Some(operation); + let mut spent = proof_info("spent", 4); + spent.state = State::Spent; + spent.used_by_operation = Some(operation); + let mut foreign = proof_info("foreign", 8); + foreign.state = State::Reserved; + foreign.used_by_operation = Some(other); + let mut legacy = proof_info("legacy reservation", 16); + legacy.state = State::Reserved; + db.update_proofs( + vec![reserved.clone(), pending.clone(), spent.clone(), foreign.clone(), legacy.clone()], + vec![], + ) + .await + .unwrap(); + db.reserve_proofs(vec![reserved.y], &operation).await.unwrap(); + // A stale legacy index must not release spent or reassigned proofs. + let index: Vec<_> = [legacy.y, spent.y, foreign.y] + .into_iter() + .map(|y| hex::encode(y.serialize())) + .collect(); + KVStore::write( + store.as_ref(), + CASHU_PRIMARY_KEY, + PROOF_RESERVATIONS_KEY, + &operation.to_string(), + serde_json::to_vec(&index).unwrap(), + ) + .await + .unwrap(); + drop(db); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + db.release_proofs(&operation).await.unwrap(); + let proofs = all_proofs(&db).await; + for y in [reserved.y, pending.y, legacy.y] { + let proof = proofs.iter().find(|p| p.y == y).unwrap(); + assert_eq!(proof.state, State::Unspent); + assert_eq!(proof.used_by_operation, None); + } + assert!(proofs.contains(&spent)); + assert!(proofs.contains(&foreign)); + drop(db); + drop(store); + std::fs::remove_dir_all(path).unwrap(); + } + #[test] fn update_proofs_cache_replaces_existing_proof_with_same_y() { let original = proof_info("same cached proof", 1); @@ -1416,17 +1844,34 @@ mod tests { let original = proof_info("duplicate added proof", 1); let replacement = proof_info("duplicate added proof", 2); let mut cache = Vec::new(); - CashuKvDatabase::update_proofs_cache( &mut cache, vec![original.clone(), replacement.clone()], vec![], ); - assert_eq!(cache, vec![replacement]); assert_eq!(cache.iter().filter(|proof| proof.y == original.y).count(), 1); } + #[tokio::test] + async fn committed_proof_reads_do_not_wait_for_the_mutation_lock() { + let store = Arc::new(ProofTestStore::default()); + let database = CashuKvDatabase::new(store).await.unwrap(); + let proof = proof_info("committed proof", 2); + database.update_proofs(vec![proof.clone()], vec![]).await.unwrap(); + let _writer = database.proofs_mutation_lock.lock().await; + tokio::time::timeout(std::time::Duration::from_secs(1), async { + assert_eq!( + database.get_proofs_by_ys(vec![proof.y]).await.unwrap(), + vec![proof.clone()] + ); + assert_eq!(all_proofs(&database).await, vec![proof]); + assert_eq!(database.get_balance(None, None, None).await.unwrap(), 2); + }) + .await + .expect("committed reads should not wait for a writer"); + } + #[test] fn update_proofs_cache_removal_wins_over_added_proof() { let proof = proof_info("removed cached proof", 1); @@ -1582,3 +2027,7 @@ mod tests { assert!(serialized.len() < 1024 * 1024); } } + +#[cfg(test)] +#[path = "cashu_store_upgrade_tests.rs"] +mod upgrade_tests; diff --git a/orange-sdk/src/trusted_wallet/cashu/cashu_store_live_upgrade_tests.rs b/orange-sdk/src/trusted_wallet/cashu/cashu_store_live_upgrade_tests.rs new file mode 100644 index 0000000..cbdd667 --- /dev/null +++ b/orange-sdk/src/trusted_wallet/cashu/cashu_store_live_upgrade_tests.rs @@ -0,0 +1,409 @@ +//! Funded old-version wallets, a real local mint, and a fake Lightning backend. +use super::*; +use axum::extract::{Request, State as HttpState}; +use axum::http::{Method, StatusCode}; +use axum::middleware::Next; +use axum::response::{IntoResponse, Response}; +use cdk::nuts::MeltQuoteState; +use cdk::wallet::types::{IssueSagaState, MeltSagaState, TransactionStatus, WalletSagaState}; +use cdk_common::payment::{MintPayment, PaymentIdentifier}; +use std::process::Stdio; +use std::time::Duration; +use tokio::sync::Notify; + +#[derive(Default)] +struct HttpControl { + hold_next_melt: AtomicBool, + hold_next_issue: AtomicBool, + response_ready: Notify, + release_response: Notify, + offline: AtomicBool, + offline_requests: AtomicUsize, + melt_requests: AtomicUsize, + restore_requests: AtomicUsize, +} + +async fn control_http( + HttpState(control): HttpState>, request: Request, next: Next, +) -> Response { + let is_melt = request.method() == Method::POST && request.uri().path() == "/v1/melt/bolt11"; + if is_melt { + control.melt_requests.fetch_add(1, Ordering::SeqCst); + } + if request.method() == Method::POST && request.uri().path() == "/v1/restore" { + control.restore_requests.fetch_add(1, Ordering::SeqCst); + } + if control.offline.load(Ordering::SeqCst) { + control.offline_requests.fetch_add(1, Ordering::SeqCst); + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + let is_issue = request.method() == Method::POST && request.uri().path() == "/v1/mint/bolt11"; + let hold = (is_melt && control.hold_next_melt.swap(false, Ordering::SeqCst)) + || (is_issue && control.hold_next_issue.swap(false, Ordering::SeqCst)); + let response = next.run(request).await; + if hold { + // The mint has handled the submission, but the old wallet cannot see its + // response. Kill the wallet here to test an actual lost acknowledgement. + control.response_ready.notify_one(); + control.release_response.notified().await; + } + response +} + +struct LiveMint { + mint: Arc, + backend: Arc, + url: String, + control: Arc, + server: tokio::task::JoinHandle<()>, +} + +impl LiveMint { + async fn new() -> Self { + let mint_db = Arc::new(cdk_sqlite::mint::memory::empty().await.unwrap()); + let backend = Arc::new(cdk_fake_wallet::FakeWallet::new( + // FakeWallet charges one sat. Reserve two to require change recovery. + cdk::types::FeeReserve { min_fee_reserve: 2.into(), percent_fee_reserve: 0.0 }, + Default::default(), + Default::default(), + 0, + CurrencyUnit::Sat, + )); + let mut builder = cdk::mint::MintBuilder::new(mint_db.clone()); + builder.configure_unit(CurrencyUnit::Sat, cdk::mint::UnitConfig::default()).unwrap(); + builder + .add_payment_processor( + CurrencyUnit::Sat, + cdk::nuts::PaymentMethod::BOLT11, + cdk::mint::MintMeltLimits::new(0, u64::MAX), + backend.clone(), + ) + .await + .unwrap(); + let mint = Arc::new(builder.build_with_seed(mint_db, &[42; 64]).await.unwrap()); + mint.start().await.unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let control = Arc::new(HttpControl::default()); + let router = cdk_axum::create_mint_router(mint.clone(), vec!["bolt11".into()]) + .await + .unwrap() + .layer(axum::middleware::from_fn_with_state(control.clone(), control_http)); + let server = tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + }); + Self { mint, backend, url, control, server } + } + + async fn old_wallet( + &self, submitted: bool, outcome: MeltQuoteState, + ) -> (std::path::PathBuf, Arc) { + self.old_wallet_operation(submitted, outcome, false).await + } + + async fn old_wallet_operation( + &self, submitted: bool, outcome: MeltQuoteState, issue: bool, + ) -> (std::path::PathBuf, Arc) { + let generator = std::env::var_os("ORANGE_CDK016_GENERATOR") + .expect("run ci/test-cashu-upgrade.sh to build the old-version generator"); + let description = cdk_fake_wallet::FakeInvoiceDescription { + pay_invoice_state: outcome, + check_payment_state: outcome, + pay_err: false, + check_err: false, + }; + let invoice = cdk_fake_wallet::create_fake_invoice( + 8000, + serde_json::to_string(&description).unwrap(), + ) + .to_string(); + let (path, store) = migration_store(); + drop(store); + let mut command = tokio::process::Command::new(generator); + command + .arg(&path) + .arg(&self.url) + .arg(if issue { "issue".to_owned() } else { invoice }) + .kill_on_drop(true) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()); + if issue { + self.control.hold_next_issue.store(true, Ordering::SeqCst); + } else if submitted { + self.control.hold_next_melt.store(true, Ordering::SeqCst); + command.arg("submit"); + } + let mut child = command.spawn().unwrap(); + if submitted || issue { + tokio::select! { + _ = self.control.response_ready.notified() => {}, + status = child.wait() => panic!("old wallet exited before the response gate: {status:?}"), + } + child.kill().await.unwrap(); + assert!(!child.wait().await.unwrap().success()); + self.control.release_response.notify_one(); + } else { + assert!(child.wait().await.unwrap().success()); + } + let store: Arc = Arc::new( + ldk_node::io::sqlite_store::SqliteStore::new( + path.clone(), + Some("orange.sqlite".into()), + None, + ) + .unwrap(), + ); + write_has_recovered(&store, true).await.unwrap(); + (path, store) + } + + async fn stop(self) { + self.server.abort(); + let _ = self.server.await; + self.mint.stop().await.unwrap(); + } +} + +fn close_wallet(cashu: crate::trusted_wallet::cashu::Cashu) { + cashu.runtime.abort_cancellable_background_tasks(); +} + +fn run_upgrade_test(test: impl Future) { + let runtime = + tokio::runtime::Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap(); + let path = runtime.block_on(async { + tokio::time::timeout(Duration::from_secs(90), test) + .await + .expect("funded wallet upgrade timed out") + }); + // Cancellation does not stop SQLite spawn_blocking work already in flight. + // Drain it before deleting the database, rather than racing its last access. + drop(runtime); + std::fs::remove_dir_all(path).unwrap(); +} + +#[test] +#[ignore = "requires the CDK 0.16 fixture generator; run ci/test-cashu-upgrade.sh"] +fn cdk016_funded_wallet_recovers_interrupted_melt_and_spends_after_upgrade() { + run_upgrade_test(async { + let mint = LiveMint::new().await; + let (path, store) = mint.old_wallet(false, MeltQuoteState::Paid).await; + let cashu = open_cashu(store.clone(), &mint.url).await.unwrap(); + let wallet = &cashu.cashu_wallet; + assert_eq!(wallet.total_balance().await.unwrap(), cdk::Amount::from(64)); + assert!(wallet.localstore.get_incomplete_sagas().await.unwrap().is_empty()); + assert!(wallet.recover_incomplete_sagas().await.unwrap().is_empty()); + let invoice = + cdk_fake_wallet::create_fake_invoice(8000, "after upgrade".into()).to_string(); + let quote = + wallet.melt_quote(cdk::nuts::PaymentMethod::BOLT11, invoice, None, None).await.unwrap(); + wallet.prepare_melt("e.id, Default::default()).await.unwrap().confirm().await.unwrap(); + assert_eq!(wallet.total_balance().await.unwrap(), cdk::Amount::from(55)); + close_wallet(cashu); + let cashu = open_cashu(store.clone(), &mint.url).await.unwrap(); + assert_eq!(cashu.cashu_wallet.total_balance().await.unwrap(), cdk::Amount::from(55)); + assert_eq!( + cashu + .cashu_wallet + .localstore + .list_transactions(None, Some(TransactionDirection::Outgoing), None) + .await + .unwrap() + .len(), + 1 + ); + close_wallet(cashu); + assert_eq!(mint.control.melt_requests.load(Ordering::SeqCst), 1); + drop(store); + mint.stop().await; + path + }); +} + +fn recover_submitted_melt(outcome: MeltQuoteState, offline_first: bool) { + run_upgrade_test(async { + let mint = LiveMint::new().await; + let (path, store) = mint.old_wallet(true, outcome).await; + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let sagas = db.get_incomplete_sagas().await.unwrap(); + assert_eq!(sagas.len(), 1); + assert_eq!(sagas[0].state, WalletSagaState::Melt(MeltSagaState::MeltRequested)); + let operation = sagas[0].id; + let quote = db.get_melt_quotes().await.unwrap().remove(0); + let quote_id = quote.id; + let invoice: ldk_node::lightning_invoice::Bolt11Invoice = quote.request.parse().unwrap(); + let reserved = db.get_reserved_proofs(&operation).await.unwrap(); + assert!(!reserved.is_empty()); + assert!(reserved.iter().all(|proof| proof.state == State::Pending)); + assert_eq!(reserved.iter().map(|proof| u64::from(proof.proof.amount)).sum::(), 10); + assert_eq!(db.get_balance(None, None, Some(vec![State::Unspent])).await.unwrap(), 54); + drop(db); + // Async melts can return PENDING before the fake backend finishes. Wait + // for the mint's actual result, independently of the lost HTTP response. + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let state = + mint.mint.check_melt_quote("e_id.parse().unwrap()).await.unwrap().state(); + if state == outcome + || (outcome == MeltQuoteState::Failed && state == MeltQuoteState::Unpaid) + { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("mint did not reach the requested outcome"); + // In particular, the failed case must reach the payment backend, rather + // than passing because the mint rejected an invalid request up front. + let payment = mint + .backend + .check_outgoing_payment(&PaymentIdentifier::PaymentHash(invoice.payment_hash().0)) + .await + .unwrap(); + assert_eq!(payment.status, outcome); + if offline_first { + mint.control.offline.store(true, Ordering::SeqCst); + let cashu = open_cashu(store.clone(), &mint.url).await.unwrap(); + assert_eq!(cashu.cashu_wallet.total_balance().await.unwrap(), cdk::Amount::from(54)); + assert_eq!( + cashu.cashu_wallet.localstore.get_reserved_proofs(&operation).await.unwrap(), + reserved + ); + assert_eq!( + cashu.cashu_wallet.localstore.get_incomplete_sagas().await.unwrap().len(), + 1 + ); + assert!(mint.control.offline_requests.load(Ordering::SeqCst) > 0); + close_wallet(cashu); + mint.control.offline.store(false, Ordering::SeqCst); + } + for _ in 0..2 { + let cashu = open_cashu(store.clone(), &mint.url).await.unwrap(); + let wallet = &cashu.cashu_wallet; + let db = &wallet.localstore; + let transactions = db + .list_transactions(None, Some(TransactionDirection::Outgoing), None) + .await + .unwrap(); + match outcome { + MeltQuoteState::Paid => { + assert_eq!(wallet.total_balance().await.unwrap(), cdk::Amount::from(55)); + assert!(db.get_incomplete_sagas().await.unwrap().is_empty()); + assert_eq!(transactions.len(), 1); + assert_eq!(transactions[0].status, TransactionStatus::Completed); + assert_eq!(transactions[0].quote_id.as_deref(), Some(quote_id.as_str())); + assert_eq!(transactions[0].amount, cdk::Amount::from(8)); + assert_eq!(transactions[0].fee, cdk::Amount::from(1)); + let spent = + db.get_proofs_by_ys(reserved.iter().map(|p| p.y).collect()).await.unwrap(); + assert_eq!(spent.len(), reserved.len()); + assert!(spent.iter().all(|p| p.state == State::Spent)); + }, + MeltQuoteState::Failed => { + assert_eq!(wallet.total_balance().await.unwrap(), cdk::Amount::from(64)); + assert!(db.get_incomplete_sagas().await.unwrap().is_empty()); + assert!(db.get_reserved_proofs(&operation).await.unwrap().is_empty()); + assert!( + transactions.len() <= 1 + && transactions.iter().all(|t| t.status == TransactionStatus::Failed) + ); + }, + MeltQuoteState::Pending => { + assert_eq!(wallet.total_balance().await.unwrap(), cdk::Amount::from(54)); + assert_eq!(db.get_reserved_proofs(&operation).await.unwrap(), reserved); + assert_eq!(db.get_incomplete_sagas().await.unwrap().len(), 1); + assert!( + transactions.len() <= 1 + && transactions.iter().all(|t| t.status == TransactionStatus::Pending) + ); + }, + _ => unreachable!(), + } + close_wallet(cashu); + assert_eq!( + mint.control.melt_requests.load(Ordering::SeqCst), + 1, + "recovery must not submit another payment" + ); + } + if outcome == MeltQuoteState::Paid { + assert!( + mint.control.restore_requests.load(Ordering::SeqCst) > 0, + "lost change must be restored from the mint" + ); + } + drop(store); + mint.stop().await; + path + }); +} + +#[test] +#[ignore = "requires the CDK 0.16 fixture generator; run ci/test-cashu-upgrade.sh"] +fn cdk016_funded_wallet_recovers_paid_melt_and_change_after_lost_response() { + recover_submitted_melt(MeltQuoteState::Paid, false); +} + +#[test] +#[ignore = "requires the CDK 0.16 fixture generator; run ci/test-cashu-upgrade.sh"] +fn cdk016_funded_wallet_releases_failed_melt_after_lost_response() { + recover_submitted_melt(MeltQuoteState::Failed, false); +} + +#[test] +#[ignore = "requires the CDK 0.16 fixture generator; run ci/test-cashu-upgrade.sh"] +fn cdk016_funded_wallet_keeps_pending_melt_reserved_after_lost_response() { + recover_submitted_melt(MeltQuoteState::Pending, false); +} + +#[test] +#[ignore = "requires the CDK 0.16 fixture generator; run ci/test-cashu-upgrade.sh"] +fn cdk016_funded_wallet_retries_recovery_after_mint_unavailability() { + recover_submitted_melt(MeltQuoteState::Paid, true); +} + +#[test] +#[ignore = "requires the CDK 0.16 fixture generator; run ci/test-cashu-upgrade.sh"] +fn cdk016_funded_wallet_opens_with_issue_saga_offline_then_recovers() { + run_upgrade_test(async { + let mint = LiveMint::new().await; + let (path, store) = mint.old_wallet_operation(false, MeltQuoteState::Paid, true).await; + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let sagas = db.get_incomplete_sagas().await.unwrap(); + assert_eq!(sagas.len(), 1); + assert_eq!(sagas[0].state, WalletSagaState::Issue(IssueSagaState::MintRequested)); + let operation = sagas[0].id; + drop(db); + mint.control.offline.store(true, Ordering::SeqCst); + let cashu = open_cashu(store.clone(), &mint.url) + .await + .expect("offline issue recovery must not block startup"); + let wallet = &cashu.cashu_wallet; + assert_eq!(wallet.total_balance().await.unwrap(), cdk::Amount::ZERO); + assert_eq!(wallet.localstore.get_incomplete_sagas().await.unwrap()[0].id, operation); + assert_eq!(wallet.recover_incomplete_sagas().await.unwrap().failed, 1); + assert!(mint.control.offline_requests.load(Ordering::SeqCst) > 0); + assert!(mint.control.restore_requests.load(Ordering::SeqCst) > 0); + close_wallet(cashu); + mint.control.offline.store(false, Ordering::SeqCst); + for _ in 0..2 { + let cashu = open_cashu(store.clone(), &mint.url).await.unwrap(); + let wallet = &cashu.cashu_wallet; + assert_eq!(wallet.total_balance().await.unwrap(), cdk::Amount::from(64)); + assert!(wallet.localstore.get_incomplete_sagas().await.unwrap().is_empty()); + let transactions = wallet + .localstore + .list_transactions(None, Some(TransactionDirection::Incoming), None) + .await + .unwrap(); + assert_eq!(transactions.len(), 1); + assert_eq!(transactions[0].status, TransactionStatus::Completed); + assert_eq!(transactions[0].amount, cdk::Amount::from(64)); + close_wallet(cashu); + } + drop(store); + mint.stop().await; + path + }); +} diff --git a/orange-sdk/src/trusted_wallet/cashu/cashu_store_upgrade_tests.rs b/orange-sdk/src/trusted_wallet/cashu/cashu_store_upgrade_tests.rs new file mode 100644 index 0000000..e3236d1 --- /dev/null +++ b/orange-sdk/src/trusted_wallet/cashu/cashu_store_upgrade_tests.rs @@ -0,0 +1,600 @@ +//! Upgrade tests exercise the public database contract and interrupted storage operations. +use super::tests::{migration_store, proof_info}; +use super::*; +use ldk_node::lightning::util::persist::{PageToken, PaginatedListResponse}; +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; + +// Exercise the same initialization entry point as Wallet::new, without an LDK node. +async fn open_cashu( + store: Arc, url: &str, +) -> Result { + use crate::trusted_wallet::cashu::{Cashu, CashuConfig}; + let cashu_config = + CashuConfig { mint_url: url.into(), unit: CurrencyUnit::Sat, npubcash_url: None }; + let config = crate::WalletConfig { + storage_config: crate::StorageConfig::LocalSQLite(String::new()), + logger_type: crate::logging::LoggerType::LogFacade, + chain_source: crate::ChainSource::Esplora { + url: "http://127.0.0.1:1".into(), + username: None, + password: None, + }, + lsp: ( + "127.0.0.1:1".parse().unwrap(), + "024369d2d22a80ecf78f3937da9d5f30c1b9f74f0c32684d583cca0fa6a61cdcfc".parse().unwrap(), + None, + ), + scorer_url: None, + rgs_url: None, + network: ldk_node::bitcoin::Network::Regtest, + seed: crate::Seed::Seed64([7; 64]), + tunables: Default::default(), + extra_config: crate::ExtraConfig::Cashu(cashu_config.clone()), + }; + let logger = Arc::new(crate::logging::Logger::new(&config.logger_type).unwrap()); + let runtime = Arc::new(crate::runtime::Runtime::new(logger.clone()).unwrap()); + let metadata = crate::store::TxMetadataStore::new(store.clone()).await; + let events = Arc::new(crate::EventQueue::new(store.clone(), metadata.clone(), logger.clone())); + Cashu::init(&config, cashu_config, store, events, metadata, logger, runtime).await +} + +#[tokio::test] +async fn missing_or_duplicate_proofs_never_partially_reserve() { + let (path, store) = migration_store(); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let present = proof_info("present", 1); + let missing = proof_info("missing", 2); + db.update_proofs(vec![present.clone()], vec![]).await.unwrap(); + let operation = uuid::Uuid::from_u128(7); + for ys in [vec![present.y, missing.y], vec![missing.y], vec![present.y, present.y]] { + assert!(matches!( + db.reserve_proofs(ys, &operation).await, + Err(cdk::cdk_database::Error::ProofNotUnspent) + )); + assert_eq!(db.get_proofs(None, None, None, None).await.unwrap(), vec![present.clone()]); + } + drop(db); + let reopened = CashuKvDatabase::new(store.clone()).await.unwrap(); + assert_eq!(reopened.get_proofs(None, None, None, None).await.unwrap(), vec![present]); + drop(reopened); + drop(store); + std::fs::remove_dir_all(path).unwrap(); +} + +#[test] +fn startup_recovers_legacy_reservations_even_when_seed_restore_is_complete() { + let runtime = + tokio::runtime::Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap(); + let path = runtime.block_on(async { + let (path, store) = migration_store(); + seed_legacy_store(&store).await; + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let mut saga = db.get_saga(&uuid::Uuid::from_u128(1)).await.unwrap().unwrap(); + saga.state = cdk::wallet::types::WalletSagaState::Melt( + cdk::wallet::types::MeltSagaState::ProofsReserved, + ); + db.add_saga(saga).await.unwrap(); + // Synthetic invoices should not be monitored; recovery itself needs no HTTP here. + db.remove_mint_quote("mint-quote").await.unwrap(); + drop(db); + write_has_recovered(&store, true).await.unwrap(); + for _ in 0..2 { + let cashu = open_cashu(store.clone(), "https://mint.example.com").await.unwrap(); + let db = &cashu.cashu_wallet.localstore; + assert_eq!(cashu.cashu_wallet.total_balance().await.unwrap(), cdk::Amount::from(3)); + assert!(db.get_incomplete_sagas().await.unwrap().is_empty()); + assert!(db.get_reserved_proofs(&uuid::Uuid::from_u128(1)).await.unwrap().is_empty()); + cashu.runtime.abort_cancellable_background_tasks(); + } + drop(store); + path + }); + // Aborted wallet tasks may have already dispatched SQLite spawn_blocking + // work. Runtime shutdown drains it before the directory is removed. + drop(runtime); + std::fs::remove_dir_all(path).unwrap(); +} + +#[test] +fn startup_preserves_reservations_on_unreadable_or_failed_recovery() { + let runtime = + tokio::runtime::Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap(); + let paths = runtime.block_on(async { + let mut paths = Vec::new(); + for malformed in [false, true] { + let (path, store) = migration_store(); + seed_legacy_store(&store).await; + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let id = uuid::Uuid::from_u128(1); + let reserved = db.get_reserved_proofs(&id).await.unwrap(); + // Avoid background requests for the synthetic invoice and seed restoration. + db.remove_mint_quote("mint-quote").await.unwrap(); + write_has_recovered(&store, true).await.unwrap(); + if malformed { + store + .write_async( + CASHU_PRIMARY_KEY, + SAGAS_KEY, + &id.to_string(), + b"invalid JSON".to_vec(), + ) + .await + .unwrap(); + } else { + let mut saga = db.get_saga(&id).await.unwrap().unwrap(); + // Valid serialization, but recovery cannot interpret melt data as a swap. + saga.state = cdk::wallet::types::WalletSagaState::Swap( + cdk::wallet::types::SwapSagaState::ProofsReserved, + ); + db.add_saga(saga).await.unwrap(); + } + drop(db); + let result = open_cashu(store.clone(), "http://127.0.0.1:1").await; + if malformed { + assert!(matches!( + result, + Err(crate::InitFailure::TrustedFailure(TrustedError::WalletOperationFailed(_))) + )); + } else { + let cashu = result.expect("individual recovery failures must not block startup"); + assert_eq!( + cashu.cashu_wallet.localstore.get_incomplete_sagas().await.unwrap().len(), + 1 + ); + cashu.runtime.abort_cancellable_background_tasks(); + } + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + assert_eq!(db.get_reserved_proofs(&id).await.unwrap(), reserved); + paths.push(path); + } + paths + }); + drop(runtime); + for path in paths { + std::fs::remove_dir_all(path).unwrap(); + } +} + +macro_rules! upstream_tests { + ($($name:ident),* $(,)?) => { $( + #[tokio::test] + async fn $name() { + let (path, store) = migration_store(); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + cdk_common::database::wallet::test::$name(db).await; + drop(store); + std::fs::remove_dir_all(path).unwrap(); + } + )* }; +} + +upstream_tests!( + add_and_get_mint, + add_mint_without_info, + remove_mint, + update_mint_url, + add_and_get_keysets, + get_keyset_by_id_in_transaction, + add_and_get_keys, + get_keys_in_transaction, + remove_keys, + add_and_get_mint_quote, + get_mint_quote_in_transaction, + remove_mint_quote, + add_and_get_melt_quote, + get_melt_quote_in_transaction, + remove_melt_quote, + add_mint_quote_optimistic_locking, + add_melt_quote_optimistic_locking, + add_and_get_proofs, + get_proofs_in_transaction, + update_proofs, + update_proofs_state, + filter_proofs_by_unit, + filter_proofs_by_state, + get_balance, + get_balance_by_state, + increment_keyset_counter, + keyset_counter_isolation, + increment_derivation_counter, + derivation_counters_are_independent, + derivation_counter_is_atomic, + add_and_get_transaction, + update_transaction_status, + same_proofs_in_different_sagas, + list_transactions, + filter_transactions_by_mint, + remove_transaction, + kvstore_write_and_read, + kvstore_list, + kvstore_update, + kvstore_remove, + kvstore_namespace_isolation, + add_and_get_saga, + update_saga_optimistic_locking, + delete_saga, + get_incomplete_sagas, + reserve_proofs, + release_proofs, + release_proofs_preserves_spent, + get_reserved_proofs, + reserve_proofs_already_reserved, + reserve_proofs_is_atomic, +); + +/// One-shot error before an operation or after it committed (lost acknowledgement). +/// Yielding every operation also models an asynchronous remote KV store. +struct FaultStore { + inner: Arc, + calls: Arc, + fail_at: usize, + after: bool, + migration_only: bool, +} + +type StoreFuture = Pin> + Send + 'static>>; +impl FaultStore { + fn wrap( + &self, namespace: &str, key: &str, future: impl FnOnce() -> StoreFuture + Send + 'static, + ) -> StoreFuture { + let tracked = !self.migration_only + || namespace == TRANSACTIONS_KEY + || key == "cdk_018_transaction_ids"; + let fail = tracked && self.calls.fetch_add(1, Ordering::SeqCst) + 1 == self.fail_at; + let after = self.after; + Box::pin(async move { + tokio::task::yield_now().await; + if fail && !after { + return Err(io::Error::new(io::ErrorKind::Other, "injected before operation")); + } + let result = future().await; + if fail && after { + return Err(io::Error::new(io::ErrorKind::Other, "injected lost acknowledgement")); + } + result + }) + } +} +impl DynStore for FaultStore { + fn read_async(&self, p: &str, s: &str, k: &str) -> StoreFuture> { + let (inner, p, ns, key) = (self.inner.clone(), p.to_owned(), s.to_owned(), k.to_owned()); + self.wrap(s, k, move || inner.read_async(&p, &ns, &key)) + } + fn write_async(&self, p: &str, s: &str, k: &str, value: Vec) -> StoreFuture<()> { + let (inner, p, ns, key) = (self.inner.clone(), p.to_owned(), s.to_owned(), k.to_owned()); + self.wrap(s, k, move || inner.write_async(&p, &ns, &key, value)) + } + fn remove_async(&self, p: &str, s: &str, k: &str, lazy: bool) -> StoreFuture<()> { + let (inner, p, ns, key) = (self.inner.clone(), p.to_owned(), s.to_owned(), k.to_owned()); + self.wrap(s, k, move || inner.remove_async(&p, &ns, &key, lazy)) + } + fn list_async(&self, p: &str, s: &str) -> StoreFuture> { + let (inner, p, ns) = (self.inner.clone(), p.to_owned(), s.to_owned()); + self.wrap(s, "", move || inner.list_async(&p, &ns)) + } + fn list_paginated_async( + &self, p: &str, s: &str, token: Option, + ) -> StoreFuture { + let (inner, p, ns) = (self.inner.clone(), p.to_owned(), s.to_owned()); + self.wrap(s, "", move || inner.list_paginated_async(&p, &ns, token)) + } +} + +#[derive(serde::Deserialize)] +struct LegacyRecord { + namespace: String, + key: String, + value: String, +} + +fn legacy_records() -> Vec { + serde_json::from_str(include_str!("../../../tests/fixtures/cdk016-wallet.json")).unwrap() +} + +async fn seed_legacy_store(store: &Arc) { + for record in legacy_records() { + store + .write_async( + CASHU_PRIMARY_KEY, + &record.namespace, + &record.key, + record.value.into_bytes(), + ) + .await + .unwrap(); + } +} + +async fn verify_upgraded_store(store: &Arc) { + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + for record in legacy_records() { + let key = if record.namespace == TRANSACTIONS_KEY { + let tx: Transaction = serde_json::from_slice(record.value.as_bytes()).unwrap(); + assert_eq!(db.get_transaction(tx.id()).await.unwrap(), Some(tx.clone())); + assert_eq!(tx.status, cdk::wallet::types::TransactionStatus::Completed); + if record.key != tx.id().to_string() { + assert!(db.get_transaction(record.key.parse().unwrap()).await.unwrap().is_none()); + } + tx.id().to_string() + } else { + record.key + }; + assert_eq!( + store.read_async(CASHU_PRIMARY_KEY, &record.namespace, &key).await.unwrap(), + record.value.as_bytes() + ); + } + assert_eq!(db.list_transactions(None, None, None).await.unwrap().len(), 2); + assert_eq!(db.get_balance(None, None, Some(vec![State::Unspent])).await.unwrap(), 1); + let proofs = db.get_proofs(None, None, None, None).await.unwrap(); + assert_eq!(proofs.len(), 4); + assert!(proofs.iter().all(|proof| proof.derivation_index.is_none())); + assert_eq!(db.get_mints().await.unwrap().len(), 1); + let quote = db.get_mint_quotes().await.unwrap().remove(0); + assert_eq!(quote.secret_key.as_ref().unwrap().to_secret_hex(), format!("{:064x}", 1)); + assert_eq!(quote.updated_at, 0); + assert_eq!(quote.version, 0); + let quote = db.get_melt_quote("melt-quote").await.unwrap().unwrap(); + assert_eq!(quote.payment_proof.as_deref(), Some("fixture-preimage")); + assert_eq!(quote.version, 2); + let sagas = db.get_incomplete_sagas().await.unwrap(); + assert_eq!(sagas.len(), 1); + assert_eq!(sagas[0].version, 0); + assert_eq!(db.get_reserved_proofs(&uuid::Uuid::from_u128(1)).await.unwrap().len(), 1); + assert_eq!( + db.increment_keyset_counter(&Id::from_str("009a1f293253e41e").unwrap(), 0).await.unwrap(), + 41 + ); +} + +#[tokio::test] +async fn database_written_by_cdk016_reopens_without_losing_records() { + let (path, store) = migration_store(); + seed_legacy_store(&store).await; + verify_upgraded_store(&store).await; + drop(store); + let reopened: Arc = Arc::new( + ldk_node::io::sqlite_store::SqliteStore::new( + path.clone(), + Some("orange.sqlite".into()), + None, + ) + .unwrap(), + ); + verify_upgraded_store(&reopened).await; + drop(reopened); + std::fs::remove_dir_all(path).unwrap(); +} + +#[tokio::test] +async fn migration_recovers_from_every_io_failure_before_and_after_commit() { + let (path, store) = migration_store(); + seed_legacy_store(&store).await; + let calls = Arc::new(AtomicUsize::new(0)); + let wrapper = Arc::new(FaultStore { + inner: store.clone(), + calls: calls.clone(), + fail_at: usize::MAX, + after: false, + migration_only: true, + }); + drop(CashuKvDatabase::new(wrapper).await.unwrap()); + let total = calls.load(Ordering::SeqCst); + assert!(total >= 8, "must cover copying, deleting and committing the migration marker"); + drop(store); + std::fs::remove_dir_all(path).unwrap(); + for after in [false, true] { + for fail_at in 1..=total { + let (path, store) = migration_store(); + seed_legacy_store(&store).await; + let wrapper = Arc::new(FaultStore { + inner: store.clone(), + calls: Arc::new(AtomicUsize::new(0)), + fail_at, + after, + migration_only: true, + }); + assert!( + CashuKvDatabase::new(wrapper).await.is_err(), + "failure {fail_at}, after={after}" + ); + // Drop every SQLite handle: recover using only the durable state. + drop(store); + let reopened: Arc = Arc::new( + ldk_node::io::sqlite_store::SqliteStore::new( + path.clone(), + Some("orange.sqlite".into()), + None, + ) + .unwrap(), + ); + verify_upgraded_store(&reopened).await; + verify_upgraded_store(&reopened).await; + drop(reopened); + std::fs::remove_dir_all(path).unwrap(); + } + } +} + +#[tokio::test] +async fn derivation_counter_errors_never_reset_or_reuse_a_committed_index() { + for after in [false, true] { + for fail_at in [1, 2] { + let (path, store) = migration_store(); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + assert_eq!(db.increment_derivation_counter("nut20_quote", 7).await.unwrap(), 7); + drop(db); + let calls = Arc::new(AtomicUsize::new(0)); + let wrapper = Arc::new(FaultStore { + inner: store.clone(), + calls: calls.clone(), + fail_at: usize::MAX, + after, + migration_only: false, + }); + let mut db = CashuKvDatabase::new(wrapper).await.unwrap(); + db.store = Arc::new(FaultStore { + inner: store.clone(), + calls: calls.clone(), + fail_at, + after, + migration_only: false, + }); + calls.store(0, Ordering::SeqCst); + assert!(db.increment_derivation_counter("nut20_quote", 1).await.is_err()); + drop(db); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let committed = if after && fail_at == 2 { 8 } else { 7 }; + assert_eq!(db.increment_derivation_counter("nut20_quote", 0).await.unwrap(), committed); + assert_eq!( + db.increment_derivation_counter("nut20_quote", 1).await.unwrap(), + committed + 1 + ); + drop(db); + drop(store); + std::fs::remove_dir_all(path).unwrap(); + } + } +} + +#[tokio::test] +async fn reservation_write_failure_never_leaves_a_partial_proof_set_after_restart() { + for after in [false, true] { + let (path, store) = migration_store(); + let mut db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let proofs = vec![proof_info("fault-proof-1", 1), proof_info("fault-proof-2", 2)]; + db.update_proofs(proofs.clone(), vec![]).await.unwrap(); + db.store = Arc::new(FaultStore { + inner: store.clone(), + calls: Arc::new(AtomicUsize::new(0)), + fail_at: 1, + after, + migration_only: false, + }); + let operation = uuid::Uuid::from_u128(1); + assert!(db.reserve_proofs(proofs.iter().map(|p| p.y).collect(), &operation).await.is_err()); + // A lost acknowledgement must not leave this still-running adapter + // presenting reserved proofs as spendable from its stale cache. + assert_eq!( + db.get_reserved_proofs(&operation).await.unwrap().len(), + if after { 2 } else { 0 } + ); + if after { + assert!(db.reserve_proofs(vec![proofs[0].y], &uuid::Uuid::from_u128(2)).await.is_err()); + } + drop(db); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + assert_eq!( + db.get_reserved_proofs(&operation).await.unwrap().len(), + if after { 2 } else { 0 } + ); + assert_eq!( + db.get_balance(None, None, Some(vec![State::Unspent])).await.unwrap(), + if after { 0 } else { 3 } + ); + drop(db); + drop(store); + std::fs::remove_dir_all(path).unwrap(); + } +} + +#[tokio::test] +async fn upgraded_legacy_reservation_cannot_be_stolen_by_a_new_operation() { + let (path, store) = migration_store(); + seed_legacy_store(&store).await; + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let owner = uuid::Uuid::from_u128(1); + let reserved = db.get_reserved_proofs(&owner).await.unwrap(); + assert!(db.reserve_proofs(vec![reserved[0].y], &uuid::Uuid::from_u128(2)).await.is_err()); + assert_eq!(db.get_reserved_proofs(&owner).await.unwrap(), reserved); + drop(db); + drop(store); + std::fs::remove_dir_all(path).unwrap(); +} + +#[cfg(feature = "_cashu-tests")] +#[path = "cashu_store_live_upgrade_tests.rs"] +mod live; + +#[tokio::test] +async fn legacy_release_retries_safely_after_every_storage_failure() { + for after in [false, true] { + // Reservation-index read, proof snapshot write, reservation-index delete. + for fail_at in 1..=3 { + let (path, store) = migration_store(); + seed_legacy_store(&store).await; + let mut db = CashuKvDatabase::new(store.clone()).await.unwrap(); + db.store = Arc::new(FaultStore { + inner: store.clone(), + calls: Arc::new(AtomicUsize::new(0)), + fail_at, + after, + migration_only: false, + }); + let operation = uuid::Uuid::from_u128(1); + assert!(db.release_proofs(&operation).await.is_err()); + drop(db); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + db.release_proofs(&operation).await.unwrap(); + db.release_proofs(&operation).await.unwrap(); + assert_eq!(db.get_balance(None, None, Some(vec![State::Unspent])).await.unwrap(), 3); + assert_eq!(db.get_balance(None, None, Some(vec![State::Pending])).await.unwrap(), 4); + assert_eq!(db.get_balance(None, None, Some(vec![State::Spent])).await.unwrap(), 8); + assert!(db.get_reserved_proofs(&operation).await.unwrap().is_empty()); + drop(db); + drop(store); + std::fs::remove_dir_all(path).unwrap(); + } + } +} + +#[tokio::test] +async fn concurrent_quote_and_saga_updates_have_one_winner() { + let (path, store) = migration_store(); + seed_legacy_store(&store).await; + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let quote = db.get_mint_quote("mint-quote").await.unwrap().unwrap(); + let (a, b) = tokio::join!(db.add_mint_quote(quote.clone()), db.add_mint_quote(quote)); + assert_eq!([a.is_ok(), b.is_ok()].into_iter().filter(|ok| *ok).count(), 1); + assert_eq!(db.get_mint_quote("mint-quote").await.unwrap().unwrap().version, 1); + let mut quote = db.get_melt_quote("melt-quote").await.unwrap().unwrap(); + quote.payment_proof = None; // A later status response may omit the proof. + let (a, b) = tokio::join!(db.add_melt_quote(quote.clone()), db.add_melt_quote(quote)); + assert_eq!([a.is_ok(), b.is_ok()].into_iter().filter(|ok| *ok).count(), 1); + let quote = db.get_melt_quote("melt-quote").await.unwrap().unwrap(); + assert_eq!(quote.version, 3); + assert_eq!(quote.payment_proof.as_deref(), Some("fixture-preimage")); + let id = uuid::Uuid::from_u128(1); + let mut saga = db.get_saga(&id).await.unwrap().unwrap(); + saga.version += 1; + let (a, b) = tokio::join!(db.update_saga(saga.clone()), db.update_saga(saga)); + assert_ne!(a.unwrap(), b.unwrap()); + assert_eq!(db.get_saga(&id).await.unwrap().unwrap().version, 1); + drop(db); + drop(store); + std::fs::remove_dir_all(path).unwrap(); +} + +#[tokio::test] +async fn legacy_keyset_counter_does_not_reset_on_read_error_or_overflow() { + let (path, store) = migration_store(); + seed_legacy_store(&store).await; + let mut db = CashuKvDatabase::new(store.clone()).await.unwrap(); + let keyset = Id::from_str("009a1f293253e41e").unwrap(); + db.store = Arc::new(FaultStore { + inner: store.clone(), + calls: Arc::new(AtomicUsize::new(0)), + fail_at: 1, + after: false, + migration_only: false, + }); + assert!(db.increment_keyset_counter(&keyset, 1).await.is_err()); + assert_eq!(db.increment_keyset_counter(&keyset, 0).await.unwrap(), 41); + assert!(db.increment_keyset_counter(&keyset, u32::MAX).await.is_err()); + assert_eq!(db.increment_keyset_counter(&keyset, 0).await.unwrap(), 41); + drop(db); + drop(store); + std::fs::remove_dir_all(path).unwrap(); +} diff --git a/orange-sdk/src/trusted_wallet/cashu/mod.rs b/orange-sdk/src/trusted_wallet/cashu/mod.rs index db1dd13..aa88d0a 100644 --- a/orange-sdk/src/trusted_wallet/cashu/mod.rs +++ b/orange-sdk/src/trusted_wallet/cashu/mod.rs @@ -12,7 +12,7 @@ use ldk_node::bitcoin::hashes::Hash; use ldk_node::bitcoin::hashes::sha256::Hash as Sha256; use ldk_node::bitcoin::hex::FromHex; use ldk_node::lightning::util::logger::Logger as _; -use ldk_node::lightning::{log_error, log_info}; +use ldk_node::lightning::{log_error, log_info, log_warn}; use ldk_node::lightning_invoice::Bolt11Invoice; use ldk_node::lightning_types::payment::{PaymentHash, PaymentPreimage}; @@ -26,7 +26,7 @@ use cdk::nuts::nut23::Amountless; use cdk::nuts::{CurrencyUnit, MeltQuoteState}; use cdk::wallet::MintQuote; use cdk::wallet::Wallet; -use cdk::wallet::types::{Transaction, TransactionDirection}; +use cdk::wallet::types::{Transaction, TransactionDirection, TransactionStatus}; use cdk::{Amount as CdkAmount, StreamExt}; use graduated_rebalancer::ReceivedLightningPayment; @@ -181,13 +181,7 @@ impl TrustedWalletInterface for Cashu { TrustedError::WalletOperationFailed(format!("Failed to list transactions: {e}")) })?; - // Convert CDK Transaction to Payment - let payments = transactions - .into_iter() - .map(|t| Self::convert_transaction_to_payment(t, &self.unit)) - .collect::, _>>()?; - - Ok(payments) + Self::convert_transactions_to_payments(transactions, &self.unit) }) } @@ -375,7 +369,8 @@ impl TrustedWalletInterface for Cashu { let hex = payment_hash.to_lower_hex_string(); let tx = txs.iter().find(|tx| { - tx.metadata.get(PAYMENT_HASH_METADATA_KEY).is_some_and(|h| h == &hex) + tx.status == TransactionStatus::Completed + && tx.metadata.get(PAYMENT_HASH_METADATA_KEY).is_some_and(|h| h == &hex) }); if let Some(tx) = tx { @@ -478,6 +473,32 @@ impl Cashu { })?, ); + // Recover before exposing the wallet or starting tasks that create operations. + // This must run on every open, independently of the seed-restore flag. + let recovery = cashu_wallet.recover_incomplete_sagas().await.map_err(|e| { + InitFailure::TrustedFailure(TrustedError::WalletOperationFailed(format!( + "Failed to recover Cashu operations: {e}" + ))) + })?; + if recovery.failed != 0 { + // A mint outage can fail individual recoveries. Keep their reservations + // intact and allow access to the rest of the wallet, including LDK funds. + log_warn!( + logger, + "Failed to recover {} Cashu operations; continuing startup, retry on next open", + recovery.failed + ); + } + if !recovery.is_empty() { + log_info!( + logger, + "Cashu recovery: {} recovered, {} compensated, {} still pending (retry on next startup)", + recovery.recovered, + recovery.compensated, + recovery.skipped + ); + } + let supports_bolt12 = Arc::new(std::sync::atomic::AtomicBool::new(false)); let supports_mpp = Arc::new(std::sync::atomic::AtomicBool::new(false)); { @@ -849,6 +870,35 @@ impl Cashu { id_array } + fn convert_transactions_to_payments( + transactions: Vec, unit: &CurrencyUnit, + ) -> Result, TrustedError> { + let mut outgoing_quotes: HashMap = HashMap::new(); + let mut selected: Vec = Vec::new(); + // CDK retains each retry under its saga ID, while Orange identifies an + // outgoing payment by quote. A settled attempt wins; otherwise expose the + // newest attempt. UUIDv7 saga IDs break ties in second-resolution timestamps. + let priority = |tx: &Transaction| { + (tx.status == TransactionStatus::Completed, tx.timestamp, tx.saga_id) + }; + for transaction in transactions { + if transaction.direction == TransactionDirection::Outgoing { + if let Some(quote_id) = &transaction.quote_id { + if let Some(&index) = outgoing_quotes.get(quote_id) { + if priority(&transaction) > priority(&selected[index]) { + selected[index] = transaction; + } + continue; + } + outgoing_quotes.insert(quote_id.clone(), selected.len()); + } + } + // Reusable incoming quotes can represent multiple actual receipts. + selected.push(transaction); + } + selected.into_iter().map(|tx| Self::convert_transaction_to_payment(tx, unit)).collect() + } + /// Convert a CDK Transaction to a Payment struct fn convert_transaction_to_payment( transaction: Transaction, unit: &CurrencyUnit, @@ -865,9 +915,11 @@ impl Cashu { let outbound = transaction.direction == TransactionDirection::Outgoing; - // For Cashu, we'll assume all completed transactions are successful - // and all others are pending. CDK doesn't have a direct status mapping. - let status = TxStatus::Completed; // Assume completed since it's in the transaction list + let status = match transaction.status { + TransactionStatus::Pending => TxStatus::Pending, + TransactionStatus::Completed => TxStatus::Completed, + TransactionStatus::Failed => TxStatus::Failed, + }; // Convert timestamp to Duration since epoch let time_since_epoch = Duration::from_secs(transaction.timestamp); @@ -941,7 +993,7 @@ impl Cashu { fee = fee.saturating_add(convert_amount(keyset_fee, &self.unit)?); } - let active_keyset = self.cashu_wallet.get_active_keyset().await.map_err(|e| { + let active_keyset = self.cashu_wallet.active_keyset().await.map_err(|e| { TrustedError::WalletOperationFailed(format!("Failed to get active keyset: {e}")) })?; let fee_and_amounts = diff --git a/orange-sdk/tests/fixtures/cdk016-generator/Cargo.toml b/orange-sdk/tests/fixtures/cdk016-generator/Cargo.toml new file mode 100644 index 0000000..c73a75e --- /dev/null +++ b/orange-sdk/tests/fixtures/cdk016-generator/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "orange-cdk016-fixture" +version = "0.0.0" +edition = "2024" +publish = false + +[workspace] + +[dependencies] +cdk = { version = "=0.16.0", default-features = false, features = ["wallet"] } +ldk-node = { git = "https://github.com/lightningdevkit/ldk-node", rev = "0cea34122767f0b86ebaf29c03f49eb38a85fc00" } +async-trait = "0.1" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1", features = ["full"] } +uuid = "1" diff --git a/orange-sdk/tests/fixtures/cdk016-generator/README.md b/orange-sdk/tests/fixtures/cdk016-generator/README.md new file mode 100644 index 0000000..dc21446 --- /dev/null +++ b/orange-sdk/tests/fixtures/cdk016-generator/README.md @@ -0,0 +1,71 @@ +# CDK 0.16 upgrade fixtures + +`../cdk016-wallet.json` is an export of the raw KV values written by Orange's +original `CashuKvDatabase` at commit `a3aa1c59`, compiled against CDK **0.16.0**. +Values are UTF-8 strings so tests can compare the exact persisted bytes. All +secrets, invoices and proofs in this static fixture are synthetic test values; +they do not represent funds. The generator calls the old adapter's write methods, +including its original proof-reservation implementation, rather than hand-writing +serialized records. + +The standalone generator is excluded from the main Cargo workspace. It takes the +old adapter sources through build-time environment variables, preventing an +accidental test against the new adapter. Cargo's configured target/cache settings +still apply. + +From the repository root, using the repository's Rust build environment: + +```sh +scratch=$(mktemp -d /data/rust/tmp/orange-cdk016.XXXXXXXX) +jj file show -r a3aa1c59 orange-sdk/src/trusted_wallet/cashu/cashu_store.rs > "$scratch/cashu_store.rs" +jj file show -r a3aa1c59 orange-sdk/src/dyn_store.rs > "$scratch/dyn_store.rs" +# include! does not accept the old module's inner documentation comments. +sed -i '/^\/\/!/d' "$scratch/dyn_store.rs" +export ORANGE_OLD_CASHU_STORE="$scratch/cashu_store.rs" +export ORANGE_OLD_DYN_STORE="$scratch/dyn_store.rs" +cargo run --manifest-path orange-sdk/tests/fixtures/cdk016-generator/Cargo.toml -- "$scratch/database" > orange-sdk/tests/fixtures/cdk016-wallet.json +``` + +The generator also supports `DATABASE_DIR MINT_URL INVOICE [submit]`. It creates +an actual old-version wallet, mints 64 sats, and prepares an 8-sat melt. Without +`submit` it exits without destructor cleanup. With `submit`, it confirms the +payment; the integration test holds the mint's response and kills the process +before that response reaches the old wallet. These tests use real Cashu +signatures from a local CDK mint and a fake Lightning backend. Passing `issue` +instead of `INVOICE` lets the test hold the initial mint response and kill the +process with its issue saga still in `MintRequested`. + +All upgrades go through Orange's production `Cashu::init`, with the seed-restore +flag already set and Orange's `Seed64([7; 64])` derivation. The suite verifies: + +- A prepared melt releases its reservation, and the upgraded wallet can spend. +- A paid melt with a lost response recovers its change and exactly one completed + transaction. Paying 8 sats plus a 1-sat fee leaves 55 sats; the 2-sat fee reserve + requires recovering 1 sat of change. +- A failed melt returns the full 64-sat balance without a completed transaction. +- A pending melt keeps its 10 sats reserved across repeated startup attempts. +- HTTP 503 responses during recovery preserve the reservation, and reopening + once the mint is available recovers the paid melt and change. +- An issue saga with a lost mint response does not block startup while the mint + returns HTTP 503; reopening online recovers all 64 sats exactly once. + +Repeated startup must neither resubmit the melt nor duplicate payment history. + +```sh +bash ci/test-cashu-upgrade.sh +``` + +Run this from the repository root in the Rust development environment. It builds +the generator from the pinned old commit and discovers Cargo's actual executable +path, honoring configured target directories. It uses Git to read the pinned sources; +the Git checkout (including a colocated JJ checkout) must contain commit `a3aa1c592625c45314004987430498307b9a1819`. These +tests remain ignored in ordinary `cargo test` runs because they need that separate +generator, but the Cashu CI job explicitly runs this script on every build. CI caches +both Cargo workspaces, including the generator's separate CDK 0.16 dependency tree; +a cold build still needs to compile both versions. + +The ordinary Cashu unit suite runs the exported-data and before/after I/O failure +tests without this executable or any external services. The asynchronous fault +wrapper covers lost acknowledgements, but is not a deployed VSS-server test. +The adapter requires one active instance per store; these tests do not establish +multi-process or multi-device transactional guarantees. diff --git a/orange-sdk/tests/fixtures/cdk016-generator/src/main.rs b/orange-sdk/tests/fixtures/cdk016-generator/src/main.rs new file mode 100644 index 0000000..c00a33e --- /dev/null +++ b/orange-sdk/tests/fixtures/cdk016-generator/src/main.rs @@ -0,0 +1,209 @@ +//! Run against the original adapter, extracted from Orange commit a3aa1c59. +#![allow(dead_code)] +mod dyn_store { + include!(env!("ORANGE_OLD_DYN_STORE")); +} +mod trusted_wallet { + #[derive(Debug)] + pub enum TrustedError { + IOError(ldk_node::lightning::io::Error), + } + pub mod cashu_store { + include!(env!("ORANGE_OLD_CASHU_STORE")); + } +} +use cdk::{ + Amount, + cdk_database::WalletDatabase, + nuts::*, + secret::Secret, + types::ProofInfo, + wallet::{MeltQuote, MintQuote, types::*}, +}; +use ldk_node::lightning::util::persist::KVStore; +use std::{str::FromStr, sync::Arc}; +use trusted_wallet::cashu_store::CashuKvDatabase; +#[tokio::main] +async fn main() { + let path = std::env::args().nth(1).expect("fresh SQLite directory"); + let store: Arc = Arc::new( + ldk_node::io::sqlite_store::SqliteStore::new( + path.into(), + Some("orange.sqlite".into()), + None, + ) + .unwrap(), + ); + let db = CashuKvDatabase::new(store.clone()).await.unwrap(); + if let Some(url) = std::env::args().nth(2) { + // Match Orange's Seed64 derivation so the new wallet can use production init. + use ldk_node::bitcoin::hashes::{Hash, sha256}; + let hash = sha256::Hash::hash(&[7; 64]).to_byte_array(); + let mut seed = [0; 64]; + seed[..32].copy_from_slice(&hash); + seed[32..].copy_from_slice(&hash); + let wallet = + cdk::wallet::Wallet::new(&url, CurrencyUnit::Sat, Arc::new(db), seed, None).unwrap(); + let quote = + wallet.mint_quote(PaymentMethod::BOLT11, Some(64.into()), None, None).await.unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(30), async { + loop { + if wallet.check_mint_quote_status("e.id).await.unwrap().state + == MintQuoteState::Paid + { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + }) + .await + .unwrap(); + let issue_only = std::env::args().nth(3).as_deref() == Some("issue"); + let result = wallet.mint("e.id, Default::default(), None).await; + if issue_only { + panic!("test server unexpectedly delivered the mint response: {result:?}"); + } + result.unwrap(); + assert_eq!(wallet.total_balance().await.unwrap(), Amount::from(64)); + let invoice = std::env::args().nth(3).expect("external fake invoice"); + let quote = wallet.melt_quote(PaymentMethod::BOLT11, invoice, None, None).await.unwrap(); + let prepared = wallet.prepare_melt("e.id, Default::default()).await.unwrap(); + println!("Prepared old-version melt {}", prepared.operation_id()); + if std::env::args().nth(4).as_deref() == Some("submit") { + // The test server holds the response after processing the payment. The + // parent kills this process there, bypassing wallet compensation/Drop. + let result = prepared.confirm().await; + panic!("test server unexpectedly delivered the melt response: {result:?}"); + } + // A process exit bypasses any Drop cleanup, modelling a crash after reservation. + std::process::exit(0); + } + let mint = cdk::mint_url::MintUrl::from_str("https://mint.example.com").unwrap(); + let keyset = Id::from_str("009a1f293253e41e").unwrap(); + db.add_mint(mint.clone(), Some(MintInfo::default())).await.unwrap(); + db.add_mint_keysets( + mint.clone(), + vec![KeySetInfo { + id: keyset, + unit: CurrencyUnit::Sat, + active: true, + input_fee_ppk: 100, + final_expiry: None, + }], + ) + .await + .unwrap(); + db.increment_keyset_counter(&keyset, 41).await.unwrap(); + let mut proofs = Vec::new(); + for (i, state) in + [State::Unspent, State::Reserved, State::Pending, State::Spent].into_iter().enumerate() + { + let proof = Proof::new( + Amount::from(1u64 << i), + keyset, + Secret::new(format!("fixture-only-secret-{i}")), + PublicKey::from_str( + "024369d2d22a80ecf78f3937da9d5f30c1b9f74f0c32684d583cca0fa6a61cdcfc", + ) + .unwrap(), + ); + proofs.push(ProofInfo::new(proof, mint.clone(), state, CurrencyUnit::Sat).unwrap()); + } + db.update_proofs(proofs.clone(), vec![]).await.unwrap(); + let operation = uuid::Uuid::from_u128(1); + db.reserve_proofs(vec![proofs[1].y], &operation).await.unwrap(); + for (i, saga_id) in [Some(operation), None].into_iter().enumerate() { + db.add_transaction(Transaction { + mint_url: mint.clone(), + direction: TransactionDirection::Outgoing, + amount: 10.into(), + fee: 1.into(), + unit: CurrencyUnit::Sat, + ys: vec![proofs[i].y], + timestamp: 42, + memo: Some("fixture".into()), + metadata: Default::default(), + quote_id: Some(format!("quote-{i}")), + payment_request: Some("fixture-invoice".into()), + payment_proof: Some("fixture-preimage".into()), + payment_method: Some(PaymentMethod::BOLT11), + saga_id, + }) + .await + .unwrap(); + } + let mut quote = MintQuote::new( + "mint-quote".into(), + mint.clone(), + PaymentMethod::BOLT11, + Some(10.into()), + CurrencyUnit::Sat, + "fixture-invoice".into(), + 123, + Some( + SecretKey::from_hex("0000000000000000000000000000000000000000000000000000000000000001") + .unwrap(), + ), + ); + quote.amount_paid = 10.into(); + quote.state = MintQuoteState::Paid; + db.add_mint_quote(quote).await.unwrap(); + db.add_melt_quote(MeltQuote { + id: "melt-quote".into(), + mint_url: Some(mint.clone()), + unit: CurrencyUnit::Sat, + amount: 10.into(), + request: "fixture-invoice".into(), + fee_reserve: 1.into(), + state: MeltQuoteState::Pending, + expiry: 123, + payment_preimage: Some("fixture-preimage".into()), + payment_method: PaymentMethod::BOLT11, + used_by_operation: Some(operation.to_string()), + version: 2, + }) + .await + .unwrap(); + let mut saga = WalletSaga::new( + operation, + WalletSagaState::Melt(MeltSagaState::PaymentPending), + 10.into(), + mint, + CurrencyUnit::Sat, + OperationData::Melt(MeltOperationData { + quote_id: "melt-quote".into(), + amount: 10.into(), + fee_reserve: 1.into(), + counter_start: Some(40), + counter_end: Some(41), + change_amount: None, + change_blinded_messages: None, + }), + ); + saga.created_at = 42; + saga.updated_at = 43; + db.add_saga(saga).await.unwrap(); + let mut records = Vec::new(); + for namespace in [ + "mints", + "mint_keysets", + "keysets_table", + "keyset_u32_mapping", + "keyset_counters", + "proofs", + "transactions", + "mint_quotes", + "melt_quotes", + "sagas", + "proof_reservations", + ] { + let mut keys = KVStore::list(store.as_ref(), "cashu_wallet", namespace).await.unwrap(); + keys.sort(); + for key in keys { + let value = + KVStore::read(store.as_ref(), "cashu_wallet", namespace, &key).await.unwrap(); + records.push(serde_json::json!({"namespace": namespace, "key": key, "value": String::from_utf8(value).unwrap()})); + } + } + println!("{}", serde_json::to_string_pretty(&records).unwrap()); +} diff --git a/orange-sdk/tests/fixtures/cdk016-wallet.json b/orange-sdk/tests/fixtures/cdk016-wallet.json new file mode 100644 index 0000000..6a4a56e --- /dev/null +++ b/orange-sdk/tests/fixtures/cdk016-wallet.json @@ -0,0 +1,67 @@ +[ + { + "key": "info_374d58f396cb3152", + "namespace": "mints", + "value": "{\"nuts\":{\"4\":{\"methods\":[],\"disabled\":false},\"5\":{\"methods\":[],\"disabled\":false},\"7\":{\"supported\":false},\"8\":{\"supported\":false},\"9\":{\"supported\":false},\"10\":{\"supported\":false},\"11\":{\"supported\":false},\"12\":{\"supported\":false},\"14\":{\"supported\":false},\"17\":{\"supported\":[]},\"19\":{\"ttl\":null,\"cached_endpoints\":[]},\"20\":{\"supported\":false}}}" + }, + { + "key": "mint_374d58f396cb3152", + "namespace": "mints", + "value": "\"https://mint.example.com\"" + }, + { + "key": "keysets_374d58f396cb3152", + "namespace": "mint_keysets", + "value": "[{\"id\":\"009a1f293253e41e\",\"unit\":\"sat\",\"active\":true,\"input_fee_ppk\":100}]" + }, + { + "key": "keyset_009a1f293253e41e", + "namespace": "keysets_table", + "value": "{\"id\":\"009a1f293253e41e\",\"unit\":\"sat\",\"active\":true,\"input_fee_ppk\":100}" + }, + { + "key": "u32_864559728", + "namespace": "keyset_u32_mapping", + "value": "009a1f293253e41e" + }, + { + "key": "009a1f293253e41e", + "namespace": "keyset_counters", + "value": "41" + }, + { + "key": "proof_set", + "namespace": "proofs", + "value": "{\"schema_version\":1,\"proofs\":[{\"proof\":{\"amount\":1,\"id\":\"009a1f293253e41e\",\"secret\":\"fixture-only-secret-0\",\"C\":\"024369d2d22a80ecf78f3937da9d5f30c1b9f74f0c32684d583cca0fa6a61cdcfc\"},\"y\":\"022485c6bc8f3f9798a8de63f8da2406107bcf3593b21d1d8e1da62774ed5b7265\",\"mint_url\":\"https://mint.example.com\",\"state\":\"UNSPENT\",\"spending_condition\":null,\"unit\":\"sat\"},{\"proof\":{\"amount\":2,\"id\":\"009a1f293253e41e\",\"secret\":\"fixture-only-secret-1\",\"C\":\"024369d2d22a80ecf78f3937da9d5f30c1b9f74f0c32684d583cca0fa6a61cdcfc\"},\"y\":\"0292729b6de3218dcd12d1ddb984e0558f57cdd9d3fb1982a50a88a06d040520b0\",\"mint_url\":\"https://mint.example.com\",\"state\":\"RESERVED\",\"spending_condition\":null,\"unit\":\"sat\"},{\"proof\":{\"amount\":4,\"id\":\"009a1f293253e41e\",\"secret\":\"fixture-only-secret-2\",\"C\":\"024369d2d22a80ecf78f3937da9d5f30c1b9f74f0c32684d583cca0fa6a61cdcfc\"},\"y\":\"025e336af57063adbf0035a8874c6ebe4538decc2efbef127c9d7345bbeb383dc1\",\"mint_url\":\"https://mint.example.com\",\"state\":\"PENDING\",\"spending_condition\":null,\"unit\":\"sat\"},{\"proof\":{\"amount\":8,\"id\":\"009a1f293253e41e\",\"secret\":\"fixture-only-secret-3\",\"C\":\"024369d2d22a80ecf78f3937da9d5f30c1b9f74f0c32684d583cca0fa6a61cdcfc\"},\"y\":\"023616cee749e6f69f7baa85cb544a60dfa3910f730789fbd42eb39c1044f71543\",\"mint_url\":\"https://mint.example.com\",\"state\":\"SPENT\",\"spending_condition\":null,\"unit\":\"sat\"}]}" + }, + { + "key": "b62568ef5f7f4037381fce92bc2a745fdb72aa6830de20e77da8c8fe2691123c", + "namespace": "transactions", + "value": "{\"mint_url\":\"https://mint.example.com\",\"direction\":\"Outgoing\",\"amount\":10,\"fee\":1,\"unit\":\"sat\",\"ys\":[\"022485c6bc8f3f9798a8de63f8da2406107bcf3593b21d1d8e1da62774ed5b7265\"],\"timestamp\":42,\"memo\":\"fixture\",\"metadata\":{},\"quote_id\":\"quote-0\",\"payment_request\":\"fixture-invoice\",\"payment_proof\":\"fixture-preimage\",\"payment_method\":\"bolt11\",\"saga_id\":\"00000000-0000-0000-0000-000000000001\"}" + }, + { + "key": "e785f0ddfaaee3184d68db33d13cad95988da3827ade370c6eb0a406e4312b64", + "namespace": "transactions", + "value": "{\"mint_url\":\"https://mint.example.com\",\"direction\":\"Outgoing\",\"amount\":10,\"fee\":1,\"unit\":\"sat\",\"ys\":[\"0292729b6de3218dcd12d1ddb984e0558f57cdd9d3fb1982a50a88a06d040520b0\"],\"timestamp\":42,\"memo\":\"fixture\",\"metadata\":{},\"quote_id\":\"quote-1\",\"payment_request\":\"fixture-invoice\",\"payment_proof\":\"fixture-preimage\",\"payment_method\":\"bolt11\",\"saga_id\":null}" + }, + { + "key": "mint-quote", + "namespace": "mint_quotes", + "value": "{\"id\":\"mint-quote\",\"mint_url\":\"https://mint.example.com\",\"payment_method\":\"bolt11\",\"amount\":10,\"unit\":\"sat\",\"request\":\"fixture-invoice\",\"state\":\"PAID\",\"expiry\":123,\"secret_key\":\"0000000000000000000000000000000000000000000000000000000000000001\",\"amount_issued\":0,\"amount_paid\":10,\"used_by_operation\":null,\"version\":0}" + }, + { + "key": "melt-quote", + "namespace": "melt_quotes", + "value": "{\"id\":\"melt-quote\",\"mint_url\":\"https://mint.example.com\",\"unit\":\"sat\",\"amount\":10,\"request\":\"fixture-invoice\",\"fee_reserve\":1,\"state\":\"PENDING\",\"expiry\":123,\"payment_preimage\":\"fixture-preimage\",\"payment_method\":\"bolt11\",\"used_by_operation\":\"00000000-0000-0000-0000-000000000001\",\"version\":2}" + }, + { + "key": "00000000-0000-0000-0000-000000000001", + "namespace": "sagas", + "value": "{\"id\":\"00000000-0000-0000-0000-000000000001\",\"kind\":\"melt\",\"state\":{\"type\":\"melt\",\"state\":\"payment_pending\"},\"amount\":10,\"mint_url\":\"https://mint.example.com\",\"unit\":\"sat\",\"quote_id\":\"melt-quote\",\"created_at\":42,\"updated_at\":43,\"data\":{\"kind\":\"melt\",\"data\":{\"quote_id\":\"melt-quote\",\"amount\":10,\"fee_reserve\":1,\"counter_start\":40,\"counter_end\":41,\"change_amount\":null}},\"version\":0}" + }, + { + "key": "00000000-0000-0000-0000-000000000001", + "namespace": "proof_reservations", + "value": "[\"0292729b6de3218dcd12d1ddb984e0558f57cdd9d3fb1982a50a88a06d040520b0\"]" + } +] diff --git a/orange-sdk/tests/test_utils.rs b/orange-sdk/tests/test_utils.rs index 9eb2e25..67a3b0c 100644 --- a/orange-sdk/tests/test_utils.rs +++ b/orange-sdk/tests/test_utils.rs @@ -517,6 +517,7 @@ async fn build_test_nodes() -> TestParams { t.local_addr().unwrap().port() }; let cdk_addr = SocketAddr::from_str(format!("127.0.0.1:{cdk_port}").as_str()).unwrap(); + let mem_db = Arc::new(cdk_sqlite::mint::memory::empty().await.unwrap()); let cdk = cdk_ldk_node::CdkLdkNodeBuilder::new( Network::Regtest, cdk_ldk_node::ChainSource::BitcoinRpc(BitcoinRpcConfig { @@ -529,6 +530,7 @@ async fn build_test_nodes() -> TestParams { tmp.to_str().unwrap().to_string(), FeeReserve { min_fee_reserve: Default::default(), percent_fee_reserve: 0.0 }, vec![cdk_addr.into()], + mem_db.clone(), ) .build() .unwrap(); @@ -544,7 +546,6 @@ async fn build_test_nodes() -> TestParams { let lsp_listen_clone = lsp_listen.clone(); let mint = { // build mint - let mem_db = Arc::new(cdk_sqlite::mint::memory::empty().await.unwrap()); let mut mint_seed: [u8; 64] = [0; 64]; rand::thread_rng().fill_bytes(&mut mint_seed); let mut builder = MintBuilder::new(mem_db.clone());