feat(linkscaler): add interest in LinkdefSet

Signed-off-by: Brooks Townsend <brooks@cosmonic.com>

address PR feedback, fix tests

Signed-off-by: Brooks Townsend <brooks@cosmonic.com>

Author: brooksmtownsend

.gitignore

@@ -1,2 +1,4 @@
 /target
 test/e2e_log/
+
+*.dump

Cargo.lock

@@ -55,6 +55,7 @@ uuid = "1"
 wasmbus-rpc = "0.14"
 wasmcloud-control-interface = "0.28.1"
 semver = { version = "1.0.16", features = ["serde"] }
+regex = "1.9.3"
 base64 = "0.21.2"
 
 [dev-dependencies]

Cargo.toml

@@ -2,10 +2,7 @@ use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
 
-use async_nats::jetstream::{
-    stream::{RetentionPolicy, Stream},
-    Context,
-};
+use async_nats::jetstream::{stream::Stream, Context};
 use clap::Parser;
 use tokio::sync::Semaphore;
 use tracing::log::debug;
@@ -202,9 +199,6 @@ async fn main() -> anyhow::Result<()> {
             "A stream that stores all events coming in on the wasmbus.evt topics in a cluster"
                 .to_string(),
         ),
-        RetentionPolicy::WorkQueue,
-        None,
-        None,
     )
     .await?;
 
@@ -215,20 +209,13 @@ async fn main() -> anyhow::Result<()> {
         COMMAND_STREAM_NAME.to_owned(),
         vec![DEFAULT_COMMANDS_TOPIC.to_owned()],
         Some("A stream that stores all commands for wadm".to_string()),
-        RetentionPolicy::WorkQueue,
-        None,
-        None,
     )
     .await?;
 
