Recover TierStore writes after a backup is removed

A write can be interrupted while a backup is configured, leaving unfinished
work in the journal. If the node then restarts without that backup, recovery
must not wait forever for a store that was deliberately removed.

Updates to existing keys were also not journaled. If the primary write
succeeded but the backup write failed, TierStore did not retain the value
needed to finish that update later.

Separately, a matching primary generation alone does not prove a backup is
current. If the local TierStore index has been replaced, the old index may
have contained unfinished operations that were lost with it, and generation
comparison alone would not detect this.

In this commit, we:

- Journal updates that must be written to both primary and backup storage.

- Allow unfinished creates, updates, and removals to finish using only the
  primary store after restarting without the backup.

- Rotate the primary synchronization generation first, ensuring that the
  missing backup is recognized as out of date when it is configured again.

- Reopen an existing TierStore index even when the backup is no longer
  configured, so unfinished journal entries can still be recovered.

- Store the index database identity alongside the primary generation in the
  backup completion record, and treat a completion from another index as
  requiring synchronization.

- Add tests for failed updates, recovery without a backup, attempts to
  recover before backup synchronization has been initialized, completion
  encoding, and index replacement.

A later commit will copy the current primary data into a new or outdated
backup.

Assisted-by: Amp (AI coding agent)

Author: enigbe

src/builder.rs

@@ -992,15 +992,27 @@ impl NodeBuilder {
 			let ts_config = self.tier_store_config.as_ref();
 			let primary_store = Arc::new(DynStoreWrapper(kv_store));
 			let mut tier_store = TierStore::new(primary_store, Arc::clone(&logger));
+			let tier_index_exists = PathBuf::from(&self.config.storage_dir_path)
+				.join(io::sqlite_store::SQLITE_TIER_INDEX_DB_FILE_NAME)
+				.exists();
+			let tier_index_required = ts_config
+				.map(|config| {
+					config.ephemeral_storage_dir_path.is_some()
+						|| config.backup_storage_dir_path.is_some()
+				})
+				.unwrap_or(false);
+			if tier_index_required || tier_index_exists {
+				let index_store = runtime
+					.block_on(setup_index_store(self.config.storage_dir_path.clone().into()))
+					.map_err(|e| {
+						log_error!(logger, "Failed to setup tier-store index: {}", e);
+						BuildError::KVStoreSetupFailed
+					})?;
+				tier_store.set_index_store(index_store);
+			}
 			if let Some(config) = ts_config {
 				if let Some(ephemeral_storage_dir_path) = config.ephemeral_storage_dir_path.as_ref()
 				{
-					let index_store = runtime
-						.block_on(setup_index_store(self.config.storage_dir_path.clone().into()))
-						.map_err(|e| {
-							log_error!(logger, "Failed to setup tier-store index: {}", e);
-							BuildError::KVStoreSetupFailed
-						})?;
 					let ephemeral_store = SqliteStore::new(
 						ephemeral_storage_dir_path.clone(),
 						Some(io::sqlite_store::SQLITE_EPHEMERAL_DB_FILE_NAME.to_string()),
@@ -1011,7 +1023,6 @@ impl NodeBuilder {
 						BuildError::KVStoreSetupFailed
 					})?;
 					let ephemeral_store: Arc<DynStore> = Arc::new(DynStoreWrapper(ephemeral_store));
-					tier_store.set_index_store(index_store);
 					tier_store.set_ephemeral_store(ephemeral_store);
 				}