feat(metrics): optional metrics layer (off by default, Prometheus/OTLP-ready)

Cargo.toml

@@ -69,8 +69,8 @@ hmac = { version = "0.13.0", default-features = false }
 iai-callgrind = "0.16"
 itoa = "1"
 log = "0.4"
+metrics = "0.24"
 portable-atomic = { version = "1", default-features = false, features = ["fallback"] }
-tracing = { version = "0.1", default-features = false, features = ["attributes"] }
 prost = { version = "0.14.3", default-features = false, features = ["std"] }
 prost-build = { version = "0.14.3", default-features = false }
 rand = "0.10"
@@ -82,6 +82,7 @@ sha2 = { version = "0.11.0", default-features = false }
 subtle = { version = "2.6", default-features = false }
 thiserror = "2.0.17"
 tokio = { version = "1.48.0", default-features = false }
+tracing = { version = "0.1", default-features = false, features = ["attributes"] }
 uuid = { version = "1", default-features = false }
 
 # Internal workspace crates
@@ -101,6 +102,10 @@ debug-snapshots = ["wacore/debug-snapshots"]
 # Emits tracing spans/events only; the application installs the subscriber
 # (and any OpenTelemetry bridge). See examples/observability.rs.
 tracing = ["dep:tracing", "wacore/tracing"]
+# Optional metrics (counters/histograms/gauges via the `metrics` facade). Off by
+# default: no dependency, zero overhead. Emits only; the application installs a
+# recorder (e.g. metrics-exporter-prometheus). See examples/metrics.rs.
+metrics = ["wacore/metrics"]
 # Render raw phone numbers in tracing fields instead of the redacted `pn#<hash>`.
 # Local debugging only; never enable in production.
 tracing-pii = ["wacore/tracing-pii", "wacore-binary/tracing-pii"]
@@ -139,7 +144,6 @@ futures = { workspace = true, features = ["std"] }
 hex = { workspace = true }
 itoa = { workspace = true }
 log = { workspace = true }
-tracing = { workspace = true, optional = true }
 moka = { version = "0.12.12", features = ["future"], optional = true }
 portable-atomic = { workspace = true }
 prost = { workspace = true }
@@ -149,6 +153,7 @@ serde = { workspace = true }
 serde_json = { workspace = true, features = ["std"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt", "sync", "time"], optional = true }
+tracing = { workspace = true, optional = true }
 wacore = { workspace = true }
 wacore-binary = { workspace = true }
 waproto = { workspace = true }
@@ -171,9 +176,10 @@ cbc = { version = "0.2", features = ["alloc", "block-padding"] }
 flate2 = { workspace = true }
 hkdf = { workspace = true }
 hmac = { workspace = true }
+metrics-exporter-prometheus = "0.16"
 sha2 = { workspace = true }
-uuid = { workspace = true, features = ["v4"] }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+uuid = { workspace = true, features = ["v4"] }
 wacore-noise = { path = "./wacore/noise", features = [
     "test-util",
     "danger-skip-cert-chain-verify",
@@ -190,6 +196,10 @@ required-features = ["danger-skip-tls-verify"]
 name = "observability"
 required-features = ["tracing"]
 
+[[example]]
+name = "metrics"
+required-features = ["metrics"]
+
 [profile.release]
 opt-level = 3
 debug = false

examples/metrics.rs

@@ -0,0 +1,38 @@
+//! Wiring metrics for `whatsapp-rust`.
+//!
+//! Run with:
+//!     cargo run --example metrics --features metrics
+//!
+//! The library only *emits* metrics through the `metrics` facade (the `wa_*`
+//! counters/histograms/gauges in `whatsapp_rust::telemetry`). It never installs a
+//! recorder or depends on Prometheus/OTLP; the application does, as shown here.
+//! With the `metrics` feature off there is no dependency and every emit is a
+//! zero-cost no-op.
+//!
+//! Metric labels are strictly categorical (outcome, kind, namespace, ...); JIDs,
+//! phone numbers and message ids are never used as labels.
+
+fn main() {
+    // Install a Prometheus recorder. `install_recorder()` sets the global recorder
+    // and returns a handle you can render from your own HTTP endpoint. Use
+    // `PrometheusBuilder::install()` instead (inside a Tokio runtime) to also serve
+    // `/metrics` on 0.0.0.0:9000 automatically.
+    let handle = metrics_exporter_prometheus::PrometheusBuilder::new()
+        .install_recorder()
+        .expect("install prometheus recorder");
+
+    // Register units/help for the wa_* metrics (optional, improves the output).
+    whatsapp_rust::telemetry::describe();
+
+    // From here you would build and run a `whatsapp_rust::Client` as usual; every
+    // wa_* metric is recorded into the recorder above. A couple of demo emits:
+    whatsapp_rust::telemetry::connect("ok");
+    whatsapp_rust::telemetry::recv("decrypted");
+    {
+        let _t = whatsapp_rust::telemetry::timer(whatsapp_rust::telemetry::IQ_DURATION);
+        // ... the IQ round-trip would happen here; the timer records on drop.
+    }
+
+    // Scrape this from your HTTP `/metrics` handler.
+    println!("{}", handle.render());
+}

src/client/app_state.rs

@@ -62,6 +62,7 @@ impl Client {
     }
 
     async fn fetch_app_state_with_retry_inner(&self, name: WAPatchName) -> anyhow::Result<()> {
+        let _t = wacore::telemetry::timer(wacore::telemetry::APPSTATE_SYNC_DURATION);
         let mut attempt = 0u32;
         loop {
             attempt += 1;
@@ -70,7 +71,10 @@ impl Client {
             // Matches WA Web which only requests snapshot when version is undefined.
             let res = self.process_app_state_sync_task(name, false).await;
             match res {
-                Ok(()) => return Ok(()),
+                Ok(()) => {
+                    wacore::telemetry::appstate_sync("ok");
+                    return Ok(());
+                }
                 Err(e) => {
                     if e.downcast_ref::<crate::appstate_sync::AppStateSyncError>()
                         .is_some_and(|ase| {
@@ -109,6 +113,7 @@ impl Client {
                         self.runtime.sleep(backoff).await;
                         continue;
                     }
+                    wacore::telemetry::appstate_sync("fail");
                     return Err(e);
                 }
             }
@@ -330,6 +335,7 @@ impl Client {
                 // (version was 0 before sync). This prevents server_sync-triggered
                 // incremental syncs from being incorrectly marked as full syncs.
                 let full_sync = was_snapshot.contains(&name);
+                wacore::telemetry::appstate_mutations(mutations.len() as u64);
                 for m in mutations {
                     self.dispatch_app_state_mutation(&m, full_sync).await;
                 }
@@ -516,6 +522,7 @@ impl Client {
             };
             self.request_missing_keys_with_dedup(missing).await;
 
+            wacore::telemetry::appstate_mutations(mutations.len() as u64);
             for m in mutations {
                 debug!(target: "Client/AppState", "Dispatching mutation kind={} index_len={} full_sync={}", m.index.first().map(|s| s.as_str()).unwrap_or(""), m.index.len(), full_sync);
                 self.dispatch_app_state_mutation(&m, full_sync).await;

src/client/lifecycle.rs

@@ -61,6 +61,7 @@ impl Client {
     /// Dispatch the Connected event and notify waiters.
     pub(crate) fn dispatch_connected(&self) {
         self.is_ready.store(true, Ordering::Relaxed);
+        wacore::telemetry::set_connected(true);
         self.core
             .event_bus
             .dispatch(Event::Connected(crate::types::events::Connected));
@@ -277,6 +278,7 @@ impl Client {
             self.expected_disconnect.store(false, Ordering::Relaxed);
 
             if let Err(connect_err) = self.connect().await {
+                wacore::telemetry::connect("fail");
                 let is_transient = connect_err
                     .downcast_ref::<crate::handshake::HandshakeError>()
                     .is_some_and(|e| e.is_transient());
@@ -286,6 +288,7 @@ impl Client {
                     error!("Failed to connect: {connect_err:#}. Will retry...");
                 }
             } else {
+                wacore::telemetry::connect("ok");
                 let unexpected_disconnect = if self.read_messages_loop().await.is_err() {
                     // Check intentional_reconnect AFTER read loop exits — reconnect()
                     // sets this flag while the loop is running, so it must be read here.
@@ -362,6 +365,7 @@ impl Client {
         if self.is_connected() {
             return Err(ClientError::AlreadyConnected.into());
         }
+        let _t = wacore::telemetry::timer(wacore::telemetry::CONNECT_DURATION);
 
         // Reset login state for new connection attempt. This ensures that
         // handle_success will properly process the <success> stanza even if
@@ -475,6 +479,7 @@ impl Client {
     )]
     pub async fn disconnect(self: &Arc<Self>) {
         info!("Disconnecting client intentionally.");
+        wacore::telemetry::set_connected(false);
         self.expected_disconnect.store(true, Ordering::Relaxed);
         self.is_running.store(false, Ordering::Relaxed);
         self.shutdown_notifier.notify();
@@ -523,6 +528,7 @@ impl Client {
     )]
     pub async fn reconnect(self: &Arc<Self>) {
         info!("Reconnecting: dropping transport for auto-reconnect.");
+        wacore::telemetry::reconnect();
         self.intentional_reconnect.store(true, Ordering::Relaxed);
         self.auto_reconnect_errors
             .store(Self::RECONNECT_BACKOFF_STEP, Ordering::Relaxed);
@@ -594,6 +600,10 @@ impl Client {
         // is_connected==true with a cleared socket. send_node() independently
         // checks the socket, but this ordering avoids a confusing state window.
         self.is_connected.store(false, Ordering::Release);
+        // Authoritative point for the gauge: every disconnect (intentional or a
+        // run-loop drop/reconnect) funnels through here, so disconnect()'s early
+        // set is just a prompt redundant signal.
+        wacore::telemetry::set_connected(false);
         // Presence doesn't survive reconnects: demote presence-driven active
         // receipts (1 -> 0), leaving a forced value (2) untouched.
         let _ =

src/client/node_io.rs

@@ -968,6 +968,7 @@ impl Client {
         tracing::instrument(name = "wa.conn.stream_error", level = "debug", skip_all)
     )]
     pub(crate) async fn handle_stream_error(&self, node: &wacore_binary::NodeRef<'_>) {
+        wacore::telemetry::stream_error();
         // is_logged_in handling: opt-in branches (515/516/401/409/conflict) clear it
         // in the disconnect block below; 429/503 clear it inline because the server
         // explicitly rejected the session and outgoing sends should bail fast; the

src/handlers/notification.rs

@@ -415,6 +415,9 @@ async fn handle_identity_change(client: &Arc<Client>, node: &NodeRef<'_>) {
         return;
     }
 
+    // Counted here, past the companion/self/no-prior gates, so it reflects actual
+    // session resets rather than every identity-change push received.
+    wacore::telemetry::identity_change();
     info!(
         "Identity change for {} (had_prior_identity=true): resetting session",
         from_jid.user

src/lib.rs

@@ -5,6 +5,8 @@
 
 pub use wacore::appstate::schemas;
 pub use wacore::client_profile::ClientProfile;
+/// Optional metrics emission (the `metrics` feature). No-op when the feature is off.
+pub use wacore::telemetry;
 pub use wacore::{
     iq::privacy as privacy_settings, proto_helpers, sticker_pack, store::traits, webp,
 };

src/message/dispatch.rs

@@ -11,6 +11,7 @@ impl Client {
         info: &Arc<MessageInfo>,
     ) {
         use wacore::proto_helpers::MessageExt;
+        wacore::telemetry::recv("decrypted");
 
         let mut info = Arc::clone(info);
         if info.ephemeral_expiration.is_none()

src/message/receive.rs

@@ -523,6 +523,9 @@ impl Client {
         let mut session_guard: Option<async_lock::MutexGuardArc<()>> =
             Some(session_mutex.lock_arc().await);
 
+        // Started after the lock so the histogram is crypto-only, not lock/queue wait.
+        let _t = wacore::telemetry::timer(wacore::telemetry::DECRYPT_DURATION);
+
         let mut adapter = self.signal_adapter().await;
         let mut rng = rand::make_rng::<rand::rngs::StdRng>();
         let mut outcome = SessionBatchOutcome::default();

src/message/retry.rs

@@ -32,6 +32,7 @@ impl Client {
             .await;
         let was_fresh = fresh.load(std::sync::atomic::Ordering::Acquire);
         if was_fresh {
+            wacore::telemetry::recv("undecryptable");
             self.core.event_bus.dispatch(Event::UndecryptableMessage(
                 crate::types::events::UndecryptableMessage {
                     info,
@@ -244,6 +245,7 @@ impl Client {
 
         let retry_sent = match self.send_retry_receipt(info, retry_count, reason).await {
             Ok(()) => {
+                wacore::telemetry::retry_receipt(reason.as_str());
                 debug!(
                     "Sent retry receipt #{} for message {} in chat {} from {} [{:?}]",
                     retry_count,

src/prekeys.rs

@@ -134,7 +134,10 @@ impl Client {
         }
 
         log::info!("Server missing prekeys (persisted flag), uploading.");
-        self.upload_pre_keys_inner().await
+        // Operation-level outcome (the login path skips the retry wrapper).
+        let r = self.upload_pre_keys_inner().await;
+        wacore::telemetry::prekey_upload(if r.is_ok() { "ok" } else { "fail" });
+        r
     }
 
     /// Ensure the server has enough pre-keys, uploading if below threshold.
@@ -340,6 +343,8 @@ impl Client {
             match self.upload_pre_keys(force).await {
                 Ok(()) => {
                     log::info!("Pre-key upload succeeded");
+                    // Operation-level outcome: one emit per logical upload, not per attempt.
+                    wacore::telemetry::prekey_upload("ok");
                     return Ok(());
                 }
                 Err(e) => {
@@ -352,6 +357,7 @@ impl Client {
 
                     // Bail if disconnected during retry wait
                     if !self.is_logged_in.load(Ordering::Relaxed) {
+                        wacore::telemetry::prekey_upload("fail");
                         return Err(anyhow::anyhow!(
                             "Connection lost during pre-key upload retry"
                         ));

src/request.rs

@@ -208,7 +208,9 @@ impl Client {
     where
         F: std::future::Future<Output = Result<(), crate::client::ClientError>>,
     {
+        let _t = wacore::telemetry::timer(wacore::telemetry::IQ_DURATION);
         if !self.is_running.load(Ordering::Relaxed) {
+            wacore::telemetry::iq("error");
             return Err(IqError::NotConnected);
         }
 
@@ -224,11 +226,13 @@ impl Client {
 
         if !self.is_running.load(Ordering::Acquire) {
             self.response_waiters.lock().await.remove(&req_id);
+            wacore::telemetry::iq("error");
             return Err(IqError::NotConnected);
         }
 
         if let Err(e) = send_fn.await {
             self.response_waiters.lock().await.remove(&req_id);
+            wacore::telemetry::iq("error");
             return match e {
                 ClientError::Socket(s_err) => Err(IqError::Socket(s_err)),
                 ClientError::EncryptSend(es_err) => Err(IqError::EncryptSend(es_err)),
@@ -240,7 +244,7 @@ impl Client {
         }
 
         let request_utils = self.get_request_utils();
-        futures::select! {
+        let result = futures::select! {
             result = rt_timeout(&*self.runtime, timeout, rx).fuse() => {
                 match result {
                     Ok(Ok(response_node)) => match request_utils.parse_iq_response(response_node.get()) {
@@ -258,6 +262,12 @@ impl Client {
                 self.response_waiters.lock().await.remove(&req_id);
                 Err(IqError::NotConnected)
             }
-        }
+        };
+        wacore::telemetry::iq(match &result {
+            Ok(_) => "ok",
+            Err(IqError::Timeout) => "timeout",
+            Err(_) => "error",
+        });
+        result
     }
 }

src/send.rs

@@ -396,6 +396,13 @@ impl Client {
         mut message: wa::Message,
         options: SendOptions,
     ) -> Result<SendResult, anyhow::Error> {
+        let _t = wacore::telemetry::timer(wacore::telemetry::SEND_DURATION);
+        wacore::telemetry::send(match to.server {
+            wacore_binary::Server::Group => "group",
+            wacore_binary::Server::Broadcast => "status",
+            wacore_binary::Server::Newsletter => "newsletter",
+            _ => "dm",
+        });
         if let Some(exp) = options.ephemeral_expiration
             && exp > 0
         {
@@ -485,6 +492,10 @@ impl Client {
             return Err(anyhow!("Cannot send status with no recipients"));
         }
 
+        // Status posts don't go through send_message_with_options, so count them here.
+        let _t = wacore::telemetry::timer(wacore::telemetry::SEND_DURATION);
+        wacore::telemetry::send("status");
+
         let to = Jid::status_broadcast();
         let request_id = self.generate_message_id().await;
 

wacore/Cargo.toml

@@ -14,6 +14,8 @@ debug-diagnostics = []
 debug-snapshots = []
 # Optional observability: emit tracing spans/events. Off by default (no dep).
 tracing = ["dep:tracing"]
+# Optional metrics via the `metrics` facade. Off by default (no dep).
+metrics = ["dep:metrics"]
 # Render raw phone numbers in tracing fields instead of redacted tokens (debug only).
 tracing-pii = ["wacore-binary/tracing-pii"]
 # Disables XEdDSA verification of the server's Noise cert chain. Required
@@ -43,7 +45,7 @@ hmac = { workspace = true }
 itoa = { workspace = true }
 log = { workspace = true }
 md5 = "0.8.0"
-tracing = { workspace = true, optional = true }
+metrics = { workspace = true, optional = true }
 portable-atomic = { workspace = true }
 prost = { workspace = true }
 rand = { workspace = true }
@@ -54,6 +56,7 @@ sha1 = { workspace = true }
 sha2 = { workspace = true }
 subtle = { workspace = true }
 thiserror = { workspace = true }
+tracing = { workspace = true, optional = true }
 typed-builder = "0.23"
 wacore-appstate = { workspace = true }
 wacore-binary = { workspace = true, features = ["serde"] }