fixup! Implement tiered storage

Respect inner sync store semantics in TierStore
and FFI store bridging

Here we rework TierStore's `KVStoreSync` implementation
so that synchronous calls no longer route through TierStore's
own runtime and async helpers. Instead, sync operations now
delegate through dedicated sync internal helpers that call
the underlying stores' `KVStoreSync` methods directly.

This preserves backend-specific sync behavior for wrapped stores,
including stores that intentionally use a different execution
strategy for synchronous operations than for async ones.
In particular, it avoids forcing inner stores through TierStore's
runtime when the inner store already defines its own sync
semantics.

As part of this change:
- remove the `Runtime` dependency from `TierStore`
- update `TierStore::new` and its call sites accordingly
- add sync internal helpers for read/write/remove/list routing
- add sync primary/backup helpers mirroring the async dual-store
  behavior
- share ephemeral routing and primary/backup result handling across
  sync and async paths to reduce duplication

We also update `FfiDynStore`'s sync write/remove path to avoid
directly calling `tokio::sync::Mutex::blocking_lock` on a Tokio
runtime worker thread. The sync path now goes through a small
helper that uses `block_in_place` when already inside a runtime,
preventing panics in runtime-backed test scenarios such as
`start_stop_reinit`, and many others.

The net effect is:
- TierStore sync operations now honor the inner store's sync
  contract
- FFI-backed sync writes/removals no longer panic when invoked
  from within a Tokio runtime
- TierStore construction and tests are simplified by dropping
  the now-unused runtime plumbing

Author: enigbe

src/builder.rs

@@ -838,8 +838,7 @@ impl NodeBuilder {
 		};
 
 		let ts_config = self.tier_store_config.as_ref();
-		let mut tier_store =
-			TierStore::new(primary_store, Arc::clone(&runtime), Arc::clone(&logger));
+		let mut tier_store = TierStore::new(primary_store, Arc::clone(&logger));
 		if let Some(config) = ts_config {
 			config.ephemeral.as_ref().map(|s| tier_store.set_ephemeral_store(Arc::clone(s)));
 			config.backup.as_ref().map(|s| tier_store.set_backup_store(Arc::clone(s)));

src/ffi/types.rs

@@ -451,8 +451,7 @@ impl FfiDynStoreInner {
 			"write",
 		)?;
 
-		let res = {
-			let mut last_written_version = inner_lock_ref.blocking_lock();
+		let res = self.with_blocking_lock(&inner_lock_ref, |last_written_version| {
 			if version <= *last_written_version {
 				Ok(())
 			} else {
@@ -463,7 +462,7 @@ impl FfiDynStoreInner {
 						*last_written_version = version;
 					})
 			}
-		};
+		});
 
 		self.clean_locks(&inner_lock_ref, locking_key);
 		res
@@ -504,8 +503,7 @@ impl FfiDynStoreInner {
 			"remove",
 		)?;
 
-		let res = {
-			let mut last_written_version = inner_lock_ref.blocking_lock();
+		let res = self.with_blocking_lock(&inner_lock_ref, |last_written_version| {
 			if version <= *last_written_version {
 				Ok(())
 			} else {
@@ -516,7 +514,7 @@ impl FfiDynStoreInner {
 						*last_written_version = version;
 					})
 			}
-		};
+		});
 
 		self.clean_locks(&inner_lock_ref, locking_key);
 		res
@@ -582,6 +580,20 @@ impl FfiDynStoreInner {
 			outer_lock.remove(&locking_key);
 		}
 	}
+
+	fn with_blocking_lock<T, F: FnOnce(&mut u64) -> T>(
+		&self, inner_lock_ref: &Arc<tokio::sync::Mutex<u64>>, f: F,
+	) -> T {
+		if tokio::runtime::Handle::try_current().is_ok() {
+			tokio::task::block_in_place(|| {
+				let mut last_written_version = inner_lock_ref.blocking_lock();
+				f(&mut last_written_version)
+			})
+		} else {
+			let mut last_written_version = inner_lock_ref.blocking_lock();
+			f(&mut last_written_version)
+		}
+	}
 }
 
 #[async_trait::async_trait]