diff --git a/rust/src/api/messages/incoming/messages.rs b/rust/src/api/messages/incoming/messages.rs index c50fd1bc..cd3662ff 100644 --- a/rust/src/api/messages/incoming/messages.rs +++ b/rust/src/api/messages/incoming/messages.rs @@ -26,6 +26,39 @@ use std::sync::LazyLock; static ALREADY_QUEUED_RECEIPTS: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); +/// How long a receipt the server has already taken is kept out of the queue. +#[cfg(not(debug_assertions))] +const RESEND_SUPPRESSION: std::time::Duration = std::time::Duration::from_secs(120); + +/// Whether the server accepted this receipt inside the suppression window. +/// +/// Every inbound message marks a peer's whole outbox for retry, so a chat busy +/// enough to keep flushing would otherwise re-send the same envelope on every +/// message that arrives. +#[cfg(not(debug_assertions))] +fn recently_sent(receipt_id: &str) -> bool { + let mut sent = ALREADY_QUEUED_RECEIPTS + .lock() + .unwrap_or_else(PoisonError::into_inner); + sent.retain(|_, time| time.elapsed() < RESEND_SUPPRESSION); + sent.contains_key(receipt_id) +} + +/// Starts the suppression window, once the server has actually taken the +/// envelope. +/// +/// Recorded on success rather than on the attempt: a send that fails has not +/// reached anyone, and locking it out for two minutes turned a transient +/// failure -- one dropped socket, one timed-out request -- into an outbox that +/// stayed stuck long after the connection came back. +#[cfg(not(debug_assertions))] +fn note_sent(receipt_id: &str) { + ALREADY_QUEUED_RECEIPTS + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(receipt_id.to_owned(), std::time::Instant::now()); +} + /// Coalesces overlapping flushes of one account's receipt queue. /// /// Every committed inbound message asks for a flush, and a flush scans the @@ -502,18 +535,11 @@ pub(crate) async fn prepare_queued_receipt( pub(crate) async fn send_queued_receipt(ctx: &Arc, receipt_id: &str) -> Result<()> { #[cfg(not(debug_assertions))] - { - let mut locks = ALREADY_QUEUED_RECEIPTS.lock().unwrap(); - if let Some(time) = locks.get(receipt_id) { - if time.elapsed() < std::time::Duration::from_secs(120) { - tracing::info!( - "Blocking queued receipt, as it was already sent within the last 120 seconds" - ); - return Ok(()); - } - } - locks.insert(receipt_id.to_owned(), std::time::Instant::now()); - locks.retain(|_, time| time.elapsed() < std::time::Duration::from_secs(120)); + if recently_sent(receipt_id) { + tracing::info!( + "Blocking queued receipt, as it was already sent within the last 120 seconds" + ); + return Ok(()); } let app_db = ctx.app_db.read().await.clone(); @@ -599,6 +625,9 @@ pub(crate) async fn send_queued_receipt(ctx: &Arc, receipt_id: &str) -> } } + #[cfg(not(debug_assertions))] + note_sent(receipt_id); + let mut t = app_db.pool.begin().await?; if let Some(message_id) = receipt.message_id { sqlx::query!( diff --git a/rust/src/api/runtime/client.rs b/rust/src/api/runtime/client.rs index 510954c8..379daa29 100644 --- a/rust/src/api/runtime/client.rs +++ b/rust/src/api/runtime/client.rs @@ -160,6 +160,27 @@ impl ApiClient { if let Err(error) = crate::api::Server::request_pending_messages(&context).await { tracing::warn!("mailbox catch-up request failed: {error}"); } + + // The outbox needs the same safety net as the mailbox. Every + // other flush is tied to an event -- authenticating, a message + // arriving, a background job -- so on a connection that stays + // up while nothing comes in, a send that failed once had + // nothing left to retry it. A one-way conversation is exactly + // the case with no inbound message to trigger the sweep. + // + // Not in a notification worker: its whole budget belongs to the + // alert, and it must not spend the one app-database connection + // on the outbox while a drain is waiting for it. + if !context.is_notification_runtime() { + if let Err(error) = + crate::api::messages::incoming::messages::retransmit_queued_receipts( + &context, + ) + .await + { + tracing::warn!("outbox catch-up flush failed: {error}"); + } + } } }); } diff --git a/rust/src/api/runtime/request.rs b/rust/src/api/runtime/request.rs index eaf6228e..31ed93f2 100644 --- a/rust/src/api/runtime/request.rs +++ b/rust/src/api/runtime/request.rs @@ -277,17 +277,20 @@ impl ApiClient { if let Some(contact_id) = contact_id { let context = self.context.upgrade().ok_or(TwonlyError::Initialization)?; let database = context.app_db.read().await.clone(); - let mut transaction = database.pool.begin().await?; + // Only the mark, never the queue. `account_deleted` is a latch + // any contact-scoped request can set, and this one may have + // been about something else entirely -- or the peer may since + // have re-registered. `send_queued_receipt` asks the server + // again before it believes the latch, and drops only what the + // server positively denies, so emptying the outbox here threw + // away the very messages that path exists to save: unsent, + // unacknowledged and unlogged. sqlx::query!( "UPDATE contacts SET account_deleted = 1 WHERE user_id = ?", contact_id ) - .execute(&mut *transaction) + .execute(&database.pool) .await?; - sqlx::query!("DELETE FROM receipts WHERE contact_id = ?", contact_id) - .execute(&mut *transaction) - .await?; - transaction.commit().await?; } } Ok(())