feat(outbound-redis): Add DEL command (#828)

* chore(outbound-redis): Add basic integration test

Signed-off-by: Konstantin Shabanov <[email protected]>

* feat(outbound-redis): Add DEL command

Closes: #801.

Signed-off-by: Konstantin Shabanov <[email protected]>

* feat(go-sdk): Add INCR and DEL support

Signed-off-by: Konstantin Shabanov <[email protected]>

Signed-off-by: Konstantin Shabanov <[email protected]>

Author: etehtsea

Cargo.toml

@@ -75,6 +75,7 @@ vergen = { version = "7", default-features = false, features = [ "build", "git"
 [features]
 default = []
 e2e-tests = []
+outbound-redis-tests = []
 
 [workspace]
 members = [

Makefile

@@ -9,7 +9,7 @@ build:
 .PHONY: test
 test: lint test-unit test-integration
 
-.PHONY: lint 
+.PHONY: lint
 lint:
 	cargo clippy --all-targets --all-features -- -D warnings
 	cargo fmt --all -- --check
@@ -30,9 +30,13 @@ test-integration:
 
 .PHONY: test-e2e
 test-e2e:
-	RUST_LOG=$(LOG_LEVEL) cargo test --test integration --features e2e-tests --no-fail-fast  -- integration_tests::test_dependencies --nocapture 
+	RUST_LOG=$(LOG_LEVEL) cargo test --test integration --features e2e-tests --no-fail-fast  -- integration_tests::test_dependencies --nocapture
 	RUST_LOG=$(LOG_LEVEL) cargo test --test integration --features e2e-tests --no-fail-fast -- --skip integration_tests::test_dependencies --nocapture
 
+.PHONY: test-outbound-redis
+test-outbound-redis:
+	RUST_LOG=$(LOG_LEVEL) cargo test --test integration --features outbound-redis-tests --no-fail-fast -- --nocapture
+
 .PHONY: test-sdk-go
 test-sdk-go:
 	$(MAKE) -C sdk/go test

build.rs

@@ -8,6 +8,7 @@ use cargo_target_dep::build_target_dep;
 
 const RUST_HTTP_INTEGRATION_TEST: &str = "tests/http/simple-spin-rust";
 const RUST_HTTP_INTEGRATION_ENV_TEST: &str = "tests/http/headers-env-routes-test";
+const RUST_OUTBOUND_REDIS_INTEGRATION_TEST: &str = "tests/outbound-redis/http-rust-outbound-redis";
 
 fn main() {
     println!("cargo:rerun-if-changed=build.rs");
@@ -53,6 +54,7 @@ error: the `wasm32-wasi` target is not installed
 
     cargo_build(RUST_HTTP_INTEGRATION_TEST);
     cargo_build(RUST_HTTP_INTEGRATION_ENV_TEST);
+    cargo_build(RUST_OUTBOUND_REDIS_INTEGRATION_TEST);
 
     let mut config = vergen::Config::default();
     *config.git_mut().sha_kind_mut() = vergen::ShaKind::Short;

crates/abi-conformance/src/outbound_redis.rs

@@ -41,6 +41,14 @@ pub struct RedisReport {
     /// function with the arguments \["127.0.0.1", "foo"\] and expect `ok(42)` as the result.  The host will assert
     /// that said function is called exactly once with the specified arguments.
     pub incr: Result<(), String>,
+
+    /// Result of the Redis `DEL` test
+    ///
+    /// The guest module should expect a call according to [`super::InvocationStyle`] with \["outbound-redis-del",
+    /// "127.0.0.1", "foo"\] as arguments. The module should call the host-implemented `outbound-redis::del`
+    /// function with the arguments \["127.0.0.1", \["foo"\]\] and expect `ok(0)` as the result.  The host will assert
+    /// that said function is called exactly once with the specified arguments.
+    pub del: Result<(), String>,
 }
 
 wit_bindgen_wasmtime::export!("../../wit/ephemeral/outbound-redis.wit");
@@ -51,6 +59,7 @@ pub(super) struct OutboundRedis {
     set_set: HashSet<(String, String, Vec<u8>)>,
     get_map: HashMap<(String, String), Vec<u8>>,
     incr_map: HashMap<(String, String), i64>,
+    del_map: HashMap<(String, String), i64>,
 }
 
 impl outbound_redis::OutboundRedis for OutboundRedis {
@@ -93,6 +102,12 @@ impl outbound_redis::OutboundRedis for OutboundRedis {
             .map(|value| value + 1)
             .ok_or(outbound_redis::Error::Error)
     }
+
+    fn del(&mut self, address: &str, keys: Vec<&str>) -> Result<i64, outbound_redis::Error> {
+        self.del_map
+            .remove(&(address.into(), format!("{keys:?}")))
+            .ok_or(outbound_redis::Error::Error)
+    }
 }
 
 pub(super) fn test(store: &mut Store<Context>, pre: &InstancePre<Context>) -> Result<RedisReport> {
@@ -183,5 +198,26 @@ pub(super) fn test(store: &mut Store<Context>, pre: &InstancePre<Context>) -> Re
                 },
             )
         },
+
+        del: {
+            store.data_mut().outbound_redis.del_map.insert(
+                ("127.0.0.1".into(), format!("{:?}", vec!["foo".to_owned()])),
+                0,
+            );
+
+            super::run_command(
+                store,
+                pre,
+                &["outbound-redis-del", "127.0.0.1", "foo"],
+                |store| {
+                    ensure!(
+                        store.data().outbound_redis.del_map.is_empty(),
+                        "expected module to call `outbound-redis::del` exactly once"
+                    );
+
+                    Ok(())
+                },
+            )
+        },
     })
 }

