feat(*): Update to newest control client

This bumps the control client version and removes some of the previous work
that was in rc.1

Signed-off-by: Taylor Thomas <taylor@cosmonic.com>

Author: thomastaylor312

Cargo.lock

@@ -1,7 +1,7 @@
 [package]
 name = "wadm"
 description = "wasmCloud Application Deployment Manager: A tool for running Wasm applications in wasmCloud"
-version = "0.8.0-rc.1"
+version = "0.8.0-rc.2"
 edition = "2021"
 authors = ["wasmCloud Team"]
 keywords = ["webassembly", "wasmcloud", "wadm"]
@@ -63,7 +63,7 @@ tracing-subscriber = { version = "0.3.7", features = [
     "json",
 ], optional = true }
 uuid = "1"
-wasmcloud-control-interface = "0.30"
+wasmcloud-control-interface = "0.31"
 
 [dev-dependencies]
 serial_test = "1"

Cargo.toml

@@ -1,70 +1,44 @@
 //! A module for connection pools and generators. This is needed because control interface clients
 //! (and possibly other things like nats connections in the future) are lattice scoped or need
 //! different credentials
-use std::marker::PhantomData;
-
-use wasmcloud_control_interface::{
-    kv::{Build, KvStore},
-    Client, ClientBuilder,
-};
+use wasmcloud_control_interface::{Client, ClientBuilder};
 
 // Copied from https://github.com/wasmCloud/control-interface-client/blob/main/src/broker.rs#L1, not public
 const DEFAULT_TOPIC_PREFIX: &str = "wasmbus.ctl";
 
-#[derive(Debug, Default, Clone)]
-pub struct ControlClientConfig {
-    /// The jetstream domain to use for the clients
-    pub js_domain: Option<String>,
-    /// The topic prefix to use for operations
-    pub topic_prefix: Option<String>,
-}
-
 /// A client constructor for wasmCloud control interface clients, identified by a lattice ID
 // NOTE: Yes, this sounds java-y. Deal with it.
 #[derive(Clone)]
