From 0f3d67ad7f28ce2af87f4fed57fc17bd521cd29e Mon Sep 17 00:00:00 2001 From: Andrew Plaza Date: Wed, 3 Jun 2026 18:31:41 -0400 Subject: [PATCH] fix(xmtp_mls): forward db-disconnect signal through SyncSummary and sibling error variants MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A dropped DB pool (PoolNeedsConnection) surfacing during a sync was misclassified as not-a-disconnect: SyncSummary had no NeedsDbReconnect impl, so any disconnect landing in its publish/post-commit/other/per-message errors got wrapped into GroupError::Sync / SyncFailedToWait / DeviceSyncError::Sync / CommitLogError::SyncError and hit `_ => false`. The worker supervisor never dropped the pool to reconnect — it retried forever on a dead pool (GroupError::SyncFailedToWait is also is_retryable, a true hot-loop). Fix — forward the signal everywhere it was dropped: - SqlKeyStoreError gains db_needs_connection() (mirrors StorageError / ConnectionError) so the OpenMLS key-store disconnect check lives in one place. - SyncSummary: NeedsDbReconnect scans all four error sources, including the per-message process.errored failures that is_retryable deliberately skips. - GroupMessageProcessingError: NeedsDbReconnect forwards the directly-wrapped DB variants and the OpenMLS state-I/O wrappers (process/merge), which carry a drop as StorageError(SqlKeyStoreError::Connection(_)). - GroupError: Sync / SyncFailedToWait / SqlKeyStore forwarded. - TaskWorkerError: Group / LoadGroup / DeviceSync forward the full classification instead of structurally matching only ::Storage. - CommitLogError: SyncError / KeystoreError forwarded. - DeviceSyncError: Sync forwarded. Every new arm bottoms out at db_needs_connection(), which matches only PoolNeedsConnection / Pool(_) — benign connection errors (e.g. DisconnectInTransaction) stay false, pinned by negative test cases. Rebased on main (post #3744 TaskRunner rework; tasks.rs merged with the new LoadGroup variant). Adds 3 disconnect-propagation tests. Co-Authored-By: Claude Fable 5 --- crates/xmtp_db/src/sql_key_store.rs | 8 ++ crates/xmtp_mls/src/groups/commit_log.rs | 6 +- crates/xmtp_mls/src/groups/error.rs | 5 + crates/xmtp_mls/src/groups/mls_sync.rs | 27 ++++ crates/xmtp_mls/src/groups/summary.rs | 18 +++ crates/xmtp_mls/src/worker.rs | 130 ++++++++++++++++++ crates/xmtp_mls/src/worker/device_sync/mod.rs | 2 + crates/xmtp_mls/src/worker/tasks.rs | 10 +- 8 files changed, 198 insertions(+), 8 deletions(-) diff --git a/crates/xmtp_db/src/sql_key_store.rs b/crates/xmtp_db/src/sql_key_store.rs index 79076d3a97..3584861199 100644 --- a/crates/xmtp_db/src/sql_key_store.rs +++ b/crates/xmtp_db/src/sql_key_store.rs @@ -311,6 +311,14 @@ pub enum SqlKeyStoreError { Connection(#[from] crate::ConnectionError), } +impl SqlKeyStoreError { + /// True when the pool can't currently hand out a connection. Mirrors + /// [`StorageError::db_needs_connection`](crate::StorageError::db_needs_connection). + pub fn db_needs_connection(&self) -> bool { + matches!(self, Self::Connection(c) if c.db_needs_connection()) + } +} + impl RetryableError for SqlKeyStoreError { fn is_retryable(&self) -> bool { use SqlKeyStoreError::*; diff --git a/crates/xmtp_mls/src/groups/commit_log.rs b/crates/xmtp_mls/src/groups/commit_log.rs index 055b0ce59e..e1381e8a5f 100644 --- a/crates/xmtp_mls/src/groups/commit_log.rs +++ b/crates/xmtp_mls/src/groups/commit_log.rs @@ -147,16 +147,18 @@ impl NeedsDbReconnect for CommitLogError { // A readd send failure wraps an inner CommitLogError; forward. Self::FailedToSendReadd { source, .. } => source.needs_db_reconnect(), Self::FailedReadds { errors } => errors.iter().any(|e| e.needs_db_reconnect()), + // A dropped pool can be wrapped inside a SyncSummary; forward. + Self::SyncError(s) => s.needs_db_reconnect(), + // OpenMLS key-store ops surface a disconnect as a Connection error. + Self::KeystoreError(e) => e.db_needs_connection(), // Remaining variants can't carry a dropped-pool signal. Self::Diesel(_) | Self::Api(_) | Self::Prost(_) - | Self::KeystoreError(_) | Self::CryptoError(_) | Self::TryFromSliceError(_) | Self::Conversion(_) | Self::GroupReaddValidationError(_) - | Self::SyncError(_) | Self::MissingLatestCommitSequenceId { .. } => false, } } diff --git a/crates/xmtp_mls/src/groups/error.rs b/crates/xmtp_mls/src/groups/error.rs index 371a08c5c3..80e43bf774 100644 --- a/crates/xmtp_mls/src/groups/error.rs +++ b/crates/xmtp_mls/src/groups/error.rs @@ -653,6 +653,11 @@ impl crate::worker::NeedsDbReconnect for GroupError { Self::MlsStore(s) => s.needs_db_reconnect(), Self::Identity(i) => i.needs_db_reconnect(), Self::DeviceSync(d) => d.needs_db_reconnect(), + // A dropped pool can be wrapped inside a SyncSummary's + // publish/post-commit/other/per-message errors; forward it. + Self::Sync(s) | Self::SyncFailedToWait(s) => s.needs_db_reconnect(), + // OpenMLS key-store ops surface a disconnect as a Connection error. + Self::SqlKeyStore(e) => e.db_needs_connection(), _ => false, } } diff --git a/crates/xmtp_mls/src/groups/mls_sync.rs b/crates/xmtp_mls/src/groups/mls_sync.rs index 3ea0f10fec..7505ec855e 100644 --- a/crates/xmtp_mls/src/groups/mls_sync.rs +++ b/crates/xmtp_mls/src/groups/mls_sync.rs @@ -273,6 +273,33 @@ impl RetryableError for GroupMessageProcessingError { } } +impl crate::worker::NeedsDbReconnect for GroupMessageProcessingError { + /// Forwards a dropped-pool signal from storage-bearing variants so a + /// per-message disconnect (landing in `SyncSummary::process.errored`) isn't lost. + fn needs_db_reconnect(&self) -> bool { + use super::app_data::ProcessMessageWithAppDataError; + match self { + Self::Storage(s) => s.db_needs_connection(), + Self::Db(c) => c.db_needs_connection(), + Self::Identity(i) => i.needs_db_reconnect(), + Self::Client(c) => c.db_needs_connection(), + // OpenMLS state-I/O carries a pool drop as + // `StorageError(SqlKeyStoreError::Connection(_))`; forward it. + Self::ClearPendingCommit(e) => e.db_needs_connection(), + Self::OpenMlsProcessMessage(ProcessMessageError::StorageError(e)) => { + e.db_needs_connection() + } + Self::OpenMlsProcessMessageWithAppData(ProcessMessageWithAppDataError::OpenMls( + ProcessMessageError::StorageError(e), + )) => e.db_needs_connection(), + Self::MergeStagedCommit(openmls::group::MergeCommitError::StorageError(e)) => { + e.db_needs_connection() + } + _ => false, + } + } +} + impl GroupMessageProcessingError { pub(crate) fn commit_result(&self) -> CommitResult { use super::app_data::ProcessMessageWithAppDataError; diff --git a/crates/xmtp_mls/src/groups/summary.rs b/crates/xmtp_mls/src/groups/summary.rs index 5cc17a4568..565e183a14 100644 --- a/crates/xmtp_mls/src/groups/summary.rs +++ b/crates/xmtp_mls/src/groups/summary.rs @@ -30,6 +30,24 @@ impl RetryableError for SyncSummary { } } +impl crate::worker::NeedsDbReconnect for SyncSummary { + /// Scans every error source (incl. per-message `process.errored`, which + /// `is_retryable` skips) so a dropped pool wrapped in a summary still surfaces. + fn needs_db_reconnect(&self) -> bool { + self.publish_errors.iter().any(|e| e.needs_db_reconnect()) + || self + .post_commit_errors + .iter() + .any(|e| e.needs_db_reconnect()) + || self.other.as_ref().is_some_and(|e| e.needs_db_reconnect()) + || self + .process + .errored + .iter() + .any(|(_, e)| e.needs_db_reconnect()) + } +} + impl SyncSummary { /// synced a single message successfully pub fn single(msg: MessageIdentifier) -> Self { diff --git a/crates/xmtp_mls/src/worker.rs b/crates/xmtp_mls/src/worker.rs index e02da70c7c..32b0c8daef 100644 --- a/crates/xmtp_mls/src/worker.rs +++ b/crates/xmtp_mls/src/worker.rs @@ -446,6 +446,10 @@ mod disconnect_propagation_tests { !CommitLogError::Connection(ConnectionError::DisconnectInTransaction) .needs_db_reconnect() ); + // A dropped pool wrapped in a SyncSummary (was `SyncError(_) => false`). + let mut summary = crate::groups::summary::SyncSummary::default(); + summary.add_other(GroupError::Db(disconnect_connection())); + assert!(CommitLogError::SyncError(summary).needs_db_reconnect()); } #[xmtp_common::test] @@ -463,6 +467,10 @@ mod disconnect_propagation_tests { DeviceSyncError::Subscribe(SubscribeError::Db(disconnect_connection())) .needs_db_reconnect() ); + // A dropped pool wrapped in a SyncSummary (was caught by `_ => false`). + let mut summary = crate::groups::summary::SyncSummary::default(); + summary.add_other(GroupError::Db(disconnect_connection())); + assert!(DeviceSyncError::Sync(Box::new(summary)).needs_db_reconnect()); assert!(!DeviceSyncError::Storage(benign_storage()).needs_db_reconnect()); assert!(!DeviceSyncError::InvalidPayload.needs_db_reconnect()); } @@ -483,6 +491,128 @@ mod disconnect_propagation_tests { ); assert!(!KeyPackagesCleanerError::Storage(benign_storage()).needs_db_reconnect()); } + + // SyncSummary previously had no NeedsDbReconnect impl, so a dropped pool in + // any of its four error sources hot-looped the worker via `_ => false`. + #[xmtp_common::test] + fn sync_summary_forwards_disconnect() { + use crate::groups::mls_sync::GroupMessageProcessingError; + use crate::groups::summary::{ProcessSummary, SyncSummary}; + use xmtp_proto::types::Cursor; + + // 1. disconnect in publish_errors + let mut s = SyncSummary::default(); + s.add_publish_err(GroupError::Storage(disconnect_storage())); + assert!(s.needs_db_reconnect(), "publish_errors disconnect"); + + // 2. disconnect in post_commit_errors + let mut s = SyncSummary::default(); + s.add_post_commit_err(GroupError::Db(disconnect_connection())); + assert!(s.needs_db_reconnect(), "post_commit_errors disconnect"); + + // 3. disconnect in `other` + let mut s = SyncSummary::default(); + s.add_other(GroupError::Storage(disconnect_storage())); + assert!(s.needs_db_reconnect(), "other disconnect"); + + // 4. disconnect in a per-message processing error (process.errored) — + // the source is_retryable() deliberately skips. + let mut process = ProcessSummary::default(); + process.errored.push(( + Cursor::new(1, 1u32), + GroupMessageProcessingError::Db(disconnect_connection()), + )); + let mut s = SyncSummary::default(); + s.add_process(process); + assert!(s.needs_db_reconnect(), "process.errored disconnect"); + + // A benign sync error must NOT trip the contract (else the worker tears + // down the pool needlessly on an ordinary per-message failure). + let mut s = SyncSummary::default(); + s.add_publish_err(GroupError::Storage(benign_storage())); + assert!(!s.needs_db_reconnect(), "benign publish error"); + + // And the forwarding must survive the GroupError wrappers the worker + // actually inspects. + let mut s = SyncSummary::default(); + s.add_publish_err(GroupError::Db(disconnect_connection())); + assert!( + GroupError::Sync(Box::new(s)).needs_db_reconnect(), + "GroupError::Sync" + ); + + let mut s = SyncSummary::default(); + s.add_other(GroupError::Storage(disconnect_storage())); + assert!( + GroupError::SyncFailedToWait(Box::new(s)).needs_db_reconnect(), + "GroupError::SyncFailedToWait" + ); + } + + // OpenMLS state-I/O wrappers carry a pool drop as + // `StorageError(SqlKeyStoreError::Connection(_))`; previously `_ => false`. + #[xmtp_common::test] + fn group_message_processing_error_forwards_openmls_disconnect() { + use crate::groups::mls_sync::GroupMessageProcessingError; + use openmls::group::MergeCommitError; + use openmls::prelude::ProcessMessageError; + use xmtp_db::sql_key_store::SqlKeyStoreError; + + let disconnect_sql = || SqlKeyStoreError::Connection(disconnect_connection()); + let benign_sql = || SqlKeyStoreError::Connection(ConnectionError::DisconnectInTransaction); + + // OpenMlsProcessMessage(StorageError(Connection(PoolNeedsConnection))) + assert!( + GroupMessageProcessingError::OpenMlsProcessMessage(ProcessMessageError::StorageError( + disconnect_sql() + )) + .needs_db_reconnect() + ); + // MergeStagedCommit(StorageError(Connection(PoolNeedsConnection))) + assert!( + GroupMessageProcessingError::MergeStagedCommit(MergeCommitError::StorageError( + disconnect_sql() + )) + .needs_db_reconnect() + ); + // A non-pool-drop connection error inside the same wrapper must NOT trip it. + assert!( + !GroupMessageProcessingError::OpenMlsProcessMessage(ProcessMessageError::StorageError( + benign_sql() + )) + .needs_db_reconnect() + ); + // The directly-wrapped paths still work. + assert!(GroupMessageProcessingError::Db(disconnect_connection()).needs_db_reconnect()); + assert!(!GroupMessageProcessingError::Storage(benign_storage()).needs_db_reconnect()); + } + + // A disconnect can reach the task worker as Group(Db)/DeviceSync(Db), not + // only as the structurally-matched ::Storage; previously those were `false`. + #[xmtp_common::test] + fn task_worker_error_forwards_disconnect() { + use crate::worker::tasks::TaskWorkerError; + assert!( + TaskWorkerError::Group(GroupError::Db(disconnect_connection())).needs_db_reconnect(), + "Group(Db) disconnect" + ); + assert!( + TaskWorkerError::Group(GroupError::MlsStore(MlsStoreError::Connection( + disconnect_connection() + ))) + .needs_db_reconnect(), + "Group(MlsStore) disconnect" + ); + assert!( + TaskWorkerError::DeviceSync(DeviceSyncError::Db(disconnect_connection())) + .needs_db_reconnect(), + "DeviceSync(Db) disconnect" + ); + assert!( + !TaskWorkerError::Group(GroupError::InvalidGroupMembership).needs_db_reconnect(), + "benign group error" + ); + } } #[cfg(test)] diff --git a/crates/xmtp_mls/src/worker/device_sync/mod.rs b/crates/xmtp_mls/src/worker/device_sync/mod.rs index e71408e324..3cc767a007 100644 --- a/crates/xmtp_mls/src/worker/device_sync/mod.rs +++ b/crates/xmtp_mls/src/worker/device_sync/mod.rs @@ -212,6 +212,8 @@ impl NeedsDbReconnect for DeviceSyncError { Self::Group(e) => e.needs_db_reconnect(), Self::MlsStore(e) => e.needs_db_reconnect(), Self::Subscribe(e) => e.needs_db_reconnect(), + // A dropped pool can be wrapped inside a SyncSummary; forward. + Self::Sync(s) => s.needs_db_reconnect(), _ => false, } } diff --git a/crates/xmtp_mls/src/worker/tasks.rs b/crates/xmtp_mls/src/worker/tasks.rs index a01636d603..74cc33d3ce 100644 --- a/crates/xmtp_mls/src/worker/tasks.rs +++ b/crates/xmtp_mls/src/worker/tasks.rs @@ -45,14 +45,12 @@ pub enum TaskWorkerError { impl NeedsDbReconnect for TaskWorkerError { fn needs_db_reconnect(&self) -> bool { match self { - TaskWorkerError::Storage(s) - | TaskWorkerError::DeviceSync(DeviceSyncError::Storage(s)) => s.db_needs_connection(), + TaskWorkerError::Storage(s) => s.db_needs_connection(), + // Forward the full classification: a dropped pool can reach these as + // `Db`/`MlsStore`/`Sync`/... too, not only as `Storage`. TaskWorkerError::LoadGroup(e) => e.needs_db_reconnect(), - // Forward through GroupError's own classifier so a dropped pool hiding - // in a `Db`/`MlsStore` (not just `Storage`) variant still restarts the - // worker instead of being retried on a dead connection. TaskWorkerError::Group(e) => e.needs_db_reconnect(), - TaskWorkerError::DeviceSync(_) => false, + TaskWorkerError::DeviceSync(e) => e.needs_db_reconnect(), TaskWorkerError::InvalidTaskData { .. } => false, TaskWorkerError::InvalidHash { .. } => false, TaskWorkerError::ReceiverLocked => false,