crates/outbound-redis/src/lib.rs

@@ -41,6 +41,12 @@ impl outbound_redis::OutboundRedis for OutboundRedis {
         let value = conn.incr(key, 1).await.map_err(log_error)?;
         Ok(value)
     }
+
+    async fn del(&mut self, address: &str, keys: Vec<&str>) -> Result<i64, Error> {
+        let conn = self.get_conn(address).await.map_err(log_error)?;
+        let value = conn.del(keys).await.map_err(log_error)?;
+        Ok(value)
+    }
 }
 
 impl OutboundRedis {

examples/tinygo-outbound-redis/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"net/http"
 	"os"
+	"strconv"
 
 	spin_http "github.com/fermyon/spin/sdk/go/http"
 	"github.com/fermyon/spin/sdk/go/redis"
@@ -41,6 +42,25 @@ func init() {
 		} else {
 			w.Write([]byte("mykey value was: "))
 			w.Write(payload)
+			w.Write([]byte("\n"))
+		}
+
+		// incr `spin-go-incr` by 1
+		if payload, err := redis.Incr(addr, "spin-go-incr"); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		} else {
+			w.Write([]byte("spin-go-incr value: "))
+			w.Write([]byte(strconv.FormatInt(payload, 10)))
+			w.Write([]byte("\n"))
+		}
+
+		// delete `spin-go-incr` and `mykey`
+		if payload, err := redis.Del(addr, []string{"spin-go-incr", "mykey", "non-existing-key"}); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		} else {
+			w.Write([]byte("deleted keys num: "))
+			w.Write([]byte(strconv.FormatInt(payload, 10)))
+			w.Write([]byte("\n"))
 		}
 	})
 }

sdk/go/redis/internals.go

@@ -66,6 +66,45 @@ func set(addr, key string, payload []byte) error {
 	return toErr(err)
 }
 
+func incr(addr, key string) (int64, error) {
+	caddr := redisStr(addr)
+	ckey := redisStr(key)
+
+	var cpayload C.int64_t
+
+	defer func() {
+		C.outbound_redis_string_free(&caddr)
+		C.outbound_redis_string_free(&ckey)
+	}()
+
+	err := C.outbound_redis_incr(&caddr, &ckey, &cpayload)
+	return int64(cpayload), toErr(err)
+}
+
+func del(addr string, keys []string) (int64, error) {
+	caddr := redisStr(addr)
+	ckeys := redisListStr(keys)
+
+	var cpayload C.int64_t
+
+	defer func() {
+		C.outbound_redis_string_free(&caddr)
+		C.outbound_redis_list_string_free(&ckeys)
+	}()
+
+	err := C.outbound_redis_del(&caddr, &ckeys, &cpayload)
+	return int64(cpayload), toErr(err)
+}
+
+func redisListStr(xs []string) C.outbound_redis_list_string_t {
+	var cxs []C.outbound_redis_string_t
+
+	for i := 0; i < len(xs); i++ {
+		cxs = append(cxs, redisStr(xs[i]))
+	}
+	return C.outbound_redis_list_string_t{ptr: &cxs[0], len: C.size_t(len(cxs))}
+}
+
 func redisStr(x string) C.outbound_redis_string_t {
 	return C.outbound_redis_string_t{ptr: C.CString(x), len: C.size_t(len(x))}
 }

