From 00b7e1798ff30cd50da4509eb57b0c80f2e5e270 Mon Sep 17 00:00:00 2001 From: archipelago Date: Sat, 25 Jul 2026 13:12:19 -0400 Subject: [PATCH] feat(lnd): closed-channels RPC, closing channels in list, streaming close with txid - lnd.closedchannels: closed-channel history via /v1/channels/closed - channel list now includes waiting-close and force-closing pending channels with their closing_txid - closechannel reads the close stream's first update with a dedicated client instead of hanging until the closing tx confirms; returns the closing txid in display byte order Co-Authored-By: Claude Fable 5 --- core/archipelago/src/api/rpc/dispatcher.rs | 1 + core/archipelago/src/api/rpc/lnd/channels.rs | 219 ++++++++++++++++--- 2 files changed, 185 insertions(+), 35 deletions(-) diff --git a/core/archipelago/src/api/rpc/dispatcher.rs b/core/archipelago/src/api/rpc/dispatcher.rs index 871949cd..7a244db3 100644 --- a/core/archipelago/src/api/rpc/dispatcher.rs +++ b/core/archipelago/src/api/rpc/dispatcher.rs @@ -124,6 +124,7 @@ impl RpcHandler { } "lnd.getinfo" => self.handle_lnd_getinfo().await, "lnd.listchannels" => self.handle_lnd_listchannels().await, + "lnd.closedchannels" => self.handle_lnd_closedchannels().await, "lnd.openchannel" => self.handle_lnd_openchannel(params).await, "lnd.closechannel" => self.handle_lnd_closechannel(params).await, "lnd.newaddress" => self.handle_lnd_newaddress().await, diff --git a/core/archipelago/src/api/rpc/lnd/channels.rs b/core/archipelago/src/api/rpc/lnd/channels.rs index 62271591..2259d7fd 100644 --- a/core/archipelago/src/api/rpc/lnd/channels.rs +++ b/core/archipelago/src/api/rpc/lnd/channels.rs @@ -30,6 +30,8 @@ struct ChannelInfo { active: bool, status: String, channel_point: String, + #[serde(skip_serializing_if = "String::is_empty")] + closing_txid: String, } #[derive(Debug, Serialize)] @@ -58,6 +60,10 @@ struct LndChannel { #[derive(Debug, Deserialize, Default)] struct LndPendingChannelsResponse { pending_open_channels: Option>, + // Cooperative closes waiting for their closing tx to confirm + waiting_close_channels: Option>, + // Force closes serving out their timelock + pending_force_closing_channels: Option>, } #[derive(Debug, Deserialize)] @@ -65,6 +71,18 @@ struct LndPendingOpenChannel { channel: Option, } +#[derive(Debug, Deserialize)] +struct LndWaitingCloseChannel { + channel: Option, + closing_txid: Option, +} + +#[derive(Debug, Deserialize)] +struct LndForceClosingChannel { + channel: Option, + closing_txid: Option, +} + #[derive(Debug, Deserialize)] struct LndPendingChannel { remote_node_pub: Option, @@ -74,6 +92,52 @@ struct LndPendingChannel { channel_point: Option, } +impl LndPendingChannel { + fn into_channel_info(self, status: &str, closing_txid: Option) -> ChannelInfo { + let parse = |s: &Option| s.as_deref().and_then(|v| v.parse().ok()).unwrap_or(0); + ChannelInfo { + chan_id: String::new(), + remote_pubkey: self.remote_node_pub.clone().unwrap_or_default(), + capacity: parse(&self.capacity), + local_balance: parse(&self.local_balance), + remote_balance: parse(&self.remote_balance), + active: false, + status: status.into(), + channel_point: self.channel_point.unwrap_or_default(), + closing_txid: closing_txid.unwrap_or_default(), + } + } +} + +#[derive(Debug, Deserialize, Default)] +struct LndClosedChannelsResponse { + channels: Option>, +} + +#[derive(Debug, Deserialize)] +struct LndClosedChannel { + chan_id: Option, + remote_pubkey: Option, + capacity: Option, + settled_balance: Option, + close_type: Option, + closing_tx_hash: Option, + channel_point: Option, + close_height: Option, +} + +#[derive(Debug, Serialize)] +struct ClosedChannelInfo { + chan_id: String, + remote_pubkey: String, + capacity: i64, + settled_balance: i64, + close_type: String, + closing_tx_hash: String, + channel_point: String, + close_height: i64, +} + impl RpcHandler { pub(in crate::api::rpc) async fn handle_lnd_listchannels(&self) -> Result { let (client, macaroon_hex) = self.lnd_client().await?; @@ -131,6 +195,7 @@ impl RpcHandler { "inactive".into() }, channel_point: ch.channel_point.unwrap_or_default(), + closing_txid: String::new(), } }) .collect(); @@ -138,31 +203,17 @@ impl RpcHandler { let mut pending_channels: Vec = Vec::new(); for pch in pending_resp.pending_open_channels.unwrap_or_default() { if let Some(ch) = pch.channel { - let capacity: i64 = ch - .capacity - .as_deref() - .and_then(|s| s.parse().ok()) - .unwrap_or(0); - let local: i64 = ch - .local_balance - .as_deref() - .and_then(|s| s.parse().ok()) - .unwrap_or(0); - let remote: i64 = ch - .remote_balance - .as_deref() - .and_then(|s| s.parse().ok()) - .unwrap_or(0); - pending_channels.push(ChannelInfo { - chan_id: String::new(), - remote_pubkey: ch.remote_node_pub.unwrap_or_default(), - capacity, - local_balance: local, - remote_balance: remote, - active: false, - status: "pending_open".into(), - channel_point: ch.channel_point.unwrap_or_default(), - }); + pending_channels.push(ch.into_channel_info("pending_open", None)); + } + } + for wch in pending_resp.waiting_close_channels.unwrap_or_default() { + if let Some(ch) = wch.channel { + pending_channels.push(ch.into_channel_info("closing", wch.closing_txid)); + } + } + for fch in pending_resp.pending_force_closing_channels.unwrap_or_default() { + if let Some(ch) = fch.channel { + pending_channels.push(ch.into_channel_info("force_closing", fch.closing_txid)); } } @@ -349,6 +400,46 @@ impl RpcHandler { Ok(body) } + pub(in crate::api::rpc) async fn handle_lnd_closedchannels(&self) -> Result { + let (client, macaroon_hex) = self.lnd_client().await?; + + let resp: LndClosedChannelsResponse = client + .get(format!("{LND_REST_BASE_URL}/v1/channels/closed")) + .header("Grpc-Metadata-macaroon", &macaroon_hex) + .send() + .await + .context("LND REST connection failed")? + .json() + .await + .context("Failed to parse LND closed channels response")?; + + let channels: Vec = resp + .channels + .unwrap_or_default() + .into_iter() + .map(|ch| ClosedChannelInfo { + chan_id: ch.chan_id.unwrap_or_default(), + remote_pubkey: ch.remote_pubkey.unwrap_or_default(), + capacity: ch + .capacity + .as_deref() + .and_then(|s| s.parse().ok()) + .unwrap_or(0), + settled_balance: ch + .settled_balance + .as_deref() + .and_then(|s| s.parse().ok()) + .unwrap_or(0), + close_type: ch.close_type.unwrap_or_default(), + closing_tx_hash: ch.closing_tx_hash.unwrap_or_default(), + channel_point: ch.channel_point.unwrap_or_default(), + close_height: ch.close_height.unwrap_or(0), + }) + .collect(); + + Ok(serde_json::json!({ "channels": channels })) + } + pub(in crate::api::rpc) async fn handle_lnd_closechannel( &self, params: Option, @@ -389,27 +480,35 @@ impl RpcHandler { "Closing Lightning channel" ); - let (client, macaroon_hex) = self.lnd_client().await?; + let (_, macaroon_hex) = self.lnd_client().await?; + + // The close endpoint is server-streaming: LND holds the connection + // open and emits updates until the closing tx CONFIRMS on-chain + // (potentially hours). Reading the whole body hangs the RPC even + // though the close already went through, and the shared lnd_client's + // 15s total timeout would abort the stream mid-read. Use a dedicated + // client and return as soon as the first streamed update arrives. + let client = reqwest::Client::builder() + .no_proxy() + .connect_timeout(std::time::Duration::from_secs(10)) + .danger_accept_invalid_certs(true) + .build() + .context("Failed to create streaming HTTP client")?; let url = format!( "{LND_REST_BASE_URL}/v1/channels/{}/{}?force={}", parts[0], parts[1], force ); - let resp = client + let mut resp = client .delete(&url) .header("Grpc-Metadata-macaroon", &macaroon_hex) .send() .await .context("Failed to close channel")?; - let status = resp.status(); - let body: serde_json::Value = resp - .json() - .await - .context("Failed to parse close channel response")?; - - if !status.is_success() { + if !resp.status().is_success() { + let body: serde_json::Value = resp.json().await.unwrap_or_default(); let msg = body .get("message") .and_then(|v| v.as_str()) @@ -417,6 +516,56 @@ impl RpcHandler { return Err(anyhow::anyhow!("Failed to close channel: {}", msg)); } - Ok(serde_json::json!({ "success": true })) + // First streamed line is {"result":{"close_pending":…}} on success or + // {"error":…} — the stream reports errors in-band after a 200. + let mut buf: Vec = Vec::new(); + let first_update = tokio::time::timeout(std::time::Duration::from_secs(25), async { + while let Some(chunk) = resp.chunk().await? { + buf.extend_from_slice(&chunk); + let line = match buf.iter().position(|&b| b == b'\n') { + Some(pos) => &buf[..pos], + None => &buf[..], + }; + if let Ok(v) = serde_json::from_slice::(line) { + return Ok::<_, anyhow::Error>(Some(v)); + } + } + Ok(None) + }) + .await; + + match first_update { + Ok(Ok(Some(update))) => { + if let Some(err) = update.get("error") { + let msg = err + .get("message") + .and_then(|v| v.as_str()) + .unwrap_or("Unknown error"); + return Err(anyhow::anyhow!("Failed to close channel: {}", msg)); + } + // txid arrives base64-encoded in internal byte order; flip it + // into the display order explorers use. + use base64::Engine as _; + let closing_txid = update + .pointer("/result/close_pending/txid") + .and_then(|v| v.as_str()) + .and_then(|b64| base64::engine::general_purpose::STANDARD.decode(b64).ok()) + .map(|mut bytes| { + bytes.reverse(); + hex::encode(bytes) + }) + .unwrap_or_default(); + info!(channel_point, closing_txid, "Channel close initiated"); + Ok(serde_json::json!({ "success": true, "closing_txid": closing_txid })) + } + Ok(Ok(None)) => Err(anyhow::anyhow!( + "LND ended the close stream without an update — check the channel list" + )), + Ok(Err(e)) => Err(e).context("Failed reading close channel response"), + // No update inside the window: the close is almost certainly still + // negotiating with the peer — report initiated, the channel list + // will show it under Closing. + Err(_) => Ok(serde_json::json!({ "success": true, "closing_txid": "" })), + } } }