fixup! Reset LSPS5 persistence_in_flight counter on persist errors

Author: tnull

lightning-liquidity/src/lsps5/service.rs

@@ -251,85 +251,91 @@ where
 			return Ok(false);
 		}
 
-		let res = self.do_persist().await;
-		debug_assert!(res.is_err() || self.persistence_in_flight.load(Ordering::Acquire) == 0);
-		self.persistence_in_flight.store(0, Ordering::Release);
-		res
-	}
-
-	async fn do_persist(&self) -> Result<bool, lightning::io::Error> {
 		let mut did_persist = false;
 
 		loop {
-			let mut need_remove = Vec::new();
-			let mut need_persist = Vec::new();
+			match self.do_persist().await {
+				Ok(pass_did_persist) => did_persist |= pass_did_persist,
+				Err(e) => {
+					self.persistence_in_flight.store(0, Ordering::Release);
+					return Err(e);
+				},
+			}
 
-			self.check_prune_stale_webhooks(&mut self.per_peer_state.write().unwrap());
-			{
-				let outer_state_lock = self.per_peer_state.read().unwrap();
-
-				for (client_id, peer_state) in outer_state_lock.iter() {
-					let is_prunable = peer_state.is_prunable();
-					let has_open_channel = self.client_has_open_channel(client_id);
-					if is_prunable && !has_open_channel {
-						need_remove.push(*client_id);
-					} else if peer_state.needs_persist {
-						need_persist.push(*client_id);
-					}
-				}
+			if self.persistence_in_flight.fetch_sub(1, Ordering::AcqRel) != 1 {
+				// If another thread incremented the state while we were running we should go
+				// around again, but only once.
+				self.persistence_in_flight.store(1, Ordering::Release);
+				continue;
 			}
+			break;
+		}
 
-			for client_id in need_persist.into_iter() {
-				debug_assert!(!need_remove.contains(&client_id));
-				self.persist_peer_state(client_id).await?;
-				did_persist = true;
+		Ok(did_persist)
+	}
+
+	async fn do_persist(&self) -> Result<bool, lightning::io::Error> {
+		let mut did_persist = false;
+		let mut need_remove = Vec::new();
+		let mut need_persist = Vec::new();
+
+		self.check_prune_stale_webhooks(&mut self.per_peer_state.write().unwrap());
+		{
+			let outer_state_lock = self.per_peer_state.read().unwrap();
+
+			for (client_id, peer_state) in outer_state_lock.iter() {
+				let is_prunable = peer_state.is_prunable();
+				let has_open_channel = self.client_has_open_channel(client_id);
+				if is_prunable && !has_open_channel {
+					need_remove.push(*client_id);
+				} else if peer_state.needs_persist {
+					need_persist.push(*client_id);
+				}
 			}
+		}
 
-			for client_id in need_remove {
-				let mut future_opt = None;
-				{
-					// We need to take the `per_peer_state` write lock to remove an entry, but also
-					// have to hold it until after the `remove` call returns (but not through
-					// future completion) to ensure that writes for the peer's state are
-					// well-ordered with other `persist_peer_state` calls even across the removal
-					// itself.
-					let mut per_peer_state = self.per_peer_state.write().unwrap();
-					if let Entry::Occupied(mut entry) = per_peer_state.entry(client_id) {
-						let state = entry.get_mut();
-						if state.is_prunable() && !self.client_has_open_channel(&client_id) {
-							entry.remove();
-							let key = client_id.to_string();
-							future_opt = Some(self.kv_store.remove(
-								LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
-								LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
-								&key,
-								true,
-							));
-						} else {
-							// If the peer was re-added, force a re-persist of the current state.
-							state.needs_persist = true;
-						}
+		for client_id in need_persist.into_iter() {
+			debug_assert!(!need_remove.contains(&client_id));
+			self.persist_peer_state(client_id).await?;
+			did_persist = true;
+		}
+
+		for client_id in need_remove {
+			let mut future_opt = None;
+			{
+				// We need to take the `per_peer_state` write lock to remove an entry, but also
+				// have to hold it until after the `remove` call returns (but not through
+				// future completion) to ensure that writes for the peer's state are
+				// well-ordered with other `persist_peer_state` calls even across the removal
+				// itself.
+				let mut per_peer_state = self.per_peer_state.write().unwrap();
+				if let Entry::Occupied(mut entry) = per_peer_state.entry(client_id) {
+					let state = entry.get_mut();
+					if state.is_prunable() && !self.client_has_open_channel(&client_id) {
+						entry.remove();
+						let key = client_id.to_string();
+						future_opt = Some(self.kv_store.remove(
+							LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+							LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+							&key,
+							true,
+						));
 					} else {
-						// This should never happen, we can only have one `persist` call
-						// in-progress at once and map entries are only removed by it.
-						debug_assert!(false);
+						// If the peer was re-added, force a re-persist of the current state.
+						state.needs_persist = true;
 					}
-				}
-				if let Some(future) = future_opt {
-					future.await?;
-					did_persist = true;
 				} else {
-					self.persist_peer_state(client_id).await?;
+					// This should never happen, we can only have one `persist` call
+					// in-progress at once and map entries are only removed by it.
+					debug_assert!(false);
 				}
 			}
-
-			if self.persistence_in_flight.fetch_sub(1, Ordering::AcqRel) != 1 {
-				// If another thread incremented the state while we were running we should go
-				// around again, but only once.
-				self.persistence_in_flight.store(1, Ordering::Release);
-				continue;
+			if let Some(future) = future_opt {
+				future.await?;
+				did_persist = true;
+			} else {
+				self.persist_peer_state(client_id).await?;
 			}
-			break;
 		}
 
 		Ok(did_persist)