sdk/go/redis/outbound-redis.c

@@ -56,9 +56,22 @@ typedef struct {
     outbound_redis_error_t err;
   } val;
 } outbound_redis_expected_payload_error_t;
+typedef struct {
+  bool is_err;
+  union {
+    int64_t ok;
+    outbound_redis_error_t err;
+  } val;
+} outbound_redis_expected_s64_error_t;
+void outbound_redis_list_string_free(outbound_redis_list_string_t *ptr) {
+  for (size_t i = 0; i < ptr->len; i++) {
+    outbound_redis_string_free(&ptr->ptr[i]);
+  }
+  canonical_abi_free(ptr->ptr, ptr->len * 8, 4);
+}
 
-__attribute__((aligned(4)))
-static uint8_t RET_AREA[12];
+__attribute__((aligned(8)))
+static uint8_t RET_AREA[16];
 __attribute__((import_module("outbound-redis"), import_name("publish")))
 void __wasm_import_outbound_redis_publish(int32_t, int32_t, int32_t, int32_t, int32_t, int32_t, int32_t);
 outbound_redis_error_t outbound_redis_publish(outbound_redis_string_t *address, outbound_redis_string_t *channel, outbound_redis_payload_t *payload) {
@@ -123,3 +136,47 @@ outbound_redis_error_t outbound_redis_set(outbound_redis_string_t *address, outb
     }
   }return expected.is_err ? expected.val.err : -1;
 }
+__attribute__((import_module("outbound-redis"), import_name("incr")))
+void __wasm_import_outbound_redis_incr(int32_t, int32_t, int32_t, int32_t, int32_t);
+outbound_redis_error_t outbound_redis_incr(outbound_redis_string_t *address, outbound_redis_string_t *key, int64_t *ret0) {
+  int32_t ptr = (int32_t) &RET_AREA;
+  __wasm_import_outbound_redis_incr((int32_t) (*address).ptr, (int32_t) (*address).len, (int32_t) (*key).ptr, (int32_t) (*key).len, ptr);
+  outbound_redis_expected_s64_error_t expected;
+  switch ((int32_t) (*((uint8_t*) (ptr + 0)))) {
+    case 0: {
+      expected.is_err = false;
+      
+      expected.val.ok = *((int64_t*) (ptr + 8));
+      break;
+    }
+    case 1: {
+      expected.is_err = true;
+      
+      expected.val.err = (int32_t) (*((uint8_t*) (ptr + 8)));
+      break;
+    }
+  }*ret0 = expected.val.ok;
+  return expected.is_err ? expected.val.err : -1;
+}
+__attribute__((import_module("outbound-redis"), import_name("del")))
+void __wasm_import_outbound_redis_del(int32_t, int32_t, int32_t, int32_t, int32_t);
+outbound_redis_error_t outbound_redis_del(outbound_redis_string_t *address, outbound_redis_list_string_t *keys, int64_t *ret0) {
+  int32_t ptr = (int32_t) &RET_AREA;
+  __wasm_import_outbound_redis_del((int32_t) (*address).ptr, (int32_t) (*address).len, (int32_t) (*keys).ptr, (int32_t) (*keys).len, ptr);
+  outbound_redis_expected_s64_error_t expected;
+  switch ((int32_t) (*((uint8_t*) (ptr + 0)))) {
+    case 0: {
+      expected.is_err = false;
+      
+      expected.val.ok = *((int64_t*) (ptr + 8));
+      break;
+    }
+    case 1: {
+      expected.is_err = true;
+      
+      expected.val.err = (int32_t) (*((uint8_t*) (ptr + 8)));
+      break;
+    }
+  }*ret0 = expected.val.ok;
+  return expected.is_err ? expected.val.err : -1;
+}

sdk/go/redis/outbound-redis.h

@@ -24,9 +24,16 @@ extern "C"
     size_t len;
   } outbound_redis_payload_t;
   void outbound_redis_payload_free(outbound_redis_payload_t *ptr);
