Add PostgreSQL node failover leases

src/builder.rs

@@ -190,6 +190,8 @@ pub enum BuildError {
 	///
 	/// [`KVStore`]: lightning::util::persist::KVStore
 	KVStoreSetupFailed,
+	/// Another node currently owns the PostgreSQL node lease.
+	NodeLeaseUnavailable,
 	/// We failed to setup the onchain wallet.
 	WalletSetupFailed,
 	/// We failed to setup the logger.
@@ -232,6 +234,7 @@ impl fmt::Display for BuildError {
 			Self::WriteFailed => write!(f, "Failed to write to store."),
 			Self::StoragePathAccessFailed => write!(f, "Failed to access the given storage path."),
 			Self::KVStoreSetupFailed => write!(f, "Failed to setup KVStore."),
+			Self::NodeLeaseUnavailable => write!(f, "The PostgreSQL node lease is unavailable."),
 			Self::WalletSetupFailed => write!(f, "Failed to setup onchain wallet."),
 			Self::LoggerSetupFailed => write!(f, "Failed to setup the logger."),
 			Self::ChainSourceSetupFailed => write!(f, "Failed to setup the chain source."),
@@ -683,6 +686,9 @@ impl NodeBuilder {
 	/// Builds a [`Node`] instance with a [PostgreSQL] backend and according to the options
 	/// previously configured.
 	///
+	/// This acquires an exclusive lease for the selected KV table before reading persisted node
+	/// state. Nodes may share a database when each node identity uses a distinct `kv_table_name`.
+	///
 	/// Connects to the PostgreSQL database at the given `connection_string`, e.g.,
 	/// `"postgres://user:password@localhost/ldk_db"`.
 	///
@@ -701,6 +707,9 @@ impl NodeBuilder {
 	/// certificates (it does not replace them). If `certificate_pem` is `None`, connections
 	/// will be unencrypted.
 	///
+	/// Returns [`BuildError::NodeLeaseUnavailable`] while another process owns the selected KV
+	/// table's lease.
+	///
 	/// [PostgreSQL]: https://www.postgresql.org
 	#[cfg(feature = "postgres")]
 	pub fn build_with_postgres_store(
@@ -709,19 +718,29 @@ impl NodeBuilder {
 	) -> Result<Node, BuildError> {
 		let logger = setup_logger(&self.log_writer_config, &self.config)?;
 		let runtime = self.setup_runtime(&logger)?;
-		let kv_store = runtime
-			.block_on(io::postgres_store::PostgresStore::new_with_logger(
-				connection_string,
-				db_name,
-				kv_table_name,
-				certificate_pem,
-				Some(Arc::clone(&logger)),
-			))
-			.map_err(|e| {
+		let kv_store = match runtime.block_on(io::postgres_store::PostgresStore::new_with_logger(
+			connection_string,
+			db_name,
+			kv_table_name,
+			certificate_pem,
+			Some(Arc::clone(&logger)),
+		)) {
+			Ok(kv_store) => kv_store,
+			Err(e) if e.kind() == lightning::io::ErrorKind::WouldBlock => {
+				return Err(BuildError::NodeLeaseUnavailable);
+			},
+			Err(e) => {
 				log_error!(logger, "Failed to set up Postgres store: {e}");
-				BuildError::KVStoreSetupFailed
-			})?;
-		self.build_with_store_runtime_and_logger(node_entropy, kv_store, runtime, logger)
+				return Err(BuildError::KVStoreSetupFailed);
+			},
+		};
+		let node_lease = kv_store.node_lease();
+		let mut node =
+			self.build_with_store_runtime_and_logger(node_entropy, kv_store, runtime, logger)?;
+		if !node.install_node_lease(node_lease) {
+			return Err(BuildError::NodeLeaseUnavailable);
+		}
+		Ok(node)
 	}
 
 	/// Builds a [`Node`] instance with a [`FilesystemStoreV2`] backend and according to the options
@@ -1217,6 +1236,9 @@ impl ArcedNodeBuilder {
 	/// Builds a [`Node`] instance with a [PostgreSQL] backend and according to the options
 	/// previously configured.
 	///
+	/// This acquires an exclusive lease for the selected KV table before reading persisted node
+	/// state. Nodes may share a database when each node identity uses a distinct `kv_table_name`.
+	///
 	/// Connects to the PostgreSQL database at the given `connection_string`, e.g.,
 	/// `"postgres://user:password@localhost/ldk_db"`.
 	///
@@ -1235,6 +1257,9 @@ impl ArcedNodeBuilder {
 	/// certificates (it does not replace them). If `certificate_pem` is `None`, connections
 	/// will be unencrypted.
 	///
+	/// Returns [`BuildError::NodeLeaseUnavailable`] while another process owns the selected KV
+	/// table's lease.
+	///
 	/// [PostgreSQL]: https://www.postgresql.org
 	#[cfg(feature = "postgres")]
 	pub fn build_with_postgres_store(
@@ -2335,6 +2360,7 @@ fn build_with_store_internal(
 		payment_store,
 		lnurl_auth,
 		is_running,
+		node_lease: None,
 		node_metrics,
 		om_mailbox,
 		async_payments_role,

src/io/mod.rs

@@ -7,6 +7,8 @@
 
 //! Objects and traits for data persistence.
 
+#[cfg_attr(not(feature = "postgres"), allow(dead_code))]
+pub(crate) mod node_lease;
 #[cfg(feature = "postgres")]
 pub mod postgres_store;
 pub mod sqlite_store;

src/io/node_lease.rs

@@ -0,0 +1,138 @@
+// 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::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use lightning::io;
+
+pub(crate) const NODE_LEASE_DURATION: Duration = Duration::from_secs(30);
+// Fail closed before the database lease expires, leaving time for process termination.
+pub(crate) const NODE_LEASE_RENEWAL_DEADLINE: Duration = Duration::from_secs(20);
+pub(crate) const NODE_LEASE_RENEWAL_INTERVAL: Duration = Duration::from_secs(10);
+pub(crate) const NODE_LEASE_RETRY_INTERVAL: Duration = Duration::from_secs(1);
+pub(crate) const NODE_LEASE_RELEASE_TIMEOUT: Duration = Duration::from_secs(5);
+
+type LeaseLossHandler = Box<dyn FnOnce() + Send>;
+
+pub(crate) struct NodeLease {
+	owner_id: [u8; 32],
+	lease_lost: AtomicBool,
+	last_confirmed_renewal: Mutex<Instant>,
+	loss_sender: tokio::sync::watch::Sender<bool>,
+	loss_handler: Mutex<Option<LeaseLossHandler>>,
+}
+
+impl NodeLease {
+	pub(crate) fn new() -> io::Result<Arc<Self>> {
+		let mut owner_id = [0u8; 32];
+		getrandom::fill(&mut owner_id).map_err(|e| {
+			io::Error::new(io::ErrorKind::Other, format!("Failed to generate lease owner ID: {e}"))
+		})?;
+		let (loss_sender, _) = tokio::sync::watch::channel(false);
+		Ok(Arc::new(Self {
+			owner_id,
+			lease_lost: AtomicBool::new(false),
+			last_confirmed_renewal: Mutex::new(Instant::now()),
+			loss_sender,
+			loss_handler: Mutex::new(None),
+		}))
+	}
+
+	pub(crate) fn owner_id(&self) -> &[u8; 32] {
+		&self.owner_id
+	}
+
+	pub(crate) fn is_lost(&self) -> bool {
+		self.lease_lost.load(Ordering::Acquire)
+	}
+
+	pub(crate) fn record_renewal(&self) {
+		if !self.is_lost() {
+			*self.last_confirmed_renewal.lock().expect("lock") = Instant::now();
+		}
+	}
+
+	pub(crate) fn renewal_deadline_elapsed(&self) -> bool {
+		self.last_confirmed_renewal.lock().expect("lock").elapsed() >= NODE_LEASE_RENEWAL_DEADLINE
+	}
+
+	pub(crate) fn ensure_operation_active(&self) -> io::Result<()> {
+		if self.is_lost() || self.renewal_deadline_elapsed() {
+			self.mark_lost();
+			Err(lease_lost_error())
+		} else {
+			Ok(())
+		}
+	}
+
+	pub(crate) fn map_operation_error(&self, error: io::Error) -> io::Error {
+		// Preserve transient database errors until they outlive the local safety margin.
+		self.ensure_operation_active().err().unwrap_or(error)
+	}
+
+	pub(crate) fn mark_lost(&self) {
+		if self
+			.lease_lost
+			.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+			.is_err()
+		{
+			return;
+		}
+
+		// Run any installed containment handler before publishing lease loss.
+		if let Some(handler) = self.loss_handler.lock().expect("lock").take() {
+			handler();
+		}
+		self.loss_sender.send_replace(true);
+	}
+
+	pub(crate) fn set_loss_handler(&self, handler: LeaseLossHandler) {
+		let mut locked_handler = self.loss_handler.lock().expect("lock");
+		if self.is_lost() {
+			drop(locked_handler);
+			handler();
+		} else {
+			*locked_handler = Some(handler);
+		}
+	}
+
+	pub(crate) async fn wait_for_loss(self: Arc<Self>) {
+		let mut receiver = self.loss_sender.subscribe();
+		let _ = receiver.wait_for(|lost| *lost).await;
+	}
+}
+
+pub(crate) fn lease_lost_error() -> io::Error {
+	io::Error::new(io::ErrorKind::PermissionDenied, "PostgreSQL node lease was lost")
+}
+
+#[cfg(test)]
+mod tests {
+	use std::sync::atomic::{AtomicBool, Ordering};
+
+	use super::*;
+
+	#[test]
+	fn expired_operation_marks_loss_before_returning_error() {
+		let lease = NodeLease::new().unwrap();
+		let handler_ran = Arc::new(AtomicBool::new(false));
+		let handler_ran_ref = Arc::clone(&handler_ran);
+		lease.set_loss_handler(Box::new(move || {
+			handler_ran_ref.store(true, Ordering::Release);
+		}));
+		*lease.last_confirmed_renewal.lock().unwrap() =
+			Instant::now() - NODE_LEASE_RENEWAL_DEADLINE;
+
+		let error = lease.map_operation_error(io::Error::from(io::ErrorKind::Other));
+
+		assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
+		assert!(lease.is_lost());
+		assert!(handler_ran.load(Ordering::Acquire));
+	}
+}