Track inbound payments by PaymentId

CHANGELOG.md

@@ -7,6 +7,11 @@
   a persisted `ChannelClosed` event.
 - Users of the VSS storage backend must upgrade their VSS server to at least version
   `v0.1.0-alpha.0` before upgrading LDK Node.
+- The `payment_id` field on the `PaymentSuccessful`, `PaymentFailed`, and
+  `PaymentReceived` events is now a required (non-optional) `PaymentId`. Events
+  persisted by LDK Node v0.2.1 or earlier (which stored `payment_id` as
+  optional) will fail to deserialize on read; users upgrading from those
+  versions need to drain pending events before the upgrade.
 
 ## Feature and API updates
 - The Bitcoin Core RPC and REST chain-source builder methods now accept an optional

benches/payments.rs

@@ -75,12 +75,8 @@ async fn send_payments(node_a: Arc<Node>, node_b: Arc<Node>) -> std::time::Durat
 	while success_count < total_payments {
 		match node_a.next_event_async().await {
 			Event::PaymentSuccessful { payment_id, payment_hash, .. } => {
-				if let Some(id) = payment_id {
-					success_count += 1;
-					println!("{}: Payment with id {:?} completed", payment_hash.0.as_hex(), id);
-				} else {
-					println!("Payment completed (no payment_id)");
-				}
+				success_count += 1;
+				println!("{}: Payment with id {:?} completed", payment_hash.0.as_hex(), payment_id);
 			},
 			Event::PaymentFailed { payment_id, payment_hash, .. } => {
 				println!("{}: Payment {:?} failed", payment_hash.unwrap().0.as_hex(), payment_id);

bindings/python/src/ldk_node/test_ldk_node.py

@@ -235,7 +235,7 @@ def test_spontaneous_payment(self):
         self.assertEqual(received_event.custom_records, custom_tlvs)
 
         sender_payment = node_1.payment(keysend_payment_id)
-        receiver_payment = node_2.payment(keysend_payment_id)
+        receiver_payment = node_2.payment(received_event.payment_id)
 
         self.assertIsNotNone(sender_payment)
         self.assertIsNotNone(receiver_payment)

src/builder.rs

@@ -2333,6 +2333,7 @@ fn build_with_store_internal(
 		scorer,
 		peer_store,
 		payment_store,
+		pending_payment_store,
 		lnurl_auth,
 		is_running,
 		node_metrics,

src/data_store.rs

@@ -112,11 +112,29 @@ 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 (removed_objects, result) = self.remove_batch_with_partial_result(ids).await;
+		result?;
+		Ok(removed_objects)
+	}
+
+	pub(crate) async fn remove_batch_with_partial_result(
+		&self, ids: &[SO::Id],
+	) -> (Vec<SO>, Result<(), 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(
+			let remove_result = KVStore::remove(
 				&*self.kv_store,
 				&self.primary_namespace,
 				&self.secondary_namespace,
@@ -134,10 +152,16 @@ where
 					e
 				);
 				Error::PersistenceFailed
-			})?;
-			self.objects.lock().expect("lock").remove(id);
+			});
+			if let Err(e) = remove_result {
+				return (removed_objects, Err(e));
+			}
+
+			if let Some(object) = self.objects.lock().expect("lock").remove(id) {
+				removed_objects.push(object);
+			}
 		}
-		Ok(())
+		(removed_objects, Ok(()))
 	}
 
 	/// Returns the current in-memory object for `id`.
@@ -422,6 +446,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] };