f - Read the pending-index gate's status inside the pending store's lock

The check that only Pending payments enter the pending index read the
payment store before taking the pending store's lock. Graduation could
write Succeeded and remove the index entry between the check and the
write, and the stale check would then re-create an entry for the
graduated payment. The next chain tip re-graduates it from the entry's
stale embedded copy — or, if that copy was still unconfirmed, keeps
rebroadcasting an already-confirmed transaction on every tip.

Move the decision into the pending store's critical section via a new
DataStore::mutate that reads, transforms, and persists an entry under
one hold of the mutation lock. Re-reading the payment's status there is
race-free because graduation writes Succeeded before removing the
entry: a read that still observes Pending precedes the removal, which
then also deletes anything inserted here. The closure runs with the
in-memory map lock released, so it may read other stores without
ordering the stores' map locks against each other.

The race spans a few instructions between two store writes and no seam
exists to schedule a graduation inside it, so no test exercises it; the
new unit tests cover the mutate primitive itself.

Generated with assistance from Claude Code.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: jkczyz

src/data_store.rs

@@ -223,6 +223,36 @@ where
 		Ok(result)
 	}
 
+	/// Atomically transforms the entry for `id` through `f` and persists the result.
+	///
+	/// `f` receives the current entry (`None` when absent) and returns the new state to write;
+	/// returning `None` leaves the store untouched. The read, the closure, and the write share
+	/// one critical section of the mutation lock, so no concurrent writer can land in between —
+	/// unlike a separate [`Self::get`] followed by an insert or update.
+	///
+	/// The closure runs on a clone of the entry with the in-memory map lock released, so it may
+	/// freely read this store or others (reads see the pre-mutation state) without ordering map
+	/// locks against each other. Keep it cheap and non-blocking.
+	///
+	/// Returns the written object, or `None` when the closure declined to write.
+	pub(crate) async fn mutate<F: FnOnce(Option<&SO>) -> Option<SO>>(
+		&self, id: &SO::Id, f: F,
+	) -> Result<Option<SO>, Error> {
+		let _guard = self.mutation_lock.lock().await;
+
+		let current = self.objects.lock().expect("lock").get(id).cloned();
+		let new_object = match f(current.as_ref()) {
+			Some(new_object) => new_object,
+			None => return Ok(None),
+		};
+		debug_assert!(new_object.id() == *id, "mutate closure must not change the object's id");
+
+		self.persist(&new_object).await?;
+		let mut locked_objects = self.objects.lock().expect("lock");
+		locked_objects.insert(new_object.id(), new_object.clone());
+		Ok(Some(new_object))
+	}
+
 	/// Returns in-memory objects matching `f`.
 	///
 	/// The async mutation lock serializes writers, but this synchronous reader cannot wait on it.
@@ -552,6 +582,131 @@ mod tests {
 		assert!(data_store.get(&new_id).is_none());
 	}
 
+	#[tokio::test]
+	async fn mutate_inserts_when_absent() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let primary_namespace = "datastore_test_primary".to_string();
+		let secondary_namespace = "datastore_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 id = TestObjectId { id: [42u8; 4] };
+		let object = TestObject { id, data: [23u8; 3] };
+		let result = data_store
+			.mutate(&id, |existing| {
+				assert!(existing.is_none());
+				Some(object)
+			})
+			.await;
+		assert_eq!(Ok(Some(object)), result);
+
+		assert_eq!(Some(object), data_store.get(&id));
+		let store_key = id.encode_to_hex_str();
+		assert!(KVStore::read(&*store, &primary_namespace, &secondary_namespace, &store_key)
+			.await
+			.is_ok());
+	}
+
+	#[tokio::test]
+	async fn mutate_transforms_existing_entry() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let id = TestObjectId { id: [42u8; 4] };
+		let existing_object = TestObject { id, data: [23u8; 3] };
+		let data_store: DataStore<TestObject, Arc<TestLogger>> = DataStore::new(
+			vec![existing_object],
+			"datastore_test_primary".to_string(),
+			"datastore_test_secondary".to_string(),
+			store,
+			logger,
+		);
+
+		// The closure sees the current entry and derives the new state from it.
+		let result = data_store
+			.mutate(&id, |existing| {
+				let mut new_object = *existing.unwrap();
+				new_object.data[0] += 1;
+				Some(new_object)
+			})
+			.await;
+		let expected = TestObject { id, data: [24u8, 23u8, 23u8] };
+		assert_eq!(Ok(Some(expected)), result);
+		assert_eq!(Some(expected), data_store.get(&id));
+	}
+
+	#[tokio::test]
+	async fn mutate_runs_the_closure_without_the_map_lock() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let id = TestObjectId { id: [42u8; 4] };
+		let existing_object = TestObject { id, data: [23u8; 3] };
+		let data_store: DataStore<TestObject, Arc<TestLogger>> = DataStore::new(
+			vec![existing_object],
+			"datastore_test_primary".to_string(),
+			"datastore_test_secondary".to_string(),
+			store,
+			logger,
+		);
+
+		// Closures gate cross-store decisions on reads of other stores, which lock their own
+		// in-memory maps. Holding this store's map lock across the closure would order it
+		// before theirs and invite lock-order inversions, so the closure must run with the map
+		// lock released.
+		let result = data_store
+			.mutate(&id, |existing| {
+				assert_eq!(Some(&existing_object), existing);
+				assert!(data_store.objects.try_lock().is_ok());
+				None
+			})
+			.await;
+		assert_eq!(Ok(None), result);
+	}
+
+	#[tokio::test]
+	async fn mutate_persists_nothing_when_closure_declines() {
+		let id = TestObjectId { id: [42u8; 4] };
+		let existing_object = TestObject { id, data: [23u8; 3] };
+		let data_store = new_failing_data_store(vec![existing_object]);
+
+		// Returning `None` must not attempt a write (the store fails all writes) nor touch memory.
+		let result = data_store
+			.mutate(&id, |existing| {
+				assert_eq!(Some(&existing_object), existing);
+				None
+			})
+			.await;
+		assert_eq!(Ok(None), result);
+		assert_eq!(Some(existing_object), data_store.get(&id));
+	}
+
+	#[tokio::test]
+	async fn mutate_does_not_mutate_memory_if_persist_fails() {
+		let existing_id = TestObjectId { id: [42u8; 4] };
+		let existing_object = TestObject { id: existing_id, data: [23u8; 3] };
+		let data_store = new_failing_data_store(vec![existing_object]);
+
+		let changed = TestObject { id: existing_id, data: [24u8; 3] };
+		assert_eq!(
+			Err(Error::PersistenceFailed),
+			data_store.mutate(&existing_id, |_| Some(changed)).await
+		);
+		assert_eq!(Some(existing_object), data_store.get(&existing_id));
+
+		let new_id = TestObjectId { id: [55u8; 4] };
+		let new_object = TestObject { id: new_id, data: [34u8; 3] };
+		assert_eq!(
+			Err(Error::PersistenceFailed),
+			data_store.mutate(&new_id, |_| Some(new_object)).await
+		);
+		assert!(data_store.get(&new_id).is_none());
+	}
+
 	#[tokio::test]
 	async fn insert_or_update_does_not_mutate_memory_if_persist_fails() {
 		let existing_id = TestObjectId { id: [42u8; 4] };

src/wallet/mod.rs

@@ -54,6 +54,7 @@ use lightning_invoice::RawBolt11Invoice;
 use persist::KVStoreWalletPersister;
 
 use crate::config::Config;
+use crate::data_store::StorableObject;
 use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator};
 use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate;
