Keep funding payment records consistent across sync and classification

Funding broadcasts are classified into payment records off the
broadcaster's queue, which runs concurrently with wallet sync -- and can
run after sync has already recorded the transaction, for instance when
the counterparty's broadcast of a shared funding transaction is observed
first. The two writers raced: a late classification overwrote
confirmation state that wallet sync had already advanced, sync could
observe a half-written classification and record a duplicate generic
payment, and graduation could roll back figures a concurrent
classification had just written.

Make each writer's decision and writes atomic against the others:

- Classification merges only the transaction type and our contribution
  figures into an existing record, leaving the confirmation state that
  wallet-sync events own in place. Once a record is confirmed, its txid
  and figures describe the candidate that actually confirmed and are
  kept on a late classification, except when the update names the
  confirmed txid itself.

- Wallet sync resolves a transaction to its funding record before
  deciding how to record it -- including transactions known only as
  earlier RBF candidates -- and serializes with classification's
  two-store write pair from that resolution through its final write, so
  neither writer observes the other's torn state.

- Graduation decides from the live record and updates only the payment
  status, so a stale snapshot cannot roll back concurrently written
  figures.

- A missing pending-store entry is recreated while the payment is still
  Pending, repairing the index after a crash or failed write between
  the two stores.

Raised by Codex in the review of #888 and hardened through subsequent
review rounds.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: jkczyz

src/data_store.rs

@@ -81,6 +81,8 @@ where
 		Ok(updated)
 	}
 
+	/// Like [`Self::insert`], but when an entry with the object's id already exists, merges the
+	/// object's full update ([`StorableObject::to_update`]) into it instead of replacing it.
 	pub(crate) async fn insert_or_update(&self, object: SO) -> Result<bool, Error> {
 		let _guard = self.mutation_lock.lock().await;
 
@@ -170,6 +172,36 @@ where
 		Ok(DataStoreUpdateResult::Updated)
 	}
 
+	/// 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.
@@ -403,6 +435,131 @@ mod tests {
 		assert_eq!(Ok(true), data_store.insert_or_update(new_iou_object).await);
 	}
 
+	#[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/payment/pending_payment_store.rs

@@ -242,4 +242,79 @@ mod tests {
 			"current txid must not remain in its own conflict list"
 		);
 	}
+
+	#[test]
+	fn funding_classification_pending_update_preserves_mirrored_confirmation() {
+		use bitcoin::BlockHash;
+
+		use crate::payment::store::PaymentDetailsUpdate;
+
+		let txid = test_txid(7);
+		let payment_id = PaymentId(txid.to_byte_array());
+
+		// A pending entry wallet sync has already mirrored a confirmation into (via
+		// `apply_funding_status_update_locked`) before classification ran.
+		let confirmed_details = PaymentDetails::new(
+			payment_id,
+			PaymentKind::Onchain {
+				txid,
+				status: ConfirmationStatus::Confirmed {
+					block_hash: BlockHash::from_byte_array([8u8; 32]),
+					height: 100,
+					timestamp: 1,
+				},
+				tx_type: None,
+			},
+			Some(2_000_000),
+			Some(999),
+			PaymentDirection::Outbound,
+			PaymentStatus::Pending,
+		);
+		let mirrored = PendingPaymentDetails::new(confirmed_details, Vec::new(), Vec::new());
+
+		// A fresh classification is always Unconfirmed and carries the candidate history; its
+		// figures are the active candidate's.
+		let fresh = pending_onchain_payment(payment_id, txid);
+		let candidates = vec![FundingTxCandidate {
+			txid,
+			amount_msat: fresh.amount_msat,
+			fee_paid_msat: fresh.fee_paid_msat,
+		}];
+
+		// The old fresh-insert path merged the full fresh record, downgrading the mirrored
+		// confirmation.
+		let mut downgraded = mirrored.clone();
+		let full_update =
+			PendingPaymentDetails::new(fresh.clone(), Vec::new(), candidates.clone()).to_update();
+		assert!(downgraded.update(full_update));
+		assert!(
+			matches!(
+				downgraded.details.kind,
+				PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. }
+			),
+			"a full merge of a fresh classification downgrades a mirrored confirmation",
+		);
+
+		// The narrow classification update merges the candidates while preserving the
+		// confirmation state wallet sync owns. It names the confirmed txid, so its
+		// contribution-derived figures replace the mirrored wallet-view ones.
+		let mut merged = mirrored.clone();
+		let narrow_update = PendingPaymentDetailsUpdate {
+			id: payment_id,
+			payment_update: Some(PaymentDetailsUpdate::funding_reclassification(fresh)),
+			conflicting_txids: None,
+			candidates: candidates.clone(),
+		};
+		assert!(merged.update(narrow_update));
+		assert!(
+			matches!(
+				merged.details.kind,
+				PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. }
+			),
+			"a narrow classification update must not downgrade a mirrored confirmation",
+		);
+		assert_eq!(merged.candidates, candidates);
+		assert_eq!(merged.details.amount_msat, Some(1_000));
+		assert_eq!(merged.details.fee_paid_msat, Some(100));
+	}
 }