     @available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
     public func values(
         in reader: DatabaseReader,
-        scheduling scheduler: ValueObservationScheduler = .async(onQueue: .main))
+        scheduling scheduler: ValueObservationScheduler = .async(onQueue: .main),
+        bufferingPolicy: BufferingPolicy = .unbounded)
     -> AsyncValueObservation<Reducer.Value>
     {
-        AsyncValueObservation { onError, onChange in
+        AsyncValueObservation(bufferingPolicy: bufferingPolicy) { onError, onChange in
             self.start(in: reader, scheduling: scheduler, onError: onError, onChange: onChange)
         }
     }

extension ValueObservation {
    // MARK: - Asynchronous Observation
    
    /// The database observation, as an asynchronous sequence of
    /// database changes.
    ///

        // cancel the observation when the Swift async runtime releases
        // the iterator.
        var cancellable: AnyDatabaseCancellable?
        let stream = AsyncThrowingStream(Element.self, bufferingPolicy: bufferingPolicy) { continuation in
            cancellable = AnyDatabaseCancellable(start(
                // onError
                { error in
