             return try fetch()
         }
     }
-    
-    private func reduce(
-        fetchedValue: DatabaseFuture<Reducer.Fetched>,
-        completion: @escaping (Result<Reducer.Value, Error>) -> Void)
-    {
-        do {
-            if let value = try reducer.value(fetchedValue.wait()) {
-                completion(.success(value))
-            }
-        } catch {
-            completion(.failure(error))
-        }
-    }
 }
 
 class ValueObserverToken<Reducer: _ValueReducer>: TransactionObserver {

        guard isChanged else { return }
        isChanged = false
        
        let fetchedValue: DatabaseFuture<Reducer.Fetched>
        
        if requiresWriteAccess || needsRecordingObservedRegion {
            // Synchronous fetch
            fetchedValue = DatabaseFuture(Result {
                try recordingObservedRegionIfNeeded(db) {
                    try reducer.fetch(db, requiringWriteAccess: requiresWriteAccess)
                }
            })
        } else {
            // Concurrent fetch
            fetchedValue = writer.concurrentRead(reducer.fetch)
        }
        
        // Wait for future fetched value in reduceQueue. This guarantees:
        // - that notifications have the same ordering as transactions.
        // - that expensive reduce operations are computed without blocking
        //   any database dispatch queue.
        reduceQueue.async {
            if self.isCancelled { return }
            self.reduce(fetchedValue: fetchedValue) { result in
                do {
                    try self.send(result.get())
                } catch {
                    self.send(error)
                }
            }
        }
    }

}

extension ValueObserver {
    /// This method must be called once, before the observer is added
    /// to the database writer.
    func fetchInitialValue(_ db: Database) throws -> Reducer.Value {
        guard let value = try fetchNextValue(db) else {
