Add pending payment batch removal

Expose a pending-store batch removal path so callers can prune
expired pending payments without repeatedly taking the
pending-store mutation lock.

Co-Authored-By: HAL 9000

Author: tnull

src/data_store.rs

@@ -112,9 +112,19 @@ where
 	}
 
 	pub(crate) async fn remove(&self, id: &SO::Id) -> Result<(), Error> {
+		self.remove_batch(std::slice::from_ref(id)).await?;
+		Ok(())
+	}
+
+	pub(crate) async fn remove_batch(&self, ids: &[SO::Id]) -> Result<Vec<SO>, Error> {
 		let _guard = self.mutation_lock.lock().await;
-		let should_remove = { self.objects.lock().expect("lock").contains_key(id) };
-		if should_remove {
+		let mut removed_objects = Vec::new();
+		for id in ids {
+			let should_remove = { self.objects.lock().expect("lock").contains_key(id) };
+			if !should_remove {
+				continue;
+			}
+
 			let store_key = id.encode_to_hex_str();
 			KVStore::remove(
 				&*self.kv_store,
@@ -135,9 +145,12 @@ where
 				);
 				Error::PersistenceFailed
 			})?;
-			self.objects.lock().expect("lock").remove(id);
+
+			if let Some(object) = self.objects.lock().expect("lock").remove(id) {
+				removed_objects.push(object);
+			}
 		}
-		Ok(())
+		Ok(removed_objects)
 	}
 
 	/// Returns the current in-memory object for `id`.
@@ -422,6 +435,45 @@ mod tests {
 		assert!(data_store.get(&new_id).is_none());
 	}
 
+	#[tokio::test]
+	async fn batch_remove_removes_persisted_objects() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let primary_namespace = "datastore_batch_remove_test_primary".to_string();
+		let secondary_namespace = "datastore_batch_remove_test_secondary".to_string();
+		let data_store: DataStore<TestObject, Arc<TestLogger>> = DataStore::new(
+			Vec::new(),
+			primary_namespace.clone(),
+			secondary_namespace.clone(),
+			Arc::clone(&store),
+			logger,
+		);
+
+		let first = TestObject { id: TestObjectId { id: [1u8; 4] }, data: [23u8; 3] };
+		let second = TestObject { id: TestObjectId { id: [2u8; 4] }, data: [42u8; 3] };
+		let missing_id = TestObjectId { id: [3u8; 4] };
+		assert_eq!(Ok(false), data_store.insert(first).await);
+		assert_eq!(Ok(false), data_store.insert(second).await);
+
+		assert_eq!(
+			Ok(vec![first, second]),
+			data_store.remove_batch(&[first.id, missing_id, second.id]).await
+		);
+		assert_eq!(None, data_store.get(&first.id));
+		assert_eq!(None, data_store.get(&second.id));
+
+		for id in [first.id, second.id] {
+			assert!(KVStore::read(
+				&*store,
+				&primary_namespace,
+				&secondary_namespace,
+				&id.encode_to_hex_str()
+			)
+			.await
+			.is_err());
+		}
+	}
+
 	#[tokio::test]
 	async fn insert_does_not_mutate_memory_if_persist_fails() {
 		let id = TestObjectId { id: [42u8; 4] };

src/payment/pending_payment_store.rs

@@ -240,11 +240,14 @@ where
 	}
 
 	pub(crate) async fn remove(&self, id: &PaymentId) -> Result<(), Error> {
+		self.remove_batch(std::slice::from_ref(id)).await
+	}
+
+	pub(crate) async fn remove_batch(&self, ids: &[PaymentId]) -> Result<(), Error> {
 		let _guard = self.mutation_lock.lock().await;
-		let before = self.inner.get(id);
-		self.inner.remove(id).await?;
-		if let Some(payment) = before.as_ref() {
-			self.remove_from_index(payment);
+		let removed_payments = self.inner.remove_batch(ids).await?;
+		for payment in removed_payments {
+			self.remove_from_index(&payment);
 		}
 		Ok(())
 	}