@@ -1412,34 +1413,52 @@ impl Wallet {
 			&candidates,
 			self.payment_store.get(&details.id).as_ref(),
 		);
-		let pending_update = PendingPaymentDetailsUpdate {
-			id: update.id,
-			payment_update: Some(update.clone()),
-			conflicting_txids: None,
-			candidates: candidates.clone(),
-		};
-		self.payment_store.update_or_insert(update, details.clone()).await?;
+		let id = update.id;
+		self.payment_store.update_or_insert(update.clone(), details.clone()).await?;
 
 		// The pending index must exist exactly while the authoritative record is Pending:
 		// graduation and rebroadcast read it, and a graduated payment must not be re-indexed.
 		// Deciding by the post-write status rather than by whether the write inserted also
 		// repairs a missing index — a crash or failed write between the two stores leaves a
 		// Pending record with no entry, and a merge alone would never recreate it, leaving the
 		// payment unable to graduate and its txids unmapped.
-		let recorded = self.payment_store.get(&details.id).unwrap_or(details);
-		if recorded.status == PaymentStatus::Pending {
-			// Wallet sync can still land between the payment-store write above and this one and
-			// mirror an advanced confirmation into the pending store, so this write makes the
-			// same atomic decision: merge narrowly into an entry that appeared, insert otherwise.
-			// The inserted entry embeds the post-write record rather than the fresh details, so a
-			// confirmation wallet sync already recorded keeps driving graduation.
-			let pending = PendingPaymentDetails::new(recorded, Vec::new(), candidates);
-			self.pending_payment_store.update_or_insert(pending_update, pending).await?;
-		} else {
-			// The payment already advanced beyond Pending: the graduation path removed the
-			// entry, and `update`'s no-op on absence must not re-create it.
-			self.pending_payment_store.update(pending_update).await?;
-		}
+		//
+		// The status must be read inside the pending store's critical section. Graduation writes
+		// `Succeeded` before removing the entry, so a read there that still observes `Pending`
+		// is ordered before the removal, which then also deletes anything inserted here. A
+		// status read taken before this write goes stale when graduation lands in between, and
+		// would re-index the graduated payment.
+		self.pending_payment_store
+			.mutate(&id, |existing| {
+				// The record was written above and payment records are never removed, so absence
+				// means the write failed out; fall back to the fresh details.
+				let recorded = self.payment_store.get(&id).unwrap_or(details);
+				match existing {
+					// The inserted entry embeds the post-write record rather than the fresh
+					// details, so a confirmation wallet sync already recorded keeps driving
+					// graduation.
+					None if recorded.status == PaymentStatus::Pending => {
+						Some(PendingPaymentDetails::new(recorded, Vec::new(), candidates))
+					},
+					// The payment already advanced beyond Pending: the graduation path removed
+					// the entry and it must not be re-created.
+					None => None,
+					// Wallet sync can land between the payment-store write above and this one
+					// and mirror an advanced confirmation into the pending store: merge only the
+					// classification into the entry that appeared.
+					Some(entry) => {
+						let pending_update = PendingPaymentDetailsUpdate {
+							id,
+							payment_update: Some(update),
+							conflicting_txids: None,
+							candidates,
+						};
+						let mut updated = entry.clone();
+						updated.update(pending_update).then_some(updated)
+					},
+				}
+			})
+			.await?;
 		Ok(())
 	}