f - Reinsert failed handouts into the address pool by index

Concurrent failed handouts complete in pop order, so pushing each
returned address back to the front of the pool reverses their segment.
A subsequent successful handout would then serve a higher index while
lower unused ones stay pooled behind it, where a from-seed restore's
full-scan stop gap could strand them behind a used address. Reinsert
returned addresses at their index position instead, keeping handouts
consuming the oldest revealed index first.

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

Author: jkczyz

src/wallet/mod.rs

@@ -604,8 +604,15 @@ impl Wallet {
 			Ok(()) => Ok(address),
 			Err(e) => {
 				// The address was never handed out, so return it for the next caller rather
-				// than leaving its index revealed but unreachable.
-				self.address_pool.lock().expect("lock").available.push_front((index, address));
+				// than leaving its index revealed but unreachable. Reinsert by index:
+				// concurrent failed handouts complete in pop order, so pushing to the front
+				// would reverse their segment and let the next successful handout skip past a
+				// lower index, stranding it behind a used address in a from-seed restore's scan.
+				{
+					let mut locked_pool = self.address_pool.lock().expect("lock");
+					let position = locked_pool.available.partition_point(|(i, _)| *i < index);
+					locked_pool.available.insert(position, (index, address));
+				}
 				// The refill may have failed after rewriting the record, which then durably
 				// excludes the pushed-back index; rewrite it from the restored pool so a crash
 				// before the next successful refill doesn't strand the index outside the pool.
@@ -2789,12 +2796,13 @@ mod tests {
 		}
 	}
 
-	/// An in-memory store whose writes can be made to park until aborted, signalling when a
-	/// write has entered the gate.
+	/// An in-memory store whose writes can be made to park until aborted or released,
+	/// signalling when a write has entered the gate, and whose writes can be made to fail.
 	#[derive(Clone)]
 	struct GatedStore {
 		inner: Arc<InMemoryStore>,
 		gate_writes: Arc<AtomicBool>,
+		fail_writes: Arc<AtomicBool>,
 		write_entered: Arc<tokio::sync::Notify>,
 		release: Arc<tokio::sync::Notify>,
 	}
@@ -2804,6 +2812,7 @@ mod tests {
 			Self {
 				inner: Arc::new(InMemoryStore::new()),
 				gate_writes: Arc::new(AtomicBool::new(false)),
+				fail_writes: Arc::new(AtomicBool::new(false)),
 				write_entered: Arc::new(tokio::sync::Notify::new()),
 				release: Arc::new(tokio::sync::Notify::new()),
 			}
@@ -2822,6 +2831,7 @@ mod tests {
 		) -> impl Future<Output = Result<(), io::Error>> + 'static + Send {
 			let inner = Arc::clone(&self.inner);
 			let gate_writes = Arc::clone(&self.gate_writes);
+			let fail_writes = Arc::clone(&self.fail_writes);
 			let write_entered = Arc::clone(&self.write_entered);
 			let release = Arc::clone(&self.release);
 			let primary_namespace = primary_namespace.to_string();
@@ -2832,6 +2842,9 @@ mod tests {
 					write_entered.notify_one();
 					release.notified().await;
 				}
+				if fail_writes.load(Ordering::Acquire) {
+					return Err(io::Error::new(io::ErrorKind::Other, "write failed"));
+				}
 				KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf).await
 			}
 		}
@@ -3212,4 +3225,49 @@ mod tests {
 			indices
 		);
 	}
+
+	#[tokio::test]
+	async fn oldest_address_still_leads_the_pool_after_concurrent_failed_handouts() {
+		let gated_store = GatedStore::new();
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(gated_store.clone()));
+		let wallet = new_test_wallet(Arc::clone(&store), false).await;
+		wallet.refill_address_pool().await.unwrap();
+		let (_, oldest_address) =
+			wallet.address_pool.lock().unwrap().available.front().cloned().unwrap();
+
+		// First handout pops index 0 and parks inside its refill's record write, holding the
+		// refill lock.
+		gated_store.gate_writes.store(true, Ordering::Release);
+		gated_store.fail_writes.store(true, Ordering::Release);
+		let first_wallet = Arc::clone(&wallet);
+		let first_handout = tokio::spawn(async move { first_wallet.get_new_address().await });
+		gated_store.write_entered.notified().await;
+
+		// Second handout pops index 1 while the first is parked, then queues on the refill lock.
+		let second_wallet = Arc::clone(&wallet);
+		let second_handout = tokio::spawn(async move { second_wallet.get_new_address().await });
+		while wallet.address_pool.lock().unwrap().available.len() > ADDRESS_POOL_TARGET_SIZE - 2 {
+			tokio::task::yield_now().await;
+		}
+
+		// Both handouts now fail and return their indices to the pool, completing out of pop
+		// order: index 0 first, index 1 second.
+		gated_store.gate_writes.store(false, Ordering::Release);
+		gated_store.release.notify_one();
+		assert!(first_handout.await.unwrap().is_err());
+		assert!(second_handout.await.unwrap().is_err());
+		gated_store.fail_writes.store(false, Ordering::Release);
+
+		// The pushed-back indices must not swap the pool out of index order: the next handout
+		// has to serve the oldest revealed index, or a lower unused index would be left sitting
+		// behind a handed-out (potentially funded) one, where a from-seed restore's stop gap
+		// could strand it.
+		let handed_out = wallet.get_new_address().await.unwrap();
+		assert_eq!(
+			handed_out,
+			oldest_address,
+			"the oldest pooled address must be handed out first, pool: {:?}",
+			pooled_indices(&wallet)
+		);
+	}
 }