Skip to content

More async interfaces #3428

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
itowlson merged 3 commits into from
Mar 24, 2026
Merged

More async interfaces #3428

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6c6a6ff
Embetter KV errors
itowlson Mar 17, 2026
4162373
Async variables
itowlson Mar 17, 2026
2684e6b
MQTT async
itowlson Mar 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

100 changes: 57 additions & 43 deletions crates/factor-key-value/src/host.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,7 +235,11 @@ impl spin_core::wasmtime::component::HasData for KeyValueDispatch {
type Data<'a> = &'a mut KeyValueDispatch;
}

impl v3::Host for KeyValueDispatch {}
impl v3::Host for KeyValueDispatch {
fn convert_error(&mut self, err: v3::Error) -> anyhow::Result<v3::Error> {
Ok(err)
}
}

impl v3::HostStore for KeyValueDispatch {
async fn drop(&mut self, store: Resource<v3::Store>) -> Result<()> {
Expand All @@ -248,19 +252,19 @@ impl v3::HostStoreWithStore for crate::KeyValueFactorData {
async fn open<T>(
accessor: &Accessor<T, Self>,
label: String,
) -> Result<Result<Resource<v3::Store>, v3::Error>> {
) -> Result<Resource<v3::Store>, v3::Error> {
let (allowed, manager) = accessor.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
(host.allowed_stores.contains(&label), host.manager.clone())
});

if !allowed {
return Ok(Err(v3::Error::AccessDenied));
return Err(v3::Error::AccessDenied);
}

let store = manager.get(&label).await?;
store.after_open().await?;
let store = manager.get(&label).await.map_err(to_v3_err)?;
store.after_open().await.map_err(to_v3_err)?;

let rsrc = accessor.with(|mut access| {
let host = access.get();
Expand All @@ -270,87 +274,97 @@ impl v3::HostStoreWithStore for crate::KeyValueFactorData {
.map_err(|()| v3::Error::StoreTableFull)
});

Ok(rsrc)
rsrc
}

async fn get<T>(
accessor: &Accessor<T, Self>,
store: Resource<v3::Store>,
key: String,
) -> Result<Result<Option<Vec<u8>>, v3::Error>> {
let store = accessor.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})?;
Ok(store
) -> Result<Option<Vec<u8>>, v3::Error> {
let store = accessor
.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})
.map_err(|_| v3::Error::NoSuchStore)?;
store
.get(&key, MAX_HOST_BUFFERED_BYTES)
.await
.map_err(to_v3_err)
.map_err(track_error_on_span_v3))
.map_err(track_error_on_span_v3)
}

async fn set<T>(
accessor: &Accessor<T, Self>,
store: Resource<v3::Store>,
key: String,
value: Vec<u8>,
) -> Result<Result<(), v3::Error>> {
let store = accessor.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})?;
Ok(store
) -> Result<(), v3::Error> {
let store = accessor
.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})
.map_err(|_| v3::Error::NoSuchStore)?;
store
.set(&key, &value)
.await
.map_err(to_v3_err)
.map_err(track_error_on_span_v3))
.map_err(track_error_on_span_v3)
}

async fn delete<T>(
accessor: &Accessor<T, Self>,
store: Resource<v3::Store>,
key: String,
) -> Result<Result<(), v3::Error>> {
let store = accessor.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})?;
Ok(store
) -> Result<(), v3::Error> {
let store = accessor
.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})
.map_err(|_| v3::Error::NoSuchStore)?;
store
.delete(&key)
.await
.map_err(to_v3_err)
.map_err(track_error_on_span_v3))
.map_err(track_error_on_span_v3)
}

async fn exists<T>(
accessor: &Accessor<T, Self>,
store: Resource<v3::Store>,
key: String,
) -> Result<Result<bool, v3::Error>> {
let store = accessor.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})?;
Ok(store
) -> Result<bool, v3::Error> {
let store = accessor
.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})
.map_err(|_| v3::Error::NoSuchStore)?;
store
.exists(&key)
.await
.map_err(to_v3_err)
.map_err(track_error_on_span_v3))
.map_err(track_error_on_span_v3)
}

async fn get_keys<T>(
accessor: &Accessor<T, Self>,
store: Resource<v3::Store>,
) -> Result<(StreamReader<String>, FutureReader<Result<(), v3::Error>>)> {
let store = accessor.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})?;
let store = accessor
.with(|mut access| {
let host = access.get();
host.otel.reparent_tracing_span();
host.get_store(store).cloned()
})
.map_err(|_| v3::Error::NoSuchStore)?;

let (keys_rx, err_rx) = store.get_keys_async(MAX_HOST_BUFFERED_BYTES).await;

Expand Down
21 changes: 21 additions & 0 deletions crates/factor-outbound-mqtt/src/allowed_hosts.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
use std::sync::Arc;

use anyhow::Result;
use spin_factor_outbound_networking::config::allowed_hosts::OutboundAllowedHosts;

#[derive(Clone)]
pub struct AllowedHostChecker {
allowed_hosts: Arc<OutboundAllowedHosts>,
}

impl AllowedHostChecker {
pub fn new(allowed_hosts: OutboundAllowedHosts) -> Self {
Self {
allowed_hosts: Arc::new(allowed_hosts),
}
}

pub async fn is_address_allowed(&self, address: &str) -> Result<bool> {
self.allowed_hosts.check_url(address, "mqtt").await
}
}
Loading
Loading