fixup! Add per-key version-locked writes to FfiDynStore

Route FfiDynStore sync calls through the async bridge

Make `FfiDynStore` async-first for Rust-side store access
so sync and async callers share the same ordered mutation
path.

`FfiDynStore` exists to preserve per-key write ordering
guarantees across the FFI boundary regardless of how a
foreign store chooses to implement its sync and async methods.
Previously, the sync path used `blocking_lock()` on a Tokio
mutex, which could panic when invoked from within a Tokio runtime.
An earlier attempt to patch that by introducing a per-instance
runtime also created a runtime-drop hazard in async contexts.

This change makes the async path canonical and bridges Rust-side
sync calls through it using a shared process-wide Tokio runtime.

Concretely:
- add a shared `OnceLock`-backed Tokio runtime for the sync bridge
- add `run_sync_via_async` to execute Rust-side sync calls through
  the async implementation
- route sync `read`/`write`/`remove`/`list` through the async
  internal methods
- remove the old sync internal methods and the `blocking_lock()`-based
  path
- document that Rust-side sync calls may invoke the foreign async methods
  rather than the foreign sync methods directly

This preserves shared write-ordering guarantees across sync and
async callers while avoiding runtime-sensitive blocking behavior
and per-instance runtime teardown issues.

Author: enigbe

src/ffi/io.rs

