Persist sweeper state as part of background process

To prepare for an async kv store trait that must be awaited, this commit
moves the kv store calls from the chain notification handlers to the
background process. It uses a dirty flag to communicate that there is
something to persist. The block height is part of the persisted data. If
that data does not make it to disk, the chain notifications are replayed
after restart.

Author: joostjager

lightning/src/util/sweep.rs

@@ -382,8 +382,10 @@ where
 		output_spender: O, change_destination_source: D, kv_store: K, logger: L,
 	) -> Self {
 		let outputs = Vec::new();
-		let sweeper_state =
-			Mutex::new(SweeperState { persistent: PersistentSweeperState { outputs, best_block } });
+		let sweeper_state = Mutex::new(SweeperState {
+			persistent: PersistentSweeperState { outputs, best_block },
+			dirty: false,
+		});
 		Self {
 			sweeper_state,
 			pending_sweep: AtomicBool::new(false),
@@ -445,9 +447,9 @@ where
 
 			state_lock.persistent.outputs.push(output_info);
 		}
-		self.persist_state(&state_lock).map_err(|e| {
-			log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
-		})
+		self.persist_state(&state_lock.persistent)
+			.inspect(|_| state_lock.dirty = false)
+			.map_err(|e| log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e))
 	}
 
 	/// Returns a list of the currently tracked spendable outputs.
@@ -504,12 +506,21 @@ where
 
 		// See if there is anything to sweep before requesting a change address.
 		{
-			let sweeper_state = self.sweeper_state.lock().unwrap();
+			let mut sweeper_state = self.sweeper_state.lock().unwrap();
 
 			let cur_height = sweeper_state.persistent.best_block.height;
 			let has_respends =
 				sweeper_state.persistent.outputs.iter().any(|o| filter_fn(o, cur_height));
 			if !has_respends {
+				// If there is nothing to sweep, we still persist the state if it is dirty.
+				if sweeper_state.dirty {
+					self.persist_state(&sweeper_state.persistent)
+						.inspect(|_| sweeper_state.dirty = false)
+						.map_err(|e| {
+							log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e)
+						})?;
+				}
+
 				return Ok(());
 			}
 		}
@@ -534,7 +545,8 @@ where
 				.collect();
 
 			if respend_descriptors.is_empty() {
-				// It could be that a tx confirmed and there is now nothing to sweep anymore.
+				// It could be that a tx confirmed and there is now nothing to sweep anymore. If there is dirty state,
+				// we'll persist it in the next cycle.
 				return Ok(());
 			}
 
@@ -567,9 +579,9 @@ where
 				output_info.status.broadcast(cur_hash, cur_height, spending_tx.clone());
 			}
 
-			self.persist_state(&sweeper_state).map_err(|e| {
-				log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
-			})?;
+			self.persist_state(&sweeper_state.persistent)
+				.inspect(|_| sweeper_state.dirty = false)
+				.map_err(|e| log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e))?;
 
 			self.broadcaster.broadcast_transactions(&[&spending_tx]);
 		}
@@ -595,6 +607,8 @@ where
 			}
 			true
 		});
+
+		sweeper_state.dirty = true;
 	}
 
 	fn persist_state(&self, sweeper_state: &PersistentSweeperState) -> Result<(), io::Error> {
@@ -648,13 +662,17 @@ where
 				}
 			}
 		}
+
+		sweeper_state.dirty = true;
 	}
 
 	fn best_block_updated_internal(
 		&self, sweeper_state: &mut SweeperState, header: &Header, height: u32,
 	) {
 		sweeper_state.persistent.best_block = BestBlock::new(header.block_hash(), height);
 		self.prune_confirmed_outputs(sweeper_state);
+
+		sweeper_state.dirty = true;
 	}
 }
 
@@ -678,12 +696,8 @@ where
 		assert_eq!(state_lock.persistent.best_block.height, height - 1,
 			"Blocks must be connected in chain-order - the connected block height must be one greater than the previous height");
 
-		self.transactions_confirmed_internal(&mut *state_lock, header, txdata, height);
-		self.best_block_updated_internal(&mut *state_lock, header, height);
-
-		let _ = self.persist_state(&*state_lock).map_err(|e| {
-			log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
-		});
+		self.transactions_confirmed_internal(&mut state_lock, header, txdata, height);
+		self.best_block_updated_internal(&mut state_lock, header, height);
 	}
 
 	fn block_disconnected(&self, header: &Header, height: u32) {
@@ -705,9 +719,7 @@ where
 			}
 		}
 
-		self.persist_state(&*state_lock).unwrap_or_else(|e| {
-			log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
-		});
+		state_lock.dirty = true;
 	}
 }
 
@@ -727,9 +739,6 @@ where
 	) {
 		let mut state_lock = self.sweeper_state.lock().unwrap();
 		self.transactions_confirmed_internal(&mut *state_lock, header, txdata, height);
-		self.persist_state(&*state_lock).unwrap_or_else(|e| {
-			log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
-		});
 	}
 
 	fn transaction_unconfirmed(&self, txid: &Txid) {
@@ -752,18 +761,13 @@ where
 				.filter(|o| o.status.confirmation_height() >= Some(unconf_height))
 				.for_each(|o| o.status.unconfirmed());
 
-			self.persist_state(&*state_lock).unwrap_or_else(|e| {
-				log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
-			});
+			state_lock.dirty = true;
 		}
 	}
 
 	fn best_block_updated(&self, header: &Header, height: u32) {
 		let mut state_lock = self.sweeper_state.lock().unwrap();
-		self.best_block_updated_internal(&mut *state_lock, header, height);
-		let _ = self.persist_state(&*state_lock).map_err(|e| {
-			log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
-		});
+		self.best_block_updated_internal(&mut state_lock, header, height);
 	}
 
 	fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option<BlockHash>)> {
@@ -792,6 +796,7 @@ where
 #[derive(Debug)]
 struct SweeperState {
 	persistent: PersistentSweeperState,
+	dirty: bool,
 }
 
 #[derive(Debug, Clone)]
@@ -856,7 +861,7 @@ where
 			}
 		}
 
-		let sweeper_state = Mutex::new(SweeperState { persistent: state });
+		let sweeper_state = Mutex::new(SweeperState { persistent: state, dirty: false });
 		Ok(Self {
 			sweeper_state,
 			pending_sweep: AtomicBool::new(false),
@@ -905,7 +910,7 @@ where
 			}
 		}
 
-		let sweeper_state = Mutex::new(SweeperState { persistent: state });
+		let sweeper_state = Mutex::new(SweeperState { persistent: state, dirty: false });
 		Ok((
 			best_block,
 			OutputSweeper {