Async Redis trigger

crates/trigger-redis/src/lib.rs

@@ -7,7 +7,8 @@ use serde::Deserialize;
 use spin_factor_variables::VariablesFactor;
 use spin_factors::RuntimeFactors;
 use spin_trigger::{cli::NoCliArgs, App, Trigger, TriggerApp};
-use spin_world::exports::fermyon::spin::inbound_redis;
+use spin_world::exports::fermyon::spin::inbound_redis as v1;
+use spin_world::exports::spin::redis::inbound_redis as v3;
 use tracing::{instrument, Level};
 
 pub struct RedisTrigger;
@@ -211,14 +212,60 @@ impl<F: RuntimeFactors> Subscriber<F> {
             .await?;
 
         let pre = instance.instance_pre(&store);
-        let guest_indices = inbound_redis::GuestIndices::new(&pre)?;
-        let guest = guest_indices.load(&mut store, &instance)?;
 
-        let payload = msg.get_payload_bytes().to_vec();
+        match HandlerType::from_instance_pre(&pre)? {
+            HandlerType::V1(guest_indices) => {
+                let guest = guest_indices.load(&mut store, &instance)?;
 
-        guest
-            .call_handle_message(&mut store, &payload)
-            .await?
-            .context("Redis handler returned an error")
+                let payload = msg.get_payload_bytes().to_vec();
+
+                guest
+                    .call_handle_message(&mut store, &payload)
+                    .await?
+                    .context("Redis handler returned an error")
+            }
+            HandlerType::V3(guest_indices) => {
+                let guest = guest_indices.load(&mut store, &instance)?;
+
+                let payload = msg.get_payload_bytes().to_vec();
+                let res = std::pin::pin!(store.as_mut().run_concurrent(async |accessor| {
+                    guest.call_handle_message(accessor, payload).await
+                }))
+                .await;
+
+                res.map_err(|e| anyhow::anyhow!("{e}"))
+                    .context("Redis handler returned an error (run_concurrent)")?
+                    .map_err(|e| anyhow::anyhow!("{e}"))
+                    .context("Redis handler returned an error")?
+                    .context("Redis handler returned an error")
+            }
+        }
+    }
+}
+
+/// The type of Redis handler export used by a component.
+pub enum HandlerType /*<S: HandlerState>*/ {
+    V1(v1::GuestIndices),
+    V3(v3::GuestIndices),
+}
+
+impl HandlerType {
+    /// Determine the handler type from the exports of a component.
+    pub fn from_instance_pre<T: 'static>(
+        pre: &spin_factors::wasmtime::component::InstancePre<T>, /*, handler_state: S*/
+    ) -> anyhow::Result<Self> {
+        let mut candidates = Vec::new();
+        if let Ok(indices) = v1::GuestIndices::new(pre) {
+            candidates.push(HandlerType::V1(indices));
+        }
+        if let Ok(indices) = v3::GuestIndices::new(pre) {
+            candidates.push(HandlerType::V3(indices));
+        }
+
+        match candidates.len() {
+            0 => anyhow::bail!("component does not export a Redis interface"),
+            1 => Ok(candidates.pop().unwrap()),
+            _ => anyhow::bail!("component exports multiple Redis interfaces"),
+        }
     }
 }

crates/world/src/lib.rs

