Reset persistence_in_flight counter on error in LSPS1/LSPS2

Previously, if any `.await?` in the persist loop returned an error, the
`?` would propagate out of `persist()` before reaching the `fetch_sub`
at the end of the loop. This left the counter permanently > 0, causing
all subsequent `persist()` calls to early-return and effectively
disabling persistence for the lifetime of the handler.

Fix this by extracting the loop into `do_persist()` and unconditionally
resetting the counter via `store(0, Release)` in the outer `persist()`
after `do_persist()` returns, regardless of success or failure.

Co-Authored-By: HAL 9000

Author: tnull

lightning-liquidity/src/lsps1/service.rs

@@ -145,14 +145,22 @@ where
 		// TODO: We should eventually persist in parallel, however, when we do, we probably want to
 		// introduce some batching to upper-bound the number of requests inflight at any given
 		// time.
-		let mut did_persist = false;
 
 		if self.persistence_in_flight.fetch_add(1, Ordering::AcqRel) > 0 {
 			// If we're not the first event processor to get here, just return early, the increment
 			// we just did will be treated as "go around again" at the end.
-			return Ok(did_persist);
+			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();

lightning-liquidity/src/lsps2/service.rs

@@ -1786,14 +1786,22 @@ where
 		// TODO: We should eventually persist in parallel, however, when we do, we probably want to
 		// introduce some batching to upper-bound the number of requests inflight at any given
 		// time.
-		let mut did_persist = false;
 
 		if self.persistence_in_flight.fetch_add(1, Ordering::AcqRel) > 0 {
 			// If we're not the first event processor to get here, just return early, the increment
 			// we just did will be treated as "go around again" at the end.
-			return Ok(did_persist);
+			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();