From 0380e6e00ed02f7d4b8154b3754e8d6f245e8923 Mon Sep 17 00:00:00 2001 From: Jean-Louis Queguiner Date: Fri, 11 Sep 2026 15:06:30 -0400 Subject: [PATCH] fix(whatsapp): never report a send as sent without server confirmation wa-rs 0.2 returns a message id as soon as the stanza bytes are written to the noise socket. Nothing waits for the server ack, so a send on a dying socket returns Ok(id) and the CLI prints "Message sent" for a message the server never saw. Measured in production on 2026-09-11. The library's ack machinery is not reachable from a consumer: Client::response_waiters is pub(crate), and incoming stanzas are consumed by handlers::basic::AckHandler without being dispatched on the public event bus (there is no Event::ServerAck). So confirm acceptance with stream ordering instead. All outbound frames go through one NoiseSocket sender task, so frames hit the TCP stream in call order. After the message, issue a w:p ping IQ and wait, bounded, for its pong: a pong proves the server consumed the stream past our message frame. Adds crates/void-whatsapp/src/connector/delivery.rs: - precheck(): fails fast when the connection has no live socket or is not logged in, before anything is written, - confirm_accepted(): 12s bounded barrier after the write, with timeout, bad-server-reply and transport outcomes kept distinguishable, - with_send_timeout(): hard cap on the write itself, which wa-rs does not bound. Wired into all four send paths in ops.rs (send, reply, and their notes-to-self variants). No error variant can be read as a success, and the two unknown cases say delivery is unknown rather than claiming loss. Co-Authored-By: Claude Opus 5 --- .../void-whatsapp/src/connector/delivery.rs | 397 ++++++++++++++++++ crates/void-whatsapp/src/connector/mod.rs | 1 + crates/void-whatsapp/src/connector/ops.rs | 39 +- crates/void-whatsapp/src/connector/tests.rs | 151 +++++++ 4 files changed, 580 insertions(+), 8 deletions(-) create mode 100644 crates/void-whatsapp/src/connector/delivery.rs diff --git a/crates/void-whatsapp/src/connector/delivery.rs b/crates/void-whatsapp/src/connector/delivery.rs new file mode 100644 index 0000000..1319d2a --- /dev/null +++ b/crates/void-whatsapp/src/connector/delivery.rs @@ -0,0 +1,397 @@ +//! Server-acceptance verification for outbound WhatsApp sends. +//! +//! `wa-rs` 0.2 hands back a message id as soon as the stanza bytes are given to +//! the noise socket: `send::send_message_with_options` generates the id locally, +//! calls `send_message_impl`, which ends on `Client::send_node`, which only +//! marshals, encrypts and writes. Nothing waits for the server ``. A send +//! on a dying socket therefore returns `Ok(id)` for a message the server never +//! saw. Measured on 2026-09-11: the CLI printed "Message sent (id: ...)" in +//! 0.66 s, the message was in no store afterwards, and the wa-rs message loop +//! exited 52 s later. +//! +//! The library does wait for acks internally, but that machinery is not +//! reachable from a consumer: `Client::response_waiters` is `pub(crate)`, and +//! incoming `` stanzas are swallowed by `handlers::basic::AckHandler` +//! without being dispatched on the public event bus (there is no +//! `Event::ServerAck`). So a consumer cannot observe the ack for its own send. +//! +//! What a consumer can do is prove the stanza reached the server, using stream +//! ordering. Every outbound frame goes through a single `NoiseSocket` sender +//! task (`socket/noise_socket.rs`: `encrypt_and_send` pushes a job onto an mpsc +//! queue and awaits that job's result), so frames hit the TCP stream in call +//! order, and the caller does not return until its own frame was handed to the +//! transport. Issuing a `w:p` ping IQ *after* the message and waiting for its +//! pong is therefore a barrier: a pong proves the server consumed the stream +//! past our message frame. No pong means we cannot claim the message was sent. +//! +//! That is what this module implements: a liveness precheck before the write, +//! and a bounded barrier after it. + +use std::time::Duration; + +use thiserror::Error; +use tracing::{debug, warn}; +use wa_rs::client::Client; +use wa_rs::request::IqError; +use wa_rs_core::iq::keepalive::KeepaliveSpec; + +/// Bounded wait for the post-send barrier to round-trip. +pub(crate) const ACK_TIMEOUT: Duration = Duration::from_secs(12); + +/// Hard ceiling on the barrier call. `send_iq` applies `ACK_TIMEOUT` to the +/// response wait only: the `send_node` that precedes it can itself block on a +/// stuck transport, so the whole call gets an outer deadline too. +const BARRIER_HARD_TIMEOUT: Duration = Duration::from_secs(15); + +/// Hard ceiling on the stanza write. `send_message_with_options` has no +/// internal timeout, and the noise sender task can hang on `transport.send`. +pub(crate) const SEND_TIMEOUT: Duration = Duration::from_secs(30); + +/// `Client::is_connected` reads the noise socket slot with `try_lock`, so it +/// reports `false` while another send briefly holds that mutex. Retry a few +/// times before declaring the connection down, to avoid failing a healthy send. +const LIVENESS_ATTEMPTS: u32 = 5; +const LIVENESS_RETRY_DELAY: Duration = Duration::from_millis(20); + +/// Why the barrier did not confirm the send. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum UnconfirmedReason { + /// No response from the server before the deadline. + Timeout, + /// The server answered, but not with a clean result. + BadServerReply(String), +} + +impl std::fmt::Display for UnconfirmedReason { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Timeout => write!(f, "no server response before the deadline"), + Self::BadServerReply(detail) => write!(f, "server replied with {detail}"), + } + } +} + +/// A send that could not be reported as successful. +/// +/// The wording matters as much as the variant: none of these may read as a +/// success, and the two "unknown" cases must not read as a confirmed loss. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub(crate) enum SendFailure { + #[error( + "WhatsApp send aborted on connection '{connection_id}': {reason}. \ + Nothing was written to the socket, the message was not sent." + )] + NotLive { + connection_id: String, + reason: &'static str, + }, + + #[error( + "WhatsApp send NOT confirmed on connection '{connection_id}': the stanza for message \ + {message_id} was written but the server did not confirm it within {}s ({reason}). \ + Delivery is unknown, do not assume the message arrived.", + .waited.as_secs() + )] + Unconfirmed { + connection_id: String, + message_id: String, + waited: Duration, + reason: UnconfirmedReason, + }, + + #[error( + "WhatsApp send NOT confirmed on connection '{connection_id}': the transport failed while \ + confirming message {message_id} ({detail}). The message very likely never reached \ + WhatsApp." + )] + Transport { + connection_id: String, + message_id: String, + detail: String, + }, + + #[error( + "WhatsApp send stalled on connection '{connection_id}': writing the stanza did not \ + complete within {}s. Delivery is unknown, do not assume the message arrived.", + .waited.as_secs() + )] + SendStalled { + connection_id: String, + waited: Duration, + }, +} + +/// Precheck decision, split from the `Client` so it is testable on its own. +/// +/// Both flags are required: `is_connected` alone is true during a reconnect +/// handshake, before the session is usable again. +pub(crate) fn liveness_failure( + connection_id: &str, + connected: bool, + logged_in: bool, +) -> Option { + let reason = match (connected, logged_in) { + (true, true) => return None, + (false, _) => "the connection has no live socket (disconnected or reconnecting)", + (true, false) => "the socket is up but the session is not logged in yet", + }; + Some(SendFailure::NotLive { + connection_id: connection_id.to_string(), + reason, + }) +} + +/// Fails fast when the connection cannot carry a stanza right now. +pub(crate) async fn precheck(client: &Client, connection_id: &str) -> Result<(), SendFailure> { + let mut failure = None; + for attempt in 0..LIVENESS_ATTEMPTS { + match liveness_failure(connection_id, client.is_connected(), client.is_logged_in()) { + None => return Ok(()), + Some(f) => { + failure = Some(f); + if attempt + 1 < LIVENESS_ATTEMPTS { + tokio::time::sleep(LIVENESS_RETRY_DELAY).await; + } + } + } + } + let failure = failure.expect("loop ran at least once without returning Ok"); + warn!(connection_id = %connection_id, error = %failure, "WhatsApp send rejected by liveness precheck"); + Err(failure) +} + +/// Turns a barrier IQ result into a send verdict. +/// +/// Split from the network call so every branch is testable without a socket. +pub(crate) fn classify_barrier( + connection_id: &str, + message_id: &str, + result: Result<(), IqError>, +) -> Result<(), SendFailure> { + let unconfirmed = |reason| SendFailure::Unconfirmed { + connection_id: connection_id.to_string(), + message_id: message_id.to_string(), + waited: ACK_TIMEOUT, + reason, + }; + let transport = |detail: String| SendFailure::Transport { + connection_id: connection_id.to_string(), + message_id: message_id.to_string(), + detail, + }; + + match result { + // The pong came back, so the server consumed the stream past our + // message frame. The stanza was accepted. + Ok(()) => Ok(()), + + Err(IqError::Timeout) => Err(unconfirmed(UnconfirmedReason::Timeout)), + + // A reply came back but was not a clean result. The stream did + // round-trip, yet we refuse to read that as proof of acceptance. + Err(e @ IqError::ServerError { .. }) | Err(e @ IqError::ParseError(_)) => Err(unconfirmed( + UnconfirmedReason::BadServerReply(e.to_string()), + )), + + Err(e) => Err(transport(e.to_string())), + } +} + +/// Waits, bounded, for proof that the server consumed the message stanza. +/// +/// Call this immediately after the send, on the same `Client`, so no other +/// stanza can be interleaved by this code path between the two. +pub(crate) async fn confirm_accepted( + client: &Client, + connection_id: &str, + message_id: &str, +) -> Result<(), SendFailure> { + let barrier = client.execute(KeepaliveSpec::with_timeout(ACK_TIMEOUT)); + let result = match tokio::time::timeout(BARRIER_HARD_TIMEOUT, barrier).await { + Ok(result) => result, + Err(_) => Err(IqError::Timeout), + }; + + match classify_barrier(connection_id, message_id, result) { + Ok(()) => { + debug!(connection_id = %connection_id, message_id = %message_id, "WhatsApp send confirmed by server"); + Ok(()) + } + Err(failure) => { + warn!(connection_id = %connection_id, message_id = %message_id, error = %failure, "WhatsApp send not confirmed"); + Err(failure) + } + } +} + +/// Runs the stanza write under a hard deadline. +pub(crate) async fn with_send_timeout(connection_id: &str, fut: F) -> anyhow::Result +where + F: std::future::Future>, +{ + with_send_deadline(connection_id, SEND_TIMEOUT, fut).await +} + +/// Deadline is a parameter so tests can exercise the stall branch without +/// burning [`SEND_TIMEOUT`] of wall clock. +async fn with_send_deadline( + connection_id: &str, + deadline: Duration, + fut: F, +) -> anyhow::Result +where + F: std::future::Future>, +{ + match tokio::time::timeout(deadline, fut).await { + Ok(result) => result, + Err(_) => Err(SendFailure::SendStalled { + connection_id: connection_id.to_string(), + waited: deadline, + } + .into()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use wa_rs::socket::error::SocketError; + use wa_rs_binary::node::{Attrs, Node}; + + const CONN: &str = "WA-french"; + const MSG: &str = "3EB01E021E254E73BF2930"; + + fn assert_never_claims_success(failure: &SendFailure) { + let text = failure.to_string(); + assert!( + !text.contains("Message sent"), + "failure text must not read as a success: {text}" + ); + assert!( + text.contains("NOT confirmed") || text.contains("aborted") || text.contains("stalled"), + "failure text must be unambiguous: {text}" + ); + } + + #[test] + fn liveness_passes_only_when_connected_and_logged_in() { + assert!(liveness_failure(CONN, true, true).is_none()); + } + + #[test] + fn liveness_rejects_dead_socket() { + let failure = liveness_failure(CONN, false, true).expect("must reject"); + assert!(matches!(failure, SendFailure::NotLive { .. })); + assert!(failure.to_string().contains("no live socket")); + assert!(failure.to_string().contains("was not sent")); + assert_never_claims_success(&failure); + } + + #[test] + fn liveness_rejects_socket_up_but_not_logged_in() { + // The state during a reconnect handshake: this is what the production + // failure on 2026-09-11 wrote into. + let failure = liveness_failure(CONN, true, false).expect("must reject"); + assert!(failure.to_string().contains("not logged in")); + assert_never_claims_success(&failure); + } + + #[test] + fn liveness_rejects_fully_down_connection() { + let failure = liveness_failure(CONN, false, false).expect("must reject"); + assert!(matches!(failure, SendFailure::NotLive { .. })); + } + + #[test] + fn barrier_pong_confirms_the_send() { + assert!(classify_barrier(CONN, MSG, Ok(())).is_ok()); + } + + #[test] + fn barrier_timeout_is_unconfirmed_not_transport() { + let failure = classify_barrier(CONN, MSG, Err(IqError::Timeout)).expect_err("must fail"); + assert!(matches!( + failure, + SendFailure::Unconfirmed { + reason: UnconfirmedReason::Timeout, + .. + } + )); + let text = failure.to_string(); + assert!(text.contains(MSG)); + assert!(text.contains("12s")); + assert!(text.contains("Delivery is unknown")); + assert_never_claims_success(&failure); + } + + #[test] + fn barrier_transport_errors_are_distinguishable_from_timeout() { + let cases = [ + IqError::NotConnected, + IqError::Socket(SocketError::Crypto("write failed".into())), + IqError::Disconnected(Node::new("stream:error", Attrs::new(), None)), + IqError::InternalChannelClosed, + ]; + for case in cases { + let label = case.to_string(); + let failure = classify_barrier(CONN, MSG, Err(case)).expect_err("must fail"); + assert!( + matches!(failure, SendFailure::Transport { .. }), + "{label} should classify as transport, got {failure:?}" + ); + assert!(failure.to_string().contains("never reached WhatsApp")); + assert_never_claims_success(&failure); + } + } + + #[test] + fn barrier_bad_server_reply_is_unconfirmed() { + let failure = classify_barrier( + CONN, + MSG, + Err(IqError::ServerError { + code: 500, + text: "internal".into(), + }), + ) + .expect_err("must fail"); + assert!(matches!( + failure, + SendFailure::Unconfirmed { + reason: UnconfirmedReason::BadServerReply(_), + .. + } + )); + assert_never_claims_success(&failure); + } + + #[test] + fn send_stalled_reports_unknown_delivery() { + let failure = SendFailure::SendStalled { + connection_id: CONN.into(), + waited: SEND_TIMEOUT, + }; + assert!(failure.to_string().contains("30s")); + assert_never_claims_success(&failure); + } + + #[tokio::test] + async fn with_send_timeout_passes_through_success() { + let value = with_send_timeout(CONN, async { Ok::<_, anyhow::Error>("id-1".to_string()) }) + .await + .expect("should pass through"); + assert_eq!(value, "id-1"); + } + + #[tokio::test] + async fn with_send_deadline_fails_when_the_write_hangs() { + let err = with_send_deadline(CONN, Duration::from_millis(20), async { + // A noise sender task stuck on `transport.send` never returns. + std::future::pending::>().await + }) + .await + .expect_err("a hanging write must not succeed"); + assert!(err.to_string().contains("stalled"), "{err}"); + assert!(!err.to_string().contains("Message sent"), "{err}"); + } +} diff --git a/crates/void-whatsapp/src/connector/mod.rs b/crates/void-whatsapp/src/connector/mod.rs index 704937b..85e70bd 100644 --- a/crates/void-whatsapp/src/connector/mod.rs +++ b/crates/void-whatsapp/src/connector/mod.rs @@ -1,6 +1,7 @@ //! WhatsApp connector: struct, Connector impl, and orchestration. mod connector_trait; +mod delivery; mod extract; mod media; mod ops; diff --git a/crates/void-whatsapp/src/connector/ops.rs b/crates/void-whatsapp/src/connector/ops.rs index 47f240a..aab855c 100644 --- a/crates/void-whatsapp/src/connector/ops.rs +++ b/crates/void-whatsapp/src/connector/ops.rs @@ -10,6 +10,7 @@ use wa_rs::client::Client; use wa_rs::send::SendOptions; use wa_rs_proto::whatsapp::ContextInfo; +use super::delivery::{confirm_accepted, precheck, with_send_timeout}; use super::media::{download_media_with_client, upload_and_build_media_message}; use super::self_chat::send_self_chat_message; use super::send::{build_wa_message, parse_jid}; @@ -74,8 +75,17 @@ impl WhatsAppConnector { let client = self.require_sync_client().await?; let identity = self.own_identity.lock().expect("mutex").clone(); + // Refuse to write into a dead socket. See `delivery` for why the + // library's own `Ok(id)` is not evidence that anything was sent. + precheck(&client, &self.config_id).await?; + if identity.should_route_as_self_chat(to) { - let msg_id = send_self_chat_message(&client, &identity, content, None).await?; + let msg_id = with_send_timeout( + &self.config_id, + send_self_chat_message(&client, &identity, content, None), + ) + .await?; + confirm_accepted(&client, &self.config_id, &msg_id).await?; super::presence::schedule_unavailable(Arc::clone(&client)); return Ok(msg_id); } @@ -102,9 +112,12 @@ impl WhatsAppConnector { _ => build_wa_message(&content, None)?, }; - let msg_id = client - .send_message_with_options(jid, msg, SendOptions::default()) - .await?; + let msg_id = with_send_timeout( + &self.config_id, + client.send_message_with_options(jid, msg, SendOptions::default()), + ) + .await?; + confirm_accepted(&client, &self.config_id, &msg_id).await?; debug!(connection_id = %self.config_id, message_id = %msg_id, "WhatsApp message sent via sync"); super::presence::schedule_unavailable(Arc::clone(&client)); Ok(msg_id) @@ -128,6 +141,8 @@ impl WhatsAppConnector { let client = self.require_sync_client().await?; let identity = self.own_identity.lock().expect("mutex").clone(); + precheck(&client, &self.config_id).await?; + if identity.should_route_as_self_chat(&chat_jid_str) { let context_info = if in_thread { Some(ContextInfo { @@ -143,7 +158,12 @@ impl WhatsAppConnector { in_thread, "sending WhatsApp notes-to-self reply via sync" ); - let msg_id = send_self_chat_message(&client, &identity, content, context_info).await?; + let msg_id = with_send_timeout( + &self.config_id, + send_self_chat_message(&client, &identity, content, context_info), + ) + .await?; + confirm_accepted(&client, &self.config_id, &msg_id).await?; super::presence::schedule_unavailable(Arc::clone(&client)); return Ok(msg_id); } @@ -178,9 +198,12 @@ impl WhatsAppConnector { _ => build_wa_message(&content, context_info)?, }; - let msg_id = client - .send_message_with_options(jid, msg, SendOptions::default()) - .await?; + let msg_id = with_send_timeout( + &self.config_id, + client.send_message_with_options(jid, msg, SendOptions::default()), + ) + .await?; + confirm_accepted(&client, &self.config_id, &msg_id).await?; debug!(connection_id = %self.config_id, message_id = %msg_id, "WhatsApp reply sent via sync"); super::presence::schedule_unavailable(Arc::clone(&client)); Ok(msg_id) diff --git a/crates/void-whatsapp/src/connector/tests.rs b/crates/void-whatsapp/src/connector/tests.rs index e55a8fd..9dab60c 100644 --- a/crates/void-whatsapp/src/connector/tests.rs +++ b/crates/void-whatsapp/src/connector/tests.rs @@ -799,3 +799,154 @@ fn is_system_message_pin_in_chat() { }; assert!(sync::is_system_message(&msg)); } + +fn text_content(body: &str) -> MessageContent { + MessageContent::Text { + body: body.to_string(), + subject: None, + append_signature: false, + signature_from: None, + cc: None, + bcc: None, + } +} + +/// Builds a real `wa_rs::Client` that has never connected, so `is_connected()` +/// is false and its noise socket slot is empty. This is the dead-socket state +/// the 2026-09-11 failure sent into. +async fn disconnected_client() -> Arc { + use wa_rs::store::persistence_manager::PersistenceManager; + use wa_rs::store::traits::Backend; + + let db = format!( + "file:void_wa_delivery_{}?mode=memory&cache=shared", + uuid::Uuid::new_v4().simple() + ); + let backend = Arc::new( + wa_rs_sqlite_storage::SqliteStore::new(&db) + .await + .expect("in-memory wa-rs store"), + ) as Arc; + let pm = Arc::new( + PersistenceManager::new(backend) + .await + .expect("persistence manager"), + ); + let (client, _sync_rx) = wa_rs::client::Client::new( + pm, + Arc::new(wa_rs_tokio_transport::TokioWebSocketTransportFactory::new()), + Arc::new(wa_rs_ureq_http::UreqHttpClient::new()), + None, + ) + .await; + client +} + +fn connector_with_client(client: Arc) -> WhatsAppConnector { + let connector = WhatsAppConnector::new("WA-french", "/nonexistent/session.db"); + *connector.client.try_lock().expect("fresh mutex") = Some(client); + connector +} + +#[tokio::test] +async fn send_on_dead_socket_fails_instead_of_reporting_success() { + let connector = connector_with_client(disconnected_client().await); + + let err = connector + .send_via_sync( + "33612345678", + text_content("this must never be reported as sent"), + ) + .await + .expect_err("a send on a dead socket must not succeed"); + + let text = err.to_string(); + assert!( + text.contains("WhatsApp send aborted"), + "unexpected error: {text}" + ); + assert!(text.contains("WA-french"), "unexpected error: {text}"); + assert!(text.contains("was not sent"), "unexpected error: {text}"); + // The old behaviour returned Ok(message_id) here. + assert!(!text.contains("Message sent"), "unexpected error: {text}"); +} + +#[tokio::test] +async fn reply_on_dead_socket_fails_instead_of_reporting_success() { + let connector = connector_with_client(disconnected_client().await); + + let err = connector + .reply_via_sync( + "33612345678@s.whatsapp.net:3EB01E021E254E73BF2930", + text_content("this must never be reported as sent"), + false, + ) + .await + .expect_err("a reply on a dead socket must not succeed"); + + assert!( + err.to_string().contains("WhatsApp send aborted"), + "unexpected error: {err}" + ); +} + +#[tokio::test] +async fn dead_socket_send_fails_fast() { + let connector = connector_with_client(disconnected_client().await); + + let started = std::time::Instant::now(); + let _ = connector + .send_via_sync("33612345678", text_content("fail fast")) + .await + .expect_err("must fail"); + let elapsed = started.elapsed(); + + // The precheck retries a few times to absorb a transient `try_lock` miss, + // then gives up. It must not sit on the 12s barrier or the 30s write cap. + assert!( + elapsed < std::time::Duration::from_secs(2), + "precheck took {elapsed:?}, expected a fast failure" + ); +} + +#[tokio::test] +async fn barrier_on_dead_socket_reports_transport_failure() { + use super::delivery::confirm_accepted; + + let client = disconnected_client().await; + let err = confirm_accepted(&client, "WA-french", "3EB01E021E254E73BF2930") + .await + .expect_err("the barrier cannot round-trip on a dead socket"); + + let text = err.to_string(); + assert!(text.contains("NOT confirmed"), "unexpected error: {text}"); + assert!( + text.contains("3EB01E021E254E73BF2930"), + "unexpected error: {text}" + ); +} + +/// The production failure of 2026-09-11, reproduced at the seam. +/// +/// `wa-rs` handed back `Ok(id)` for a stanza the server never saw, and the CLI +/// printed "Message sent (id: 3EB01E021E254E73BF2930)". Here the write is +/// stubbed to succeed exactly like that, and the barrier that follows it cannot +/// round-trip. The composed result must be an error. +#[tokio::test] +async fn a_send_that_returns_an_id_but_is_never_confirmed_is_a_failure() { + use super::delivery::{confirm_accepted, with_send_timeout}; + + let client = disconnected_client().await; + + let msg_id = with_send_timeout("WA-french", async { + Ok::<_, anyhow::Error>("3EB01E021E254E73BF2930".to_string()) + }) + .await + .expect("the library reports success as soon as the bytes are written"); + assert_eq!(msg_id, "3EB01E021E254E73BF2930"); + + let err = confirm_accepted(&client, "WA-french", &msg_id) + .await + .expect_err("an unconfirmed send must not be reported as sent"); + assert!(err.to_string().contains("NOT confirmed"), "{err}"); +}