diff --git a/Cargo.lock b/Cargo.lock index 642e983b3..b33d5a1df 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1133,6 +1133,15 @@ version = "0.4.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "113b30b4cd05f7c06868fdb2854f66a7b9fece9a48425351cd532e810d74024f" +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "md5" version = "0.8.0" @@ -1219,6 +1228,15 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num-conv" version = "0.2.2" @@ -1716,6 +1734,15 @@ dependencies = [ "digest", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "2.0.1" @@ -1869,6 +1896,15 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3bf63baf9f5039dadc247375c29eb13706706cfde997d0330d05aa63a77d8820" +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "time" version = "0.3.47" @@ -2002,6 +2038,67 @@ dependencies = [ "winnow 1.0.3", ] +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", +] + [[package]] name = "typed-builder" version = "0.23.2" @@ -2107,6 +2204,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "vcpkg" version = "0.2.15" @@ -2155,6 +2258,7 @@ dependencies = [ "subtle", "thiserror", "tokio", + "tracing", "typed-builder", "wacore-appstate", "wacore-binary", @@ -2414,6 +2518,8 @@ dependencies = [ "sha2", "thiserror", "tokio", + "tracing", + "tracing-subscriber", "uuid", "wacore", "wacore-binary", diff --git a/Cargo.toml b/Cargo.toml index fde994648..71c879b7e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -70,6 +70,7 @@ iai-callgrind = "0.16" itoa = "1" log = "0.4" 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" @@ -96,6 +97,13 @@ yoke = { version = "0.8", features = ["derive"] } [features] debug-diagnostics = ["wacore/debug-diagnostics"] debug-snapshots = ["wacore/debug-snapshots"] +# Optional observability. Off by default: no `tracing` dep, zero overhead. +# Emits tracing spans/events only; the application installs the subscriber +# (and any OpenTelemetry bridge). See examples/observability.rs. +tracing = ["dep:tracing", "wacore/tracing"] +# Render raw phone numbers in tracing fields instead of the redacted `pn#`. +# Local debugging only; never enable in production. +tracing-pii = ["wacore/tracing-pii", "wacore-binary/tracing-pii"] danger-skip-tls-verify = ["whatsapp-rust-tokio-transport?/danger-skip-tls-verify"] danger-skip-cert-chain-verify = ["wacore/danger-skip-cert-chain-verify"] default = [ @@ -131,6 +139,7 @@ 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 } @@ -164,6 +173,7 @@ hkdf = { workspace = true } hmac = { workspace = true } sha2 = { workspace = true } uuid = { workspace = true, features = ["v4"] } +tracing-subscriber = { version = "0.3", features = ["env-filter"] } wacore-noise = { path = "./wacore/noise", features = [ "test-util", "danger-skip-cert-chain-verify", @@ -176,6 +186,10 @@ workspace = true name = "benchmark" required-features = ["danger-skip-tls-verify"] +[[example]] +name = "observability" +required-features = ["tracing"] + [profile.release] opt-level = 3 debug = false diff --git a/examples/observability.rs b/examples/observability.rs new file mode 100644 index 000000000..b8136d3fc --- /dev/null +++ b/examples/observability.rs @@ -0,0 +1,62 @@ +//! Wiring observability for `whatsapp-rust`. +//! +//! Run with: +//! cargo run --example observability --features tracing +//! +//! The library only *emits* `tracing` spans/events (and keeps its existing `log` +//! calls). It never installs a subscriber and never depends on OpenTelemetry — +//! that is the application's job, shown here. +//! +//! Two things happen below: +//! +//! 1. A `tracing-subscriber` is installed. Its default `tracing-log` feature +//! bridges the library's existing `log::{info,warn,error}!` calls into +//! tracing, so they become events attached to the active `wa.*` span. +//! 2. Span/level/target filtering is driven by `RUST_LOG` (EnvFilter), e.g. +//! `RUST_LOG="info,whatsapp_rust=debug,wacore=debug"`. The library groups +//! spans under `wa.*` names and reuses its `target: "Client/AppState"`-style +//! targets, so you can filter per area. +//! +//! IMPORTANT: do NOT enable the `log` feature on the `tracing` crate together +//! with a log->tracing bridge — that recurses. This crate already pins +//! `tracing` with `default-features = false` so the hazard cannot happen. +//! +//! PII note: the bridged `log` lines surface alongside the redacted `wa.*` spans. +//! The library renders JIDs and Signal addresses in its own log messages through +//! `Jid::observe()` / `observe_protocol_address()` (phone numbers become +//! `pn#`), so the `whatsapp_rust`/`wacore` log lines carry the same +//! redaction as the span fields. Your own application code is a separate leak +//! path: any raw JID/phone you log reaches the exporter under your own targets, +//! and dropping the library targets does nothing for it — scrub your app's logs +//! with `Jid::observe()` too. The `tracing-pii` cargo feature (off) renders raw +//! numbers for local debugging only. + +fn main() { + use tracing_subscriber::prelude::*; + + let filter = tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info,whatsapp_rust=debug")); + + tracing_subscriber::registry() + .with(filter) + .with(tracing_subscriber::fmt::layer()) + // ── OpenTelemetry (OTLP) ──────────────────────────────────────────── + // Add the application deps `opentelemetry`, `opentelemetry-otlp` and + // `tracing-opentelemetry`, then append a layer here: + // + // let tracer = opentelemetry_otlp::new_pipeline() + // .tracing() + // .with_exporter(opentelemetry_otlp::new_exporter().tonic()) + // .install_batch(opentelemetry_sdk::runtime::Tokio)?; + // .with(tracing_opentelemetry::layer().with_tracer(tracer)) + // + // Every `wa.*` span is then exported as an OTLP span with its fields + // (chat/peer/msg_id are already privacy-redacted via `Jid::observe()`). + .init(); + + tracing::info!("observability initialized — RUST_LOG drives filtering"); + + // From here you would build and run a `whatsapp_rust::Client` as usual; all + // connect / recv / decrypt / send / iq / appstate / pair / media spans and + // the bridged log events will flow into the subscriber above. +} diff --git a/src/bot.rs b/src/bot.rs index 324c70a58..777533336 100644 --- a/src/bot.rs +++ b/src/bot.rs @@ -58,6 +58,7 @@ impl MessageContext { Some(Self::from_arc(Arc::clone(msg), info, client)) } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.bot.send_message", level = "debug", skip_all, fields(chat = %self.info.source.chat.observe()), err(Debug)))] pub async fn send_message( &self, message: wa::Message, @@ -94,6 +95,7 @@ impl MessageContext { } } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.bot.edit_message", level = "debug", skip_all, fields(chat = %self.info.source.chat.observe()), err(Debug)))] pub async fn edit_message( &self, original_message_id: impl Into, @@ -109,6 +111,7 @@ impl MessageContext { } /// Delete a message for everyone in the chat. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.bot.revoke_message", level = "debug", skip_all, fields(chat = %self.info.source.chat.observe()), err(Debug)))] pub async fn revoke_message( &self, message_id: String, @@ -122,6 +125,7 @@ impl MessageContext { /// React to the incoming message. An empty `emoji` removes a previous /// reaction. The target key (including the group/status participant) is /// taken from [`MessageContext::message_key`]. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.bot.react", level = "debug", skip_all, fields(chat = %self.info.source.chat.observe()), err(Debug)))] pub async fn react(&self, emoji: &str) -> Result { self.client .send_reaction(&self.info.source.chat, self.message_key(), emoji) @@ -220,6 +224,10 @@ impl Bot { self.client.clone() } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.bot.run", level = "debug", skip_all, err(Debug)) + )] pub async fn run(&mut self) -> Result { if let Some(receiver) = self.sync_task_receiver.take() { let worker_client = Arc::downgrade(&self.client); @@ -701,6 +709,10 @@ impl BotBuilder { // ── build() — only available when all 4 required fields are Provided ───── impl BotBuilder { + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.bot.build", level = "debug", skip_all, err(Debug)) + )] pub async fn build(self) -> std::result::Result { // Destructure to extract required fields — typestate guarantees all are Some. let (Some(runtime), Some(backend), Some(transport_factory), Some(http_client)) = ( diff --git a/src/client/app_state.rs b/src/client/app_state.rs index 597f64841..24877a219 100644 --- a/src/client/app_state.rs +++ b/src/client/app_state.rs @@ -18,6 +18,10 @@ impl Client { } /// Public entry point for processing [`MajorSyncTask`] from the sync channel. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.appstate.sync_task", level = "debug", skip_all) + )] pub async fn process_sync_task(self: &Arc, task: crate::sync_task::MajorSyncTask) { match task { crate::sync_task::MajorSyncTask::HistorySync { @@ -36,6 +40,7 @@ impl Client { } } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.fetch", level = "debug", skip_all, fields(name = ?name), err(Debug)))] pub(crate) async fn fetch_app_state_with_retry(&self, name: WAPatchName) -> anyhow::Result<()> { // In-flight dedup: skip if this collection is already being synced. // Matches WA Web's WAWebSyncdCollectionsStateMachine which tracks in-flight syncs @@ -113,6 +118,7 @@ impl Client { /// Sync multiple collections in a single IQ request, re-fetching those with `has_more_patches`. /// Matches WA Web's `serverSync()` outer loop (`3JJWKHeu5-P.js:54278-54305`). /// Max 5 iterations (WA Web's `C=5` constant). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.sync_batched", level = "debug", skip_all, fields(count = collections.len()), err(Debug)))] pub(crate) async fn sync_collections_batched( &self, collections: Vec, @@ -359,6 +365,7 @@ impl Client { Ok(()) } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.sync", level = "debug", skip_all, fields(name = ?name, full_sync = full_sync), err(Debug)))] pub(crate) async fn process_app_state_sync_task( &self, name: WAPatchName, @@ -560,6 +567,7 @@ impl Client { } } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.request_keys", level = "debug", skip_all, fields(count = raw_key_ids.len()), err(Debug)))] async fn request_app_state_keys(&self, raw_key_ids: &[Vec]) -> Result<(), anyhow::Error> { if raw_key_ids.is_empty() { return Ok(()); @@ -604,6 +612,7 @@ impl Client { /// Send an app state patch to the server for a given collection. /// /// Builds the IQ stanza and sends it. Returns the updated hash state. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.send_patch", level = "debug", skip_all, fields(name = %collection_name, count = mutations.len()), err(Debug)))] pub(crate) async fn send_app_state_patch( &self, collection_name: &str, @@ -733,6 +742,7 @@ impl Client { } } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.clean_dirty", level = "debug", skip_all, fields(bit = ?bit), err(Debug)))] pub async fn clean_dirty_bits( &self, bit: wacore::iq::dirty::DirtyBit, diff --git a/src/client/device_registry.rs b/src/client/device_registry.rs index f913cbaaa..d149e4c51 100644 --- a/src/client/device_registry.rs +++ b/src/client/device_registry.rs @@ -150,6 +150,15 @@ impl Client { /// Update the device list for a user. /// Stores under LID when mapping is known, otherwise under PN. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.update_device_list", + level = "debug", + skip_all, + err(Debug) + ) + )] pub(crate) async fn update_device_list( &self, mut record: wacore::store::traits::DeviceListRecord, @@ -203,6 +212,7 @@ impl Client { /// collapses into a single transaction. Used by usync after fetching /// device lists for many users at once, where the per-row commit /// dominated wall-clock time on large groups. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.update_device_lists", level = "debug", skip_all, fields(count = records.len()), err(Debug)))] pub(crate) async fn update_device_lists( &self, records: Vec, @@ -284,6 +294,14 @@ impl Client { /// /// Removes all device registry cache entries (all LID/PN aliases) so the /// next lookup falls through to the database or network. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.invalidate_device_cache", + level = "debug", + skip_all + ) + )] pub(crate) async fn invalidate_device_cache(&self, user: &str) { let lookup = self.resolve_lookup_keys(user).await; @@ -312,6 +330,10 @@ impl Client { /// New devices need no explicit cache invalidation: `resolve_skdm_targets` /// queries the registry on each send and `device_has_key()` returns `None` /// for unseen device IDs, dropping them into `needs_skdm` automatically. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.session.patch_device_add", level = "debug", skip_all) + )] pub(crate) async fn patch_device_add( &self, user: &str, @@ -421,6 +443,10 @@ impl Client { /// (matches WA Web's single-threaded model). A concurrent encrypt to one of /// those companions can re-store a session right after the wipe, but that is /// self-healing: the next send re-establishes it via `process_prekey_bundle`. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.session.clear_device_record", level = "debug", skip_all) + )] pub(crate) async fn clear_device_record( &self, user: &str, @@ -452,6 +478,7 @@ impl Client { /// (`UpdateDeviceTableApi`): deletes Signal sessions for the device, /// then invalidates the sender key device cache so SKDM will be /// redistributed on the next group send. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.patch_device_remove", level = "debug", skip_all, fields(device_id = device_id)))] pub(crate) async fn patch_device_remove(&self, user: &str, device_id: u32) { if let Some(mut record) = self.load_device_record(user).await { let before = record.devices.len(); @@ -507,6 +534,7 @@ impl Client { /// Cache eviction runs only after the DB delete succeeds; on failure the /// error is propagated so the caller can leave both DB and cache in their /// pre-call state rather than half-applying the cleanup. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.delete_sender_key_rows", level = "debug", skip_all, fields(device_id = device_id), err(Debug)))] async fn delete_sender_key_rows_for_device( &self, user: &str, @@ -536,6 +564,10 @@ impl Client { } /// Update key_index for a device in the registry. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.session.patch_device_update", level = "debug", skip_all) + )] pub(crate) async fn patch_device_update( &self, user: &str, @@ -593,6 +625,7 @@ impl Client { /// /// This follows the same 2-tier pattern as [`has_device`]: registry cache first, /// then the backend database. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.get_devices_from_registry", level = "trace", skip_all, fields(peer = %jid.observe())))] pub(crate) async fn get_devices_from_registry(&self, jid: &Jid) -> Option> { // Use the borrowed `&str` keys directly: both the moka cache and the // backend take `&str`, so going through `get_lookup_keys` (which re-owns @@ -669,6 +702,14 @@ impl Client { } /// Migrate device registry entries from PN key to LID key. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.migrate_device_registry", + level = "debug", + skip_all + ) + )] pub(crate) async fn migrate_device_registry_on_lid_discovery(&self, pn: &str, lid: &str) { let backend = self.persistence_manager.backend(); diff --git a/src/client/lid_pn.rs b/src/client/lid_pn.rs index 7d2b293e5..34c23d098 100644 --- a/src/client/lid_pn.rs +++ b/src/client/lid_pn.rs @@ -39,6 +39,15 @@ impl Client { /// Warm up the LID-PN cache from persistent storage. /// This is called during client initialization to populate the in-memory cache /// with previously learned LID-PN mappings. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.warm_up_lid_pn_cache", + level = "debug", + skip_all, + err(Debug) + ) + )] pub(crate) async fn warm_up_lid_pn_cache(&self) -> Result<(), anyhow::Error> { let backend = self.persistence_manager.backend(); let entries = backend.get_all_lid_mappings().await?; @@ -56,6 +65,15 @@ impl Client { /// Awaits the persist + any device/session migrations. Hot paths should /// prefer [`learn_lid_pn_mapping_fast`]. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.add_lid_pn_mapping", + level = "debug", + skip_all, + err(Debug) + ) + )] pub(crate) async fn add_lid_pn_mapping( &self, lid: &str, @@ -90,6 +108,7 @@ impl Client { /// record is gone /// - `migrate_signal_sessions_on_lid_discovery` no-ops after the sessions /// are migrated + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.learn_lid_pn_fast", level = "trace", skip_all, fields(is_offline = is_offline)))] pub(crate) async fn learn_lid_pn_mapping_fast( self: &Arc, lid: &str, @@ -140,6 +159,7 @@ impl Client { /// into the `LidPnEntry` stored in the cache, then (via `into_iter`) into /// the `LidPnMappingEntry` that's persisted — no clones on either step. /// The `Vec` itself is consumed, so no copy of the outer container either. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.learn_lid_pn_batch", level = "debug", skip_all, fields(count = mappings.len(), is_offline = is_offline)))] pub(crate) async fn learn_lid_pn_mappings_batch( self: &Arc, mappings: Vec<(String, String)>, @@ -218,6 +238,7 @@ impl Client { (entry, is_new_mapping) } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.persist_migrate_lid_pn", level = "debug", skip_all, fields(is_new = is_new_mapping), err(Debug)))] async fn persist_and_migrate_lid_pn( &self, entry: LidPnEntry, @@ -261,6 +282,7 @@ impl Client { Ok(()) } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.persist_migrate_lid_pn_batch", level = "debug", skip_all, fields(count = entries.len()), err(Debug)))] async fn persist_and_migrate_lid_pn_batch( &self, entries: Vec, @@ -398,6 +420,14 @@ impl Client { /// Callers must NOT hold `session_lock_for()` for any device /// in [0, 100) — `async_lock::Mutex` is not reentrant. The decrypt path /// drops its address lock around the call (`try_pn_to_lid_migration_decrypt`). + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.migrate_signal_sessions", + level = "debug", + skip_all + ) + )] pub(crate) async fn migrate_signal_sessions_on_lid_discovery(&self, pn: &str, lid: &str) { use log::{info, warn}; use wacore::types::jid::JidExt; @@ -510,6 +540,7 @@ impl Client { /// /// Backend errors are propagated — callers can distinguish "no mapping" /// (`Ok(None)`) from "lookup failed" (`Err(_)`). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.get_lid_pn_entry", level = "trace", skip_all, fields(peer = %jid.observe()), err(Debug)))] pub async fn get_lid_pn_entry(&self, jid: &Jid) -> Result> { let (hit, is_lid) = if jid.is_lid() { (self.lid_pn_cache.get_entry_by_lid(&jid.user).await, true) @@ -559,7 +590,7 @@ impl Client { Err(e) => { log::warn!( "resolve_recipient_to_lid: LID lookup for {} failed: {:?}", - jid, + jid.observe(), e ); None diff --git a/src/client/lifecycle.rs b/src/client/lifecycle.rs index ce23f1922..4a90a73b7 100644 --- a/src/client/lifecycle.rs +++ b/src/client/lifecycle.rs @@ -253,11 +253,26 @@ impl Client { (arc, rx) } + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.conn.run", + level = "info", + skip_all, + fields(account = tracing::field::Empty) + ) + )] pub async fn run(self: &Arc) { if self.is_running.swap(true, Ordering::SeqCst) { warn!("Client `run` method called while already running."); return; } + // Tag the session-root span with our own (pseudonymous) account id so + // connection-lifecycle traces are attributable per account. + #[cfg(feature = "tracing")] + if let Some(lid) = self.get_lid().await { + tracing::Span::current().record("account", tracing::field::display(lid.observe())); + } while self.is_running.load(Ordering::Relaxed) { self.expected_disconnect.store(false, Ordering::Relaxed); @@ -331,6 +346,10 @@ impl Client { info!("Client run loop has shut down."); } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.connect", level = "info", skip_all, err(Debug)) + )] pub async fn connect(self: &Arc) -> Result<(), anyhow::Error> { if self.is_connecting.swap(true, Ordering::SeqCst) { return Err(ClientError::AlreadyConnected.into()); @@ -422,6 +441,10 @@ impl Client { /// Deregister this companion device and disconnect. /// Does NOT wipe stored keys. Delete the storage backend to fully clear credentials. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.logout", level = "info", skip_all, err(Debug)) + )] pub async fn logout(self: &Arc) -> Result<()> { use wacore::iq::devices::RemoveCompanionDeviceSpec; @@ -446,6 +469,10 @@ impl Client { Ok(()) } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.disconnect", level = "info", skip_all) + )] pub async fn disconnect(self: &Arc) { info!("Disconnecting client intentionally."); self.expected_disconnect.store(true, Ordering::Relaxed); @@ -490,6 +517,10 @@ impl Client { /// - Handling network changes (e.g., Wi-Fi → cellular) /// - Forcing a fresh server session /// - Testing offline message delivery + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.reconnect", level = "info", skip_all) + )] pub async fn reconnect(self: &Arc) { info!("Reconnecting: dropping transport for auto-reconnect."); self.intentional_reconnect.store(true, Ordering::Relaxed); @@ -512,6 +543,10 @@ impl Client { /// Unlike [`reconnect`], which introduces a deliberate offline window, /// this method sets the `expected_disconnect` flag so the run loop /// skips the backoff delay and reconnects as fast as possible. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.reconnect_immediately", level = "info", skip_all) + )] pub async fn reconnect_immediately(self: &Arc) { info!("Reconnecting immediately (expected disconnect)."); self.expected_disconnect.store(true, Ordering::Relaxed); @@ -527,6 +562,10 @@ impl Client { } } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.cleanup", level = "debug", skip_all) + )] pub(crate) async fn cleanup_connection_state(&self) { // Note: node_waiters are intentionally NOT cleared here — they are // cross-connection (callers may register a waiter before an action that diff --git a/src/client/messaging.rs b/src/client/messaging.rs index d060c1527..84e1a0d41 100644 --- a/src/client/messaging.rs +++ b/src/client/messaging.rs @@ -21,6 +21,7 @@ impl Client { Ok(()) } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.node", level = "debug", skip_all, fields(tag = %node.tag), err(Debug)))] pub async fn send_node(&self, node: Node) -> Result<(), ClientError> { debug!(target: "Client/Send", "{}", DisplayableNode(&node)); if self.sent_node_waiter_count.load(Ordering::Acquire) > 0 { @@ -35,6 +36,10 @@ impl Client { self.send_raw_bytes(plaintext_buf).await } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.send.unified_session", level = "debug", skip_all) + )] pub(crate) async fn send_unified_session(&self) { if !self.is_connected() { debug!(target: "Client/UnifiedSession", "Skipping: not connected"); @@ -51,6 +56,7 @@ impl Client { } } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.edit", level = "debug", skip_all, fields(to = %to.observe()), err(Debug)))] pub async fn edit_message( &self, to: Jid, @@ -104,6 +110,7 @@ impl Client { } /// Send a server-side reaction (used by both newsletter and status reactions). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.server_reaction", level = "debug", skip_all, fields(to = %to.observe()), err(Debug)))] pub(crate) async fn send_server_reaction( &self, to: &Jid, @@ -149,6 +156,10 @@ impl Client { } } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.send.protocol_receipt", level = "debug", skip_all) + )] pub(crate) async fn send_protocol_receipt( &self, id: String, @@ -194,6 +205,10 @@ impl Client { /// Dispatch a parsed chatstate stanza to registered handlers. /// /// Called by `ChatstateHandler` after parsing the incoming stanza. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.notif.chatstate", level = "debug", skip_all) + )] pub(crate) async fn dispatch_chatstate_event( &self, stanza: wacore::iq::chatstate::ChatstateStanza, diff --git a/src/client/node_io.rs b/src/client/node_io.rs index 5f0bf534a..f3a428e4f 100644 --- a/src/client/node_io.rs +++ b/src/client/node_io.rs @@ -30,6 +30,10 @@ impl Client { .fetch_add(1, Ordering::SeqCst); } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.read_loop", level = "debug", skip_all, err(Debug)) + )] pub(crate) async fn read_messages_loop(self: &Arc) -> Result<(), anyhow::Error> { debug!("Starting message processing loop..."); @@ -147,6 +151,10 @@ impl Client { /// Decrypt a frame and return the parsed node as a zero-copy OwnedNodeRef. /// This must be called sequentially due to noise protocol counter requirements. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.decrypt_frame", level = "trace", skip_all) + )] pub(crate) async fn decrypt_frame( self: &Arc, encrypted_frame: bytes::BytesMut, @@ -197,6 +205,10 @@ impl Client { } /// Process a node wrapped in Arc. Handlers receive the Arc and can share/store it cheaply. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.node", level = "trace", skip_all, fields(tag = %node.get().tag.as_ref())) + )] pub(crate) async fn process_node(self: &Arc, node: Arc) { use wacore::xml::DisplayableNodeRef; let nr = node.get(); @@ -415,6 +427,10 @@ impl Client { } /// Build and send an node corresponding to the given stanza. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.ack", level = "trace", skip_all, err(Debug)) + )] pub(crate) async fn send_ack_for( &self, node: &wacore_binary::NodeRef<'_>, @@ -495,6 +511,10 @@ impl Client { }); } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.success", level = "debug", skip_all) + )] pub(crate) async fn handle_success(self: &Arc, node: &wacore_binary::NodeRef<'_>) { // Skip processing if an expected disconnect is pending (e.g., 515 received). // This prevents race conditions where a spawned success handler runs after @@ -547,7 +567,7 @@ impl Client { let device_snapshot = client_clone.persistence_manager.get_device_snapshot().await; if device_snapshot.lid.as_ref() != Some(&lid) { - debug!("Updating LID from server to '{lid}'"); + debug!("Updating LID from server to '{}'", lid.observe()); client_clone .persistence_manager .process_command(DeviceCommand::SetLid(Some(lid))) @@ -879,6 +899,10 @@ impl Client { /// /// If an ack with an ID that matches a pending task in `response_waiters`, /// the task is resolved and the function returns `true`. Otherwise, returns `false`. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.ack_response", level = "debug", skip_all) + )] pub(crate) async fn handle_ack_response(&self, node: &wacore_binary::NodeRef<'_>) -> bool { // Surface server nack codes for diagnosability. A nacked send still // resolves Ok to the caller, so without this the failure is invisible. @@ -939,6 +963,10 @@ impl Client { false } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.stream_error", level = "debug", skip_all) + )] pub(crate) async fn handle_stream_error(&self, node: &wacore_binary::NodeRef<'_>) { // 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 @@ -1091,6 +1119,10 @@ impl Client { } } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.connect_failure", level = "debug", skip_all) + )] pub(crate) async fn handle_connect_failure(&self, node: &wacore_binary::NodeRef<'_>) { self.expected_disconnect.store(true, Ordering::Relaxed); self.notify_connection_shutdown(); @@ -1156,6 +1188,10 @@ impl Client { } } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.iq_in", level = "debug", skip_all) + )] pub(crate) async fn handle_iq(self: &Arc, node: &wacore_binary::NodeRef<'_>) -> bool { if node.get_attr("type").is_some_and(|s| s.as_str() == "get") && (node.get_optional_child("ping").is_some() diff --git a/src/client/offline_resume.rs b/src/client/offline_resume.rs index c0d67ba51..854e0352c 100644 --- a/src/client/offline_resume.rs +++ b/src/client/offline_resume.rs @@ -90,6 +90,7 @@ pub(crate) fn build_offline_batch_request(count: u32) -> Node { } /// Called from `IbHandler` on `` with N > 0. +#[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.offline_first_batch", level = "debug", skip_all, fields(total = total)))] pub(crate) async fn send_first_batch(client: Arc, total: usize) { let generation = client.connection_generation.load(Ordering::Acquire); client.offline_batch.arm(generation); diff --git a/src/client/sender_keys.rs b/src/client/sender_keys.rs index 4b6ed0eb1..154b67fc4 100644 --- a/src/client/sender_keys.rs +++ b/src/client/sender_keys.rs @@ -8,6 +8,7 @@ use waproto::whatsapp as wa; use super::Client; impl Client { + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.set_sender_key_status", level = "debug", skip_all, fields(count = device_jids.len(), has_key = has_key), err(Debug)))] pub(crate) async fn set_sender_key_status_for_devices( &self, group_jid: &str, @@ -57,6 +58,7 @@ impl Client { /// Mark device JIDs as needing fresh SKDM (has_key = false). /// Filters out our own devices (WA Web: `!isMeDevice(e)` check). /// Called from handle_retry_receipt for group/status messages. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.mark_forget_sender_key", level = "debug", skip_all, fields(count = device_jids.len()), err(Debug)))] pub(crate) async fn mark_forget_sender_key( &self, group_jid: &str, @@ -73,6 +75,7 @@ impl Client { /// the group and wipe `sender_key_devices` so the next send takes the /// `force_skdm=true` path (`!key_exists`) and redistributes to all /// remaining participants. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.rotate_sender_key_on_remove", level = "debug", skip_all, fields(removed = removed_user_ids.len())))] pub(crate) async fn rotate_sender_key_on_participant_remove( &self, group_jid: &str, @@ -141,6 +144,7 @@ impl Client { /// alternate PN/LID key, `alternate_chat` contains the namespace that /// matched -- the caller should use it for session operations instead of /// `resolve_encryption_jid` (which would map back to the primary). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.take_recent_message", level = "debug", skip_all, fields(peer = %to.observe())))] pub(crate) async fn take_recent_message( &self, to: &Jid, @@ -164,9 +168,9 @@ impl Client { if let Some(alt_chat) = alt_chat { log::debug!( "Primary key miss for {}:{}, trying alternate {}", - primary_key.chat, + primary_key.chat.observe(), id, - alt_chat + alt_chat.observe() ); let alt_key = ChatMessageId { chat: alt_chat, @@ -203,7 +207,7 @@ impl Client { } log::warn!( "Failed to decode cached message for {}:{}, trying DB", - key.chat, + key.chat.observe(), key.id ); } @@ -220,7 +224,7 @@ impl Client { Err(e) => { log::warn!( "Failed to decode DB message for {}:{}: {}", - key.chat, + key.chat.observe(), key.id, e ); @@ -231,7 +235,7 @@ impl Client { Err(e) => { log::warn!( "Failed to read sent message from DB for {}:{}: {}", - key.chat, + key.chat.observe(), key.id, e ); @@ -244,6 +248,7 @@ impl Client { /// is enabled (capacity > 0) also stores in-memory for fast retrieval. /// In DB-only mode (capacity = 0), the DB write is awaited to guarantee persistence. /// With L1 cache, the DB write is backgrounded since the cache serves reads immediately. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.add_recent_message", level = "debug", skip_all, fields(peer = %to.observe())))] pub(crate) async fn add_recent_message(&self, to: &Jid, id: &str, msg: &wa::Message) { use prost::Message; let key = self.make_chat_message_id(to, id).await; diff --git a/src/client/sessions.rs b/src/client/sessions.rs index badb93c36..8b7c0ea00 100644 --- a/src/client/sessions.rs +++ b/src/client/sessions.rs @@ -146,6 +146,7 @@ impl Client { /// Ensure E2E sessions exist for the given device JIDs. /// Waits for offline delivery, resolves LID mappings, then batches prekey fetches. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.ensure", level = "debug", skip_all, fields(count = device_jids.len()), err(Debug)))] pub(crate) async fn ensure_e2e_sessions(&self, device_jids: &[Jid]) -> Result<()> { if device_jids.is_empty() { return Ok(()); @@ -158,6 +159,7 @@ impl Client { /// Like `ensure_e2e_sessions` but skips `resolve_lid_mappings`. Use when the /// caller already resolved JIDs to the correct namespace (e.g., after /// alternate PN/LID key normalization in retry handling). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.ensure_resolved", level = "debug", skip_all, fields(count = jids.len()), err(Debug)))] pub(crate) async fn ensure_e2e_sessions_resolved(&self, jids: &[Jid]) -> Result<()> { if jids.is_empty() { return Ok(()); @@ -167,6 +169,7 @@ impl Client { } /// Core session-check + prekey-fetch logic shared by both entry points. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.ensure_inner", level = "debug", skip_all, fields(count = jids.len()), err(Debug)))] async fn ensure_sessions_inner(&self, jids: Vec) -> Result<()> { use wacore::types::jid::JidExt; @@ -185,7 +188,7 @@ impl Client { { Ok(true) => {} Ok(false) => jids_needing_sessions.push(jid), - Err(e) => log::warn!("Failed to check session for {}: {}", jid, e), + Err(e) => log::warn!("Failed to check session for {}: {}", jid.observe(), e), } } } @@ -203,6 +206,7 @@ impl Client { /// Fetch prekeys and establish sessions for a batch of JIDs. /// Returns the number of sessions successfully established. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.fetch_establish", level = "debug", skip_all, fields(count = jids.len()), err(Debug)))] async fn fetch_and_establish_sessions(&self, jids: &[Jid]) -> Result { use wacore::libsignal::protocol::{UsePQRatchet, process_prekey_bundle}; use wacore::types::jid::JidExt; @@ -241,7 +245,7 @@ impl Client { { Ok(identity_change) => { success_count += 1; - log::debug!("Successfully established session with {}", jid); + log::debug!("Successfully established session with {}", jid.observe()); if identity_change == wacore::libsignal::protocol::IdentityChange::ReplacedExisting { @@ -250,15 +254,18 @@ impl Client { } Err(e) => { failed_count += 1; - log::warn!("Failed to establish session with {}: {}", jid, e); + log::warn!("Failed to establish session with {}: {}", jid.observe(), e); } } } else { missing_count += 1; if jid.device == 0 { - log::warn!("Server did not return prekeys for primary phone {}", jid); + log::warn!( + "Server did not return prekeys for primary phone {}", + jid.observe() + ); } else { - log::debug!("Server did not return prekeys for {}", jid); + log::debug!("Server did not return prekeys for {}", jid.observe()); } } } @@ -283,6 +290,15 @@ impl Client { /// Log primary phone (device 0) session state at login. /// Migration is lazy via try_pn_to_lid_migration_decrypt on first message. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.primary_phone_check", + level = "debug", + skip_all, + err(Debug) + ) + )] pub(crate) async fn establish_primary_phone_session_immediate(&self) -> Result<()> { let device_snapshot = self.persistence_manager.get_device_snapshot().await; @@ -309,7 +325,7 @@ impl Client { .unwrap_or(false); match (lid_exists, pn_exists) { - (true, _) => log::debug!("LID session with {} exists", primary_phone_lid), + (true, _) => log::debug!("LID session with {} exists", primary_phone_lid.observe()), (false, true) => { log::debug!("PN-only session for own device 0 — will migrate on first message") } @@ -330,7 +346,7 @@ impl Client { device_guard .contains_session(&signal_addr) .await - .map_err(|e| anyhow::anyhow!("Failed to check session for {}: {}", jid, e)) + .map_err(|e| anyhow::anyhow!("Failed to check session for {}: {}", jid.observe(), e)) } } diff --git a/src/download.rs b/src/download.rs index 3bf91da62..c45c57332 100644 --- a/src/download.rs +++ b/src/download.rs @@ -232,6 +232,10 @@ impl Client { /// Only needed when you need the plaintext bytes (processing, transcoding, /// re-upload). To forward existing media unchanged, reuse the original /// message's CDN fields directly, no round-trip required. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.media.download", level = "debug", skip_all, err(Debug)) + )] pub async fn download(&self, downloadable: &dyn Downloadable) -> Result> { download_media_with_retry( |force| self.prepare_requests(downloadable, force), @@ -241,6 +245,15 @@ impl Client { .await } + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.media.download_to_file", + level = "debug", + skip_all, + err(Debug) + ) + )] pub async fn download_to_file( &self, downloadable: &dyn Downloadable, @@ -257,6 +270,15 @@ impl Client { /// Each returned [`wacore::sticker_pack::StickerPackItem`] is [`Downloadable`], /// so individual stickers can be fetched with [`Self::download`]. The locale /// only affects localized pack names; `"en"` mirrors whatsmeow's default. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.media.fetch_sticker_pack", + level = "debug", + skip_all, + err(Debug) + ) + )] pub async fn fetch_sticker_pack( &self, pack_id: &str, @@ -278,6 +300,7 @@ impl Client { } /// Downloads and decrypts media from raw parameters without needing the original message. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.media.download_from_params", level = "debug", skip_all, fields(kind = ?media_type), err(Debug)))] pub async fn download_from_params( &self, direct_path: &str, @@ -361,6 +384,15 @@ impl Client { /// blocking thread. The writer is seeked back to position 0 before returning. /// /// Memory usage: ~40KB regardless of file size (8KB read buffer + decrypt state). + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.media.download_to_writer", + level = "debug", + skip_all, + err(Debug) + ) + )] pub async fn download_to_writer( &self, downloadable: &dyn Downloadable, @@ -378,6 +410,7 @@ impl Client { /// Streaming variant of `download_from_params` that writes to a writer /// instead of buffering in memory. #[allow(clippy::too_many_arguments)] + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.media.download_from_params_to_writer", level = "debug", skip_all, fields(kind = ?media_type), err(Debug)))] pub async fn download_from_params_to_writer( &self, direct_path: &str, diff --git a/src/handlers/basic.rs b/src/handlers/basic.rs index 77d6f8664..44d58445d 100644 --- a/src/handlers/basic.rs +++ b/src/handlers/basic.rs @@ -15,6 +15,10 @@ impl StanzaHandler for SuccessHandler { "success" } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.success", level = "debug", skip_all) + )] async fn handle( &self, client: Arc, @@ -37,6 +41,10 @@ impl StanzaHandler for FailureHandler { "failure" } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.failure", level = "debug", skip_all) + )] async fn handle( &self, client: Arc, @@ -59,6 +67,10 @@ impl StanzaHandler for StreamErrorHandler { "stream:error" } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.stream_error", level = "debug", skip_all) + )] async fn handle( &self, client: Arc, diff --git a/src/handlers/call.rs b/src/handlers/call.rs index df3d233b4..676036737 100644 --- a/src/handlers/call.rs +++ b/src/handlers/call.rs @@ -24,6 +24,10 @@ impl StanzaHandler for CallHandler { "call" } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.call", level = "debug", skip_all) + )] async fn handle( &self, client: Arc, @@ -51,6 +55,7 @@ impl StanzaHandler for CallHandler { } } +#[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.call_offer_ack", level = "debug", skip_all, fields(peer = %call.from.observe()), err(Debug)))] async fn send_offer_ack_receipt(client: &Client, call: &IncomingCall) -> anyhow::Result<()> { let own_from = match call.from.server { Server::Lid => client.get_lid().await, diff --git a/src/handlers/chatstate.rs b/src/handlers/chatstate.rs index 13f9bff7f..a2b2a9adb 100644 --- a/src/handlers/chatstate.rs +++ b/src/handlers/chatstate.rs @@ -53,6 +53,10 @@ impl StanzaHandler for ChatstateHandler { "chatstate" } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.chatstate", level = "debug", skip_all) + )] async fn handle( &self, client: Arc, diff --git a/src/handlers/ib.rs b/src/handlers/ib.rs index 23bb432b7..9b43adac7 100644 --- a/src/handlers/ib.rs +++ b/src/handlers/ib.rs @@ -36,6 +36,10 @@ impl StanzaHandler for IbHandler { } } +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.ib", level = "debug", skip_all) +)] async fn handle_ib_impl(client: Arc, node: &wacore_binary::NodeRef<'_>) { for child in node.children().unwrap_or_default() { match child.tag.as_ref() { diff --git a/src/handlers/iq.rs b/src/handlers/iq.rs index a33551274..290f0e7a0 100644 --- a/src/handlers/iq.rs +++ b/src/handlers/iq.rs @@ -22,6 +22,10 @@ impl StanzaHandler for IqHandler { "iq" } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.iq", level = "debug", skip_all) + )] async fn handle( &self, client: Arc, diff --git a/src/handlers/message.rs b/src/handlers/message.rs index baa5cadb3..7c9b8cb85 100644 --- a/src/handlers/message.rs +++ b/src/handlers/message.rs @@ -22,6 +22,10 @@ impl StanzaHandler for MessageHandler { "message" } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.message_enqueue", level = "debug", skip_all) + )] async fn handle( &self, client: Arc, diff --git a/src/handlers/notification.rs b/src/handlers/notification.rs index a46337e6d..c702cd6ed 100755 --- a/src/handlers/notification.rs +++ b/src/handlers/notification.rs @@ -48,6 +48,10 @@ impl StanzaHandler for NotificationHandler { /// Dispatch notification by type. Each arm calls a separate async fn so the /// compiler doesn't size this future for all arms simultaneously. +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.notif.dispatch", level = "debug", skip_all) +)] async fn handle_notification_impl(client: &Arc, node: Arc) { let nr = node.get(); let notification_type = nr.attrs().optional_string("type"); @@ -300,6 +304,10 @@ fn handle_digest_key(client: &Arc) { /// /// WA Web defers this when offline. We process immediately because all cleanup /// is local-only, and `ensure_e2e_sessions` self-defers via `wait_for_offline_delivery_end`. +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.notif.identity_change", level = "debug", skip_all) +)] async fn handle_identity_change(client: &Arc, node: &NodeRef<'_>) { let from_jid = crate::require_from_jid!(node, "Identity change notification"); @@ -307,7 +315,7 @@ async fn handle_identity_change(client: &Arc, node: &NodeRef<'_>) { if from_jid.device != 0 { debug!( "Ignoring identity change from companion device {}", - from_jid + from_jid.observe() ); return; } @@ -390,7 +398,8 @@ async fn handle_identity_change(client: &Arc, node: &NodeRef<'_>) { Ok(None) => {} Err(e) => { warn!( - "Identity change: failed reading stored identity for {cand}: {e}; proceeding with reset" + "Identity change: failed reading stored identity for {}: {e}; proceeding with reset", + wacore::types::jid::observe_protocol_address(cand) ); had_prior_identity = true; break; @@ -511,6 +520,7 @@ async fn handle_identity_change(client: &Arc, node: &NodeRef<'_>) { /// server push's job (which reliably follows). This matches WA Web, where the /// local `handleNewIdentity` omits those steps that only the server-push /// `handleE2eIdentityChange` performs. +#[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.notif.local_identity_change", level = "debug", skip_all, fields(sender = %sender.observe())))] pub(crate) async fn handle_local_identity_change(client: &Arc, sender: Jid) { // Only a peer's primary-device identity change matters; companion devices // carry their own identities (WA Web ignores them on this path). @@ -574,6 +584,10 @@ pub(crate) async fn handle_local_identity_change(client: &Arc, sender: J /// /// /// ``` +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.notif.devices", level = "debug", skip_all) +)] async fn handle_devices_notification(client: &Arc, node: &NodeRef<'_>) { let notification = match DeviceNotification::try_parse(node) { Ok(n) => n, @@ -721,7 +735,7 @@ async fn handle_account_sync_devices( warn!( target: "Client/AccountSync", "Received account_sync devices for non-self user: {} (our PN: {:?}, LID: {:?})", - from_jid, + from_jid.observe(), own_pn.map(|j| j.user.as_str()), own_lid.map(|j| j.user.as_str()) ); @@ -791,7 +805,7 @@ async fn handle_account_sync_devices( debug!( target: "Client/AccountSync", " Device: {} (key-index: {:?})", - device.jid, + device.jid.observe(), device.key_index ); } @@ -810,6 +824,10 @@ async fn handle_account_sync_devices( /// /// /// ``` +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.notif.privacy_token", level = "debug", skip_all) +)] async fn handle_privacy_token_notification(client: &Arc, node: &NodeRef<'_>) { use wacore::iq::tctoken::parse_privacy_token_notification; use wacore::store::traits::TcTokenEntry; @@ -847,7 +865,7 @@ async fn handle_privacy_token_notification(client: &Arc, node: &NodeRef< debug!( target: "Client/TcToken", "Cannot resolve LID for privacy_token sender {}, storing under PN", - from + from.observe() ); &from.user } @@ -940,7 +958,7 @@ async fn handle_privacy_token_notification(client: &Arc, node: &NodeRef< && let Some(from) = &from_jid && let Err(e) = client.presence().re_subscribe_when_active(from).await { - debug!(target: "Client/TcToken", "Failed to re-subscribe presence for {from}: {e}"); + debug!(target: "Client/TcToken", "Failed to re-subscribe presence for {}: {e}", from.observe()); } } @@ -957,7 +975,7 @@ async fn handle_business_notification(client: &Arc, node: &NodeRef<'_>) debug!( target: "Client/Business", "Business notification: from={}, type={}, jid={:?}", - notification.from, + notification.from.observe(), notification.notification_type, notification.jid ); @@ -986,7 +1004,7 @@ async fn handle_business_notification(client: &Arc, node: &NodeRef<'_>) info!( target: "Client/Business", "Contact {} is no longer a business account", - notification.from + notification.from.observe() ); } wacore::stanza::business::BusinessNotificationType::VerifiedNameJid @@ -999,7 +1017,7 @@ async fn handle_business_notification(client: &Arc, node: &NodeRef<'_>) info!( target: "Client/Business", "Contact {} verified business name: {}", - notification.from, + notification.from.observe(), name ); } @@ -1009,7 +1027,7 @@ async fn handle_business_notification(client: &Arc, node: &NodeRef<'_>) debug!( target: "Client/Business", "Contact {} business profile updated (hash: {:?})", - notification.from, + notification.from.observe(), notification.hash ); } @@ -1055,7 +1073,7 @@ fn handle_picture_notification(client: &Arc, node: &NodeRef<'_>) { if set_node.attrs().optional_string("hash").is_some() { debug!( target: "Client/Picture", - "Hash-based picture notification (no jid), using from={}", from + "Hash-based picture notification (no jid), using from={}", from.observe() ); } from.clone() @@ -1091,7 +1109,7 @@ fn handle_picture_notification(client: &Arc, node: &NodeRef<'_>) { .and_then(|c| c.first().map(|n| n.tag.as_ref())); debug!( target: "Client/Picture", - "Ignoring picture notification with child {:?} from {}", child_tag, from + "Ignoring picture notification with child {:?} from {}", child_tag, from.observe() ); return; } @@ -1101,7 +1119,7 @@ fn handle_picture_notification(client: &Arc, node: &NodeRef<'_>) { target: "Client/Picture", "Picture {}: jid={}, author={:?}, pic_id={:?}", if removed { "removed" } else { "updated" }, - jid, author, picture_id + jid.observe(), author, picture_id ); let event = Event::PictureUpdate(PictureUpdate { @@ -1142,7 +1160,7 @@ fn handle_status_notification(client: &Arc, node: &NodeRef<'_>) { debug!( target: "Client/Status", - "Status update from {} (length={})", from, status_text.len() + "Status update from {} (length={})", from.observe(), status_text.len() ); let event = Event::UserAboutUpdate(UserAboutUpdate { @@ -1154,7 +1172,7 @@ fn handle_status_notification(client: &Arc, node: &NodeRef<'_>) { } else { debug!( target: "Client/Status", - "Status notification from {} without child, ignoring", from + "Status notification from {} without child, ignoring", from.observe() ); } } @@ -1193,7 +1211,7 @@ async fn learn_contact_modify_mappings( warn!( target: "Client/Contacts", "Failed to add LID-PN mapping lid={} pn={}: {e}", - lid, pn + lid.observe(), pn.observe() ); } } @@ -1201,7 +1219,7 @@ async fn learn_contact_modify_mappings( debug!( target: "Client/Contacts", "Contacts modify without old_lid/new_lid, skipping LID-PN mapping (old={}, new={})", - old_pn, new_pn + old_pn.observe(), new_pn.observe() ); } } @@ -1240,7 +1258,7 @@ async fn handle_contacts_notification(client: &Arc, node: &NodeRef<'_>) return; }; - debug!(target: "Client/Contacts", "Contact updated for {}", jid); + debug!(target: "Client/Contacts", "Contact updated for {}", jid.observe()); client .core .event_bus @@ -1272,7 +1290,7 @@ async fn handle_contacts_notification(client: &Arc, node: &NodeRef<'_>) debug!( target: "Client/Contacts", "Contact number changed: {} -> {} (old_lid={:?}, new_lid={:?})", - old_jid, new_jid, old_lid, new_lid + old_jid.observe(), new_jid.observe(), old_lid, new_lid ); client .core @@ -1327,6 +1345,10 @@ async fn handle_contacts_notification(client: &Arc, node: &NodeRef<'_>) /// and dispatches typed `Event::GroupUpdate` events for each. /// /// Reference: WhatsApp Web `WAWebHandleGroupNotification` (Ri7Gf1BxhsX.js:12556-12962) +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.notif.group", level = "debug", skip_all) +)] async fn handle_group_notification(client: &Arc, node: Arc) { let notification = match GroupNotification::try_from_node_ref(node.get()) { Some(n) => n, @@ -1361,7 +1383,7 @@ async fn handle_group_notification(client: &Arc, node: Arc debug!( target: "Client/Group", "Patched group cache for {}: added {} participants", - notification.group_jid, participants.len() + notification.group_jid.observe(), participants.len() ); } } @@ -1377,7 +1399,7 @@ async fn handle_group_notification(client: &Arc, node: Arc debug!( target: "Client/Group", "Patched group cache for {}: removed {} participants", - notification.group_jid, participants.len() + notification.group_jid.observe(), participants.len() ); } client @@ -1393,7 +1415,7 @@ async fn handle_group_notification(client: &Arc, node: Arc debug!( target: "Client/Group", "Group notification: group={}, action={}", - notification.group_jid, action.tag_name() + notification.group_jid.observe(), action.tag_name() ); client @@ -1595,7 +1617,9 @@ fn handle_disappearing_mode_notification(client: &Arc, node: &NodeRef<'_ debug!( "Disappearing mode changed for {}: duration={}s, t={}", - from, duration, setting_timestamp + from.observe(), + duration, + setting_timestamp ); client diff --git a/src/handlers/presence.rs b/src/handlers/presence.rs index 71ca1ac8b..918d1301d 100644 --- a/src/handlers/presence.rs +++ b/src/handlers/presence.rs @@ -20,6 +20,10 @@ impl StanzaHandler for PresenceHandler { "presence" } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.presence", level = "debug", skip_all) + )] async fn handle( &self, client: Arc, @@ -49,7 +53,7 @@ impl StanzaHandler for PresenceHandler { debug!( target: "PresenceHandler", "Received presence from {}: unavailable={}", - from_jid, unavailable + from_jid.observe(), unavailable ); client diff --git a/src/handshake.rs b/src/handshake.rs index 1235294dd..f6cfdfe23 100644 --- a/src/handshake.rs +++ b/src/handshake.rs @@ -152,6 +152,10 @@ fn should_persist_cert_chain(device: &wacore::store::Device) -> bool { device.is_registered() } +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.handshake", level = "debug", skip_all, err(Debug)) +)] pub async fn do_handshake( runtime: Arc, persistence_manager: &PersistenceManager, @@ -231,6 +235,10 @@ pub async fn do_handshake( } } +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.handshake.xx", level = "debug", skip_all, err(Debug)) +)] async fn run_xx_handshake( runtime: &Arc, device: &wacore::store::Device, @@ -268,6 +276,10 @@ async fn run_xx_handshake( /// `fallback_taken` is set to `true` once we pivot from IK to XXfallback, /// before any operation that could fail. +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.handshake.ik", level = "debug", skip_all, err(Debug)) +)] async fn run_ik_handshake( runtime: &Arc, device: &wacore::store::Device, diff --git a/src/history_sync.rs b/src/history_sync.rs index da940c22b..d4273934a 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -8,6 +8,7 @@ use waproto::whatsapp::message::HistorySyncNotification; use crate::client::Client; impl Client { + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.media.history_sync", level = "debug", skip_all, fields(msg_id = %message_id)))] pub(crate) async fn handle_history_sync( self: &Arc, message_id: String, @@ -60,6 +61,7 @@ impl Client { /// Process history sync: decompress, extract internal data (tctokens, /// pushname, nct_salt), then dispatch a single `Event::HistorySync` /// with the full decompressed blob for on-demand consumer decoding. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.media.history_sync_task", level = "debug", skip_all, fields(msg_id = %message_id)))] pub(crate) async fn process_history_sync_task( self: &Arc, message_id: String, @@ -375,6 +377,7 @@ impl Client { /// download failure; the encrypted payload is the same `ServerErrorReceipt` /// used for media retries. Exposed for consumers that detect an undownloadable /// or unwanted history-sync chunk and want the phone to re-send it. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.media.history_sync_error_receipt", level = "debug", skip_all, fields(msg_id = %message_id), err(Debug)))] pub async fn send_history_sync_server_error_receipt( &self, message_id: &str, diff --git a/src/keepalive.rs b/src/keepalive.rs index e856ed9bc..874cd01af 100644 --- a/src/keepalive.rs +++ b/src/keepalive.rs @@ -50,6 +50,10 @@ impl Client { /// the pong's `t` attribute using RTT-adjusted midpoint calculation. /// /// WA Web: `sendPing` → `onClockSkewUpdate(Math.round((start + rtt/2) / 1000 - serverTime))` + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.keepalive.ping", level = "debug", skip_all) + )] async fn send_keepalive(&self) -> KeepaliveResult { if !self.is_connected() { return KeepaliveResult::FatalFailure; @@ -92,6 +96,10 @@ impl Client { } } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.conn.keepalive", level = "debug", skip_all) + )] pub(crate) async fn keepalive_loop(self: Arc) { let mut error_count = 0u32; let mut cleanup_counter = 0u32; diff --git a/src/lib.rs b/src/lib.rs index 8f08d844c..77f37e129 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,8 @@ +// Instrumenting large async fns (e.g. process_sync_task) wraps them in deep +// `Instrumented` future types; the default depth limit overflows when the +// `tracing` + `tracing-pii` paths combine. Raise it (compile-time only). +#![recursion_limit = "512"] + pub use wacore::appstate::schemas; pub use wacore::client_profile::ClientProfile; pub use wacore::{ diff --git a/src/main.rs b/src/main.rs index 87d9ddf78..d9b333ddb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,3 +1,7 @@ +// See the matching note in lib.rs: instrumented large async fns need a deeper +// recursion limit when the `tracing` + `tracing-pii` paths combine. +#![recursion_limit = "512"] + use log::{error, info}; use std::sync::Arc; use wacore::proto_helpers::MessageExt; diff --git a/src/mediaconn.rs b/src/mediaconn.rs index dcd881e4b..febf3e907 100644 --- a/src/mediaconn.rs +++ b/src/mediaconn.rs @@ -50,6 +50,16 @@ impl Client { *self.media_conn.write().await = None; } + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.media.refresh_conn", + level = "debug", + skip_all, + fields(force), + err(Debug) + ) + )] pub async fn refresh_media_conn(&self, force: bool) -> Result { { let guard = self.media_conn.read().await; diff --git a/src/message/dispatch.rs b/src/message/dispatch.rs index b8f57461b..92a477937 100644 --- a/src/message/dispatch.rs +++ b/src/message/dispatch.rs @@ -4,6 +4,7 @@ use super::*; impl Client { /// Dispatches a successfully parsed message to the event bus and sends a delivery receipt. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.dispatch", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id)))] pub(crate) async fn dispatch_parsed_message( self: &Arc, msg: wa::Message, diff --git a/src/message/msg_secret.rs b/src/message/msg_secret.rs index 1637bbedd..31d62969a 100644 --- a/src/message/msg_secret.rs +++ b/src/message/msg_secret.rs @@ -5,6 +5,7 @@ use super::*; impl Client { /// Capture embedded `MessageContextInfo.message_secret` for add-on /// decrypts. Bot DMs keep the legacy LID key as a second entry. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.capture_secret", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), msg_id = %info.id)))] pub(crate) async fn maybe_capture_inbound_msg_secret( self: &Arc, msg: &wa::Message, @@ -153,6 +154,7 @@ impl Client { } } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.decrypt_secret", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), msg_id = %info.id)))] pub(crate) async fn maybe_decrypt_secret_encrypted_message( self: &Arc, msg: &wa::Message, @@ -407,6 +409,7 @@ impl Client { /// outbound `messageSecret` we persisted at send time and runs the /// dual-HKDF + AES-GCM open from [`wacore::bot_message`]. Failures /// (missing secret, GCM tag fail, malformed proto) nack with code 495. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.msmsg", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id)))] pub(crate) async fn handle_msmsg_payload( self: &Arc, info: &Arc, @@ -638,7 +641,7 @@ impl Client { log::info!( "[msg:{}] Successfully decrypted msmsg bot reply from {}", info.id, - info.source.sender + info.source.sender.observe() ); self.dispatch_parsed_message(msg, info).await; } diff --git a/src/message/receive.rs b/src/message/receive.rs index 98c6a6531..36bbb44c4 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -3,6 +3,10 @@ use super::*; impl Client { + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.incoming", level = "debug", skip_all) + )] pub(crate) async fn handle_incoming_message(self: Arc, node: Arc) { // Phase 1: classify borrows the node tree, extracts owned payloads, returns quickly. // Phase 2: process_classified_message holds no node borrows across heavy .await points, @@ -16,6 +20,10 @@ impl Client { self.process_classified_message(classified).await; } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.classify", level = "debug", skip_all) + )] pub(crate) async fn classify_incoming_message( self: &Arc, node: &OwnedNodeRef, @@ -222,7 +230,7 @@ impl Client { "[msg:{}] Protocol violation: skmsg is first in multi-enc message from {}. \ Expected pkmsg/msg first (containing SKDM).", info.id, - info.source.sender + info.source.sender.observe() ); } @@ -263,6 +271,10 @@ impl Client { } /// Phase 2: acquire permit, decrypt payloads, flush. No node borrows. + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.process", level = "debug", skip_all) + )] pub(crate) async fn process_classified_message(self: Arc, msg: ClassifiedMessage) { let ClassifiedMessage { info, @@ -343,7 +355,7 @@ impl Client { log::debug!( "Skipping {} session messages from group sender {}", session_payloads.len(), - sender_encryption_jid + sender_encryption_jid.observe() ); } SessionBatchOutcome::default() @@ -388,8 +400,8 @@ impl Client { log::warn!( "[msg:{}] Batch group decrypt from {} in {} failed: {e:?}", info.id, - info.source.sender, - info.source.chat + info.source.sender.observe(), + info.source.chat.observe() ); } } @@ -400,14 +412,14 @@ impl Client { log::debug!( "[msg:{}] Silently dropping expired status from {}", info.id, - info.source.sender + info.source.sender.observe() ); } else { log::log!( decrypt_fail_log_level(decrypt_fail_mode), "Skipping skmsg decryption for message {} from {} because pkmsg failed to decrypt.", info.id, - info.source.sender + info.source.sender.observe() ); if !session_dispatched_undecryptable { self.dispatch_undecryptable_event( @@ -438,7 +450,7 @@ impl Client { decrypt_fail_log_level(decrypt_fail_mode), "Message {} from {} failed to decrypt and has no group content. Dispatching UndecryptableMessage event.", info.id, - info.source.sender + info.source.sender.observe() ); // Dispatch UndecryptableMessage event for messages that failed to decrypt // (This should not cause double-dispatching since process_session_enc_batch @@ -485,6 +497,7 @@ impl Client { .await; } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.session_decrypt", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %sender_encryption_jid.observe(), msg_id = %info.id)))] pub(crate) async fn process_session_enc_batch( self: Arc, payloads: &[EncPayload], @@ -534,7 +547,7 @@ impl Client { log::error!( "[msg:{}] Failed to parse PreKeySignalMessage from {}: {e:?}. Sending nack.", info.id, - info.source.sender + info.source.sender.observe() ); // |= so a later dedup'd return (false) can't clobber // a true set by a prior iteration in this batch. @@ -558,7 +571,7 @@ impl Client { log::error!( "[msg:{}] Failed to parse SignalMessage from {}: {e:?}. Sending nack.", info.id, - info.source.sender + info.source.sender.observe() ); outcome.had_failure = true; outcome.undecryptable |= self @@ -657,7 +670,7 @@ impl Client { log::warn!( "[msg:{}] Failed processing plaintext from {}: {e:?}", info.id, - info.source.sender + info.source.sender.observe() ); outcome.decrypted = true; outcome.plaintext_failed = true; @@ -672,7 +685,7 @@ impl Client { if let SignalProtocolError::DuplicatedMessage(chain, counter) = e { log::debug!( "Skipping already-processed message from {} (chain {}, counter {}). This is normal during reconnection.", - info.source.sender, + info.source.sender.observe(), chain, counter ); @@ -701,7 +714,10 @@ impl Client { // Flush immediately so the backend is updated BEFORE the retry decrypt below. // Device::is_trusted_identity reads from backend, not cache. if let Err(e) = self.flush_signal_cache().await { - log::warn!("Failed to flush identity deletion for {}: {e:?}", address); + log::warn!( + "Failed to flush identity deletion for {}: {e:?}", + wacore::types::jid::observe_protocol_address(address) + ); outcome.had_failure = true; continue; } @@ -917,7 +933,9 @@ impl Client { debug!( "[msg:{}] No session found for {} message from {}. Sending retry receipt to request session establishment.", - info.id, enc_type, info.source.sender + info.id, + enc_type, + info.source.sender.observe() ); outcome.had_failure = true; outcome.undecryptable |= self @@ -974,7 +992,7 @@ impl Client { Sending retry receipt.", info.id, enc_type, - info.source.sender + info.source.sender.observe() ); outcome.had_failure = true; @@ -1024,7 +1042,7 @@ impl Client { Sending retry receipt with fresh prekeys.", info.id, enc_type, - info.source.sender + info.source.sender.observe() ); // Send retry receipt with fresh prekeys @@ -1043,7 +1061,7 @@ impl Client { "[msg:{}] Batch session decrypt failed (type: {}) from {}: {:?}. Sending nack.", info.id, enc_type, - info.source.sender, + info.source.sender.observe(), e ); outcome.had_failure = true; @@ -1064,6 +1082,7 @@ impl Client { outcome } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.group_decrypt", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id), err(Debug)))] async fn process_group_enc_batch( self: Arc, payloads: &[EncPayload], @@ -1090,9 +1109,9 @@ impl Client { log::debug!( "Looking up sender key for group {} with sender address {} (from sender JID: {})", - info.source.chat, + info.source.chat.observe(), sender_address, - info.source.sender + info.source.sender.observe() ); let decrypt_result = @@ -1107,7 +1126,8 @@ impl Client { if !self.is_from_known_device(&info.source.sender).await { debug!( "[msg:{}] Unknown device {}, triggering device sync", - info.id, info.source.sender + info.id, + info.source.sender.observe() ); self.handle_unknown_device_sync(info).await; } @@ -1128,8 +1148,8 @@ impl Client { Err(SignalProtocolError::DuplicatedMessage(iteration, counter)) => { log::debug!( "Skipping already-processed sender key message from {} in group {} (iteration {}, counter {}). This is normal during reconnection.", - info.source.sender, - info.source.chat, + info.source.sender.observe(), + info.source.chat.observe(), iteration, counter ); @@ -1145,7 +1165,7 @@ impl Client { log::debug!( "[msg:{}] Skipping retry for expired status from {}", info.id, - info.source.sender + info.source.sender.observe() ); continue; } @@ -1159,7 +1179,9 @@ impl Client { debug!( "No sender key state for group message [msg:{}] from {}: {}. Sending retry receipt.", - info.id, info.source.sender, msg + info.id, + info.source.sender.observe(), + msg ); if is_unknown_device { @@ -1174,7 +1196,7 @@ impl Client { log::debug!( "[msg:{}] Ignoring decrypt error for expired status from {}: {:?}", info.id, - info.source.sender, + info.source.sender.observe(), e ); continue; @@ -1208,6 +1230,7 @@ impl Client { } /// WA Web: online → `syncDeviceListJob`, offline → `OfflinePendingDeviceCache`. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.unknown_device_sync", level = "debug", skip_all, fields(sender = %info.source.sender.observe(), msg_id = %info.id)))] async fn handle_unknown_device_sync(self: &Arc, info: &MessageInfo) { let user_jid = info.source.sender.to_non_ad(); @@ -1217,9 +1240,15 @@ impl Client { } if info.is_offline { - log::debug!("Queueing {} for pending device sync (offline)", user_jid); + log::debug!( + "Queueing {} for pending device sync (offline)", + user_jid.observe() + ); } else { - log::debug!("Triggering immediate device sync for {}", user_jid); + log::debug!( + "Triggering immediate device sync for {}", + user_jid.observe() + ); let client = Arc::clone(self); self.runtime .spawn(Box::pin(async move { @@ -1232,6 +1261,7 @@ impl Client { } } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.handle_plaintext", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id, enc_type = %enc_type), err(Debug)))] pub(crate) async fn handle_decrypted_plaintext( self: Arc, enc_type: &str, @@ -1243,7 +1273,7 @@ impl Client { log::debug!( "[msg:{}] Successfully decrypted message from {}: type={} [batch path]", info.id, - info.source.sender, + info.source.sender.observe(), enc_type ); @@ -1252,7 +1282,8 @@ impl Client { if original_msg.device_sent_message.is_some() && !info.source.is_from_me { warn!( "[msg:{}] DeviceSentMessage present but sender {} is not self", - info.id, info.source.sender, + info.id, + info.source.sender.observe(), ); } @@ -1303,7 +1334,8 @@ impl Client { } else { warn!( "[msg:{}] Dropping app_state_sync_key_share from non-self sender {}", - info.id, info.source.sender + info.id, + info.source.sender.observe() ); } } @@ -1333,7 +1365,8 @@ impl Client { } else { warn!( "[msg:{}] Dropping history_sync_notification from non-self sender {}", - info.id, info.source.sender + info.id, + info.source.sender.observe() ); } } @@ -1370,6 +1403,7 @@ impl Client { /// decrypt and replaces the caller's `session_guard` on the way out /// so the next payload in the batch stays serialized. #[allow(clippy::too_many_arguments)] + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.pn_lid_migration_decrypt", level = "debug", skip_all, fields(sender = %sender_jid.observe(), msg_id = %info.id, enc_type = %enc_type)))] async fn try_pn_to_lid_migration_decrypt( self: &Arc, sender_jid: &Jid, @@ -1421,7 +1455,7 @@ impl Client { log::info!( "[msg:{}] Decrypted after PN→LID session migration for {}", info.id, - info.source.sender + info.source.sender.observe() ); if let Some(prekey_id) = decrypted.consumed_prekey_id { adapter diff --git a/src/message/retry.rs b/src/message/retry.rs index 2111616e4..7981c3df6 100644 --- a/src/message/retry.rs +++ b/src/message/retry.rs @@ -11,6 +11,7 @@ impl Client { /// /// Returns `true` if this call dispatched the event, `false` if a /// previous call already did. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.undecryptable", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), msg_id = %info.id)))] pub(crate) async fn dispatch_undecryptable_event( &self, info: Arc, @@ -61,6 +62,7 @@ impl Client { /// resulting duplicate ack. /// /// Returns `true` to be assigned to `dispatched_undecryptable` flag. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.decrypt_failure", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id, reason = ?reason)))] pub(crate) async fn handle_decrypt_failure( self: &Arc, info: &Arc, @@ -101,6 +103,7 @@ impl Client { true } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.plaintext_failure", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), msg_id = %info.id)))] pub(crate) async fn handle_plaintext_failure( self: &Arc, info: &Arc, @@ -205,6 +208,7 @@ impl Client { /// Returns whether the caller should send the ack: `false` when we intended /// to retry but the send failed (so the stanza stays queued for another try), /// `true` when the resend went out or we deliberately gave up at the cap. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.retry_receipt", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id, reason = ?reason)))] async fn run_retry_receipt( self: &Arc, info: &Arc, @@ -219,7 +223,7 @@ impl Client { "Max retries ({}) reached for message {} from {} [{:?}]. Sending immediate PDO request.", MAX_DECRYPT_RETRIES, info.id, - info.source.sender, + info.source.sender.observe(), reason ); // Capped: give up and clear the backlog regardless of PDO outcome. @@ -232,8 +236,8 @@ impl Client { "High retry count ({}) for message {} in chat {} from {} [{:?}]", retry_count, info.id, - info.source.chat, - info.source.sender, + info.source.chat.observe(), + info.source.sender.observe(), reason ); } @@ -242,7 +246,11 @@ impl Client { Ok(()) => { debug!( "Sent retry receipt #{} for message {} in chat {} from {} [{:?}]", - retry_count, info.id, info.source.chat, info.source.sender, reason + retry_count, + info.id, + info.source.chat.observe(), + info.source.sender.observe(), + reason ); true } diff --git a/src/message/special.rs b/src/message/special.rs index e942cb83b..a64c76570 100644 --- a/src/message/special.rs +++ b/src/message/special.rs @@ -7,6 +7,7 @@ impl Client { /// Newsletters are not E2E encrypted and use the tag directly. /// They never carry a `secret_encrypted_message`, so no messageSecret is /// stored or retained for newsletter chats (no newsletter retention class). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.newsletter", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), msg_id = %info.id)))] pub(crate) async fn handle_newsletter_message( self: &Arc<Self>, node: &NodeRef<'_>, @@ -27,7 +28,7 @@ impl Client { log::info!( "[msg:{}] Received newsletter plaintext message from {}", info.id, - info.source.chat + info.source.chat.observe() ); self.dispatch_parsed_message(msg, info).await; } @@ -42,11 +43,15 @@ impl Client { log::debug!( "[msg:{}] Newsletter <plaintext> node from {} had no content bytes; skipping decode", info.id, - info.source.chat + info.source.chat.observe() ); } } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.recv.appstate_key_share", level = "debug", skip_all) + )] pub(crate) async fn handle_app_state_sync_key_share( &self, keys: &wa::message::AppStateSyncKeyShare, @@ -119,6 +124,7 @@ impl Client { } } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.skdm", level = "debug", skip_all, fields(group = %group_jid.observe(), sender = %sender_jid.observe())))] pub(crate) async fn handle_sender_key_distribution_message( self: &Arc<Self>, group_jid: &Jid, @@ -137,7 +143,7 @@ impl Client { ) else { log::warn!( "Go SKDM from {} missing required fields (signing_key={}, id={}, iteration={}, chain_key={})", - sender_jid, + sender_jid.observe(), go_msg.signing_key.is_some(), go_msg.id.is_some(), go_msg.iteration.is_some(), @@ -151,7 +157,7 @@ impl Client { log::error!( "Invalid chain_key length {} from Go SKDM from {}", chain_key.len(), - sender_jid + sender_jid.observe() ); return; } @@ -169,7 +175,7 @@ impl Client { Err(e) => { log::error!( "Failed to construct SKDM from Go format from {}: {:?} (original parse error: {:?})", - sender_jid, + sender_jid.observe(), e, e1 ); @@ -180,7 +186,7 @@ impl Client { Err(e) => { log::error!( "Failed to parse public key from Go SKDM for {}: {:?} (original parse error: {:?})", - sender_jid, + sender_jid.observe(), e, e1 ); @@ -191,7 +197,7 @@ impl Client { Err(e2) => { log::error!( "Failed to parse SenderKeyDistributionMessage (standard and Go fallback) from {}: primary: {:?}, fallback: {:?}", - sender_jid, + sender_jid.observe(), e1, e2 ); @@ -218,14 +224,14 @@ impl Client { { log::error!( "Failed to process SenderKeyDistributionMessage from {}: {:?}", - sender_jid, + sender_jid.observe(), e ); } else { log::debug!( "Successfully processed sender key distribution for group {} from {}", - group_jid, - sender_jid + group_jid.observe(), + sender_jid.observe() ); } } diff --git a/src/pair.rs b/src/pair.rs index 427b8796c..61cf1a0cb 100644 --- a/src/pair.rs +++ b/src/pair.rs @@ -35,6 +35,10 @@ pub fn make_qr_data_with_client_type( PairUtils::make_qr_data(&device_state, ref_str, client_type) } +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.pair.handle_iq", level = "debug", skip_all) +)] pub async fn handle_iq(client: &Arc<Client>, node: &NodeRef<'_>) -> bool { // Server JID is "s.whatsapp.net" (no @ prefix for server-only JIDs) if node @@ -147,6 +151,10 @@ pub async fn handle_iq(client: &Arc<Client>, node: &NodeRef<'_>) -> bool { false } +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.pair.success", level = "debug", skip_all) +)] async fn handle_pair_success<'a>( client: &Arc<Client>, request_node: &NodeRef<'a>, @@ -334,7 +342,7 @@ async fn handle_pair_success<'a>( client.expected_disconnect.store(true, Ordering::Relaxed); - info!("Successfully paired {jid}"); + info!("Successfully paired {}", jid.observe()); let success_event = PairSuccess { id: jid, @@ -369,6 +377,10 @@ async fn handle_pair_success<'a>( } } +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.pair.qr", level = "debug", skip_all, err(Debug)) +)] pub async fn pair_with_qr_code(client: &Arc<Client>, qr_code: &str) -> Result<(), anyhow::Error> { info!(target: "Client/PairTest", "Master client attempting to pair with QR code."); diff --git a/src/pair_code.rs b/src/pair_code.rs index fa8bcfa20..da674bded 100644 --- a/src/pair_code.rs +++ b/src/pair_code.rs @@ -108,6 +108,10 @@ impl Client { /// # Ok(()) /// # } /// ``` + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.pair.code", level = "debug", skip_all, err(Debug)) + )] pub async fn pair_with_code( self: &Arc<Self>, options: PairCodeOptions, @@ -237,6 +241,10 @@ impl Client { /// /// This is called when the user enters the code on their phone. The notification /// contains the primary device's encrypted ephemeral public key and identity public key. +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.pair.code_notification", level = "debug", skip_all) +)] pub(crate) async fn handle_pair_code_notification( client: &Arc<Client>, node: &NodeRef<'_>, diff --git a/src/pdo.rs b/src/pdo.rs index e871ac3de..caad0411b 100644 --- a/src/pdo.rs +++ b/src/pdo.rs @@ -60,6 +60,7 @@ impl Client { /// # Returns /// * `Ok(())` if the request was sent successfully /// * `Err` if we couldn't send the request (e.g., not logged in) + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.pdo.placeholder_resend", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id), err(Debug)))] pub async fn send_pdo_placeholder_resend_request( self: &Arc<Self>, info: &Arc<MessageInfo>, @@ -100,7 +101,8 @@ impl Client { if self.pdo_pending_requests.get(&cache_key).await.is_some() { debug!( "PDO request already pending for message {} from {}", - info.id, info.source.sender + info.id, + info.source.sender.observe() ); return Ok(()); } @@ -149,7 +151,10 @@ impl Client { info!( "Sending PDO placeholder resend request for message {} from {} in {} to {}", - info.id, info.source.sender, info.source.chat, peer_target + info.id, + info.source.sender.observe(), + info.source.chat.observe(), + peer_target.observe() ); if let Err(e) = self @@ -174,6 +179,7 @@ impl Client { } /// Request on-demand message history from the primary phone via PDO. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.pdo.fetch_history", level = "debug", skip_all, fields(chat = %chat_jid.observe(), count), err(Debug)))] pub async fn fetch_message_history( self: &Arc<Self>, chat_jid: &Jid, @@ -217,7 +223,9 @@ impl Client { info!( "Sending PDO history sync on-demand request for chat {} (count={}) to {}", - chat_jid, count, peer_target + chat_jid.observe(), + count, + peer_target.observe() ); self.ensure_e2e_sessions(std::slice::from_ref(&peer_target)) @@ -227,6 +235,7 @@ impl Client { /// Sends a peer message (message to our own devices). /// This is used for PDO requests and similar device-to-device communication. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.pdo.send_peer_message", level = "debug", skip_all, fields(to = %to.observe()), err(Debug)))] async fn send_peer_message( self: &Arc<Self>, to: Jid, @@ -256,6 +265,7 @@ impl Client { /// # Arguments /// * `response` - The PDO response message /// * `info` - The MessageInfo for the PDO response message itself + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.pdo.handle_response", level = "debug", skip_all, fields(sender = %pdo_msg_info.source.sender.observe())))] pub async fn handle_pdo_response( self: &Arc<Self>, response: &wa::message::PeerDataOperationRequestResponseMessage, @@ -265,7 +275,7 @@ impl Client { if pdo_msg_info.source.sender.device != 0 { debug!( "Ignoring PDO response from non-primary device {}", - pdo_msg_info.source.sender + pdo_msg_info.source.sender.observe() ); return; } @@ -365,7 +375,9 @@ impl Client { info!( "Dispatching PDO-recovered message {} from {} via phone (request_id={})", - message_info.id, message_info.source.sender, request_id + message_info.id, + message_info.source.sender.observe(), + request_id ); self.core @@ -462,6 +474,7 @@ impl Client { /// NOT ack, so the stanza stays in the offline queue for another attempt. /// Age-skip counts as a deliberate give-up (`true`), so ancient stanzas are /// still cleared. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.pdo.run_request", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id)))] pub(crate) async fn run_pdo_request(self: &Arc<Self>, info: &Arc<MessageInfo>) -> bool { // Skip ancient messages (14d, matching the AB prop), compared in seconds // like WA Web's `age_s > i`. Uses the wacore time primitive (mockable). @@ -479,7 +492,9 @@ impl Client { Err(e) => { warn!( "Failed to send PDO request for message {} from {}: {:?}", - info.id, info.source.sender, e + info.id, + info.source.sender.observe(), + e ); false } diff --git a/src/prekeys.rs b/src/prekeys.rs index be0a20430..226eaa31b 100644 --- a/src/prekeys.rs +++ b/src/prekeys.rs @@ -62,6 +62,7 @@ fn clamp_wanted_pre_key_count(n: usize) -> usize { } impl Client { + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.fetch_pre_keys", level = "debug", skip_all, fields(count = jids.len()), err(Debug)))] pub(crate) async fn fetch_pre_keys( &self, jids: &[Jid], @@ -75,13 +76,22 @@ impl Client { let bundles = self.execute(spec).await?; for jid in bundles.keys() { - log::debug!("Successfully parsed pre-key bundle for {jid}"); + log::debug!("Successfully parsed pre-key bundle for {}", jid.observe()); } Ok(bundles) } /// Query the WhatsApp server for how many pre-keys it currently has for this device. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.server_pre_key_count", + level = "debug", + skip_all, + err(Debug) + ) + )] pub(crate) async fn get_server_pre_key_count(&self) -> Result<usize, crate::request::IqError> { let response = self.execute(PreKeyCountSpec::new()).await?; Ok(response.count) @@ -89,6 +99,15 @@ impl Client { /// Upload prekeys at login if the persisted flag indicates they're needed. /// Matches WA Web's PassiveTasks.js:30 which checks `getServerHasPreKeys()`. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.upload_pre_keys_at_login", + level = "debug", + skip_all, + err(Debug) + ) + )] pub(crate) async fn upload_pre_keys_at_login(&self) -> Result<(), anyhow::Error> { let has_prekeys = self .persistence_manager @@ -120,6 +139,7 @@ impl Client { /// Ensure the server has enough pre-keys, uploading if below threshold. /// When `force` is true, skips the count guard (used by digest key repair). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.upload_pre_keys", level = "debug", skip_all, fields(force = force), err(Debug)))] pub(crate) async fn upload_pre_keys(&self, force: bool) -> Result<(), anyhow::Error> { // Decision is should_upload_pre_keys(force, count), but a forced upload short-circuits // and skips the server-count IQ entirely: WA Web's handlePreKeyLow uploads @@ -146,6 +166,15 @@ impl Client { /// (the same namespace as uploads) and advance it, so a retry-receipt prekey can never /// collide with a live pool key. The caller must hold `prekey_upload_lock` to serialize the /// allocate+bump with the upload path. Mirrors WA Web's `getOrGenSinglePreKey -> NEXT_PK_ID`. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.allocate_prekey_id", + level = "debug", + skip_all, + err(Debug) + ) + )] pub(crate) async fn allocate_next_one_time_prekey_id(&self) -> Result<u32, anyhow::Error> { let next_pre_key_id = self .persistence_manager @@ -169,6 +198,15 @@ impl Client { /// Generate and upload the configured number of pre-keys (see /// [`Client::set_wanted_pre_key_count`]). Shared by `upload_pre_keys` and /// `upload_pre_keys_at_login` to avoid redundant server count queries. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.upload_pre_keys_inner", + level = "debug", + skip_all, + err(Debug) + ) + )] async fn upload_pre_keys_inner(&self) -> Result<(), anyhow::Error> { let device_snapshot = self.persistence_manager.get_device_snapshot().await; let device_store = self.persistence_manager.get_device_arc().await; @@ -289,6 +327,7 @@ impl Client { /// Verified against WA Web JS: `{ algo: { type: "fibonacci", first: 1e3, second: 2e3 }, max: 61e4 }` /// /// When `force` is true, bypasses the count guard (used by digest repair path). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.upload_pre_keys_retry", level = "debug", skip_all, fields(force = force), err(Debug)))] pub(crate) async fn upload_pre_keys_with_retry( &self, force: bool, @@ -338,6 +377,15 @@ impl Client { /// /// Acquires `prekey_upload_lock` for the duration so this force-upload /// cannot race on `start_id` with the count-based and digest-repair paths. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.refresh_pre_keys", + level = "debug", + skip_all, + err(Debug) + ) + )] pub async fn refresh_pre_keys(&self) -> Result<(), anyhow::Error> { let _guard = self.prekey_upload_lock.lock().await; self.upload_pre_keys_with_retry(true).await @@ -353,6 +401,15 @@ impl Client { /// 5. If validation fails (regId mismatch, missing prekey, hash mismatch): logs warning, /// does NOT re-upload — WA Web catches all `validateLocalKeyBundle` exceptions without /// re-uploading; the normal `RotateKeyJob` will eventually refresh keys + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.session.validate_digest_key", + level = "debug", + skip_all, + err(Debug) + ) + )] pub(crate) async fn validate_digest_key(&self) -> Result<(), anyhow::Error> { // Hold the lock across the whole pass so the 404 re-upload can't race with // `upload_pre_keys_at_login`, `handle_prekey_low`, or `refresh_pre_keys` on diff --git a/src/receipt.rs b/src/receipt.rs index 400fd1dd9..d4756ad72 100644 --- a/src/receipt.rs +++ b/src/receipt.rs @@ -132,6 +132,10 @@ impl Client { || info.source.is_self_fanout() } + #[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.receipt.handle", level = "debug", skip_all) + )] pub(crate) async fn handle_receipt(self: &Arc<Self>, node: Arc<OwnedNodeRef>) { let nr = node.get(); let mut attrs = nr.attrs(); @@ -180,8 +184,9 @@ impl Client { .or_else(|| agg_key.clone()) .unwrap_or_else(|| stanza_id.clone()); debug!( - "Aggregated receipt from {from}: stanza={stanza_id} \ + "Aggregated receipt from {}: stanza={stanza_id} \ message_id={agg_msg_id:?} key={agg_key:?} users={}", + from.observe(), users.len() ); for user in users { @@ -226,8 +231,9 @@ impl Client { wacore::stanza::receipt::collect_simple_message_ids(nr, &stanza_id, is_view); debug!( - "Received receipt type '{receipt_type:?}' for {} message(s) from {from}", - message_ids.len() + "Received receipt type '{receipt_type:?}' for {} message(s) from {}", + message_ids.len(), + from.observe() ); let receipt = Receipt { @@ -275,7 +281,7 @@ impl Client { .optional_string("call-id") .as_deref() .unwrap_or_default(), - receipt.source.chat, + receipt.source.chat.observe(), child_attrs .optional_string("call-creator") .as_deref() @@ -305,6 +311,7 @@ impl Client { /// `Send/DeliveryReceiptJob.js`); these are NOT skipped anymore. /// - Newsletters and messages without an ID are skipped (newsletters are /// handled by the ack gate, not here). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.receipt.send_delivery", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id)))] pub(crate) async fn send_delivery_receipt(&self, info: &crate::types::message::MessageInfo) { if !Self::should_send_delivery_receipt(info) { return; @@ -324,7 +331,7 @@ impl Client { ReceiptType::Delivered }; debug!(target: "Client/Receipt", "Sending {} receipt for message {} to {}", - receipt_kind.as_wire_str(), info.id, info.source.sender); + receipt_kind.as_wire_str(), info.id, info.source.sender.observe()); if let Err(e) = self.send_node(receipt_node).await && !matches!(e, crate::client::ClientError::NotConnected) @@ -353,6 +360,7 @@ impl Client { /// Emits a nack so the server stops retransmitting an unrecoverable /// failure. Prefer [`Client::send_retry_receipt`] for recoverable /// errors (BadMac, NoSession, etc). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.receipt.send_nack", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id, reason = ?reason)))] pub(crate) async fn send_nack( &self, info: &MessageInfo, @@ -374,7 +382,7 @@ impl Client { let nack = build_nack_node(info, &own_pn, reason, failure_reason); debug!(target: "Client/Receipt", "Sending nack (reason={:?}, code={}) for message {} from {}", - reason, reason.code(), info.id, info.source.sender); + reason, reason.code(), info.id, info.source.sender.observe()); if let Err(e) = self.send_node(nack).await && !matches!(e, crate::client::ClientError::NotConnected) @@ -387,6 +395,7 @@ impl Client { /// Sends read receipts for one or more messages. /// /// For group messages, pass the message sender as `sender`. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.receipt.mark_as_read", level = "debug", skip_all, fields(chat = %chat.observe()), err(Debug)))] pub async fn mark_as_read( &self, chat: &Jid, @@ -420,7 +429,7 @@ impl Client { let node = builder.build(); - debug!(target: "Client/Receipt", "Sending read receipt for {} message(s) to {}", message_ids.len(), chat); + debug!(target: "Client/Receipt", "Sending read receipt for {} message(s) to {}", message_ids.len(), chat.observe()); self.send_node(node) .await diff --git a/src/request.rs b/src/request.rs index 0edf44687..80f69c6f1 100644 --- a/src/request.rs +++ b/src/request.rs @@ -132,6 +132,7 @@ impl Client { /// # Ok(()) /// # } /// ``` + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.iq", level = "debug", skip_all, fields(ns = %query.namespace, kind = ?query.query_type), err(Debug)))] pub async fn send_iq( &self, query: InfoQuery<'_>, diff --git a/src/retry.rs b/src/retry.rs index d89b6ec49..f8d8012c2 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -192,6 +192,7 @@ fn build_retry_processing_key(chat: &Jid, message_id: &str, participant_jid: &Ji } impl Client { + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.retry.handle_receipt", level = "debug", skip_all, fields(chat = %receipt.source.chat.observe(), sender = %receipt.source.sender.observe()), err(Debug)))] pub(crate) async fn handle_retry_receipt( self: &Arc<Self>, receipt: &Receipt, @@ -218,7 +219,10 @@ impl Client { if retry_count >= MAX_RETRY_COUNT { warn!( "Refusing retry #{} for message {} from {}: exceeds max attempts ({})", - retry_count, message_id, receipt.source.sender, MAX_RETRY_COUNT + retry_count, + message_id, + receipt.source.sender.observe(), + MAX_RETRY_COUNT ); return Ok(()); } @@ -326,7 +330,7 @@ impl Client { log::warn!( "Failed to fetch group info for retry of msg {} in {}: {e}", message_id, - info.chat + info.chat.observe() ); None } @@ -348,7 +352,7 @@ impl Client { log::warn!( "Unknown device {} in group {} — forcing full sender key rotation \ (matches WA Web's rotateKey behavior)", - info.requester, + info.requester.observe(), group_jid ); @@ -427,7 +431,10 @@ impl Client { .should_recreate_session(retry_count, &resolved_jid) .await { - info!("Recreating session with {resolved_jid} for retry of {message_id}: {reason}"); + info!( + "Recreating session with {} for retry of {message_id}: {reason}", + resolved_jid.observe() + ); self.signal_cache.delete_session(&signal_address).await; drop(guard); self.flush_signal_cache_logged("should_recreate_session", Some(&message_id)) @@ -448,7 +455,9 @@ impl Client { info!( "Resending message {} to {} (retry #{})", - message_id, info.chat, retry_count + message_id, + info.chat.observe(), + retry_count ); if info.chat.is_group() { @@ -547,6 +556,7 @@ impl Client { /// retry>2 when the base key already changed (session was regenerated /// legitimately). The subsequent `ensure_e2e_sessions_resolved` call in /// `handle_retry_receipt` rebuilds any session this function deleted. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.retry.update_local_session", level = "debug", skip_all, fields(chat = %info.chat.observe(), peer = %resolved_jid.observe(), retry = retry_count)))] async fn update_local_signal_session( &self, info: &RetryChatInfo, @@ -573,12 +583,14 @@ impl Client { }; info!( "Marked {} for fresh SKDM in {} {} due to retry receipt", - info.requester, chat_type, group_jid + info.requester.observe(), + chat_type, + group_jid ); } Err(e) => log::warn!( "Failed to mark sender key forget for {} in {}: {}", - info.requester, + info.requester.observe(), group_jid, e ), @@ -600,7 +612,7 @@ impl Client { if !key_bundle_processed && keys_node_present { log::warn!( "Key bundle present but rejected for {}: {:?} — skipping regId mismatch deletion", - resolved_jid, + resolved_jid.observe(), key_bundle_result.as_ref().err() ); } @@ -610,7 +622,7 @@ impl Client { // only warn when a regId mismatch triggers a delete below. log::debug!( "No key bundle in retry receipt for {}: {}. Checking for reg ID mismatch.", - resolved_jid, + resolved_jid.observe(), e ); } @@ -635,7 +647,9 @@ impl Client { info!( "Registration ID mismatch for {} (stored: {}, received: {}). \ Deleting session since no key bundle provided.", - signal_address, stored_reg_id, received_reg_id + wacore::types::jid::observe_protocol_address(&signal_address), + stored_reg_id, + received_reg_id ); let lock = self.session_lock_for(signal_address.as_str()).await; let _guard = lock.lock().await; @@ -676,9 +690,14 @@ impl Client { { Ok(()) => info!( "Saved base key for {} at retry #{} for collision detection", - signal_address, retry_count + wacore::types::jid::observe_protocol_address(&signal_address), + retry_count + ), + Err(e) => warn!( + "Failed to save base key for {}: {}", + wacore::types::jid::observe_protocol_address(&signal_address), + e ), - Err(e) => warn!("Failed to save base key for {}: {}", signal_address, e), } return; } @@ -693,7 +712,8 @@ impl Client { warn!( "Base key collision detected for {} at retry #{}. \ Session hasn't been regenerated. Forcing fresh session.", - signal_address, retry_count + wacore::types::jid::observe_protocol_address(&signal_address), + retry_count ); let _ = device_guard .backend @@ -713,7 +733,8 @@ impl Client { Ok(false) => { info!( "Base key changed for {} at retry #{} - session regenerated", - signal_address, retry_count + wacore::types::jid::observe_protocol_address(&signal_address), + retry_count ); let _ = device_guard .backend @@ -721,7 +742,11 @@ impl Client { .await; } Err(e) => { - warn!("Failed to check base key for {}: {}", signal_address, e); + warn!( + "Failed to check base key for {}: {}", + wacore::types::jid::observe_protocol_address(&signal_address), + e + ); } } } @@ -807,6 +832,7 @@ impl Client { /// * `node` - The retry receipt node containing the key bundle /// * `requester_jid` - The JID of the device requesting the retry /// * `is_peer` - Whether this is a peer device (our own device) + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.retry.process_key_bundle", level = "debug", skip_all, fields(peer = %requester_jid.observe(), is_peer), err(Debug)))] async fn process_retry_key_bundle( &self, node: &NodeRef<'_>, @@ -961,6 +987,7 @@ impl Client { /// know which attempt this is. The sender may use this to decide whether to resend. /// * `reason` - The retry reason code (matches WhatsApp Web's RetryReason enum). This helps /// the sender understand why the message couldn't be decrypted. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.retry.send_receipt", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), retry = retry_count), err(Debug)))] pub(crate) async fn send_retry_receipt( &self, info: &crate::types::message::MessageInfo, @@ -977,15 +1004,19 @@ impl Client { log::debug!( "Skipping retry receipt for message {} from bot {} in non-bot chat {}", info.id, - info.source.sender, - info.source.chat + info.source.sender.observe(), + info.source.chat.observe() ); return Ok(()); } debug!( "Sending retry receipt #{} for message {} in chat {} from {} (reason: {:?})", - retry_count, info.id, info.source.chat, info.source.sender, reason + retry_count, + info.id, + info.source.chat.observe(), + info.source.sender.observe(), + reason ); // Build the retry element with the error code (matches WhatsApp Web's format) @@ -1117,6 +1148,7 @@ impl Client { /// WA Web reference: `ENC_RETRY_RECEIPT_ATTRS.GROUP_CALL = "enc_rekey_retry"`, /// constructed in `WAWebVoipSignalingEnums` module. #[allow(dead_code)] // Will be used when call handling is implemented (#345) + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.retry.send_enc_rekey_receipt", level = "debug", skip_all, fields(peer = %peer_jid.observe(), retry = retry_count), err(Debug)))] pub(crate) async fn send_enc_rekey_retry_receipt( &self, stanza_id: &str, @@ -1149,7 +1181,9 @@ impl Client { info!( "Sending enc_rekey_retry receipt for call-id={} to {} (count={})", - call_id, peer_jid, retry_count + call_id, + peer_jid.observe(), + retry_count ); self.send_node(receipt_node).await?; diff --git a/src/send.rs b/src/send.rs index dfcc3a667..4eb873a87 100644 --- a/src/send.rs +++ b/src/send.rs @@ -389,6 +389,7 @@ impl Client { } /// Send a message with additional options. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.message", level = "debug", skip_all, fields(to = %to.observe()), err(Debug)))] pub async fn send_message_with_options( &self, to: Jid, @@ -470,6 +471,7 @@ impl Client { /// `AddressingMode::Lid`; `prepare_group_stanza` signs with `own_lid` /// and emits `addressing_mode="lid"` on the stanza. Errors only if no /// recipient could be resolved. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.status", level = "debug", skip_all, fields(count = recipients.len()), err(Debug)))] pub(crate) async fn send_status_message( &self, message: wa::Message, @@ -720,6 +722,7 @@ impl Client { /// For LID mode, uses `group_info.phone_jid_for_lid_user` to query devices /// via PN when available (LID usync is unreliable for own JID), then /// converts the result back to LID. Same fallback as `prepare_group_stanza`. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.resolve_skdm_targets", level = "debug", skip_all, fields(group = %wacore_binary::jid::observe_str(group_jid))))] async fn resolve_skdm_targets( &self, group_jid: &str, @@ -898,6 +901,7 @@ impl Client { /// Cold path of [`spawn_phash_validation`](Self::spawn_phash_validation): the /// server's phash disagreed with ours, so invalidate the relevant /// device/group caches and (for groups) force sender-key redistribution. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.phash_mismatch", level = "debug", skip_all, fields(jid = %jid.observe())))] async fn handle_phash_mismatch( &self, jid: &Jid, @@ -906,7 +910,8 @@ impl Client { invalidate_group_cache: bool, ) { log::warn!( - "Phash mismatch for {jid}: ours={our_phash}, server={server_phash}. Invalidating caches." + "Phash mismatch for {}: ours={our_phash}, server={server_phash}. Invalidating caches.", + jid.observe() ); // DM phash covers both recipient + own devices // (WA Web: syncDeviceListJob([recipient, me])) @@ -974,6 +979,7 @@ impl Client { /// * `message_id` - The ID of the message to delete /// * `revoke_type` - Use `RevokeType::Sender` to delete your own message, /// or `RevokeType::Admin { original_sender }` to delete another user's message as group admin + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.revoke", level = "debug", skip_all, fields(to = %to.observe()), err(Debug)))] pub async fn revoke_message( &self, to: Jid, @@ -1062,6 +1068,7 @@ impl Client { .await } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.pin", level = "debug", skip_all, fields(chat = %chat.observe()), err(Debug)))] async fn send_pin( &self, chat: Jid, @@ -1095,6 +1102,7 @@ impl Client { .await } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.impl", level = "debug", skip_all, fields(to = %to.observe()), err(Debug)))] #[allow(clippy::too_many_arguments)] pub(crate) async fn send_message_impl( &self, @@ -1252,7 +1260,8 @@ impl Client { if needs_rotation { log::info!( - "Periodic sender-key rotation for {to} (chain iteration ≥ {SENDER_KEY_ROTATION_THRESHOLD})" + "Periodic sender-key rotation for {} (chain iteration ≥ {SENDER_KEY_ROTATION_THRESHOLD})", + to.observe() ); self.signal_cache .delete_sender_key(sender_key_name.cache_key()) @@ -1327,7 +1336,10 @@ impl Client { if let Some(SignalProtocolError::NoSenderKeyState(_)) = e.downcast_ref::<SignalProtocolError>() { - log::warn!("No sender key for group {}, forcing distribution.", to); + log::warn!( + "No sender key for group {}, forcing distribution.", + to.observe() + ); if let Err(e) = self .persistence_manager @@ -1418,7 +1430,10 @@ impl Client { } } Err(e) => { - log::warn!("LID query failed for {}, falling back to PN: {e:?}", to); + log::warn!( + "LID query failed for {}, falling back to PN: {e:?}", + to.observe() + ); } } } @@ -1527,7 +1542,7 @@ impl Client { } } if should_issue_tc_token_after_send { - debug!(target: "Client/TcToken", "Scheduled tc token issuance after send for {}", to); + debug!(target: "Client/TcToken", "Scheduled tc token issuance after send for {}", to.observe()); } let lock_jids = self.build_session_lock_keys(&all_dm_jids).await; @@ -1725,6 +1740,7 @@ impl Client { /// /// Returns whether we should issue a new tc token after send, and the cache key /// of the attached valid tc token when that token should be marked as used. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.maybe_tc_token", level = "debug", skip_all, fields(to = %to.observe())))] async fn maybe_include_tc_token( &self, to: &Jid, @@ -1820,9 +1836,9 @@ impl Client { wacore_binary::Jid::new(*lid_user, Server::Lid).to_string(); let cs_token = compute_cs_token(salt, &recipient_lid); extra_nodes.push(build_cs_token_node(&cs_token)); - log::debug!(target: "Client/CsToken", "Attached cstoken for {} (NCT fallback)", to); + log::debug!(target: "Client/CsToken", "Attached cstoken for {} (NCT fallback)", to.observe()); } else { - log::debug!(target: "Client/CsToken", "No tctoken or NCT salt/LID available for {}", to); + log::debug!(target: "Client/CsToken", "No tctoken or NCT salt/LID available for {}", to.observe()); } } } @@ -1832,6 +1848,7 @@ impl Client { } /// Returns `true` if the issuance IQ succeeded. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.issue_tc_token", level = "debug", skip_all, fields(to = %to.observe())))] async fn issue_tc_token_after_send(&self, to: &Jid) -> bool { use wacore::iq::tctoken::IssuePrivacyTokensSpec; @@ -1847,7 +1864,7 @@ impl Client { ))) .await else { - log::debug!(target: "Client/TcToken", "Failed to issue tc_token for {}", issuance_jid); + log::debug!(target: "Client/TcToken", "Failed to issue tc_token for {}", issuance_jid.observe()); return false; }; @@ -1870,7 +1887,7 @@ impl Client { let mut any_stored = false; for received in tokens { if received.token.is_empty() { - log::warn!(target: "Client/TcToken", "Server returned empty tc_token for {}, skipping", received.jid); + log::warn!(target: "Client/TcToken", "Server returned empty tc_token for {}, skipping", received.jid.observe()); continue; } @@ -1945,6 +1962,7 @@ impl Client { /// Re-issue tctoken after a contact's device identity changes. /// Only re-issues if we previously sent a token (sender_timestamp valid). /// Uses session_locks to deduplicate concurrent spawns for the same sender. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.reissue_tc_token", level = "debug", skip_all, fields(sender = %sender.observe())))] pub(crate) async fn reissue_tc_token_after_identity_change(&self, sender: &Jid) { use wacore::iq::tctoken::{IssuePrivacyTokensSpec, is_sender_tc_token_expired}; @@ -1994,14 +2012,14 @@ impl Client { log::debug!( target: "Client/TcToken", "Re-issued tctoken after identity change for {}", - sender + sender.observe() ); } Err(e) => { log::debug!( target: "Client/TcToken", "Failed to re-issue tctoken after identity change for {}: {e}", - sender + sender.observe() ); } } diff --git a/src/upload.rs b/src/upload.rs index 13a1d7ef1..0f8a23c98 100644 --- a/src/upload.rs +++ b/src/upload.rs @@ -349,6 +349,7 @@ impl Client { /// /// Only needed for new or modified media. To forward existing media unchanged, /// reuse the original message's CDN fields directly, no round-trip required. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.media.upload", level = "debug", skip_all, fields(kind = ?media_type, len = data.len()), err(Debug)))] pub async fn upload( &self, data: Vec<u8>, @@ -400,6 +401,7 @@ impl Client { /// storage of your choice, then pass that storage as `source` plus the /// returned [`wacore::upload::EncryptedMediaInfo`]. The caller owns where the /// ciphertext lives (temp file, memory, …); this method never touches disk. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.media.upload_stream", level = "debug", skip_all, fields(kind = ?media_type), err(Debug)))] pub async fn upload_stream<S>( &self, source: S, diff --git a/src/usync.rs b/src/usync.rs index 355aba215..22c5ad654 100644 --- a/src/usync.rs +++ b/src/usync.rs @@ -9,6 +9,7 @@ use wacore::iq::usync::{DeviceListResponse, DeviceListSpec}; use wacore_binary::Jid; impl Client { + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.usync.get_user_devices", level = "debug", skip_all, fields(users = jids.len()), err(Debug)))] pub(crate) async fn get_user_devices(&self, jids: &[Jid]) -> Result<Vec<Jid>, anyhow::Error> { let mut jids_to_fetch: HashSet<Jid> = HashSet::with_capacity(jids.len()); let mut all_devices = Vec::with_capacity(jids.len() * 2); @@ -49,6 +50,7 @@ impl Client { /// Users the server OMITS — unchanged ones, when we sent a `device_hash` — are /// simply absent here, so their cached records are left untouched (the /// merge-safe behavior the `device_hash` optimization depends on). + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.usync.process_device_list", level = "debug", skip_all, fields(users = response.device_lists.len())))] async fn process_device_list_response(&self, response: &DeviceListResponse) -> Vec<Jid> { // Learn LID↔PN mappings via the same batched, guarded learner query_info // uses (one detached transaction, skipping already-durable pairs), so @@ -206,6 +208,15 @@ impl Client { /// (by omitting the user) instead of returning the full list on every reconnect. /// On a changed list the server returns it and we update; omitted users keep /// their cache. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.usync.sync_own_device_list", + level = "debug", + skip_all, + err(Debug) + ) + )] pub(crate) async fn sync_own_device_list(&self) -> Result<(), anyhow::Error> { let device_snapshot = self.persistence_manager.get_device_snapshot().await; @@ -241,6 +252,14 @@ impl Client { } /// WA Web: `doPendingDeviceSync()` — flush batched unknown-device users. + #[cfg_attr( + feature = "tracing", + tracing::instrument( + name = "wa.usync.flush_pending_device_sync", + level = "debug", + skip_all + ) + )] pub(crate) async fn flush_pending_device_sync(&self) { let pending = self.pending_device_sync.take_all().await; if pending.is_empty() { diff --git a/wacore/Cargo.toml b/wacore/Cargo.toml index 1cfbea286..2f3a8266c 100644 --- a/wacore/Cargo.toml +++ b/wacore/Cargo.toml @@ -12,6 +12,10 @@ default = ["simd"] simd = ["wacore-appstate/simd"] debug-diagnostics = [] debug-snapshots = [] +# Optional observability: emit tracing spans/events. Off by default (no dep). +tracing = ["dep:tracing"] +# 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 # for the e2e mock server (self-signed certs). NEVER enable in prod. danger-skip-cert-chain-verify = ["wacore-noise/danger-skip-cert-chain-verify"] @@ -39,6 +43,7 @@ hmac = { workspace = true } itoa = { workspace = true } log = { workspace = true } md5 = "0.8.0" +tracing = { workspace = true, optional = true } portable-atomic = { workspace = true } prost = { workspace = true } rand = { workspace = true } diff --git a/wacore/binary/Cargo.toml b/wacore/binary/Cargo.toml index 31452b7b1..399f7d06f 100644 --- a/wacore/binary/Cargo.toml +++ b/wacore/binary/Cargo.toml @@ -17,6 +17,9 @@ crate-type = ["rlib"] default = ["simd"] simd = [] serde = ["dep:serde", "compact_str/serde"] +# Render raw phone numbers in `Jid::observe()` instead of the redacted `pn#<hash>`. +# Local debugging only; never enable in production. +tracing-pii = [] [dependencies] bytes = { workspace = true } diff --git a/wacore/binary/src/jid.rs b/wacore/binary/src/jid.rs index c4667c8f8..1188bf3b6 100644 --- a/wacore/binary/src/jid.rs +++ b/wacore/binary/src/jid.rs @@ -236,6 +236,17 @@ impl Server { pub fn renders_agent(self) -> bool { !matches!(self, Self::Pn | Self::Lid | Self::Hosted | Self::HostedLid) } + + /// Whether the `user` part is (or can be) a real phone number, i.e. PII that + /// must be redacted in tracing fields. LID-family/group/broadcast/newsletter/ + /// bot/call users are pseudonymous or non-personal and are safe to render raw. + #[inline] + pub fn carries_phone_number(self) -> bool { + matches!( + self, + Self::Pn | Self::Hosted | Self::Legacy | Self::Messenger | Self::Interop + ) + } } impl fmt::Display for Server { @@ -833,6 +844,104 @@ impl fmt::Display for Jid { } } +/// Privacy-aware [`Display`] wrapper for a [`Jid`], for use in tracing fields. +/// +/// Pseudonymous (LID) and non-personal (newsletter/bot/call, modern group ids) +/// JIDs render in full, so the same peer/chat correlates across spans. JIDs whose +/// `user` is a phone number ([`Server::carries_phone_number`]) render the user as +/// a `pn#<token>` instead — preserving correlation without leaking the number. +/// Legacy group/broadcast ids of the form `<creator-phone>-<timestamp>` get only +/// the numeric prefix redacted (`pn#<token>-<timestamp>`), keeping the timestamp. +/// +/// The token is a keyed hash (SipHash via a process-lifetime random key), not a +/// plain digest of the number: the phone-number search space is small, so an +/// unkeyed hash would be reversible by precomputation. The random key lives only +/// in process memory, so exported traces cannot be brute-forced back to numbers. +/// It is stable within a process run (correlation works) but not across restarts +/// (a fresh key each start). Enable the `tracing-pii` feature to render raw +/// numbers (local debugging only). The token is computed only while formatting an +/// already-enabled span, so it costs nothing on disabled call sites. +pub struct ObservedJid<'a>(&'a Jid); + +impl Jid { + /// Privacy-aware display for tracing spans/fields. See [`ObservedJid`]. + #[inline] + pub fn observe(&self) -> ObservedJid<'_> { + ObservedJid(self) + } +} + +/// Per-process keyed token for a sensitive string: a SipHash with a random key +/// created once per process. Stable within a run (so the same value correlates +/// across spans) but not precomputable from the input, so exported traces cannot +/// be brute-forced back to the original (e.g. an E.164 phone number). Public so +/// other layers can redact non-`Jid` identifiers (e.g. a Signal `ProtocolAddress` +/// name, which embeds a phone number) with the same keyed scheme. +pub fn observe_token(s: &str) -> u64 { + use std::hash::BuildHasher; + static KEY: std::sync::OnceLock<std::collections::hash_map::RandomState> = + std::sync::OnceLock::new(); + KEY.get_or_init(std::collections::hash_map::RandomState::new) + .hash_one(s) +} + +/// Privacy-aware redaction of a JID supplied as a string (e.g. a group jid `&str` +/// in a span field). Parses it and applies [`Jid::observe`]; if it does not parse, +/// falls back to a keyed token so a raw number can never leak. Honors `tracing-pii`. +pub fn observe_str(s: &str) -> String { + if cfg!(feature = "tracing-pii") { + return s.to_string(); + } + match s.parse::<Jid>() { + Ok(jid) => jid.observe().to_string(), + Err(_) => format!("?#{:016x}", observe_token(s)), + } +} + +impl fmt::Display for ObservedJid<'_> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let jid = self.0; + if jid.user.is_empty() || cfg!(feature = "tracing-pii") { + return fmt::Display::fmt(jid, f); + } + // Decide the privacy-safe `user` rendering. + let redacted: Option<String> = if jid.server.carries_phone_number() { + // The whole user is a phone number. + Some(format!("pn#{:016x}", observe_token(jid.user.as_str()))) + } else if matches!(jid.server, Server::Group | Server::Broadcast) { + // Legacy group/broadcast ids embed the creator phone as + // "<phone>-<timestamp>". Redact the numeric prefix and keep the + // timestamp (not PII) so the group still correlates across spans. + match jid.user.find('-') { + Some(i) if i > 0 && jid.user.as_bytes()[..i].iter().all(|b| b.is_ascii_digit()) => { + Some(format!( + "pn#{:016x}{}", + observe_token(&jid.user[..i]), + &jid.user[i..] + )) + } + _ => None, + } + } else { + None + }; + let Some(user) = redacted else { + // Pseudonymous (LID) or non-personal: safe to render in full. + return fmt::Display::fmt(jid, f); + }; + f.write_str(&user)?; + // Preserve agent where it is part of the identity (e.g. Interop/Messenger), + // mirroring the normal JID display so distinct IDs stay distinct. + if jid.agent > 0 && jid.server.renders_agent() { + write!(f, ".{}", jid.agent)?; + } + if jid.device > 0 { + write!(f, ":{}", jid.device)?; + } + write!(f, "@{}", jid.server.as_str()) + } +} + impl<'a> fmt::Display for JidRef<'a> { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write_jid!(fallible f, &*self.user, self.server, self.agent, self.device) @@ -863,6 +972,47 @@ mod tests { use super::*; use std::str::FromStr; + /// `observe()` must never leak a raw phone number, must keep pseudonymous / + /// non-personal JIDs intact for correlation, and must preserve device. + #[test] + #[cfg(not(feature = "tracing-pii"))] + fn observe_redacts_phone_but_not_lid_or_group() { + let pn = Jid::from_str("5511999998888:7@s.whatsapp.net").unwrap(); + let shown = pn.observe().to_string(); + assert!(shown.starts_with("pn#"), "{shown}"); + assert!( + !shown.contains("5511999998888"), + "raw number leaked: {shown}" + ); + assert!(shown.ends_with(":7@s.whatsapp.net"), "device lost: {shown}"); + // Stable within the process so the same peer correlates across spans. + assert_eq!(shown, pn.observe().to_string()); + + // LID is pseudonymous and modern group ids are non-personal: rendered in full. + let lid = Jid::from_str("123456789@lid").unwrap(); + assert_eq!(lid.observe().to_string(), lid.to_string()); + let group = Jid::from_str("120363012345678901@g.us").unwrap(); + assert_eq!(group.observe().to_string(), group.to_string()); + + // Legacy group id embeds the creator phone ("<phone>-<ts>"): redact the + // numeric prefix, keep the timestamp. + let legacy = Jid::from_str("123456789-1620000000@g.us").unwrap(); + let ls = legacy.observe().to_string(); + assert!( + ls.starts_with("pn#") && ls.ends_with("-1620000000@g.us"), + "{ls}" + ); + // Exact no-leak invariant: the creator phone must not appear anywhere. + assert!(!ls.contains("123456789"), "creator phone leaked: {ls}"); + // The redacted prefix is the fixed-width keyed token, not the raw number. + let mid = &ls["pn#".len()..ls.find('-').unwrap()]; + assert_eq!(mid.len(), 16, "token width: {ls}"); + assert!( + mid.bytes().all(|b| b.is_ascii_hexdigit()), + "token hex: {ls}" + ); + } + /// Helper function to test a full parsing and display round-trip. fn assert_jid_roundtrip( input: &str, diff --git a/wacore/src/appstate_sync.rs b/wacore/src/appstate_sync.rs index e5b03f131..9a6245242 100644 --- a/wacore/src/appstate_sync.rs +++ b/wacore/src/appstate_sync.rs @@ -154,6 +154,7 @@ impl AppStateProcessor { /// Process an already-parsed single PatchList: download external blobs via /// `download`, then decode + apply. Lets a caller that parsed the response for /// pre-download avoid re-parsing it. See [`decode_patch_list_ref`]. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.process_parsed", level = "debug", skip_all, fields(name = ?pl.name), err(Debug)))] pub async fn process_parsed_patch_list<FDownload>( &self, mut pl: PatchList, @@ -214,6 +215,7 @@ impl AppStateProcessor { /// Process already-parsed patch lists, downloading any external blobs via /// `download`. Lets callers that already parsed the IQ response (e.g. to /// pre-download blobs) avoid re-parsing it. See [`decode_multi_patch_list_ref`]. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.process_lists", level = "debug", skip_all, fields(count = patch_lists.len()), err(Debug)))] pub async fn process_patch_lists<FDownload>( &self, patch_lists: Vec<PatchList>, @@ -242,6 +244,7 @@ impl AppStateProcessor { Ok(results) } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.process_list", level = "debug", skip_all, fields(name = ?pl.name), err(Debug)))] pub async fn process_patch_list( &self, pl: PatchList, @@ -389,6 +392,7 @@ impl AppStateProcessor { /// Returns `(patch_bytes, base_version)` where `base_version` is the collection /// version before the patch (for the IQ `version` attribute). Does NOT persist /// state — the caller must only persist after the server acknowledges the patch. + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.build_patch", level = "debug", skip_all, fields(name = %collection_name, count = mutations.len()), err(Debug)))] pub async fn build_patch( &self, collection_name: &str, @@ -466,6 +470,7 @@ impl AppStateProcessor { Ok(missing) } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.sync_collection", level = "debug", skip_all, fields(name = ?name), err(Debug)))] pub async fn sync_collection<D, FDownload>( &self, driver: &D, diff --git a/wacore/src/send/dm.rs b/wacore/src/send/dm.rs index 216a80823..a55bd4b25 100644 --- a/wacore/src/send/dm.rs +++ b/wacore/src/send/dm.rs @@ -45,6 +45,7 @@ pub struct PreparedDmStanza { pub message_secret: Option<[u8; crate::reporting_token::MESSAGE_SECRET_SIZE]>, } +#[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.dm_prepare", level = "debug", skip_all, fields(to = %to_jid.observe()), err(Debug)))] #[allow(clippy::too_many_arguments)] pub async fn prepare_dm_stanza< 'a, @@ -237,6 +238,7 @@ where /// `recipient_jid` is propagated verbatim from the retry receipt /// (`f && (k.recipient = f)` in `WAWebHandleRetryRequest`); pass `None` /// when the incoming receipt didn't carry it. +#[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.dm_retry", level = "debug", skip_all, fields(to = %to_jid.observe()), err(Debug)))] #[allow(clippy::too_many_arguments)] pub async fn prepare_dm_retry_stanza<S, I>( session_store: &mut S, diff --git a/wacore/src/send/encrypt.rs b/wacore/src/send/encrypt.rs index 44136cc35..b55524426 100644 --- a/wacore/src/send/encrypt.rs +++ b/wacore/src/send/encrypt.rs @@ -5,6 +5,10 @@ use super::*; /// Caller must hold `SenderKeyStore::sender_key_lock` for `sender_key_name` /// across the surrounding SKDM creation + this encrypt, so a concurrent send /// can't split the key between the SKDM and the skmsg. +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.send.encrypt_group", level = "debug", skip_all, err(Debug)) +)] pub async fn encrypt_group_message<S, R>( sender_key_store: &mut S, sender_key_name: &SenderKeyName, @@ -284,6 +288,7 @@ fn push_encrypt_result( /// /// Callers must hold per-device session locks before calling this function — /// concurrent ratchet mutations will corrupt Signal session state. +#[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.encrypt_fanout", level = "debug", skip_all, fields(count = devices.len()), err(Debug)))] pub async fn encrypt_for_devices<'a, S, I, P, SP>( runtime: &dyn Runtime, stores: &mut SignalStores<'a, S, I, P, SP>, @@ -325,8 +330,8 @@ where if stores.session_store.has_session(&reusable_addr).await? { log::debug!( "Using LID session {} for PN {} (LID-first lookup)", - lid_jid, - device_jid + lid_jid.observe(), + device_jid.observe() ); encryption_overrides[idx] = Some(lid_jid); continue; @@ -347,8 +352,8 @@ where let lid_jid = Jid::lid_device(lid_user, device_jid.device); log::debug!( "Will create LID session {} for PN {} (no existing session)", - lid_jid, - device_jid + lid_jid.observe(), + device_jid.observe() ); encryption_overrides[idx] = Some(lid_jid); } diff --git a/wacore/src/send/group.rs b/wacore/src/send/group.rs index 70201568a..1969c4029 100644 --- a/wacore/src/send/group.rs +++ b/wacore/src/send/group.rs @@ -5,6 +5,7 @@ use super::*; /// Pairwise-encrypted retry stanza for a single group participant. /// WA Web sends retries to the failing device only (RetryMsgJob.js:71), /// NOT as a sender-key broadcast to all participants. +#[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.group_retry", level = "debug", skip_all, fields(group = %group_jid.observe()), err(Debug)))] #[allow(clippy::too_many_arguments)] pub async fn prepare_group_retry_stanza<S, I>( session_store: &mut S, @@ -111,6 +112,7 @@ pub struct PreparedGroupStanza { pub sender_identity: Jid, } +#[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.group_prepare", level = "debug", skip_all, fields(to = %to_jid.observe()), err(Debug)))] #[allow(clippy::too_many_arguments)] pub async fn prepare_group_stanza< 'a, @@ -183,7 +185,7 @@ pub async fn prepare_group_stanza< log::debug!( "SKDM distribution to {} specific devices for group {}", target_devices.len(), - to_jid + to_jid.observe() ); Some(target_devices) } @@ -202,8 +204,8 @@ pub async fn prepare_group_stanza< { log::debug!( "Using phone number {} for LID {} device query", - phone_jid, - base_jid + phone_jid.observe(), + base_jid.observe() ); return phone_jid.to_non_ad(); } @@ -249,7 +251,7 @@ pub async fn prepare_group_stanza< log::debug!( "Converted {} devices to LID addressing for group {}", resolved_list.len(), - to_jid + to_jid.observe() ); } @@ -285,7 +287,7 @@ pub async fn prepare_group_stanza< log::debug!( "SKDM distribution list for {} resolved to {} devices", - to_jid, + to_jid.observe(), resolved_list.len(), ); @@ -312,13 +314,19 @@ pub async fn prepare_group_stanza< let phash_set = build_group_phash_set(src, &own_sending_jid); match MessageUtils::participant_list_hash(&phash_set) { Ok(phash) => phash_for_stanza = Some(phash), - Err(e) => log::warn!("Failed to compute group phash for {}: {:?}", to_jid, e), + Err(e) => { + log::warn!( + "Failed to compute group phash for {}: {:?}", + to_jid.observe(), + e + ) + } } } } else if let Some(ref distribution_list) = distribution_list { match MessageUtils::participant_list_hash(distribution_list) { Ok(phash) => phash_for_stanza = Some(phash), - Err(e) => log::warn!("Failed to compute phash for {}: {:?}", to_jid, e), + Err(e) => log::warn!("Failed to compute phash for {}: {:?}", to_jid.observe(), e), } } @@ -383,7 +391,7 @@ pub async fn prepare_group_stanza< Err(e) => { log::warn!( "SKDM distribution failed for group {}, continuing without it: {e}", - to_jid + to_jid.observe() ); if is_device_unregistered_error(&e) { had_unregistered_devices = true; @@ -527,6 +535,10 @@ pub(crate) fn collect_stale_device_users( /// Caller must hold `SenderKeyStore::sender_key_lock` for `sender_key_name` /// across this creation + the matching skmsg encrypt (see `encrypt_group_message`). +#[cfg_attr( + feature = "tracing", + tracing::instrument(name = "wa.send.skdm_create", level = "debug", skip_all, err(Debug)) +)] pub async fn create_sender_key_distribution_message_for_group( store: &mut (dyn SenderKeyStore + Send + Sync), sender_key_name: &SenderKeyName, diff --git a/wacore/src/send/peer.rs b/wacore/src/send/peer.rs index 8ca5ec381..4d1016398 100644 --- a/wacore/src/send/peer.rs +++ b/wacore/src/send/peer.rs @@ -30,6 +30,7 @@ where .await } +#[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.peer_prepare", level = "debug", skip_all, fields(to = %transport_jid.observe()), err(Debug)))] #[allow(clippy::too_many_arguments)] pub async fn prepare_peer_stanza_with_options<S, I>( session_store: &mut S, diff --git a/wacore/src/types/jid.rs b/wacore/src/types/jid.rs index fe90b1030..3a75afdfb 100644 --- a/wacore/src/types/jid.rs +++ b/wacore/src/types/jid.rs @@ -126,6 +126,23 @@ impl JidExt for Jid { } } +/// Privacy-aware rendering of a Signal [`ProtocolAddress`] for tracing/logs. +/// +/// The address name embeds the peer JID (a phone number for PN peers) plus the +/// device, so logging it directly leaks PII. This replaces the whole name with a +/// keyed token (same per-process scheme as `Jid::observe`): stable per peer-device +/// for correlation, but not reversible to the number. (The Signal `device_id` is +/// always 0 here — the device lives inside the name — so it is not shown.) +pub fn observe_protocol_address(addr: &ProtocolAddress) -> String { + if cfg!(feature = "tracing-pii") { + return addr.name().to_string(); + } + format!( + "addr#{:016x}", + wacore_binary::jid::observe_token(addr.name()) + ) +} + #[cfg(test)] mod tests { use super::*; diff --git a/wacore/src/usync.rs b/wacore/src/usync.rs index 38a18a2d9..b9be8d07d 100644 --- a/wacore/src/usync.rs +++ b/wacore/src/usync.rs @@ -67,7 +67,7 @@ pub fn parse_get_user_devices_response_with_phash(resp_node: &Node) -> Result<Ve let user_jid = user_node.attrs().jid("jid"); let device_list_node = user_node .get_optional_child_by_tag(&["devices", "device-list"]) - .ok_or_else(|| anyhow!("<device-list> not found for user {user_jid}"))?; + .ok_or_else(|| anyhow!("<device-list> not found for user {}", user_jid.observe()))?; // Extract phash from device-list node attributes let phash = device_list_node @@ -98,7 +98,7 @@ pub fn parse_get_user_devices_response_with_phash(resp_node: &Node) -> Result<Ve let device_id: u16 = match device_id_str.parse() { Ok(id) => id, Err(_) => { - log::warn!(target: "usync", "invalid device id '{device_id_str}' for user {user_jid}, skipping"); + log::warn!(target: "usync", "invalid device id '{device_id_str}' for user {}, skipping", user_jid.observe()); continue; } }; @@ -119,7 +119,8 @@ pub fn parse_get_user_devices_response_with_phash(resp_node: &Node) -> Result<Ve if has_companion && key_index_bytes.is_none() { log::warn!( target: "usync", - "User {user_jid} has companion devices but no signedKeyIndexBytes, skipping" + "User {} has companion devices but no signedKeyIndexBytes, skipping", + user_jid.observe() ); continue; }