fix(xmtp_mls): forward db-disconnect signal through SyncSummary and sibling error variants

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 <noreply@anthropic.com>

Author: insipx

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::*;

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,
         }
     }

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,
         }
     }

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;

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 {

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)]

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,
         }
     }

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,