fix: messages not being sent, waiting forever for decryption to finish - WPB-17866 (#3066)

Author: jullianm

WireDomain/Sources/WireDomain/Synchronization/Protocols/PullPendingUpdateEventsSyncProtocol.swift

@@ -27,6 +27,7 @@ public protocol PullPendingUpdateEventsSyncProtocol {
     /// Pull pending update events from the remote, decrypt (if needed),
     /// and store them locally.
 
+    @discardableResult
     func pull() async throws -> AsyncStream<[UpdateEvent]>
 
 }

WireDomain/Sources/WireDomain/Synchronization/SyncState.swift

@@ -36,6 +36,10 @@ public enum SyncState: Equatable {
 
     case liveSyncing
 
+    /// Sync was suspended
+
+    case suspended
+
     public enum InitialSyncState: Equatable {
 
         case pullLastEventID

WireDomain/Sources/WireDomainSupport/Sourcery/generated/AutoMockable.generated.swift

@@ -57,7 +57,9 @@ final class IncrementalSyncObserver: IncrementalSyncObserverProtocol {
                     switch syncState {
                     case .incrementalSyncing(.pullPendingEvents):
                         self?.decryptionState = .inProgress
-                    case .incrementalSyncing(.processPendingEvents), .liveSyncing:
+                    case .incrementalSyncing(.processPendingEvents),
+                         .suspended,
+                         .liveSyncing:
                         self?.decryptionState = .done
                     default:
                         self?.decryptionState = .notStarted

wire-ios-sync-engine/Source/Synchronization/IncrementalSyncObserver.swift

@@ -132,6 +132,7 @@ final class SyncAgent: NSObject, SyncAgentProtocol {
         WireLogger.sync.debug("suspending sync")
         await incrementalSyncToken?.suspend()
         incrementalSyncToken = nil
+        syncStateSubject.send(.suspended)
     }
 
     /// Performs the appropriate sync depending in the local state.
@@ -210,6 +211,7 @@ final class SyncAgent: NSObject, SyncAgentProtocol {
                 }
             } catch {
                 WireLogger.sync.error("failed to perform new incremental sync: \(String(describing: error))")
+                syncStateSubject.send(.suspended)
                 throw error
             }
         } else {

wire-ios-sync-engine/Source/Synchronization/SyncAgent.swift

@@ -208,4 +208,74 @@ final class SyncAgentTests: XCTestCase, InitialSyncProvider, IncrementalSyncProv
         XCTAssertEqual(incrementalSync.perform_Invocations.count, 1)
     }
 
+    func testPerformIncrementalSync_Sync_State_Update_To_Suspended_When_Throwing_Error() async throws {
+        // Given
+        journal[.isSyncV2Enabled] = true
+        let expectation = XCTestExpectation()
+
+        enum Failure: Error {
+            case failed
+        }
+
+        // Mock
+        incrementalSync.perform_MockMethod = {
+            throw Failure.failed
+        }
+
+        var cancellable: AnyCancellable?
+
+        cancellable = syncStateSubject
+            .dropFirst()
+            .sink { state in
+                switch state {
+                case .suspended:
+                    // Then
+                    expectation.fulfill()
+                default:
+                    XCTFail("Sync should be suspended when an error is thrown")
+                }
+            }
+
+        do {
+            // When
+            try await sut.performIncrementalSync()
+        } catch {}
+
+        await fulfillment(of: [expectation])
+    }
+
+    func testPerformIncrementalSync_Sync_State_Update_To_Suspended() async throws {
+        // Given
+        journal[.isSyncV2Enabled] = true
+        let expectation = XCTestExpectation()
+
+        enum Failure: Error {
+            case failed
+        }
+
+        // Mock
+        incrementalSync.perform_MockMethod = {
+            throw Failure.failed
+        }
+
+        var cancellable: AnyCancellable?
+
+        cancellable = syncStateSubject
+            .dropFirst()
+            .sink { state in
+                switch state {
+                case .suspended:
+                    // Then
+                    expectation.fulfill()
+                default:
+                    XCTFail("Sync should be suspended")
+                }
+            }
+
+        // When
+        sut.suspend()
+
+        await fulfillment(of: [expectation])
+    }
+
 }