try async kvstore in sweeper

Author: joostjager

lightning/src/util/sweep.rs

@@ -430,24 +430,35 @@ where
 			return Ok(());
 		}
 
-		let mut state_lock = self.sweeper_state.lock().unwrap();
-		for descriptor in relevant_descriptors {
-			let output_info = TrackedSpendableOutput {
-				descriptor,
-				channel_id,
-				status: OutputSpendStatus::PendingInitialBroadcast {
-					delayed_until_height: delay_until_height,
-				},
-			};
-
-			let mut outputs = state_lock.persistent.outputs.iter();
-			if outputs.find(|o| o.descriptor == output_info.descriptor).is_some() {
-				continue;
+		let encoded;
+		let flush_fut;
+		{
+			let mut state_lock = self.sweeper_state.lock().unwrap();
+			for descriptor in relevant_descriptors {
+				let output_info = TrackedSpendableOutput {
+					descriptor,
+					channel_id,
+					status: OutputSpendStatus::PendingInitialBroadcast {
+						delayed_until_height: delay_until_height,
+					},
+				};
+
+				let mut outputs = state_lock.persistent.outputs.iter();
+				if outputs.find(|o| o.descriptor == output_info.descriptor).is_some() {
+					continue;
+				}
+
+				state_lock.persistent.outputs.push(output_info);
 			}
+			encoded = state_lock.persistent.encode();
 
-			state_lock.persistent.outputs.push(output_info);
+			// Not safe, because not yet persisted...
+			state_lock.dirty = false;
+
+			// Hopefully this fixates the ordering?
+			flush_fut = self.flush_state(encoded);
 		}
-		self.flush_state(&mut state_lock).await.map_err(|e| {
+		flush_fut.await.map_err(|e| {
 			log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
 		})
 	}
@@ -481,9 +492,12 @@ where
 			// If there is still dirty state, we need to persist it.
 			let mut sweeper_state = self.sweeper_state.lock().unwrap();
 			if sweeper_state.dirty {
-				self.flush_state(&mut sweeper_state).await.map_err(|e| {
-					log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
-				})
+				// TODO: Move outside lock.
+				// self.flush_state(&mut sweeper_state).await.map_err(|e| {
+				// 	log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
+				// })
+
+				Ok(())
 			} else {
 				Ok(())
 			}
@@ -533,6 +547,8 @@ where
 			self.change_destination_source.get_change_destination_script().await?;
 
 		// Sweep the outputs.
+		let flush_fut;
+		let encoded;
 		{
 			let mut sweeper_state = self.sweeper_state.lock().unwrap();
 
@@ -581,13 +597,18 @@ where
 				output_info.status.broadcast(cur_hash, cur_height, spending_tx.clone());
 			}
 
-			self.flush_state(&mut sweeper_state).await.map_err(|e| {
-				log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
-			})?;
+			encoded = sweeper_state.persistent.encode();
+			sweeper_state.dirty = false;
+
+			flush_fut = self.flush_state(encoded);
 
 			self.broadcaster.broadcast_transactions(&[&spending_tx]);
 		}
 
+		flush_fut.await.map_err(|e| {
+			log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
+		})?;
+
 		Ok(())
 	}
 
@@ -614,13 +635,13 @@ where
 	}
 
 	/// Flushes the current state to the persistence layer and marks the state as clean.
-	async fn flush_state(&self, sweeper_state: &mut SweeperState) -> Result<(), io::Error> {
+	async fn flush_state(&self, sweeper_state_encoded: Vec<u8>) -> Result<(), io::Error> {
 		self.kv_store
 			.write(
 				OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE,
 				OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
 				OUTPUT_SWEEPER_PERSISTENCE_KEY,
-				&sweeper_state.persistent.encode(),
+				&sweeper_state_encoded,
 			)
 			.await
 			.map_err(|e| {
@@ -634,9 +655,6 @@ where
 				);
 				e
 			})
-			.map(|_| {
-				sweeper_state.dirty = false;
-			})
 	}
 
 	fn spend_outputs(