Merge pull request #1364 from groue/issue1362-test

Fix ValueObservation deadlock

Author: groue

GRDB/ValueObservation/Observers/ValueConcurrentObserver.swift

@@ -101,6 +101,19 @@ final class ValueConcurrentObserver<Reducer: ValueReducer, Scheduler: ValueObser
         }
     }
 
+    /// The fetching state for observation of constant regions.
+    enum FetchingState {
+        /// No need to fetch.
+        case idle
+        
+        /// Waiting for a fetched value.
+        case fetching
+        
+        /// Waiting for a fetched value, and for a subsequent  fetch after
+        /// that, because a change has been detected as we were fetching.
+        case fetchingAndNeedsFetch
+    }
+    
     /// Ability to notify observation events
     private struct NotificationCallbacks {
         let events: ValueObservationEvents
@@ -132,6 +145,9 @@ final class ValueConcurrentObserver<Reducer: ValueReducer, Scheduler: ValueObser
     /// Ability to notify observation events, protected by `lock`.
     private var notificationCallbacks: NotificationCallbacks?
 
+    /// The fetching state for observation of constant regions.
+    @LockedBox private var fetchingState = FetchingState.idle
+    
     /// Support for `TransactionObserver`, protected by the serialized writer
     /// dispatch queue.
     private var observationState = ObservationState.notObserving
@@ -692,13 +708,10 @@ extension ValueConcurrentObserver: TransactionObserver {
         events.databaseDidChange?()
 
         // Fetch
-        let future: DatabaseFuture<Reducer.Fetched>
-        
         switch trackingMode {
         case .constantRegion, .constantRegionRecordedFromSelection:
-            future = databaseAccess.dbPool.concurrentRead { db in
-                try databaseAccess.fetch(db)
-            }
+            setNeedsFetching(databaseAccess: databaseAccess)
+            
         case .nonConstantRegionRecordedFromSelection:
             // When the tracked region is not constant, we can't perform
             // concurrent fetches of observed values.
@@ -723,26 +736,63 @@ extension ValueConcurrentObserver: TransactionObserver {
                 }
 
                 observationState.region = observedRegion
-                future = DatabaseFuture(.success(fetchedValue))
+                reduce(.success(fetchedValue))
             } catch {
                 stopDatabaseObservation(writerDB)
                 notifyError(error)
                 return
             }
         }
-        
-        // Reduce
-        //
-        // Reducing is performed asynchronously, so that we do not lock
-        // the writer dispatch queue longer than necessary.
-        //
-        // Important: reduceQueue.async guarantees the same ordering between
-        // transactions and notifications!
+    }
+    
+    private func setNeedsFetching(databaseAccess: DatabaseAccess) {
+        $fetchingState.update { state in
+            switch state {
+            case .idle:
+                state = .fetching
+                asyncFetch(databaseAccess: databaseAccess)
+                
+            case .fetching:
+                state = .fetchingAndNeedsFetch
+                
+            case .fetchingAndNeedsFetch:
+                break
+            }
+        }
+    }
+    
+    private func asyncFetch(databaseAccess: DatabaseAccess) {
+        databaseAccess.dbPool.asyncRead { [self] dbResult in
+            let isNotifying = self.lock.synchronized { self.notificationCallbacks != nil }
+            guard isNotifying else { return /* Cancelled */ }
+            
+            let fetchResult = dbResult.flatMap { db in
+                Result { try databaseAccess.fetch(db) }
+            }
+            
+            self.reduce(fetchResult)
+            
+            $fetchingState.update { state in
+                switch state {
+                case .idle:
+                    // GRDB bug
+                    preconditionFailure()
+                    
+                case .fetching:
+                    state = .idle
+                    
+                case .fetchingAndNeedsFetch:
+                    state = .fetching
+                    asyncFetch(databaseAccess: databaseAccess)
+                }
+            }
+        }
+    }
+    
+    private func reduce(_ fetchResult: Result<Reducer.Fetched, Error>) {
         reduceQueue.async {
             do {
-                // Wait until fetch has completed
-                // TODO: find a way to guarantee correct ordering without waiting for a semaphore and blocking a thread.
-                let fetchedValue = try future.wait()
+                let fetchedValue = try fetchResult.get()
 
                 let isNotifying = self.lock.synchronized { self.notificationCallbacks != nil }
                 guard isNotifying else { return /* Cancelled */ }

Makefile

@@ -52,14 +52,17 @@ MAX_IOS_DESTINATION := $(shell xcrun simctl list -j devices available | Scripts/
 MIN_TVOS_DESTINATION := $(shell xcrun simctl list -j devices available | Scripts/destination.rb | grep tvOS | sort -n | head -1 | cut -wf 3 | sed 's/\(.*\)/"platform=tvOS Simulator,id=\1"/')
 MAX_TVOS_DESTINATION := $(shell xcrun simctl list -j devices available | Scripts/destination.rb | grep tvOS | sort -rn | head -1 | cut -wf 3 | sed 's/\(.*\)/"platform=tvOS Simulator,id=\1"/')
 
-# If xcbeautify or xcpretty is available, use it for xcodebuild output
-XCPRETTY = 
-XCBEAUTIFY_PATH := $(shell command -v xcbeautify 2> /dev/null)
-XCPRETTY_PATH := $(shell command -v xcpretty 2> /dev/null)
-ifdef XCBEAUTIFY_PATH
-  XCPRETTY = | xcbeautify
-else ifdef XCPRETTY_PATH
-  XCPRETTY = | xcpretty -c
+  # If xcbeautify or xcpretty is available, use it for xcodebuild output, except in CI.
+XCPRETTY =
+ifeq ($(CI),true)
+else
+  XCBEAUTIFY_PATH := $(shell command -v xcbeautify 2> /dev/null)
+  XCPRETTY_PATH := $(shell command -v xcpretty 2> /dev/null)
+  ifdef XCBEAUTIFY_PATH
+    XCPRETTY = | xcbeautify
+  else ifdef XCPRETTY_PATH
+    XCPRETTY = | xcpretty -c
+  endif
 endif
 
 # =====

Tests/GRDBTests/ValueObservationRecorder.swift

@@ -310,43 +310,43 @@ extension XCTestCase {
     /// - `[0, 1, 2]` (unexpected value)
     /// - `[1, 0, 1]` (unexpected value)
     func assertValueObservationRecordingMatch<R, E>(
-        recorded recordedValues: R,
-        expected expectedValues: E,
-        allowMissingLastValue: Bool = false,
+        recorded: R,
+        expected: E,
         _ message: @autoclosure () -> String = "",
         file: StaticString = #file,
         line: UInt = #line)
         where
-        R: BidirectionalCollection,
-        E: BidirectionalCollection,
+        R: Collection,
+        E: Collection,
         R.Element == E.Element,
         R.Element: Equatable
     {
-        guard let value = expectedValues.last else {
-            if !recordedValues.isEmpty {
-                XCTFail("unexpected recorded prefix \(Array(recordedValues)) - \(message())", file: file, line: line)
-            }
-            return
+        XCTAssertTrue(
+            valueObservationRecordingMatch(recorded: recorded, expected: expected),
+            "Unexpected recording \(Array(recorded)) - \(message())",
+            file: file, line: line)
+    }
+    
+    func valueObservationRecordingMatch<R, E>(
+        recorded: R,
+        expected: E)
+    -> Bool
+    where R: Collection,
+          E: Collection,
+          R.Element == E.Element,
+          R.Element: Equatable
+    {
+        guard let first = recorded.first else {
+            return expected.isEmpty
         }
 
-        let recordedSuffix = recordedValues.reversed().prefix(while: { $0 == value })
-        let expectedSuffix = expectedValues.reversed().prefix(while: { $0 == value })
-        if !allowMissingLastValue {
-            // Both missing and repeated values are allowed in the recorded values.
-            // This is because of asynchronous DatabasePool observations.
-            if recordedSuffix.isEmpty {
-                XCTFail("missing expected value \(value) - \(message()) in \(recordedValues)", file: file, line: line)
+        return expected.indices.lazy
+            .filter { expected[$0] == first }
+            .contains {
+                valueObservationRecordingMatch(
+                    recorded: recorded.drop(while: { $0 == first }),
+                    expected: expected[$0...].dropFirst())
             }
-        }
-        
-        let remainingRecordedValues = recordedValues.prefix(recordedValues.count - recordedSuffix.count)
-        let remainingExpectedValues = expectedValues.prefix(expectedValues.count - expectedSuffix.count)
-        assertValueObservationRecordingMatch(
-            recorded: remainingRecordedValues,
-            expected: remainingExpectedValues,
-            // Other values can be missed
-            allowMissingLastValue: true,
-            message(), file: file, line: line)
     }
 }