diff --git a/crates/xmtp_mls/src/groups/mls_sync.rs b/crates/xmtp_mls/src/groups/mls_sync.rs index 3ea0f10fec..073242171a 100644 --- a/crates/xmtp_mls/src/groups/mls_sync.rs +++ b/crates/xmtp_mls/src/groups/mls_sync.rs @@ -1075,11 +1075,63 @@ where if message_expire_at_ns.is_some() { *disappearing_stored = true; } - self.process_own_leave_request_message(mls_group, storage, &id); - self.process_own_delete_message(storage, &id); + // Best-effort follow-ups on our own just-published message. The message + // is already published and its delivery status persisted; these only + // update local state (leave / delete). A retryable storage failure must + // still roll the sync back and retry, matching the rest of + // process_message_inner; a non-retryable failure is logged and swallowed + // because re-publishing won't help and the local reaction is genuinely + // best-effort here. + self.resolve_own_reaction( + self.process_own_leave_request_message(mls_group, storage, &id), + intent, + &id, + "leave request", + )?; + self.resolve_own_reaction( + self.process_own_delete_message(storage, &id), + intent, + &id, + "delete", + )?; Ok(Some(id)) } + /// Resolve the result of a best-effort follow-up reaction on our own + /// just-published message. + /// + /// Retryable failures become an [`IntentResolutionError`] so the caller + /// rolls the sync transaction back and retries (the message was already + /// published, so the optimistic row is untouched on rollback). Non-retryable + /// failures are logged and swallowed: the publish already succeeded and + /// re-running won't help, so we don't wedge the intent. + fn resolve_own_reaction( + &self, + result: Result<(), GroupMessageProcessingError>, + intent: &StoredGroupIntent, + message_id: &[u8], + reaction: &str, + ) -> Result<(), IntentResolutionError> { + match result { + Ok(()) => Ok(()), + Err(err) if err.is_retryable() => Err(IntentResolutionError { + processing_error: err, + // Unused on the retryable path: the caller returns the + // processing_error directly (rollback + retry) rather than + // transitioning the intent to this state. + next_intent_state: intent.state, + }), + Err(err) => { + tracing::warn!( + group_id = %self.group_id, + message_id = hex::encode(message_id), + "non-retryable error processing own {reaction} reaction, skipping: {err}" + ); + Ok(()) + } + } + } + #[tracing::instrument(level = "trace", skip(mls_group, envelope))] async fn validate_and_process_external_message( &self, @@ -1596,46 +1648,58 @@ where mls_group: &OpenMlsGroup, storage: &impl XmtpMlsStorageProvider, message_id: &[u8], - ) { - if let Ok(Some(message)) = storage.db().get_group_message(message_id) - && message.content_type == ContentType::LeaveRequest - { - match self.process_leave_request_message(mls_group, storage, &message, None) { - Ok(()) => { - debug!("Successfully processed leave request message"); - } - Err(e) => { - debug!("Failed to process leave request message: {}", e); - } - } + ) -> Result<(), GroupMessageProcessingError> { + #[cfg(any(test, feature = "test-utils"))] + crate::utils::test_mocks_helpers::maybe_mock_own_reaction_retryable_error()?; + + // Propagate read failures (`?`) instead of dropping them: a retryable + // storage error here must roll the sync back and retry. A missing row is + // still a benign no-op. + let Some(message) = storage.db().get_group_message(message_id)? else { + return Ok(()); + }; + // Not a leave request: correct early return, not an error. + if message.content_type != ContentType::LeaveRequest { + return Ok(()); } + + self.process_leave_request_message(mls_group, storage, &message, None) } - fn process_own_delete_message(&self, storage: &impl XmtpMlsStorageProvider, message_id: &[u8]) { + fn process_own_delete_message( + &self, + storage: &impl XmtpMlsStorageProvider, + message_id: &[u8], + ) -> Result<(), GroupMessageProcessingError> { + #[cfg(any(test, feature = "test-utils"))] + crate::utils::test_mocks_helpers::maybe_mock_own_reaction_retryable_error()?; + let db = storage.db(); - let Ok(Some(message)) = db.get_group_message(message_id) else { - return; + // Propagate read failures (`?`); a missing row remains a benign no-op, + // handled by the `else` arms below. + let Some(message) = db.get_group_message(message_id)? else { + return Ok(()); }; if message.content_type != ContentType::DeleteMessage { - return; + return Ok(()); } - let Ok(Some(deletion)) = db.get_message_deletion(message_id) else { + let Some(deletion) = db.get_message_deletion(message_id)? else { tracing::warn!( message_id = hex::encode(message_id), "Deletion record not found for own delete message" ); - return; + return Ok(()); }; - let Ok(Some(original_msg)) = db.get_group_message(&deletion.deleted_message_id) else { + let Some(original_msg) = db.get_group_message(&deletion.deleted_message_id)? else { tracing::debug!( deleted_message_id = hex::encode(&deletion.deleted_message_id), "Original message not found for deletion event (may be out-of-order)" ); - return; + return Ok(()); }; let _ = self @@ -1644,6 +1708,8 @@ where .send(crate::subscriptions::LocalEvents::MsgsDeleted(vec![ original_msg, ])); + + Ok(()) } fn process_leave_request_message( @@ -4893,6 +4959,109 @@ pub(crate) mod tests { ); } + /// The self-publish follow-up reaction handlers must *surface* a retryable + /// storage failure (so the sync transaction rolls back and retries) instead + /// of silently swallowing it. Regression guard for the dropped-error paths + /// in `process_own_leave_request_message` / `process_own_delete_message`. + /// + /// Not on wasm: the fault injection uses an env-var flag, and wasm targets + /// cannot set environment variables (same gating as the other + /// `test_mocks_helpers` based tests). + #[cfg(not(target_arch = "wasm32"))] + #[xmtp_common::test(flavor = "current_thread", unwrap_try = true)] + async fn test_own_reaction_handlers_surface_retryable_errors() { + use crate::tester; + use crate::utils::test_mocks_helpers::set_test_mode_own_reaction_retryable_error; + + tester!(alix); + let alix_group = alix.create_group(None, None)?; + let storage = alix.context.mls_storage(); + let message_id = vec![1, 2, 3]; + + // Inject a transient (retryable) storage error into both handlers. + set_test_mode_own_reaction_retryable_error(true); + + let leave_result = alix_group.load_mls_group_with_lock(storage, |mls_group| { + Ok::<_, crate::groups::GroupError>(alix_group.process_own_leave_request_message( + &mls_group, + storage, + &message_id, + )) + })?; + assert!( + matches!(&leave_result, Err(e) if e.is_retryable()), + "own leave-request handler should surface the retryable error, got {leave_result:?}" + ); + + let delete_result = alix_group.process_own_delete_message(storage, &message_id); + assert!( + matches!(&delete_result, Err(e) if e.is_retryable()), + "own delete handler should surface the retryable error, got {delete_result:?}" + ); + + // With the fault cleared, a message id with no stored row is a benign + // no-op (Ok), not an error — the missing-row guards still hold. + set_test_mode_own_reaction_retryable_error(false); + + let leave_ok = alix_group.load_mls_group_with_lock(storage, |mls_group| { + Ok::<_, crate::groups::GroupError>(alix_group.process_own_leave_request_message( + &mls_group, + storage, + &message_id, + )) + })?; + assert!( + leave_ok.is_ok(), + "expected benign no-op for missing leave-request row, got {leave_ok:?}" + ); + assert!( + alix_group + .process_own_delete_message(storage, &message_id) + .is_ok(), + "expected benign no-op for missing delete row" + ); + + // End-to-end: the same fault must surface through the real sync path, + // not just the handler return value. Publish an own message, inject + // the retryable failure, and receive: the sync transaction must roll + // back and report the failure in the process summary instead of + // silently succeeding. + use crate::groups::send_message_opts::SendMessageOpts; + + alix_group.sync().await?; + alix_group.send_message_optimistic(b"own message", SendMessageOpts::default())?; + alix_group.publish_intents().await?; + + set_test_mode_own_reaction_retryable_error(true); + let summary = alix_group.receive().await?; + set_test_mode_own_reaction_retryable_error(false); + assert!( + !summary.errored.is_empty(), + "retryable failure in an own-reaction handler should be reported by the sync, got {summary:?}" + ); + assert!( + summary.errored.iter().all(|(_, e)| e.is_retryable()), + "the injected failure should still be retryable when surfaced, got {summary:?}" + ); + assert!( + summary.new_messages.is_empty(), + "the message must not count as processed while its reaction fails retryably, got {summary:?}" + ); + + // The rollback left the cursor untouched: with the fault cleared, the + // same message is re-processed successfully on the next sync. + let summary = alix_group.receive().await?; + assert!( + summary.errored.is_empty(), + "re-sync after clearing the fault should succeed, got {summary:?}" + ); + assert_eq!( + summary.new_messages.len(), + 1, + "the rolled-back message should be processed on retry, got {summary:?}" + ); + } + /// Pin the `CommitResult` mapping for each arm of the AppData-aware /// wrapper error so future refactors of `commit_result()` can't /// silently reshuffle what a receiver will write to the remote diff --git a/crates/xmtp_mls/src/utils/test/test_mocks_helpers.rs b/crates/xmtp_mls/src/utils/test/test_mocks_helpers.rs index 82dc83af59..f1a3044df6 100644 --- a/crates/xmtp_mls/src/utils/test/test_mocks_helpers.rs +++ b/crates/xmtp_mls/src/utils/test/test_mocks_helpers.rs @@ -151,3 +151,32 @@ pub fn maybe_mock_package_lifetime() -> Lifetime { Lifetime::default() } } + +/* ---------- own-reaction follow-up storage fault ---------- */ + +const OWN_REACTION_RETRYABLE_ERROR: EnvFlag = EnvFlag("TEST_MODE_OWN_REACTION_RETRYABLE_ERROR"); + +/// Toggle injection of a retryable storage error into the self-publish +/// follow-up reaction handlers (`process_own_leave_request_message` and +/// `process_own_delete_message`). Used to prove those handlers surface a +/// transient failure instead of silently swallowing it. +#[inline] +pub fn set_test_mode_own_reaction_retryable_error(enable: bool) { + OWN_REACTION_RETRYABLE_ERROR.set(enable); +} + +#[inline] +pub fn is_test_mode_own_reaction_retryable_error() -> bool { + OWN_REACTION_RETRYABLE_ERROR.get() +} + +/// If the flag is set, fail with a *retryable* storage error, standing in for a +/// transient DB read failure inside the own-reaction follow-up handlers. +pub fn maybe_mock_own_reaction_retryable_error() -> Result<(), GroupMessageProcessingError> { + if is_test_mode_own_reaction_retryable_error() { + return Err(GroupMessageProcessingError::Db( + xmtp_db::ConnectionError::ReconnectInTransaction, + )); + } + Ok(()) +}