+  typedef struct {
+    outbound_redis_string_t *ptr;
+    size_t len;
+  } outbound_redis_list_string_t;
+  void outbound_redis_list_string_free(outbound_redis_list_string_t *ptr);
   outbound_redis_error_t outbound_redis_publish(outbound_redis_string_t *address, outbound_redis_string_t *channel, outbound_redis_payload_t *payload);
   outbound_redis_error_t outbound_redis_get(outbound_redis_string_t *address, outbound_redis_string_t *key, outbound_redis_payload_t *ret0);
   outbound_redis_error_t outbound_redis_set(outbound_redis_string_t *address, outbound_redis_string_t *key, outbound_redis_payload_t *value);
+  outbound_redis_error_t outbound_redis_incr(outbound_redis_string_t *address, outbound_redis_string_t *key, int64_t *ret0);
+  outbound_redis_error_t outbound_redis_del(outbound_redis_string_t *address, outbound_redis_list_string_t *keys, int64_t *ret0);
   #ifdef __cplusplus
 }
 #endif

sdk/go/redis/redis.go

@@ -36,3 +36,16 @@ func Get(addr, key string) ([]byte, error) {
 func Set(addr, key string, payload []byte) error {
 	return set(addr, key, payload)
 }
+
+// Increments the number stored at key by one. If the key does not exist,
+// it is set to 0 before performing the operation. An error is returned if
+// the key contains a value of the wrong type or contains a string that can not
+// be represented as integer.
+func Incr(addr, key string) (int64, error) {
+	return incr(addr, key)
+}
+
+// Removes the specified keys. A key is ignored if it does not exist.
+func Del(addr string, keys []string) (int64, error) {
+	return del(addr, keys)
+}

tests/integration.rs

@@ -602,6 +602,30 @@ mod integration_tests {
         }
     }
 
+    #[cfg(feature = "outbound-redis-tests")]
+    mod outbound_pg_tests {
+        use super::*;
+
+        const RUST_OUTBOUND_REDIS_INTEGRATION_TEST: &str =
+            "tests/outbound-redis/http-rust-outbound-redis";
+
+        #[tokio::test]
+        async fn test_outbound_redis_rust_local() -> Result<()> {
+            let s = SpinTestController::with_manifest(
+                &format!(
+                    "{}/{}",
+                    RUST_OUTBOUND_REDIS_INTEGRATION_TEST, DEFAULT_MANIFEST_LOCATION
+                ),
+                &[],
+                None,
+            )
+            .await?;
+
+            assert_status(&s, "/test", 204).await?;
+            Ok(())
+        }
+    }
+
     #[tokio::test]
     async fn test_simple_rust_local() -> Result<()> {
         let s = SpinTestController::with_manifest(

tests/outbound-redis/http-rust-outbound-redis/.cargo/config.toml

@@ -0,0 +1,2 @@
+[build]
+target = "wasm32-wasi"

tests/outbound-redis/http-rust-outbound-redis/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "http-rust-outbound-redis"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+crate-type = [ "cdylib" ]
+
+[dependencies]
+# Useful crate to handle errors.
+anyhow = "1"
+# Crate to simplify working with bytes.
+bytes = "1"
+# General-purpose crate with common HTTP types.
+http = "0.2"
+# The Spin SDK.
+spin-sdk = { path = "../../../sdk/rust" }
+# Crate that generates Rust Wasm bindings from a WebAssembly interface.
+wit-bindgen-rust = { git = "https://github.com/bytecodealliance/wit-bindgen", rev = "cb871cfa1ee460b51eb1d144b175b9aab9c50aba" }
+
+[workspace]

tests/outbound-redis/http-rust-outbound-redis/spin.toml

@@ -0,0 +1,15 @@
+spin_version = "1"
+authors = ["Fermyon Engineering <[email protected]>"]
+name = "rust-outbound-redis-example"
+trigger = { type = "http", base = "/" }
+version = "0.1.0"
+
+[[component]]
+environment = { REDIS_ADDRESS = "redis://127.0.0.1:6379" }
+id = "outbound-redis"
+source = "target/wasm32-wasi/release/http_rust_outbound_redis.wasm"
+[component.trigger]
+route = "/test"
+[component.build]
+command = "cargo build --target wasm32-wasi --release"
+