-pub struct ControlClientConstructor<T> {
+pub struct ControlClientConstructor {
     client: async_nats::Client,
-    config: ControlClientConfig,
-    marker: PhantomData<T>,
+    /// The topic prefix to use for operations
+    topic_prefix: Option<String>,
 }
 
-impl<T: KvStore + Build + Clone> ControlClientConstructor<T> {
-    /// Creates a new client pool that is all backed using the same NATS client. The given NATS
-    /// client should be using credentials that can access all desired lattices.
+impl ControlClientConstructor {
+    /// Creates a new client pool that is all backed using the same NATS client and an optional
+    /// topic prefix. The given NATS client should be using credentials that can access all desired
+    /// lattices.
     pub fn new(
         client: async_nats::Client,
-        config: ControlClientConfig,
-    ) -> ControlClientConstructor<T> {
+        topic_prefix: Option<String>,
+    ) -> ControlClientConstructor {
         ControlClientConstructor {
             client,
-            config,
-            marker: PhantomData,
+            topic_prefix,
         }
     }
 
     /// Get the client for the given lattice ID
-    pub async fn get_connection(
-        &self,
-        id: &str,
-        multitenant_prefix: Option<&str>,
-    ) -> anyhow::Result<Client<T>> {
-        let builder: ClientBuilder<T> =
-            ClientBuilder::new_generic(self.client.clone()).lattice_prefix(id);
-        let builder = if let Some(domain) = self.config.js_domain.as_deref() {
-            builder.js_domain(domain)
-        } else {
-            builder
-        };
+    pub fn get_connection(&self, id: &str, multitenant_prefix: Option<&str>) -> Client {
+        let builder = ClientBuilder::new(self.client.clone()).lattice_prefix(id);
 
         let builder = builder.topic_prefix(topic_prefix(
             multitenant_prefix,
-            self.config.topic_prefix.as_deref(),
+            self.topic_prefix.as_deref(),
         ));
 
-        builder
-            .build()
-            .await
-            .map_err(|e| anyhow::anyhow!("Error building client for {id}: {e:?}"))
+        builder.build()
     }
 }
 

bin/connections.rs

@@ -27,8 +27,7 @@ mod logging;
 mod nats;
 mod observer;
 
-use connections::{ControlClientConfig, ControlClientConstructor};
-use wasmcloud_control_interface::kv::{CachedKvStore, DirectKvStore};
+use connections::ControlClientConstructor;
 
 const EVENT_STREAM_NAME: &str = "wadm_events";
 const COMMAND_STREAM_NAME: &str = "wadm_commands";
@@ -181,22 +180,7 @@ async fn main() -> anyhow::Result<()> {
     .await?;
 
     // TODO: We will probably need to set up all the flags (like lattice prefix and topic prefix) down the line
-    let connection_pool: ControlClientConstructor<CachedKvStore> = ControlClientConstructor::new(
-        client.clone(),
-        ControlClientConfig {
-            js_domain: args.domain.clone(),
-            topic_prefix: None,
-        },
-    );
-
-    let direct_connection_pool: ControlClientConstructor<DirectKvStore> =
-        ControlClientConstructor::new(
-            client.clone(),
-            ControlClientConfig {
-                js_domain: args.domain,
-                topic_prefix: None,
-            },
-        );
+    let connection_pool = ControlClientConstructor::new(client.clone(), None);
 
     let trimmer: &[_] = &['.', '>', '*'];
 
@@ -286,7 +270,7 @@ async fn main() -> anyhow::Result<()> {
     let event_worker_creator = EventWorkerCreator {
         state_store: state_storage.clone(),
         manifest_store: manifest_storage.clone(),
-        pool: connection_pool,
+        pool: connection_pool.clone(),
         command_topic_prefix: DEFAULT_COMMANDS_TOPIC.trim_matches(trimmer).to_owned(),
         publisher: context.clone(),
         notify_stream,
@@ -302,7 +286,7 @@ async fn main() -> anyhow::Result<()> {
     debug!("Creating command consumer manager");
 
     let command_worker_creator = CommandWorkerCreator {
-        pool: direct_connection_pool,
+        pool: connection_pool,
     };
     let commands_manager: ConsumerManager<CommandConsumer> = ConsumerManager::new(
         permit_pool.clone(),
@@ -363,30 +347,29 @@ async fn main() -> anyhow::Result<()> {
 
 #[derive(Clone)]
 struct CommandWorkerCreator {
-    pool: ControlClientConstructor<DirectKvStore>,
+    pool: ControlClientConstructor,
 }
 
 #[async_trait::async_trait]
 impl WorkerCreator for CommandWorkerCreator {
-    type Output = CommandWorker<DirectKvStore>;
+    type Output = CommandWorker;
 
     async fn create(
         &self,
         lattice_id: &str,
         multitenant_prefix: Option<&str>,
     ) -> anyhow::Result<Self::Output> {
-        self.pool
-            .get_connection(lattice_id, multitenant_prefix)
-            .await
-            .map(CommandWorker::new)
+        let client = self.pool.get_connection(lattice_id, multitenant_prefix);
+
+        Ok(CommandWorker::new(client))
     }
 }
 
 #[derive(Clone)]
 struct EventWorkerCreator<StateStore> {
     state_store: StateStore,
     manifest_store: async_nats::jetstream::kv::Store,
-    pool: ControlClientConstructor<CachedKvStore>,
+    pool: ControlClientConstructor,
     command_topic_prefix: String,
     publisher: Context,
     notify_stream: Stream,
@@ -397,49 +380,38 @@ impl<StateStore> WorkerCreator for EventWorkerCreator<StateStore>
 where
     StateStore: wadm::storage::Store + Send + Sync + Clone + 'static,
 {
-    type Output =
-        EventWorker<StateStore, wasmcloud_control_interface::Client<CachedKvStore>, Context>;
+    type Output = EventWorker<StateStore, wasmcloud_control_interface::Client, Context>;
 
     async fn create(
         &self,
         lattice_id: &str,
         multitenant_prefix: Option<&str>,
     ) -> anyhow::Result<Self::Output> {
-        match self
-            .pool
-            .get_connection(lattice_id, multitenant_prefix)
-            .await
-        {
-            Ok(client) => {
-                let command_publisher = CommandPublisher::new(
-                    self.publisher.clone(),
-                    &format!("{}.{lattice_id}", self.command_topic_prefix),
-                );
-                let status_publisher = StatusPublisher::new(
-                    self.publisher.clone(),
-                    &format!("wadm.status.{lattice_id}"),
-                );
-                let manager = ScalerManager::new(
-                    self.publisher.clone(),
-                    self.notify_stream.clone(),
-                    lattice_id,
-                    multitenant_prefix,
-                    self.state_store.clone(),
-                    self.manifest_store.clone(),
-                    command_publisher.clone(),
-                    status_publisher.clone(),
-                    client.clone(),
-                )
-                .await?;
-                Ok(EventWorker::new(
-                    self.state_store.clone(),
-                    client,
-                    command_publisher,
-                    status_publisher,
-                    manager,
-                ))
-            }
-            Err(e) => Err(e),
-        }
+        let client = self.pool.get_connection(lattice_id, multitenant_prefix);
+        let command_publisher = CommandPublisher::new(
+            self.publisher.clone(),
+            &format!("{}.{lattice_id}", self.command_topic_prefix),
+        );
+        let status_publisher =
+            StatusPublisher::new(self.publisher.clone(), &format!("wadm.status.{lattice_id}"));
+        let manager = ScalerManager::new(
+            self.publisher.clone(),
+            self.notify_stream.clone(),
+            lattice_id,
+            multitenant_prefix,
+            self.state_store.clone(),
+            self.manifest_store.clone(),
+            command_publisher.clone(),
+            status_publisher.clone(),
+            client.clone(),
+        )
+        .await?;
+        Ok(EventWorker::new(
+            self.state_store.clone(),
+            client,
+            command_publisher,
+            status_publisher,
+            manager,
+        ))
     }
 }

bin/main.rs

@@ -1,5 +1,5 @@
 use tracing::{instrument, trace};
-use wasmcloud_control_interface::{kv::KvStore, CtlOperationAck};
+use wasmcloud_control_interface::CtlOperationAck;
 
 use crate::{
     commands::*,
@@ -14,19 +14,19 @@ use super::insert_managed_annotations;
 
 /// A worker implementation for handling incoming commands
 #[derive(Clone)]
-pub struct CommandWorker<T: Clone> {
-    client: wasmcloud_control_interface::Client<T>,
+pub struct CommandWorker {
+    client: wasmcloud_control_interface::Client,
 }
 
-impl<T: Clone> CommandWorker<T> {
+impl CommandWorker {
     /// Creates a new command worker with the given connection pool.
-    pub fn new(ctl_client: wasmcloud_control_interface::Client<T>) -> CommandWorker<T> {
+    pub fn new(ctl_client: wasmcloud_control_interface::Client) -> CommandWorker {
         CommandWorker { client: ctl_client }
     }
 }
 
 #[async_trait::async_trait]
-impl<T: KvStore + Clone + Send + Sync> Worker for CommandWorker<T> {
+impl Worker for CommandWorker {
     type Message = Command;
 
     #[instrument(level = "trace", skip_all)]

src/workers/command.rs

@@ -2,7 +2,7 @@ use std::collections::{BTreeMap, HashMap};
 use std::fmt::Debug;
 
 use tracing::{instrument, warn};
-use wasmcloud_control_interface::{kv::KvStore, HostInventory, LinkDefinition};
+use wasmcloud_control_interface::{HostInventory, LinkDefinition};
 
 use crate::{commands::Command, publisher::Publisher, server::StatusInfo, APP_SPEC_ANNOTATION};
 
@@ -43,7 +43,7 @@ pub trait LinkSource {
 }
 
 #[async_trait::async_trait]
-impl<T: KvStore + Clone + Send + Sync> ClaimsSource for wasmcloud_control_interface::Client<T> {
+impl ClaimsSource for wasmcloud_control_interface::Client {
     async fn get_claims(&self) -> anyhow::Result<HashMap<String, Claims>> {
         Ok(self
             .get_claims()
@@ -72,7 +72,7 @@ impl<T: KvStore + Clone + Send + Sync> ClaimsSource for wasmcloud_control_interf
 }
 
 #[async_trait::async_trait]
-impl<T: KvStore + Clone + Send + Sync> InventorySource for wasmcloud_control_interface::Client<T> {
+impl InventorySource for wasmcloud_control_interface::Client {
     async fn get_inventory(&self, host_id: &str) -> anyhow::Result<HostInventory> {
         Ok(self
             .get_host_inventory(host_id)
@@ -86,7 +86,7 @@ impl<T: KvStore + Clone + Send + Sync> InventorySource for wasmcloud_control_int
 // the KV store for updates. This would allow us to not have to fetch every time we need to get
 // links
 #[async_trait::async_trait]
-impl<T: KvStore + Clone + Send + Sync> LinkSource for wasmcloud_control_interface::Client<T> {
+impl LinkSource for wasmcloud_control_interface::Client {
     async fn get_links(&self) -> anyhow::Result<Vec<LinkDefinition>> {
         self.query_links()
             .await

src/workers/event_helpers.rs

@@ -38,7 +38,7 @@ services:
 
   # Have hosts in two different accounts
   wasmcloud_east:
-    image: wasmcloud/wasmcloud:0.78.0
+    image: wasmcloud/wasmcloud:0.79.0
     depends_on:
       - nats-leaf-a
     deploy:
@@ -51,7 +51,7 @@ services:
       WASMCLOUD_CLUSTER_SEED: SCAOGJWX53TGI4233T6GAXWYWBIB5ZDGPTCO6ODJQYELS52YCQCBQSRPA4
       HOST_region: us-brooks-east
   wasmcloud_west:
-    image: wasmcloud/wasmcloud:0.78.0
+    image: wasmcloud/wasmcloud:0.79.0
     depends_on:
       - nats-leaf-b
     deploy:

test/docker-compose-e2e-multitenant.yaml

@@ -5,7 +5,7 @@ services:
     ports:
       - 4222:4222
   wasmcloud:
-    image: wasmcloud/wasmcloud:0.78.0
+    image: wasmcloud/wasmcloud:0.79.0
     depends_on:
       - nats
     deploy: