Fixing broken CI signals

Includes updating the generated test certificates for TLS integration tests and a minor addition for stats

ractor/src/common_test.rs

@@ -12,6 +12,7 @@ use crate::concurrency::sleep;
 use crate::concurrency::Duration;
 use crate::concurrency::Instant;
 
+/// Periodic check for condition
 pub async fn periodic_check<F>(check: F, timeout: Duration)
 where
     F: Fn() -> bool,
@@ -28,6 +29,7 @@ where
     assert!(check(), "Periodic check failed.\n{:?}", backtrace);
 }
 
+/// Periodic check of Future for condition
 pub async fn periodic_async_check<F, Fut>(check: F, timeout: Duration)
 where
     F: Fn() -> Fut,

ractor/src/factory/factoryimpl.rs

@@ -225,6 +225,9 @@ where
     dead_mans_switch: Option<DeadMansSwitchConfiguration>,
     capacity_controller: Option<Box<dyn WorkerCapacityController>>,
     lifecycle_hooks: Option<Box<dyn FactoryLifecycleHooks<TKey, TMsg>>>,
+    // Local counter to avoid having to sum over the worker states for more performant metrics capturing
+    // in large worker-count factories
+    processing_messages: usize,
 }
 
 impl<TKey, TMsg, TWorkerStart, TWorker, TRouter, TQueue> Debug
@@ -301,6 +304,7 @@ where
         if let Some(worker) = self.pool.get_mut(&worker_hint).filter(|f| f.is_available()) {
             if let Some(mut job) = self.queue.pop_front() {
                 job.accept();
+                self.processing_messages += 1;
                 worker.enqueue_job(job)?;
             }
         } else {
@@ -315,6 +319,7 @@ where
                 .and_then(|wid| self.pool.get_mut(&wid));
             if let (Some(mut job), Some(worker)) = (self.queue.pop_front(), target_worker) {
                 job.accept();
+                self.processing_messages += 1;
                 worker.enqueue_job(job)?;
             }
         }
@@ -505,6 +510,9 @@ where
             {
                 // workers are busy, we need to queue a job
                 self.maybe_enqueue(busy_job);
+            } else {
+                // message was routed
+                self.processing_messages += 1;
             }
         } else {
             tracing::debug!("Factory is draining but a job was received");
@@ -517,6 +525,10 @@ where
     }
 
     fn worker_finished_job(&mut self, who: WorkerId, key: TKey) -> Result<(), ActorProcessingErr> {
+        if self.processing_messages > 0 {
+            self.processing_messages -= 1;
+        }
+
         let (is_worker_draining, should_drop_worker) = if let Some(worker) = self.pool.get_mut(&who)
         {
             if let Some(job_options) = worker.worker_complete(key)? {
@@ -573,12 +585,12 @@ where
             }
         }
 
+        let qlen = self.queue.len();
+        self.stats.record_queue_depth(&self.factory_name, qlen);
+        self.stats
+            .record_processing_messages_count(&self.factory_name, self.processing_messages);
         self.stats
-            .record_queue_depth(&self.factory_name, self.queue.len());
-        self.stats.record_processing_messages_count(
-            &self.factory_name,
-            self.pool.values().filter(|f| f.is_working()).count(),
-        );
+            .record_in_flight_messages_count(&self.factory_name, self.processing_messages + qlen);
         self.stats
             .record_worker_count(&self.factory_name, self.pool_size);
 
@@ -778,6 +790,7 @@ where
             queue,
             router,
             stats,
+            processing_messages: 0,
         })
     }
 

ractor/src/factory/stats.rs

@@ -51,6 +51,12 @@ pub trait FactoryStatsLayer: Send + Sync + 'static {
     /// Fixed-period recording of the factory's number of processed messages
     fn record_processing_messages_count(&self, factory: &str, count: usize);
 
+    /// Fixed-period recording of the factory's in-flight message count (processing + queued)
+    ///
+    /// Default empty implemention for backwards compatibility
+    #[allow(unused_variables)]
+    fn record_in_flight_messages_count(&self, factory: &str, count: usize) {}
+
     /// Fixed-period recording of the factory's number of workers
     fn record_worker_count(&self, factory: &str, count: usize);
 
@@ -129,6 +135,13 @@ impl FactoryStatsLayer for Option<Arc<dyn FactoryStatsLayer>> {
         }
     }
 
+    /// Fixed-period recording of the factory's in-flight message count (processing + queued)
+    fn record_in_flight_messages_count(&self, factory: &str, count: usize) {
+        if let Some(s) = self {
+            s.record_in_flight_messages_count(factory, count);
+        }
+    }
+
     /// Fixed-period recording of the factory's number of workers
     fn record_worker_count(&self, factory: &str, count: usize) {
         if let Some(s) = self {

ractor/src/port/output/mod.rs

@@ -278,7 +278,6 @@ impl OutputPortSubscription {
 ///     subscriber_actor_handle.await.unwrap();
 /// }
 /// ```
-
 pub type OutputPortSubscriber<InputMessage> = Box<dyn OutputPortSubscriberTrait<InputMessage>>;
 /// A trait for subscribing to an [OutputPort]
 pub trait OutputPortSubscriberTrait<I>: Send

ractor_cluster/Cargo.toml

@@ -15,7 +15,8 @@ build = "src/build.rs"
 rust-version = "1.64"
 
 [build-dependencies]
-protobuf-src = "2"
+# protobuf-src = "2"
+protoc-bin-vendored = "3"
 prost-build = { version = "0.13" }
 
 [dependencies]

ractor_cluster/src/build.rs

@@ -12,7 +12,8 @@ const PROTOBUF_BASE_DIRECTORY: &str = "src/protocol";
 const PROTOBUF_FILES: [&str; 4] = ["meta", "node", "auth", "control"];
 
 fn build_protobufs() {
-    std::env::set_var("PROTOC", protobuf_src::protoc());
+    let path = protoc_bin_vendored::protoc_bin_path().expect("Failed to find protoc installation");
+    std::env::set_var("PROTOC", path);
 
     let mut protobuf_files = Vec::with_capacity(PROTOBUF_FILES.len());
 

ractor_cluster/src/macros.rs

@@ -14,6 +14,7 @@
 #[macro_export]
 macro_rules! derive_serialization_for_prost_type {
     {$ty:ty} => {
+        #[allow(non_local_definitions)]
         impl $crate::BytesConvertable for $ty {
             fn into_bytes(self) -> Vec<u8> {
                 <Self as prost::Message>::encode_length_delimited_to_vec(&self)

ractor_cluster_integration_tests/src/tests/encryption.rs

@@ -33,6 +33,7 @@ pub struct EncryptionConfig {
     client_host: Option<String>,
 }
 
+#[allow(elided_named_lifetimes)]
 fn load_certs(path_str: &'static str) -> io::Result<Vec<CertificateDer>> {
     let path = PathBuf::from(path_str);
     let certs: Vec<_> = certs(&mut BufReader::new(File::open(path)?))
@@ -46,6 +47,7 @@ fn load_certs(path_str: &'static str) -> io::Result<Vec<CertificateDer>> {
     }
 }
 
+#[allow(elided_named_lifetimes)]
 fn load_keys(path_str: &'static str) -> io::Result<Vec<PrivateKeyDer>> {
     let path = PathBuf::from(path_str);
     let keys: Vec<PrivateKeyDer> = rsa_private_keys(&mut BufReader::new(File::open(path)?))

ractor_cluster_integration_tests/test-ca/build-a-pki.sh

@@ -7,7 +7,7 @@ mkdir -p rsa/ ecdsa/ eddsa/
 
 openssl req -nodes \
           -x509 \
-          -days 3650 \
+          -days 36500 \
           -newkey rsa:4096 \
           -keyout rsa/ca.key \
           -out rsa/ca.cert \
@@ -58,7 +58,7 @@ openssl req -nodes \
           -out ecdsa/ca.cert \
           -sha256 \
           -batch \
-          -days 3650 \
+          -days 36500 \
           -subj "/CN=ponytown ECDSA CA"
 
 openssl req -nodes \
@@ -67,7 +67,7 @@ openssl req -nodes \
           -out ecdsa/inter.req \
           -sha256 \
           -batch \
-          -days 3000 \
+          -days 36500 \
           -subj "/CN=ponytown ECDSA level 2 intermediate"
 
 openssl req -nodes \
@@ -76,7 +76,7 @@ openssl req -nodes \
           -out ecdsa/end.req \
           -sha256 \
           -batch \
-          -days 2000 \
+          -days 36500 \
           -subj "/CN=testserver.com"
 
 openssl req -nodes \
@@ -85,7 +85,7 @@ openssl req -nodes \
           -out ecdsa/client.req \
           -sha256 \
           -batch \
-          -days 2000 \
+          -days 36500 \
           -subj "/CN=ponytown client"
 
 # eddsa
@@ -100,7 +100,7 @@ openssl req -nodes \
           -out eddsa/ca.cert \
           -sha256 \
           -batch \
-          -days 3650 \
+          -days 36500 \
           -subj "/CN=ponytown EdDSA CA"
 
 openssl genpkey -algorithm Ed25519 -out eddsa/inter.key
@@ -152,7 +152,7 @@ for kt in rsa ecdsa eddsa ; do
             -CA $kt/inter.cert \
             -CAkey $kt/inter.key \
             -sha256 \
-            -days 2000 \
+            -days 36500 \
             -set_serial 456 \
             -extensions v3_end -extfile openssl.cnf
 
@@ -162,7 +162,7 @@ for kt in rsa ecdsa eddsa ; do
             -CA $kt/inter.cert \
             -CAkey $kt/inter.key \
             -sha256 \
-            -days 2000 \
+            -days 36500 \
             -set_serial 789 \
             -extensions v3_client -extfile openssl.cnf
 

ractor_cluster_integration_tests/test-ca/ecdsa/ca.cert

@@ -1,12 +1,12 @@
 -----BEGIN CERTIFICATE-----
-MIIByjCCAVCgAwIBAgIUSA11/39PY7uM9Nc2ITnV1eHzaKYwCgYIKoZIzj0EAwIw
-HDEaMBgGA1UEAwwRcG9ueXRvd24gRUNEU0EgQ0EwHhcNMTkwNjA5MTcxNTEyWhcN
-MjkwNjA2MTcxNTEyWjAcMRowGAYDVQQDDBFwb255dG93biBFQ0RTQSBDQTB2MBAG
-ByqGSM49AgEGBSuBBAAiA2IABLsXWEKs2xXCgW1OcC63pCPjQo0q3VnPc1J24n6m
-Xwxpg398nzR4n3iHcYA0pKgEneBstSOsXOhbNZ09DAvEr3iSc8ByWWntEbWVjY3g
-9Kt6Q6Y1sXGkaUIiP9be5lIQRaNTMFEwHQYDVR0OBBYEFKD72TTU/GXhb3/D1/Z7
-hD/ZG6lKMB8GA1UdIwQYMBaAFKD72TTU/GXhb3/D1/Z7hD/ZG6lKMA8GA1UdEwEB
-/wQFMAMBAf8wCgYIKoZIzj0EAwIDaAAwZQIxAL9FtbNV7i9trxukhakfTvbXCHgE
-2pIOT5r/Vc5kSrPU4vJu2MOJz6X/JCX15IbZlQIwJxYfsD8QTQf8J9bP9Pq4SY71
-obja/vQ6UBixlRB5vDSG0UuukL4kzlyUKpHkwUcj
+MIIBzDCCAVKgAwIBAgIUHiwk5uIddEKTg21i3Zz7MFQK6bswCgYIKoZIzj0EAwIw
+HDEaMBgGA1UEAwwRcG9ueXRvd24gRUNEU0EgQ0EwIBcNMjQxMjA1MTcxMDA5WhgP
+MjEyNDExMTExNzEwMDlaMBwxGjAYBgNVBAMMEXBvbnl0b3duIEVDRFNBIENBMHYw
+EAYHKoZIzj0CAQYFK4EEACIDYgAEqv33hz22eM5t/nUXiua5t2H/S1Nb2bttslq0
+GIP/M5r1CkaDcUq4lBMnQlYAEQHlEXvze49ndUaXHts/BGmTDrPHXPCkRmXC1wtW
+MWQaLUR4DbZoPNrXVqREns48K7RWo1MwUTAdBgNVHQ4EFgQU0pt7k69yNs2PFBrT
+INzChAutF20wHwYDVR0jBBgwFoAU0pt7k69yNs2PFBrTINzChAutF20wDwYDVR0T
+AQH/BAUwAwEB/zAKBggqhkjOPQQDAgNoADBlAjAd4T6RpzJ5RJMi/6mPvk7UanYI
+UnM3Eo756hSEKcEWf2JUh/MLqWYQI3DFainOZ+4CMQC7ZQC6Jto/HmfW+q13yOcE
+nu+dnK9uv5p2PMzWhJwH+fm+OsXlvBwfRkml6H/fx1U=
 -----END CERTIFICATE-----

ractor_cluster_integration_tests/test-ca/ecdsa/ca.key

@@ -1,6 +1,6 @@
 -----BEGIN PRIVATE KEY-----
-MIG2AgEAMBAGByqGSM49AgEGBSuBBAAiBIGeMIGbAgEBBDDl30Srs7laSdaAOzoB
-kCiehcf1HXv7NqGQBECqshrtHxGEX6bAnBB7JgyDs28NvPGhZANiAAS7F1hCrNsV
-woFtTnAut6Qj40KNKt1Zz3NSduJ+pl8MaYN/fJ80eJ94h3GANKSoBJ3gbLUjrFzo
-WzWdPQwLxK94knPAcllp7RG1lY2N4PSrekOmNbFxpGlCIj/W3uZSEEU=
+MIG2AgEAMBAGByqGSM49AgEGBSuBBAAiBIGeMIGbAgEBBDDwQeF2O6t82KL2A33k
+juFLN7IFhR0OZg2avDqUTHa3vIiJ0TUJTLYCawixVAT1yjShZANiAASq/feHPbZ4
+zm3+dReK5rm3Yf9LU1vZu22yWrQYg/8zmvUKRoNxSriUEydCVgARAeURe/N7j2d1
+Rpce2z8EaZMOs8dc8KRGZcLXC1YxZBotRHgNtmg82tdWpESezjwrtFY=
 -----END PRIVATE KEY-----

ractor_cluster_integration_tests/test-ca/ecdsa/client.cert

@@ -1,13 +1,13 @@
 -----BEGIN CERTIFICATE-----
-MIIB8jCCAZegAwIBAgICAxUwCgYIKoZIzj0EAwIwLjEsMCoGA1UEAwwjcG9ueXRv
-d24gRUNEU0EgbGV2ZWwgMiBpbnRlcm1lZGlhdGUwHhcNMTkwNjA5MTcxNTEyWhcN
-MjQxMTI5MTcxNTEyWjAaMRgwFgYDVQQDDA9wb255dG93biBjbGllbnQwdjAQBgcq
-hkjOPQIBBgUrgQQAIgNiAATx0R97foSC0Ra9a13pJzfI1hh3G6476MIMslLHxg5w
-wCG8k5mMHia2hGOBbdGjoY0C1wJLNrUSov5SfcsYX6/VjHQH/elmb/KOO1AGwPD7
-1yD1+DG/cjK1okLZIVhbSQyjgZswgZgwDAYDVR0TAQH/BAIwADALBgNVHQ8EBAMC
-BsAwFgYDVR0lAQH/BAwwCgYIKwYBBQUHAwIwHQYDVR0OBBYEFFBkko+0OE2piFRx
-h9m2UonFYQFEMEQGA1UdIwQ9MDuAFD93gjUQ7CX28Dy5NlFYfYh8XlKSoSCkHjAc
-MRowGAYDVQQDDBFwb255dG93biBFQ0RTQSBDQYIBezAKBggqhkjOPQQDAgNJADBG
-AiEAvyquOUQlqAWkSlfwH3nYNmmEG9CT/jjzNs1OBr1RD6ACIQDtmqdbttqgqKAZ
-Wi5lCzftwM6Hy5aA0qy1v80H4xBJyw==
+MIIB8zCCAZmgAwIBAgICAxUwCgYIKoZIzj0EAwIwLjEsMCoGA1UEAwwjcG9ueXRv
+d24gRUNEU0EgbGV2ZWwgMiBpbnRlcm1lZGlhdGUwIBcNMjQxMjA1MTcxMDEwWhgP
+MjEyNDExMTExNzEwMTBaMBoxGDAWBgNVBAMMD3Bvbnl0b3duIGNsaWVudDB2MBAG
+ByqGSM49AgEGBSuBBAAiA2IABJUMxu6BUfe8S35aP8i/TjYorlpA9KlVzOjZb4Jc
+ekWeqr9XKJM+M1o1djTbopvlXiLGKjvOT7NTJTe4uVKNcV7cTUWevwCp6PA1VSZK
+YmjxgrJX3u8bKgKY7E8ootPY9aOBmzCBmDAMBgNVHRMBAf8EAjAAMAsGA1UdDwQE
+AwIGwDAWBgNVHSUBAf8EDDAKBggrBgEFBQcDAjAdBgNVHQ4EFgQUWQhODUmvh/Ev
+f5WSZsMGLbilY6MwRAYDVR0jBD0wO4AUZbsuPp0XuoFATxqNUkm8ZgSv8JKhIKQe
+MBwxGjAYBgNVBAMMEXBvbnl0b3duIEVDRFNBIENBggF7MAoGCCqGSM49BAMCA0gA
+MEUCIQCJjtjLNdCclIvzB7WTjLuAmrBLBy9avE8ZdSYdqNj+igIgDPlIFe1fjNie
+r5AyEYzJuuAB8KN3K5u1UgJ8Q+05ohw=
 -----END CERTIFICATE-----

ractor_cluster_integration_tests/test-ca/ecdsa/client.chain

@@ -1,24 +1,24 @@
 -----BEGIN CERTIFICATE-----
-MIIBuDCCAT2gAwIBAgIBezAKBggqhkjOPQQDAjAcMRowGAYDVQQDDBFwb255dG93
-biBFQ0RTQSBDQTAeFw0xOTA2MDkxNzE1MTJaFw0yOTA2MDYxNzE1MTJaMC4xLDAq
+MIIB2TCCAV6gAwIBAgIBezAKBggqhkjOPQQDAjAcMRowGAYDVQQDDBFwb255dG93
+biBFQ0RTQSBDQTAeFw0yNDEyMDUxNzEwMTBaFw0zNDEyMDMxNzEwMTBaMC4xLDAq
 BgNVBAMMI3Bvbnl0b3duIEVDRFNBIGxldmVsIDIgaW50ZXJtZWRpYXRlMFkwEwYH
-KoZIzj0CAQYIKoZIzj0DAQcDQgAEYtRlPykhT0YLnjcSsbe8rfmJ7ojfWuHImDGx
-DpF5vJ259giO99qFEcZTi7dNvQGBQC6bsUWddTl3Bc7gxiCr3aNeMFwwHQYDVR0O
-BBYEFD93gjUQ7CX28Dy5NlFYfYh8XlKSMCAGA1UdJQEB/wQWMBQGCCsGAQUFBwMB
-BggrBgEFBQcDAjAMBgNVHRMEBTADAQH/MAsGA1UdDwQEAwIB/jAKBggqhkjOPQQD
-AgNpADBmAjEAxdSnB7ryhG+y7tshwxqrFoZEWXpDLQDZGad0+Wf+7hiNoNCDDdIv
-MhYxzCDbTS/lAjEAwjsfrp4gxwoz/6fNfUvHyiA3j9jMd64tapzWy2hoqubKBEum
-EVczk9vVmsiJA5J3
+KoZIzj0CAQYIKoZIzj0DAQcDQgAEWG/Vj1f3GiVR7ZQePQfDziToyYkJ9a3LIW2W
+XcrbxfdlIUq83lD1rE5DUHN1ORYKCmgV/ZkK0/qGZ0kMe2ylS6N/MH0wHQYDVR0O
+BBYEFGW7Lj6dF7qBQE8ajVJJvGYEr/CSMCAGA1UdJQEB/wQWMBQGCCsGAQUFBwMB
+BggrBgEFBQcDAjAMBgNVHRMEBTADAQH/MAsGA1UdDwQEAwIB/jAfBgNVHSMEGDAW
+gBTSm3uTr3I2zY8UGtMg3MKEC60XbTAKBggqhkjOPQQDAgNpADBmAjEA3LU4nBmy
+4SA+te0JecOLT7nHy64KnFy5SCYdXYk6MIev/GfARya+G574kgT9OTLrAjEAgN92
+CsjLZi5diwkhFKDoUl63zfb0E5Zqgoh0m8O3xUC6Tf48jOWsU/yPUWNF4sCd
 -----END CERTIFICATE-----
 -----BEGIN CERTIFICATE-----
-MIIByjCCAVCgAwIBAgIUSA11/39PY7uM9Nc2ITnV1eHzaKYwCgYIKoZIzj0EAwIw
-HDEaMBgGA1UEAwwRcG9ueXRvd24gRUNEU0EgQ0EwHhcNMTkwNjA5MTcxNTEyWhcN
-MjkwNjA2MTcxNTEyWjAcMRowGAYDVQQDDBFwb255dG93biBFQ0RTQSBDQTB2MBAG
-ByqGSM49AgEGBSuBBAAiA2IABLsXWEKs2xXCgW1OcC63pCPjQo0q3VnPc1J24n6m
-Xwxpg398nzR4n3iHcYA0pKgEneBstSOsXOhbNZ09DAvEr3iSc8ByWWntEbWVjY3g
-9Kt6Q6Y1sXGkaUIiP9be5lIQRaNTMFEwHQYDVR0OBBYEFKD72TTU/GXhb3/D1/Z7
-hD/ZG6lKMB8GA1UdIwQYMBaAFKD72TTU/GXhb3/D1/Z7hD/ZG6lKMA8GA1UdEwEB
-/wQFMAMBAf8wCgYIKoZIzj0EAwIDaAAwZQIxAL9FtbNV7i9trxukhakfTvbXCHgE
-2pIOT5r/Vc5kSrPU4vJu2MOJz6X/JCX15IbZlQIwJxYfsD8QTQf8J9bP9Pq4SY71
-obja/vQ6UBixlRB5vDSG0UuukL4kzlyUKpHkwUcj
+MIIBzDCCAVKgAwIBAgIUHiwk5uIddEKTg21i3Zz7MFQK6bswCgYIKoZIzj0EAwIw
+HDEaMBgGA1UEAwwRcG9ueXRvd24gRUNEU0EgQ0EwIBcNMjQxMjA1MTcxMDA5WhgP
+MjEyNDExMTExNzEwMDlaMBwxGjAYBgNVBAMMEXBvbnl0b3duIEVDRFNBIENBMHYw
+EAYHKoZIzj0CAQYFK4EEACIDYgAEqv33hz22eM5t/nUXiua5t2H/S1Nb2bttslq0
+GIP/M5r1CkaDcUq4lBMnQlYAEQHlEXvze49ndUaXHts/BGmTDrPHXPCkRmXC1wtW
+MWQaLUR4DbZoPNrXVqREns48K7RWo1MwUTAdBgNVHQ4EFgQU0pt7k69yNs2PFBrT
+INzChAutF20wHwYDVR0jBBgwFoAU0pt7k69yNs2PFBrTINzChAutF20wDwYDVR0T
+AQH/BAUwAwEB/zAKBggqhkjOPQQDAgNoADBlAjAd4T6RpzJ5RJMi/6mPvk7UanYI
+UnM3Eo756hSEKcEWf2JUh/MLqWYQI3DFainOZ+4CMQC7ZQC6Jto/HmfW+q13yOcE
+nu+dnK9uv5p2PMzWhJwH+fm+OsXlvBwfRkml6H/fx1U=
 -----END CERTIFICATE-----