fix import

Author: jbesraa

.github/workflows/build.yml

@@ -180,6 +180,15 @@ jobs:
           cargo check --no-default-features --features=futures --release
           cargo doc --release
           RUSTDOCFLAGS="--cfg=anchors" cargo doc --release
+      - name: Run cargo check for Taproot build.
+        run: |
+          cargo check --release
+          cargo check --no-default-features --features=no-std --release
+          cargo check --no-default-features --features=futures --release
+          cargo doc --release
+        env:
+          RUSTFLAGS: '--cfg=anchors --cfg=taproot'
+          RUSTDOCFLAGS: '--cfg=anchors --cfg=taproot'
 
   fuzz:
     runs-on: ubuntu-latest

ci/ci-tests.sh

@@ -4,8 +4,9 @@ set -eox pipefail
 RUSTC_MINOR_VERSION=$(rustc --version | awk '{ split($2,a,"."); print a[2] }')
 HOST_PLATFORM="$(rustc --version --verbose | grep "host:" | awk '{ print $2 }')"
 
-# Tokio MSRV on versions newer than 1.14 is rustc 1.49
+# Tokio MSRV on versions 1.17 through 1.26 is rustc 1.49. Above 1.26 MSRV is 1.56.
 [ "$RUSTC_MINOR_VERSION" -lt 49 ] && cargo update -p tokio --precise "1.14.0" --verbose
+[[ "$RUSTC_MINOR_VERSION" -gt 48  &&  "$RUSTC_MINOR_VERSION" -lt 56 ]] && cargo update -p tokio --precise "1.26.0" --verbose
 [ "$LDK_COVERAGE_BUILD" != "" ] && export RUSTFLAGS="-C link-dead-code"
 
 export RUST_BACKTRACE=1
@@ -93,4 +94,6 @@ fi
 echo -e "\n\nTest anchors builds"
 pushd lightning
 RUSTFLAGS="$RUSTFLAGS --cfg=anchors" cargo test --verbose --color always -p lightning
+echo -e "\n\nTest Taproot builds"
+RUSTFLAGS="$RUSTFLAGS --cfg=anchors --cfg=taproot" cargo test --verbose --color always -p lightning
 popd

fuzz/src/chanmon_consistency.rs

@@ -38,17 +38,17 @@ use lightning::chain::channelmonitor::{ChannelMonitor, MonitorEvent};
 use lightning::chain::transaction::OutPoint;
 use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
 use lightning::chain::keysinterface::{KeyMaterial, InMemorySigner, Recipient, EntropySource, NodeSigner, SignerProvider};
+use lightning::events;
+use lightning::events::MessageSendEventsProvider;
 use lightning::ln::{PaymentHash, PaymentPreimage, PaymentSecret};
 use lightning::ln::channelmanager::{ChainParameters, ChannelDetails, ChannelManager, PaymentSendFailure, ChannelManagerReadArgs, PaymentId};
 use lightning::ln::channel::FEE_SPIKE_BUFFER_FEE_INCREASE_MULTIPLE;
 use lightning::ln::msgs::{self, CommitmentUpdate, ChannelMessageHandler, DecodeError, UpdateAddHTLC, Init};
 use lightning::ln::script::ShutdownScript;
 use lightning::util::enforcing_trait_impls::{EnforcingSigner, EnforcementState};
 use lightning::util::errors::APIError;
-use lightning::util::events;
 use lightning::util::logger::Logger;
 use lightning::util::config::UserConfig;
-use lightning::util::events::MessageSendEventsProvider;
 use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 use lightning::routing::router::{InFlightHtlcs, Route, RouteHop, RouteParameters, Router};
 
@@ -526,7 +526,18 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out) {
 					msg.clone()
 				} else { panic!("Wrong event type"); }
 			};
+			let events = $dest.get_and_clear_pending_events();
+			assert_eq!(events.len(), 1);
+			if let events::Event::ChannelPending{ ref counterparty_node_id, .. } = events[0] {
+				assert_eq!(counterparty_node_id, &$source.get_our_node_id());
+			} else { panic!("Wrong event type"); }
+
 			$source.handle_funding_signed(&$dest.get_our_node_id(), &funding_signed);
