Implement concurrent broadcast tolerance for distributed watchtowers

With a distrbuted watchtowers deployment, where each monitor is plugged
to its own chain view, there is no guarantee that block are going to be
seen in same order. Watchtower may diverge in their acceptance of a
submitted `commitment_signed` update due to a block timing-out a HTLC
and provoking a subset but yet not seen by the other watchtower subset.
Any update reject by one of the watchtower must block offchain coordinator
to move channel state forward and release revocation secret for previous
state.

In this case, we want any watchtower from the rejection subset to still
be able to claim outputs if the concurrent state, has accepted by the
other subset, is confirming. This improve overall watchtower system
fault-tolerance.

This change stores local commitment transaction unconditionally and fail
the update if there is knowledge of an already signed commitment
transaction (ChannelMonitor.local_tx_signed=true).

Author: Antoine Riard

lightning/src/ln/channelmonitor.rs

@@ -823,6 +823,10 @@ pub struct ChannelMonitor<ChanSigner: ChannelKeys> {
 	// Set once we've signed a local commitment transaction and handed it over to our
 	// OnchainTxHandler. After this is set, no future updates to our local commitment transactions
 	// may occur, and we fail any such monitor updates.
+	//
+	// In case of update rejection due to a locally already signed commitment transaction, we
+	// nevertheless store update content to track in case of concurrent broadcast by another
+	// remote monitor out-of-order with regards to the block view.
 	local_tx_signed: bool,
 
 	// We simply modify last_block_hash in Channel's block_connected so that serialization is
@@ -887,6 +891,11 @@ pub trait ManyChannelMonitor: Send + Sync {
 	///
 	/// Any spends of outputs which should have been registered which aren't passed to
 	/// ChannelMonitors via block_connected may result in FUNDS LOSS.
+	///
+	/// In case of distributed watchtowers deployment, even if an Err is return, the new version
+	/// must be written to disk, as state may have been stored but rejected due to a block forcing
+	/// a commitment broadcast. This storage is used to claim outputs of rejected state confirmed
+	/// onchain by another watchtower, lagging behind on block processing.
 	fn update_monitor(&self, funding_txo: OutPoint, monitor: ChannelMonitorUpdate) -> Result<(), ChannelMonitorUpdateErr>;
 
 	/// Used by ChannelManager to get list of HTLC resolved onchain and which needed to be updated
@@ -1166,12 +1175,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 			feerate_per_kw: initial_local_commitment_tx.feerate_per_kw,
 			htlc_outputs: Vec::new(), // There are never any HTLCs in the initial commitment transactions
 		};
-		// Returning a monitor error before updating tracking points means in case of using
-		// a concurrent watchtower implementation for same channel, if this one doesn't
-		// reject update as we do, you MAY have the latest local valid commitment tx onchain
-		// for which you want to spend outputs. We're NOT robust again this scenario right
-		// now but we should consider it later.
-		onchain_tx_handler.provide_latest_local_tx(initial_local_commitment_tx).unwrap();
+		onchain_tx_handler.provide_latest_local_tx(initial_local_commitment_tx);
 
 		ChannelMonitor {
 			latest_update_id: 0,
@@ -1326,9 +1330,6 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 	/// up-to-date as our local commitment transaction is updated.
 	/// Panics if set_on_local_tx_csv has never been called.
 	pub(super) fn provide_latest_local_commitment_tx_info(&mut self, commitment_tx: LocalCommitmentTransaction, htlc_outputs: Vec<(HTLCOutputInCommitment, Option<Signature>, Option<HTLCSource>)>) -> Result<(), MonitorUpdateError> {
-		if self.local_tx_signed {
-			return Err(MonitorUpdateError("A local commitment tx has already been signed, no new local commitment txn can be sent to our counterparty"));
-		}
 		let txid = commitment_tx.txid();
 		let sequence = commitment_tx.unsigned_tx.input[0].sequence as u64;
 		let locktime = commitment_tx.unsigned_tx.lock_time as u64;
@@ -1342,17 +1343,13 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 			feerate_per_kw: commitment_tx.feerate_per_kw,
 			htlc_outputs: htlc_outputs,
 		};
-		// Returning a monitor error before updating tracking points means in case of using
-		// a concurrent watchtower implementation for same channel, if this one doesn't
-		// reject update as we do, you MAY have the latest local valid commitment tx onchain
-		// for which you want to spend outputs. We're NOT robust again this scenario right
-		// now but we should consider it later.
-		if let Err(_) = self.onchain_tx_handler.provide_latest_local_tx(commitment_tx) {
-			return Err(MonitorUpdateError("Local commitment signed has already been signed, no further update of LOCAL commitment transaction is allowed"));
-		}
+		self.onchain_tx_handler.provide_latest_local_tx(commitment_tx);
 		self.current_local_commitment_number = 0xffff_ffff_ffff - ((((sequence & 0xffffff) << 3*8) | (locktime as u64 & 0xffffff)) ^ self.commitment_transaction_number_obscure_factor);
 		mem::swap(&mut new_local_commitment_tx, &mut self.current_local_commitment_tx);
 		self.prev_local_signed_commitment_tx = Some(new_local_commitment_tx);
+		if self.local_tx_signed {
+			return Err(MonitorUpdateError("Latest local commitment signed has already been signed, update is rejected"));
+		}
 		Ok(())
 	}
 

lightning/src/ln/onchaintx.rs

@@ -877,18 +877,9 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 		}
 	}
 
-	pub(super) fn provide_latest_local_tx(&mut self, tx: LocalCommitmentTransaction) -> Result<(), ()> {
-		// To prevent any unsafe state discrepancy between offchain and onchain, once local
-		// commitment transaction has been signed due to an event (either block height for
-		// HTLC-timeout or channel force-closure), don't allow any further update of local
-		// commitment transaction view to avoid delivery of revocation secret to counterparty
-		// for the aformentionned signed transaction.
-		if self.local_htlc_sigs.is_some() || self.prev_local_htlc_sigs.is_some() {
-			return Err(());
-		}
+	pub(super) fn provide_latest_local_tx(&mut self, tx: LocalCommitmentTransaction) {
 		self.prev_local_commitment = self.local_commitment.take();
 		self.local_commitment = Some(tx);
-		Ok(())
 	}
 
 	fn sign_latest_local_htlcs(&mut self) {