From d80494c206f0faa2d5787939f0035f60adaf4fcf Mon Sep 17 00:00:00 2001 From: otsmr Date: Sat, 29 Aug 2026 23:42:53 +0200 Subject: [PATCH] fix multiple issues --- rust/src/api/messages/incoming.rs | 63 ++++++++++ .../api/websocket/client_to_server.proto | 17 +++ .../api/websocket/server_to_client.proto | 14 +++ rust/src/api/runtime/auth.rs | 5 + rust/src/api/runtime/client.rs | 114 +++++++++++++++++- rust/src/api/server/transport.rs | 16 +++ rust/src/services/notifications.rs | 5 + rust/tests/api.rs | 11 +- rust/tests/api/contacts.rs | 5 +- rust/tests/api/group_resilience.rs | 13 +- rust/tests/api/media.rs | 11 +- rust/tests/api/notifications.rs | 11 +- rust/tests/api/recovery.rs | 11 +- rust/tests/api/server_api.rs | 3 +- rust/tests/api/session_recovery.rs | 3 +- rust/tests/api/tester.rs | 33 +++++ rust/tests/api/user_discovery.rs | 3 +- 17 files changed, 286 insertions(+), 52 deletions(-) diff --git a/rust/src/api/messages/incoming.rs b/rust/src/api/messages/incoming.rs index e00b155c..ce44d193 100644 --- a/rust/src/api/messages/incoming.rs +++ b/rust/src/api/messages/incoming.rs @@ -64,6 +64,9 @@ pub(crate) async fn handle_server_message( } ok::Ok::None(true) } + Kind::PendingMessagesV2(batch) => { + return Ok(acknowledge_pending_messages(ctx, batch).await); + } Kind::SealedSenderMessage(message) => { if let Err(error) = handle_sealed_message(ctx, message.body).await { tracing::warn!("failed to process sealed-sender message: {error}"); @@ -97,6 +100,66 @@ pub(crate) async fn handle_server_message( }) } +/// Processes a reliable-mailbox batch and reports back exactly which delivery +/// IDs are now durable locally. The server deletes only those rows, so anything +/// left out is redelivered on the next drain. +/// +/// Deduplication is the `received_receipts` claim inside +/// [`handle_decoded_server_message`]: it is committed in the same transaction +/// that persists the message, is keyed on the end-to-end receipt ID, and is +/// never purged. A redelivered envelope is therefore recognised before it is +/// decrypted, whichever transport carried it. +async fn acknowledge_pending_messages( + ctx: &Arc, + batch: server_to_client::PendingMessagesV2, +) -> client_to_server::Response { + let mut delivery_ids = Vec::with_capacity(batch.messages.len()); + + for message in batch.messages { + let delivery_id = message.delivery_id; + let result = handle_new_server_message( + ctx, + NewMessage { + from_user_id: message.from_user_id, + body: message.body, + }, + ) + .await; + + match result { + // Committed, or recognised as a duplicate. Either way it is durable. + Ok(()) => delivery_ids.push(delivery_id), + // An envelope that cannot be decoded will never decode. Acknowledge + // it so one poisoned row cannot be redelivered forever. + Err( + error @ (TwonlyError::ProtobufDecode(_) | TwonlyError::UnknownProtobufEnumValue(_)), + ) => { + tracing::warn!( + delivery_id, + "dropping an undecodable mailbox message: {error}" + ); + delivery_ids.push(delivery_id); + } + // Anything else (storage, network, Signal state) may succeed later. + // Leaving the ID out keeps the row on the server for a retry. + Err(error) => { + tracing::warn!( + delivery_id, + "mailbox message not persisted, will retry: {error}" + ); + } + } + } + + client_to_server::Response { + response: Some(Response::Ok(client_to_server::response::Ok { + ok: Some(ok::Ok::AcknowledgedPendingMessages( + client_to_server::response::AcknowledgedPendingMessages { delivery_ids }, + )), + })), + } +} + pub(crate) async fn handle_new_server_message( ctx: &Arc, server_message: NewMessage, diff --git a/rust/src/api/proto/api/websocket/client_to_server.proto b/rust/src/api/proto/api/websocket/client_to_server.proto index 43991cf9..0b3eb66e 100644 --- a/rust/src/api/proto/api/websocket/client_to_server.proto +++ b/rust/src/api/proto/api/websocket/client_to_server.proto @@ -75,6 +75,9 @@ message Handshake { optional string app_version = 3; optional int64 device_id = 4; optional bool in_background = 5; + // Set by clients that understand PendingMessagesV2 / RequestPendingMessages. + // Absent or false means the legacy NewMessage(s) mailbox protocol is used. + optional bool supports_mailbox_v2 = 6; } message AuthenticateWithLoginToken { @@ -83,6 +86,8 @@ message Handshake { string app_version = 3; int64 device_id = 4; bool in_background = 5; + // See Authenticate.supports_mailbox_v2. + optional bool supports_mailbox_v2 = 6; } message GetServerKeyForPasswordLessRecovery { @@ -260,6 +265,10 @@ message ApplicationData { repeated bytes token_requests = 1; } + // Asks the server to (re)start a mailbox drain for the authenticated user. + // The server responds immediately; messages arrive as PendingMessagesV2. + message RequestPendingMessages {} + oneof ApplicationData { TextMessage textMessage = 1; GetUserByUsername getUserByUsername = 2; @@ -291,6 +300,7 @@ message ApplicationData { UploadPqcPreKeys upload_pqc_prekeys = 40; GetPrivacyPassParameters get_privacy_pass_parameters = 41; IssuePrivacyPassTokens issue_privacy_pass_tokens = 42; + RequestPendingMessages request_pending_messages = 43; } } @@ -309,11 +319,18 @@ message Response { repeated ApplicationData.PqcPreKey prekeys = 1; } + // Response to a PendingMessagesV2 batch. Contains only the delivery IDs the + // client has durably persisted; the server deletes exactly those rows. + message AcknowledgedPendingMessages { + repeated int64 delivery_ids = 1; + } + message Ok { oneof Ok { bool None = 1; Prekeys prekeys = 2; PqcPrekeys prekeys_pqc = 3; + AcknowledgedPendingMessages acknowledged_pending_messages = 4; } } diff --git a/rust/src/api/proto/api/websocket/server_to_client.proto b/rust/src/api/proto/api/websocket/server_to_client.proto index 1cb35722..f9edc79d 100644 --- a/rust/src/api/proto/api/websocket/server_to_client.proto +++ b/rust/src/api/proto/api/websocket/server_to_client.proto @@ -21,6 +21,7 @@ message V0 { SealedSenderMessage sealedSenderMessage = 9; SealedSenderMessages sealedSenderMessages = 10; bool mailboxDrained = 11; + PendingMessagesV2 pendingMessagesV2 = 12; } } @@ -33,6 +34,19 @@ message NewMessages { repeated NewMessage newMessages = 1; } +// Reliable mailbox delivery. `delivery_id` is a server-generated mailbox row +// identifier used only to acknowledge delivery; it is unrelated to and reveals +// nothing about the end-to-end encrypted message ID inside `body`. +message PendingMessageV2 { + int64 delivery_id = 1; + int64 from_user_id = 2; + bytes body = 3; +} + +message PendingMessagesV2 { + repeated PendingMessageV2 messages = 1; +} + message SealedSenderMessage { string message_id = 1; bytes body = 2; diff --git a/rust/src/api/runtime/auth.rs b/rust/src/api/runtime/auth.rs index 6767a441..f5452b95 100644 --- a/rust/src/api/runtime/auth.rs +++ b/rust/src/api/runtime/auth.rs @@ -115,6 +115,7 @@ impl ApiAuthHandshaker { app_version, device_id, in_background: self.in_background, + supports_mailbox_v2: Some(true), }, ); self.request_handshake(sender, receiver, handshake) @@ -144,6 +145,7 @@ impl ApiAuthHandshaker { app_version: Some(app_version), device_id: Some(device_id), in_background: Some(self.in_background), + supports_mailbox_v2: Some(true), }, ); match self.request_handshake(sender, receiver, handshake).await { @@ -286,6 +288,9 @@ impl ApiAuthHandshaker { if let Some(client) = self.api_client.upgrade() { tokio::spawn(async move { client.set_state(ApiConnectionState::Authenticated).await; + // Runs on every (re)connect, so this covers both the initial + // authentication and reconnection catch-up. + client.request_catch_up().await; }); } diff --git a/rust/src/api/runtime/client.rs b/rust/src/api/runtime/client.rs index 4f4aaf06..cba471ca 100644 --- a/rust/src/api/runtime/client.rs +++ b/rust/src/api/runtime/client.rs @@ -10,13 +10,29 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, LazyLock, Weak}; use std::time::Duration; -use stream_tungstenite::WebSocketClient; -use tokio::sync::{broadcast, oneshot, Mutex, RwLock}; +use stream_tungstenite::{ClientConfig, WebSocketClient}; +use tokio::sync::{broadcast, mpsc, oneshot, Mutex, RwLock}; use super::auth::ApiAuthHandshaker; pub(super) type PendingRequests = Arc>>>>; +/// Base delay of the safety catch-up while the socket stays connected. Every +/// other trigger (authentication, reconnection, foregrounding, a push) fires a +/// catch-up immediately, so this only has to cover a silently stalled drain. +const CATCH_UP_INTERVAL: Duration = Duration::from_secs(60); +/// Spread across clients so reconnect storms do not line their pulls up. +const CATCH_UP_JITTER: Duration = Duration::from_secs(15); + +/// First reconnect attempt after a drop. Kept well below a second so a server +/// restart or a short network blip is over before the user notices missing +/// messages. +const RECONNECT_INITIAL_DELAY: Duration = Duration::from_millis(250); +const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(20); +const RECEIVE_TIMEOUT: Duration = Duration::from_secs(45); +const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +const HANDSHAKE_RETRY_DELAY: Duration = Duration::from_secs(1); + pub(crate) static API_EVENTS: LazyLock> = LazyLock::new(|| broadcast::channel(256).0); pub(crate) static API_PERMANENTLY_REJECTED: AtomicBool = AtomicBool::new(false); @@ -33,6 +49,10 @@ pub(crate) struct ApiClient { pub(crate) in_background: AtomicBool, pub(crate) network_available: AtomicBool, pub(crate) is_authenticated: Arc, + /// Wakes the per-connection catch-up loop. Capacity one, sent with + /// `try_send`, so repeated triggers collapse into a single pull instead of + /// queueing one request each. + catch_up_tx: Mutex>>, } impl ApiClient { @@ -50,9 +70,79 @@ impl ApiClient { in_background: AtomicBool::new(in_background), network_available: AtomicBool::new(true), is_authenticated: Arc::new(AtomicBool::new(false)), + catch_up_tx: Mutex::const_new(None), }) } + /// Asks the server to redeliver anything still queued for this user. + /// + /// Coalescing happens in the channel: while a pull is in flight further + /// triggers set the single pending slot, so the loop issues exactly one + /// more request afterwards no matter how many arrived. + pub(crate) async fn request_catch_up(&self) { + if let Some(sender) = self.catch_up_tx.lock().await.as_ref() { + let _ = sender.try_send(()); + } + } + + /// Runs the catch-up loop for one connection. It exits as soon as that + /// connection is replaced or closed, so a disconnected client never + /// reconnects just to poll — reconnection triggers its own catch-up. + fn spawn_catch_up_loop( + self: &Arc, + ws_client: &Arc, + mut receiver: mpsc::Receiver<()>, + ) { + let client = Arc::downgrade(self); + let connection = Arc::downgrade(ws_client); + tokio::spawn(async move { + loop { + let delay = CATCH_UP_INTERVAL + + Duration::from_millis( + rand::random::() % (CATCH_UP_JITTER.as_millis() as u64).max(1), + ); + tokio::select! { + trigger = receiver.recv() => { + if trigger.is_none() { + break; + } + } + () = tokio::time::sleep(delay) => {} + } + + let (Some(client), Some(connection)) = (client.upgrade(), connection.upgrade()) + else { + break; + }; + + // Stop once this task no longer belongs to the live connection. + let is_current = client + .ws_client + .lock() + .await + .as_ref() + .is_some_and(|current| Arc::ptr_eq(current, &connection)); + if !is_current { + break; + } + + if !client.is_authenticated.load(Ordering::Acquire) { + continue; + } + + // Fold any triggers that piled up into the request below. + while receiver.try_recv().is_ok() {} + + let Some(context) = client.context.upgrade() else { + break; + }; + if let Err(error) = crate::api::Server::request_pending_messages(&context).await { + tracing::warn!("mailbox catch-up request failed: {error}"); + } + } + }); + } + pub(crate) async fn set_state(&self, state: ApiConnectionState) { let mut guard = self.state.write().await; if *guard != state { @@ -97,8 +187,17 @@ impl ApiClient { events: self.events.clone(), }; + // `config` replaces the whole `ClientConfig`, so it has to come before + // `receive_timeout`, which only overwrites that one field. let client = WebSocketClient::builder(host) - .receive_timeout(Duration::from_secs(60)) + .config( + ClientConfig::default() + .with_connect_timeout(CONNECT_TIMEOUT) + .with_handshake_retry_delay(HANDSHAKE_RETRY_DELAY) + .with_nodelay(true), + ) + .receive_timeout(RECEIVE_TIMEOUT) + .exponential_backoff(RECONNECT_INITIAL_DELAY, RECONNECT_MAX_DELAY, 2.0) .handshaker(handshaker) .build(); @@ -109,6 +208,10 @@ impl ApiClient { *client_guard = Some(ws_arc.clone()); drop(client_guard); + let (catch_up_tx, catch_up_rx) = mpsc::channel(1); + *self.catch_up_tx.lock().await = Some(catch_up_tx); + self.spawn_catch_up_loop(&ws_arc, catch_up_rx); + // Spawn client runner tokio::spawn({ let c = ws_arc.clone(); @@ -169,6 +272,8 @@ impl ApiClient { pub async fn close(&self) { self.deliberately_closed.store(true, Ordering::Release); self.is_authenticated.store(false, Ordering::Release); + // Dropping the sender ends the catch-up loop for this connection. + *self.catch_up_tx.lock().await = None; let client = self.ws_client.lock().await.take(); if let Some(client) = client { if let Err(error) = client.shutdown_graceful(Duration::from_secs(5)).await { @@ -193,6 +298,9 @@ impl ApiClient { self.set_state(ApiConnectionState::Suspended).await; } else if self.ws_client.lock().await.is_some() { self.set_state(ApiConnectionState::Authenticated).await; + // Returning to the foreground: pick up whatever arrived while the + // socket was suspended. + self.request_catch_up().await; } else if self.network_available.load(Ordering::Acquire) { self.connect().await?; } diff --git a/rust/src/api/server/transport.rs b/rust/src/api/server/transport.rs index 1c800267..0ba9ed8a 100644 --- a/rust/src/api/server/transport.rs +++ b/rust/src/api/server/transport.rs @@ -83,3 +83,19 @@ impl Server { ApiRuntime::request_binary(ctx, Self::handshake_request(value)).await } } + +impl Server { + /// Asks the server to (re)start a mailbox drain. It answers immediately; + /// the messages themselves arrive as unsolicited `PendingMessagesV2` + /// batches, so this must not be treated as a delivery round trip. + pub(crate) async fn request_pending_messages(ctx: &Arc) -> Result<()> { + Self::application( + ctx, + client_to_server::application_data::ApplicationData::RequestPendingMessages( + client_to_server::application_data::RequestPendingMessages {}, + ), + ) + .await + .map(|_| ()) + } +} diff --git a/rust/src/services/notifications.rs b/rust/src/services/notifications.rs index e4bb6a5e..ff665902 100644 --- a/rust/src/services/notifications.rs +++ b/rust/src/services/notifications.rs @@ -409,6 +409,11 @@ pub async fn process_wakeup( if !initial.additions.is_empty() { return Ok(initial); } + // Flutter already owns the socket. Nudge the server to redeliver in + // case the push raced a drain that had already finished. + if let Ok(client) = ApiRuntime::client(&ctx).await { + client.request_catch_up().await; + } let generation = ctx.incoming_generation(); tokio::time::timeout(deadline, ctx.wait_for_incoming_after(generation)) .await diff --git a/rust/tests/api.rs b/rust/tests/api.rs index 9a66a888..251c2627 100644 --- a/rust/tests/api.rs +++ b/rust/tests/api.rs @@ -24,18 +24,11 @@ use rust_lib_twonly::database::app::tables::Group; use rust_lib_twonly::services::contacts::ContactService; use rust_lib_twonly::services::groups::GroupService; use rust_lib_twonly::services::messages::MessageService; -pub(crate) use tester::Tester; +pub(crate) use tester::{init_tracing, Tester}; #[tokio::test] async fn test_connect_to_dev_server() -> anyhow::Result<()> { - let _ = tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .with_ansi(true) - .event_format(rust_lib_twonly::log::ShortEventFormatter::ansi()) - .try_init(); + init_tracing(); let mut tester_a = { let mut tester = Tester::new().await?; diff --git a/rust/tests/api/contacts.rs b/rust/tests/api/contacts.rs index 540bdefd..6a3ea081 100644 --- a/rust/tests/api/contacts.rs +++ b/rust/tests/api/contacts.rs @@ -1,4 +1,4 @@ -use super::Tester; +use super::{init_tracing, Tester}; use rust_lib_twonly::api::Server; use rust_lib_twonly::bridge::api::ApiConnectionState; use rust_lib_twonly::database::app::tables::Group; @@ -15,6 +15,7 @@ async fn create_authenticated_tester() -> anyhow::Result { #[tokio::test] async fn test_contact_cross_request_auto_accept() -> anyhow::Result<()> { + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; @@ -55,6 +56,7 @@ async fn test_contact_cross_request_auto_accept() -> anyhow::Result<()> { #[tokio::test] async fn test_unknown_sender_auto_contact_discovery() -> anyhow::Result<()> { + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; @@ -85,6 +87,7 @@ async fn test_unknown_sender_auto_contact_discovery() -> anyhow::Result<()> { #[tokio::test] async fn test_check_for_deleted_usernames() -> anyhow::Result<()> { + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; diff --git a/rust/tests/api/group_resilience.rs b/rust/tests/api/group_resilience.rs index c5d4464f..d42de826 100644 --- a/rust/tests/api/group_resilience.rs +++ b/rust/tests/api/group_resilience.rs @@ -1,4 +1,4 @@ -use super::Tester; +use super::{init_tracing, Tester}; use rust_lib_twonly::bridge::api::ApiConnectionState; use rust_lib_twonly::services::contacts::ContactService; use rust_lib_twonly::services::groups::GroupService; @@ -14,14 +14,7 @@ async fn create_authenticated_tester() -> anyhow::Result { #[tokio::test] async fn test_group_membership_error_healing() -> anyhow::Result<()> { - let _ = tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .with_ansi(true) - .event_format(rust_lib_twonly::log::ShortEventFormatter::ansi()) - .try_init(); + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; @@ -98,6 +91,7 @@ async fn test_group_membership_error_healing() -> anyhow::Result<()> { #[tokio::test] async fn test_add_hidden_contact() -> anyhow::Result<()> { + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; @@ -125,6 +119,7 @@ async fn test_add_hidden_contact() -> anyhow::Result<()> { #[tokio::test] async fn test_admin_and_non_admin_leave_group() -> anyhow::Result<()> { + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; let tester_c = create_authenticated_tester().await?; diff --git a/rust/tests/api/media.rs b/rust/tests/api/media.rs index a97848ba..274117b7 100644 --- a/rust/tests/api/media.rs +++ b/rust/tests/api/media.rs @@ -1,4 +1,4 @@ -use super::Tester; +use super::{init_tracing, Tester}; use prost::Message as _; use rust_lib_twonly::api::messages::outgoing::send_c2c_message_to_contact; use rust_lib_twonly::api::proto::client::{self as proto, encrypted_content}; @@ -18,14 +18,7 @@ async fn create_authenticated_tester() -> anyhow::Result { #[tokio::test] async fn test_media_lifecycle_actions_and_reupload() -> anyhow::Result<()> { - let _ = tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .with_ansi(true) - .event_format(rust_lib_twonly::log::ShortEventFormatter::ansi()) - .try_init(); + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; diff --git a/rust/tests/api/notifications.rs b/rust/tests/api/notifications.rs index a16c7fee..7d6896bb 100644 --- a/rust/tests/api/notifications.rs +++ b/rust/tests/api/notifications.rs @@ -7,7 +7,7 @@ //! hand, so classification, deduplication, claiming and clearing are all //! exercised end to end. -use crate::Tester; +use crate::{init_tracing, Tester}; use rust_lib_twonly::bridge::api::ApiConnectionState; use rust_lib_twonly::database::app::tables::Group; use rust_lib_twonly::services::contacts::ContactService; @@ -24,14 +24,7 @@ async fn ready_tester() -> anyhow::Result { #[tokio::test] async fn test_notification_outbox_end_to_end() -> anyhow::Result<()> { - let _ = tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .with_ansi(true) - .event_format(rust_lib_twonly::log::ShortEventFormatter::ansi()) - .try_init(); + init_tracing(); let mut tester_a = ready_tester().await?; let tester_b = ready_tester().await?; diff --git a/rust/tests/api/recovery.rs b/rust/tests/api/recovery.rs index 63b4ba3d..91f92115 100644 --- a/rust/tests/api/recovery.rs +++ b/rust/tests/api/recovery.rs @@ -1,4 +1,4 @@ -use super::Tester; +use super::{init_tracing, Tester}; use prost::Message as _; use rust_lib_twonly::api::messages::incoming::recovery::perform_heartbeat; use rust_lib_twonly::api::messages::outgoing::send_c2c_message_to_contact; @@ -16,14 +16,7 @@ async fn create_authenticated_tester() -> anyhow::Result { #[tokio::test] async fn test_passwordless_recovery_share_heartbeat_and_delete() -> anyhow::Result<()> { - let _ = tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .with_ansi(true) - .event_format(rust_lib_twonly::log::ShortEventFormatter::ansi()) - .try_init(); + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; diff --git a/rust/tests/api/server_api.rs b/rust/tests/api/server_api.rs index e449677b..8bd7a237 100644 --- a/rust/tests/api/server_api.rs +++ b/rust/tests/api/server_api.rs @@ -1,4 +1,4 @@ -use super::Tester; +use super::{init_tracing, Tester}; use rust_lib_twonly::api::Server; use rust_lib_twonly::bridge::api::{ApiConnectionState, ServerResult}; @@ -12,6 +12,7 @@ async fn create_authenticated_tester() -> anyhow::Result { #[tokio::test] async fn test_server_account_and_user_endpoints() -> anyhow::Result<()> { + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; diff --git a/rust/tests/api/session_recovery.rs b/rust/tests/api/session_recovery.rs index 016dd9ae..51cef751 100644 --- a/rust/tests/api/session_recovery.rs +++ b/rust/tests/api/session_recovery.rs @@ -1,4 +1,4 @@ -use super::Tester; +use super::{init_tracing, Tester}; use rust_lib_twonly::bridge::api::ApiConnectionState; use rust_lib_twonly::database::app::tables::Group; use rust_lib_twonly::services::contacts::ContactService; @@ -14,6 +14,7 @@ async fn create_authenticated_tester() -> anyhow::Result { #[tokio::test] async fn test_signal_session_auto_recovery_on_missing_session() -> anyhow::Result<()> { + init_tracing(); let tester_a = create_authenticated_tester().await?; let tester_b = create_authenticated_tester().await?; diff --git a/rust/tests/api/tester.rs b/rust/tests/api/tester.rs index 29081b3d..4627009a 100644 --- a/rust/tests/api/tester.rs +++ b/rust/tests/api/tester.rs @@ -9,6 +9,39 @@ use std::sync::Arc; use tempfile::TempDir; use tokio::time::{sleep, Duration}; +/// Installs the tracing subscriber shared by every integration test. +/// +/// Logging is entirely opt-in through the environment: without `RUST_LOG` no +/// subscriber is installed at all, so a plain `cargo test` run stays quiet. +/// `RUST_LOG` selects the filter, and `NO_COLOR` (or `TWONLY_LOG_ANSI=0`) +/// switches the formatter from ansi to plain for CI logs. +/// +/// Calling it more than once is a no-op, so every test can call it. +pub(crate) fn init_tracing() { + let Ok(filter) = tracing_subscriber::EnvFilter::try_from_default_env() else { + return; + }; + + let ansi = match std::env::var("TWONLY_LOG_ANSI") { + Ok(value) => !matches!(value.trim(), "" | "0" | "false" | "no"), + Err(_) => std::env::var_os("NO_COLOR").is_none_or(|value| value.is_empty()), + }; + + let subscriber = tracing_subscriber::fmt() + .with_env_filter(filter) + .with_ansi(ansi); + + let _ = if ansi { + subscriber + .event_format(rust_lib_twonly::log::ShortEventFormatter::ansi()) + .try_init() + } else { + subscriber + .event_format(rust_lib_twonly::log::ShortEventFormatter::plain()) + .try_init() + }; +} + pub(crate) struct Tester { pub context: Arc, pub username: String, diff --git a/rust/tests/api/user_discovery.rs b/rust/tests/api/user_discovery.rs index 5b0aecdb..efe9b0bf 100644 --- a/rust/tests/api/user_discovery.rs +++ b/rust/tests/api/user_discovery.rs @@ -5,7 +5,7 @@ use rust_lib_twonly::services::contacts::ContactService; use rust_lib_twonly::services::messages::MessageService; use tokio::time::sleep; -use super::Tester; +use super::{init_tracing, Tester}; async fn create_tester() -> anyhow::Result { let mut tester = Tester::new().await?; @@ -96,6 +96,7 @@ async fn wait_for_discovery( #[tokio::test] async fn user_discovery_reconstructs_an_unknown_user_from_three_contacts() -> anyhow::Result<()> { + init_tracing(); let observer = create_tester().await?; let relay_a = create_tester().await?; let relay_b = create_tester().await?;