+			let events = $source.get_and_clear_pending_events();
+			assert_eq!(events.len(), 1);
+			if let events::Event::ChannelPending{ ref counterparty_node_id, .. } = events[0] {
+				assert_eq!(counterparty_node_id, &$dest.get_our_node_id());
+			} else { panic!("Wrong event type"); }
 
 			funding_output
 		} }

fuzz/src/full_stack.rs

@@ -35,6 +35,7 @@ use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget,
 use lightning::chain::chainmonitor;
 use lightning::chain::transaction::OutPoint;
 use lightning::chain::keysinterface::{InMemorySigner, Recipient, KeyMaterial, EntropySource, NodeSigner, SignerProvider};
+use lightning::events::Event;
 use lightning::ln::{PaymentHash, PaymentPreimage, PaymentSecret};
 use lightning::ln::channelmanager::{ChainParameters, ChannelDetails, ChannelManager, PaymentId};
 use lightning::ln::peer_handler::{MessageHandler,PeerManager,SocketDescriptor,IgnoringMessageHandler};
@@ -46,7 +47,6 @@ use lightning::routing::router::{find_route, InFlightHtlcs, PaymentParameters, R
 use lightning::routing::scoring::FixedPenaltyScorer;
 use lightning::util::config::UserConfig;
 use lightning::util::errors::APIError;
-use lightning::util::events::Event;
 use lightning::util::enforcing_trait_impls::{EnforcingSigner, EnforcementState};
 use lightning::util::logger::Logger;
 use lightning::util::ser::{Readable, Writeable};

lightning-background-processor/Cargo.toml

@@ -14,7 +14,7 @@ all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
 [features]
-futures = [ "futures-util" ]
+futures = [ ]
 std = ["lightning/std", "lightning-rapid-gossip-sync/std"]
 
 default = ["std"]
@@ -23,7 +23,6 @@ default = ["std"]
 bitcoin = { version = "0.29.0", default-features = false }
 lightning = { version = "0.0.114", path = "../lightning", default-features = false }
 lightning-rapid-gossip-sync = { version = "0.0.114", path = "../lightning-rapid-gossip-sync", default-features = false }
-futures-util = { version = "0.3", default-features = false, features = ["async-await-macro"], optional = true }
 
 [dev-dependencies]
 lightning = { version = "0.0.114", path = "../lightning", features = ["_test_utils"] }

lightning-background-processor/src/lib.rs

@@ -7,7 +7,7 @@
 #![deny(private_intra_doc_links)]
 
 #![deny(missing_docs)]
-#![deny(unsafe_code)]
+#![cfg_attr(not(feature = "futures"), deny(unsafe_code))]
 
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 
@@ -26,18 +26,20 @@ use lightning::chain;
 use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
 use lightning::chain::chainmonitor::{ChainMonitor, Persist};
 use lightning::chain::keysinterface::{EntropySource, NodeSigner, SignerProvider};
+use lightning::events::{Event, PathFailure};
+#[cfg(feature = "std")]
+use lightning::events::{EventHandler, EventsProvider};
 use lightning::ln::channelmanager::ChannelManager;
 use lightning::ln::msgs::{ChannelMessageHandler, OnionMessageHandler, RoutingMessageHandler};
 use lightning::ln::peer_handler::{CustomMessageHandler, PeerManager, SocketDescriptor};
 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
 use lightning::routing::utxo::UtxoLookup;
 use lightning::routing::router::Router;
 use lightning::routing::scoring::{Score, WriteableScore};
-use lightning::util::events::{Event, PathFailure};
-#[cfg(feature = "std")]
-use lightning::util::events::{EventHandler, EventsProvider};
 use lightning::util::logger::Logger;
 use lightning::util::persist::Persister;
+#[cfg(feature = "std")]
+use lightning::util::wakers::Sleeper;
 use lightning_rapid_gossip_sync::RapidGossipSync;
 
 use core::ops::Deref;
@@ -52,8 +54,6 @@ use std::thread::{self, JoinHandle};
 #[cfg(feature = "std")]
 use std::time::Instant;
 
-#[cfg(feature = "futures")]
-use futures_util::{select_biased, future::FutureExt, task};
 #[cfg(not(feature = "std"))]
 use alloc::vec::Vec;
 
@@ -80,7 +80,7 @@ use alloc::vec::Vec;
 /// unilateral chain closure fees are at risk.
 ///
 /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor
-/// [`Event`]: lightning::util::events::Event
+/// [`Event`]: lightning::events::Event
 #[cfg(feature = "std")]
 #[must_use = "BackgroundProcessor will immediately stop on drop. It should be stored until shutdown."]
 pub struct BackgroundProcessor {
@@ -116,6 +116,13 @@ const FIRST_NETWORK_PRUNE_TIMER: u64 = 60;
 #[cfg(test)]
 const FIRST_NETWORK_PRUNE_TIMER: u64 = 1;
 
+#[cfg(feature = "futures")]
+/// core::cmp::min is not currently const, so we define a trivial (and equivalent) replacement
+const fn min_u64(a: u64, b: u64) -> u64 { if a < b { a } else { b } }
+#[cfg(feature = "futures")]
+const FASTEST_TIMER: u64 = min_u64(min_u64(FRESHNESS_TIMER, PING_TIMER),
+	min_u64(SCORER_PERSIST_TIMER, FIRST_NETWORK_PRUNE_TIMER));
+
 /// Either [`P2PGossipSync`] or [`RapidGossipSync`].
 pub enum GossipSync<
 	P: Deref<Target = P2PGossipSync<G, U, L>>,
@@ -258,7 +265,8 @@ macro_rules! define_run_body {
 	($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr,
 	 $channel_manager: ident, $process_channel_manager_events: expr,
 	 $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident,
-	 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr)
+	 $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr,
+	 $check_slow_await: expr)
 	=> { {
 		log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
 		$channel_manager.timer_tick_occurred();
@@ -288,9 +296,10 @@ macro_rules! define_run_body {
 
 			// We wait up to 100ms, but track how long it takes to detect being put to sleep,
 			// see `await_start`'s use below.
-			let mut await_start = $get_timer(1);
+			let mut await_start = None;
+			if $check_slow_await { await_start = Some($get_timer(1)); }
 			let updates_available = $await;
-			let await_slow = $timer_elapsed(&mut await_start, 1);
+			let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false };
 
 			if updates_available {
 				log_trace!($logger, "Persisting ChannelManager...");
@@ -384,6 +393,59 @@ macro_rules! define_run_body {
 	} }
 }
 
+#[cfg(feature = "futures")]
+pub(crate) mod futures_util {
+	use core::future::Future;
+	use core::task::{Poll, Waker, RawWaker, RawWakerVTable};
+	use core::pin::Pin;
+	use core::marker::Unpin;
+	pub(crate) struct Selector<
+		A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
+	> {
+		pub a: A,
+		pub b: B,
+		pub c: C,
+	}
+	pub(crate) enum SelectorOutput {
+		A, B, C(bool),
+	}
+
+	impl<
+		A: Future<Output=()> + Unpin, B: Future<Output=()> + Unpin, C: Future<Output=bool> + Unpin
+	> Future for Selector<A, B, C> {
+		type Output = SelectorOutput;
+		fn poll(mut self: Pin<&mut Self>, ctx: &mut core::task::Context<'_>) -> Poll<SelectorOutput> {
+			match Pin::new(&mut self.a).poll(ctx) {
+				Poll::Ready(()) => { return Poll::Ready(SelectorOutput::A); },
+				Poll::Pending => {},
+			}
+			match Pin::new(&mut self.b).poll(ctx) {
+				Poll::Ready(()) => { return Poll::Ready(SelectorOutput::B); },
+				Poll::Pending => {},
+			}
+			match Pin::new(&mut self.c).poll(ctx) {
+				Poll::Ready(res) => { return Poll::Ready(SelectorOutput::C(res)); },
+				Poll::Pending => {},
+			}
+			Poll::Pending
+		}
+	}
+
+	// If we want to poll a future without an async context to figure out if it has completed or
+	// not without awaiting, we need a Waker, which needs a vtable...we fill it with dummy values
+	// but sadly there's a good bit of boilerplate here.
+	fn dummy_waker_clone(_: *const ()) -> RawWaker { RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE) }
+	fn dummy_waker_action(_: *const ()) { }
+
+	const DUMMY_WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
+		dummy_waker_clone, dummy_waker_action, dummy_waker_action, dummy_waker_action);
+	pub(crate) fn dummy_waker() -> Waker { unsafe { Waker::from_raw(RawWaker::new(core::ptr::null(), &DUMMY_WAKER_VTABLE)) } }
+}
+#[cfg(feature = "futures")]
+use futures_util::{Selector, SelectorOutput, dummy_waker};
+#[cfg(feature = "futures")]
+use core::task;
+
 /// Processes background events in a future.
 ///
 /// `sleeper` should return a future which completes in the given amount of time and returns a
@@ -396,6 +458,11 @@ macro_rules! define_run_body {
 /// feature, doing so will skip calling [`NetworkGraph::remove_stale_channels_and_tracking`],
 /// you should call [`NetworkGraph::remove_stale_channels_and_tracking_with_time`] regularly
 /// manually instead.
+///
+/// The `mobile_interruptable_platform` flag should be set if we're currently running on a
+/// mobile device, where we may need to check for interruption of the application regularly. If you
+/// are unsure, you should set the flag, as the performance impact of it is minimal unless there
+/// are hundreds or thousands of simultaneous process calls running.
 #[cfg(feature = "futures")]
 pub async fn process_events_async<
 	'a,
@@ -431,7 +498,7 @@ pub async fn process_events_async<
 >(
 	persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM,
 	gossip_sync: GossipSync<PGS, RGS, G, UL, L>, peer_manager: PM, logger: L, scorer: Option<S>,
-	sleeper: Sleeper,
+	sleeper: Sleeper, mobile_interruptable_platform: bool,
 ) -> Result<(), lightning::io::Error>
 where
 	UL::Target: 'static + UtxoLookup,
@@ -470,19 +537,25 @@ where
 		chain_monitor, chain_monitor.process_pending_events_async(async_event_handler).await,
 		channel_manager, channel_manager.process_pending_events_async(async_event_handler).await,
 		gossip_sync, peer_manager, logger, scorer, should_break, {
-			select_biased! {
-				_ = channel_manager.get_persistable_update_future().fuse() => true,
-				exit = sleeper(Duration::from_millis(100)).fuse() => {
+			let fut = Selector {
+				a: channel_manager.get_persistable_update_future(),
+				b: chain_monitor.get_update_future(),
+				c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }),
+			};
+			match fut.await {
+				SelectorOutput::A => true,
+				SelectorOutput::B => false,
+				SelectorOutput::C(exit) => {
 					should_break = exit;
 					false
 				}
 			}
 		}, |t| sleeper(Duration::from_secs(t)),
 		|fut: &mut SleepFuture, _| {
-			let mut waker = task::noop_waker();
+			let mut waker = dummy_waker();
 			let mut ctx = task::Context::from_waker(&mut waker);
 			core::pin::Pin::new(fut).poll(&mut ctx).is_ready()
-		})
+		}, mobile_interruptable_platform)
 }
 
 #[cfg(feature = "std")]
