fixup! Implement tiered storage

Record the attempted write version even when a multi-store
peration fails, as one store may already contain the newer state.
This prevents a previously started write that completes later
from overwriting a newer primary write whose corresponding backup
write failed.

Add a regression test covering a successful primary write paired
with a failed backup write.

Author: enigbe

src/io/tier_store.rs

@@ -367,9 +367,9 @@ impl TierStoreInner {
 				Ok(())
 			} else {
 				let res = callback().await;
-				if res.is_ok() {
-					*last_written_version = version;
-				}
+				// A failed multi-store operation may still have updated one of its stores. We record
+				// the attempted version regardless so an older operation cannot overwrite newer state.
+				*last_written_version = version;
 				res
 			}
 		};
@@ -614,6 +614,7 @@ mod tests {
 	use std::future::Future;
 	use std::panic::RefUnwindSafe;
 	use std::path::PathBuf;
+	use std::sync::atomic::{AtomicUsize, Ordering};
 	use std::sync::Arc;
 
 	use lightning::util::logger::Level;
@@ -647,20 +648,25 @@ mod tests {
 		TierStore::new(primary_store, logger)
 	}
 
-	/// A store whose `list`/`list_paginated` always fail while every other operation is delegated
-	/// to an inner [`InMemoryStore`]. Used to prove that a failing ephemeral list does not sink a
-	/// listing for a namespace that can never hold an ephemeral-cached key.
-	struct FailingListStore {
+	enum FailureMode {
+		List,
+		Write { attempts: Arc<AtomicUsize> },
+	}
+
+	/// A store that injects a selected failure while delegating other operations to an inner
+	/// [`InMemoryStore`].
+	struct FailingStore {
 		inner: InMemoryStore,
+		failure_mode: FailureMode,
 	}
 
-	impl FailingListStore {
-		fn new() -> Self {
-			Self { inner: InMemoryStore::new() }
+	impl FailingStore {
+		fn new(failure_mode: FailureMode) -> Self {
+			Self { inner: InMemoryStore::new(), failure_mode }
 		}
 	}
 
-	impl KVStore for FailingListStore {
+	impl KVStore for FailingStore {
 		fn read(
 			&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 		) -> impl Future<Output = Result<Vec<u8>, io::Error>> + 'static + Send {
@@ -669,26 +675,62 @@ mod tests {
 		fn write(
 			&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
 		) -> impl Future<Output = Result<(), io::Error>> + 'static + Send {
-			KVStore::write(&self.inner, primary_namespace, secondary_namespace, key, buf)
+			let write = if let FailureMode::Write { attempts } = &self.failure_mode {
+				attempts.fetch_add(1, Ordering::Relaxed);
+				None
+			} else {
+				Some(KVStore::write(&self.inner, primary_namespace, secondary_namespace, key, buf))
+			};
+			async move {
+				match write {
+					Some(write) => write.await,
+					None => Err(io::Error::new(io::ErrorKind::Other, "write failed")),
+				}
+			}
 		}
 		fn remove(
 			&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
 		) -> impl Future<Output = Result<(), io::Error>> + 'static + Send {
 			KVStore::remove(&self.inner, primary_namespace, secondary_namespace, key, lazy)
 		}
 		fn list(
-			&self, _primary_namespace: &str, _secondary_namespace: &str,
+			&self, primary_namespace: &str, secondary_namespace: &str,
 		) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + Send {
-			async { Err(io::Error::new(io::ErrorKind::Other, "list failed")) }
+			let list = match &self.failure_mode {
+				FailureMode::List => None,
+				FailureMode::Write { .. } => {
+					Some(KVStore::list(&self.inner, primary_namespace, secondary_namespace))
+				},
+			};
+			async move {
+				match list {
+					Some(list) => list.await,
+					None => Err(io::Error::new(io::ErrorKind::Other, "list failed")),
+				}
+			}
 		}
 	}
 
-	impl PaginatedKVStore for FailingListStore {
+	impl PaginatedKVStore for FailingStore {
 		fn list_paginated(
-			&self, _primary_namespace: &str, _secondary_namespace: &str,
-			_page_token: Option<PageToken>,
+			&self, primary_namespace: &str, secondary_namespace: &str,
+			page_token: Option<PageToken>,
 		) -> impl Future<Output = Result<PaginatedListResponse, io::Error>> + 'static + Send {
-			async { Err(io::Error::new(io::ErrorKind::Other, "list_paginated failed")) }
+			let list = match &self.failure_mode {
+				FailureMode::List => None,
+				FailureMode::Write { .. } => Some(PaginatedKVStore::list_paginated(
+					&self.inner,
+					primary_namespace,
+					secondary_namespace,
+					page_token,
+				)),
+			};
+			async move {
+				match list {
+					Some(list) => list.await,
+					None => Err(io::Error::new(io::ErrorKind::Other, "list_paginated failed")),
+				}
+			}
 		}
 	}
 
@@ -1026,7 +1068,8 @@ mod tests {
 		let mut tier = setup_tier_store(Arc::clone(&primary_store), logger);
 
 		// An ephemeral store whose `list`/`list_paginated` always fail.
-		let ephemeral_store: Arc<DynStore> = Arc::new(DynStoreWrapper(FailingListStore::new()));
+		let ephemeral_store: Arc<DynStore> =
+			Arc::new(DynStoreWrapper(FailingStore::new(FailureMode::List)));
 		tier.set_ephemeral_store(Arc::clone(&ephemeral_store));
 
 		// A durable key in a namespace that can never hold an ephemeral-cached key.
@@ -1165,6 +1208,56 @@ mod tests {
 		assert_eq!(persisted, new_data);
 	}
 
+	#[tokio::test]
+	async fn failed_newer_backup_write_still_supersedes_older_write() {
+		let base_dir = random_storage_path();
+		let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned();
+		let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap());
+
+		let _cleanup = CleanupDir(base_dir);
+
+		let primary_store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let mut tier = setup_tier_store(Arc::clone(&primary_store), logger);
+
+		let backup_write_attempts = Arc::new(AtomicUsize::new(0));
+		let backup_store: Arc<DynStore> =
+			Arc::new(DynStoreWrapper(FailingStore::new(FailureMode::Write {
+				attempts: Arc::clone(&backup_write_attempts),
+			})));
+		tier.set_backup_store(backup_store);
+
+		let old_data = vec![1u8; 32];
+		let new_data = vec![2u8; 32];
+		let old_write = tier.write(
+			CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+			CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
+			CHANNEL_MANAGER_PERSISTENCE_KEY,
+			old_data,
+		);
+		let new_write = tier.write(
+			CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+			CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
+			CHANNEL_MANAGER_PERSISTENCE_KEY,
+			new_data.clone(),
+		);
+
+		// The primary write succeeds, but the same newer write fails on the backup.
+		assert!(new_write.await.is_err());
+		// The older operation must be treated as stale even though the newer operation failed.
+		old_write.await.unwrap();
+
+		let persisted = primary_store
+			.read(
+				CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+				CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
+				CHANNEL_MANAGER_PERSISTENCE_KEY,
+			)
+			.await
+			.unwrap();
+		assert_eq!(persisted, new_data);
+		assert_eq!(backup_write_attempts.load(Ordering::Relaxed), 1);
+	}
+
 	#[tokio::test]
 	async fn ephemeral_writes_preserve_latest_call_order() {
 		let base_dir = random_storage_path();