lfg2025/archy
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases 11 Wiki Activity

Merge current main and make purchase filing atomic under concurrent writes

Browse Source
This commit is contained in:
archipelago
2026-09-30 07:26:51 -04:00
parent 33477f284b f12042f194
commit 0677924a64
55 changed files with 3089 additions and 493 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/api/rpc/content.rs
+103 -96
View File
@@ -22,9 +22,9 @@ const FILE_CATALOG_PROTOCOL: &str = "https://archipelago.dev/protocols/file-cata
/// Best-effort reclaim of an ecash payment token that was minted but the sale
/// didn't complete (seller unreachable or couldn't redeem it), so the buyer
/// doesn't lose the value. For Fedimint the spender can reissue its own
/// un-redeemed notes; for Cashu the proofs are received back. Fails silently if
/// the seller already claimed the token (then the value is genuinely gone).
async fn reclaim_spent_ecash(data_dir: &std::path::Path, token: &str, backend: &str) {
/// un-redeemed notes; for Cashu the proofs are received back. Report the actual
/// recovered amount, or explicitly say when a refund could not be confirmed.
async fn reclaim_spent_ecash(data_dir: &std::path::Path, token: &str, backend: &str) -> String {
let res = match backend {
"fedimint" => crate::wallet::fedimint_client::reissue_into_any(data_dir, token)
.await
@@ -32,16 +32,62 @@ async fn reclaim_spent_ecash(data_dir: &std::path::Path, token: &str, backend: &
_ => ecash::receive_token(data_dir, token).await,
};
match res {
Ok(sats) => tracing::info!(
"paid download: reclaimed {sats} sats of unspent {backend} ecash after a failed sale"
),
Err(e) => tracing::warn!(
"paid download: could not reclaim {backend} ecash (the peer may have already \
claimed it): {e:#}"
),
Ok(sats) => {
tracing::info!("paid download: reclaimed {sats} sats after failed sale");
format!("Refunded {sats} sats to your wallet.")
}
Err(e) => {
tracing::warn!("paid download: refund not confirmed: {e}");
"Your refund could not be confirmed. The seller may have received the payment. Do not pay again until this is checked.".to_string()
}
}
}
/// Keep first purchases and cached repeats compatible with both existing clients.
fn paid_content_response(bytes: &[u8], mime: &str, paid_sats: u64) -> serde_json::Value {
use base64::Engine;
let data = base64::engine::general_purpose::STANDARD.encode(bytes);
serde_json::json!({
"data": data, "data_base64": data,
"size": bytes.len(), "size_bytes": bytes.len(),
"mime_type": mime, "paid_sats": paid_sats, "owned": true,
})
}
/// File purchases through an atomic no-clobber write in Files' own namespace.
async fn file_purchase_in_files(
data_dir: &std::path::Path,
filename: &str,
mime: &str,
bytes: &[u8],
) -> Result<String> {
let folder = if mime.starts_with("image/") || mime.starts_with("video/") {
"Photos"
} else if mime.starts_with("audio/") {
"Music"
} else {
"Documents"
};
let root = data_dir.join("filebrowser");
anyhow::ensure!(
tokio::fs::metadata(&root).await?.is_dir(),
"Files storage is unavailable"
);
let name = std::path::Path::new(filename)
.file_name()
.and_then(|n| n.to_str())
.filter(|n| !n.is_empty())
.unwrap_or("download");
let path =
crate::container::filebrowser::save_new_file(&root.join(folder), name, bytes).await?;
Ok(format!(
"{folder}/{}",
path.file_name()
.and_then(|n| n.to_str())
.context("Invalid Files name")?
))
}
impl RpcHandler {
/// List content I'm sharing.
pub(super) async fn handle_content_list_mine(&self) -> Result<serde_json::Value> {
@@ -463,17 +509,10 @@ impl RpcHandler {
crate::content_owned::read_owned(&self.config.data_dir, &o.onion, &o.content_id)
.await
{
use base64::Engine;
return Ok(serde_json::json!({
"owned": true,
"already_owned": true,
"filename": o.filename,
"mime_type": mime,
"size_bytes": bytes.len(),
"paid_sats": 0,
"data_base64":
base64::engine::general_purpose::STANDARD.encode(&bytes),
}));
let mut result = paid_content_response(&bytes, &mime, 0);
result["already_owned"] = serde_json::json!(true);
result["filename"] = serde_json::json!(o.filename);
return Ok(result);
}
// Cache record exists but bytes are gone — fall through and
// repurchase rather than stranding the user.
@@ -547,29 +586,27 @@ impl RpcHandler {
// Surface a real reason instead of the generic sanitized error (#30):
// the dial already tries FIPS/mesh then falls back to Tor, so a failure
// here means the peer is genuinely unreachable on both transports.
let (response, transport) = match crate::fips::dial::PeerRequest::new(
fips_npub.as_deref(),
onion,
&path,
)
.service(crate::settings::transport::PeerService::PeerFiles)
.header("X-Federation-DID", local_did)
.header("X-Payment-Token", token_str.clone())
.timeout(std::time::Duration::from_secs(900))
.send_get()
.await
{
Ok(v) => v,
Err(e) => {
tracing::warn!("paid peer download dial failed for {}: {:#}", onion, e);
// The token was already minted/spent — reclaim it so the buyer
// doesn't lose the value when the seller was simply unreachable.
reclaim_spent_ecash(&self.config.data_dir, &token_str, used_backend).await;
return Ok(serde_json::json!({
"error": "Could not reach the peer over mesh or Tor — it may be offline. Your ecash was refunded to your wallet. Please try again."
}));
}
};
let (response, transport) =
match crate::fips::dial::PeerRequest::new(fips_npub.as_deref(), onion, &path)
.service(crate::settings::transport::PeerService::PeerFiles)
.header("X-Federation-DID", local_did)
.header("X-Payment-Token", token_str.clone())
.timeout(std::time::Duration::from_secs(900))
.send_get()
.await
{
Ok(v) => v,
Err(e) => {
tracing::warn!("paid peer download dial failed for {}: {:#}", onion, e);
// The token was already minted/spent — reclaim it so the buyer
// doesn't lose the value when the seller was simply unreachable.
let refund =
reclaim_spent_ecash(&self.config.data_dir, &token_str, used_backend).await;
return Ok(serde_json::json!({
"error": format!("Could not reach the peer over mesh or Tor. {refund}")
}));
}
};
// Record which transport actually reached the peer (B14).
if let Err(e) = crate::federation::record_peer_transport(
&self.config.data_dir,
@@ -583,25 +620,17 @@ impl RpcHandler {
}
if response.status() == reqwest::StatusCode::PAYMENT_REQUIRED {
// Payment was rejected by the seller. Surface the most likely cause
// per backend — for ecash both sides must share a redemption network
// (a Cashu mint, or a Fedimint federation).
// A 402 can mean mint validation, network failure, underpayment,
// or an unaccepted mint. Do not invent a mint-mismatch diagnosis.
let body = response.text().await.unwrap_or_default();
tracing::warn!(
"paid download: seller {onion} rejected {used_backend} payment of {price_sats} sats: {body}"
);
// Seller couldn't redeem the token — reclaim it so the buyer keeps
// their funds (the spent-but-unredeemed-notes case the user hit).
reclaim_spent_ecash(&self.config.data_dir, &token_str, used_backend).await;
let hint = match used_backend {
"fedimint" => "the seller isn't in the same Fedimint federation as you",
_ => "the seller doesn't accept your Cashu mint",
};
let refund = reclaim_spent_ecash(&self.config.data_dir, &token_str, used_backend).await;
return Ok(serde_json::json!({
"error": format!(
"Payment rejected by the seller — {hint}. Your ecash was refunded to \
your wallet. Try the other ecash type, or use a shared mint/federation."
)
"error": format!("The seller could not verify the payment. {refund}")
}));
}
@@ -609,9 +638,9 @@ impl RpcHandler {
let status = response.status();
let body = response.text().await.unwrap_or_default();
tracing::warn!("paid download: seller {onion} returned {status}: {body}");
reclaim_spent_ecash(&self.config.data_dir, &token_str, used_backend).await;
let refund = reclaim_spent_ecash(&self.config.data_dir, &token_str, used_backend).await;
return Ok(serde_json::json!({
"error": format!("Peer returned an error ({status}). Your ecash was refunded to your wallet.")
"error": format!("Peer returned an error ({status}). {refund}")
}));
}
@@ -658,47 +687,21 @@ impl RpcHandler {
tracing::warn!("paid download: failed to cache purchased content (non-fatal): {e:#}");
}
// Auto-file the purchase into the user's Files area (2026-07-22):
// Photos for images/video, Music for audio, Documents otherwise —
// same buckets the Cloud view uses. The in-app viewer still plays
// from the purchase cache; this makes the file ALSO show up where
// files live, on every device, without relying on a browser
// download. Best-effort: never fail a paid download over it.
{
let folder = if mime_type.starts_with("image/") || mime_type.starts_with("video/") {
"Photos"
} else if mime_type.starts_with("audio/") {
"Music"
} else {
"Documents"
};
let base = std::path::Path::new(&filename)
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("download")
.to_string();
let dir = self.config.data_dir.join("filebrowser").join(folder);
match crate::container::filebrowser::save_new_file(&dir, &base, &bytes).await {
Ok(path) => tracing::info!("paid download: filed into {}", path.display()),
Err(e) => tracing::warn!(
"paid download: filing into {} failed (non-fatal): {e:#}",
dir.display()
),
}
// The durable purchased-content cache above is primary. A Files copy
// remains optional: a stopped FileBrowser must not undo a paid download.
let filed =
file_purchase_in_files(&self.config.data_dir, &filename, &mime_type, &bytes).await;
match filed {
Ok(path) => tracing::info!("paid download: filed into Files/{path}"),
Err(error) => tracing::warn!(
"paid download: optional Files copy failed; purchase cache retained: {error}"
),
}
use base64::Engine;
let encoded = base64::engine::general_purpose::STANDARD.encode(&bytes);
tracing::info!("paid download: received {} bytes from {onion} (paid {price_sats} sats via {used_backend})", bytes.len());
Ok(serde_json::json!({
"data": encoded,
"size": bytes.len(),
"paid_sats": price_sats,
"ecash_backend": used_backend,
"mime_type": mime_type,
"owned": true,
}))
let mut result = paid_content_response(&bytes, &mime_type, price_sats);
result["ecash_backend"] = serde_json::json!(used_backend);
Ok(result)
}
/// Buyer side (#46): ask the selling node to mint a Lightning invoice for a
@@ -1371,3 +1374,7 @@ impl RpcHandler {
}
}
}
#[cfg(test)]
#[path = "content_tests.rs"]
mod tests;
core/archipelago/src/api/rpc/content_tests.rs
+56
View File
@@ -0,0 +1,56 @@
use super::*;
#[test]
fn first_and_cached_paid_downloads_have_the_same_client_payload_contract() {
use base64::Engine;
for paid in [0, 1] {
let response = paid_content_response(&[0, 255, 123], "application/octet-stream", paid);
assert_eq!(response["data"], response["data_base64"]);
assert_eq!(
base64::engine::general_purpose::STANDARD
.decode(response["data"].as_str().unwrap())
.unwrap(),
[0, 255, 123]
);
assert_eq!(response["size"], 3);
assert_eq!(response["size_bytes"], 3);
assert_eq!(response["paid_sats"], paid);
assert_eq!(response["owned"], true);
}
}
#[tokio::test]
async fn files_copy_routes_media_and_sanitizes_the_filename() {
let dir = tempfile::tempdir().unwrap();
tokio::fs::create_dir(dir.path().join("filebrowser"))
.await
.unwrap();
for (mime, folder) in [
("image/png", "Photos"),
("video/mp4", "Photos"),
("audio/mpeg", "Music"),
("text/plain", "Documents"),
] {
let relative = file_purchase_in_files(dir.path(), "../name #?.bin", mime, b"paid")
.await
.unwrap();
assert!(relative.starts_with(&format!("{folder}/name #?")));
assert_eq!(
tokio::fs::read(dir.path().join("filebrowser").join(relative))
.await
.unwrap(),
b"paid"
);
}
}
#[tokio::test]
async fn unavailable_files_storage_is_reported_without_creating_a_fake_installation() {
let dir = tempfile::tempdir().unwrap();
assert!(
file_purchase_in_files(dir.path(), "name", "text/plain", b"bytes")
.await
.is_err()
);
assert!(!dir.path().join("filebrowser").exists());
}
core/archipelago/src/api/rpc/lnd/info.rs
+84
View File
@@ -109,7 +109,50 @@ fn checked_balances(
))
}
fn bitcoin_wait_state(
installed: bool,
running: bool,
fresh: bool,
ibd: Option<bool>,
) -> (&'static str, &'static str) {
if !installed {
("waiting_install", "Waiting for Bitcoin to be installed")
} else if !running {
("waiting_start", "Waiting for Bitcoin to start")
} else if !fresh || ibd.is_none() {
("waiting_start", "Waiting for Bitcoin to start")
} else if ibd == Some(true) {
("waiting_sync", "Waiting for Bitcoin to sync")
} else {
("bitcoin_ready", "Bitcoin is ready")
}
}
impl RpcHandler {
pub(crate) async fn handle_lnd_readiness(&self) -> serde_json::Value {
let (data, _) = self.state_manager.get_snapshot().await;
if !data.server_info.status_info.containers_scanned {
return serde_json::json!({"state":"checking", "message":"Checking Bitcoin availability"});
}
let nodes: Vec<_> = ["bitcoin-core", "bitcoin-knots", "bitcoin"]
.iter()
.filter_map(|id| data.package_data.get(*id))
.collect();
let installed = !nodes.is_empty();
let running = nodes
.iter()
.any(|p| p.state == crate::data_model::PackageState::Running);
let bitcoin = crate::bitcoin_status::get_bitcoin_status().await;
let ibd = bitcoin
.blockchain_info
.as_ref()
.and_then(|v| v.get("initialblockdownload"))
.and_then(|v| v.as_bool());
let (state, message) =
bitcoin_wait_state(installed, running, bitcoin.ok && !bitcoin.stale, ibd);
serde_json::json!({"state": state, "message": message})
}
pub(in crate::api::rpc) async fn handle_lnd_getinfo(&self) -> Result<serde_json::Value> {
let macaroon_bytes = read_lnd_admin_macaroon().await?;
let macaroon_hex = hex::encode(&macaroon_bytes);
@@ -419,3 +462,44 @@ mod tests {
assert!(!is_valid_identity_pubkey(&"g".repeat(66)));
}
}
#[cfg(test)]
mod dependency_readiness_tests {
use super::bitcoin_wait_state;
#[test]
fn waiting_states_cover_install_start_sync_outage_and_recovery() {
assert_eq!(
bitcoin_wait_state(false, false, false, None).0,
"waiting_install"
);
assert_eq!(
bitcoin_wait_state(true, false, false, None).0,
"waiting_start"
);
assert_eq!(
bitcoin_wait_state(true, true, false, None).0,
"waiting_start"
);
assert_eq!(
bitcoin_wait_state(true, true, true, Some(true)).0,
"waiting_sync"
);
assert_eq!(
bitcoin_wait_state(true, true, true, Some(false)).0,
"bitcoin_ready"
);
// Previously synced cached information must not hide a current outage.
assert_eq!(
bitcoin_wait_state(true, true, false, Some(false)).0,
"waiting_start"
);
assert_eq!(
bitcoin_wait_state(true, true, true, None).0,
"waiting_start"
);
assert_eq!(
bitcoin_wait_state(true, true, true, Some(false)).0,
"bitcoin_ready"
);
}
}
core/archipelago/src/api/rpc/lnd/mod.rs
+62 -1
View File
@@ -133,12 +133,36 @@ async fn stream_lnd_transactions(sm: &crate::state::StateManager) -> Result<()>
/// RPC-unreachable and locked-wallet states are deliberately NOT handled
/// here — container-down is crash-recovery's job, and unlocking needs the
/// operator.
fn bitcoin_ready_for_lnd_watchdog(status: &crate::bitcoin_status::BitcoinNodeStatus) -> bool {
status.ok
&& !status.stale
&& status.age_ms < 30_000
&& status
.blockchain_info
.as_ref()
.and_then(|v| v.get("initialblockdownload"))
.and_then(|v| v.as_bool())
== Some(false)
}
pub(crate) fn spawn_lnd_health_watchdog() {
tokio::spawn(async move {
let mut bad_minutes: u32 = 0;
let mut last_restart: Option<tokio::time::Instant> = None;
let mut last_height: Option<u64> = None;
loop {
tokio::time::sleep(std::time::Duration::from_secs(60)).await;
// Initial Bitcoin sync, warmup, and outages are dependencies to
// wait for, never evidence that LND is wedged. Do not accumulate
// restart pressure during a days-long initial block download.
let bitcoin = crate::bitcoin_status::get_bitcoin_status().await;
if !bitcoin_ready_for_lnd_watchdog(&bitcoin)
|| crate::app_ops::lifecycle_op_in_flight("lnd")
{
bad_minutes = 0;
last_height = None;
continue;
}
let Ok(bytes) = read_lnd_admin_macaroon().await else {
bad_minutes = 0; // no LND on this node (or not set up yet)
continue;
@@ -161,6 +185,10 @@ pub(crate) fn spawn_lnd_health_watchdog() {
bad_minutes = 0; // down/locked — not the wedge signature
continue;
};
if !resp.status().is_success() {
bad_minutes = 0;
continue;
}
let Ok(info) = resp.json::<serde_json::Value>().await else {
bad_minutes = 0;
continue;
@@ -182,7 +210,12 @@ pub(crate) fn spawn_lnd_health_watchdog() {
.get("num_pending_channels")
.and_then(|v| v.as_u64())
.unwrap_or(0);
let wedged = !synced || (channels > 0 && peers == 0);
let height = info.get("block_height").and_then(|v| v.as_u64());
let progressing = height
.zip(last_height)
.is_some_and(|(now, before)| now > before);
last_height = height;
let wedged = !progressing && (!synced || (channels > 0 && peers == 0));
if !wedged {
bad_minutes = 0;
continue;
@@ -239,3 +272,31 @@ impl RpcHandler {
Ok((client, macaroon_hex))
}
}
#[cfg(test)]
mod watchdog_dependency_tests {
use super::bitcoin_ready_for_lnd_watchdog;
use crate::bitcoin_status::BitcoinNodeStatus;
use serde_json::json;
#[test]
fn initial_sync_warmup_outage_stale_and_unknown_never_trigger_lnd_restart() {
let mut status = BitcoinNodeStatus::default();
assert!(!bitcoin_ready_for_lnd_watchdog(&status));
status.ok = true;
status.blockchain_info = Some(json!({"initialblockdownload":true}));
assert!(!bitcoin_ready_for_lnd_watchdog(&status));
status.blockchain_info = Some(json!({"initialblockdownload":false}));
assert!(bitcoin_ready_for_lnd_watchdog(&status));
status.stale = true;
assert!(!bitcoin_ready_for_lnd_watchdog(&status));
status.stale = false;
status.ok = false;
assert!(!bitcoin_ready_for_lnd_watchdog(&status));
status.ok = true;
status.age_ms = 30_000;
assert!(!bitcoin_ready_for_lnd_watchdog(&status));
status.age_ms = 0;
status.blockchain_info = Some(json!({}));
assert!(!bitcoin_ready_for_lnd_watchdog(&status));
}
}
core/archipelago/src/api/rpc/package/install.rs
+4
View File
@@ -326,6 +326,10 @@ impl RpcHandler {
// an older version pins it so install_fresh resolves that image and the
// update badge stays suppressed. See docs/bitcoin-multi-version-design.md.
if matches!(package_id, "bitcoin-core" | "bitcoin-knots") {
if let Some(value) = params.get("prune") {
let prune = value.as_bool().context("prune must be a boolean")?;
crate::settings::bitcoin_storage::save(&self.config.data_dir, prune).await?;
}
if let Some(version) = params.get("version").and_then(|v| v.as_str()) {
persist_install_version_selection(package_id, version).await;
}
core/archipelago/src/api/rpc/package/set_config.rs
+10
View File
@@ -153,8 +153,18 @@ impl RpcHandler {
let default = app_catalog::catalog_default_version(app_id);
let cfg = version_config::read(app_id);
let installed = installed_version(app_id).await;
let bitcoin_prune = if matches!(app_id, "bitcoin-core" | "bitcoin-knots") {
Some(
crate::settings::bitcoin_storage::load(&self.config.data_dir)
.await?
.prune,
)
} else {
None
};
Ok(serde_json::json!({
"bitcoinPrune": bitcoin_prune,
"id": app_id,
"supportsVersions": supports_versions(app_id),
"default": default,