feat: add native async provider API

CHANGELOG.md

@@ -9,6 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Rust integrations can implement the awaitable `AsyncProvider` trait or use
+  `AsyncProviderAdapter` to call existing providers without blocking async tasks.
+  `SyncProviderAdapter` exposes native async backends to the existing sync API;
+  `ProviderBase` carries their shared configuration and identity.
+  Async SDK, HTTP, and external IPC providers implement the trait directly,
+  retaining batch reads, metadata, authentication, and retry behavior. The
+  async URI factory selects native implementations automatically.
+
 - Bitwarden Password Manager can load its `session` credential from another
   provider, such as keyring, instead of requiring `BW_SESSION` in the shell.
 

Cargo.toml

@@ -92,6 +92,7 @@ kube = { version = "4.2", default-features = false, features = ["client", "rustl
 k8s-openapi = { version = "0.28", features = ["v1_36"] }
 json-patch = "4.2.0"
 async-trait = "0.1"
+futures-util = "0.3"
 tokio-util = { version = "0.7", features = ["rt"] }
 zeroize = { version = "1.8", features = ["derive"] }
 secretspec-ipc = { version = "0.21.1", path = "./secretspec-ipc" }

conformance/ipc/runner/src/bin/ipc-provider-endpoint-rust.rs

@@ -47,6 +47,7 @@ struct StoredValue {
 struct MemoryProvider {
     values: Mutex<HashMap<String, StoredValue>>,
     errors_seen: Mutex<HashSet<String>>,
+    async_attempts: Mutex<HashMap<String, usize>>,
     crash_on_get_once: Option<PathBuf>,
 }
 
@@ -107,6 +108,16 @@ impl MemoryProvider {
         }
     }
 
+    // Faults used by direct async client tests. Counters let tests distinguish
+    // a surfaced lost acknowledgement from an automatically replayed mutation.
+    fn async_attempt(&self, operation: &str, key: &str) -> usize {
+        let key = key.rsplit('/').next().unwrap();
+        let mut counts = self.async_attempts.lock().unwrap();
+        let count = counts.entry(format!("{operation}/{key}")).or_default();
+        *count += 1;
+        *count
+    }
+
     fn one_shot_error(&self, key: &str) -> Option<RpcError> {
         let kind = if key.ends_with("/__INTERACTION_REQUIRED__") {
             ErrorKind::InteractionRequired
@@ -198,6 +209,24 @@ impl ProviderHandler for MemoryProvider {
         address: Address,
     ) -> RpcResult<Option<ProvidedSecret>> {
         let key = Self::key(&address);
+        if key.ends_with("/__ASYNC_COUNTS__") {
+            let counts = serde_json::to_string(&*self.async_attempts.lock().unwrap()).unwrap();
+            return Ok(Some(ProvidedSecret::new(counts, None)));
+        }
+        if key.ends_with("/__ASYNC_RETRY_ONCE__") {
+            if self.async_attempt("get", &key) == 1 {
+                return Err(RpcError::unavailable(Some(1000)));
+            }
+            return Ok(Some(ProvidedSecret::new("recovered".into(), None)));
+        }
+        if key.ends_with("/__ASYNC_RETRY_ALWAYS__") {
+            self.async_attempt("get", &key);
+            return Err(RpcError::unavailable(Some(1)));
+        }
+        if key.ends_with("/__ASYNC_DENIED__") {
+            self.async_attempt("get", &key);
+            return Err(RpcError::new(ErrorKind::PermissionDenied));
+        }
         if key.ends_with("/__BLOCK__") {
             context.cancellation.cancelled().await;
             return Err(RpcError::new(ErrorKind::Cancelled));
@@ -254,6 +283,10 @@ impl ProviderHandler for MemoryProvider {
             return Err(error);
         }
         let key = Self::key(&address);
+        let lost_ack = key.ends_with("/__ASYNC_SET__");
+        if lost_ack {
+            self.async_attempt("set", &key);
+        }
         let secret_expires_at_unix_ms = key
             .ends_with("/SECRET_EXPIRY")
             .then(|| {
@@ -285,6 +318,9 @@ impl ProviderHandler for MemoryProvider {
                 revision,
             },
         );
+        if lost_ack {
+            return Err(RpcError::unavailable(Some(1)));
+        }
         Ok(())
     }
 
@@ -295,25 +331,37 @@ impl ProviderHandler for MemoryProvider {
         value: SecretValue,
         ttl_ms: u64,
     ) -> RpcResult<()> {
+        let key = Self::key(&address);
+        let lost_ack = key.ends_with("/__ASYNC_SET_EXPIRING__");
+        if lost_ack {
+            self.async_attempt("set_expiring", &key);
+        }
         self.values.lock().unwrap().insert(
-            Self::key(&address),
+            key,
             StoredValue {
                 value: value.expose().to_string(),
                 storage_expires_at: Some(Instant::now() + Duration::from_millis(ttl_ms)),
                 secret_expires_at_unix_ms: None,
                 revision: None,
             },
         );
+        if lost_ack {
+            return Err(RpcError::unavailable(Some(1)));
+        }
         Ok(())
     }
 
     async fn delete(&self, _context: RequestContext, address: Address) -> RpcResult<bool> {
-        Ok(self
-            .values
-            .lock()
-            .unwrap()
-            .remove(&Self::key(&address))
-            .is_some())
+        let key = Self::key(&address);
+        let lost_ack = key.ends_with("/__ASYNC_DELETE__");
+        if lost_ack {
+            self.async_attempt("delete", &key);
+        }
+        let removed = self.values.lock().unwrap().remove(&key).is_some();
+        if lost_ack {
+            return Err(RpcError::unavailable(Some(1)));
+        }
+        Ok(removed)
     }
 
     async fn clear(&self, _context: RequestContext, params: ClearParams) -> RpcResult<usize> {
@@ -448,6 +496,7 @@ async fn main() {
     let handler = MemoryProvider {
         values: Mutex::new(HashMap::new()),
         errors_seen: Mutex::new(HashSet::new()),
+        async_attempts: Mutex::new(HashMap::new()),
         crash_on_get_once,
     };
     if serve_provider(

conformance/ipc/runner/tests/async_provider.rs

@@ -0,0 +1,215 @@
+use secretspec::{
+    Address, AsyncProvider, DiscoveryContext, ExternalProvider, Provider, ProviderCredentialBroker,
+    ProviderCredentialPrincipal, ProviderCredentialRequest, ProviderEndpoint, RetryPolicy,
+    SecretBytes, SecretSpecError,
+};
+use secretspec_ipc::error::ErrorKind;
+use std::cell::Cell;
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+struct NoHostCredentials;
+impl ProviderCredentialBroker for NoHostCredentials {
+    fn get(
+        &self,
+        _principal: &ProviderCredentialPrincipal,
+        _request: &ProviderCredentialRequest,
+    ) -> secretspec::Result<Option<SecretBytes>> {
+        Ok(None)
+    }
+}
+
+fn provider(attempts: u32) -> (Option<tempfile::TempDir>, ExternalProvider) {
+    let endpoint = PathBuf::from(env!("CARGO_BIN_EXE_ipc-provider-endpoint-rust"));
+    #[cfg(windows)]
+    let (endpoint, directory) = {
+        let directory = tempfile::tempdir().unwrap();
+        secretspec::windows_security::make_path_private(directory.path()).unwrap();
+        let staged = directory.path().join(endpoint.file_name().unwrap());
+        std::fs::copy(&endpoint, &staged).unwrap();
+        (staged, Some(directory))
+    };
+    #[cfg(not(windows))]
+    let directory = None;
+    let mut provider = ExternalProvider::new(
+        ProviderEndpoint {
+            scheme: "memory".into(),
+            executable: endpoint,
+            arguments: Vec::new(),
+            environment: Vec::new(),
+        },
+        "memory://conformance",
+    )
+    .unwrap();
+    provider.with_credential_broker(Arc::new(NoHostCredentials));
+    Provider::set_retry_policy(&mut provider, RetryPolicy::new(attempts).unwrap());
+    (directory, provider)
+}
+
+fn address(key: &str) -> Address<'_> {
+    Address::convention("async-tests", "default", key)
+}
+
+async fn counts(provider: &dyn AsyncProvider) -> HashMap<String, usize> {
+    let value = provider
+        .get(address("__ASYNC_COUNTS__"))
+        .await
+        .unwrap()
+        .unwrap();
+    serde_json::from_slice(value.expose_secret()).unwrap()
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn native_ipc_crud_batches_and_metadata() {
+    let (_directory, provider) = provider(3);
+    provider.initialize_async().await.unwrap();
+    let provider: &dyn AsyncProvider = provider.as_async();
+    assert!(provider.supports_read());
+    assert!(provider.supports_delete());
+    let token = address("TOKEN");
+    assert!(provider.get(token).await.unwrap().is_none());
+    provider.check_writable(token).await.unwrap();
+    provider.set(token, &"value".into()).await.unwrap();
+    assert!(provider.exists(token).await.unwrap());
+    assert_eq!(
+        provider.get(token).await.unwrap().unwrap().expose_secret(),
+        b"value"
+    );
+    let requests = [
+        ("first", token),
+        ("alias", token),
+        ("missing", address("MISSING")),
+    ];
+    let batch = provider.get_many(&requests).await.unwrap();
+    assert_eq!(batch.len(), 2);
+    assert_eq!(batch["first"].expose_secret(), b"value");
+    assert_eq!(batch["alias"].expose_secret(), b"value");
+    let present = provider.exists_many(&requests).await.unwrap();
+    assert_eq!(present.len(), 2);
+    assert!(present.contains("first") && present.contains("alias"));
+
+    let expiry = address("SECRET_EXPIRY");
+    provider.set(expiry, &"expiring".into()).await.unwrap();
+    let metadata = provider.get_with_metadata(expiry).await.unwrap().unwrap();
+    assert!(metadata.expires_at_unix_ms.is_some());
+    assert!(metadata.revision.is_some());
+    let batch = provider
+        .get_many_with_metadata(&[("expiry", expiry)])
+        .await
+        .unwrap();
+    assert_eq!(batch["expiry"].revision, metadata.revision);
+    assert_eq!(
+        batch["expiry"].expires_at_unix_ms,
+        metadata.expires_at_unix_ms
+    );
+    provider
+        .set_expiring(token, &"retained".into(), Duration::from_secs(60))
+        .await
+        .unwrap();
+    let retained = provider.get_with_metadata(token).await.unwrap().unwrap();
+    assert_eq!(retained.value.expose_secret(), b"retained");
+    assert!(retained.expires_at_unix_ms.is_none());
+    assert!(
+        provider
+            .reflect(DiscoveryContext::new("async-tests", "default"))
+            .await
+            .unwrap()
+            .contains_key("TOKEN")
+    );
+    provider.check_deletable(token).await.unwrap();
+    assert!(provider.delete(token).await.unwrap());
+    assert!(!provider.delete(token).await.unwrap());
+    assert!(!provider.exists(token).await.unwrap());
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn native_ipc_retry_backoff_yields_to_the_callers_runtime() {
+    let (_directory, provider) = provider(3);
+    provider.initialize_async().await.unwrap();
+    let provider: &dyn AsyncProvider = provider.as_async();
+    // The endpoint requests one second of backoff. A blocking sleep would keep
+    // this current-thread runtime from polling the heartbeat until it finishes.
+    let started = Instant::now();
+    let finished = Cell::new(false);
+    let read = async {
+        let result = provider.get(address("__ASYNC_RETRY_ONCE__")).await;
+        finished.set(true);
+        result
+    };
+    let heartbeat = async {
+        let mut previous = started;
+        loop {
+            tokio::time::sleep(Duration::from_millis(20)).await;
+            assert!(previous.elapsed() < Duration::from_millis(750));
+            previous = Instant::now();
+            if finished.get() {
+                break;
+            }
+        }
+    };
+    let (value, ()) = tokio::join!(read, heartbeat);
+    assert!(started.elapsed() >= Duration::from_secs(1));
+    assert_eq!(value.unwrap().unwrap().expose_secret(), b"recovered");
+    assert_eq!(counts(provider).await["get/__ASYNC_RETRY_ONCE__"], 2);
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn native_ipc_obeys_retry_budget_and_does_not_retry_permission_errors() {
+    for attempts in [1, 3] {
+        let (_directory, provider) = provider(attempts);
+        let provider: &dyn AsyncProvider = provider.as_async();
+        let error = provider
+            .get(address("__ASYNC_RETRY_ALWAYS__"))
+            .await
+            .unwrap_err();
+        assert_eq!(error.protocol_kind(), Some(ErrorKind::Unavailable));
+        assert!(matches!(
+            error,
+            SecretSpecError::ProviderTransient {
+                exhausted: true,
+                ..
+            }
+        ));
+        let error = provider.get(address("__ASYNC_DENIED__")).await.unwrap_err();
+        assert_eq!(error.protocol_kind(), Some(ErrorKind::PermissionDenied));
+        let counts = counts(provider).await;
+        assert_eq!(counts["get/__ASYNC_RETRY_ALWAYS__"], attempts as usize);
+        assert_eq!(counts["get/__ASYNC_DENIED__"], 1);
+    }
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn native_ipc_does_not_replay_mutations_after_lost_acknowledgements() {
+    let (_directory, provider) = provider(3);
+    let provider: &dyn AsyncProvider = provider.as_async();
+    let value = "stored once".into();
+    let set = address("__ASYNC_SET__");
+    let error = provider.set(set, &value).await.unwrap_err();
+    assert_eq!(error.protocol_kind(), Some(ErrorKind::Unavailable));
+    assert_eq!(
+        provider.get(set).await.unwrap().unwrap().expose_secret(),
+        b"stored once"
+    );
+    let expiring = address("__ASYNC_SET_EXPIRING__");
+    let error = provider
+        .set_expiring(expiring, &value, Duration::from_secs(60))
+        .await
+        .unwrap_err();
+    assert_eq!(error.protocol_kind(), Some(ErrorKind::Unavailable));
+    assert!(provider.exists(expiring).await.unwrap());
+    let delete = address("__ASYNC_DELETE__");
+    provider.set(delete, &value).await.unwrap();
+    let error = provider.delete(delete).await.unwrap_err();
+    assert_eq!(error.protocol_kind(), Some(ErrorKind::Unavailable));
+    assert!(!provider.exists(delete).await.unwrap());
+    let counts = counts(provider).await;
+    for operation in [
+        "set/__ASYNC_SET__",
+        "set_expiring/__ASYNC_SET_EXPIRING__",
+        "delete/__ASYNC_DELETE__",
+    ] {
+        assert_eq!(counts[operation], 1, "replayed {operation}");
+    }
+}