ID @Fetch*.load events to fix cancel race

Sources/SQLiteData/Fetch.swift

@@ -39,12 +39,16 @@ public struct Fetch<Value: Sendable>: Sendable {
     private let box: FetchBox<Value>
     private let state: SwiftUI.State<FetchBox<Value>>
     private let generation = SwiftUI.State(wrappedValue: 0)
+
+    var loadGeneration: LoadGeneration { state.wrappedValue.loadGeneration }
   #else
     /// The underlying shared reader powering the property wrapper.
     ///
     /// Shared readers come from the [Sharing](https://github.com/pointfreeco/swift-sharing)
     /// package, a general solution to observing and persisting changes to external data sources.
     public let sharedReader: SharedReader<Value>
+
+    let loadGeneration = LoadGeneration()
   #endif
 
   /// Data associated with the underlying query.
@@ -58,7 +62,10 @@ public struct Fetch<Value: Sendable>: Sendable {
   /// ``isLoading``, and ``publisher``.
   public var projectedValue: Self {
     get { self }
-    nonmutating set { sharedReader.projectedValue = newValue.sharedReader.projectedValue }
+    nonmutating set {
+      loadGeneration.invalidate()
+      sharedReader.projectedValue = newValue.sharedReader.projectedValue
+    }
   }
 
   /// Returns a ``sharedReader`` for the given key path.
@@ -127,8 +134,15 @@ public struct Fetch<Value: Sendable>: Sendable {
     _ request: some FetchKeyRequest<Value>,
     database: (any DatabaseReader)? = nil
   ) async throws -> FetchSubscription {
-    try await sharedReader.load(.fetch(request, database: database))
-    return FetchSubscription(sharedReader: sharedReader)
+    try await withSubscription {
+      try await sharedReader.load(.fetch(request, database: database))
+    }
+  }
+
+  private func withSubscription(_ load: () async throws -> Void) async throws -> FetchSubscription {
+    let token = loadGeneration.begin()
+    try await load()
+    return FetchSubscription(sharedReader: sharedReader, token: token)
   }
 
   #if !canImport(SwiftUI)
@@ -183,8 +197,9 @@ extension Fetch {
     database: (any DatabaseReader)? = nil,
     scheduler: some ValueObservationScheduler & Hashable
   ) async throws -> FetchSubscription {
-    try await sharedReader.load(.fetch(request, database: database, scheduler: scheduler))
-    return FetchSubscription(sharedReader: sharedReader)
+    try await withSubscription {
+      try await sharedReader.load(.fetch(request, database: database, scheduler: scheduler))
+    }
   }
 }
 
@@ -249,8 +264,9 @@ extension Fetch: Equatable where Value: Equatable {
       database: (any DatabaseReader)? = nil,
       animation: Animation?
     ) async throws -> FetchSubscription {
-      try await sharedReader.load(.fetch(request, database: database, animation: animation))
-      return FetchSubscription(sharedReader: sharedReader)
+      try await withSubscription {
+        try await sharedReader.load(.fetch(request, database: database, animation: animation))
+      }
     }
   }
 #endif

Sources/SQLiteData/FetchAll+Sections.swift