-    let status_stream = nats::ensure_stream(
+    let status_stream = nats::ensure_status_stream(
         &context,
         STATUS_STREAM_NAME.to_owned(),
         vec![DEFAULT_STATUS_TOPIC.to_owned()],
-        Some("A stream that stores all status updates for wadm applications".to_string()),
-        RetentionPolicy::Limits,
-        Some(std::time::Duration::from_nanos(0)),
-        Some(10),
     )
     .await?;
 
@@ -249,9 +236,6 @@ async fn main() -> anyhow::Result<()> {
         mirror_stream.to_owned(),
         event_stream_topics.clone(),
         Some("A stream that publishes all events to the same stream".to_string()),
-        RetentionPolicy::WorkQueue,
-        None,
-        None,
     )
     .await?;
 

bin/main.rs

@@ -1,12 +1,11 @@
 use std::path::PathBuf;
-use std::time::Duration;
 
 use anyhow::{anyhow, Result};
 use async_nats::{
     jetstream::{
         self,
         kv::{Config as KvConfig, Store},
-        stream::{Config as StreamConfig, RetentionPolicy, Stream},
+        stream::{Config as StreamConfig, Stream},
         Context,
     },
     Client, ConnectOptions,
@@ -111,19 +110,15 @@ pub async fn ensure_stream(
     name: String,
     subjects: Vec<String>,
     description: Option<String>,
-    retention: RetentionPolicy,
-    max_age: Option<Duration>,
-    max_messages_per_subject: Option<i64>,
 ) -> Result<Stream> {
     context
         .get_or_create_stream(StreamConfig {
             name,
             description,
             num_replicas: 1,
-            retention,
+            retention: async_nats::jetstream::stream::RetentionPolicy::WorkQueue,
             subjects,
-            max_messages_per_subject: max_messages_per_subject.unwrap_or(0),
-            max_age: max_age.unwrap_or(DEFAULT_EXPIRY_TIME),
+            max_age: DEFAULT_EXPIRY_TIME,
             storage: async_nats::jetstream::stream::StorageType::File,
             allow_rollup: false,
             ..Default::default()
@@ -132,6 +127,30 @@ pub async fn ensure_stream(
         .map_err(|e| anyhow::anyhow!("{e:?}"))
 }
 
+pub async fn ensure_status_stream(
+    context: &Context,
+    name: String,
+    subjects: Vec<String>,
+) -> Result<Stream> {
+    context
+        .get_or_create_stream(StreamConfig {
+            name,
+            description: Some(
+                "A stream that stores all status updates for wadm applications".into(),
+            ),
+            num_replicas: 1,
+            allow_direct: true,
+            retention: async_nats::jetstream::stream::RetentionPolicy::Limits,
+            max_messages_per_subject: 10,
+            subjects,
+            max_age: std::time::Duration::from_nanos(0),
+            storage: async_nats::jetstream::stream::StorageType::File,
+            ..Default::default()
+        })
+        .await
+        .map_err(|e| anyhow::anyhow!("{e:?}"))
+}
+
 /// A helper that ensures that the notify stream exists
 pub async fn ensure_notify_stream(
     context: &Context,

bin/nats.rs

@@ -85,7 +85,7 @@ impl<S: ReadStore + Send + Sync + Clone> Scaler for SimpleActorScaler<S> {
             config,
             store: self.store.clone(),
             id: self.id.clone(),
-            status: StatusInfo::compensating("Cleaning up"),
+            status: StatusInfo::compensating(""),
         };
 
         cleanerupper.compute_actor_commands(&self.store).await
@@ -112,7 +112,7 @@ impl<S: ReadStore + Send + Sync> SimpleActorScaler<S> {
                 model_name,
             },
             id,
-            status: StatusInfo::compensating("Initializing simplescaler"),
+            status: StatusInfo::compensating(""),
         }
     }
 

src/scaler/simplescaler.rs

@@ -7,7 +7,9 @@ use tracing::{instrument, trace};
 
 use crate::{
     commands::{Command, DeleteLinkdef, PutLinkdef},
-    events::{Event, LinkdefDeleted, ProviderHealthCheckPassed, ProviderHealthCheckStatus},
+    events::{
+        Event, LinkdefDeleted, LinkdefSet, ProviderHealthCheckPassed, ProviderHealthCheckStatus,
+    },
     model::TraitProperty,
     scaler::Scaler,
     server::StatusInfo,
@@ -94,6 +96,7 @@ where
                 self.reconcile().await
             }
             Event::LinkdefDeleted(LinkdefDeleted { linkdef })
+            | Event::LinkdefSet(LinkdefSet { linkdef })
                 if linkdef.contract_id == self.config.provider_contract_id
                     && linkdef.actor_id == self.actor_id().await.unwrap_or_default()
                     && linkdef.provider_id == self.provider_id().await.unwrap_or_default()
@@ -152,7 +155,9 @@ where
             // };
 
             let commands = if !exists {
-                *self.status.write().await = StatusInfo::compensating("Creating link definition");
+                *self.status.write().await = StatusInfo::compensating(&format!(
+                    "Putting link definition between {actor_id} and {provider_id}"
+                ));
                 vec![Command::PutLinkdef(PutLinkdef {
                     actor_id: actor_id.to_owned(),
                     provider_id: provider_id.to_owned(),
@@ -227,7 +232,7 @@ impl<S: ReadStore + Send + Sync, L: LinkSource> LinkScaler<S, L> {
             },
             ctl_client,
             id,
-            status: RwLock::new(StatusInfo::compensating("Initializing")),
+            status: RwLock::new(StatusInfo::compensating("")),
         }
     }
 

src/scaler/spreadscaler/link.rs

@@ -257,7 +257,7 @@ impl<S: ReadStore + Send + Sync + Clone> Scaler for ActorSpreadScaler<S> {
             spread_requirements,
             actor_id: self.actor_id.clone(),
             id: self.id.clone(),
-            status: RwLock::new(StatusInfo::compensating("Cleaning up")),
+            status: RwLock::new(StatusInfo::compensating("")),
         };
 
         cleanerupper.reconcile().await
@@ -286,7 +286,7 @@ impl<S: ReadStore + Send + Sync> ActorSpreadScaler<S> {
                 model_name,
             },
             id,
-            status: RwLock::new(StatusInfo::compensating("Initializing")),
+            status: RwLock::new(StatusInfo::compensating("")),
         }
     }
 

src/scaler/spreadscaler/mod.rs

@@ -225,6 +225,9 @@ impl<S: ReadStore + Send + Sync + Clone> Scaler for ProviderSpreadScaler<S> {
                             .collect::<Vec<Command>>();
 
                         if commands.len() < num_to_start {
+                            // NOTE(brooksmtownsend): We're reporting status for the entire spreadscaler here, not just this individual
+                            // command, so we want to consider the providers that are already running for the spread over the 
+                            // total expected count.
                             let msg = format!("Could not satisfy spread {} for {}, {}/{} eligible hosts found.", spread.name, self.config.provider_reference, running_for_spread.len(), count);
                             spread_status.push(StatusInfo::failed(&msg));
                         }
@@ -268,7 +271,7 @@ impl<S: ReadStore + Send + Sync + Clone> Scaler for ProviderSpreadScaler<S> {
             spread_requirements,
             provider_id: self.provider_id.clone(),
             id: self.id.clone(),
-            status: RwLock::new(StatusInfo::compensating("Cleaning up")),
+            status: RwLock::new(StatusInfo::compensating("")),
         };
 
         cleanerupper.reconcile().await
@@ -288,7 +291,7 @@ impl<S: ReadStore + Send + Sync> ProviderSpreadScaler<S> {
             provider_id: OnceCell::new(),
             config,
             id,
-            status: RwLock::new(StatusInfo::compensating("Initializing")),
+            status: RwLock::new(StatusInfo::compensating("")),
         }
     }
 

src/scaler/spreadscaler/provider.rs

@@ -1,6 +1,8 @@
 use async_nats::{jetstream::stream::Stream, Client, Message};
 use base64::{engine::general_purpose::STANDARD as B64decoder, Engine};
+use regex::Regex;
 use serde_json::json;
+use tokio::sync::OnceCell;
 use tracing::{debug, error, instrument, log::warn, trace};
 
 use crate::{
@@ -19,6 +21,8 @@ use super::{
     StatusResponse, StatusResult, UndeployModelRequest, VersionInfo, VersionResponse,
 };
 
+static MANIFEST_NAME_REGEX: OnceCell<Regex> = OnceCell::const_new();
+
 pub(crate) struct Handler<P> {
     pub(crate) store: ModelStorage,
     pub(crate) client: Client,
@@ -79,7 +83,12 @@ impl<P: Publisher> Handler<P> {
         }
 
         let manifest_name = manifest.metadata.name.trim().to_string();
-        if manifest_name.contains(' ') || manifest_name.contains('.') {
+        if !MANIFEST_NAME_REGEX
+            // SAFETY: We know this is valid Regex
+            .get_or_init(|| async { Regex::new(r"^[-\w]+$").unwrap() })
+            .await
+            .is_match(&manifest_name)
+        {
             self.send_error(
                 msg.reply,
                 format!(
@@ -784,6 +793,8 @@ impl<P: Publisher> Handler<P> {
     }
 
     async fn get_manifest_status(&self, lattice_id: &str, name: &str) -> Option<StatusInfo> {
+        // NOTE(brooksmtownsend): We're getting the last raw message instead of direct get here
+        // to ensure we fetch the latest message from the cluster leader.
         match self
             .status_stream
             .get_last_raw_message_by_subject(&format!("wadm.status.{lattice_id}.{name}",))
@@ -799,3 +810,30 @@ impl<P: Publisher> Handler<P> {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    #[tokio::test]
+    async fn manifest_name_regex_works() {
+        let regex = super::MANIFEST_NAME_REGEX
+            .get_or_init(|| async { regex::Regex::new(r"^[-\w]+$").unwrap() })
+            .await;
+
+        // Acceptable manifest names
+        let word = "mymanifest";
+        let word_with_dash = "my-manifest";
+        let word_with_underscore = "my_manifest";
+        let word_with_numbers = "mymanifest-v2-v3-final";
+
+        assert!(regex.is_match(word));
+        assert!(regex.is_match(word_with_dash));
+        assert!(regex.is_match(word_with_underscore));
+        assert!(regex.is_match(word_with_numbers));
+
+        // Not acceptable manifest names
+        let word_with_period = "my.manifest";
+        let word_with_space = "my manifest";
+        assert!(!regex.is_match(word_with_period));
+        assert!(!regex.is_match(word_with_space));
+    }
+}

src/server/handlers.rs

@@ -724,7 +724,7 @@ where
             .publish_status(data.manifest.metadata.name.as_ref(), status)
             .await
         {
-            warn!(error = ?e, "Failed to set status to compensating");
+            warn!("Failed to set manifest status: {e:}");
         };
 
         // Now handle the result from reconciliation

src/workers/event.rs

@@ -100,16 +100,16 @@ impl LinkSource for wasmcloud_control_interface::Client {
 pub struct StatusPublisher<Pub> {
     publisher: Pub,
     // Topic prefix, e.g. wadm.status.default
-    topic: String,
+    topic_prefix: String,
 }
 
 impl<Pub> StatusPublisher<Pub> {
     /// Creates an new status publisher configured with the given publisher that will send to the
-    /// specified topic
-    pub fn new(publisher: Pub, topic: &str) -> StatusPublisher<Pub> {
+    /// manifest status topic using the given prefix
+    pub fn new(publisher: Pub, topic_prefix: &str) -> StatusPublisher<Pub> {
         StatusPublisher {
             publisher,
-            topic: topic.to_owned(),
+            topic_prefix: topic_prefix.to_owned(),
         }
     }
 }
@@ -120,7 +120,7 @@ impl<Pub: Publisher> StatusPublisher<Pub> {
         self.publisher
             .publish(
                 serde_json::to_vec(&status)?,
-                Some(&format!("{}.{name}", self.topic)),
+                Some(&format!("{}.{name}", self.topic_prefix)),
             )
             .await
     }

src/workers/event_helpers.rs

@@ -509,7 +509,7 @@ pub fn check_providers(
     Ok(())
 }
 
-pub(crate) async fn get_manifest_status(
+pub async fn get_manifest_status(
     stream: &Stream,
     lattice_id: &str,
     name: &str,