Fix join flow: skip stale groups in discovery + add sync fallback (#517)

* Skip self-created single-member groups in discoverNewConversations

When consumeInboxOnly() is used for the scanner flow, the pre-created
'New Convo' group exists in XMTP but may be cleaned up from the local DB.
discoverNewConversations() would then re-discover it and write it back
with isUnused: false, making it visible as 'New Convo' in the list.

Skip groups where we are the creator and the only member — these are
unused pre-created conversations that should not be re-introduced.

* Add discovery loop fallback for join flow

The XMTP conversation stream does not call sync_welcomes() at startup
(unlike StreamAllMessages), so it cannot discover groups the client is
added to after stream subscription. This causes the join flow to get
stuck at 'Verifying' indefinitely when the stream fails to deliver.

Add requestDiscovery() to SyncingManager that runs syncAllConversations
+ discoverNewConversations on demand. Thread it through InboxStateMachine
and InboxStateManager protocols.

In ConversationStateMachine.handleJoin, start a discovery loop alongside
the DB observation that periodically syncs welcomes with backoff
(3s, 6s, 12s, 15s cap). The DB observation fires as soon as the group
is written by discoverNewConversations. The loop is cancelled on success,
cancellation, or error.

* Remove accidentally committed .mote-preview.html

* Add .mote-preview.html to gitignore

* Remove investigation and plan docs

* Fix duplicate MessagingService creation in messagingServiceSync

messagingServiceSync could create duplicate services for the same
clientId when called concurrently. The nonisolated getAwakeService
check and the async registerExternalService had a race window where
two callers would both see nil and both create services.

Add getOrSetAwakeService to InboxLifecycleManager that atomically
checks the lock-based cache and stores the service if no entry exists.
If a duplicate is detected, the redundant service is stopped immediately.

* Revert "Fix duplicate MessagingService creation in messagingServiceSync"

This reverts commit 630481d.

* Stop orphaned duplicate services in registerExternalService

When messagingServiceSync is called concurrently for the same clientId,
duplicate MessagingService instances can be created before the first
one is registered. The second call to registerExternalService finds
the inbox already tracked and previously just silently returned,
leaving the duplicate's SyncingManager running orphaned streams.

Now stop the rejected service so its streams and auth flow are cleaned up.

* Address PR feedback: stop duplicate at call site, add deinit cleanup

Move duplicate service stop() from registerExternalService to the
caller in SessionManager. registerExternalService now returns Bool
indicating success — the caller stops the orphan if registration
was rejected. This avoids stopping a service that another caller
may already be using.

Also add discoveryTask?.cancel() to ConversationStateMachine deinit
for consistency with other task cleanup.

Author: yewreeka

.gitignore

@@ -129,3 +129,4 @@ qa/cxdb/qa.sqlite
 qa/cxdb/qa.sqlite-wal
 qa/cxdb/qa.sqlite-shm
 .DS_Store
+.mote-preview.html

ConvosCore/Sources/ConvosCore/Inboxes/ConversationStateMachine.swift

@@ -95,6 +95,9 @@ public actor ConversationStateMachine {
     // Database observation task for tracking conversation join
     private var observationTask: Task<String, Error>?
 
+    // Discovery loop task for syncing welcomes during join flow
+    private var discoveryTask: Task<Void, Never>?
+
     // MARK: - State Observation
 
     private var stateContinuations: [AsyncStream<State>.Continuation] = []
@@ -169,6 +172,7 @@ public actor ConversationStateMachine {
         messageStreamContinuation?.finish()
         messageProcessingTask?.cancel()
         observationTask?.cancel()
+        discoveryTask?.cancel()
     }
 
     private func setupMessageStream() {
@@ -668,12 +672,35 @@ public actor ConversationStateMachine {
             inviteTag: invite.invitePayload.tag
         )
 
+        // The XMTP conversation stream does not call sync_welcomes() on startup,
+        // so it cannot discover groups the client is added to after subscription.
+        // Run a discovery loop that periodically syncs welcomes and processes any
+        // newly discovered groups. The DB observation above fires as soon as the
+        // group is written to the database by discoverNewConversations.
+        discoveryTask = Task { [weak self] in
+            var interval: Duration = .seconds(3)
+            let maxInterval: Duration = .seconds(15)
+            while !Task.isCancelled {
+                do {
+                    try await Task.sleep(for: interval)
+                    try Task.checkCancellation()
+                    Log.info("Join flow: triggering discovery sync...")
+                    await self?.inboxStateManager.requestDiscovery()
+                    interval = min(interval * 2, maxInterval)
+                } catch {
+                    break
+                }
+            }
+        }
+
         do {
             guard let task = observationTask else {
                 throw ConversationStateMachineError.timedOut
             }
             let conversationId = try await task.value
             observationTask = nil
+            discoveryTask?.cancel()
+            discoveryTask = nil
 
             Log.info("Conversation joined successfully: \(conversationId)")
             QAEvent.emit(.conversation, "joined", ["id": conversationId])
@@ -691,6 +718,8 @@ public actor ConversationStateMachine {
             )))
         } catch is CancellationError {
             observationTask = nil
+            discoveryTask?.cancel()
+            discoveryTask = nil
             await clearInviteJoinErrorHandler()
             Log.debug("Conversation join observation cancelled")
             guard case .joinFailed = _state else {
@@ -699,6 +728,8 @@ public actor ConversationStateMachine {
             Log.debug("Already in joinFailed state, not propagating cancellation")
         } catch {
             observationTask = nil
+            discoveryTask?.cancel()
+            discoveryTask = nil
             await clearInviteJoinErrorHandler()
             Log.error("Error waiting for conversation to join: \(error)")
             guard case .joinFailed = _state else {
@@ -757,6 +788,8 @@ extension ConversationStateMachine {
         messageProcessingTask?.cancel()
         observationTask?.cancel()
         observationTask = nil
+        discoveryTask?.cancel()
+        discoveryTask = nil
 
         if let conversationId {
             let inboxReady = try await inboxStateManager.waitForInboxReadyResult()
@@ -863,6 +896,8 @@ extension ConversationStateMachine {
         messageProcessingTask?.cancel()
         observationTask?.cancel()
         observationTask = nil
+        discoveryTask?.cancel()
+        discoveryTask = nil
         await clearInviteJoinErrorHandler()
         emitStateChange(.uninitialized)
     }

ConvosCore/Sources/ConvosCore/Inboxes/InboxLifecycleManager.swift

@@ -60,7 +60,8 @@ public protocol InboxLifecycleManagerProtocol: Actor {
     func getOrCreateService(clientId: String, inboxId: String) -> any MessagingServiceProtocol
 
     func getOrWake(clientId: String, inboxId: String) async throws -> any MessagingServiceProtocol
-    func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async
+    @discardableResult
+    func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async -> Bool
     func isAwake(clientId: String) -> Bool
     func isSleeping(clientId: String) -> Bool
     func rebalance() async
@@ -352,10 +353,11 @@ public actor InboxLifecycleManager: InboxLifecycleManagerProtocol {
         return try await wake(clientId: clientId, inboxId: inboxId, reason: .userInteraction)
     }
 
-    public func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async {
+    @discardableResult
+    public func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async -> Bool {
         guard awakeInboxes[clientId] == nil else {
             Log.debug("Inbox already tracked, skipping external registration: \(clientId)")
-            return
+            return false
         }
         if awakeInboxes.count >= maxAwakeInboxes {
             let freed = await sleepLeastRecentlyUsed(excluding: [clientId])
@@ -365,12 +367,13 @@ public actor InboxLifecycleManager: InboxLifecycleManagerProtocol {
         }
         guard awakeInboxes[clientId] == nil else {
             Log.debug("Inbox registered by another task during eviction, skipping: \(clientId)")
-            return
+            return false
         }
         awakeInboxes[clientId] = service
         _sleepingClientIds.remove(clientId)
         _sleepTimes.removeValue(forKey: clientId)
         Log.debug("Registered external service: \(clientId), total awake: \(awakeInboxes.count)")
+        return true
     }
 
     public func isAwake(clientId: String) -> Bool {

ConvosCore/Sources/ConvosCore/Inboxes/InboxStateMachine.swift

@@ -309,6 +309,11 @@ public actor InboxStateMachine {
         await syncingManager.setInviteJoinErrorHandler(handler)
     }
 
+    func requestDiscovery() async {
+        guard let syncingManager else { return }
+        await syncingManager.requestDiscovery()
+    }
+
     // MARK: - Private
 
     private func enqueueAction(_ action: Action) {

ConvosCore/Sources/ConvosCore/Inboxes/InboxStateManager.swift

@@ -15,6 +15,8 @@ public protocol InboxStateManagerProtocol: AnyObject, Sendable {
     func waitForDeletionComplete() async
     func setInviteJoinErrorHandler(_ handler: (any InviteJoinErrorHandler)?) async throws
 
+    func requestDiscovery() async
+
     func addObserver(_ observer: InboxStateObserver)
     func removeObserver(_ observer: InboxStateObserver)
 
@@ -163,6 +165,11 @@ public final class InboxStateManager: InboxStateManagerProtocol, @unchecked Send
         await stateMachine.setInviteJoinErrorHandler(handler)
     }
 
+    public func requestDiscovery() async {
+        guard let stateMachine else { return }
+        await stateMachine.requestDiscovery()
+    }
+
     public func reauthorize(inboxId: String, clientId: String) async throws -> InboxReadyResult {
         guard let stateMachine = stateMachine else {
             throw InboxStateError.inboxNotReady

ConvosCore/Sources/ConvosCore/Mocks/MockInboxLifecycleManager.swift

@@ -95,10 +95,12 @@ public actor MockInboxLifecycleManager: InboxLifecycleManagerProtocol {
         _sleepingClientIds.removeAll()
     }
 
-    public func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async {
+    @discardableResult
+    public func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async -> Bool {
         mockServices[clientId] = service
         _awakeClientIds.insert(clientId)
         _sleepingClientIds.remove(clientId)
+        return true
     }
 
     public func prepareUnusedConversationIfNeeded() async {}

ConvosCore/Sources/ConvosCore/Mocks/MockInboxStateManager.swift

@@ -45,6 +45,9 @@ public final class MockInboxStateManager: InboxStateManagerProtocol, @unchecked
     public func setInviteJoinErrorHandler(_ handler: (any InviteJoinErrorHandler)?) async throws {
     }
 
+    public func requestDiscovery() async {
+    }
+
     public func addObserver(_ observer: any InboxStateObserver) {
         observers.removeAll { $0.observer == nil }
         observers.append(WeakStateObserver(observer: observer))

ConvosCore/Sources/ConvosCore/Sessions/SessionManager.swift

@@ -284,7 +284,11 @@ public final class SessionManager: SessionManagerProtocol, @unchecked Sendable {
             apiClient: apiClient
         )
         Task { [lifecycleManager] in
-            await lifecycleManager.registerExternalService(service, clientId: clientId)
+            let registered = await lifecycleManager.registerExternalService(service, clientId: clientId)
+            if !registered {
+                Log.warning("Stopping duplicate MessagingService for \(clientId)")
+                await service.stop()
+            }
         }
         return service
     }

ConvosCore/Sources/ConvosCore/Syncing/SyncingManager.swift

@@ -10,6 +10,7 @@ public protocol SyncingManagerProtocol: Actor {
     func stop() async
     func pause() async
     func resume() async
+    func requestDiscovery() async
     func setInviteJoinErrorHandler(_ handler: (any InviteJoinErrorHandler)?) async
 }
 
@@ -440,6 +441,12 @@ actor SyncingManager: SyncingManagerProtocol {
             var discoveredCount: Int = 0
             for group in groups where !existingIds.contains(group.id) {
                 do {
+                    let creatorInboxId = try await group.creatorInboxId()
+                    let memberCount = try await group.members.count
+                    if creatorInboxId == params.client.inboxId && memberCount <= 1 {
+                        Log.debug("Skipping self-created single-member group: \(group.id)")
+                        continue
+                    }
                     try await streamProcessor.processConversation(group, params: params)
                     discoveredCount += 1
                 } catch {
@@ -610,6 +617,22 @@ actor SyncingManager: SyncingManagerProtocol {
         await processJoinRequestsAfterSync(params: params)
     }
 
+    func requestDiscovery() async {
+        guard case .ready(let params) = _state else {
+            Log.debug("requestDiscovery ignored - not in ready state (\(_state))")
+            return
+        }
+        do {
+            let syncStart = CFAbsoluteTimeGetCurrent()
+            _ = try await params.client.conversationsProvider.syncAllConversations(consentStates: params.consentStates)
+            let syncElapsed = String(format: "%.0f", (CFAbsoluteTimeGetCurrent() - syncStart) * 1000)
+            Log.info("[PERF] sync.requestDiscovery: \(syncElapsed)ms")
+            await discoverNewConversations(params: params)
+        } catch {
+            Log.error("requestDiscovery failed: \(error)")
+        }
+    }
+
     private func handleStop() async throws {
         Log.info("Stopping sync...")
         emitStateChange(.stopping)

ConvosCore/Tests/ConvosCoreTests/SleepingInboxMessageCheckerTests.swift

@@ -649,9 +649,11 @@ actor TestableInboxLifecycleManager: InboxLifecycleManagerProtocol {
         fatalError("Not implemented for tests")
     }
 
-    func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async {
+    @discardableResult
+    func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async -> Bool {
         _awakeClientIds.insert(clientId)
         _sleepingClientIds.remove(clientId)
+        return true
     }
 
     func prepareUnusedConversationIfNeeded() async {}