@@ -597,8 +670,11 @@ impl BackgroundProcessor {
 			define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler),
 				channel_manager, channel_manager.process_pending_events(&event_handler),
 				gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire),
-				channel_manager.await_persistable_update_timeout(Duration::from_millis(100)),
-				|_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur)
+				Sleeper::from_two_futures(
+					channel_manager.get_persistable_update_future(),
+					chain_monitor.get_update_future()
+				).wait_timeout(Duration::from_millis(100)),
+				|_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false)
 		});
 		Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
 	}
@@ -663,7 +739,8 @@ mod tests {
 	use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 	use lightning::chain::keysinterface::{InMemorySigner, KeysManager};
 	use lightning::chain::transaction::OutPoint;
-	use lightning::get_event_msg;
+	use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
+	use lightning::{get_event_msg, get_event};
 	use lightning::ln::PaymentHash;
 	use lightning::ln::channelmanager;
 	use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId};
@@ -674,7 +751,6 @@ mod tests {
 	use lightning::routing::router::{DefaultRouter, RouteHop};
 	use lightning::routing::scoring::{ChannelUsage, Score};
 	use lightning::util::config::UserConfig;
-	use lightning::util::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
 	use lightning::util::ser::Writeable;
 	use lightning::util::test_utils;
 	use lightning::util::persist::KVStorePersister;
@@ -1012,7 +1088,10 @@ mod tests {
 		($node_a: expr, $node_b: expr, $temporary_channel_id: expr, $tx: expr) => {{
 			$node_a.node.funding_transaction_generated(&$temporary_channel_id, &$node_b.node.get_our_node_id(), $tx.clone()).unwrap();
 			$node_b.node.handle_funding_created(&$node_a.node.get_our_node_id(), &get_event_msg!($node_a, MessageSendEvent::SendFundingCreated, $node_b.node.get_our_node_id()));
+			get_event!($node_b, Event::ChannelPending);
+
 			$node_a.node.handle_funding_signed(&$node_b.node.get_our_node_id(), &get_event_msg!($node_b, MessageSendEvent::SendFundingSigned, $node_a.node.get_our_node_id()));
+			get_event!($node_a, Event::ChannelPending);
 		}}
 	}
 

lightning-block-sync/Cargo.toml

@@ -20,7 +20,6 @@ rpc-client = [ "serde_json", "chunked_transfer" ]
 [dependencies]
 bitcoin = "0.29.0"
 lightning = { version = "0.0.114", path = "../lightning" }
-futures-util = { version = "0.3" }
 tokio = { version = "1.0", features = [ "io-util", "net", "time" ], optional = true }
 serde_json = { version = "1.0", optional = true }
 chunked_transfer = { version = "1.4", optional = true }

lightning-block-sync/src/rest.rs

@@ -7,32 +7,34 @@ use crate::http::{BinaryResponse, HttpEndpoint, HttpClient, JsonResponse};
 use bitcoin::hash_types::BlockHash;
 use bitcoin::hashes::hex::ToHex;
 
-use futures_util::lock::Mutex;
-
 use std::convert::TryFrom;
 use std::convert::TryInto;
+use std::sync::Mutex;
 
 /// A simple REST client for requesting resources using HTTP `GET`.
 pub struct RestClient {
 	endpoint: HttpEndpoint,
-	client: Mutex<HttpClient>,
+	client: Mutex<Option<HttpClient>>,
 }
 
 impl RestClient {
 	/// Creates a new REST client connected to the given endpoint.
 	///
 	/// The endpoint should contain the REST path component (e.g., http://127.0.0.1:8332/rest).
 	pub fn new(endpoint: HttpEndpoint) -> std::io::Result<Self> {
-		let client = Mutex::new(HttpClient::connect(&endpoint)?);
-		Ok(Self { endpoint, client })
+		Ok(Self { endpoint, client: Mutex::new(None) })
 	}
 
 	/// Requests a resource encoded in `F` format and interpreted as type `T`.
 	pub async fn request_resource<F, T>(&self, resource_path: &str) -> std::io::Result<T>
 	where F: TryFrom<Vec<u8>, Error = std::io::Error> + TryInto<T, Error = std::io::Error> {
 		let host = format!("{}:{}", self.endpoint.host(), self.endpoint.port());
 		let uri = format!("{}/{}", self.endpoint.path().trim_end_matches("/"), resource_path);
-		self.client.lock().await.get::<F>(&uri, &host).await?.try_into()
+		let mut client = if let Some(client) = self.client.lock().unwrap().take() { client }
+			else { HttpClient::connect(&self.endpoint)? };
+		let res = client.get::<F>(&uri, &host).await?.try_into();
+		*self.client.lock().unwrap() = Some(client);
+		res
 	}
 }