@@ -14,6 +14,7 @@ wasmtime::component::bindgen!({
         include spin:up/platform@3.4.0;
         include spin:up/platform@3.7.0;
         include wasi:keyvalue/imports@0.2.0-draft2;
+        export spin:redis/inbound-redis@3.0.0;
     }
     "#,
     path: "../../wit",

tests/integration.rs

@@ -214,6 +214,64 @@ mod integration_tests {
         Ok(())
     }
 
+    #[test]
+    #[cfg(feature = "extern-dependencies-tests")]
+    #[allow(dependency_on_unit_never_type_fallback)]
+    /// Test that basic redis trigger support works
+    fn redis_async_trigger_smoke_test() -> anyhow::Result<()> {
+        use anyhow::Context;
+        use redis::Commands;
+        run_test(
+            "redis-async-trigger-smoke-test",
+            SpinConfig {
+                binary_path: spin_binary(),
+                spin_up_args: Vec::new(),
+                app_type: SpinAppType::Redis,
+            },
+            ServicesConfig::new(vec!["redis"])?,
+            move |env| {
+                let redis_port = env
+                    .services_mut()
+                    .get_port(6379)?
+                    .context("no redis port was exposed by test services")?;
+
+                let mut redis = redis::Client::open(format!("redis://localhost:{redis_port}"))
+                    .context("could not connect to redis in test")?;
+                redis
+                    .publish::<_, _, ()>("my-channel", "msg-from-test")
+                    .context("could not publish test message to redis")?;
+                assert_eventually!(
+                    {
+                        match env.read_file(".spin/logs/hello_stdout.txt") {
+                            Ok(logs) => {
+                                let logs = String::from_utf8_lossy(&logs);
+                                logs.contains("Got message: 'msg-from-test'")
+                            }
+                            Err(e)
+                                if e.downcast_ref()
+                                    .map(|e: &std::io::Error| {
+                                        e.kind() == std::io::ErrorKind::NotFound
+                                    })
+                                    .unwrap_or_default() =>
+                            {
+                                false
+                            }
+                            Err(e) => {
+                                return Err(
+                                    anyhow::anyhow!("could not read stdout file: {e}").into()
+                                );
+                            }
+                        }
+                    },
+                    2
+                );
+                Ok(())
+            },
+        )?;
+
+        Ok(())
+    }
+
     #[test]
     #[cfg(feature = "extern-dependencies-tests")]
     /// Test that basic otel tracing works

tests/test-components/components/redis-async-trigger-smoke-test/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "redis-async-trigger-smoke-test"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+anyhow = "1"
+bytes = "1"
+wit-bindgen = "0.54"

tests/test-components/components/redis-async-trigger-smoke-test/src/lib.rs

@@ -0,0 +1,25 @@
+use std::str::from_utf8;
+
+const KV_KEY: &str = "message";
+
+wit_bindgen::generate!({
+    path: "../../../../wit",
+    world: "spin:up/redis-trigger@3.7.0",
+    generate_all,
+});
+
+struct Guest;
+
+impl exports::spin::redis::inbound_redis::Guest for Guest {
+    async fn handle_message(message: Vec<u8>) -> Result<(), spin::redis::redis::Error> {
+        // Do some async stuff to prove it works
+        let kv = spin::key_value::key_value::Store::open("default".to_string()).await.expect("should have had access to default KV store");
+        kv.set(KV_KEY.to_string(), message.clone()).await.expect("should have set KV entry");
+        let message = kv.get(KV_KEY.to_string()).await.expect("should have read KV entry").expect("KV entry should have existed");
+
+        println!("Got message: '{}'", from_utf8(&message).unwrap_or("<MESSAGE NOT UTF8>"));
+        Ok(())
+    }
+}
+
+export!(Guest);

tests/testcases/redis-async-trigger-smoke-test/spin.toml

@@ -0,0 +1,20 @@
+spin_manifest_version = 2
+
+[application]
+authors = ["Spin Framework Contributors"]
+description = "A simple redis application that exercises the Rust SDK in the current branch"
+name = "redis-async-trigger-smoke-test"
+version = "1.0.0"
+
+[application.trigger.redis]
+address = "redis://localhost:%{port=6379}"
+
+[[trigger.redis]]
+channel = "my-channel"
+component = "hello"
+
+[component.hello]
+source = "%{source=redis-async-trigger-smoke-test}"
+key_value_stores = ["default"]
+[component.hello.build]
+command = "cargo build --target wasm32-wasip2 --release"

wit/deps/spin-redis@3.0.0/redis.wit

@@ -70,3 +70,10 @@ interface redis {
       binary(payload)
   }
 }
+
+interface inbound-redis {
+    use redis.{payload, error};
+
+    // The entrypoint for a Redis handler.
+    handle-message: async func(message: payload) -> result<_, error>;
+}

wit/world.wit

@@ -6,6 +6,12 @@ world http-trigger {
   export wasi:http/handler@0.3.0-rc-2026-03-15;
 }
 
+/// The full world of a guest targeting a redis-trigger
+world redis-trigger {
+  include platform;
+  export spin:redis/inbound-redis@3.0.0;
+}
+
 /// The imports needed for a guest to run on a Spin host
 world platform {
   include wasi:cli/imports@0.2.6;