@@ -8,7 +8,7 @@ use std::collections::HashMap;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, OnceLock};
 
 use crate::{
 	io::utils::check_namespace_key_validity, DynStoreTrait, DynStoreWrapper, SyncAndAsyncKVStore,
@@ -114,6 +114,23 @@ impl std::fmt::Display for IOError {
 
 /// FFI-safe version of [`DynStoreTrait`].
 ///
+/// Foreign implementations must provide both synchronous and asynchronous store
+/// methods so they can be used across the Rust and language-binding surfaces.
+///
+/// Internally, [`FfiDynStore`] treats the asynchronous write/remove path as the
+/// canonical implementation for ordered mutations. Its synchronous methods are
+/// bridged through that async path to preserve per-key write ordering guarantees
+/// across both sync and async callers.
+///
+/// As a result, Rust-side synchronous calls routed through [`FfiDynStore`] may
+/// invoke the async methods of a foreign implementation rather than its sync
+/// methods directly.
+///
+/// Note: Foreign implementations are free to make either the sync or async methods
+/// their primary implementation and have the other delegate to it. [`FfiDynStore`]
+/// does not assume which side is primary, but its Rust-side sync bridge executes
+/// through the async interface.
+///
 /// [`DynStoreTrait`]: crate::types::DynStoreTrait
 #[uniffi::export(with_foreign)]
 #[async_trait::async_trait]
@@ -145,6 +162,16 @@ pub trait FfiDynStoreTrait: Send + Sync {
 	) -> Result<Vec<String>, IOError>;
 }
 
+/// Bridges a foreign [`FfiDynStoreTrait`] implementation into Rust's
+/// [`DynStoreTrait`] while enforcing per-key write ordering.
+///
+/// Ordered writes/removals are coordinated within [`FfiDynStore`] so the
+/// underlying foreign store does not need to provide its own cross-call ordering
+/// guarantees.
+///
+/// The async mutation path is canonical. Sync operations are executed by running
+/// the corresponding async operation on an internal runtime, which preserves the
+/// same ordering guarantees for both sync and async callers.
 #[derive(Clone, uniffi::Object)]
 pub struct FfiDynStore {
 	inner: Arc<FfiDynStoreInner>,
@@ -156,6 +183,7 @@ impl FfiDynStore {
 	#[uniffi::constructor]
 	pub fn from_store(store: Arc<dyn FfiDynStoreTrait>) -> Self {
 		let inner = Arc::new(FfiDynStoreInner::new(store));
+
 		Self { inner, next_write_version: Arc::new(AtomicU64::new(1)) }
 	}
 }
@@ -183,6 +211,34 @@ impl FfiDynStore {
 
 		(inner_lock_ref, version)
 	}
+
+	/// Runs a synchronous Rust-side store call via FfiDynStore's canonical async path.
+	///
+	/// This keeps sync callers aligned with the same ordered mutation logic used by
+	/// async callers and avoids blocking directly on async coordination primitives.
+	///
+	/// If already inside a Tokio runtime, the async operation is run from a
+	/// dedicated OS thread using a shared process-wide runtime.
+	fn run_sync_via_async<T, F>(&self, fut: F) -> Result<T, bitcoin::io::Error>
+	where
+		T: Send + 'static,
+		F: Future<Output = Result<T, bitcoin::io::Error>> + Send + 'static,
+	{
+		let runtime = ffi_dynstore_runtime();
+
+		// If we are already inside a Tokio runtime, avoid blocking the current Tokio runtime
+		// thread directly and run the async operation on a dedicated OS thread.
+		if tokio::runtime::Handle::try_current().is_ok() {
+			std::thread::spawn(move || runtime.block_on(fut)).join().unwrap_or_else(|_| {
+				Err(bitcoin::io::Error::new(
+					bitcoin::io::ErrorKind::Other,
+					"FfiDynStore sync bridge thread panicked",
+				))
+			})
+		} else {
+			runtime.block_on(fut)
+		}
+	}
 }
 
 impl DynStoreTrait for FfiDynStore {
@@ -280,7 +336,12 @@ impl DynStoreTrait for FfiDynStore {
 		let secondary_namespace = secondary_namespace.to_string();
 		let key = key.to_string();
 
-		store.read_internal(primary_namespace, secondary_namespace, key)
+		self.run_sync_via_async(async move {
+			store
+				.read_internal_async(primary_namespace, secondary_namespace, key)
+				.await
+				.map_err(|e| e.into())
+		})
 	}
 
 	fn write(
@@ -295,15 +356,20 @@ impl DynStoreTrait for FfiDynStore {
 		let locking_key = self.build_locking_key(&primary_namespace, &secondary_namespace, &key);
 		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(locking_key.clone());
 
-		store.write_internal(
-			inner_lock_ref,
-			locking_key,
-			version,
-			primary_namespace,
-			secondary_namespace,
-			key,
-			buf,
-		)
+		self.run_sync_via_async(async move {
+			store
+				.write_internal_async(
+					inner_lock_ref,
+					locking_key,
+					version,
+					primary_namespace,
+					secondary_namespace,
+					key,
+					buf,
+				)
+				.await
+				.map_err(|e| e.into())
+		})
 	}
 
 	fn remove(
@@ -318,15 +384,20 @@ impl DynStoreTrait for FfiDynStore {
 		let locking_key = self.build_locking_key(&primary_namespace, &secondary_namespace, &key);
 		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(locking_key.clone());
 
-		store.remove_internal(
-			inner_lock_ref,
-			locking_key,
-			version,
-			primary_namespace,
-			secondary_namespace,
-			key,
-			lazy,
-		)
+		self.run_sync_via_async(async move {
+			store
+				.remove_internal_async(
+					inner_lock_ref,
+					locking_key,
+					version,
+					primary_namespace,
+					secondary_namespace,
+					key,
+					lazy,
+				)
+				.await
+				.map_err(|e| e.into())
+		})
 	}
 
 	fn list(
@@ -337,7 +408,12 @@ impl DynStoreTrait for FfiDynStore {
 		let primary_namespace = primary_namespace.to_string();
 		let secondary_namespace = secondary_namespace.to_string();
 
-		store.list_internal(primary_namespace, secondary_namespace)
+		self.run_sync_via_async(async move {
+			store
+				.list_internal_async(primary_namespace, secondary_namespace)
+				.await
+				.map_err(|e| e.into())
+		})
 	}
 }
 
@@ -366,13 +442,6 @@ impl FfiDynStoreInner {
 			.map_err(|e| e.into())
 	}
 
-	fn read_internal(
-		&self, primary_namespace: String, secondary_namespace: String, key: String,
-	) -> bitcoin::io::Result<Vec<u8>> {
-		check_namespace_key_validity(&primary_namespace, &secondary_namespace, Some(&key), "read")?;
-		self.ffi_store.read(primary_namespace, secondary_namespace, key).map_err(|e| e.into())
-	}
-
 	async fn write_internal_async(
 		&self, inner_lock_ref: Arc<tokio::sync::Mutex<u64>>, locking_key: String, version: u64,
 		primary_namespace: String, secondary_namespace: String, key: String, buf: Vec<u8>,
@@ -397,34 +466,6 @@ impl FfiDynStoreInner {
 		.await
 	}
 
-	fn write_internal(
-		&self, inner_lock_ref: Arc<tokio::sync::Mutex<u64>>, locking_key: String, version: u64,
-		primary_namespace: String, secondary_namespace: String, key: String, buf: Vec<u8>,
-	) -> bitcoin::io::Result<()> {
-		check_namespace_key_validity(
-			&primary_namespace,
-			&secondary_namespace,
-			Some(&key),
-			"write",
-		)?;
-
-		let res = self.with_blocking_lock(&inner_lock_ref, |last_written_version| {
-			if version <= *last_written_version {
-				Ok(())
-			} else {
-				self.ffi_store
-					.write(primary_namespace, secondary_namespace, key, buf)
-					.map_err(|e| e.into())
-					.map(|_| {
-						*last_written_version = version;
-					})
-			}
-		});
-
-		self.clean_locks(&inner_lock_ref, locking_key);
-		res
-	}
-
 	async fn remove_internal_async(
 		&self, inner_lock_ref: Arc<tokio::sync::Mutex<u64>>, locking_key: String, version: u64,
 		primary_namespace: String, secondary_namespace: String, key: String, lazy: bool,
@@ -449,34 +490,6 @@ impl FfiDynStoreInner {
 		.await
 	}
 
-	fn remove_internal(
-		&self, inner_lock_ref: Arc<tokio::sync::Mutex<u64>>, locking_key: String, version: u64,
-		primary_namespace: String, secondary_namespace: String, key: String, lazy: bool,
-	) -> bitcoin::io::Result<()> {
-		check_namespace_key_validity(
-			&primary_namespace,
-			&secondary_namespace,
-			Some(&key),
-			"remove",
-		)?;
-
-		let res = self.with_blocking_lock(&inner_lock_ref, |last_written_version| {
-			if version <= *last_written_version {
-				Ok(())
-			} else {
-				self.ffi_store
-					.remove(primary_namespace, secondary_namespace, key, lazy)
-					.map_err(|e| <IOError as Into<bitcoin::io::Error>>::into(e))
-					.map(|_| {
-						*last_written_version = version;
-					})
-			}
-		});
-
-		self.clean_locks(&inner_lock_ref, locking_key);
-		res
-	}
-
 	async fn list_internal_async(
 		&self, primary_namespace: String, secondary_namespace: String,
 	) -> bitcoin::io::Result<Vec<String>> {
@@ -487,13 +500,6 @@ impl FfiDynStoreInner {
 			.map_err(|e| e.into())
 	}
 
-	fn list_internal(
-		&self, primary_namespace: String, secondary_namespace: String,
-	) -> bitcoin::io::Result<Vec<String>> {
-		check_namespace_key_validity(&primary_namespace, &secondary_namespace, None, "list")?;
-		self.ffi_store.list(primary_namespace, secondary_namespace).map_err(|e| e.into())
-	}
-
 	async fn execute_locked_write<
 		F: Future<Output = Result<(), bitcoin::io::Error>>,
 		FN: FnOnce() -> F,
@@ -537,20 +543,6 @@ 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]
@@ -620,3 +612,18 @@ impl<T: SyncAndAsyncKVStore + Send + Sync + 'static> From<T> for FfiDynStore {
 		Self::from_store(Arc::new(DynStoreWrapper(store)))
 	}
 }
+
+/// Returns the process-wide runtime used to bridge Rust-side synchronous
+/// `FfiDynStore` calls through the canonical async store path.
+fn ffi_dynstore_runtime() -> &'static tokio::runtime::Runtime {
+	static RUNTIME: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
+
+	RUNTIME.get_or_init(|| {
+		tokio::runtime::Builder::new_multi_thread()
+			.enable_all()
+			.worker_threads(2)
+			.max_blocking_threads(2)
+			.build()
+			.expect("Failed to build FfiDynStore runtime")
+	})
+}