@@ -196,7 +196,8 @@ extension FetchAll {
   ///   - database: The database to read from. A value of `nil` will use the default database
   ///     (`@Dependency(\.defaultDatabase)`).
   public init<
-    V: QueryRepresentable, From: StructuredQueriesCore.Table
+    V: QueryRepresentable,
+    From: StructuredQueriesCore.Table
   >(
     wrappedValue: [Element] = [],
     _ statement: Select<V, From, ()>,
@@ -232,7 +233,9 @@ extension FetchAll {
   ///     (`@Dependency(\.defaultDatabase)`).
   @_documentation(visibility: private)
   public init<
-    V: QueryRepresentable, From: StructuredQueriesCore.Table, J: StructuredQueriesCore.Table
+    V: QueryRepresentable,
+    From: StructuredQueriesCore.Table,
+    J: StructuredQueriesCore.Table
   >(
     wrappedValue: [Element] = [],
     _ statement: Select<V, From, J>,
@@ -371,7 +374,8 @@ extension FetchAll {
   /// - Returns: A subscription associated with the observation.
   @discardableResult
   public func load<
-    V: QueryRepresentable, From: StructuredQueriesCore.Table
+    V: QueryRepresentable,
+    From: StructuredQueriesCore.Table
   >(
     _ statement: Select<V, From, ()>,
     @_SectionBuilder<String?> sectionBy sectioning: (From.TableColumns) -> _Sectioning<String?>?,
@@ -404,7 +408,9 @@ extension FetchAll {
   @_documentation(visibility: private)
   @discardableResult
   public func load<
-    V: QueryRepresentable, From: StructuredQueriesCore.Table, J: StructuredQueriesCore.Table
+    V: QueryRepresentable,
+    From: StructuredQueriesCore.Table,
+    J: StructuredQueriesCore.Table
   >(
     _ statement: Select<V, From, J>,
     @_SectionBuilder<String?> sectionBy sectioning: (From.TableColumns, J.TableColumns) ->
@@ -522,14 +528,15 @@ extension FetchAll {
     guard let sectioning else {
       removeSections()
       let statement: Select<From, From, ()> = statement.selectStar()
-      try await sharedReader.load(
-        FetchKey(
-          request: FetchAllStatementValueRequest(statement: statement),
-          database: database,
-          scheduler: scheduler
+      return try await withSubscription {
+        try await sharedReader.load(
+          FetchKey(
+            request: FetchAllStatementValueRequest(statement: statement),
+            database: database,
+            scheduler: scheduler
+          )
         )
-      )
-      return FetchSubscription(sharedReader: sharedReader)
+      }
     }
     return try await loadSections(
       request: FetchAllSectionedStatementValueRequest(statement: statement, sectionBy: sectioning),
@@ -553,14 +560,15 @@ extension FetchAll {
     defer {
       sharedReader.projectedValue = sectionedReader.elements.projectedValue
     }
-    try await sectionedReader.load(
-      FetchKey(
-        request: request,
-        database: database,
-        scheduler: scheduler
+    return try await withSubscription {
+      try await sectionedReader.load(
+        FetchKey(
+          request: request,
+          database: database,
+          scheduler: scheduler
+        )
       )
-    )
-    return FetchSubscription(sharedReader: sharedReader, sectionedReader: sectionedReader)
+    }
   }
 }
 
@@ -714,7 +722,8 @@ extension FetchAll {
   ///   - scheduler: The scheduler to observe from. By default, database observation is performed
   ///     asynchronously on the main queue.
   public init<
-    V: QueryRepresentable, From: StructuredQueriesCore.Table
+    V: QueryRepresentable,
+    From: StructuredQueriesCore.Table
   >(
     wrappedValue: [Element] = [],
     _ statement: Select<V, From, ()>,
@@ -753,7 +762,9 @@ extension FetchAll {
   ///     asynchronously on the main queue.
   @_documentation(visibility: private)
   public init<
-    V: QueryRepresentable, From: StructuredQueriesCore.Table, J: StructuredQueriesCore.Table
+    V: QueryRepresentable,
+    From: StructuredQueriesCore.Table,
+    J: StructuredQueriesCore.Table
   >(
     wrappedValue: [Element] = [],
     _ statement: Select<V, From, J>,
@@ -905,7 +916,8 @@ extension FetchAll {
   /// - Returns: A subscription associated with the observation.
   @discardableResult
   public func load<
-    V: QueryRepresentable, From: StructuredQueriesCore.Table
+    V: QueryRepresentable,
+    From: StructuredQueriesCore.Table
   >(
     _ statement: Select<V, From, ()>,
     @_SectionBuilder<String?> sectionBy sectioning: (From.TableColumns) -> _Sectioning<String?>?,
@@ -941,7 +953,9 @@ extension FetchAll {
   @_documentation(visibility: private)
   @discardableResult
   public func load<
-    V: QueryRepresentable, From: StructuredQueriesCore.Table, J: StructuredQueriesCore.Table
+    V: QueryRepresentable,
+    From: StructuredQueriesCore.Table,
+    J: StructuredQueriesCore.Table
   >(
     _ statement: Select<V, From, J>,
     @_SectionBuilder<String?> sectionBy sectioning: (From.TableColumns, J.TableColumns) ->
@@ -1155,7 +1169,8 @@ extension FetchAll {
     ///     the fetched results.
     @available(iOS 17, macOS 14, tvOS 17, watchOS 10, *)
     public init<
-      V: QueryRepresentable, From: StructuredQueriesCore.Table
+      V: QueryRepresentable,
+      From: StructuredQueriesCore.Table
     >(
       wrappedValue: [Element] = [],
       _ statement: Select<V, From, ()>,
@@ -1191,7 +1206,9 @@ extension FetchAll {
     @_documentation(visibility: private)
     @available(iOS 17, macOS 14, tvOS 17, watchOS 10, *)
     public init<
-      V: QueryRepresentable, From: StructuredQueriesCore.Table, J: StructuredQueriesCore.Table
+      V: QueryRepresentable,
+      From: StructuredQueriesCore.Table,
+      J: StructuredQueriesCore.Table
     >(
       wrappedValue: [Element] = [],
       _ statement: Select<V, From, J>,
@@ -1340,7 +1357,8 @@ extension FetchAll {
     @available(iOS 17, macOS 14, tvOS 17, watchOS 10, *)
     @discardableResult
     public func load<
-      V: QueryRepresentable, From: StructuredQueriesCore.Table
+      V: QueryRepresentable,
+      From: StructuredQueriesCore.Table
     >(
       _ statement: Select<V, From, ()>,
       @_SectionBuilder<String?> sectionBy sectioning: (From.TableColumns) -> _Sectioning<String?>?,
@@ -1374,7 +1392,9 @@ extension FetchAll {
     @available(iOS 17, macOS 14, tvOS 17, watchOS 10, *)
     @discardableResult
     public func load<
-      V: QueryRepresentable, From: StructuredQueriesCore.Table, J: StructuredQueriesCore.Table
+      V: QueryRepresentable,
+      From: StructuredQueriesCore.Table,
+      J: StructuredQueriesCore.Table
     >(
       _ statement: Select<V, From, J>,
       @_SectionBuilder<String?> sectionBy sectioning: (From.TableColumns, J.TableColumns) ->

Sources/SQLiteData/FetchAll.swift

@@ -48,6 +48,8 @@ public struct FetchAll<Element: Sendable>: Sendable {
     private let box: FetchAllBox<Element>
     private let state: SwiftUI.State<FetchAllBox<Element>>
     private let generation = SwiftUI.State(wrappedValue: 0)
+
+    var loadGeneration: LoadGeneration { state.wrappedValue.loadGeneration }
   #else
     /// The underlying shared reader powering the property wrapper.
     ///
@@ -59,6 +61,8 @@ public struct FetchAll<Element: Sendable>: Sendable {
       SharedReader(value: ResultsSectionCollection())
 
     let sectioning = LockIsolated<_Sectioning<String?>?>(nil)
+
+    let loadGeneration = LoadGeneration()
   #endif
 
   /// A collection of data associated with the underlying query.
@@ -73,6 +77,7 @@ public struct FetchAll<Element: Sendable>: Sendable {
   public var projectedValue: Self {
     get { self }
     nonmutating set {
+      loadGeneration.invalidate()
       sharedReader.projectedValue = newValue.sharedReader.projectedValue
       sectionedReader.projectedValue = newValue.sectionedReader.projectedValue
       sectioning.setValue(newValue.sectioning.value)
@@ -247,13 +252,22 @@ public struct FetchAll<Element: Sendable>: Sendable {
     V.QueryOutput: Sendable
   {
     removeSections()
-    try await sharedReader.load(
-      .fetch(
-        FetchAllStatementValueRequest(statement: statement),
-        database: database
+    return try await withSubscription {
+      try await sharedReader.load(
+        .fetch(
+          FetchAllStatementValueRequest(statement: statement),
+          database: database
+        )
       )
-    )
-    return FetchSubscription(sharedReader: sharedReader)
+    }
+  }
+
+  func withSubscription(_ load: () async throws -> Void) async throws -> FetchSubscription {
+    let token = loadGeneration.begin()
+    try await load()
+    return sectioning.value == nil
+      ? FetchSubscription(sharedReader: sharedReader, token: token)
+      : FetchSubscription(sharedReader: sharedReader, sectionedReader: sectionedReader, token: token)
   }
 
   #if !canImport(SwiftUI)
@@ -435,14 +449,15 @@ extension FetchAll {
     V.QueryOutput: Sendable
   {
     removeSections()
-    try await sharedReader.load(
-      .fetch(
-        FetchAllStatementValueRequest(statement: statement),
-        database: database,
-        scheduler: scheduler
+    return try await withSubscription {
+      try await sharedReader.load(
+        .fetch(
+          FetchAllStatementValueRequest(statement: statement),
+          database: database,
+          scheduler: scheduler
+        )
       )
-    )
-    return FetchSubscription(sharedReader: sharedReader)
+    }
   }
 }
 
@@ -635,14 +650,15 @@ extension FetchAll: Equatable where Element: Equatable {
       V.QueryOutput: Sendable
     {
       removeSections()
-      try await sharedReader.load(
-        .fetch(
-          FetchAllStatementValueRequest(statement: statement),
-          database: database,
-          animation: animation
+      return try await withSubscription {
+        try await sharedReader.load(
+          .fetch(
+            FetchAllStatementValueRequest(statement: statement),
+            database: database,
+            animation: animation
+          )
         )
-      )
-      return FetchSubscription(sharedReader: sharedReader)
+      }
     }
   }
 #endif