fix(xmtp_db): make TestInstrumentation database-lock panic opt-in

crates/xmtp_db/src/encrypted_store/database/instrumentation.rs

@@ -3,12 +3,33 @@ use diesel::connection::InstrumentationEvent;
 use diesel::result::Error as DieselError;
 use std::fmt::Write;
 
-// Test instrumentatiomn
+// Test instrumentation
 // prints out query on error
-// panics on database lock
+// panics on database lock ONLY when explicitly opted in via the
+// `XMTP_PANIC_ON_DB_LOCK` env var (accepts "1" or "true").
+//
+// `database is locked` is a transient, retryable condition in this codebase
+// (SQLite is configured with `PRAGMA busy_timeout` and the error is classified
+// as retryable). Panicking on it by default aborts the worker thread and crashes
+// downstream test harnesses (e.g. the node-sdk Vitest fork workers), so the panic
+// is opt-in for developers actively debugging lock contention.
 #[allow(unused)]
 pub struct TestInstrumentation;
 
+/// Returns true when the caller has explicitly opted in to panicking on
+/// `database is locked` via the `XMTP_PANIC_ON_DB_LOCK` env var.
+fn panic_on_db_lock_enabled() -> bool {
+    panic_on_db_lock_from_env(std::env::var("XMTP_PANIC_ON_DB_LOCK"))
+}
+
+/// Pure decision helper: the opt-in is enabled only when the env var resolves to
+/// exactly `"1"` or `"true"`. Anything else (unset, empty, other values) keeps the
+/// default non-panicking behavior. Kept separate from the env read so it can be
+/// unit-tested without mutating the process environment.
+fn panic_on_db_lock_from_env(var: Result<String, std::env::VarError>) -> bool {
+    matches!(var, Ok(s) if s == "1" || s == "true")
+}
+
 impl Instrumentation for TestInstrumentation {
     fn on_connection_event(&mut self, event: InstrumentationEvent<'_>) {
         use InstrumentationEvent::*;
@@ -32,7 +53,7 @@ impl Instrumentation for TestInstrumentation {
                             tracing::error!("details: {},", details);
                         }
                         tracing::error!("{}", s);
-                        if s.contains("database is locked") {
+                        if s.contains("database is locked") && panic_on_db_lock_enabled() {
                             panic!("database locked");
                         }
                     }
@@ -52,3 +73,32 @@ impl Instrumentation for TestInstrumentation {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::panic_on_db_lock_from_env;
+    use std::env::VarError;
+
+    #[test]
+    fn db_lock_panic_disabled_by_default() {
+        // Unset env var (the default in CI / node-sdk Vitest workers) must NOT
+        // opt in to panicking, so a `database is locked` error stays recoverable.
+        assert!(!panic_on_db_lock_from_env(Err(VarError::NotPresent)));
+    }
+
+    #[test]
+    fn db_lock_panic_opt_in_values() {
+        assert!(panic_on_db_lock_from_env(Ok("1".to_string())));
+        assert!(panic_on_db_lock_from_env(Ok("true".to_string())));
+    }
+
+    #[test]
+    fn db_lock_panic_ignores_other_values() {
+        for v in ["0", "false", "", "yes", "TRUE", "True"] {
+            assert!(
+                !panic_on_db_lock_from_env(Ok(v.to_string())),
+                "value {v:?} should not enable the panic"
+            );
+        }
+    }
+}

crates/xmtp_mls/src/subscriptions/stream_messages.rs

@@ -596,6 +596,7 @@ pub mod tests {
     use crate::tester;
     use futures::stream::StreamExt;
     use rstest::*;
+    use xmtp_db::group_message::GroupMessageKind;
 
     #[xmtp_common::timeout(std::time::Duration::from_secs(30))]
     #[rstest]
@@ -613,15 +614,24 @@ pub mod tests {
         let bob_group = bob_groups.first().unwrap();
         alice_group.sync().await.unwrap();
 
-        let stream = alice_group.stream().await.unwrap();
+        // The `group_updated` membership-change commit from `add_members` above can
+        // race into the stream ahead of the application messages, which made this
+        // test flaky (see xmtp/libxmtp#3765). Filter to application messages so the
+        // assertions below are deterministic regardless of whether the commit
+        // message is delivered through the stream.
+        let stream = alice_group.stream().await.unwrap().filter(|msg| {
+            futures::future::ready(
+                msg.as_ref()
+                    .map(|m| m.kind == GroupMessageKind::Application)
+                    .unwrap_or(true),
+            )
+        });
         futures::pin_mut!(stream);
         bob_group
             .send_message(b"hello", SendMessageOpts::default())
             .await
             .unwrap();
 
-        // group updated msg/bob is added
-        // assert_msg_exists!(stream);
         assert_msg!(stream, "hello");
 
         bob_group