From 71635aa366b6120a4081e0a5ccc531bdb3b22e41 Mon Sep 17 00:00:00 2001 From: maxie-agent <330537927+maxie-agent@users.noreply.github.com> Date: Fri, 9 Oct 2026 23:37:31 +0000 Subject: [PATCH 1/5] feat(trade): lock-free discover, serve-handled list/take/cancel, unresponsive warning, readable status, fee dry-run - discover: ephemeral relay key, no owner.lock/journal/home creation; per-lot expected fees (unknown when a mint is unreachable) - serve binds /serve.sock (0600, same-uid peers, one bounded request per connection); list/take/cancel hand off with the client's exact bounds and run inside serve's loop (single writer) - counterparty_unresponsive display state after 180 s (take, serve log, status); display only - status prints a human summary; status --json unchanged (golden test) - --dry-run on list/take: both legs' fees, gross and max debit, zero side effects - tests/ux.rs safety tests --- crates/maxplayer-trade/src/coordinator.rs | 26 +- crates/maxplayer-trade/src/dryrun.rs | 267 ++++++ crates/maxplayer-trade/src/lib.rs | 5 + crates/maxplayer-trade/src/main.rs | 359 ++++++- crates/maxplayer-trade/src/observe.rs | 275 ++++++ crates/maxplayer-trade/src/report.rs | 222 +++++ crates/maxplayer-trade/src/serve.rs | 606 ++++++++++++ crates/maxplayer-trade/tests/receive.rs | 11 +- crates/maxplayer-trade/tests/ux.rs | 1055 +++++++++++++++++++++ 9 files changed, 2799 insertions(+), 27 deletions(-) create mode 100644 crates/maxplayer-trade/src/dryrun.rs create mode 100644 crates/maxplayer-trade/src/observe.rs create mode 100644 crates/maxplayer-trade/src/report.rs create mode 100644 crates/maxplayer-trade/src/serve.rs create mode 100644 crates/maxplayer-trade/tests/ux.rs diff --git a/crates/maxplayer-trade/src/coordinator.rs b/crates/maxplayer-trade/src/coordinator.rs index 90faebcfd..698941951 100644 --- a/crates/maxplayer-trade/src/coordinator.rs +++ b/crates/maxplayer-trade/src/coordinator.rs @@ -103,7 +103,7 @@ pub struct Swap { #[serde(default)] pub refund_generation: u32, } -fn terminal(s: &Swap) -> bool { +pub fn terminal(s: &Swap) -> bool { [ "complete", "complete_unclaimed", @@ -993,11 +993,33 @@ async fn recover_pass(home: &Path, j: &Journal, m: &Market) -> Result { Ok(deferred) } pub async fn run(home: &Path, j: &Journal, m: &mut Market, until: Option<&str>) -> Result<()> { + run_with(home, j, m, until, None).await +} +/// [`run`], also executing requests handed to `serve` over its local socket. Every request +/// runs here, in this one loop, between inbox messages and recovery ticks: serve stays the +/// single writer and each request keeps journal-before-effect ordering. +pub async fn run_with( + home: &Path, + j: &Journal, + m: &mut Market, + until: Option<&str>, + mut jobs: Option>, +) -> Result<()> { let mut tick = tokio::time::interval(std::time::Duration::from_secs(3)); + let mut warned = crate::observe::Warned::default(); loop { tokio::select! {e=m.inbox.recv()=>{if let Some(e)=e{match tokio::time::timeout(std::time::Duration::from_secs(60), handle(home,j,m,&e)).await { Ok(Ok(())) => {}, Ok(Err(err)) => eprintln!("trade message rejected: {err}"), Err(_) => eprintln!("trade message timed out; dropped; journal retained") - }}},_=tick.tick()=>recover(home,j,m).await?} + }}}, + Some(job) = async { match jobs.as_mut() { Some(rx) => rx.recv().await, None => std::future::pending().await } } => crate::serve::execute(home, j, m, job).await, + _=tick.tick()=>{ + recover(home,j,m).await?; + // Display only: derived from the journal, never fed back into the state machine. + let now = now(); + for u in warned.due(&crate::observe::unresponsive(home, now).unwrap_or_default()) { + eprintln!("{}", u.json(now)); + } + }} if let Some(id) = until { if let Some(s) = j.get::("swap", id).await? { if terminal(&s) { diff --git a/crates/maxplayer-trade/src/dryrun.rs b/crates/maxplayer-trade/src/dryrun.rs new file mode 100644 index 000000000..764f1d40b --- /dev/null +++ b/crates/maxplayer-trade/src/dryrun.rs @@ -0,0 +1,267 @@ +//! Fee previews: `list --dry-run`, `take --dry-run` and the per-lot fees in `discover`. +//! +//! Read-only by construction: published keysets come from the mint's `/v1/keysets`, input +//! selection reads the wallet SQLite file with `SQLITE_OPEN_READ_ONLY`, and the lot comes from a +//! relay query. Nothing is journaled, reserved, locked, created, published or sent to a maker. +//! The arithmetic is the coordinator's own (`gross`, `fee`, `check_lock_gross`) applied in the +//! same order as `mint::plan`, so the preview equals the plan the real command would build +//! against the same wallet and keysets. +use crate::{Asset, Lot, fee, gross, observe}; +use anyhow::{Context, Result, ensure}; +use sha2::{Digest, Sha256}; +use std::{collections::BTreeMap, path::Path, time::Duration}; + +/// One mint's published sat keysets: id -> (active, input_fee_ppk). +#[derive(Clone, Debug, Default)] +pub struct Keysets(pub BTreeMap); +impl Keysets { + /// Active sat keyset with the lowest input fee, as `Wallet::fetch_active_keyset` picks it. + pub fn active(&self) -> Option<(&str, u64)> { + self.0 + .iter() + .filter(|(_, (active, _))| *active) + .min_by_key(|(_, (_, ppk))| *ppk) + .map(|(id, (_, ppk))| (id.as_str(), *ppk)) + } +} + +pub async fn keysets(mint: &str, timeout: Duration) -> Result { + Asset::new(mint)?.fence()?; + let v: serde_json::Value = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(timeout) + .build()? + .get(format!("{}/v1/keysets", mint.trim_end_matches('/'))) + .send() + .await? + .error_for_status()? + .json() + .await?; + let mut out = BTreeMap::new(); + for k in v["keysets"].as_array().context("no keysets")? { + if k["unit"] != "sat" { + continue; + } + let (Some(id), Some(active)) = (k["id"].as_str(), k["active"].as_bool()) else { + continue; + }; + out.insert( + id.to_string(), + (active, k["input_fee_ppk"].as_u64().unwrap_or(0)), + ); + } + Ok(Keysets(out)) +} + +/// Fee-inclusive amounts for one leg, sender-funds-net: the sender locks `gross` so the +/// receiver's claim (input fee `claim_fee`) nets exactly `net`. +pub fn leg(net: u64, ppk: u64) -> serde_json::Value { + match gross(net, ppk) { + Ok((g, claim)) => serde_json::json!({"net":net,"ppk":ppk,"gross":g,"claim_fee":claim}), + Err(e) => serde_json::json!({"net":net,"ppk":ppk,"error":unsolvable(net, ppk, &e)}), + } +} +fn unsolvable(net: u64, ppk: u64, e: &anyhow::Error) -> String { + format!( + "{e}: no gross amount covers exactly {net} net sats plus its own claim fee at {ppk} ppk; choose a different amount" + ) +} + +/// What `mint::plan` would select, computed from a read-only wallet snapshot. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Preview { + pub keyset: String, + pub ppk: u64, + pub net: u64, + pub gross: u64, + pub claim_fee: u64, + pub lock_fee: Option, + pub debit: Option, + pub inputs: usize, + pub error: Option, +} +impl Preview { + pub fn json(&self, mint: &str) -> serde_json::Value { + serde_json::json!({"mint":mint,"keyset":self.keyset,"ppk":self.ppk,"net":self.net, + "gross":self.gross,"lock_fee":self.lock_fee,"claim_fee":self.claim_fee, + "debit":self.debit,"inputs":self.inputs,"error":self.error}) + } +} + +/// Unreserved plain proofs `(amount, keyset)` in the order `mint::plan` sorts them. +pub fn spendable(home: &Path, mint: &str) -> Result> { + let asset = hex::encode(Sha256::digest(format!("{mint}|sat"))); + let Some(db) = observe::open(&home.join(format!("{asset}.sqlite")))? else { + return Ok(vec![]); + }; + let mut stmt = db.prepare( + "SELECT amount, keyset_id FROM proof WHERE mint_url=?1 AND unit='sat' AND state='UNSPENT' \ + AND used_by_operation IS NULL AND spending_condition IS NULL ORDER BY rowid", + )?; + let mut rows = stmt + .query_map([mint], |r| { + Ok((r.get::<_, i64>(0)?, r.get::<_, String>(1)?)) + })? + .map(|r| { + let (a, k) = r?; + Ok((u64::try_from(a)?, k)) + }) + .collect::>>()?; + rows.sort_by_key(|(a, _)| *a); + Ok(rows) +} + +/// `mint::plan` without the wallet: same gross, same greedy input walk, same fee cap. +pub fn plan(proofs: &[(u64, String)], ks: &Keysets, net: u64, max_fee: u64) -> Result { + let (keyset, ppk) = ks.active().context("mint has no active sat keyset")?; + let (g, claim_fee) = gross(net, ppk).map_err(|e| anyhow::anyhow!(unsolvable(net, ppk, &e)))?; + let mut p = Preview { + keyset: keyset.into(), + ppk, + net, + gross: g, + claim_fee, + lock_fee: None, + debit: None, + inputs: 0, + error: None, + }; + if let Err(e) = crate::real_money::check_lock_gross(g) { + p.error = Some(e.to_string()); + return Ok(p); + } + let mut total = 0u64; + let mut ppks = vec![]; + for (amount, id) in proofs { + total = total.checked_add(*amount).context("overflow")?; + let (_, kppk) = + ks.0.get(id) + .with_context(|| format!("unknown keyset {id} for a wallet proof"))?; + ppks.push(*kppk); + let lock_fee = fee(ppks.clone())?; + if total >= g + lock_fee { + p.inputs = ppks.len(); + p.lock_fee = Some(lock_fee); + p.debit = Some(g + lock_fee); + if ppks.len() > 128 { + p.error = Some("too many inputs".into()); + } else if claim_fee + lock_fee > max_fee { + p.error = Some(format!( + "fee cap exceeded: lock fee {lock_fee} + claim fee {claim_fee} > --max-fees {max_fee}" + )); + } + return Ok(p); + } + } + p.inputs = ppks.len(); + p.error = Some("insufficient unreserved balance (overlisting refused)".into()); + Ok(p) +} + +/// Non-blocking per-lot fee estimate for `discover`: `"unknown"` when a mint is unreachable. +pub async fn discover_fees(lots: &[Lot]) -> Vec { + let mut mints: BTreeMap> = BTreeMap::new(); + for l in lots { + mints.insert(l.give.asset.mint_url.clone(), None); + mints.insert(l.want.asset.mint_url.clone(), None); + } + let mut tasks = tokio::task::JoinSet::new(); + for mint in mints.keys().cloned() { + tasks.spawn(async move { + let ks = tokio::time::timeout( + Duration::from_secs(6), + keysets(&mint, Duration::from_secs(5)), + ) + .await; + (mint, ks.ok().and_then(Result::ok)) + }); + } + while let Some(Ok((mint, ks))) = tasks.join_next().await { + mints.insert(mint, ks); + } + let side = |mint: &str, net: u64| match mints + .get(mint) + .and_then(Option::as_ref) + .and_then(Keysets::active) + { + Some((_, ppk)) => leg(net, ppk), + None => serde_json::json!("unknown"), + }; + lots.iter() + .map(|l| { + let mut want = side(&l.want.asset.mint_url, l.want.net); + if let Some(o) = want.as_object_mut() { + if o.contains_key("gross") { + o.insert( + "taker_lock_fee".into(), + "depends on the taker's inputs; use take --dry-run".into(), + ); + } + } + serde_json::json!({ + // Maker sends `give` (funds the taker's claim fee); taker sends `want`. + "give": side(&l.give.asset.mint_url, l.give.net), + "want": want, + }) + }) + .collect() +} + +/// Lot terms from a validated lot event's content. +pub fn lot_of(content: &str) -> Result { + let l: Lot = serde_json::from_str(content)?; + ensure!(l.trade_v == 1, "unknown protocol version"); + Ok(l) +} + +#[cfg(test)] +mod tests { + use super::*; + fn ks(entries: &[(&str, bool, u64)]) -> Keysets { + Keysets( + entries + .iter() + .map(|(i, a, p)| (i.to_string(), (*a, *p))) + .collect(), + ) + } + #[test] + fn active_is_lowest_fee_active() { + let k = ks(&[("a", true, 500), ("b", true, 100), ("c", false, 0)]); + assert_eq!(k.active(), Some(("b", 100))); + } + #[test] + fn plan_matches_coordinator_arithmetic() { + let k = ks(&[("old", false, 1000), ("new", true, 100)]); + let proofs = vec![ + (8, "old".to_string()), + (16, "new".into()), + (32, "new".into()), + ]; + let p = plan(&proofs, &k, 24, 16).unwrap(); + let (g, c) = gross(24, 100).unwrap(); + assert_eq!((p.gross, p.claim_fee), (g, c)); + // 8+16 = 24 < 25 + fee; 8+16+32 covers it with fee(1000,100,100) = 2. + assert_eq!((p.inputs, p.lock_fee, p.debit), (3, Some(2), Some(g + 2))); + assert!(p.error.is_none()); + let capped = plan(&proofs, &k, 24, 2).unwrap(); + assert!(capped.error.unwrap().contains("fee cap exceeded")); + let poor = plan(&proofs[..1], &k, 24, 16).unwrap(); + assert!(poor.error.unwrap().contains("insufficient")); + } + #[test] + fn unsolvable_amount_is_reported() { + let k = ks(&[("x", true, 1000)]); + let e = plan(&[], &k, 2, 16).unwrap_err().to_string(); + assert!( + e.contains("cannot solve bounded fee-inclusive split"), + "{e}" + ); + assert!( + leg(2, 1000)["error"] + .as_str() + .unwrap() + .contains("choose a different amount") + ); + } +} diff --git a/crates/maxplayer-trade/src/lib.rs b/crates/maxplayer-trade/src/lib.rs index 9b095db69..785d55344 100644 --- a/crates/maxplayer-trade/src/lib.rs +++ b/crates/maxplayer-trade/src/lib.rs @@ -477,3 +477,8 @@ pub mod wallet; pub mod money; pub mod real_money; pub mod receive; + +pub mod dryrun; +pub mod observe; +pub mod report; +pub mod serve; diff --git a/crates/maxplayer-trade/src/main.rs b/crates/maxplayer-trade/src/main.rs index 285216114..50ba896de 100644 --- a/crates/maxplayer-trade/src/main.rs +++ b/crates/maxplayer-trade/src/main.rs @@ -32,8 +32,11 @@ enum Command { /// Maximum combined lock and claim mint fees, in sats. #[arg(long, default_value_t = 16)] max_fees: u64, + /// Print both legs' expected fees, gross amounts and total debit; change nothing. + #[arg(long)] + dry_run: bool, }, - /// Discover verified available lots across all configured relays. + /// Discover verified available lots (lock-free, read-only; works while serve runs). Discover, /// Cancel an inactive lot and release its reserved inputs. Cancel { lot: String }, @@ -51,6 +54,9 @@ enum Command { /// Maximum combined lock and claim mint fees, in sats. #[arg(long, default_value_t = 16)] max_fees: u64, + /// Print both legs' expected fees, gross amounts and total debit; change nothing. + #[arg(long)] + dry_run: bool, }, /// One recovery pass; exit 3 if items remain pending. Does not wait for lock deadlines. Recover, @@ -80,8 +86,12 @@ enum Command { #[arg(long)] token_file: Option, }, - /// Read public journal states without locking the home or advancing payments. - Status, + /// Summarize lots, swaps and money records without locking the home or advancing payments. + Status { + /// Print the raw public records (machine-readable) instead of the summary. + #[arg(long)] + json: bool, + }, /// Read spendable wallet balance without submitting existing money authorizations. Balance { mint: String }, } @@ -93,6 +103,9 @@ async fn main() -> std::process::ExitCode { match execute().await { Ok(()) => std::process::ExitCode::SUCCESS, Err(error) => { + if let Some(f) = error.downcast_ref::() { + return std::process::ExitCode::from(f.0); + } eprintln!("{error}"); std::process::ExitCode::from(if error.is::() { 3 @@ -115,10 +128,92 @@ async fn execute() -> Result<()> { ); return Ok(()); } - if matches!(cli.command, Command::Status) { - println!("{}", maxplayer_trade::wallet::read_status(&cli.home)?); + if let Command::Status { json } = cli.command { + if json { + println!("{}", maxplayer_trade::wallet::read_status(&cli.home)?); + } else { + println!("{}", maxplayer_trade::report::render(&cli.home)?); + } return Ok(()); } + let relays: Vec = if cli.relay.is_empty() { + maxplayer_trade::market::DEFAULT_RELAYS + .iter() + .map(|s| s.to_string()) + .collect() + } else { + cli.relay.clone() + }; + // Lock-free, journal-free commands: they only read relays, mints and SQLite snapshots. + match &cli.command { + Command::Discover => return lockfree::discover(&relays).await, + Command::List { + give_mint, + give, + want_mint, + want, + max_fees, + dry_run: true, + } => { + return lockfree::list(&cli.home, give_mint, *give, want_mint, *want, *max_fees).await; + } + Command::Take { + lot, + max_give, + min_receive, + max_fees, + dry_run: true, + } => { + return lockfree::take(&cli.home, &relays, lot, *max_give, *min_receive, *max_fees) + .await; + } + _ => {} + } + // A running serve owns the home: hand list/take/cancel to it, with these exact bounds. + let handoff = match &cli.command { + Command::List { + give_mint, + give, + want_mint, + want, + max_fees, + .. + } => Some(maxplayer_trade::serve::Command::List { + give_mint: give_mint.clone(), + give: *give, + want_mint: want_mint.clone(), + want: *want, + max_fees: *max_fees, + }), + Command::Take { + lot, + max_give, + min_receive, + max_fees, + .. + } => Some(maxplayer_trade::serve::Command::Take { + lot: lot.clone(), + max_give: *max_give, + min_receive: *min_receive, + max_fees: *max_fees, + }), + Command::Cancel { lot } => { + Some(maxplayer_trade::serve::Command::Cancel { lot: lot.clone() }) + } + _ => None, + } + .map(|command| maxplayer_trade::serve::Request { + v: maxplayer_trade::serve::PROTOCOL_V, + relays: relays.clone(), + command, + }); + if let Some(request) = &handoff { + if cli.home.is_dir() { + if let Some(code) = maxplayer_trade::serve::forward(&cli.home, request).await? { + return forwarded(code); + } + } + } fs::create_dir_all(&cli.home)?; fs::set_permissions(&cli.home, fs::Permissions::from_mode(0o700))?; let lock = OpenOptions::new() @@ -128,8 +223,35 @@ async fn execute() -> Result<()> { .write(true) .mode(0o600) .open(cli.home.join("owner.lock"))?; - lock.try_lock_exclusive() - .context("home is already in use")?; + if let Err(busy) = lock.try_lock_exclusive() { + // Another process owns the home: a serve (possibly still binding its socket) gets the + // request; a short-lived writer such as recover is waited out. Never two writers. + let Some(request) = &handoff else { + return Err(busy).context("home is already in use"); + }; + let end = tokio::time::Instant::now() + maxplayer_trade::serve::CONNECT_WAIT; + loop { + if let Some(code) = maxplayer_trade::serve::forward(&cli.home, request).await? { + return forwarded(code); + } + if lock.try_lock_exclusive().is_ok() { + break; + } + if tokio::time::Instant::now() >= end { + return Err(busy).context("home is already in use"); + } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + } + } + #[cfg(feature = "lab")] + let _trace = lab_trace::Writer::start(); + // Bind while holding owner.lock (an existing socket file is stale) and before the startup + // recovery pass, so hand-offs queue instead of failing while serve recovers. + let mut jobs = if matches!(cli.command, Command::Serve) { + Some(maxplayer_trade::serve::bind(&cli.home, relays.clone())?) + } else { + None + }; let money_journal = Journal::open(&cli.home).await?; // Inspection must not submit an earlier payment authorization. Newly credited // money proofs remain reserved until explicit recovery settles their journal. @@ -155,14 +277,6 @@ async fn execute() -> Result<()> { | Command::Receive { .. } ) { let keys = maxplayer_trade::wallet::identity(&cli.home)?; - let relays = if cli.relay.is_empty() { - maxplayer_trade::market::DEFAULT_RELAYS - .iter() - .map(|s| s.to_string()) - .collect() - } else { - cli.relay.clone() - }; let j = Journal::open(&cli.home).await?; let mut m = Market::connect(keys, &relays).await?; if !matches!(cli.command, Command::Recover) { @@ -175,6 +289,7 @@ async fn execute() -> Result<()> { want_mint, want, max_fees, + .. } => { coordinator::list( &cli.home, @@ -193,20 +308,13 @@ async fn execute() -> Result<()> { .await?; Ok(()) } - Command::Discover => { - let lots = m.discover(None).await?; - let listings = lots.iter().map(|e| { - Ok(serde_json::json!({"lot_id":e.id,"maker":e.pubkey,"terms":serde_json::from_str::(&e.content)?})) - }).collect::>>()?; - println!("{}", serde_json::json!({"status":"ok","listings":listings})); - Ok(()) - } Command::Cancel { lot } => coordinator::cancel(&cli.home, &j, &m, &lot).await, Command::Take { lot, max_give, min_receive, max_fees, + .. } => { let id = coordinator::start_take( &cli.home, @@ -220,7 +328,7 @@ async fn execute() -> Result<()> { .await?; coordinator::run(&cli.home, &j, &mut m, Some(&id)).await } - Command::Serve => coordinator::run(&cli.home, &j, &mut m, None).await, + Command::Serve => coordinator::run_with(&cli.home, &j, &mut m, None, jobs.take()).await, Command::Recover => coordinator::recover_until_settled(&cli.home, &j, &mut m).await, _ => unreachable!(), }; @@ -341,3 +449,206 @@ async fn execute() -> Result<()> { unreachable!() } + +fn forwarded(code: u8) -> Result<()> { + if code == 0 { + Ok(()) + } else { + Err(maxplayer_trade::serve::Forwarded(code).into()) + } +} + +/// Commands that never take owner.lock, never open the journal and never create files. +mod lockfree { + use anyhow::{Context, Result, bail, ensure}; + use maxplayer_trade::{Asset, dryrun, market::Market, parse_lot, wallet}; + use std::path::Path; + /// Ephemeral relay identity: discovery needs no trade key and must not create one. + async fn market(relays: &[String]) -> Result { + Market::connect(nostr_sdk::Keys::generate(), relays).await + } + pub async fn discover(relays: &[String]) -> Result<()> { + let m = market(relays).await?; + let lots = m.discover(None).await?; + let now = maxplayer_trade::coordinator::now(); + let terms = lots + .iter() + .map(|e| parse_lot(e, now)) + .collect::>>()?; + let fees = dryrun::discover_fees(&terms).await; + let listings = lots.iter().zip(fees).map(|(e, fees)| { + Ok(serde_json::json!({"lot_id":e.id,"maker":e.pubkey,"terms":serde_json::from_str::(&e.content)?,"expected_fees":fees})) + }).collect::>>()?; + println!("{}", serde_json::json!({"status":"ok","listings":listings})); + Ok(()) + } + async fn preflight(mints: [&str; 2]) -> Result<()> { + for mint in mints { + Asset::new(mint)?.fence()?; + wallet::preflight(mint).await?; + } + Ok(()) + } + fn finish(out: serde_json::Value, refusal: Option) -> Result<()> { + println!("{out}"); + match refusal { + Some(r) => bail!("dry run: the real command would refuse: {r}"), + None => Ok(()), + } + } + pub async fn list( + home: &Path, + give_mint: &str, + give: u64, + want_mint: &str, + want: u64, + max_fees: u64, + ) -> Result<()> { + let (give_asset, want_asset) = (Asset::new(give_mint)?, Asset::new(want_mint)?); + ensure!(give_asset != want_asset, "same-asset swap"); + preflight([&give_asset.mint_url, &want_asset.mint_url]).await?; + let gks = dryrun::keysets(&give_asset.mint_url, std::time::Duration::from_secs(15)).await?; + let wks = dryrun::keysets(&want_asset.mint_url, std::time::Duration::from_secs(15)).await?; + let p = dryrun::plan( + &dryrun::spendable(home, &give_asset.mint_url)?, + &gks, + give, + max_fees, + )?; + let incoming = dryrun::leg( + want, + wks.active() + .context("want mint has no active sat keyset")? + .1, + ); + let mut refusal = p.error.clone(); + if let Some(c) = incoming["claim_fee"].as_u64() { + if c > max_fees && refusal.is_none() { + refusal = Some(format!( + "every taker would be refused: incoming claim fee {c} exceeds --max-fees {max_fees}" + )); + } + } else if refusal.is_none() { + refusal = incoming["error"].as_str().map(str::to_owned); + } + finish( + serde_json::json!({"dry_run":true,"command":"list", + "you_give":p.json(&give_asset.mint_url), + "you_receive":{"mint":want_asset.mint_url,"leg":incoming,"note":"the taker funds your claim fee; you net exactly the wanted amount"}, + "max_total_debit":p.debit,"max_fees":max_fees, + "fees_total":p.lock_fee.map(|l| l + p.claim_fee), + "side_effects":"none: nothing journaled, reserved, locked or published"}), + refusal, + ) + } + pub async fn take( + home: &Path, + relays: &[String], + lot: &str, + max_give: u64, + min_receive: u64, + max_fees: u64, + ) -> Result<()> { + let m = market(relays).await?; + let e = m + .discover(Some(lot.parse()?)) + .await? + .into_iter() + .next() + .context("lot unavailable, invalid, expired or cancelled")?; + let l = parse_lot(&e, maxplayer_trade::coordinator::now())?; + // Self-trade check without creating a trade key. + let own = std::fs::read_to_string(home.join("trade.key")) + .ok() + .and_then(|k| nostr_sdk::Keys::parse(&k).ok()); + ensure!(own.is_none_or(|k| k.public_key() != e.pubkey), "self trade"); + preflight([&l.give.asset.mint_url, &l.want.asset.mint_url]).await?; + let gks = + dryrun::keysets(&l.give.asset.mint_url, std::time::Duration::from_secs(15)).await?; + let wks = + dryrun::keysets(&l.want.asset.mint_url, std::time::Duration::from_secs(15)).await?; + let p = dryrun::plan( + &dryrun::spendable(home, &l.want.asset.mint_url)?, + &wks, + l.want.net, + max_fees, + )?; + let incoming = dryrun::leg( + l.give.net, + gks.active() + .context("give mint has no active sat keyset")? + .1, + ); + let mut refusal = None; + if l.give.net < min_receive { + refusal = Some(format!( + "minimum receive cap: lot gives {} net < --min-receive {min_receive}", + l.give.net + )); + } + refusal = refusal.or(p.error.clone()); + if refusal.is_none() { + if let Some(d) = p.debit.filter(|d| *d > max_give) { + refusal = Some(format!( + "maximum give cap: debit {d} > --max-give {max_give}" + )); + } + } + if refusal.is_none() { + match incoming["claim_fee"].as_u64() { + Some(c) if c > max_fees => { + refusal = Some(format!( + "incoming claim fee {c} exceeds taker admission cap --max-fees {max_fees}" + )) + } + Some(_) => {} + None => refusal = incoming["error"].as_str().map(str::to_owned), + } + } + finish( + serde_json::json!({"dry_run":true,"command":"take","lot_id":e.id, + "you_give":p.json(&l.want.asset.mint_url), + "you_receive":{"mint":l.give.asset.mint_url,"leg":incoming,"note":"the maker funds your claim fee; you net exactly this amount"}, + "max_total_debit":p.debit,"max_give":max_give,"min_receive":min_receive,"max_fees":max_fees, + "fees_total":p.lock_fee.map(|lf| lf + p.claim_fee), + "side_effects":"none: nothing journaled, reserved, locked, published or sent to the maker"}), + refusal, + ) + } +} + +/// Lab-only writer trace for the single-writer race test: one line per owner.lock tenure. +#[cfg(feature = "lab")] +mod lab_trace { + use std::io::Write; + pub struct Writer(Option); + fn line(path: &std::path::Path, what: &str) { + if let Ok(mut f) = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path) + { + let t = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + let _ = writeln!(f, "{what} {} {t}", std::process::id()); + } + } + impl Writer { + pub fn start() -> Self { + let path = std::env::var_os("TRADE_LAB_WRITER_TRACE").map(std::path::PathBuf::from); + if let Some(p) = &path { + line(p, "acquire"); + } + Self(path) + } + } + impl Drop for Writer { + fn drop(&mut self) { + if let Some(p) = &self.0 { + line(p, "release"); + } + } + } +} diff --git a/crates/maxplayer-trade/src/observe.rs b/crates/maxplayer-trade/src/observe.rs new file mode 100644 index 000000000..dcf6b4c00 --- /dev/null +++ b/crates/maxplayer-trade/src/observe.rs @@ -0,0 +1,275 @@ +//! Read-only views of a trade home: no owner lock, no journal writes, no network. +//! +//! Everything here is DERIVED display state. Nothing in this module feeds back into the +//! swap state machine, its deadlines, lock durations or refund decisions. +use crate::coordinator::Swap; +use anyhow::Result; +use std::{collections::HashMap, path::Path}; + +/// A swap waiting on its counterparty is reported `counterparty_unresponsive` after this long. +pub const UNRESPONSIVE_SECONDS: u64 = 180; +/// Repeat an unresponsive warning in live output at most this often. +pub const REPEAT_SECONDS: u64 = 300; + +/// Threshold in seconds. Lab builds may shorten it for loopback tests only. +pub fn unresponsive_after() -> u64 { + #[cfg(feature = "lab")] + if let Some(seconds) = std::env::var("TRADE_LAB_UNRESPONSIVE_SECONDS") + .ok() + .and_then(|v| v.parse().ok()) + { + return seconds; + } + UNRESPONSIVE_SECONDS +} + +/// Open an SQLite file read-only, or `None` when it does not exist (nothing is created). +pub fn open(path: &Path) -> Result> { + if !path.exists() { + return Ok(None); + } + let db = + rusqlite::Connection::open_with_flags(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?; + db.busy_timeout(std::time::Duration::from_secs(2))?; + Ok(Some(db)) +} + +/// `(key, value, updated_time)` for one journal namespace, read-only. +pub fn journal_rows(home: &Path, ns: &str) -> Result, u64)>> { + let Some(db) = open(&home.join("trade.sqlite"))? else { + return Ok(vec![]); + }; + let mut stmt = db.prepare( + "SELECT key, value, updated_time FROM kv_store WHERE primary_namespace='trade-v1' \ + AND secondary_namespace=?1 ORDER BY created_time, key", + )?; + let rows = stmt.query_map([ns], |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, Vec>(1)?, + r.get::<_, i64>(2)?, + )) + })?; + rows.map(|row| { + let (k, v, t) = row?; + Ok((k, v, u64::try_from(t).unwrap_or(0))) + }) + .collect() +} + +/// True when the journal holds a swap that is not terminal (read-only). +pub fn open_swap(home: &Path) -> Result { + Ok(journal_rows(home, "swap")?.iter().any(|(_, v, _)| { + serde_json::from_slice::(v).map_or(true, |s| !crate::coordinator::terminal(&s)) + })) +} + +/// What a swap in `state` is waiting for from its counterparty, if anything. +pub fn waiting_for(role: &str, state: &str) -> Option<&'static str> { + match (role, state) { + ("taker", "requested") => Some("maker_quote"), + ("taker", "first_locked") => Some("maker_second_lock"), + ("maker", "quoted") => Some("taker_first_lock"), + ("maker", "second_locked") => Some("taker_claim"), + _ => None, + } +} + +/// Earliest unix time at which our own outgoing lock becomes refundable, if we hold one. +/// Mirrors the refund gates in the coordinator (`time > long + margin` for the taker, +/// `time > short + margin` for the maker); it does not decide anything. +pub fn refund_available_at(s: &Swap) -> Option { + let q = s.quote.as_ref()?; + match (s.role.as_str(), s.state.as_str()) { + ("taker", "first_locked") => Some(q.long.saturating_add(q.margin).saturating_add(1)), + ("maker", "second_locked") => Some(q.short.saturating_add(q.margin).saturating_add(1)), + _ => None, + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Unresponsive { + pub swap_id: String, + pub role: String, + pub state: String, + pub waiting_for: &'static str, + pub elapsed_seconds: u64, + pub refund_available_at: Option, +} +impl Unresponsive { + pub fn json(&self, now: u64) -> serde_json::Value { + serde_json::json!({ + "warning": "counterparty_unresponsive", + "swap_id": self.swap_id, + "role": self.role, + "state": self.state, + "waiting_for": self.waiting_for, + "elapsed_seconds": self.elapsed_seconds, + "refund_available_at": self.refund_available_at, + "refund_available_in_seconds": self.refund_available_at.map(|t| t.saturating_sub(now)), + }) + } +} + +/// Pure derivation. `entered` is when the swap record last changed, i.e. when it entered its +/// current waiting state; nothing from the counterparty has advanced it since. +pub fn derive( + swap_id: &str, + role: &str, + state: &str, + entered: u64, + refund_available_at: Option, + now: u64, + after: u64, +) -> Option { + let waiting_for = waiting_for(role, state)?; + let elapsed_seconds = now.saturating_sub(entered); + (elapsed_seconds > after).then(|| Unresponsive { + swap_id: swap_id.into(), + role: role.into(), + state: state.into(), + waiting_for, + elapsed_seconds, + refund_available_at, + }) +} + +/// Every swap currently waiting on an unresponsive counterparty (read-only). +pub fn unresponsive(home: &Path, now: u64) -> Result> { + let after = unresponsive_after(); + let mut out = vec![]; + for (_, v, updated) in journal_rows(home, "swap")? { + let Ok(s) = serde_json::from_slice::(&v) else { + continue; + }; + if let Some(u) = derive( + &s.id, + &s.role, + &s.state, + updated, + refund_available_at(&s), + now, + after, + ) { + out.push(u); + } + } + Ok(out) +} + +/// Rate limiter for live warnings: first crossing, then every [`REPEAT_SECONDS`]. +#[derive(Default)] +pub struct Warned(HashMap); +impl Warned { + pub fn due(&mut self, current: &[Unresponsive]) -> Vec { + self.0 + .retain(|id, _| current.iter().any(|u| &u.swap_id == id)); + let mut out = vec![]; + for u in current { + let due = match self.0.get(&u.swap_id) { + None => true, + Some(last) => u.elapsed_seconds >= last.saturating_add(REPEAT_SECONDS), + }; + if due { + self.0.insert(u.swap_id.clone(), u.elapsed_seconds); + out.push(u.clone()); + } + } + out + } +} + +/// `YYYY-MM-DD HH:MM:SS UTC` without a date dependency. +pub fn utc(ts: u64) -> String { + let days = (ts / 86400) as i64; + let secs = ts % 86400; + let z = days + 719_468; + let era = z.div_euclid(146_097); + let doe = z.rem_euclid(146_097); + let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = doy - (153 * mp + 2) / 5 + 1; + let m = if mp < 10 { mp + 3 } else { mp - 9 }; + let y = yoe + era * 400 + i64::from(m <= 2); + format!( + "{y:04}-{m:02}-{d:02} {:02}:{:02}:{:02} UTC", + secs / 3600, + secs / 60 % 60, + secs % 60 + ) +} + +/// `1h02m03s` / `4m05s` / `6s`. +pub fn span(seconds: u64) -> String { + let (h, m, s) = (seconds / 3600, seconds / 60 % 60, seconds % 60); + if h > 0 { + format!("{h}h{m:02}m{s:02}s") + } else if m > 0 { + format!("{m}m{s:02}s") + } else { + format!("{s}s") + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn warning_only_after_threshold_and_only_while_waiting() { + let after = UNRESPONSIVE_SECONDS; + assert_eq!(after, 180); + assert!(derive("s", "taker", "first_locked", 1000, Some(5000), 1180, after).is_none()); + let w = derive("s", "taker", "first_locked", 1000, Some(5000), 1181, after).unwrap(); + assert_eq!( + (w.elapsed_seconds, w.waiting_for, w.refund_available_at), + (181, "maker_second_lock", Some(5000)) + ); + assert_eq!( + derive("s", "maker", "second_locked", 0, None, 999, after) + .unwrap() + .waiting_for, + "taker_claim" + ); + // Our own pending actions and terminal states never warn. + for (role, state) in [ + ("taker", "accepted"), + ("taker", "second_validated"), + ("taker", "claimed"), + ("taker", "complete"), + ("maker", "first_validated"), + ("maker", "claiming"), + ("maker", "settling"), + ("maker", "refunded"), + ] { + assert!(derive("s", role, state, 0, None, 100_000, after).is_none()); + } + assert_eq!(w.json(4000)["refund_available_in_seconds"], 1000); + } + #[test] + fn repeat_rate_limited() { + let mut w = Warned::default(); + let u = |e| Unresponsive { + swap_id: "a".into(), + role: "taker".into(), + state: "first_locked".into(), + waiting_for: "maker_second_lock", + elapsed_seconds: e, + refund_available_at: None, + }; + assert_eq!(w.due(&[u(181)]).len(), 1); + assert!(w.due(&[u(400)]).is_empty()); + assert_eq!(w.due(&[u(481)]).len(), 1); + assert!(w.due(&[]).is_empty()); + assert_eq!(w.due(&[u(500)]).len(), 1, "a fresh wait warns again"); + } + #[test] + fn utc_and_span_format() { + assert_eq!(utc(0), "1970-01-01 00:00:00 UTC"); + assert_eq!(utc(1_791_504_000), "2026-10-09 00:00:00 UTC"); + assert_eq!(utc(951_825_600), "2000-02-29 12:00:00 UTC"); + assert_eq!(span(3723), "1h02m03s"); + assert_eq!(span(245), "4m05s"); + assert_eq!(span(6), "6s"); + } +} diff --git a/crates/maxplayer-trade/src/report.rs b/crates/maxplayer-trade/src/report.rs new file mode 100644 index 000000000..3b6807398 --- /dev/null +++ b/crates/maxplayer-trade/src/report.rs @@ -0,0 +1,222 @@ +//! Human-readable `status`: one line per lot, swap, funding, withdrawal and receive. +//! +//! Read-only (see [`observe`]); prints only public identifiers, mints, states and amounts — +//! never keys, preimages, proofs, tokens, quotes' secrets or invoices. +use crate::{ + coordinator::{Listing, Swap, now, terminal}, + lifecycle, + money::{Funding, Withdrawal}, + observe, + receive::{Receipt, ReceiveState}, +}; +use anyhow::Result; +use std::path::Path; + +fn lot_terms(e: &crate::Event) -> Option { + serde_json::from_str(&e.content).ok() +} +fn lot_status(home: &Path, lot: &str) -> Option { + observe::journal_rows(home, "listing") + .ok()? + .into_iter() + .find(|(k, _, _)| k == lot) + .and_then(|(_, v, _)| serde_json::from_slice::(&v).ok()) + .and_then(|l| lifecycle(&l.event, &l.statuses).ok()) + .map(|s| format!("{s:?}").to_lowercase()) +} + +fn swap_line(home: &Path, s: &Swap) -> String { + let lot = s.lot.id.to_hex(); + let mut line = match (s.role.as_str(), lot_status(home, &lot)) { + ("maker", Some(status)) => format!("lot {lot} {status}; swap {} {} (maker)", s.id, s.state), + (role, _) => format!("swap {} {} ({role}) on lot {lot}", s.id, s.state), + }; + let Some(l) = lot_terms(&s.lot) else { + return line; + }; + // Incoming net is fixed by the lot (the sender funds our claim fee); outgoing is our debit. + let (incoming, outgoing) = if s.role == "taker" { + (&l.give, &l.want) + } else { + (&l.want, &l.give) + }; + let fees = s.plan.debit.saturating_sub(s.plan.net); + match s.state.as_str() { + "complete" | "complete_unclaimed" => line.push_str(&format!( + "; +{} {} / -{} {}; fees {fees}", + incoming.net, incoming.asset.mint_url, s.plan.debit, outgoing.asset.mint_url + )), + "refunded" => line.push_str(&format!( + "; own lock of {} {} refunded (refund mint fees not itemized)", + s.plan.debit, outgoing.asset.mint_url + )), + "expired" => line.push_str("; nothing locked; reservation released"), + _ => line.push_str(&format!( + "; expects +{} {} / -{} {}; fees {fees}", + incoming.net, incoming.asset.mint_url, s.plan.debit, outgoing.asset.mint_url + )), + } + if s.state.ends_with("_quarantined") { + line.push_str("; MANUAL RECOVERY: quarantined outputs retained"); + } + line +} + +pub fn render(home: &Path) -> Result { + let now = now(); + let mut unresolved = vec![]; + let mut settled = vec![]; + let mut manual = 0usize; + let mut swap_lots = std::collections::BTreeSet::new(); + for (_, v, _) in observe::journal_rows(home, "swap")? { + let Ok(s) = serde_json::from_slice::(&v) else { + unresolved.push("swap record unreadable by this version".to_string()); + continue; + }; + if s.role == "maker" { + swap_lots.insert(s.lot.id.to_hex()); + } + manual += usize::from(s.state.ends_with("_quarantined")); + let line = swap_line(home, &s); + if terminal(&s) && !s.state.ends_with("_quarantined") { + settled.push(line) + } else { + unresolved.push(line) + } + } + for (id, v, _) in observe::journal_rows(home, "listing")? { + if swap_lots.contains(&id) { + continue; + } + let Ok(l) = serde_json::from_slice::(&v) else { + continue; + }; + let status = lifecycle(&l.event, &l.statuses) + .map(|s| format!("{s:?}").to_lowercase()) + .unwrap_or_else(|_| "invalid".into()); + let mut line = format!("lot {id} {status}"); + if l.cancelled && status == "available" { + line.push_str(" (cancel pending publication)"); + } + if let Some(t) = lot_terms(&l.event) { + line.push_str(&format!( + "; gives {} {} for {} {}", + t.give.net, t.give.asset.mint_url, t.want.net, t.want.asset.mint_url + )); + } + if status == "available" && !l.cancelled { + line.push_str(&format!( + "; reserves debit {} {}", + l.plan.debit, l.plan.mint + )); + } + settled.push(line); + } + for (_, v, _) in observe::journal_rows(home, "funding")? { + let Ok(f) = serde_json::from_slice::(&v) else { + continue; + }; + let state = if f.done { + "done" + } else if f.expired_unpaid { + "expired_unpaid" + } else { + "pending (invoice not yet paid or not yet issued)" + }; + let line = format!("funding {} {} {} sats: {state}", f.id, f.mint, f.amount); + if f.done || f.expired_unpaid { + settled.push(line) + } else { + unresolved.push(line) + } + } + for (_, v, _) in observe::journal_rows(home, "withdrawal")? { + let Ok(a) = serde_json::from_slice::(&v) else { + continue; + }; + let s = a.summary(); + let n = |k: &str| s[k].as_u64().map_or("?".to_string(), |n| n.to_string()); + let line = format!( + "withdrawal {} {} {}: invoice {} sats; mint fee {}; lightning fee {}; change {}", + a.id, + a.mint, + s["state"].as_str().unwrap_or("?"), + n("amount"), + n("mint_fee"), + n("lightning_fee"), + n("change") + ); + if a.terminal() { + settled.push(line) + } else { + unresolved.push(line) + } + } + for (_, v, _) in observe::journal_rows(home, "receive")? { + let Ok(r) = serde_json::from_slice::(&v) else { + continue; + }; + let quarantined = r.state == ReceiveState::Quarantined; + manual += usize::from(quarantined); + let s = r.summary(); + let mut line = format!( + "receive {} {} {}: token {} sats; fee {}; credited +{}", + r.id, + r.mint, + s["state"].as_str().unwrap_or("?"), + r.amount, + r.fee, + s["credited"] + ); + if quarantined { + line.push_str("; MANUAL RECOVERY: quarantined outputs retained"); + } + if r.terminal() && !quarantined { + settled.push(line) + } else { + unresolved.push(line) + } + } + let warnings = observe::unresponsive(home, now)?; + let mut out = String::new(); + for w in &warnings { + out.push_str(&format!( + "WARNING counterparty_unresponsive: swap {} ({} {}) waiting for {} for {}", + w.swap_id, + w.role, + w.state, + w.waiting_for, + observe::span(w.elapsed_seconds) + )); + match w.refund_available_at { + Some(t) if t > now => out.push_str(&format!( + "; refund available at {} (in {}); keep serve running", + observe::utc(t), + observe::span(t - now) + )), + Some(t) => out.push_str(&format!( + "; refund available since {}; keep serve running or run recover", + observe::utc(t) + )), + None => out.push_str("; nothing of ours is locked yet"), + } + out.push('\n'); + } + if unresolved.is_empty() { + out.push_str("UNRESOLVED: none\n"); + } else { + out.push_str(&format!("UNRESOLVED ({}):\n", unresolved.len())); + for l in &unresolved { + out.push_str(&format!(" {l}\n")); + } + } + for l in &settled { + out.push_str(&format!("{l}\n")); + } + out.push_str(&format!( + "summary: {} unresolved, {} need manual recovery. Exit codes: recover/take exit 3 while unresolved items remain (keep serve running or rerun recover), 4 when manual recovery is needed, 0 when everything is settled. status itself always exits 0; use status --json for machine-readable records.", + unresolved.len(), + manual + )); + Ok(out) +} diff --git a/crates/maxplayer-trade/src/serve.rs b/crates/maxplayer-trade/src/serve.rs new file mode 100644 index 000000000..7d6f0d674 --- /dev/null +++ b/crates/maxplayer-trade/src/serve.rs @@ -0,0 +1,606 @@ +//! Local hand-off from a CLI invocation to a running `serve` over `/serve.sock`. +//! +//! `serve` holds `owner.lock` and stays the only writer: a handed-off `list`, `take` or `cancel` +//! is queued to serve's own loop and executed there, between inbox messages and recovery ticks, +//! by the same coordinator functions (journal before effect) the CLI would have called, with +//! exactly the bounds the client sent. The client only relays output. A client that disconnects +//! never cancels a queued or running request; a take keeps being driven by serve. +//! +//! The socket is mode 0600 inside the 0700 home, and every peer must have the uid that owns the +//! socket (SO_PEERCRED). One request per connection; the request is at most +//! [`MAX_REQUEST_BYTES`]; reads and writes are bounded by [`IO_TIMEOUT`]; at most +//! [`MAX_CONNECTIONS`] connections are served at once. +use crate::{Asset, Leg, coordinator, journal::Journal, market::Market, observe}; +use anyhow::{Context, Result, bail, ensure}; +use serde::{Deserialize, Serialize}; +use std::{ + os::unix::fs::{MetadataExt, PermissionsExt}, + path::{Path, PathBuf}, + time::Duration, +}; +use tokio::{ + io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}, + net::{UnixListener, UnixStream}, + sync::{mpsc, oneshot}, +}; + +pub const SOCKET: &str = "serve.sock"; +pub const MAX_REQUEST_BYTES: usize = 16 * 1024; +pub const MAX_REPLY_LINE_BYTES: usize = 64 * 1024; +pub const IO_TIMEOUT: Duration = Duration::from_secs(10); +pub const MAX_CONNECTIONS: usize = 8; +/// Server keepalive cadence while a request runs or a take is streamed. +pub const KEEPALIVE: Duration = Duration::from_secs(15); +/// The client gives up on a silent serve after this long (keepalives arrive every 15 s). +pub const CLIENT_IDLE: Duration = Duration::from_secs(90); +/// How long a client whose home is locked keeps looking for a serve socket (serve may be starting). +pub const CONNECT_WAIT: Duration = Duration::from_secs(5); +pub const PROTOCOL_V: u8 = 1; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", deny_unknown_fields)] +pub enum Command { + List { + give_mint: String, + give: u64, + want_mint: String, + want: u64, + max_fees: u64, + }, + Take { + lot: String, + max_give: u64, + min_receive: u64, + max_fees: u64, + }, + Cancel { + lot: String, + }, +} +impl Command { + fn name(&self) -> &'static str { + match self { + Command::List { .. } => "list", + Command::Take { .. } => "take", + Command::Cancel { .. } => "cancel", + } + } +} +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Request { + pub v: u8, + /// The client's resolved relay set; serve refuses a request for a different set. + pub relays: Vec, + pub command: Command, +} +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", deny_unknown_fields)] +pub enum Reply { + Stdout(String), + Stderr(String), + Keepalive, + Exit { code: u8, error: Option }, +} + +/// A request queued to serve's loop. +pub struct Job { + pub command: Command, + reply: oneshot::Sender>, +} +#[derive(Debug)] +pub struct Outcome { + pub stdout: Vec, + pub swap: Option, +} +#[derive(Debug)] +pub struct Failure { + pub code: u8, + pub message: String, +} +/// Same exit-code mapping as the CLI. +pub fn exit_code(error: &anyhow::Error) -> u8 { + if error.is::() { + 3 + } else if error.is::() { + 4 + } else { + 1 + } +} + +/// Path used to bind/connect. Unix socket paths are limited to ~107 bytes; a longer home is +/// reached through its open directory descriptor (`/proc/self/fd/N/serve.sock`, Linux). +struct SocketPath { + path: PathBuf, + _dir: Option, +} +fn socket_path(home: &Path) -> Result { + let direct = home.join(SOCKET); + if direct.as_os_str().len() < 100 { + return Ok(SocketPath { + path: direct, + _dir: None, + }); + } + use std::os::fd::AsRawFd; + let dir = std::fs::File::open(home).context("open home for socket")?; + Ok(SocketPath { + path: PathBuf::from(format!("/proc/self/fd/{}/{SOCKET}", dir.as_raw_fd())), + _dir: Some(dir), + }) +} + +/// Bind the socket. Call only while holding `owner.lock`: any existing socket file is then +/// stale (its serve no longer holds the lock) and is replaced. +pub fn bind(home: &Path, relays: Vec) -> Result> { + let file = home.join(SOCKET); + match std::fs::symlink_metadata(&file) { + Ok(meta) => { + ensure!( + meta.file_type().is_socket_like(), + "{} exists and is not a socket; refusing to replace it", + file.display() + ); + std::fs::remove_file(&file).context("remove stale serve socket")?; + eprintln!("{}", serde_json::json!({"serve_socket":"stale_removed"})); + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(e.into()), + } + let sp = socket_path(home)?; + let listener = UnixListener::bind(&sp.path).context("bind serve socket")?; + std::fs::set_permissions(&file, std::fs::Permissions::from_mode(0o600))?; + let owner = std::fs::symlink_metadata(&file)?.uid(); + let (tx, rx) = mpsc::channel(MAX_CONNECTIONS); + let permits = std::sync::Arc::new(tokio::sync::Semaphore::new(MAX_CONNECTIONS)); + let home = home.to_path_buf(); + tokio::spawn(async move { + let _dir = sp; + loop { + let Ok((stream, _)) = listener.accept().await else { + continue; + }; + let Ok(permit) = permits.clone().try_acquire_owned() else { + drop(stream); + continue; + }; + let tx = tx.clone(); + let relays = relays.clone(); + let home = home.clone(); + tokio::spawn(async move { + let _permit = permit; + if let Err(e) = connection(stream, owner, &home, &relays, tx).await { + eprintln!( + "{}", + serde_json::json!({"serve_client":"closed","reason":e.to_string()}) + ); + } + }); + } + }); + eprintln!("{}", serde_json::json!({"serve_socket":file})); + Ok(rx) +} +trait SocketLike { + fn is_socket_like(&self) -> bool; +} +impl SocketLike for std::fs::FileType { + fn is_socket_like(&self) -> bool { + use std::os::unix::fs::FileTypeExt; + self.is_socket() + } +} + +/// One bounded line from a PERSISTENT buffered reader (a per-call reader would drop any +/// read-ahead lines). +async fn read_line_bounded( + r: &mut R, + max: usize, +) -> Result> { + let mut buf = Vec::new(); + let n = (&mut *r) + .take(max as u64 + 1) + .read_until(b'\n', &mut buf) + .await?; + if n == 0 { + return Ok(None); + } + ensure!( + buf.last() == Some(&b'\n'), + "unterminated line or line exceeds {max} bytes" + ); + buf.pop(); + ensure!(buf.len() <= max, "line exceeds {max} bytes"); + Ok(Some(String::from_utf8(buf)?)) +} +async fn write(stream: &mut UnixStream, reply: &Reply) -> Result<()> { + let mut line = serde_json::to_string(reply)?; + line.push('\n'); + tokio::time::timeout(IO_TIMEOUT, stream.write_all(line.as_bytes())) + .await + .context("client write timed out")??; + Ok(()) +} + +async fn connection( + mut stream: UnixStream, + owner: u32, + home: &Path, + relays: &[String], + tx: mpsc::Sender, +) -> Result<()> { + let peer = stream.peer_cred()?; + ensure!(peer.uid() == owner, "peer uid {} refused", peer.uid()); + // One request per connection: nothing after its line is read. + let line = tokio::time::timeout( + IO_TIMEOUT, + read_line_bounded(&mut BufReader::new(&mut stream), MAX_REQUEST_BYTES), + ) + .await + .context("request read timed out")?? + .context("empty request")?; + let request: Request = match serde_json::from_str(&line) { + Ok(r) => r, + Err(e) => { + let error = format!("invalid serve request: {e}"); + write( + &mut stream, + &Reply::Exit { + code: 1, + error: Some(error.clone()), + }, + ) + .await?; + bail!(error) + } + }; + if request.v != PROTOCOL_V || request.relays != relays { + let error = if request.v != PROTOCOL_V { + "serve protocol version mismatch".to_string() + } else { + format!( + "serve uses relays {relays:?}; this command asked for {:?}. Use the same --relay set as serve", + request.relays + ) + }; + write( + &mut stream, + &Reply::Exit { + code: 1, + error: Some(error.clone()), + }, + ) + .await?; + bail!(error) + } + let (reply, rx) = oneshot::channel(); + tx.send(Job { + command: request.command, + reply, + }) + .await + .context("serve loop stopped")?; + // The job runs in serve's loop whether or not this client stays connected. + tokio::pin!(rx); + let outcome = loop { + tokio::select! { + r = &mut rx => break r.context("serve loop dropped the request")?, + _ = tokio::time::sleep(KEEPALIVE) => write(&mut stream, &Reply::Keepalive).await?, + } + }; + let outcome = match outcome { + Ok(o) => o, + Err(f) => { + return write( + &mut stream, + &Reply::Exit { + code: f.code, + error: Some(f.message), + }, + ) + .await; + } + }; + for line in outcome.stdout { + write(&mut stream, &Reply::Stdout(line)).await?; + } + let Some(id) = outcome.swap else { + return write( + &mut stream, + &Reply::Exit { + code: 0, + error: None, + }, + ) + .await; + }; + write( + &mut stream, + &Reply::Stderr(format!( + "serve is watching swap {id}; it keeps driving it to settlement even if this command exits. Follow it with `status`." + )), + ) + .await?; + stream_take(&mut stream, home, &id).await +} + +/// Stream one swap's state from the journal (read-only) until it is terminal. +async fn stream_take(stream: &mut UnixStream, home: &Path, id: &str) -> Result<()> { + let mut last = String::new(); + let mut warned = observe::Warned::default(); + let mut quiet = tokio::time::Instant::now(); + loop { + let row = observe::journal_rows(home, "swap")? + .into_iter() + .find(|(k, _, _)| k == id); + if let Some((_, v, updated)) = row { + let s: coordinator::Swap = serde_json::from_slice(&v)?; + if s.state != last { + last = s.state.clone(); + write( + stream, + &Reply::Stderr(serde_json::json!({"swap_id":id,"state":s.state}).to_string()), + ) + .await?; + quiet = tokio::time::Instant::now(); + } + let now = coordinator::now(); + let current: Vec<_> = observe::derive( + &s.id, + &s.role, + &s.state, + updated, + observe::refund_available_at(&s), + now, + observe::unresponsive_after(), + ) + .into_iter() + .collect(); + for u in warned.due(¤t) { + write(stream, &Reply::Stderr(u.json(now).to_string())).await?; + } + if coordinator::terminal(&s) { + for line in terminal_lines(&s) { + write(stream, &Reply::Stdout(line)).await?; + } + let ok = ["complete", "complete_unclaimed"].contains(&s.state.as_str()); + return write( + stream, + &Reply::Exit { + code: if ok { 0 } else { 1 }, + error: (!ok).then(|| format!("trade ended {}", s.state)), + }, + ) + .await; + } + } + if quiet.elapsed() >= KEEPALIVE { + write(stream, &Reply::Keepalive).await?; + quiet = tokio::time::Instant::now(); + } + tokio::time::sleep(Duration::from_millis(500)).await; + } +} +/// The stdout lines a CLI `take` prints when its swap ends in this state. +pub fn terminal_lines(s: &coordinator::Swap) -> Vec { + let mut out = vec![]; + if s.state == "complete" { + out.push(serde_json::json!({"swap_id":s.id,"state":"complete","role":s.role}).to_string()); + } + if s.state.ends_with("_quarantined") { + out.push( + serde_json::json!({"swap_id":s.id,"state":s.state,"manual_recovery":true}).to_string(), + ); + } + out.push(serde_json::json!({"swap_id":s.id,"state":s.state}).to_string()); + out +} + +/// Execute one queued request inside serve's loop. Never panics the loop. +pub async fn execute(home: &Path, j: &Journal, m: &Market, job: Job) { + let name = job.command.name(); + eprintln!( + "{}", + serde_json::json!({"serve_request":name,"bounds":job.command}) + ); + let result = run_command(home, j, m, &job.command) + .await + .map_err(|e| Failure { + code: exit_code(&e), + message: e.to_string(), + }); + match &result { + Ok(o) => eprintln!( + "{}", + serde_json::json!({"serve_request":name,"result":"ok","swap_id":o.swap}) + ), + Err(f) => eprintln!( + "{}", + serde_json::json!({"serve_request":name,"result":"error","exit":f.code,"error":f.message}) + ), + } + let _ = job.reply.send(result); +} +async fn run_command(home: &Path, j: &Journal, m: &Market, c: &Command) -> Result { + match c { + Command::List { + give_mint, + give, + want_mint, + want, + max_fees, + } => { + let id = coordinator::list( + home, + j, + m, + Leg { + asset: Asset::new(give_mint)?, + net: *give, + }, + Leg { + asset: Asset::new(want_mint)?, + net: *want, + }, + *max_fees, + ) + .await?; + let l: coordinator::Listing = j.get("listing", &id).await?.context("listing")?; + Ok(Outcome { + stdout: vec![serde_json::json!({"lot_id":id,"status_event_id":l.statuses[0].id,"status":"available","debit":l.plan.debit}).to_string()], + swap: None, + }) + } + Command::Cancel { lot } => { + coordinator::cancel(home, j, m, lot).await?; + let l: coordinator::Listing = j.get("listing", lot).await?.context("listing")?; + Ok(Outcome { + stdout: vec![serde_json::json!({"lot_id":l.event.id,"status_event_ids":l.statuses.iter().map(|e|e.id.to_hex()).collect::>(),"status":crate::lifecycle(&l.event,&l.statuses)?}).to_string()], + swap: None, + }) + } + Command::Take { + lot, + max_give, + min_receive, + max_fees, + } => { + let id = coordinator::start_take(home, j, m, lot, *max_give, *min_receive, *max_fees) + .await?; + let s: coordinator::Swap = j.get("swap", &id).await?.context("swap")?; + Ok(Outcome { + stdout: vec![serde_json::json!({"swap_id":s.id,"state":s.state}).to_string()], + swap: Some(id), + }) + } + } +} + +/// Exit code relayed from serve. The message was already printed. +#[derive(Debug)] +pub struct Forwarded(pub u8); +impl std::fmt::Display for Forwarded { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "handled by serve; exit {}", self.0) + } +} +impl std::error::Error for Forwarded {} + +/// Hand `request` to a running serve. `Ok(None)`: nothing is listening (no serve). +pub async fn forward(home: &Path, request: &Request) -> Result> { + let sp = socket_path(home)?; + let mut stream = match UnixStream::connect(&sp.path).await { + Ok(s) => s, + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused + ) => + { + return Ok(None); + } + Err(e) => return Err(e).context("connect to serve socket"), + }; + let owner = std::fs::symlink_metadata(home.join(SOCKET))?.uid(); + let peer = stream.peer_cred()?; + ensure!( + peer.uid() == owner, + "serve socket peer uid {} does not own the socket; refusing", + peer.uid() + ); + let mut line = serde_json::to_string(request)?; + ensure!(line.len() <= MAX_REQUEST_BYTES, "request too large"); + line.push('\n'); + tokio::time::timeout(IO_TIMEOUT, stream.write_all(line.as_bytes())) + .await + .context("serve write timed out")??; + eprintln!( + "{}", + serde_json::json!({"handed_to_serve":request.command.name()}) + ); + let mut swap: Option = None; + let mut stream = BufReader::new(stream); + loop { + let next = tokio::time::timeout( + CLIENT_IDLE, + read_line_bounded(&mut stream, MAX_REPLY_LINE_BYTES), + ) + .await; + let line = match next { + Ok(Ok(Some(line))) => line, + other => { + let why = match other { + Err(_) => "serve went silent".to_string(), + Ok(Err(e)) => e.to_string(), + _ => "serve closed the connection".to_string(), + }; + match &swap { + Some(id) => eprintln!( + "{why}; swap {id} is journaled and stays with the home: keep serve running or run recover" + ), + None => eprintln!( + "{why} before reporting a result; the request may have been applied. Check status" + ), + } + return Ok(Some(3)); + } + }; + match serde_json::from_str::(&line).context("invalid serve reply")? { + Reply::Stdout(s) => { + if swap.is_none() && matches!(request.command, Command::Take { .. }) { + swap = serde_json::from_str::(&s) + .ok() + .and_then(|v| v["swap_id"].as_str().map(str::to_owned)); + } + println!("{s}") + } + Reply::Stderr(s) => eprintln!("{s}"), + Reply::Keepalive => {} + Reply::Exit { code, error } => { + if let Some(e) = error { + eprintln!("{e}"); + } + return Ok(Some(code)); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn request_rejects_unknown_fields_and_keeps_bounds() { + let r = Request { + v: 1, + relays: vec!["ws://127.0.0.1:1".into()], + command: Command::Take { + lot: "x".into(), + max_give: 27, + min_receive: 32, + max_fees: 3, + }, + }; + let text = serde_json::to_string(&r).unwrap(); + assert_eq!(serde_json::from_str::(&text).unwrap(), r); + let loose = text.replace("\"max_fees\":3", "\"max_fees\":3,\"max_fees_override\":99"); + assert!(serde_json::from_str::(&loose).is_err()); + let missing = text.replace(",\"max_fees\":3", ""); + assert!( + serde_json::from_str::(&missing).is_err(), + "SAFETY: no serve-side default may fill a missing bound" + ); + } + #[tokio::test] + async fn bounded_lines() { + let mut ok: &[u8] = b"abc\n"; + assert_eq!(read_line_bounded(&mut ok, 3).await.unwrap().unwrap(), "abc"); + let mut long: &[u8] = b"abcd\n"; + assert!(read_line_bounded(&mut long, 3).await.is_err()); + let mut open: &[u8] = b"abc"; + assert!(read_line_bounded(&mut open, 8).await.is_err()); + let mut empty: &[u8] = b""; + assert!(read_line_bounded(&mut empty, 8).await.unwrap().is_none()); + } +} diff --git a/crates/maxplayer-trade/tests/receive.rs b/crates/maxplayer-trade/tests/receive.rs index 490013764..bb7a60278 100644 --- a/crates/maxplayer-trade/tests/receive.rs +++ b/crates/maxplayer-trade/tests/receive.rs @@ -366,11 +366,20 @@ async fn cli_never_prints_token_and_never_takes_it_as_argv() { assert_secret_free(&pending, &token2, &proofs2); m.faults.lose_reply.store(false, SeqCst); let status = std::process::Command::new(env!("CARGO_BIN_EXE_maxplayer-trade")) - .args(["--home", h.path().to_str().unwrap(), "status"]) + .args(["--home", h.path().to_str().unwrap(), "status", "--json"]) .output() .unwrap(); assert_secret_free(&status, &token2, &proofs2); assert!(String::from_utf8_lossy(&status.stdout).contains("\"receive\"")); + // The default readable summary is secret-free too and lists the unresolved receive. + let human = std::process::Command::new(env!("CARGO_BIN_EXE_maxplayer-trade")) + .args(["--home", h.path().to_str().unwrap(), "status"]) + .output() + .unwrap(); + assert_secret_free(&human, &token2, &proofs2); + let text = String::from_utf8_lossy(&human.stdout); + assert!(text.starts_with("UNRESOLVED ("), "{text}"); + assert!(text.contains("receive "), "{text}"); } #[cfg(feature = "lab")] diff --git a/crates/maxplayer-trade/tests/ux.rs b/crates/maxplayer-trade/tests/ux.rs new file mode 100644 index 000000000..614103077 --- /dev/null +++ b/crates/maxplayer-trade/tests/ux.rs @@ -0,0 +1,1055 @@ +//! Agent-UX safety tests: lock-free discover, serve hand-off, unresponsive warning, +//! readable status (with a byte-for-byte `status --json` golden) and fee dry-runs. +//! Every SAFETY assertion comes first in its test. +mod support; +use maxplayer_trade::{coordinator, coordinator::Swap, market::Market, money}; +use nostr_sdk::prelude::{Filter, Keys}; +use std::{collections::BTreeSet, path::Path, process::Stdio, time::Duration}; +use support::*; +use tokio::process::{Child, Command}; + +const BIN: &str = env!("CARGO_BIN_EXE_maxplayer-trade"); + +fn cmd(home: &Path, relay: &str, args: &[&str], envs: &[(&str, &str)]) -> Command { + let mut c = Command::new(BIN); + c.arg("--home") + .arg(home) + .args(["--relay", relay]) + .args(args); + for (k, v) in envs { + c.env(k, v); + } + c.stdin(Stdio::null()).kill_on_drop(true); + c +} +async fn run(home: &Path, relay: &str, args: &[&str]) -> std::process::Output { + tokio::time::timeout( + Duration::from_secs(120), + cmd(home, relay, args, &[]).output(), + ) + .await + .expect("command timed out") + .unwrap() +} +fn text(o: &std::process::Output) -> (String, String) { + ( + String::from_utf8_lossy(&o.stdout).into_owned(), + String::from_utf8_lossy(&o.stderr).into_owned(), + ) +} + +type KvRow = (String, String, Vec, i64, i64); +/// Every journal row including its timestamps: any write shows up here. +fn kv_dump(home: &Path) -> Vec { + let path = home.join("trade.sqlite"); + if !path.exists() { + return vec![]; + } + let db = + rusqlite::Connection::open_with_flags(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY) + .unwrap(); + let mut s = db + .prepare("SELECT secondary_namespace, key, value, created_time, updated_time FROM kv_store ORDER BY 1, 2") + .unwrap(); + s.query_map([], |r| { + Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)) + }) + .unwrap() + .map(Result::unwrap) + .collect() +} +/// Every wallet proof row (state and reservation) for every wallet DB in the home. +fn wallet_dump(home: &Path) -> Vec<(String, Vec, String, Option)> { + let mut out = vec![]; + for f in files(home) { + if f.ends_with(".sqlite") && f != "trade.sqlite" { + let db = rusqlite::Connection::open_with_flags( + home.join(&f), + rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY, + ) + .unwrap(); + let mut s = db + .prepare("SELECT y, state, used_by_operation FROM proof ORDER BY y") + .unwrap(); + out.extend( + s.query_map([], |r| Ok((f.clone(), r.get(0)?, r.get(1)?, r.get(2)?))) + .unwrap() + .map(Result::unwrap), + ); + } + } + out +} +fn files(home: &Path) -> BTreeSet { + std::fs::read_dir(home) + .map(|d| { + d.map(|e| e.unwrap().file_name().to_string_lossy().into_owned()) + .filter(|n| !n.ends_with("-wal") && !n.ends_with("-shm")) + .collect() + }) + .unwrap_or_default() +} +async fn relay_events(url: &str) -> BTreeSet { + let m = Market::connect(Keys::generate(), &[url.to_string()]) + .await + .unwrap(); + let (events, partial) = m.query(Filter::new().limit(4000)).await.unwrap(); + assert_eq!(partial, 0); + events.iter().map(|e| e.id.to_hex()).collect() +} +struct Snapshot { + kv: Vec, + wallet: Vec<(String, Vec, String, Option)>, + files: BTreeSet, +} +fn snapshot(home: &Path) -> Snapshot { + Snapshot { + kv: kv_dump(home), + wallet: wallet_dump(home), + files: files(home), + } +} +fn assert_unchanged(home: &Path, before: &Snapshot, what: &str) { + let after = snapshot(home); + assert!(after.kv == before.kv, "SAFETY: {what} changed the journal"); + assert!( + after.wallet == before.wallet, + "SAFETY: {what} changed wallet proofs/reservations" + ); + assert_eq!(after.files, before.files, "SAFETY: {what} created files"); +} + +async fn start_serve(home: &Path, relay: &str, log: &Path, envs: &[(&str, &str)]) -> Child { + let mut c = cmd(home, relay, &["serve"], envs); + c.stdout(Stdio::null()) + .stderr(std::fs::File::create(log).unwrap()); + let child = c.spawn().unwrap(); + let start = std::time::Instant::now(); + while !std::fs::read_to_string(log) + .unwrap_or_default() + .contains("{\"serve_socket\":\"") + { + assert!(start.elapsed() < Duration::from_secs(30), "serve socket"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + child +} +/// Step the in-process maker (and its recovery ticks) until `child` exits. +async fn drive_maker(f: &mut Fixture, child: &mut Child, limit: u64) -> std::process::ExitStatus { + let start = std::time::Instant::now(); + let mut next = std::time::Instant::now() + Duration::from_secs(3); + loop { + f.step(true).await; + if let Some(s) = child.try_wait().unwrap() { + return s; + } + if std::time::Instant::now() >= next { + let _ = coordinator::recover(&f.maker, &f.jm, &f.mm).await; + next = std::time::Instant::now() + Duration::from_secs(3); + } + assert!(start.elapsed().as_secs() < limit, "client did not finish"); + } +} + +/// (debit, total fees) the taker's dry run reports for `lot` with generous bounds. +async fn dry(f: &Fixture, lot: &str) -> (u64, u64) { + let out = run( + &f.taker, + &f.relay_url, + &[ + "take", + lot, + "--max-give", + "1000", + "--min-receive", + "1", + "--max-fees", + "100", + "--dry-run", + ], + ) + .await; + assert!(out.status.success(), "{}", text(&out).1); + let v: serde_json::Value = serde_json::from_str(&text(&out).0).unwrap(); + ( + v["max_total_debit"].as_u64().unwrap(), + v["fees_total"].as_u64().unwrap(), + ) +} +/// SAFETY: discover takes no lock, runs no recovery and writes nothing, even when the home +/// holds a swap that recovery WOULD change (a `requested` swap past its 60 s window), both +/// while another process owns `owner.lock` (serve mid-swap) and when nothing does. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discover_is_lock_free_and_never_touches_the_journal() { + let f = Fixture::new(0).await; + let lot = f.list(false).await; + let id = f.start(&lot).await; + let mut s = f.jt.get::("swap", &id).await.unwrap().unwrap(); + s.created = coordinator::now() - 120; + f.jt.put("swap", &id, &s).await.unwrap(); + let relay_before = relay_events(&f.relay_url).await; + std::fs::File::create(f.taker.join("owner.lock")).unwrap(); + let before = snapshot(&f.taker); + { + use fs2::FileExt; + let held = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(f.taker.join("owner.lock")) + .unwrap(); + held.lock_exclusive().unwrap(); + let before = snapshot(&f.taker); + let out = run(&f.taker, &f.relay_url, &["discover"]).await; + let (stdout, stderr) = text(&out); + assert!( + out.status.success(), + "SAFETY: serve must not block discover: {stderr}" + ); + assert_unchanged(&f.taker, &before, "discover while the home is owned"); + let v: serde_json::Value = serde_json::from_str(&stdout).unwrap(); + let l = &v["listings"][0]; + assert_eq!(l["lot_id"], lot); + assert_eq!(l["expected_fees"]["give"]["gross"], 32); + assert_eq!(l["expected_fees"]["want"]["claim_fee"], 0); + } + let out = run(&f.taker, &f.relay_url, &["discover"]).await; + assert!(out.status.success()); + assert_unchanged(&f.taker, &before, "discover on an unowned home"); + assert_eq!( + f.state(false, &id).await.as_deref(), + Some("requested"), + "SAFETY: discover ran no recovery" + ); + assert_eq!( + relay_events(&f.relay_url).await, + relay_before, + "SAFETY: discover published" + ); + // A fresh home is not created, keyed or journaled by discover. + let fresh = f.root.path().join("fresh"); + let out = run(&fresh, &f.relay_url, &["discover"]).await; + assert!(out.status.success()); + assert!(!fresh.exists(), "SAFETY: discover created a home"); +} + +/// Mint keysets unreachable or hanging: discover still answers, fees marked unknown. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn discover_marks_unreachable_mint_fees_unknown_without_blocking() { + let f = Fixture::new(100).await; + let lot = f.list(false).await; + f.b.faults + .hang_keysets + .store(true, std::sync::atomic::Ordering::SeqCst); + let start = std::time::Instant::now(); + let out = run(&f.root.path().join("x"), &f.relay_url, &["discover"]).await; + assert!(out.status.success(), "{}", text(&out).1); + assert!( + start.elapsed() < Duration::from_secs(30), + "discover blocked on a mint" + ); + let v: serde_json::Value = serde_json::from_str(&text(&out).0).unwrap(); + let l = &v["listings"][0]; + assert_eq!(l["lot_id"], lot); + assert_eq!(l["expected_fees"]["want"], "unknown"); + // 32 net at 100 ppk: gross 33, claim fee 1 (sender-funds-net). + assert_eq!(l["expected_fees"]["give"]["gross"], 33); + assert_eq!(l["expected_fees"]["give"]["claim_fee"], 1); +} + +/// SAFETY: a take handed to serve uses EXACTLY the client's bounds. Each looser-than-needed +/// value is refused by serve (with serve's own default 16-sat fee cap never substituted), and +/// the accepted take journals the client's exact numbers. serve then drives it to settlement. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn take_via_serve_enforces_client_bounds_exactly_and_settles() { + let mut f = Fixture::new(100).await; + let lot = f.list(false).await; + let (debit, fees) = dry(&f, &lot).await; + assert!(fees >= 2, "fee-bearing fixture"); + let (d, d1, fz, fz1) = ( + debit.to_string(), + (debit - 1).to_string(), + fees.to_string(), + (fees - 1).to_string(), + ); + let log = f.root.path().join("serve.log"); + let mut serve = start_serve(&f.taker, &f.relay_url, &log, &[]).await; + // Exactly one bound one sat too tight each time: serve must refuse, never loosen it + // (serve's own default of 16 would admit every one of these). + for (bounds, why) in [ + ([d1.as_str(), "32", "16"], "maximum give cap"), + ([d.as_str(), "33", "16"], "minimum receive cap"), + ([d.as_str(), "32", fz1.as_str()], "fee cap exceeded"), + ] { + let out = run( + &f.taker, + &f.relay_url, + &[ + "take", + &lot, + "--max-give", + bounds[0], + "--min-receive", + bounds[1], + "--max-fees", + bounds[2], + ], + ) + .await; + let (_, stderr) = text(&out); + assert!(stderr.contains("handed_to_serve"), "{stderr}"); + assert_eq!( + out.status.code(), + Some(1), + "SAFETY: {why} must refuse: {stderr}" + ); + assert!(stderr.contains(why), "{stderr}"); + assert!( + f.jt.all::("swap").await.unwrap().is_empty(), + "SAFETY: refused take journaled a swap" + ); + } + assert!( + serve.try_wait().unwrap().is_none(), + "serve survives refusals" + ); + // While serve owns the home mid-take, discover still works. + let mut client = cmd( + &f.taker, + &f.relay_url, + &[ + "take", + &lot, + "--max-give", + &d, + "--min-receive", + "32", + "--max-fees", + &fz, + ], + &[], + ) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let mut id = None; + for _ in 0..200 { + f.step(true).await; + if let Some(s) = f.jt.all::("swap").await.unwrap().pop() { + id = Some(s.id); + break; + } + } + let id = id.expect("serve journaled the take"); + let s = f.jt.get::("swap", &id).await.unwrap().unwrap(); + assert_eq!( + ( + s.request.max_give, + s.request.min_receive, + s.request.max_fees, + s.max_fees + ), + (debit, 32, fees, fees), + "SAFETY: serve journaled bounds other than the client's" + ); + let d = run(&f.taker, &f.relay_url, &["discover"]).await; + assert!(d.status.success(), "discover during serve: {}", text(&d).1); + let status = drive_maker(&mut f, &mut client, 150).await; + let out = client.wait_with_output().await.unwrap(); + let (stdout, stderr) = text(&out); + assert!(status.success(), "{stdout}\n{stderr}"); + assert!( + stderr.contains(&format!("serve is watching swap {id}")), + "{stderr}" + ); + // Same lines (serde_json orders keys) as a CLI-driven take prints. + assert!( + stdout.contains(&serde_json::json!({"swap_id":id,"state":"requested"}).to_string()), + "{stdout}" + ); + assert!( + stdout.contains(&serde_json::json!({"swap_id":id,"state":"complete"}).to_string()), + "{stdout}" + ); + assert_eq!(f.state(false, &id).await.as_deref(), Some("complete")); + serve.kill().await.unwrap(); + let log = std::fs::read_to_string(&log).unwrap(); + assert!( + log.contains(&format!("\"max_give\":{debit}")) + && log.contains(&format!("\"max_fees\":{fees}")), + "{log}" + ); +} + +/// A client that disconnects right after handing off never half-applies the request: +/// serve completes the journaled take on its own. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn disconnected_client_leaves_take_with_serve() { + let mut f = Fixture::new(0).await; + let lot = f.list(false).await; + let log = f.root.path().join("serve.log"); + let mut serve = start_serve(&f.taker, &f.relay_url, &log, &[]).await; + let mut client = cmd( + &f.taker, + &f.relay_url, + &["take", &lot, "--max-give", "40", "--min-receive", "32"], + &[], + ) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let mut id = None; + for _ in 0..200 { + f.step(true).await; + if let Some(s) = f.jt.all::("swap").await.unwrap().pop() { + id = Some(s.id); + break; + } + } + client.kill().await.unwrap(); + let id = id.unwrap(); + let start = std::time::Instant::now(); + let mut next = std::time::Instant::now(); + while f.state(false, &id).await.as_deref() != Some("complete") { + f.step(true).await; + if std::time::Instant::now() >= next { + let _ = coordinator::recover(&f.maker, &f.jm, &f.mm).await; + next = std::time::Instant::now() + Duration::from_secs(3); + } + assert!( + start.elapsed() < Duration::from_secs(150), + "serve did not settle" + ); + } + assert_eq!(balance(&f.taker, &f.a.url).await, 32); + serve.kill().await.unwrap(); +} + +/// A dead serve's socket file never blocks the CLI; a new serve replaces it; a non-socket +/// file is never removed; an oversized or malformed request is refused and serve stays up. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn stale_socket_and_bounded_protocol() { + let f = Fixture::new(0).await; + let lot = f.list(false).await; + // Stale: a bound-then-dropped listener leaves a socket file nobody serves. + drop(std::os::unix::net::UnixListener::bind(f.maker.join("serve.sock")).unwrap()); + assert!(f.maker.join("serve.sock").exists()); + let out = run(&f.maker, &f.relay_url, &["cancel", &lot]).await; + let (stdout, stderr) = text(&out); + assert!( + out.status.success(), + "stale socket blocked cancel: {stderr}" + ); + assert!(!stderr.contains("handed_to_serve")); + assert!(stdout.contains("cancelled"), "{stdout}"); + let log = f.root.path().join("serve.log"); + let mut serve = start_serve(&f.maker, &f.relay_url, &log, &[]).await; + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(f.maker.join("serve.sock")) + .unwrap() + .permissions() + .mode(); + assert_eq!(mode & 0o777, 0o600, "SAFETY: socket mode"); + // Oversized request: refused, connection closed, serve alive and still serving. + { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut s = tokio::net::UnixStream::connect(f.maker.join("serve.sock")) + .await + .unwrap(); + let _ = s.write_all(&vec![b'x'; 64 * 1024]).await; + let mut buf = vec![]; + let _ = tokio::time::timeout(Duration::from_secs(15), s.read_to_end(&mut buf)).await; + let mut s = tokio::net::UnixStream::connect(f.maker.join("serve.sock")) + .await + .unwrap(); + s.write_all( + b"{\"v\":1,\"relays\":[],\"command\":{\"cancel\":{\"lot\":\"x\",\"extra\":1}}}\n", + ) + .await + .unwrap(); + let mut reply = String::new(); + tokio::time::timeout(Duration::from_secs(15), s.read_to_string(&mut reply)) + .await + .unwrap() + .unwrap(); + assert!(reply.contains("invalid serve request"), "{reply}"); + } + // A request for a different relay set is refused, not silently re-routed. + let other = run(&f.maker, "ws://127.0.0.1:9", &["cancel", &lot]).await; + assert_eq!(other.status.code(), Some(1)); + assert!( + text(&other).1.contains("same --relay set"), + "{}", + text(&other).1 + ); + let out = run(&f.maker, &f.relay_url, &["cancel", &lot]).await; + let (_, stderr) = text(&out); + assert!(stderr.contains("handed_to_serve"), "{stderr}"); + assert_eq!( + out.status.code(), + Some(1), + "terminal lot: same exit as the CLI" + ); + assert!(stderr.contains("lot is terminal"), "{stderr}"); + assert!(serve.try_wait().unwrap().is_none()); + serve.kill().await.unwrap(); + serve.wait().await.unwrap(); + // A non-socket file at the socket path is never removed. + std::fs::remove_file(f.maker.join("serve.sock")).unwrap(); + std::fs::write(f.maker.join("serve.sock"), b"not a socket").unwrap(); + let out = run(&f.maker, &f.relay_url, &["serve"]).await; + assert!(!out.status.success()); + assert!(text(&out).1.contains("not a socket")); + assert_eq!( + std::fs::read(f.maker.join("serve.sock")).unwrap(), + b"not a socket" + ); +} + +#[cfg(feature = "lab")] +fn now_ns() -> u128 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() +} +/// SAFETY: two clients and a starting serve race for one home. Writer tenures recorded by +/// the lab trace never overlap, every client either succeeded or was refused before doing +/// anything, and the journaled listings reserve disjoint inputs. +#[cfg(feature = "lab")] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_clients_and_serve_never_make_two_writers() { + let f = Fixture::new(0).await; + for _ in 0..3 { + fund(&f.maker, &f.a.url, 64).await; + } + let trace = f.root.path().join("writers.trace"); + let trace_s = trace.to_str().unwrap().to_string(); + let env = [("TRADE_LAB_WRITER_TRACE", trace_s.as_str())]; + let list = [ + "list", + "--give-mint", + &f.a.url, + "--give", + "16", + "--want-mint", + &f.b.url, + "--want", + "8", + ]; + let mut ok = 0; + let mut killed = std::collections::BTreeMap::new(); + for round in 0..3 { + let log = f.root.path().join(format!("serve{round}.log")); + let mut serve = cmd(&f.maker, &f.relay_url, &["serve"], &env); + serve + .stdout(Stdio::null()) + .stderr(std::fs::File::create(&log).unwrap()); + let mut serve = serve.spawn().unwrap(); + let c1 = cmd(&f.maker, &f.relay_url, &list, &env).output(); + let c2 = cmd(&f.maker, &f.relay_url, &list, &env).output(); + let (o1, o2) = tokio::join!(c1, c2); + for o in [o1.unwrap(), o2.unwrap()] { + let (_, stderr) = text(&o); + if o.status.success() { + ok += 1; + } else { + assert!(stderr.contains("home is already in use"), "{stderr}"); + } + } + tokio::time::sleep(Duration::from_millis(300)).await; + let pid = serve.id().unwrap().to_string(); + if let Some(s) = serve.try_wait().unwrap() { + assert!(!s.success()); + assert!( + std::fs::read_to_string(&log) + .unwrap() + .contains("home is already in use") + ); + } else { + serve.kill().await.unwrap(); + serve.wait().await.unwrap(); + // SIGKILL logs no release; the kernel dropped the lock no later than now. + killed.insert(pid, now_ns()); + } + } + // Tenures: (pid, acquire, release-or-end). A killed serve never logs release. + let raw = std::fs::read_to_string(&trace).unwrap(); + let mut open: std::collections::BTreeMap = Default::default(); + let mut spans = vec![]; + for line in raw.lines() { + let p: Vec<_> = line.split(' ').collect(); + let t: u128 = p[2].parse().unwrap(); + if p[0] == "acquire" { + open.insert(p[1].into(), t); + } else { + spans.push((open.remove(p[1]).unwrap(), t)); + } + } + for (pid, start) in open { + spans.push(( + start, + *killed + .get(&pid) + .expect("unreleased tenure of a live process"), + )); + } + spans.sort(); + for w in spans.windows(2) { + assert!( + w[1].0 >= w[0].1, + "SAFETY: two writers: {:?} overlaps {:?}\n{raw}", + w[0], + w[1] + ); + } + let listings = f.jm.all::("listing").await.unwrap(); + // One listing from Fixture::list is absent here (none was made); every success journaled one. + assert_eq!( + listings.len(), + ok, + "SAFETY: success count equals journaled listings" + ); + let mut ys = BTreeSet::new(); + for l in &listings { + for p in &l.plan.inputs { + assert!( + ys.insert(p.secret.to_string()), + "SAFETY: one input reserved twice" + ); + } + } + assert!(ok >= 3, "races produced too few successful lists ({ok})"); +} + +/// While serve lives, its lock blocks every CLI writer: they hand off instead of acquiring. +#[cfg(feature = "lab")] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn live_serve_owns_every_write() { + let f = Fixture::new(0).await; + fund(&f.maker, &f.a.url, 64).await; + let trace = f.root.path().join("writers.trace"); + let trace_s = trace.to_str().unwrap().to_string(); + let env = [("TRADE_LAB_WRITER_TRACE", trace_s.as_str())]; + let log = f.root.path().join("serve.log"); + let mut serve = start_serve(&f.maker, &f.relay_url, &log, &env).await; + let list = [ + "list", + "--give-mint", + &f.a.url, + "--give", + "16", + "--want-mint", + &f.b.url, + "--want", + "8", + ]; + let (o1, o2) = tokio::join!( + cmd(&f.maker, &f.relay_url, &list, &env).output(), + cmd(&f.maker, &f.relay_url, &list, &env).output() + ); + for o in [o1.unwrap(), o2.unwrap()] { + let (stdout, stderr) = text(&o); + assert!(o.status.success(), "{stderr}"); + assert!(stderr.contains("handed_to_serve"), "{stderr}"); + assert!(stdout.contains("\"status\":\"available\""), "{stdout}"); + } + let raw = std::fs::read_to_string(&trace).unwrap(); + assert_eq!( + raw.lines().filter(|l| l.starts_with("acquire")).count(), + 1, + "SAFETY: only serve may acquire while it runs:\n{raw}" + ); + serve.kill().await.unwrap(); +} + +/// SAFETY: `--dry-run` has zero side effects (journal, reservations, wallet DB, relay, +/// maker inbox, mint swaps, files), and its numbers equal the real plan built afterwards. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn dry_run_has_zero_side_effects_and_matches_real_plan() { + let mut f = Fixture::new(100).await; + let lot = f.list(false).await; + let relay_before = relay_events(&f.relay_url).await; + let swaps = ( + f.a.faults.swaps.load(std::sync::atomic::Ordering::SeqCst), + f.b.faults.swaps.load(std::sync::atomic::Ordering::SeqCst), + ); + let (taker0, maker0) = (snapshot(&f.taker), snapshot(&f.maker)); + let take = run( + &f.taker, + &f.relay_url, + &[ + "take", + &lot, + "--max-give", + "1000", + "--min-receive", + "32", + "--max-fees", + "100", + "--dry-run", + ], + ) + .await; + let list = run( + &f.maker, + &f.relay_url, + &[ + "list", + "--give-mint", + &f.a.url, + "--give", + "32", + "--want-mint", + &f.b.url, + "--want", + "24", + "--max-fees", + "16", + "--dry-run", + ], + ) + .await; + assert_unchanged(&f.taker, &taker0, "take --dry-run"); + assert_unchanged(&f.maker, &maker0, "list --dry-run"); + assert_eq!( + relay_events(&f.relay_url).await, + relay_before, + "SAFETY: dry-run published" + ); + assert!( + tokio::time::timeout(Duration::from_millis(500), f.mm.inbox.recv()) + .await + .is_err(), + "SAFETY: dry-run sent the maker a message" + ); + assert_eq!( + ( + f.a.faults.swaps.load(std::sync::atomic::Ordering::SeqCst), + f.b.faults.swaps.load(std::sync::atomic::Ordering::SeqCst) + ), + swaps, + "SAFETY: dry-run swapped at a mint" + ); + assert!(take.status.success(), "{}", text(&take).1); + assert!(list.status.success(), "{}", text(&list).1); + let t: serde_json::Value = serde_json::from_str(&text(&take).0).unwrap(); + assert_eq!(t["you_receive"]["leg"]["claim_fee"], 1); + assert_eq!(t["you_receive"]["leg"]["gross"], 33); + let l: serde_json::Value = serde_json::from_str(&text(&list).0).unwrap(); + assert_eq!(l["you_receive"]["leg"]["claim_fee"], 1); + // The real paths build the same plans. + let id = f.start(&lot).await; + let s = f.jt.get::("swap", &id).await.unwrap().unwrap(); + let g = &t["you_give"]; + assert_eq!( + ( + g["gross"].as_u64(), + g["lock_fee"].as_u64(), + g["claim_fee"].as_u64(), + t["max_total_debit"].as_u64() + ), + ( + Some(s.plan.gross), + Some(s.plan.lock_fee), + Some(s.plan.claim_fee), + Some(s.plan.debit) + ) + ); + // The real list right after the dry run, against the same (unchanged) wallet. + let real = coordinator::list( + &f.maker, + &f.jm, + &f.mm, + maxplayer_trade::Leg { + asset: maxplayer_trade::Asset::new(&f.a.url).unwrap(), + net: 32, + }, + maxplayer_trade::Leg { + asset: maxplayer_trade::Asset::new(&f.b.url).unwrap(), + net: 24, + }, + 16, + ) + .await + .unwrap(); + let listing: coordinator::Listing = f.jm.get("listing", &real).await.unwrap().unwrap(); + let g = &l["you_give"]; + assert_eq!( + ( + g["gross"].as_u64(), + g["lock_fee"].as_u64(), + l["max_total_debit"].as_u64() + ), + ( + Some(listing.plan.gross), + Some(listing.plan.lock_fee), + Some(listing.plan.debit) + ) + ); +} + +/// A refused or unsolvable dry-run says why, exits 1, and still changes nothing; a dry-run +/// on a fresh home creates no home, key or wallet. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn dry_run_refusals_and_unsolvable_amounts_are_reported() { + let f = Fixture::new(100).await; + let lot = f.list(false).await; + let (debit, _) = dry(&f, &lot).await; + let before = snapshot(&f.taker); + let tight = (debit - 1).to_string(); + let out = run( + &f.taker, + &f.relay_url, + &[ + "take", + &lot, + "--max-give", + &tight, + "--min-receive", + "32", + "--dry-run", + ], + ) + .await; + assert_unchanged(&f.taker, &before, "refused take --dry-run"); + assert_eq!(out.status.code(), Some(1)); + assert!( + text(&out).1.contains(&format!( + "maximum give cap: debit {debit} > --max-give {tight}" + )), + "{}", + text(&out).1 + ); + let fresh = f.root.path().join("fresh"); + let out = run( + &fresh, + &f.relay_url, + &[ + "take", + &lot, + "--max-give", + "1000", + "--min-receive", + "32", + "--dry-run", + ], + ) + .await; + assert!(!fresh.exists(), "SAFETY: dry-run created a home"); + assert!(text(&out).1.contains("insufficient unreserved balance")); + // 1000 ppk: 2 net sats has no fee-inclusive split. + let m = MintFixture::start(1000).await; + let out = run( + &fresh, + &f.relay_url, + &[ + "list", + "--give-mint", + &m.url, + "--give", + "2", + "--want-mint", + &f.b.url, + "--want", + "8", + "--dry-run", + ], + ) + .await; + assert!(!fresh.exists(), "SAFETY: dry-run created a home"); + assert_eq!(out.status.code(), Some(1)); + assert!( + text(&out) + .1 + .contains("cannot solve bounded fee-inclusive split"), + "{}", + text(&out).1 + ); +} + +/// `status` before this change, verbatim from 7d8cd74 `wallet::read_status` — the golden. +fn status_golden(home: &Path) -> String { + let path = home.join("trade.sqlite"); + if !path.exists() { + return format!("{}\n", serde_json::json!({"records":[]})); + } + let db = + rusqlite::Connection::open_with_flags(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY) + .unwrap(); + let mut stmt = db.prepare("SELECT secondary_namespace, value FROM kv_store WHERE primary_namespace='trade-v1' AND secondary_namespace IN ('swap','funding','withdrawal','receive')").unwrap(); + let mut records = vec![]; + for row in stmt + .query_map([], |r| { + Ok((r.get::<_, String>(0)?, r.get::<_, Vec>(1)?)) + }) + .unwrap() + { + let (kind, bytes) = row.unwrap(); + let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + let public = if kind == "withdrawal" { + serde_json::from_slice::(&bytes) + .unwrap() + .summary() + } else { + serde_json::json!({"kind":kind,"id":v["id"],"state":v["state"],"done":v["done"],"expired_unpaid":v["expired_unpaid"]}) + }; + records.push(public); + } + format!("{}\n", serde_json::json!({"records":records})) +} + +/// SAFETY: `status --json` is byte-for-byte the old `status`; the default summary is +/// readable, lists unresolved work first, explains exit codes and prints no secrets. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn status_json_matches_golden_and_summary_is_readable_and_secret_free() { + let mut f = Fixture::new(0).await; + let empty = f.root.path().join("empty"); + let out = std::process::Command::new(BIN) + .args(["--home", empty.to_str().unwrap(), "status", "--json"]) + .output() + .unwrap(); + assert_eq!( + String::from_utf8(out.stdout).unwrap(), + status_golden(&empty) + ); + let lot = f.list(false).await; + let id = f.start(&lot).await; + f.pump(&id, "complete").await; + let _ = money::fund(&f.taker, &f.jt, &f.b.url, 10).await.unwrap(); + for home in [&f.taker, &f.maker] { + let json = std::process::Command::new(BIN) + .args(["--home", home.to_str().unwrap(), "status", "--json"]) + .output() + .unwrap(); + assert!(json.status.success()); + assert_eq!( + String::from_utf8(json.stdout.clone()).unwrap(), + status_golden(home), + "SAFETY: status --json drifted from the golden" + ); + let human = std::process::Command::new(BIN) + .args(["--home", home.to_str().unwrap(), "status"]) + .output() + .unwrap(); + assert!(human.status.success()); + let h = String::from_utf8(human.stdout).unwrap(); + for j in [&f.jt, &f.jm] { + for s in j.all::("swap").await.unwrap() { + assert!(!h.contains(&s.key), "SAFETY: key printed"); + if let Some(p) = &s.preimage { + assert!(!h.contains(p.as_str()), "SAFETY: preimage printed"); + } + for p in s.outgoing.iter().chain(&s.incoming) { + assert!(!h.contains(&p.secret.to_string()), "SAFETY: proof printed"); + } + } + } + assert!(h.contains(&format!("swap {id} complete")), "{h}"); + assert!(h.contains("Exit codes:"), "{h}"); + if home == &f.taker { + assert!( + h.contains(&format!("+32 {} / -24 {}; fees 0", f.a.url, f.b.url)), + "{h}" + ); + assert!(h.starts_with("UNRESOLVED (1):\n funding "), "{h}"); + let invoice = f.jt.all::("funding").await.unwrap()[0] + .invoice + .clone() + .unwrap(); + assert!(!h.contains(&invoice), "SAFETY: invoice printed"); + } else { + assert!( + h.contains(&format!("lot {lot} sold; swap {id} complete (maker)")), + "{h}" + ); + assert!( + h.contains(&format!("+24 {} / -32 {}; fees 0", f.b.url, f.a.url)), + "{h}" + ); + assert!(h.contains("UNRESOLVED: none"), "{h}"); + } + } +} + +/// counterparty_unresponsive is derived display state: it appears in status, in a +/// serve-handled take's live output and in serve's log, and changes no swap state. +#[cfg(feature = "lab")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn counterparty_unresponsive_warning_is_display_only() { + let mut f = Fixture::new(0).await; + let lot = f.list(false).await; + let id = f.start(&lot).await; + f.step(true).await; + f.step(false).await; + assert_eq!(f.state(false, &id).await.as_deref(), Some("first_locked")); + let before = kv_dump(&f.taker); + tokio::time::sleep(Duration::from_millis(2500)).await; + let status = |envs: &'static [(&'static str, &'static str)], home: std::path::PathBuf| { + let mut c = std::process::Command::new(BIN); + c.args(["--home", home.to_str().unwrap(), "status"]); + for (k, v) in envs { + c.env(k, v); + } + String::from_utf8(c.output().unwrap().stdout).unwrap() + }; + let quiet = status(&[], f.taker.clone()); + assert!( + !quiet.contains("counterparty_unresponsive"), + "default threshold is 180 s" + ); + let warned = status(&[("TRADE_LAB_UNRESPONSIVE_SECONDS", "1")], f.taker.clone()); + assert!( + warned.starts_with(&format!( + "WARNING counterparty_unresponsive: swap {id} (taker first_locked) waiting for maker_second_lock for " + )), + "{warned}" + ); + assert!(warned.contains("; refund available at "), "{warned}"); + assert!(kv_dump(&f.taker) == before, "SAFETY: status wrote"); + assert_eq!(f.state(false, &id).await.as_deref(), Some("first_locked")); + // Live: a serve-handled take whose maker never answers. + let g = Fixture::new(0).await; + let lot = g.list(false).await; + let log = g.root.path().join("serve.log"); + let env = [("TRADE_LAB_UNRESPONSIVE_SECONDS", "1")]; + let mut serve = start_serve(&g.taker, &g.relay_url, &log, &env).await; + let mut client = cmd( + &g.taker, + &g.relay_url, + &["take", &lot, "--max-give", "40", "--min-receive", "32"], + &env, + ) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + use tokio::io::AsyncBufReadExt; + let mut lines = tokio::io::BufReader::new(client.stderr.take().unwrap()).lines(); + let found = tokio::time::timeout(Duration::from_secs(30), async { + while let Some(l) = lines.next_line().await.unwrap() { + if l.contains("\"warning\":\"counterparty_unresponsive\"") { + return l; + } + } + panic!("client ended without a warning") + }) + .await + .expect("no live warning"); + assert!(found.contains("\"waiting_for\":\"maker_quote\""), "{found}"); + assert!(found.contains("\"elapsed_seconds\""), "{found}"); + client.kill().await.unwrap(); + tokio::time::sleep(Duration::from_secs(4)).await; + assert!( + std::fs::read_to_string(&log) + .unwrap() + .contains("counterparty_unresponsive"), + "serve log" + ); + let s = g.jt.all::("swap").await.unwrap(); + assert_eq!(s.len(), 1); + assert_eq!(s[0].state, "requested", "SAFETY: warning changed state"); + serve.kill().await.unwrap(); +} From 43b11936fd65159a66a2a708a1ca38efc1841d4b Mon Sep 17 00:00:00 2001 From: maxie-agent <330537927+maxie-agent@users.noreply.github.com> Date: Fri, 9 Oct 2026 23:37:31 +0000 Subject: [PATCH 2/5] docs(trade): skill, commands and recovery references for agent UX --- crates/maxplayer-trade/README.md | 12 ++- .../skills/maxplayer-trade/SKILL.md | 76 +++++++++---------- .../maxplayer-trade/references/commands.md | 51 +++++++++++-- .../maxplayer-trade/references/recovery.md | 14 +++- 4 files changed, 105 insertions(+), 48 deletions(-) diff --git a/crates/maxplayer-trade/README.md b/crates/maxplayer-trade/README.md index 67b206790..f30250f19 100644 --- a/crates/maxplayer-trade/README.md +++ b/crates/maxplayer-trade/README.md @@ -20,7 +20,14 @@ Earlier dated verification sections are historical snapshots, not current covera never-submitted quotes that expired, are terminal `refused`, retryable explicitly with the same invoice. No passive recovery submits an unsent quote; an explicitly failed pre-POST withdrawal cannot pay later on a serve tick. Submitted ambiguity retains exact authorization and replays it until definitive. -- Read-only `status` and `balance` use SQLite snapshots without the writer lock. Keep +- Read-only `status` (readable summary; `--json` keeps the raw records byte-for-byte) and + `balance` use SQLite snapshots without the writer lock; `discover` reads only relays and + mint keysets (per-lot `expected_fees`, `unknown` if a mint is slow) with no lock, home or + journal access. While `serve` runs, `list`/`take`/`cancel` hand off to it over + `/serve.sock` (0600, same uid, 16 KiB, one request per connection) and serve executes + them with the client's exact bounds as the single writer. `list`/`take --dry-run` preview + both legs' fees, gross amounts and max total debit with zero side effects. A swap waiting + >180 s on its counterparty shows `counterparty_unresponsive` (display only). Keep serve alive from second lock through settlement. Exits: 0 success, 1 command error, pre-submission refusal, or definitive terminal `unpaid_released`, 2 usage, 3 unresolved work, 4 terminal manual recovery. Non-final withdrawals never exit 0; a submitted (past `quote_created`) non-terminal withdrawal always @@ -40,7 +47,7 @@ sections below describe their dated runs, not current coverage totals. ## Implemented -- `list`, `discover`, `cancel`, `serve`, `take`, `recover`, `preflight`, `fund`, `balance`, `status`, `withdraw`. +- `list`, `discover`, `cancel`, `serve`, `take`, `recover`, `preflight`, `fund`, `balance`, `status`, `withdraw`, `receive`. - Immutable signed 3410 lots (24 hours), hash-chained 3411 statuses, signature/content/tag validation, fork/gap quarantine, no terminal reopening. Every configured relay is queried; discovery unions their results. No EOSE is an error, not an empty market; partial relay @@ -239,6 +246,7 @@ LOT="$(python3 -c 'import json,sys; print([json.loads(l)["lot_id"] for l in sys. "$TRADE" --home "$RUN/maker" serve > "$RUN/maker.log" 2>&1 & MAKER_PID=$! "$TRADE" --home "$RUN/taker" discover +"$TRADE" --home "$RUN/taker" take "$LOT" --max-give 40 --min-receive 32 --dry-run "$TRADE" --home "$RUN/taker" take "$LOT" --max-give 40 --min-receive 32 # Only stop the maker after both sides are complete; otherwise keep recovery running. kill "$MAKER_PID" diff --git a/crates/maxplayer-trade/skills/maxplayer-trade/SKILL.md b/crates/maxplayer-trade/skills/maxplayer-trade/SKILL.md index 7bea9d294..1588ad99a 100644 --- a/crates/maxplayer-trade/skills/maxplayer-trade/SKILL.md +++ b/crates/maxplayer-trade/skills/maxplayer-trade/SKILL.md @@ -5,43 +5,43 @@ description: Trade Cashu tokens for a human with the standalone maxplayer-trade # Trade Cashu for the human -> Round-five operating policy. Read the scoped evidence and -> verification limits in [verification](references/verification.md). +> Evidence and limits: [verification](references/verification.md). ## 1. Establish custody and scope -Use the standalone binary, not Maxplayer jobs or the Maxplayer wallet/daemon. Read -[commands and limits](references/commands.md) before constructing a command and +Use the standalone binary, not Maxplayer jobs or wallet/daemon. Read +[commands](references/commands.md) before a command and [recovery](references/recovery.md) before handling an interruption. Choose **one private, persistent absolute home** for this human's trade wallet. Put `--home ` before **every** command. Preserve that same home, keys, databases, and journals across restarts. Never delete it while funds, reservations, unresolved funding/withdrawals, or trades exist; retain it for historical recovery afterward. -Never use a fresh home to evade caps. Only one writer may own it. Read-only `status` and `balance` work while `serve` runs. +Never use a fresh home to evade caps. Only one writer owns it: while `serve` runs, +`list`/`take`/`cancel` hand off to it with your exact bounds. `status`, `balance` and +`discover` are lock-free and work while `serve` runs; never stop serve to look. **Do not stop serve between `second_locked` and `settling`**; a maker can lose its claim window. Defer other writing commands until live obligations settle. Never copy its seed into another active wallet. -Never print secrets, seeds, keys, tokens, proofs, preimages, or raw journals; never -attach a home or database to chat. Report only public IDs, mint URLs, amounts, states, -and redacted errors. Invoice delivery belongs in the human's private conversation. +Never print or attach secrets, seeds, keys, tokens, proofs, preimages, raw journals, +homes or databases. Report only public IDs, mints, amounts, states, redacted errors. Invoice delivery belongs in the human's private conversation. **Done:** the correct home and authorized mint identities (URL + sat unit) are known. ## 2. Check readiness and get explicit authorization -Use `balance ` and `preflight ` for each relevant mint. Passing preflight -allows a mint, but does not establish honesty or complete compatibility. The CLI +Run `balance ` and `preflight ` per mint. Passing preflight +does not establish honesty or full compatibility. The CLI also preflights before fund/list/take/withdraw: NUT-07/09/11/12/14, an active sat keyset, reachable mint, and clock skew at most 60 seconds. Real mints need no opt-in setting. Never bypass a failed check. Gross locks and cumulative funding per mint per home are capped at **100,000 sats**, as is each withdrawal invoice. The withdrawal Lightning fee reserve ceiling is **max(32 sats, 2% of invoice sats rounded up)**. Pass -`--max-debit` to cap invoice amount + Cashu input fee + Lightning reserve. Fees can make an otherwise -eligible net lock amount too large. +`--max-debit` to cap invoice amount + Cashu input fee + Lightning reserve. Fees can push a net lock over the cap. **Before every real-money `fund`, `list`, `take`, or `withdraw`, present the exact -operation and get an explicit yes.** A broad “trade for me”, previous trade approval, +operation and get an explicit yes.** For `list`/`take`, first run the same command +with `--dry-run` (changes nothing) and show its fees, gross amounts and max total debit. A broad “trade for me”, previous trade approval, or a published listing is not approval for a new operation. State: - Both mint URLs and direction (or the single funding/withdrawal mint and destination). @@ -50,15 +50,15 @@ or a published listing is not approval for a new operation. State: - Worst-case normal trade lock budget: **about 75 minutes**, not a guaranteed release time. Protocol locks are 60/15 minutes plus margins; outages or ambiguous evidence can hold funds longer, indefinitely with a dishonest mint. Funding and withdrawal - have their own unresolved-payment risk, not a guaranteed 75-minute timeout. + have their own unresolved-payment risk. - A mint can steal its tokens. If the counterparty vanishes, funds stay locked until the refund deadline and successful recovery. `serve` must stay running while locks are live (the active `take` also watches its trade). NUT-14 claims remain valid after locktime. Nutshell refund status and limits: [verification](references/verification.md). NUT-07 advertisement alone does not prove witness emission. The maker relies on the chosen mint reporting the HTLC witness if the taker withholds its notice. - Mint choice is the buyer’s/seller’s responsibility; there is no rating or verification gate. + Mint choice is the trader's responsibility; there is no rating gate. -Use the human's price and fee budget, never more permissive limits. If fees are -unspecified, propose a numeric cap and wait for yes; the CLI default is not consent. +Use the human's price and fee budget, never looser limits. If fees are +unspecified, propose a numeric cap and wait for yes; CLI defaults are not consent. If a command cannot enforce the approved bounds, **do not run it**: explain the missing enforcement and escalate. Do not substitute a guessed flag or a larger cap. **Done:** an explicit yes binds this exact action, amounts, limits, and risks. @@ -68,12 +68,12 @@ missing enforcement and escalate. Do not substitute a guessed flag or a larger c ### Fund my trade wallet After confirmation, `fund --amount `. On a real mint it prints a BOLT11 -invoice and never auto-pays. Give that invoice privately to the human to pay from -**their own wallet**; the agent must never pay without separate explicit consent. -Do not treat printing an invoice as funded balance. Preserve the quote ID; after +invoice and never auto-pays. Give it privately to the human to pay from **their own +wallet**; never pay it without separate explicit consent. +Preserve the quote ID; after interruption use `recover`, and when needed resume that same funding quote with -`fund --amount --quote ` after confirmation. Never create -or pay a replacement merely because a reply or issuance timed out. +`fund --amount --quote ` after confirmation. Never create or pay +a replacement because a reply or issuance timed out. **Done:** issuance is reconciled and `balance ` confirms the result, or the retained quote is reported as unresolved without another payment. @@ -82,16 +82,16 @@ retained quote is reported as unresolved without another payment. Needs the human's yes (mint URL, token amount); it charges the per-mint cap. Save the token to a private file; run `receive --token-file ` (or stdin). Never put a token in argv, chat or logs. Only plain sat proofs from that -exact mint, at most 100,000 sats; the input fee is deducted. Exit 3 is not failure: -`recover`, never re-import elsewhere. **Done:** state `done` and `balance `. +exact mint, at most 100,000 sats; the input fee is deducted. Exit 3: `recover`; never +re-import elsewhere. **Done:** state `done` and `balance `. ### Sell/list X of mint A for Y of mint B After confirmation, `list --give-mint --give --want-mint --want ---max-fees `. X and Y are net lot amounts, not interchangeable “bitcoin” assets. +--max-fees `. X and Y are net lot amounts; mints are not interchangeable. The seller's maximum debit is X plus the approved outgoing fee cap. Record the lot -ID and keep `serve` running until complete or refunded. Listing authorizes serving -that fixed lot; it does not authorize further listings or price changes. +ID and keep `serve` running until complete or refunded. Listing authorizes only +that fixed lot, not more listings or price changes. `cancel ` is allowed only with **no active quote/swap**. Cancellation never revokes an authorized HTLC lock. On rejection, continue recovery; do not force-release @@ -100,16 +100,17 @@ inactive cancellation is reported; an unresolved lock is never called cancelled. ### Find listings / buy that lot -`discover`, inspect the exact mint pair and public lot ID, and match the human's -price. Discovery alone authorizes no purchase. For “receive X A for at most Y B”, +`discover` (shows expected fees per lot, or `unknown`), inspect the exact mint pair +and public lot ID, and match the human's price. Discovery authorizes no purchase. For “receive X A for at most Y B”, set `--min-receive X` and `--max-give Y` when Y is the total spending ceiling. If Y was explicitly a net price plus a separately approved fee F, the total ceiling can be Y + F; otherwise never silently add fees. A stricter limit is fine, a looser one is not. See the worked example in the command reference. After confirmation, `take --max-give --min-receive ---max-fees `. Keep it running through settlement. Never repeat a take to “fix” -a timeout. **Done:** verify the terminal result and per-mint balances, or enter recovery. +--max-fees `. Keep it (or serve) running through settlement. Never repeat a take +to “fix” a timeout. `counterparty_unresponsive` is a warning, not a failure: keep +serve up and wait for settlement or the printed refund time. **Done:** verify the terminal result and per-mint balances, or enter recovery. ### Withdraw @@ -133,21 +134,20 @@ funding/withdrawal/receive/swap item**, without waiting for lock deadlines: exit all terminal with no deferred work, **3** unresolved/deferred, **4** terminal manual recovery, **1** command error, pre-submission refusal, or definitive terminal `unpaid_released` (never proof a submitted withdrawal was refused; a submitted non-terminal withdrawal exits **3**), **2** CLI usage error. Timeout does not undo an RPC or release reservations. Restore `serve` while locks -are live; a single recovery pass is not a watcher. See the recovery reference for -quarantine and errors. +are live; one recovery pass is not a watcher. **Done:** classify every reported outcome using the recovery reference. ## 4. Verify and report -Check public swap/withdrawal state with `status` and balances with `balance` -without stopping the owner process. A zero exit, elapsed deadline, missing listing, or mint saying PAID is not -by itself reconciled success. Keep the home. Report spent/received amounts and fees +Check state with `status` (summary, unresolved first; `--json` for raw records) +and balances with `balance` without stopping the owner process. A zero exit, elapsed deadline, missing listing or PAID mint is not alone +reconciled success. Keep the home. Report spent/received amounts and fees only when known; distinguish complete, refunded, unresolved, and manual recovery. On `refund_quarantined` or `claim_quarantined`, stop new money actions and escalate to the human immediately. These are terminal **manual recovery**, not success or spendable funds. Never credit/import unverifiable outputs, edit journal records, replace attempts, or treat quarantine as refund permission. Persistent RPC failure, -missing witnesses, unexpected states, or accounting discrepancies also need a human; -retain safe recovery for other obligations, without weakening any check. +missing witnesses, unexpected states or accounting gaps also need a human; keep safe +recovery for other obligations without weakening any check. **Done:** the human has an accurate outcome and any unresolved custody obligations. diff --git a/crates/maxplayer-trade/skills/maxplayer-trade/references/commands.md b/crates/maxplayer-trade/skills/maxplayer-trade/references/commands.md index c595ed6f3..e6c015ebf 100644 --- a/crates/maxplayer-trade/skills/maxplayer-trade/references/commands.md +++ b/crates/maxplayer-trade/skills/maxplayer-trade/references/commands.md @@ -15,14 +15,14 @@ Canonical command signatures (placeholders are not literal values): ```text preflight balance -status +status [--json] fund --amount [--quote ] receive [--token-file ] -list --give-mint --give --want-mint --want --max-fees +list --give-mint --give --want-mint --want --max-fees [--dry-run] discover serve cancel -take --max-give --min-receive --max-fees +take --max-give --min-receive --max-fees [--dry-run] withdraw --invoice [--max-debit ] recover ``` @@ -31,7 +31,8 @@ Use `-h` / `--help` at the root or on any application subcommand. The built-in `help [COMMAND]` also shows root or command help. There is no version option. `--home ` is required for operational commands; `--relay ` is repeatable. All command-specific flags in the signatures above are required except -`--quote`, `--max-debit`, `--token-file` (stdin when omitted), and `--max-fees` (default 16 on list/take). Positional mint arguments are +`--quote`, `--max-debit`, `--token-file` (stdin when omitted), `--dry-run`, `--json`, +and `--max-fees` (default 16 on list/take). Positional mint arguments are canonical URLs; lot arguments are public lot IDs; `--quote` is a saved funding quote for the same mint and amount. Never use a lab-feature binary for the human's funds. @@ -72,9 +73,45 @@ requires enabled NUT-05 bolt11/sat. The explicit `preflight ` checks the common trading capabilities. Diagnostic preflight output goes to stderr. No opt-in is required; passing advertisement checks is not independent refund interoperability evidence. `balance` and `status` are lock-free read-only SQLite snapshots, available during -`serve`, without network calls or submission of earlier authorizations. -Market commands (including `discover` and `cancel`) first recover existing -authorizations; do not treat them as read-only on a home with pending obligations. +`serve`, without network calls or submission of earlier authorizations. `status` +prints one readable line per lot/swap/funding/withdrawal/receive, unresolved items +first, any `counterparty_unresponsive` warning, and the exit-code meanings; +`status --json` prints the unchanged raw public records for scripts. + +`discover` is lock-free and read-only too: it reads relays only, with a throwaway +relay identity, never runs recovery, never takes the home lock, never creates or +touches the home or journal, and works while `serve` or `take` runs. Each listing +carries `expected_fees` per leg (`net`, `ppk`, `gross`, `claim_fee`) from the +mints' published keysets, or `"unknown"` when a mint did not answer within ~5 s +(discover never blocks on a mint). The taker's own lock fee depends on its inputs: +use `take --dry-run`. Relay union, EOSE and `timeout_or_partial` reporting are unchanged. + +## Fee preview: `--dry-run` + +`list ... --dry-run` and `take ... --dry-run` run the normal preflight, then print one +JSON object: `you_give` (keyset, ppk, net, gross, `lock_fee`, `claim_fee`, `debit`, +inputs that would be selected), `you_receive` (incoming leg's gross/claim fee, funded +by the counterparty), `max_total_debit` and `fees_total`. The numbers use the same +arithmetic and input selection as the real command against the current wallet and +the mints' published keysets. Nothing is journaled, reserved, locked, created or +published, and no quote request reaches the maker. Exit 0 when the real command +would pass the same bounds; exit 1 with the reason (e.g. `maximum give cap`, +`fee cap exceeded`, insufficient balance, or `cannot solve bounded fee-inclusive +split` for an amount with no exact fee-inclusive gross) otherwise. Run it before +asking for approval, then run the identical command without `--dry-run`. Balances +can change in between; the real command enforces your bounds again. + +## Hand-off to a running serve + +While `serve` owns the home, `list`, `take` and `cancel` hand their request to it +over `/serve.sock` (mode 0600, same user only) and print serve's result with +the usual output and exit codes. serve executes it with exactly the bounds given; +it never fills in defaults. Use the same `--relay` set as serve or the request is +refused. A handed-off `take` prints its swap id and that serve is watching it, then +streams states until settlement; if the command is interrupted, serve keeps driving +the swap (check `status`). Without a running serve these commands behave as before. +`recover` and the money commands still need the home lock; stop serve first. +`cancel` first recovers existing authorizations when it runs without serve. `serve` continuously accepts trades and recovers obligations; keep it running while locks are live. `recover` is a bounded single pass, not a replacement watcher; see [recovery](recovery.md) for exit codes and states. diff --git a/crates/maxplayer-trade/skills/maxplayer-trade/references/recovery.md b/crates/maxplayer-trade/skills/maxplayer-trade/references/recovery.md index d7255bd61..1201af3af 100644 --- a/crates/maxplayer-trade/skills/maxplayer-trade/references/recovery.md +++ b/crates/maxplayer-trade/skills/maxplayer-trade/references/recovery.md @@ -18,7 +18,19 @@ recovery indefinitely. An offline seller can miss its recovery opportunity; NUT- receiver claims remain valid after locktime. Keep the watcher alive. **Do not stop serve between `second_locked` and `settling`.** Use lock-free -`status` / `balance` for observation while the writer runs. +`status` / `balance` / `discover` for observation while the writer runs; hand new +`list`/`take`/`cancel` to it instead of stopping it. + +### `counterparty_unresponsive` + +When a swap has waited more than 180 seconds for its counterparty (taker in +`requested` awaiting a quote or in `first_locked` awaiting the maker's lock; maker in +`quoted` awaiting the taker's lock or in `second_locked` awaiting the claim), `take`'s +live output, serve's log and `status` show `counterparty_unresponsive` with the +elapsed time and, once our own lock exists, when its refund becomes available. It is +display only: no state, deadline, lock time (60/15 min) or refund rule changes. Keep +serve running; settlement or the timed refund proceeds as usual. Do not start another +take or stop serve because of it. ## Bounded pass and exit status From 0194bb5f2f6ba9bfbacd3d54170467cb04361c5f Mon Sep 17 00:00:00 2001 From: maxie-agent <330537927+maxie-agent@users.noreply.github.com> Date: Fri, 9 Oct 2026 23:54:42 +0000 Subject: [PATCH 3/5] fix(trade): identical lists in the same second no longer share a lot id The lot id hashes second-granular terms; two identical list requests handed to serve back-to-back produced the same id and the second journal row overwrote the first, orphaning its reservation. list now waits for a fresh id (or refuses). Found by the concurrent clients/serve race test, which now prints client and serve output on failure. --- crates/maxplayer-trade/src/coordinator.rs | 21 ++++++++++++++++++++- crates/maxplayer-trade/tests/ux.rs | 16 ++++++++++++++-- 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/crates/maxplayer-trade/src/coordinator.rs b/crates/maxplayer-trade/src/coordinator.rs index 698941951..eb3f85681 100644 --- a/crates/maxplayer-trade/src/coordinator.rs +++ b/crates/maxplayer-trade/src/coordinator.rs @@ -178,7 +178,26 @@ pub async fn list( ) -> Result { give.asset.fence()?; want.asset.fence()?; - let event = lot_event(&m.keys, give, want)?; + // A lot id is a hash of second-granular terms: two identical lists in the same second + // (back-to-back requests handed to serve) would share an id, and the second journal row + // would overwrite the first, orphaning its reservation. Wait for a fresh id instead. + let mut event = lot_event(&m.keys, give.clone(), want.clone())?; + for _ in 0..3 { + if j.get::("listing", &event.id.to_hex()) + .await? + .is_none() + { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(1100)).await; + event = lot_event(&m.keys, give.clone(), want.clone())?; + } + ensure!( + j.get::("listing", &event.id.to_hex()) + .await? + .is_none(), + "an identical lot was just listed; retry" + ); let lot = parse_lot(&event, now())?; preflight(&lot).await?; let plan = mint::plan(home, &lot.give.asset.mint_url, lot.give.net, max_fees).await?; diff --git a/crates/maxplayer-trade/tests/ux.rs b/crates/maxplayer-trade/tests/ux.rs index 614103077..746992d0c 100644 --- a/crates/maxplayer-trade/tests/ux.rs +++ b/crates/maxplayer-trade/tests/ux.rs @@ -540,6 +540,7 @@ async fn concurrent_clients_and_serve_never_make_two_writers() { "8", ]; let mut ok = 0; + let mut outputs = String::new(); let mut killed = std::collections::BTreeMap::new(); for round in 0..3 { let log = f.root.path().join(format!("serve{round}.log")); @@ -552,7 +553,11 @@ async fn concurrent_clients_and_serve_never_make_two_writers() { let c2 = cmd(&f.maker, &f.relay_url, &list, &env).output(); let (o1, o2) = tokio::join!(c1, c2); for o in [o1.unwrap(), o2.unwrap()] { - let (_, stderr) = text(&o); + let (stdout, stderr) = text(&o); + outputs.push_str(&format!( + "round {round} exit {:?}\nstdout: {stdout}\nstderr: {stderr}\n", + o.status.code() + )); if o.status.success() { ok += 1; } else { @@ -610,7 +615,14 @@ async fn concurrent_clients_and_serve_never_make_two_writers() { assert_eq!( listings.len(), ok, - "SAFETY: success count equals journaled listings" + "SAFETY: success count equals journaled listings\n{outputs}\n{}", + (0..3) + .map( + |r| std::fs::read_to_string(f.root.path().join(format!("serve{r}.log"))) + .unwrap_or_default() + ) + .collect::>() + .join("\n---\n") ); let mut ys = BTreeSet::new(); for l in &listings { From a3e20db5a0c6c2c807a6ad40da493ae9a103217c Mon Sep 17 00:00:00 2001 From: maxie-agent <330537927+maxie-agent@users.noreply.github.com> Date: Fri, 9 Oct 2026 23:57:00 +0000 Subject: [PATCH 4/5] test(trade): status --json golden follows #1119's receive summary; SKILL under 10 kB --- crates/maxplayer-trade/skills/maxplayer-trade/SKILL.md | 2 +- crates/maxplayer-trade/tests/ux.rs | 6 ++++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/maxplayer-trade/skills/maxplayer-trade/SKILL.md b/crates/maxplayer-trade/skills/maxplayer-trade/SKILL.md index 59c706594..ce23ea77b 100644 --- a/crates/maxplayer-trade/skills/maxplayer-trade/SKILL.md +++ b/crates/maxplayer-trade/skills/maxplayer-trade/SKILL.md @@ -104,7 +104,7 @@ inactive cancellation is reported; an unresolved lock is never called cancelled. ### Find listings / buy that lot `discover` (per-lot fees or `unknown`), inspect the mint pair -and lot ID, and match the human's price. Discovery authorizes no purchase. For “X A for at most Y B”, +and lot ID, and match the human's price. Discovery approves nothing. For “X A for at most Y B”, set `--min-receive X` and `--max-give Y` when Y is the total spending ceiling. If Y was explicitly a net price plus a separately approved fee F, the total ceiling can be Y + F; otherwise never silently add fees. A stricter limit is fine, a looser one diff --git a/crates/maxplayer-trade/tests/ux.rs b/crates/maxplayer-trade/tests/ux.rs index 746992d0c..94c1249d1 100644 --- a/crates/maxplayer-trade/tests/ux.rs +++ b/crates/maxplayer-trade/tests/ux.rs @@ -905,6 +905,12 @@ fn status_golden(home: &Path) -> String { serde_json::from_slice::(&bytes) .unwrap() .summary() + } else if kind == "receive" { + let mut s = serde_json::from_slice::(&bytes) + .unwrap() + .summary(); + s["kind"] = kind.clone().into(); + s } else { serde_json::json!({"kind":kind,"id":v["id"],"state":v["state"],"done":v["done"],"expired_unpaid":v["expired_unpaid"]}) }; From 81e6ea6acbfcd2674a10d6e8433b9c8ac9366638 Mon Sep 17 00:00:00 2001 From: maxie-agent <330537927+maxie-agent@users.noreply.github.com> Date: Sat, 10 Oct 2026 00:01:55 +0000 Subject: [PATCH 5/5] test(trade): #1119's receive status assertion reads status --json --- crates/maxplayer-trade/tests/receive.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/maxplayer-trade/tests/receive.rs b/crates/maxplayer-trade/tests/receive.rs index ace26babe..0623c5e6f 100644 --- a/crates/maxplayer-trade/tests/receive.rs +++ b/crates/maxplayer-trade/tests/receive.rs @@ -716,9 +716,9 @@ async fn cli_exit_codes_for_journaled_refused_and_quarantined() { assert_eq!(state_of(&out)["manual_recovery"], true); assert_secret_free(&out, &token2, &proofs2); m.faults.omit_dleq.store(false, SeqCst); - // status: same public fields as recover for receives, never the token. + // status --json: same public fields as recover for receives, never the token. let status = std::process::Command::new(env!("CARGO_BIN_EXE_maxplayer-trade")) - .args(["--home", h.path().to_str().unwrap(), "status"]) + .args(["--home", h.path().to_str().unwrap(), "status", "--json"]) .output() .unwrap(); assert_secret_free(&status, &token2, &proofs2);