TRY SPAWN

Author: tnull

src/io/vss_store.rs

@@ -44,6 +44,7 @@ type CustomRetryPolicy = FilteredRetryPolicy<
 >;
 
 const INTERNAL_RUNTIME_WORKERS: usize = 2;
+const VSS_IO_TIMEOUT: Duration = Duration::from_secs(5);
 
 /// A [`KVStoreSync`] implementation that writes to and reads from a [VSS](https://github.com/lightningdevkit/vss-server/blob/main/README.md) backend.
 pub struct VssStore {
@@ -120,7 +121,14 @@ impl KVStoreSync for VssStore {
 		let inner = Arc::clone(&self.inner);
 		let fut =
 			async move { inner.read_internal(primary_namespace, secondary_namespace, key).await };
-		tokio::task::block_in_place(move || internal_runtime.block_on(fut))
+		let spawned_fut = internal_runtime.spawn(async move {
+			tokio::time::timeout(VSS_IO_TIMEOUT, fut).await.map_err(|_| {
+				let msg = "VssStore::read timed out";
+				Error::new(ErrorKind::Other, msg)
+			})
+		});
+		tokio::task::block_in_place(move || internal_runtime.block_on(spawned_fut))
+			.expect("We should always finish")?
 	}
 
 	fn write(
@@ -150,7 +158,14 @@ impl KVStoreSync for VssStore {
 				)
 				.await
 		};
-		tokio::task::block_in_place(move || internal_runtime.block_on(fut))
+		let spawned_fut = internal_runtime.spawn(async move {
+			tokio::time::timeout(VSS_IO_TIMEOUT, fut).await.map_err(|_| {
+				let msg = "VssStore::write timed out";
+				Error::new(ErrorKind::Other, msg)
+			})
+		});
+		tokio::task::block_in_place(move || internal_runtime.block_on(spawned_fut))
+			.expect("We should always finish")?
 	}
 
 	fn remove(
@@ -179,7 +194,14 @@ impl KVStoreSync for VssStore {
 				)
 				.await
 		};
-		tokio::task::block_in_place(move || internal_runtime.block_on(fut))
+		let spawned_fut = internal_runtime.spawn(async move {
+			tokio::time::timeout(VSS_IO_TIMEOUT, fut).await.map_err(|_| {
+				let msg = "VssStore::remove timed out";
+				Error::new(ErrorKind::Other, msg)
+			})
+		});
+		tokio::task::block_in_place(move || internal_runtime.block_on(spawned_fut))
+			.expect("We should always finish")?
 	}
 
 	fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> {
@@ -192,7 +214,14 @@ impl KVStoreSync for VssStore {
 		let secondary_namespace = secondary_namespace.to_string();
 		let inner = Arc::clone(&self.inner);
 		let fut = async move { inner.list_internal(primary_namespace, secondary_namespace).await };
-		tokio::task::block_in_place(move || internal_runtime.block_on(fut))
+		let spawned_fut = internal_runtime.spawn(async move {
+			tokio::time::timeout(VSS_IO_TIMEOUT, fut).await.map_err(|_| {
+				let msg = "VssStore::list timed out";
+				Error::new(ErrorKind::Other, msg)
+			})
+		});
+		tokio::task::block_in_place(move || internal_runtime.block_on(spawned_fut))
+			.expect("We should always finish")?
 	}
 }