From dca749e4b697d873cf6e1eecf42471c411443df9 Mon Sep 17 00:00:00 2001 From: Dimitri Bouniol Date: Tue, 28 Jul 2026 05:04:20 -0700 Subject: [PATCH 1/7] Added test showcasing that transactions could start at the same time --- .../DiskTransactionTests.swift | 158 ++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/Tests/CodableDatastoreTests/DiskTransactionTests.swift b/Tests/CodableDatastoreTests/DiskTransactionTests.swift index aff22d5..a56ead7 100644 --- a/Tests/CodableDatastoreTests/DiskTransactionTests.swift +++ b/Tests/CodableDatastoreTests/DiskTransactionTests.swift @@ -70,4 +70,162 @@ final class DiskTransactionTests: XCTestCase, @unchecked Sendable { XCTAssertEqual(existingDescriptor, descriptor) } } + + func testSerialMutableRootTransactions() async throws { + let persistence = try DiskPersistence(readWriteURL: temporaryStoreURL) + + struct TestFormat: DatastoreFormat { + enum Version: Int, CaseIterable { + case zero + } + + struct Instance: Codable {} + typealias Identifier = UUID + + static let defaultKey: DatastoreKey = "test" + static let currentVersion = Version.zero + } + + var tasks: [Task] = [] + + let iterations = 5000 + nonisolated(unsafe) var count = 0 + nonisolated(unsafe) var lastStartedIndex = 0 + for index in 0..] = [] + + let iterations = 5000 + nonisolated(unsafe) var count = 0 + for index in 0..] = [] + + let iterations = 5000 + nonisolated(unsafe) var count = 0 + nonisolated(unsafe) var lastStartedIndex = 0 + try await persistence._withTransaction(actionName: nil, options: []) { _, _ in + for index in 0..] = [] + + let iterations = 5000 + nonisolated(unsafe) var count = 0 + try await persistence._withTransaction(actionName: nil, options: []) { _, _ in + for index in 0.. Date: Fri, 24 Jul 2026 06:17:43 -0700 Subject: [PATCH 2/7] Added dependency for `QuestionableConcurrency` 0.2.1 --- Package.swift | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/Package.swift b/Package.swift index fbbefc2..67e037d 100644 --- a/Package.swift +++ b/Package.swift @@ -38,19 +38,26 @@ let package = Package( dependencies: [ .package(url: "https://github.com/mochidev/AsyncSequenceReader.git", .upToNextMinor(from: "0.5.0")), .package(url: "https://github.com/mochidev/Bytes.git", .upToNextMinor(from: "0.6.2")), + .package(url: "https://github.com/mochidev/swift-questionable-concurrency", .upToNextMinor(from: "0.2.1")), ], targets: [ .target( name: "CodableDatastore", dependencies: [ "AsyncSequenceReader", - "Bytes" + "Bytes", + .product(name: "QuestionableConcurrency", package: "swift-questionable-concurrency") ], swiftSettings: swiftSettings ), .testTarget( name: "CodableDatastoreTests", - dependencies: ["CodableDatastore"], + dependencies: [ + "AsyncSequenceReader", + "CodableDatastore", + "Bytes", + .product(name: "QuestionableConcurrency", package: "swift-questionable-concurrency") + ], swiftSettings: swiftSettings ), ] From cdae54e59d324a54dbb8535f3729977dcb4a83ab Mon Sep 17 00:00:00 2001 From: Dimitri Bouniol Date: Mon, 27 Jul 2026 03:51:17 -0700 Subject: [PATCH 3/7] Replaced snapshot manifest locking with a transaction stream --- Package.swift | 2 +- .../Disk Persistence/Snapshot/Snapshot.swift | 102 +++++------------- .../CodableDatastoreTests/SnapshotTests.swift | 7 +- 3 files changed, 30 insertions(+), 81 deletions(-) diff --git a/Package.swift b/Package.swift index 67e037d..4aaac5c 100644 --- a/Package.swift +++ b/Package.swift @@ -38,7 +38,7 @@ let package = Package( dependencies: [ .package(url: "https://github.com/mochidev/AsyncSequenceReader.git", .upToNextMinor(from: "0.5.0")), .package(url: "https://github.com/mochidev/Bytes.git", .upToNextMinor(from: "0.6.2")), - .package(url: "https://github.com/mochidev/swift-questionable-concurrency", .upToNextMinor(from: "0.2.1")), + .package(url: "https://github.com/mochidev/swift-questionable-concurrency", .upToNextMinor(from: "0.2.2")), ], targets: [ .target( diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift index 55f93c2..bdd0839 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift @@ -8,6 +8,7 @@ // import Foundation +import QuestionableConcurrency typealias SnapshotIdentifier = DatedIdentifier> @@ -34,8 +35,8 @@ actor Snapshot { /// A cached instance of the current iteration as last loaded from disk. var cachedIteration: SnapshotIteration? - /// A pointer to the last manifest updater, so updates can be serialized after the last request. - var lastUpdateManifestTask: Task? + /// A transaction stream for manifest updates, so reads and writes can be serialized in request order. + var manifestTransactionStream = TransactionStream() /// The loaded datastores. var datastores: [DatastoreIdentifier: DiskPersistence.Datastore] = [:] @@ -170,35 +171,28 @@ extension Snapshot { cachedIteration = iteration } - /// Load and update the manifest in an updater, returning the task for the updater. + /// Load and update the manifest in an updater. /// /// This method loads the ``SnapshotManifest`` from cache, offers it to be mutated, then writes it back to disk, if it changed. It is up to the caller to update the modification date of the store. /// /// - Note: Calling this method when no manifest exists on disk will create it, even if no changes occur in the block. /// - Parameter updater: An updater that takes a mutable reference to a manifest, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateManifest( - updater: @Sendable @escaping (_ manifest: inout SnapshotManifest, _ iteration: inout SnapshotIteration) async throws -> T - ) -> Task where AccessMode == ReadWrite { + /// - Returns: The value returned from the `updater`. + func updatingManifest( + updater: (_ manifest: inout SnapshotManifest, _ iteration: inout SnapshotIteration) async throws -> T + ) async throws -> T where AccessMode == ReadWrite { if let (manifest, iteration) = SnapshotTaskLocals.manifest(for: persistence) { - return Task { - var updatedManifest = manifest - var updatedIteration = iteration - let returnValue = try await updater(&updatedManifest, &updatedIteration) - - guard updatedManifest == manifest, updatedIteration == iteration - else { throw DiskPersistenceInternalError.nestedSnapshotWrite } - - return returnValue - } + var updatedManifest = manifest + var updatedIteration = iteration + let returnValue = try await updater(&updatedManifest, &updatedIteration) + + guard updatedManifest == manifest, updatedIteration == iteration + else { throw DiskPersistenceInternalError.nestedSnapshotWrite } + + return returnValue } - /// Grab the last task so we can chain off of it in a serial manner. - let lastUpdaterTask = lastUpdateManifestTask - let updaterTask = Task { - /// We don't care if the last request throws an error or not, but we do want it to complete first. - _ = try? await lastUpdaterTask?.value - + return try await manifestTransactionStream.withTransaction { /// Load the manifest so we have a fresh copy, unless we have a cached copy already. var manifest = try cachedManifest ?? self.loadManifest() var iteration: SnapshotIteration @@ -233,32 +227,23 @@ extension Snapshot { } return returnValue } - /// Assign the task to our pointer so we can depend on it the next time. Also, re-wrap it so we can keep proper type information when returning from this method. - lastUpdateManifestTask = Task { try await updaterTask.value } - - return updaterTask } - /// Load the manifest in an accessor, returning the task for the updater. + /// Load the manifest in an updater. /// /// This method loads the ``SnapshotManifest`` from cache. /// /// - Parameter accessor: An accessor that takes an immutable reference to a manifest, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func readManifest( - accessor: @Sendable @escaping (_ manifest: SnapshotManifest, _ iteration: SnapshotIteration) async throws -> T - ) -> Task { - + /// - Returns: The value returned from the `accessor`. + @_disfavoredOverload + func readingManifest( + accessor: (_ manifest: SnapshotManifest, _ iteration: SnapshotIteration) async throws -> T + ) async throws -> T { if let (manifest, iteration) = SnapshotTaskLocals.manifest(for: persistence) { - return Task { try await accessor(manifest, iteration) } + return try await accessor(manifest, iteration) } - /// Grab the last task so we can chain off of it in a serial manner. - let lastUpdaterTask = lastUpdateManifestTask - let readerTask = Task { - /// We don't care if the last request throws an error or not, but we do want it to complete first. - _ = try? await lastUpdaterTask?.value - + return try await manifestTransactionStream.withTransaction { /// Load the manifest so we have a fresh copy, unless we have a cached copy already. let manifest = try cachedManifest ?? self.loadManifest() var iteration: SnapshotIteration @@ -276,40 +261,6 @@ extension Snapshot { try await accessor(manifest, iteration) } } - /// Assign the task to our pointer so we can depend on it the next time. Also, re-wrap it so we can keep proper type information when returning from this method. - lastUpdateManifestTask = Task { try await readerTask.value } - - return readerTask - } - - /// Load and update the manifest in an updater. - /// - /// This method loads the ``SnapshotManifest`` from cache, offers it to be mutated, then writes it back to disk, if it changed. It is up to the caller to update the modification date of the store. - /// - /// - Note: Calling this method when no manifest exists on disk will create it, even if no changes occur in the block. - /// - Parameter updater: An updater that takes a mutable reference to a manifest, and will forward the returned value to the caller. - /// - Returns: The value returned from the `updater`. - func updatingManifest( - updater: @Sendable (_ manifest: inout SnapshotManifest, _ iteration: inout SnapshotIteration) async throws -> T - ) async throws -> T where AccessMode == ReadWrite { - try await withoutActuallyEscaping(updater) { escapingClosure in - try await updateManifest(updater: escapingClosure).value - } - } - - /// Load the manifest in an updater. - /// - /// This method loads the ``SnapshotManifest`` from cache. - /// - /// - Parameter accessor: An accessor that takes an immutable reference to a manifest, and will forward the returned value to the caller. - /// - Returns: The value returned from the `accessor`. - @_disfavoredOverload - func readingManifest( - accessor: @Sendable (_ manifest: SnapshotManifest, _ iteration: SnapshotIteration) async throws -> T - ) async throws -> T { - try await withoutActuallyEscaping(accessor) { escapingClosure in - try await readManifest(accessor: escapingClosure).value - } } } @@ -322,6 +273,7 @@ private enum SnapshotTaskLocals { } static func with( + isolation actor: isolated (any Actor)? = #isolation, manifest: SnapshotManifest, iteration: SnapshotIteration, for persistence: DiskPersistence, @@ -389,7 +341,7 @@ extension Snapshot { /// Iterate through each datastore and copy the data over for (_, datastoreInfo) in iteration.dataStores { - let (datastore, _) = await loadDatastore(for: datastoreInfo.key, from: iteration) + let (datastore, _) = loadDatastore(for: datastoreInfo.key, from: iteration) try await datastore.copy( rootIdentifier: datastoreInfo.root, datastoreKey: datastoreInfo.key, diff --git a/Tests/CodableDatastoreTests/SnapshotTests.swift b/Tests/CodableDatastoreTests/SnapshotTests.swift index 2b9b2c0..3cf9ee5 100644 --- a/Tests/CodableDatastoreTests/SnapshotTests.swift +++ b/Tests/CodableDatastoreTests/SnapshotTests.swift @@ -143,20 +143,17 @@ final class SnapshotTests: XCTestCase, @unchecked Sendable { let date1 = Date(timeIntervalSince1970: 0) let date2 = Date(timeIntervalSince1970: 10) - let task1 = await snapshot.updateManifest { manifest, _ in + try await snapshot.updatingManifest { manifest, _ in sleep(1) XCTAssertNotEqual(manifest.modificationDate, date2) manifest.modificationDate = date1 } - let task2 = await snapshot.updateManifest { manifest, _ in + try await snapshot.updatingManifest { manifest, _ in XCTAssertEqual(manifest.modificationDate, date1) manifest.modificationDate = date2 } - try await task1.value - try await task2.value - let currentManifest = await snapshot.cachedManifest XCTAssertEqual(currentManifest?.modificationDate, date2) } From ac21d71e7e60b50207c29804660fed4d1aa6245f Mon Sep 17 00:00:00 2001 From: Dimitri Bouniol Date: Mon, 27 Jul 2026 05:11:01 -0700 Subject: [PATCH 4/7] Replaced store info locking with a transaction stream --- .../Datastore/PersistenceDatastore.swift | 2 - .../Disk Persistence/DiskPersistence.swift | 158 +++++------------- .../DiskPersistenceTests.swift | 7 +- 3 files changed, 43 insertions(+), 124 deletions(-) diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift index 45f22b9..d3030d3 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift @@ -27,8 +27,6 @@ extension DiskPersistence { var cachedRootObject: DatastoreRootManifest? - var lastUpdateDescriptorTask: Task? - /// The root objects that are being tracked in memory. var trackedRootObjects: [RootObject.ID : WeakValue] = [:] var trackedIndexes: [Index.ID : WeakValue] = [:] diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift index 58ae78f..23f5e20 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift @@ -8,6 +8,7 @@ // public import Foundation +import QuestionableConcurrency public actor DiskPersistence: Persistence { /// The location of this persistence. @@ -16,8 +17,8 @@ public actor DiskPersistence: Persistence { /// A cached instance of the store info as last loaded from disk. var cachedStoreInfo: StoreInfo? - /// A pointer to the last store info updater, so updates can be serialized after the last request - var lastUpdateStoreInfoTask: Task? + /// A transaction stream for storeInfo updates, so reads and writes can be serialized in request order. + var storeInfoTransactionStream = TransactionStream() /// The loaded Snapshots var snapshots: [SnapshotIdentifier: Snapshot] = [:] @@ -166,35 +167,28 @@ extension DiskPersistence { cachedStoreInfo = storeInfo } - /// Load and update the store info in an updater, returning the task for the updater. + /// Load and update the store info in an updater. /// /// This method loads the ``StoreInfo`` from cache, offers it to be mutated, then writes it back to disk, if it changed. It is up to the caller to update the modification date of the store. /// /// - Note: Calling this method when no store info exists on disk will create it, even if no changes occur in the block. /// - Parameter updater: An updater that takes a mutable reference to a store info, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateStoreInfo( - @_inheritActorContext updater: @Sendable @escaping (_ storeInfo: inout StoreInfo) async throws -> T - ) -> Task where AccessMode == ReadWrite { + /// - Returns: The value returned from the `updater`. + func withStoreInfo( + updater: (_ storeInfo: inout StoreInfo) async throws -> T + ) async throws -> T where AccessMode == ReadWrite { if let storeInfo = DiskPersistenceTaskLocals.storeInfo(for: self) { - return Task { - var updatedStoreInfo = storeInfo - let returnValue = try await updater(&updatedStoreInfo) - - guard updatedStoreInfo == storeInfo else { - throw DiskPersistenceInternalError.nestedStoreWrite - } - - return returnValue + var updatedStoreInfo = storeInfo + let returnValue = try await updater(&updatedStoreInfo) + + guard updatedStoreInfo == storeInfo else { + throw DiskPersistenceInternalError.nestedStoreWrite } + + return returnValue } - /// Grab the last task so we can chain off of it in a serial manner. - let lastUpdaterTask = lastUpdateStoreInfoTask - let updaterTask = Task { - /// We don't care if the last request throws an error or not, but we do want it to complete first. - _ = try? await lastUpdaterTask?.value - + return try await storeInfoTransactionStream.withTransaction { /// Load the store info so we have a fresh copy, unless we have a cached copy already. var storeInfo = try cachedStoreInfo ?? self.loadStoreInfo() @@ -209,31 +203,23 @@ extension DiskPersistence { } return returnValue } - /// Assign the task to our pointer so we can depend on it the next time. Also, re-wrap it so we can keep proper type information when returning from this method. - lastUpdateStoreInfoTask = Task { try await updaterTask.value } - - return updaterTask } - /// Load the store info in an accessor, returning the task for the updater. + /// Load the store info in an updater. /// /// This method loads the ``StoreInfo`` from cache. /// /// - Parameter accessor: An accessor that takes an immutable reference to a store info, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateStoreInfo( - @_inheritActorContext accessor: @Sendable @escaping (_ storeInfo: StoreInfo) async throws -> T - ) -> Task { + /// - Returns: The value returned from the `accessor`. + @_disfavoredOverload + func withStoreInfo( + accessor: (_ storeInfo: StoreInfo) async throws -> T + ) async throws -> T { if let storeInfo = DiskPersistenceTaskLocals.storeInfo(for: self) { - return Task { try await accessor(storeInfo) } + return try await accessor(storeInfo) } - /// Grab the last task so we can chain off of it in a serial manner. - let lastUpdaterTask = lastUpdateStoreInfoTask - let updaterTask = Task { - /// We don't care if the last request throws an error or not, but we do want it to complete first. - _ = try? await lastUpdaterTask?.value - + return try await storeInfoTransactionStream.withTransaction { /// Load the store info so we have a fresh copy, unless we have a cached copy already. let storeInfo = try cachedStoreInfo ?? self.loadStoreInfo() @@ -242,40 +228,6 @@ extension DiskPersistence { try await accessor(storeInfo) } } - /// Assign the task to our pointer so we can depend on it the next time. Also, re-wrap it so we can keep proper type information when returning from this method. - lastUpdateStoreInfoTask = Task { try await updaterTask.value } - - return updaterTask - } - - /// Load and update the store info in an updater. - /// - /// This method loads the ``StoreInfo`` from cache, offers it to be mutated, then writes it back to disk, if it changed. It is up to the caller to update the modification date of the store. - /// - /// - Note: Calling this method when no store info exists on disk will create it, even if no changes occur in the block. - /// - Parameter updater: An updater that takes a mutable reference to a store info, and will forward the returned value to the caller. - /// - Returns: The value returned from the `updater`. - func withStoreInfo( - updater: @Sendable (_ storeInfo: inout StoreInfo) async throws -> T - ) async throws -> T where AccessMode == ReadWrite { - try await withoutActuallyEscaping(updater) { escapingClosure in - try await updateStoreInfo(updater: escapingClosure).value - } - } - - /// Load the store info in an updater. - /// - /// This method loads the ``StoreInfo`` from cache. - /// - /// - Parameter accessor: An accessor that takes an immutable reference to a store info, and will forward the returned value to the caller. - /// - Returns: The value returned from the `accessor`. - @_disfavoredOverload - func withStoreInfo( - accessor: @Sendable (_ storeInfo: StoreInfo) async throws -> T - ) async throws -> T { - try await withoutActuallyEscaping(accessor) { escapingClosure in - try await updateStoreInfo(accessor: escapingClosure).value - } } } @@ -304,22 +256,22 @@ extension DiskPersistence { return snapshot } - /// Load and update the current snapshot in an updater, returning the task for the updater. + /// Load the current snapshot in an updater. /// /// This method loads the current ``Snapshot`` so it can be updated. /// /// - Note: Calling this method when no store info exists on disk will create it. /// - Parameter dateUpdate: The method to which to update the date of the main store with. /// - Parameter updater: An updater that takes a reference to the current ``Snapshot``, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateCurrentSnapshot( + /// - Returns: The value returned from the `accessor`. + func updatingCurrentSnapshot( dateUpdate: ModificationUpdate = .updateOnWrite, - updater: @escaping (_ snapshot: Snapshot) async throws -> T - ) -> Task where AccessMode == ReadWrite { + updater: sending (_ snapshot: Snapshot) async throws -> T + ) async throws -> T where AccessMode == ReadWrite { /// Grab access to the store info to load and update it. - return updateStoreInfo { storeInfo in + try await withStoreInfo { storeInfo in /// Grab the current snapshot from the store info - let snapshot = self.loadSnapshot(from: storeInfo) + let snapshot = loadSnapshot(from: storeInfo) /// Load a modification date to use let modificationDate = dateUpdate.modificationDate(for: storeInfo.modificationDate) @@ -335,40 +287,6 @@ extension DiskPersistence { } } - /// Load the current snapshot in an accessor, returning the task for the accessor. - /// - /// - Parameter accessor: An accessor that takes a reference to the current ``Snapshot``, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateCurrentSnapshot( - accessor: @escaping (_ snapshot: Snapshot) async throws -> T - ) -> Task { - /// Grab access to the store info to load and update it. - return updateStoreInfo { storeInfo in - /// Grab the current snapshot from the store info - let snapshot = self.loadSnapshot(from: storeInfo) - - /// Let the accessor do what it needs to do with the snapshot - return try await accessor(snapshot) - } - } - - /// Load the current snapshot in an updater. - /// - /// This method loads the current ``Snapshot`` so it can be updated. - /// - /// - Note: Calling this method when no store info exists on disk will create it. - /// - Parameter dateUpdate: The method to which to update the date of the main store with. - /// - Parameter updater: An updater that takes a reference to the current ``Snapshot``, and will forward the returned value to the caller. - /// - Returns: The value returned from the `accessor`. - func updatingCurrentSnapshot( - dateUpdate: ModificationUpdate = .updateOnWrite, - updater: @Sendable (_ snapshot: Snapshot) async throws -> T - ) async throws -> T where AccessMode == ReadWrite { - try await withoutActuallyEscaping(updater) { escapingClosure in - try await updateCurrentSnapshot(dateUpdate: dateUpdate, updater: escapingClosure).value - } - } - /// Load the current snapshot in an accessor. /// /// This method loads the current ``Snapshot`` so it can be accessed. @@ -379,10 +297,15 @@ extension DiskPersistence { /// - Returns: The value returned from the `accessor`. @_disfavoredOverload func readingCurrentSnapshot( - accessor: @Sendable (_ snapshot: Snapshot) async throws -> T + accessor: sending @Sendable (_ snapshot: Snapshot) async throws -> T ) async throws -> T { - try await withoutActuallyEscaping(accessor) { escapingClosure in - try await updateCurrentSnapshot(accessor: escapingClosure).value + /// Grab access to the store info to load and update it. + try await withStoreInfo { storeInfo in + /// Grab the current snapshot from the store info + let snapshot = self.loadSnapshot(from: storeInfo) + + /// Let the accessor do what it needs to do with the snapshot + return try await accessor(snapshot) } } @@ -405,7 +328,7 @@ extension DiskPersistence { var currentSnapshot: Snapshot { // TODO: This should return a readonly snapshot, but we need to be able to make a read-only copy from the persistence first. get async throws { - try await withStoreInfo { await loadSnapshot(from: $0) } + try await withStoreInfo { loadSnapshot(from: $0) } } } @@ -726,6 +649,7 @@ private enum DiskPersistenceTaskLocals { } static func with( + isolation actor: isolated (any Actor)? = #isolation, storeInfo: StoreInfo, for persistence: DiskPersistence, operation: () async throws -> R diff --git a/Tests/CodableDatastoreTests/DiskPersistenceTests.swift b/Tests/CodableDatastoreTests/DiskPersistenceTests.swift index 60a8a64..590eff6 100644 --- a/Tests/CodableDatastoreTests/DiskPersistenceTests.swift +++ b/Tests/CodableDatastoreTests/DiskPersistenceTests.swift @@ -131,20 +131,17 @@ final class DiskPersistenceTests: XCTestCase, @unchecked Sendable { let date1 = Date(timeIntervalSince1970: 0) let date2 = Date(timeIntervalSince1970: 10) - let task1 = await persistence.updateStoreInfo { storeInfo in + try await persistence.withStoreInfo { storeInfo in sleep(1) XCTAssertNotEqual(storeInfo.modificationDate, date2) storeInfo.modificationDate = date1 } - let task2 = await persistence.updateStoreInfo { storeInfo in + try await persistence.withStoreInfo { storeInfo in XCTAssertEqual(storeInfo.modificationDate, date1) storeInfo.modificationDate = date2 } - try await task1.value - try await task2.value - let currentStoreInfo = await persistence.cachedStoreInfo XCTAssertEqual(currentStoreInfo?.modificationDate, date2) } From b457b9ce6a6ad79f2667d0a601aed80f70062ea2 Mon Sep 17 00:00:00 2001 From: Dimitri Bouniol Date: Tue, 28 Jul 2026 03:12:15 -0700 Subject: [PATCH 5/7] Fixed an issue where transactions could conflict and cause data loss if created at the exact same time --- .../Datastore/Datastore.swift | 10 +- .../Disk Persistence/DiskPersistence.swift | 68 +++++-- .../Transaction/Transaction.swift | 185 +++++++++--------- .../MemoryPersistence.swift | 2 +- .../Persistence/Persistence.swift | 2 +- 5 files changed, 150 insertions(+), 117 deletions(-) diff --git a/Sources/CodableDatastore/Datastore/Datastore.swift b/Sources/CodableDatastore/Datastore/Datastore.swift index a16410b..2c87f25 100644 --- a/Sources/CodableDatastore/Datastore/Datastore.swift +++ b/Sources/CodableDatastore/Datastore/Datastore.swift @@ -125,7 +125,7 @@ extension Datastore { return descriptor } - func decoder(for version: Version) throws -> @Sendable (_ data: Data) async throws -> (id: IdentifierType, instance: InstanceType) { + nonisolated func decoder(for version: Version) throws -> @Sendable (_ data: Data) async throws -> (id: IdentifierType, instance: InstanceType) { guard let decoder = decoders[version] else { throw DatastoreError.missingDecoder(version: String(describing: version)) } @@ -533,7 +533,7 @@ extension Datastore { datastoreKey: self.key ) { versionData, instanceData in let entryVersion = try Version(versionData) - let decoder = try await self.decoder(for: entryVersion) + let decoder = try self.decoder(for: entryVersion) let decodedValue = try await decoder(instanceData) try await provider.yield(decodedValue) @@ -676,7 +676,7 @@ extension Datastore { datastoreKey: self.key ) { versionData, instanceData in let entryVersion = try Version(versionData) - let decoder = try await self.decoder(for: entryVersion) + let decoder = try self.decoder(for: entryVersion) let instance = try await decoder(instanceData).instance try await provider.yield(instance) @@ -690,7 +690,7 @@ extension Datastore { let persistedEntry = try await transaction.primaryIndexCursor(for: identifier, datastoreKey: self.key) let entryVersion = try Version(persistedEntry.versionData) - let decoder = try await self.decoder(for: entryVersion) + let decoder = try self.decoder(for: entryVersion) let instance = try await decoder(persistedEntry.instanceData).instance try await provider.yield(instance) @@ -953,7 +953,7 @@ extension Datastore { }.compactMap { event in try? await event.mapEntries { entry in let version = try Version(entry.versionData) - let decoder = try await self.decoder(for: version) + let decoder = try self.decoder(for: version) let instance = try await decoder(entry.instanceData).instance return instance } diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift index 23f5e20..c26a623 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift @@ -25,7 +25,8 @@ public actor DiskPersistence: Persistence { var registeredDatastores: [DatastoreKey : [WeakDatastore]] = [:] - var lastTransaction: Transaction? + var lastMutatingTransaction: Transaction? + var rootTransactionStream = TransactionStream() /// Shared caches across all snapshots and datastores. var rollingRootObjectCacheIndex = 0 @@ -472,39 +473,64 @@ extension DiskPersistence { public func _withTransaction( actionName: String?, options: UnsafeTransactionOptions, - transaction: @Sendable (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + transaction operation: sending (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T { - try await withoutActuallyEscaping(transaction) { escapingTransaction in - /// If the transaction is starting in the context of another persistence's transaction, make sure it is a read-only one. Otherwise assert and throw an error as it likely indicates a mistake and could lead to unexpected consistency violations if one persistence succeeds while the other fails. - if - Transaction.isTransactingExternally(to: self), - !options.contains(.readOnly) - { + /// If the transaction is starting in the context of another persistence's transaction, make sure it is a read-only one. Otherwise assert and throw an error as it likely indicates a mistake and could lead to unexpected consistency violations if one persistence succeeds while the other fails. + if isTransactingExternally() { + guard options.contains(.readOnly) + else { assertionFailure(DatastoreInterfaceError.transactingWithinExternalPersistence.localizedDescription) throw DatastoreInterfaceError.transactingWithinExternalPersistence } - - let currentCounter = nextTransactionCounter() -// print("[CDS] [\(storeURL.lastPathComponent)] Starting transaction \(currentCounter) “\(actionName ?? "")” - \(options)") - let (transaction, task) = await Transaction.makeTransaction( + } + + let currentCounter = nextTransactionCounter() +// print("[CDS] [\(storeURL.lastPathComponent)] Starting transaction \(currentCounter) “\(actionName ?? "")” - \(options)") + + if let parent = Transaction.unsafeCurrentTransaction(for: self) { +// print("[CDS] [\(storeURL.lastPathComponent)] Found parent \(parent.transactionIndex), making child \(currentCounter)") + assert(!parent.options.contains(.readOnly) || options.contains(.readOnly), "A child transaction was declared read-write, even though its parent was read-only!") + let transaction = Transaction( persistence: self, transactionIndex: currentCounter, - lastTransaction: lastTransaction, + parent: parent, actionName: actionName, options: options - ) { interface, isDurable in - try await escapingTransaction(interface, isDurable) - } + ) - /// Save the last non-concurrent top-level transaction from the list. Note that disk persistence currently does not support concurrent idempotent transactions. - if !options.contains(.readOnly), transaction.parent == nil { - lastTransaction = transaction - } + /// Enqueue and get the last non-concurrent transaction from the parent's list. Note that disk persistence currently does not support concurrent idempotent transactions. + let lastChildTransaction = try await parent.enqueue(childTransaction: transaction) - let result = try await task.value + let result = try await transaction.run( + lastTransaction: !options.contains(.readOnly) ? lastChildTransaction : nil, + isDurable: false, + operation: operation + ) // print("[CDS] [\(storeURL.lastPathComponent)] Finished transaction \(currentCounter) “\(actionName ?? "")” - \(options)") return result } + + let transaction = Transaction( + persistence: self, + transactionIndex: currentCounter, + parent: nil, + actionName: actionName, + options: options + ) + + /// Save the last non-concurrent root transaction from the list. Note that disk persistence currently does not support concurrent idempotent transactions. + let lastTransaction = lastMutatingTransaction + if !options.contains(.readOnly) { + self.lastMutatingTransaction = transaction + } + + let result = try await transaction.run( + lastTransaction: !options.contains(.readOnly) ? lastTransaction : nil, + isDurable: options.isDisjoint(with: [.collateWrites, .readOnly]), + operation: operation + ) +// print("[CDS] [\(storeURL.lastPathComponent)] Finished transaction \(currentCounter) “\(actionName ?? "")” - \(options)") + return result } func persist( diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift index 50320a6..aa9be3f 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift @@ -7,15 +7,21 @@ // mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 // -import Foundation import Bytes +import Foundation +import QuestionableConcurrency extension DiskPersistence { actor Transaction: AnyDiskTransaction { unowned let persistence: DiskPersistence unowned let parent: Transaction? - weak var lastReadWriteChildTransaction: Transaction? + weak var lastMutatingChildTransaction: Transaction? + + var workDidFinishPromise: Promise? + var transactionDidPersistPromise: Promise? + let workDidFinishResult: AsyncResult + let transactionDidPersistResult: AsyncResult private(set) var task: Task! let transactionIndex: Int @@ -37,7 +43,7 @@ extension DiskPersistence { var isActive = false - private init( + init( persistence: DiskPersistence, transactionIndex: Int, parent: Transaction?, @@ -49,39 +55,85 @@ extension DiskPersistence { self.actionName = actionName self.options = options self.transactionIndex = transactionIndex + + let workDidFinishPromise = Promise(name: "DiskPersistence.Transaction.workDidFinish - \"\(actionName ?? "")\"") + let transactionDidPersistPromise = Promise(name: "DiskPersistence.Transaction.transactionDidPersist - \"\(actionName ?? "")\"") + + self.workDidFinishResult = workDidFinishPromise.future + self.transactionDidPersistResult = transactionDidPersistPromise.future + self.workDidFinishPromise = consume workDidFinishPromise + self.transactionDidPersistPromise = consume transactionDidPersistPromise } - private func attachTask( - options: UnsafeTransactionOptions, - @_inheritActorContext handler: @Sendable @escaping () async throws -> T - ) async -> Task { - let task = Task { + /// An internal method to attach an operation to the transaction. + /// - Warning: Do not call this method more than once for a single transaction. + /// - Parameters: + /// - lastTransaction: The last transaction this one should be chained to. + /// - isDurable: A flag that reports whether or not the operation was durable. + /// - operation: The operation itself, which takes a reference to the owning transaction and the `isDurable` flag + /// - Returns: The return value of the operation, if any. + func run( + lastTransaction: Transaction?, + isDurable: Bool, + operation: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + ) async throws -> T { + guard let workDidFinishPromise = workDidFinishPromise.take() + else { fatalError("workDidFinishPromise was already consumed. Please report this as a critical bug!") } + + let returnValue: T + do { isActive = true - let returnValue = try await TransactionTaskLocals.with(transaction: self, for: persistence) { - try await handler() + + /// If provided, wait for the last transaction to properly finish before startign the next one. + if let lastTransaction { + await lastTransaction.yieldPersistenceCompletion() } - isActive = false - /// If we don't care to collate our writes, go ahead and wait for the persistence to stick - if !options.contains(.collateWrites) { - try await self.persist() + /// Perform the operation with the current transaction set as the task-local. + returnValue = try await TransactionTaskLocals.with(transaction: self, for: persistence) { + try await operation(self, isDurable) } - return returnValue + isActive = false + + /// Mark the work as done so opportunistic transaction may begin. + workDidFinishPromise.resume() + } catch { + /// If the operation failed, make sure to resume both continuations so the next transaction can start. + workDidFinishPromise.resume() + + guard let transactionDidPersistPromise = transactionDidPersistPromise.take() + else { fatalError("transactionDidPersistPromise was already consumed. Please report this as a critical bug!") } + transactionDidPersistPromise.resume() + + throw error } - self.task = Task { - _ = try await task.value - - /// If we previously skipped persisting, go ahead and do so now. - if options.contains(.collateWrites) { + /// Persist the work that was just completed to signal to the next transaction that it can start, but check for the requested timing option first. ``persist()`` takes care of signaling when the persistence is finished. + if options.contains(.collateWrites) { + /// If we are skipping immediate writes, kick off persistence in a separate task. + Task { try await self.persist() } + } else { + /// If we don't care to collate our writes, go ahead and wait for the persistence to stick + try await self.persist() } - return task + return returnValue + } + + /// Yield and wait for the main work of the transaction to be complete. + func yieldWorkCompletion() async { + await workDidFinishResult.yield() + } + + /// Yield and wait for the transaction to finish persisting, which is the ideal point to start another mutating transaction. + func yieldPersistenceCompletion() async { + await transactionDidPersistResult.yield() } + /// Ensure the receiving transaction is currenrtly active. func checkIsActive() throws { guard isActive else { assertionFailure(DatastoreInterfaceError.transactionInactive.localizedDescription) @@ -137,6 +189,10 @@ extension DiskPersistence { deletedRootObjects.removeAll() deletedIndexes.removeAll() deletedPages.removeAll() + + guard let transactionDidPersistPromise = transactionDidPersistPromise.take() + else { fatalError("transactionDidPersistPromise was already consumed. Please report this as a critical bug!") } + transactionDidPersistPromise.resume() } /// If the transaction is read-only, stop here without applying anything to the parent. @@ -203,76 +259,17 @@ extension DiskPersistence { } } - static func makeTransaction( - persistence: DiskPersistence, - transactionIndex: Int, - lastTransaction: Transaction?, - actionName: String?, - options: UnsafeTransactionOptions, - @_inheritActorContext handler: @Sendable @escaping (_ transaction: Transaction, _ isDurable: Bool) async throws -> T - ) async -> (Transaction, Task) { - if let parent = Self.unsafeCurrentTransaction(for: persistence) { -// print("[CDS] [\(persistence.storeURL.lastPathComponent)] Found parent \(parent.transactionIndex), making child \(transactionIndex)") - let (child, task) = await parent.childTransaction( - transactionIndex: transactionIndex, - actionName: actionName, - options: options, - handler: handler - ) - return (child, task) - } - - let transaction = Transaction( - persistence: persistence, - transactionIndex: transactionIndex, - parent: nil, - actionName: actionName, - options: options - ) - - let task = await transaction.attachTask(options: options) { - /// If the transaction is not read only, wait for the last transaction to properly finish before starting the next one. - if !options.contains(.readOnly) { - try? await lastTransaction?.task.value - } - return try await handler(transaction, options.isDisjoint(with: [.collateWrites, .readOnly])) - } - - return (transaction, task) - } - - func childTransaction( - transactionIndex: Int, - actionName: String?, - options: UnsafeTransactionOptions, - @_inheritActorContext handler: @Sendable @escaping (_ transaction: Transaction, _ isDurable: Bool) async throws -> T - ) async -> (Transaction, Task) { - assert(!self.options.contains(.readOnly) || options.contains(.readOnly), "A child transaction was declared read-write, even though its parent was read-only!") - let childTransaction = Transaction( - persistence: persistence, - transactionIndex: transactionIndex, - parent: self, - actionName: actionName, - options: options - ) + /// Replace the last mutating child transaction with the specified transaction, and return the previous one. + func enqueue(childTransaction: Transaction) throws -> Transaction? { + /// Make sure the parent transaction actually started, as we'll be dumping state into it as new child transactions complete. + try checkIsActive() - /// Get the last non-concurrent transaction from the list. Note that disk persistence currently does not support concurrent idempotent transactions. - let lastChild = lastReadWriteChildTransaction + /// Enqueue and get the last non-concurrent transaction from the parent's list. Note that disk persistence currently does not support concurrent idempotent transactions. + let lastChildTransaction = lastMutatingChildTransaction if !childTransaction.options.contains(.readOnly) { - lastReadWriteChildTransaction = childTransaction - } - - let task = await childTransaction.attachTask(options: options) { - try self.checkIsActive() - - /// If the transaction is not read only, wait for the last transaction to properly finish before starting the next one. - if !options.contains(.readOnly) { - _ = try? await lastChild?.task.value - } - return try await handler(childTransaction, false) + lastMutatingChildTransaction = childTransaction } - - return (childTransaction, task) + return lastChildTransaction } func rootObject(for datastoreKey: DatastoreKey) async throws -> Datastore.RootObject? { @@ -1370,7 +1367,16 @@ extension DiskPersistence.Transaction { } } -// MARK: - Helper Types +// MARK: - Task Locals + +extension DiskPersistence { + /// Check to see if we are in a transaction that pertains to a different persistence than the one provided. + /// + /// This is determined by checking if the only transactions present belong to other persistences. If there are no other transactions, or a transaction for the persistence is already registered, it has already been vetted and is no longer considered external. + nonisolated func isTransactingExternally() -> Bool { + !TransactionTaskLocals.transactionStorage.isEmpty && TransactionTaskLocals.transactionStorage[ObjectIdentifier(self)] == nil + } +} fileprivate protocol AnyDiskTransaction: Sendable {} @@ -1383,6 +1389,7 @@ fileprivate enum TransactionTaskLocals { } static func with( + isolation actor: isolated (any Actor)? = #isolation, transaction: any AnyDiskTransaction, for persistence: DiskPersistence, operation: () async throws -> R diff --git a/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift b/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift index e1a94cc..53499c6 100644 --- a/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift +++ b/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift @@ -17,7 +17,7 @@ extension MemoryPersistence { public func _withTransaction( actionName: String?, options: UnsafeTransactionOptions, - transaction: @Sendable (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + transaction: sending (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T { preconditionFailure("Unimplemented") } diff --git a/Sources/CodableDatastore/Persistence/Persistence.swift b/Sources/CodableDatastore/Persistence/Persistence.swift index c80d41d..d3aaca8 100644 --- a/Sources/CodableDatastore/Persistence/Persistence.swift +++ b/Sources/CodableDatastore/Persistence/Persistence.swift @@ -21,7 +21,7 @@ public protocol Persistence: Sendable { func _withTransaction( actionName: String?, options: UnsafeTransactionOptions, - @_inheritActorContext transaction: @Sendable (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + transaction: sending (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T } From 6667ea70719131e87b6ade0277b4d4b2bfe5fcdc Mon Sep 17 00:00:00 2001 From: Dimitri Bouniol Date: Tue, 28 Jul 2026 03:15:35 -0700 Subject: [PATCH 6/7] Added support for `NonisolatedNonsendingByDefault` --- Package.swift | 1 + 1 file changed, 1 insertion(+) diff --git a/Package.swift b/Package.swift index 4aaac5c..d3e596a 100644 --- a/Package.swift +++ b/Package.swift @@ -18,6 +18,7 @@ let swiftSettings: [PackageDescription.SwiftSetting] = [ .enableUpcomingFeature("InternalImportsByDefault"), .enableUpcomingFeature("MemberImportVisibility"), .enableUpcomingFeature("InferIsolatedConformances"), + .enableUpcomingFeature("NonisolatedNonsendingByDefault"), .enableUpcomingFeature("ImmutableWeakCaptures"), ] From 9d288c39c01ec0414f0b3a971cdbc7f77260dc77 Mon Sep 17 00:00:00 2001 From: Dimitri Bouniol Date: Tue, 28 Jul 2026 03:55:47 -0700 Subject: [PATCH 7/7] Updated all transaction callers to use an isolated context by default --- .../Disk Persistence/DiskPersistence.swift | 30 +++++++++++------ .../Transaction/Transaction.swift | 33 ++++++++++++------- .../MemoryPersistence.swift | 3 +- .../Persistence/Persistence.swift | 28 ++++++++++++++-- 4 files changed, 69 insertions(+), 25 deletions(-) diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift index c26a623..3da6c28 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift @@ -38,7 +38,8 @@ public actor DiskPersistence: Persistence { var rollingPageCacheIndex = 0 var rollingPageCache: [Datastore.Page] = [] - var transactionCounter = 0 + nonisolated(unsafe) var transactionCounter = 0 + let transactionCounterLock = UnfairLock() /// Initialize a ``DiskPersistence`` with a read-write URL. /// @@ -464,16 +465,28 @@ extension DiskPersistence { // MARK: - Transactions extension DiskPersistence { - func nextTransactionCounter() -> Int { - let transactionIndex = transactionCounter - transactionCounter += 1 - return transactionIndex + nonisolated func nextTransactionCounter() -> Int { + transactionCounterLock.withLock { + let transactionIndex = transactionCounter + transactionCounter += 1 + return transactionIndex + } + } + + /// Replace the last mutating root transaction with the specified transaction, and return the previous one. + func enqueue(rootTransaction: Transaction) -> Transaction? { + let lastTransaction = lastMutatingTransaction + if !rootTransaction.options.contains(.readOnly) { + self.lastMutatingTransaction = rootTransaction + } + return lastTransaction } public func _withTransaction( + isolation actor: isolated (any Actor)? = #isolation, actionName: String?, options: UnsafeTransactionOptions, - transaction operation: sending (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + transaction operation: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T { /// If the transaction is starting in the context of another persistence's transaction, make sure it is a read-only one. Otherwise assert and throw an error as it likely indicates a mistake and could lead to unexpected consistency violations if one persistence succeeds while the other fails. if isTransactingExternally() { @@ -519,10 +532,7 @@ extension DiskPersistence { ) /// Save the last non-concurrent root transaction from the list. Note that disk persistence currently does not support concurrent idempotent transactions. - let lastTransaction = lastMutatingTransaction - if !options.contains(.readOnly) { - self.lastMutatingTransaction = transaction - } + let lastTransaction = await enqueue(rootTransaction: transaction) let result = try await transaction.run( lastTransaction: !options.contains(.readOnly) ? lastTransaction : nil, diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift index aa9be3f..7ea84bc 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift @@ -73,16 +73,14 @@ extension DiskPersistence { /// - operation: The operation itself, which takes a reference to the owning transaction and the `isDurable` flag /// - Returns: The return value of the operation, if any. func run( + isolation actor: isolated (any Actor)? = #isolation, lastTransaction: Transaction?, isDurable: Bool, operation: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T { - guard let workDidFinishPromise = workDidFinishPromise.take() - else { fatalError("workDidFinishPromise was already consumed. Please report this as a critical bug!") } - let returnValue: T do { - isActive = true + await setIsActive(true) /// If provided, wait for the last transaction to properly finish before startign the next one. if let lastTransaction { @@ -94,17 +92,14 @@ extension DiskPersistence { try await operation(self, isDurable) } - isActive = false + await setIsActive(false) /// Mark the work as done so opportunistic transaction may begin. - workDidFinishPromise.resume() + await reportWorkDidFinish() } catch { /// If the operation failed, make sure to resume both continuations so the next transaction can start. - workDidFinishPromise.resume() - - guard let transactionDidPersistPromise = transactionDidPersistPromise.take() - else { fatalError("transactionDidPersistPromise was already consumed. Please report this as a critical bug!") } - transactionDidPersistPromise.resume() + await reportWorkDidFinish() + await reportTransactionDidPersist() throw error } @@ -123,6 +118,18 @@ extension DiskPersistence { return returnValue } + private func reportWorkDidFinish() { + guard let workDidFinishPromise = workDidFinishPromise.take() + else { fatalError("workDidFinishPromise was already consumed. Please report this as a critical bug!") } + workDidFinishPromise.resume() + } + + private func reportTransactionDidPersist() { + guard let transactionDidPersistPromise = transactionDidPersistPromise.take() + else { fatalError("transactionDidPersistPromise was already consumed. Please report this as a critical bug!") } + transactionDidPersistPromise.resume() + } + /// Yield and wait for the main work of the transaction to be complete. func yieldWorkCompletion() async { await workDidFinishResult.yield() @@ -133,6 +140,10 @@ extension DiskPersistence { await transactionDidPersistResult.yield() } + private func setIsActive(_ isActive: Bool) { + self.isActive = isActive + } + /// Ensure the receiving transaction is currenrtly active. func checkIsActive() throws { guard isActive else { diff --git a/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift b/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift index 53499c6..8f79eb9 100644 --- a/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift +++ b/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift @@ -15,9 +15,10 @@ public actor MemoryPersistence: Persistence { extension MemoryPersistence { public func _withTransaction( + isolation actor: isolated (any Actor)? = #isolation, actionName: String?, options: UnsafeTransactionOptions, - transaction: sending (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + transaction: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T { preconditionFailure("Unimplemented") } diff --git a/Sources/CodableDatastore/Persistence/Persistence.swift b/Sources/CodableDatastore/Persistence/Persistence.swift index d3aaca8..2420051 100644 --- a/Sources/CodableDatastore/Persistence/Persistence.swift +++ b/Sources/CodableDatastore/Persistence/Persistence.swift @@ -19,13 +19,33 @@ public protocol Persistence: Sendable { /// - options: The options to use while building the transaction. /// - transaction: A closure representing the transaction with which to perform operations on. You should not escape the provided transaction. func _withTransaction( + isolation actor: isolated (any Actor)?, actionName: String?, options: UnsafeTransactionOptions, - transaction: sending (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + transaction: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T } extension Persistence { + /// Perform a transaction on the persistence with the specified options. + /// - Parameters: + /// - actionName: The name of the action to use in an undo operation. Only the names of top-level transactions are recorded. We recommend you use the Base localization name here, as you may no longer have the localized version in future versions of your app. + /// - options: The options to use while building the transaction. + /// - transaction: A closure representing the transaction with which to perform operations on. You should not escape the provided transaction. + func _withTransaction( + isolation actor: isolated (any Actor)? = #isolation, + actionName: String?, + options: UnsafeTransactionOptions, + transaction: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + ) async throws -> T { + try await _withTransaction( + isolation: actor, + actionName: actionName, + options: options, + transaction: transaction + ) + } + /// Perform a set of operations as a single transaction. /// /// Within the transaction block, perform operations on multiple ``Datastore``s such that if any one of them were to fail, none will be persisted, and if all of them succeed, they will be persisted atomically. @@ -39,9 +59,10 @@ extension Persistence { /// - transaction: A closure with the set of operations to perform. Parameters include a reference to the persistence, and a flag indicating if the transaction is durable. /// - SeeAlso: ``Persistence/perform(actionName:options:transaction:)-1tpvd`` public func perform( + isolation actor: isolated (any Actor)? = #isolation, actionName: String? = nil, options: TransactionOptions = [], - _inheritActorContext transaction: @Sendable (_ persistence: Self, _ isDurable: Bool) async throws -> T + transaction: (_ persistence: Self, _ isDurable: Bool) async throws -> T ) async throws -> T { try await _withTransaction( actionName: actionName, @@ -64,9 +85,10 @@ extension Persistence { /// - transaction: A closure with the set of operations to perform. /// - SeeAlso: ``Persistence/perform(actionName:options:transaction:)-5476l`` public func perform( + isolation actor: isolated (any Actor)? = #isolation, actionName: String? = nil, options: TransactionOptions = [], - @_inheritActorContext transaction: @Sendable () async throws -> T + transaction: () async throws -> T ) async throws -> T { try await _withTransaction( actionName: actionName,