Add --wait to open-channel in the CLI

e2e-tests/src/lib.rs

@@ -749,10 +749,11 @@ impl Drop for McpHandle {
 	}
 }
 
-/// Run a CLI command against the given server handle and return raw stdout as a string.
-pub fn run_cli_raw(handle: &LdkServerHandle, args: &[&str]) -> String {
+/// Run a CLI command against the given server handle and return its output, whatever the exit
+/// status.
+pub fn run_cli_output(handle: &LdkServerHandle, args: &[&str]) -> std::process::Output {
 	let cli_path = cli_binary_path();
-	let output = Command::new(&cli_path)
+	Command::new(&cli_path)
 		.arg("--base-url")
 		.arg(handle.base_url())
 		.arg("--macaroon")
@@ -761,7 +762,12 @@ pub fn run_cli_raw(handle: &LdkServerHandle, args: &[&str]) -> String {
 		.arg(handle.tls_cert_path.to_str().unwrap())
 		.args(args)
 		.output()
-		.unwrap_or_else(|e| panic!("Failed to run CLI at {:?}: {}", cli_path, e));
+		.unwrap_or_else(|e| panic!("Failed to run CLI at {:?}: {}", cli_path, e))
+}
+
+/// Run a CLI command against the given server handle and return raw stdout as a string.
+pub fn run_cli_raw(handle: &LdkServerHandle, args: &[&str]) -> String {
+	let output = run_cli_output(handle, args);
 
 	if !output.status.success() {
 		let stderr = String::from_utf8_lossy(&output.stderr);

e2e-tests/tests/e2e.rs

@@ -13,7 +13,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use e2e_tests::{
 	assert_replacement, close_channel, find_available_port, list_payments, mine_and_sync,
-	payment_for_tx, run_cli, run_cli_raw, run_cli_with_config, send_bolt11_payment,
+	payment_for_tx, run_cli, run_cli_output, run_cli_raw, run_cli_with_config, send_bolt11_payment,
 	setup_funded_channel, splice_txid, wait_for_channels, wait_for_event, wait_for_gossip,
 	wait_for_onchain_balance, wait_for_settled_balance, wait_for_transaction,
 	wait_for_usable_channel, wait_for_wallet_sync, LdkServerConfig, LdkServerHandle, TestBitcoind,
@@ -675,7 +675,7 @@ async fn test_cli_list_peers() {
 
 // === CLI tests: Group 4 — Two-node with channel ===
 
-async fn open_channel_via_cli(channel_amount: &str) {
+async fn open_channel_via_cli(channel_amount: &str, extra_args: &[&str]) -> serde_json::Value {
 	let bitcoind = TestBitcoind::new();
 	let server_a = LdkServerHandle::start(&bitcoind).await;
 	let server_b = LdkServerHandle::start(&bitcoind).await;
@@ -691,21 +691,77 @@ async fn open_channel_via_cli(channel_amount: &str) {
 
 	// Open channel via CLI
 	let addr = format!("127.0.0.1:{}", server_b.p2p_port);
-	let output = run_cli(
-		&server_a,
-		&["open-channel", server_b.node_id(), &addr, channel_amount, "--announce-channel"],
-	);
+	let mut args =
+		vec!["open-channel", server_b.node_id(), &addr, channel_amount, "--announce-channel"];
+	args.extend(extra_args);
+	let output = run_cli(&server_a, &args);
 	assert!(!output["user_channel_id"].as_str().unwrap().is_empty());
+	output
 }
 
 #[tokio::test]
 async fn test_cli_open_channel() {
-	open_channel_via_cli("100000sat").await;
+	open_channel_via_cli("100000sat", &[]).await;
 }
 
 #[tokio::test]
 async fn test_cli_open_channel_with_all() {
-	open_channel_via_cli("all").await;
+	open_channel_via_cli("all", &[]).await;
+}
+
+#[tokio::test]
+async fn test_cli_open_channel_with_wait() {
+	let output = open_channel_via_cli("100000sat", &["--wait", "--wait-timeout", "60"]).await;
+
+	// `--wait` prints the `ChannelStateChanged` event once the funding transaction is broadcast.
+	assert!(!output["channel_id"].as_str().unwrap().is_empty());
+	let funding_txo = output["funding_txo"].as_str().unwrap();
+	let (txid, vout) = funding_txo.split_once(':').unwrap();
+	assert_eq!(txid.len(), 64);
+	vout.parse::<u32>().unwrap();
+}
+
+#[tokio::test]
+async fn test_cli_open_channel_with_wait_rejected() {
+	let bitcoind = TestBitcoind::new();
+	let server_a = LdkServerHandle::start(&bitcoind).await;
+	let server_b = LdkServerHandle::start(&bitcoind).await;
+
+	// Only fund server_a. ldk-node rejects inbound anchor channels when it lacks the on-chain
+	// reserve, so server_b turns the open down.
+	let addr_a = server_a.client().onchain_receive(OnchainReceiveRequest {}).await.unwrap().address;
+	bitcoind.fund_address(&addr_a, 1.0);
+	mine_and_sync(&bitcoind, &[&server_a, &server_b], 6).await;
+	wait_for_onchain_balance(server_a.client(), Duration::from_secs(30)).await;
+
+	let addr = format!("127.0.0.1:{}", server_b.p2p_port);
+	let output = run_cli_output(
+		&server_a,
+		&[
+			"open-channel",
+			server_b.node_id(),
+			&addr,
+			"100000sat",
+			// `--disable-counterparty-reserve` keeps the rejection visible. Without it, LDK retries the
+			// rejected open as a non-anchor channel, which needs no on-chain anchor reserve, and server_b
+			// accepts. LDK won't make a channel with no channel reserve non-anchor, so the open fails.
+			"--disable-counterparty-reserve",
+			"--wait",
+			"--wait-timeout",
+			"60",
+		],
+	);
+	let stdout = String::from_utf8(output.stdout).unwrap();
+	let stderr = String::from_utf8(output.stderr).unwrap();
+	// Check that the CLI exits with code 2 (`EXIT_WAIT_FAILED`), meaning the open failed.
+	assert_eq!(output.status.code(), Some(2), "stdout: {stdout}\nstderr: {stderr}");
+
+	// The `ChannelStateChanged` event is printed to stdout so scripts can read the reason.
+	let event: serde_json::Value = serde_json::from_str(&stdout).unwrap();
+	assert!(!event["user_channel_id"].as_str().unwrap().is_empty());
+	let message = event["reason"]["message"].as_str().unwrap();
+	assert!(message.contains("Channel request rejected"), "unexpected reason: {message}");
+	assert!(stderr.contains("Channel request rejected"), "stderr: {stderr}");
 }
 
 #[tokio::test]

ldk-server-cli/src/event_wait.rs

@@ -0,0 +1,58 @@
+// This file is Copyright its original authors, visible in version control
+// history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE
+// or http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
+// You may not use this file except in accordance with one or both of these
+// licenses.
+
+use std::time::Duration;
+
+use ldk_server_client::client::EventStream;
+use ldk_server_client::error::LdkServerError;
+use ldk_server_client::error::LdkServerErrorCode::InternalError;
+use ldk_server_client::ldk_server_grpc::events::event_envelope::Event;
+
+pub(crate) const EXIT_WAIT_FAILED: i32 = 2;
+pub(crate) const EXIT_WAIT_TIMEOUT: i32 = 3;
+
+pub(crate) enum WaitOutcome<T> {
+	Matched(T),
+	TimedOut,
+}
+
+pub(crate) async fn wait_for_event<T>(
+	events: &mut EventStream, subject: &str, timeout: Option<Duration>,
+	mut matcher: impl FnMut(Event) -> Option<T>,
+) -> Result<WaitOutcome<T>, LdkServerError> {
+	let deadline = timeout.map(|timeout| tokio::time::Instant::now() + timeout);
+
+	loop {
+		let next = events.next_message();
+		let message = match deadline {
+			Some(deadline) => match tokio::time::timeout_at(deadline, next).await {
+				Ok(message) => message,
+				Err(_) => return Ok(WaitOutcome::TimedOut),
+			},
+			None => next.await,
+		};
+
+		let envelope = match message {
+			Some(Ok(envelope)) => envelope,
+			Some(Err(e)) => return Err(e),
+			None => {
+				return Err(LdkServerError::new(
+					InternalError,
+					format!("event stream ended before {subject} resolved"),
+				))
+			},
+		};
+
+		if let Some(event) = envelope.event {
+			if let Some(matched) = matcher(event) {
+				return Ok(WaitOutcome::Matched(matched));
+			}
+		}
+	}
+}

ldk-server-cli/src/main.rs

@@ -78,6 +78,8 @@ use types::{
 	CliPaginatedResponse, Preimage,
 };
 
+mod event_wait;
+mod open_channel_wait;
 mod pay_wait;
 mod types;
 
@@ -602,6 +604,24 @@ enum Commands {
 			help = "The difference in the CLTV value between incoming HTLCs and an outbound HTLC forwarded over the channel. This can be updated by using update-channel-config."
 		)]
 		cltv_expiry_delta: Option<u32>,
+		/// Wait for the counterparty to accept or reject the channel.
+		///
+		/// With no `--wait-timeout`, waits until the funding transaction is broadcast or the open
+		/// fails. Does not wait for the funding transaction to confirm.
+		#[arg(
+			long,
+			help = "Wait until the funding transaction is broadcast or the open fails. Does not wait for confirmations"
+		)]
+		wait: bool,
+		/// Optional timeout in seconds for `--wait`. Omit to wait indefinitely.
+		#[arg(
+			long,
+			value_name = "SECS",
+			requires = "wait",
+			value_parser = clap::value_parser!(u64).range(1..),
+			help = "Seconds to wait when --wait is set. Omit to wait until the open resolves (minimum: 1)"
+		)]
+		wait_timeout: Option<u64>,
 	},
 	#[command(
 		about = "Increase the channel balance by the given amount, funds will come from the node's on-chain wallet"
@@ -1352,6 +1372,8 @@ async fn main() {
 			forwarding_fee_proportional_millionths,
 			forwarding_fee_base_msat,
 			cltv_expiry_delta,
+			wait,
+			wait_timeout,
 		} => {
 			let amount = match channel_amount.to_sat().unwrap_or_else(|e| handle_error_msg(e)) {
 				Some(amount_sats) => open_channel_request::Amount::ChannelAmountSats(amount_sats),
@@ -1370,19 +1392,23 @@ async fn main() {
 				));
 			}
 
-			handle_response_result::<_, OpenChannelResponse>(
-				client
-					.open_channel(OpenChannelRequest {
-						node_pubkey,
-						address,
-						amount: Some(amount),
-						push_to_counterparty_msat,
-						channel_config,
-						announce_channel,
-						disable_counterparty_reserve,
-					})
-					.await,
-			);
+			let request = OpenChannelRequest {
+				node_pubkey,
+				address,
+				amount: Some(amount),
+				push_to_counterparty_msat,
+				channel_config,
+				announce_channel,
+				disable_counterparty_reserve,
+			};
+			if wait {
+				let timeout = wait_timeout.map(std::time::Duration::from_secs);
+				open_channel_wait::open_channel_and_wait(&client, request, timeout).await;
+			} else {
+				handle_response_result::<_, OpenChannelResponse>(
+					client.open_channel(request).await,
+				);
+			}
 		},
 		Commands::SpliceIn { user_channel_id, counterparty_node_id, splice_amount } => {
 			let amount = match splice_amount.to_sat().unwrap_or_else(|e| handle_error_msg(e)) {

ldk-server-cli/src/open_channel_wait.rs

@@ -0,0 +1,126 @@
+// This file is Copyright its original authors, visible in version control
+// history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE
+// or http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
+// You may not use this file except in accordance with one or both of these
+// licenses.
+
+use std::time::Duration;
+
+use ldk_server_client::client::LdkServerClient;
+use ldk_server_client::ldk_server_grpc::api::{ListChannelsRequest, OpenChannelRequest};
+use ldk_server_client::ldk_server_grpc::events::event_envelope::Event;
+use ldk_server_client::ldk_server_grpc::events::{ChannelState, ChannelStateChanged};
+use ldk_server_client::ldk_server_grpc::types::Channel;
+
+use crate::event_wait::{wait_for_event, WaitOutcome, EXIT_WAIT_FAILED, EXIT_WAIT_TIMEOUT};
+use crate::{handle_error, print_response, sanitize_for_terminal};
+
+/// Subscribe before opening, then wait until the funding transaction is broadcast or the open
+/// fails.
+///
+/// The event stream is live-only, so the subscription is opened before `open_channel`. A dropped
+/// stream is an error; it is not replayed. `timeout` of `None` waits until the open resolves.
+/// This does not wait for the funding transaction to confirm.
+/// Exit codes: 0 pending, 1 request or event stream error, 2 failed, 3 timed out.
+pub(crate) async fn open_channel_and_wait(
+	client: &LdkServerClient, request: OpenChannelRequest, timeout: Option<Duration>,
+) {
+	let mut events = client.subscribe_events().await.unwrap_or_else(|e| handle_error(e));
+
+	let response = client.open_channel(request).await.unwrap_or_else(|e| handle_error(e));
+	let user_channel_id = response.user_channel_id;
+
+	let waiting_for = match timeout {
+		Some(timeout) => format!("waiting up to {}s for the counterparty", timeout.as_secs()),
+		None => "waiting for the counterparty".to_string(),
+	};
+	eprintln!(
+		"{}",
+		sanitize_for_terminal(format!("Channel open initiated: {user_channel_id} ({waiting_for})"))
+	);
+
+	let outcome =
+		wait_for_event(&mut events, &format!("channel open {user_channel_id}"), timeout, |event| {
+			classify(event, &user_channel_id)
+		})
+		.await
+		.unwrap_or_else(|e| handle_error(e));
+
+	match outcome {
+		WaitOutcome::Matched(Outcome::Pending(event)) => print_response(&event),
+		WaitOutcome::Matched(Outcome::Failed(event)) => {
+			let reason = event
+				.reason
+				.as_ref()
+				.map(|reason| reason.message.as_str())
+				.filter(|message| !message.is_empty())
+				.unwrap_or("no reason given");
+			eprintln!(
+				"{}",
+				sanitize_for_terminal(format!("Channel open {user_channel_id} failed: {reason}"))
+			);
+			print_response(&event);
+			std::process::exit(EXIT_WAIT_FAILED);
+		},
+		WaitOutcome::TimedOut => {
+			let secs = timeout.unwrap_or_default().as_secs();
+			eprintln!(
+				"{}",
+				sanitize_for_terminal(format!(
+					"Timed out after {secs}s waiting for channel open {user_channel_id}"
+				))
+			);
+			match fetch_channel(client, &user_channel_id).await {
+				Some(channel) => print_response(&channel),
+				None => eprintln!(
+					"{}",
+					sanitize_for_terminal(format!(
+						"Channel {user_channel_id} is not listed. The open may have failed; check the server log."
+					))
+				),
+			}
+			std::process::exit(EXIT_WAIT_TIMEOUT);
+		},
+	}
+}
+
+enum Outcome {
+	/// The funding transaction was broadcast.
+	Pending(ChannelStateChanged),
+	Failed(ChannelStateChanged),
+}
+
+/// `None` means the event is for another channel or is not a final state for the open.
+fn classify(event: Event, user_channel_id: &str) -> Option<Outcome> {
+	let Event::ChannelStateChanged(event) = event else {
+		return None;
+	};
+
+	if event.user_channel_id != user_channel_id {
+		return None;
+	}
+	match event.state() {
+		// `Ready` only arrives first if the `Pending` event was missed.
+		ChannelState::Pending | ChannelState::Ready => Some(Outcome::Pending(event)),
+		ChannelState::OpenFailed | ChannelState::Closed => Some(Outcome::Failed(event)),
+		ChannelState::Unspecified => None,
+	}
+}
+
+async fn fetch_channel(client: &LdkServerClient, user_channel_id: &str) -> Option<Channel> {
+	match client.list_channels(ListChannelsRequest {}).await {
+		Ok(response) => {
+			response.channels.into_iter().find(|channel| channel.user_channel_id == user_channel_id)
+		},
+		Err(e) => {
+			eprintln!(
+				"{}",
+				sanitize_for_terminal(format!("Warning: failed to list channels: {}", e.message))
+			);
+			None
+		},
+	}
+}