diff --git a/README.md b/README.md index bf8ec4b..584cfeb 100644 --- a/README.md +++ b/README.md @@ -261,8 +261,7 @@ Note that in the example above, even though the author is persisted first, if an As this project matures towards release, the project will focus on the functionality and work listed below: - Force migration methods -- Composite indexes (via macros?) -- Cleaning up old resources on disk +- Composite indexes - Ranged deletes - Controls for the edit history - Helper types to use with SwiftUI/Observability/Combine that can make data available on the main actor and filter and stay up to date @@ -271,7 +270,7 @@ As this project matures towards release, the project will focus on the functiona - An example app - A memory persistence useful for testing apps with - A pre-configured data store tuned to storing pure Data, useful for types like Images -- Cleaning up memory leaks +- Cleaning up memory and file descriptor leaks The above list will be kept up to date during development and will likely see additions during that process. diff --git a/Sources/CodableDatastore/Helpers/FileManager+Helpers.swift b/Sources/CodableDatastore/Helpers/FileManager+Helpers.swift new file mode 100644 index 0000000..169c4db --- /dev/null +++ b/Sources/CodableDatastore/Helpers/FileManager+Helpers.swift @@ -0,0 +1,33 @@ +// +// FileManager+Helpers.swift +// https://github.com/mochidev/CodableDatastore +// +// Created by Dimitri Bouniol on 2024-09-08. +// Copyright © 2023-26 Mochi Development, Inc. All rights reserved. +// mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 +// + +import Foundation + +enum DirectoryRemovalError: Error { + case missingEnumerator +} + +extension FileManager { + @discardableResult + func removeDirectoryIfEmpty(url: URL, recursivelyRemoveParents: Bool) throws -> Bool { + guard let enumerator = self.enumerator(at: url, includingPropertiesForKeys: [], options: [.skipsHiddenFiles, .skipsSubdirectoryDescendants, .skipsPackageDescendants, .includesDirectoriesPostOrder]) + else { throw DirectoryRemovalError.missingEnumerator } + + for case _ as URL in enumerator { + /// If this is called a single time, then we don't have an empty directory, and can stop + return false + } + + try self.removeItem(at: url) + + guard recursivelyRemoveParents else { return true } + try self.removeDirectoryIfEmpty(url: url.deletingLastPathComponent(), recursivelyRemoveParents: recursivelyRemoveParents) + return true + } +} diff --git a/Sources/CodableDatastore/Helpers/Swift6.1+Compatibility.swift b/Sources/CodableDatastore/Helpers/Swift6.1+Compatibility.swift index 0d62306..16c3cbc 100644 --- a/Sources/CodableDatastore/Helpers/Swift6.1+Compatibility.swift +++ b/Sources/CodableDatastore/Helpers/Swift6.1+Compatibility.swift @@ -8,10 +8,45 @@ // #if compiler(<6.2) -extension Task where Failure == Never { +extension Task { @discardableResult - init(name: String?, priority: TaskPriority? = nil, operation: sending @escaping @isolated(any) () async -> Success) { + init( + name: String?, + priority: TaskPriority? = nil, + @_inheritActorContext @_implicitSelfCapture operation: sending @escaping @isolated(any) () async -> Success + ) where Failure == Never { self.init(priority: priority, operation: operation) } + + @discardableResult + init( + name: String?, + priority: TaskPriority? = nil, + @_inheritActorContext @_implicitSelfCapture operation: sending @escaping @isolated(any) () async throws -> Success + ) where Failure == any Error { + self.init(priority: priority, operation: operation) + } + + @discardableResult + public static func detached( + name: String?, + priority: TaskPriority? = nil, + operation: sending @escaping @isolated(any) () async -> Success + ) -> Task where Failure == Never { + Task.detached(priority: priority, operation: operation) + } + + @discardableResult + public static func detached( + name: String?, + priority: TaskPriority? = nil, + operation: sending @escaping @isolated(any) () async throws -> Success + ) -> Task where Failure == any Error { + Task.detached(priority: priority, operation: operation) + } +} + +public func extendLifetime(_ x: borrowing T) where T : ~Copyable { + withExtendedLifetime(x) {} } #endif diff --git a/Sources/CodableDatastore/Helpers/URLCollector.swift b/Sources/CodableDatastore/Helpers/URLCollector.swift new file mode 100644 index 0000000..a297508 --- /dev/null +++ b/Sources/CodableDatastore/Helpers/URLCollector.swift @@ -0,0 +1,30 @@ +// +// URLCollector.swift +// https://github.com/mochidev/CodableDatastore +// +// Created by Dimitri Bouniol on 2026-09-06. +// Copyright © 2023-26 Mochi Development, Inc. All rights reserved. +// mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 +// + +import Foundation +import QuestionableConcurrency + +class URLCollector: @unchecked Sendable { + private var gate = UnfairLock() + private var urls: Set = [] + + init() {} + + func insertURL(_ url: URL) { + gate.withLock { + _ = urls.insert(url) + } + } + + /// Return all URLs, sorted by longest. + func removeAllURLs() -> [URL] { + let urls = gate.withLock { self.urls } + return urls.sorted { $0.absoluteString > $1.absoluteString } + } +} diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift index 779df1c..2b2b0e8 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift @@ -91,6 +91,15 @@ extension DatastoreIndexManifest { } } +extension DatastoreIndexManifest { + func pagesToPrune(for mode: SnapshotPruneMode) -> Set { + switch mode { + case .pruneRemoved: Set(removedPageIDs) + case .pruneAdded: Set(addedPageIDs) + } + } +} + // MARK: - Decoding extension DatastoreIndexManifest { diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreRootManifest.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreRootManifest.swift index b1676cc..7a92ab7 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreRootManifest.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreRootManifest.swift @@ -110,3 +110,25 @@ extension DatastoreRootManifest { } } } + +extension DatastoreRootManifest { + func indexesToPrune(for mode: SnapshotPruneMode) -> Set { + switch mode { + case .pruneRemoved: removedIndexes + case .pruneAdded: addedIndexes + } + } + + func indexManifestsToPrune( + for mode: SnapshotPruneMode, + options: SnapshotPruneOptions + ) -> Set { + switch (mode, options) { + case (.pruneRemoved, .pruneAndDelete): removedIndexManifests + case (.pruneAdded, .pruneAndDelete): addedIndexManifests + /// Flip the results when we aren't deleting, but only when removing from the bottom end. + case (.pruneRemoved, .pruneOnly): addedIndexManifests + case (.pruneAdded, .pruneOnly): [] + } + } +} diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift index a7c9c0d..e750fe6 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift @@ -159,6 +159,109 @@ extension DiskPersistence.Datastore { } } + func pruneRootObject(with identifier: RootObject.ID, mode: SnapshotPruneMode, shouldDelete: Bool) async throws { + let fileManager = FileManager() + let rootObject = try loadRootObject(for: identifier, shouldCache: false) + + /// Collect the indexes and related manifests we'll be deleting. + /// - For indexes, only collect the ones we'll be deleting since the ones we are keeping won't be making references to other deletable assets. + /// - For the manifests, we'll be deleting the entries that are being removed (relative to the direction we are removing from, so the removed ones from the oldest edge, and the added ones from the newest edge, as determined by the caller), while we'll be checking for pages to remove from entries that have just been added, but only when removing from the oldest edge. We only do this for the oldest edge because pages that have been "removed" from the newest edge are actually being _restored_ and not replaced, which maintains symmetry in a non-obvious way. + let indexesToPruneAndDelete = rootObject.indexesToPrune(for: mode) + let indexManifestsToPruneAndDelete = rootObject.indexManifestsToPrune(for: mode, options: .pruneAndDelete) + let indexManifestsToPrune = rootObject.indexManifestsToPrune(for: mode, options: .pruneOnly) + + /// Delete the index manifests and pages we know to be removed. + for indexManifestID in indexManifestsToPruneAndDelete { + let indexID = Index.ID(indexManifestID) + defer { + trackedIndexes.removeValue(forKey: indexID) + loadedIndexes.remove(indexID) + } + /// Skip any manifests for indexes being deleted, since we'll just unlink the whole directory in that case. + guard !indexesToPruneAndDelete.contains(indexID.indexID) else { continue } + + let manifestURL = manifestURL(for: indexID) + let manifest: DatastoreIndexManifest? + do { + manifest = try await DatastoreIndexManifest(contentsOf: manifestURL, id: indexID.manifestID) + } catch FileNotFoundError() { + manifest = nil + } catch { + print("Uncaught Manifest Error: \(error)") + throw error + } + + guard let manifest else { continue } + + /// Only delete the pages we know to be removed + let pagesToPruneAndDelete = manifest.pagesToPrune(for: mode) + for pageID in pagesToPruneAndDelete { + let indexedPageID = Page.ID(index: indexID, page: pageID) + defer { + trackedPages.removeValue(forKey: indexedPageID.withoutManifest) + loadedPages.remove(indexedPageID.withoutManifest) + } + + let pageURL = pageURL(for: indexedPageID) + + try? fileManager.removeItem(at: pageURL) + snapshot.persistence.directoriesToRemove.insertURL(pageURL.deletingLastPathComponent()) + } + + try? fileManager.removeItem(at: manifestURL) + } + + /// Prune the index manifests that were just added, as they themselves refer to other deleted pages. + for indexManifestID in indexManifestsToPrune { + let indexID = Index.ID(indexManifestID) + /// Skip any manifests for indexes being deleted, since we'll just unlink the whole directory in that case. + guard !indexesToPruneAndDelete.contains(indexID.indexID) else { continue } + + let manifestURL = manifestURL(for: indexID) + let manifest: DatastoreIndexManifest? + do { + manifest = try await DatastoreIndexManifest(contentsOf: manifestURL, id: indexID.manifestID) + } catch FileNotFoundError() { + manifest = nil + } catch { + print("Uncaught Manifest Error: \(error)") + throw error + } + + guard let manifest else { continue } + + /// Only delete the pages we know to be removed + let pagesToPruneAndDelete = manifest.pagesToPrune(for: mode) + for pageID in pagesToPruneAndDelete { + let indexedPageID = Page.ID(index: indexID, page: pageID) + defer { + trackedPages.removeValue(forKey: indexedPageID.withoutManifest) + loadedPages.remove(indexedPageID.withoutManifest) + } + + let pageURL = pageURL(for: indexedPageID) + + try? fileManager.removeItem(at: pageURL) + snapshot.persistence.directoriesToRemove.insertURL(pageURL.deletingLastPathComponent()) + } + } + + /// Delete any indexes in their entirety. + for indexID in indexesToPruneAndDelete { + try? fileManager.removeItem(at: indexURL(for: indexID)) + } + + /// If we are deleting the root object itself, do so at the very end as everything else would have been cleaned up. + if shouldDelete { + trackedRootObjects.removeValue(forKey: identifier) + loadedRootObjects.remove(identifier) + + let rootURL = rootURL(for: rootObject.id) + try? fileManager.removeItem(at: rootURL) + snapshot.persistence.directoriesToRemove.insertURL(rootURL.deletingLastPathComponent()) + } + } + func index(for identifier: Index.ID) -> Index { if let index = trackedIndexes[identifier]?.value { return index @@ -220,14 +323,16 @@ extension DiskPersistence.Datastore { extension DiskPersistence.Datastore { /// Load the root object from disk for the given identifier. - func loadRootObject(for rootIdentifier: DatastoreRootIdentifier) throws -> DatastoreRootManifest { + func loadRootObject(for rootIdentifier: DatastoreRootIdentifier, shouldCache: Bool = true) throws -> DatastoreRootManifest { let rootObjectURL = rootURL(for: rootIdentifier) let data = try Data(contentsOf: rootObjectURL) let root = try JSONDecoder.shared.decode(DatastoreRootManifest.self, from: data) - cachedRootObject = root + if shouldCache { + cachedRootObject = root + } return root } diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift index 97e24a2..0c78f45 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift @@ -28,6 +28,10 @@ public actor DiskPersistence: Persistence { var lastMutatingTransaction: Transaction? var rootTransactionStream = TransactionStream() + var _transactionRetentionPolicy: SnapshotRetentionPolicy = .indefinite + + let directoriesToRemove = URLCollector() + /// Shared caches across all snapshots and datastores. var rollingRootObjectCacheIndex = 0 var rollingRootObjectCache: [Datastore.RootObject] = [] @@ -61,6 +65,12 @@ public actor DiskPersistence: Persistence { storeURL = readOnlyURL } + deinit { + for (_, snapshot) in snapshots { + snapshot.cancelPruning() + } + } + /// The default URL to use for disk persistences. static var defaultURL: URL { // TODO: Make non-throwing: https://github.com/mochidev/CodableDatastore/issues/15 @@ -262,7 +272,7 @@ extension DiskPersistence { return snapshot } - let snapshot = Snapshot(id: snapshotID, persistence: self) + let snapshot = Snapshot(id: snapshotID, persistence: self, isExtendedIterationCacheEnabled: !_transactionRetentionPolicy.isIndefinite) snapshots[snapshotID] = snapshot return snapshot @@ -588,7 +598,7 @@ extension DiskPersistence { else { throw DiskPersistenceError.cannotWrite } /// If we are read-write, apply the updated root objects to the snapshot. - try await self.updatingCurrentSnapshot { snapshot in + let (currentSnapshot, persistedIteration) = try await self.updatingCurrentSnapshot { snapshot in try await snapshot.updatingManifest { manifest, iteration in iteration.actionName = actionName iteration.addedDatastoreRoots = addedDatastoreRoots @@ -600,9 +610,89 @@ extension DiskPersistence { root: root.id ) } + return (snapshot, iteration) + } + } + + /// Let the snapshot know it should start pruning from the iteration we just persisted. + await currentSnapshot.enforce(retentionPolicy: _transactionRetentionPolicy, fromIteration: persistedIteration.id) + } +} + +// MARK: - Retention Policy + +extension DiskPersistence where AccessMode == ReadWrite { + /// The current transaction retention policy for snapshot iterations written to disk. + public var transactionRetentionPolicy: SnapshotRetentionPolicy { + get async { + _transactionRetentionPolicy + } + } + + /// Update the transaction retention policy for snapshot iterations written to disk. + /// + /// If a snapshot is currently pruning old iterations, it will be allowed to finish with the retention policy in place at the time of the transaction that triggered the policy enforcement. The new policy takes effect once the next transaction is written. + /// + /// - Parameter policy: The new policy to enforce on write. + /// + /// - SeeAlso: ``SnapshotRetentionPolicy``. + public func setTransactionRetentionPolicy(_ policy: SnapshotRetentionPolicy) async { + _transactionRetentionPolicy = policy + let currentSnapshot = try? await self.readingCurrentSnapshot { $0 } + /// Configure snapshots to start caching iterations to speed up pruning. + for (_, snapshot) in snapshots { + if snapshot === currentSnapshot { + await snapshot.setExtendedIterationCacheEnabled(!_transactionRetentionPolicy.isIndefinite) + } else { + await snapshot.setExtendedIterationCacheEnabled(false) } } } + + /// Enforce the retention policy on the persistence immediately. + /// + /// - Note: Transaction retention policies are enforced after every write transaction, so calling this method directly is often unecessary. However, it can be useful if the user requires disk resources immediately. + public func enforceRetentionPolicy() async { + let info = try? await self.readingCurrentSnapshot { snapshot -> (snapshot: Snapshot, iteration: SnapshotIteration)? in + guard let snapshot else { return nil } + return try await snapshot.readingManifest { manifest, iteration in + (snapshot: snapshot, iteration: iteration) + } + } + + if let (snapshot, iteration) = info { + await snapshot.enforce(retentionPolicy: _transactionRetentionPolicy, fromIteration: iteration.id, taskPriority: Task.currentPriority).value + } + } + + func removeEmptyDirectories() { + var directoriesToRemove = directoriesToRemove.removeAllURLs() + var allDirectories = Set(directoriesToRemove) + + while let directory = allDirectories.popFirst() { + guard (try? FileManager.default.removeDirectoryIfEmpty(url: directory, recursivelyRemoveParents: false)) == true + else { continue } + + let parent = directory.deletingLastPathComponent() + if !allDirectories.contains(parent) { + directoriesToRemove.append(parent) + allDirectories.insert(parent) + } + } + } +} + +extension DiskPersistence { + /// Await any cleanup since the last complete write transaction to the persistence. + /// + /// - Note: An application is not required to await cleanup, as it'll be eventually completed on future runs. It is however useful to wait for this to complete in cases when disk resources must be cleared before progressing. + public func checkTransactionCleanupFinished() async { + for (_, snapshot) in snapshots { + /// Await it twice so we catch any immediately scheduled work since we first suspended. + await snapshot.checkPruningFinished() + await snapshot.checkPruningFinished() + } + } } // MARK: - Persistence-wide Caches diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/FileNotFoundError.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/FileNotFoundError.swift index a7cc139..64b91a6 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/FileNotFoundError.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/FileNotFoundError.swift @@ -7,11 +7,13 @@ // mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 // +import Bytes import Foundation struct FileNotFoundError: Error { static func ~= (lhs: FileNotFoundError, rhs: any Error) -> Bool { (rhs as any FileError).isFileNotFound == true + || ((rhs as? any ByteIterationError)?.iterationError as? any FileError)?.isFileNotFound == true } } @@ -24,3 +26,16 @@ extension NSError: FileError { URLError.fileDoesNotExist ~= self || CocoaError.fileReadNoSuchFile ~= self || CocoaError.fileNoSuchFile ~= self || POSIXError.ENOENT ~= self } } + +protocol ByteIterationError { + var iterationError: (any Error)? { get } +} + +extension BytesError.IterationError: ByteIterationError { + var iterationError: (any Error)? { + switch self { + case .castingFailure: nil + case .iterationFailure(let error): error + } + } +} diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift index 49fa4c5..9e107d7 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift @@ -36,8 +36,11 @@ actor Snapshot { /// A cached instance of the manifest as last loaded from disk. var cachedManifest: SnapshotManifest? - /// A cached instance of the current iteration as last loaded from disk. - var cachedIteration: SnapshotIteration? + /// Cache for the loaded iterations as last loaded from disk. ``isExtendedIterationCacheEnabled`` controls if multiple iterations are cached or not. + var cachedIterations: [SnapshotIterationIdentifier : SnapshotIteration] = [:] + var isExtendedIterationCacheEnabled: Bool + private var nextSnapshotIterationCandidateToEnforce: (iterationID: SnapshotIteration.ID, retentionPolicy: SnapshotRetentionPolicy, taskPriority: TaskPriority)? + private var snapshotIterationPruningTask: Task? /// A transaction stream for manifest updates, so reads and writes can be serialized in request order. var manifestTransactionStream = TransactionStream() @@ -45,14 +48,30 @@ actor Snapshot { /// The loaded datastores. var datastores: [DatastoreIdentifier: DiskPersistence.Datastore] = [:] + /// The chain of iterations + var iterationChain: SparseIterationChain + var iterationChainState: SparseIterationChain.State + + private var pruningWatermark = 0 + private var lastPruningTask: Task? + init( id: SnapshotIdentifier, persistence: DiskPersistence, - isBackup: Bool = false + isBackup: Bool = false, + isExtendedIterationCacheEnabled: Bool = false ) { self.id = id self.persistence = persistence self.isBackup = isBackup + self.isExtendedIterationCacheEnabled = isExtendedIterationCacheEnabled + + self.iterationChain = SparseIterationChain() + self.iterationChainState = .forwardEditsOnly + } + + deinit { + snapshotIterationPruningTask?.cancel() } } @@ -130,20 +149,419 @@ extension Snapshot { } } - /// Load an iteration from disk, or create a suitable starting value if such a file does not exist. - private func loadIteration(for iterationID: SnapshotIterationIdentifier) throws -> SnapshotIteration { + func setExtendedIterationCacheEnabled(_ isEnabled: Bool) async { + isExtendedIterationCacheEnabled = isEnabled + + /// If the extended cache is being disabled, and we are currently pruning, immediately stop and cancel the process. + if !isEnabled { + snapshotIterationPruningTask?.cancel() + await snapshotIterationPruningTask?.value + } + + await invalidateIterationChainState() + } + + private func invalidateIterationChainState() async { + let persistence = persistence + switch iterationChainState { + case .forwardEditsOnly: + /// If we are currently only collecting forward edits, and the extended cache was just enabled, start the crawling process. Leave it empty until we make our first iteration read though. + if isExtendedIterationCacheEnabled, iterationChain.first != nil { + iterationChainState = .crawling(Task(name: "Iteration Chain Crawler") { + do { + try await crawlIterations() + iterationChainState = .complete + } catch { + print("Error crawling iterations: \(error)") + iterationChainState = .forwardEditsOnly + } + extendLifetime(persistence) + }) + } + case .crawling(let task): + /// If we are currently crawling, but the extended cache was just disabled, cancel the crawling process and swap back to the incomplete state. Everything we have should still be valid. + if !isExtendedIterationCacheEnabled { + /// The state is managed by the task, and doesn't need to be set here, so long as we wait for it to complete up to the cancellation point. + task.cancel() + await task.value + } + case .complete: + break + } + } + + private func crawlIterations() async throws { + guard let currentIterationID = iterationChain.last?.iteration + else { throw CrawlingCouldNotStartError() } + + /// Make sure the last known iteration is fresh from disk. + var currentIteration = try self.loadIterationNoCache(for: currentIterationID) + + /// Walk the preceding iteration chain to the oldest iteration we can open, collecting the ones that form the train. + while let precedingIterationID = currentIteration.precedingIteration, let precedingIteration = try? await loadIteration(for: precedingIterationID) { + try Task.checkCancellation() + + currentIteration = precedingIteration + iterationChain.append(iteration: precedingIteration) + } + } + + /// Load an iteration from cache or disk, or create a suitable starting value if such a file does not exist. + func loadIteration(for iterationID: SnapshotIterationIdentifier?) async throws -> SnapshotIteration? { + guard let iterationID else { return nil } + if let iteration = cachedIterations[iterationID] { + return iteration + } + return try loadIterationNoCache(for: iterationID) + } + + /// Load an iteration from disk ignoring the current cached value, or create a suitable starting value if such a file does not exist. + func loadIterationNoCache(for iterationID: SnapshotIterationIdentifier) throws -> SnapshotIteration { do { let data = try Data(contentsOf: iterationURL(for: iterationID)) - + let iteration = try JSONDecoder.shared.decode(SnapshotIteration.self, from: data) - - cachedIteration = iteration + + if !isExtendedIterationCacheEnabled { + cachedIterations.removeAll() + } + /// Make sure not to grow the cache unecessarily. 256 represents the smallest chunk in the chain that we care to have in memory at once + if cachedIterations.count >= 256, let firstKey = cachedIterations.keys.first { + cachedIterations.removeValue(forKey: firstKey) + } + cachedIterations[iteration.id] = iteration return iteration } catch { throw error } } + /// Let the snapshot know it should enforce the specified retention policy from a given iteration. A task that can be awaited is returned + @discardableResult + func enforce( + retentionPolicy: SnapshotRetentionPolicy, + fromIteration iterationID: SnapshotIteration.ID, + taskPriority: TaskPriority = .background + ) -> Task where AccessMode == ReadWrite { +// print("Enforcing based on \(iterationID)") + /// Since a previous request may have used a higher priority, make sure we maintain that priority since we are replacing that work. + let resolvedTaskPriority = max(nextSnapshotIterationCandidateToEnforce?.taskPriority ?? taskPriority, taskPriority) + nextSnapshotIterationCandidateToEnforce = nil + + let persistence = persistence + + if let pruningTask = snapshotIterationPruningTask { + /// A pruning task is already in progress, so bookmark the iteration that needs to cleanup, and simply wait for the pruning task that eventually replaces the current one. + nextSnapshotIterationCandidateToEnforce = ( + iterationID: iterationID, + retentionPolicy: retentionPolicy, + taskPriority: resolvedTaskPriority, + ) + return Task.detached(name: "RetentionPolicyEnforcementWatcher") { + await pruningTask.value + await self.snapshotIterationPruningTask?.value + extendLifetime(persistence) + } + } + + let pruningTask = Task.detached(name: "RetentionPolicyEnforcement", priority: resolvedTaskPriority) { + do { + try await self._enforce(retentionPolicy: retentionPolicy, fromIteration: iterationID) + } catch { + print("Error pruning: \(error)") + + /// Wait for any in-progress pruning tasks to finish. + try? await self.drainPrunedIterations() + + /// The iteration chain is no longer complete, so set it back to `.forwardEditsOnly` so we can re-build it the next time. + await self.resetIterationChainState() + } + /// Either enqueue the next policy enforcement, or reset task state if there is no more work slated. + await self.enqueueNextPolicyEnforcement() + extendLifetime(persistence) + } + + snapshotIterationPruningTask = pruningTask + return pruningTask + } + + /// A private method for scheduling pruning tasks based on the retention policy to enforce. + private func _enforce( + retentionPolicy: SnapshotRetentionPolicy, + fromIteration iterationID: SnapshotIteration.ID + ) async throws where AccessMode == ReadWrite { +// print("Pruning started for \(iterationID).") + guard !retentionPolicy.isIndefinite else { +// print("Current policy doesn't require any pruning, stopping early.") + return + } + + /// Enable the extended cache, and wait for it to be filled out before doing any work. + await setExtendedIterationCacheEnabled(true) + switch iterationChainState { + case .forwardEditsOnly: + /// The chain isn't in a state that can support proper pruning. Stop here. + throw CancellationError() + case .crawling(let task): + await task.value + /// If the chain doesn't settle on a `.complete`state, then it was prematurely cancelled. Simply pass that failure state along. + guard case .complete = iterationChainState else { + throw CancellationError() + } + case .complete: + break + } + + try Task.checkCancellation() + + let now = Date() + print("Chain has \(iterationChain.count) entries over \(iterationChain.groups.count) groups.") + + /// Get a starting point from which we will start pruning iterations, without walking the entire graph. + let (startingDistance, removedGroups) = iterationChain.removeIterations(failing: retentionPolicy, from: iterationID, now: now) + + var totalIterationCount = iterationChain.count + var iterations: [SnapshotIteration.ID] = [] + var distance = startingDistance + var nextIterationID = removedGroups.first?.first.iteration + var mainlineRootIteration = try await loadIteration(for: iterationChain.last?.iteration) + + /// Walk the preceding iteration chain to the oldest iteration we can open, collecting the ones that should be pruned, and re-adding the ones that shouldn't back to the iteration chain. + while let precedingIterationID = nextIterationID, let precedingIteration = try? await loadIteration(for: precedingIterationID) { + try Task.checkCancellation() + + if !iterations.isEmpty || retentionPolicy.shouldIterationBePruned(now: now, creationDate: precedingIteration.creationDate, distance: distance) { + iterations.append(precedingIteration.id) + } else { + /// The iteration isn't actually ready to be pruned, so add it back to the chain. + iterationChain.append(iteration: precedingIteration) + totalIterationCount += 1 + mainlineRootIteration = precedingIteration + } + + nextIterationID = precedingIteration.precedingIteration + + distance += 1 + + if (totalIterationCount + iterations.count) % 100 == 0 { + print("Found \(iterations.count) iterations to prune. Keeping \(totalIterationCount) iterations.") + } + } + + guard + iterations.count > 0, + let mainlineRootIteration + else { + print("There were no iteration to prune, stopping early.") + return + } + + print("Will prune \(iterations.count) iterations. Keeping \(totalIterationCount) iterations.") + + /// Prune iterations from oldest to newest along the mainline. + while let iterationID = iterations.popLast(), let iteration = try await loadIteration(for: iterationID) { + /// The current index, since we just removed the last element. + let index = iterations.count + let mainlineSuccessorIterationID = index > 0 ? iterations[index-1] : mainlineRootIteration.id + + if index % 100 == 0 { + print("\(index) iterations left to delete.") + } + + var iterationsToPrune: [SnapshotIteration] = [] + var successorCandidatesToCheck = iteration.successiveIterations + successorCandidatesToCheck.removeAll { $0 == mainlineSuccessorIterationID } + + /// Walk the non-mainline successor candidates all the way back up so newer iterations are pruned before the ones that reference them. We pull items off from the end, and add new ones to the beginning to make sure they stay in graph order. + while let successorCandidateID = successorCandidatesToCheck.popLast() { + try Task.checkCancellation() + guard let successorIteration = try? await loadIteration(for: successorCandidateID) + else { continue } + + iterationsToPrune.append(successorIteration) + successorCandidatesToCheck.insert(contentsOf: successorIteration.successiveIterations, at: 0) + } + + /// First, remove the branch of iterations based on the one we are removing, but representing a history that was previously reverted (the non-mainline successors, from newest to oldest). + /// Prune the iterations in atomic tasks so they don't get cancelled mid-way, and instead check for cancellation in between iterations. + while let iteration = iterationsToPrune.popLast() { + try await pruneIteration(iteration, mode: .pruneAdded, shouldDelete: true) + } + + /// Finally, prune and delete the iteration itself. + try await pruneIteration(iteration, mode: .pruneRemoved, shouldDelete: true) + } + + /// Once we deleted all iterations that fall outside the set policy, prune the last iteration that we are keeping. + try await pruneIteration(mainlineRootIteration, mode: .pruneRemoved, shouldDelete: false) + /// Wait for all in-progress pruning operations to finish. + try await drainPrunedIterations() + + await persistence.removeEmptyDirectories() + print("Pruning complete!") + } + + /// Reset the iteration chain from an unknown `.complete` state to a known waiting state. + private func resetIterationChainState() { + switch iterationChainState { + case .forwardEditsOnly: + break + case .crawling: + preconditionFailure("Iteration chain is currently crawling when it should have been complete.") + case .complete: + break + } + + /// The iteration chain is no longer complete, so set it back to `.forwardEditsOnly` so we can re-build it the next time. + iterationChainState = .forwardEditsOnly + } + + /// Swap the current pruning task with one for the next candidate to enforce, if available. + private func enqueueNextPolicyEnforcement() where AccessMode == ReadWrite { + snapshotIterationPruningTask = nil + if let nextCandidate = nextSnapshotIterationCandidateToEnforce { + enforce( + retentionPolicy: nextCandidate.retentionPolicy, + fromIteration: nextCandidate.iterationID, + taskPriority: nextCandidate.taskPriority, + ) + } + } + + /// Wait for all pruning tasks currently enqueued to finish. + func checkPruningFinished() async { + while let pruningTask = snapshotIterationPruningTask { + await pruningTask.value + } + } + + /// Cancel the current pruning task. + nonisolated func cancelPruning() { + Task { + await snapshotIterationPruningTask?.cancel() + } + } + + /// Concurrently prune iterations, but force deletions to happen serially, and only after their associated prune succeeds. + func pruneIteration(_ iteration: SnapshotIteration, mode: SnapshotPruneMode, shouldDelete: Bool) async throws { + let persistence = persistence + let pruneTask = Task(name: "Concurrent Prune Iteration \(iteration.id)") { + try await pruneIteration(iteration, mode: mode) + extendLifetime(persistence) + return iteration + } + lastPruningTask = Task(name: "Serial Prune Iteration \(iteration.id)") { [lastPruningTask] in + try await lastPruningTask?.value + let iteration = try await pruneTask.value + if shouldDelete { + deleteIteration(iteration) + } + extendLifetime(persistence) + } + pruningWatermark += 1 + + /// If we've enqueued at least 64 tasks, pause before returning control so we can drain the pool, checking for cancellation in the process. + if pruningWatermark >= 64 { + try Task.checkCancellation() + try await drainPrunedIterations() + } + } + + /// An internal method for making sure all pruning tasks complete before returning. + func drainPrunedIterations() async throws { + pruningWatermark = 0 + try await lastPruningTask?.value + } + + private func pruneIteration(_ iteration: SnapshotIteration, mode: SnapshotPruneMode) async throws { + /// Collect the datastores and related roots we'll be deleting. + /// - For datastores, only collect the ones we'll be deleting since the ones we are keeping won't be making references to other deletable assets. + /// - For the datastore roots, we'll be deleting the entries that are being removed (relative to the direction we are removing from, so the removed ones from the oldest edge, and the added ones from the newest edge, as determined by the caller), while we'll be checking for more assets to remove from entries that have just been added, but only when removing from the oldest edge. We only do this for the oldest edge because entries that have been "removed" from the newest edge are actually being _restored_ and not replaced, which maintains symmetry in a non-obvious way. + let datastoresToPruneAndDelete = iteration.datastoresToPrune(for: mode) + var datastoreRootsToPruneAndDelete = iteration.datastoreRootsToPrune(for: mode, options: .pruneAndDelete) + var datastoreRootsToPrune = iteration.datastoreRootsToPrune(for: mode, options: .pruneOnly) + + /// Start by deleting and pruning roots as needed. We attempt to do this twice, as older versions of the persistence (prior to 0.4) didn't record the datastore ID along with the root id, which would therefore require extra work. + /// First, delete the root entries we know to be removed. + for datastoreRoot in datastoreRootsToPruneAndDelete { + guard let datastoreID = datastoreRoot.datastoreID else { continue } + let datastore = datastores[datastoreID] ?? DiskPersistence.Datastore(id: datastoreID, snapshot: self) + do { + try await datastore.pruneRootObject(with: datastoreRoot.datastoreRootID, mode: mode, shouldDelete: true) + } catch FileNotFoundError() { + /// This datastore root is already gone. + } catch { + print("Could not delete datastore root \(datastoreRoot): \(error)") + throw error + } + datastoreRootsToPruneAndDelete.remove(datastoreRoot) + } + /// Prune the root entries that were just added, as they themselves refer to other deleted assets. + for datastoreRoot in datastoreRootsToPrune { + guard let datastoreID = datastoreRoot.datastoreID else { continue } + let datastore = datastores[datastoreID] ?? DiskPersistence.Datastore(id: datastoreID, snapshot: self) + do { + try await datastore.pruneRootObject(with: datastoreRoot.datastoreRootID, mode: mode, shouldDelete: false) + } catch FileNotFoundError() { + /// This datastore root is already gone. + } catch { + print("Could not prune datastore root \(datastoreRoot): \(error)") + throw error + } + datastoreRootsToPrune.remove(datastoreRoot) + } + + /// If any references remain, funnel into this code path for very old persistences. + if !datastoreRootsToPruneAndDelete.isEmpty || !datastoreRootsToPrune.isEmpty { + for (_, datastoreInfo) in iteration.dataStores { + /// Skip any roots for datastores being deleted, since we'll just unlink the whole directory in that case. + guard !datastoresToPruneAndDelete.contains(datastoreInfo.id) else { continue } + + let datastore = datastores[datastoreInfo.id] ?? DiskPersistence.Datastore(id: datastoreInfo.id, snapshot: self) + + /// Delete the root entries we know to be removed. + for datastoreRoot in datastoreRootsToPruneAndDelete { + do { + try await datastore.pruneRootObject(with: datastoreRoot.datastoreRootID, mode: mode, shouldDelete: true) + datastoreRootsToPruneAndDelete.remove(datastoreRoot) + } catch FileNotFoundError() { + /// This datastore did not contain the specified root, skip it for now. + } catch { + print("Could not delete datastore root \(datastoreRoot): \(error).") + throw error + } + } + + /// Prune the root entries that were just added, as they themselves refer to other deleted assets. + for datastoreRoot in datastoreRootsToPrune { + do { + try await datastore.pruneRootObject(with: datastoreRoot.datastoreRootID, mode: mode, shouldDelete: false) + datastoreRootsToPrune.remove(datastoreRoot) + } catch FileNotFoundError() { + /// This datastore did not contain the specified root, skip it for now. + } catch { + print("Could not prune datastore root \(datastoreRoot): \(error).") + throw error + } + } + } + } + + /// Delete any datastores in their entirety. + for datastoreID in datastoresToPruneAndDelete { + try? FileManager.default.removeItem(at: datastoreURL(for: datastoreID)) + } + } + + /// Delete the iteration. Note that an iteration should be pruned first to delete related files that are specific to the iteration itself. + private func deleteIteration(_ iteration: SnapshotIteration) { + cachedIterations.removeValue(forKey: iteration.id) + + let iterationURL = iterationURL(for: iteration.id) + try? FileManager.default.removeItem(at: iterationURL) + persistence.directoriesToRemove.insertURL(iterationURL.deletingLastPathComponent()) + } + /// Write the specified manifest to the store, and cache the results in ``Snapshot/cachedManifest``. private func write(manifest: SnapshotManifest) throws where AccessMode == ReadWrite { /// Make sure the directories exists first. @@ -161,7 +579,7 @@ extension Snapshot { cachedManifest = manifest } - /// Write the specified iteration to the store, and cache the results in ``Snapshot/cachedIteration``. + /// Write the specified iteration to the store, and cache the results in ``Snapshot/cachedIterations``. private func write(iteration: SnapshotIteration) throws where AccessMode == ReadWrite { let iterationURL = iterationURL(for: iteration.id) /// Make sure the directories exists first. @@ -172,7 +590,10 @@ extension Snapshot { try data.write(to: iterationURL, options: []) /// Update the cache since we know what it should be. - cachedIteration = iteration + if !isExtendedIterationCacheEnabled { + cachedIterations.removeAll() + } + cachedIterations[iteration.id] = iteration } /// Load and update the manifest in an updater. @@ -199,15 +620,8 @@ extension Snapshot { 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 - if let cachedIteration, cachedIteration.id == manifest.currentIteration { - iteration = cachedIteration - } else if let iterationID = manifest.currentIteration { - iteration = try self.loadIteration(for: iterationID) - } else { - let date = Date() - iteration = SnapshotIteration(id: SnapshotIterationIdentifier(date: date), creationDate: date) - } + let precedingIteration = try await self.loadIteration(for: manifest.currentIteration) + var iteration = precedingIteration ?? SnapshotIteration() /// Let the updater do something with the manifest, storing the variable on the Task Local stack. let returnValue = try await SnapshotTaskLocals.with(manifest: manifest, iteration: iteration, for: persistence) { @@ -215,10 +629,10 @@ extension Snapshot { } /// Only write to the store if we changed the manifest for any reason - if iteration.isMeaningfullyChanged(from: cachedIteration) { + if iteration.isMeaningfullyChanged(from: precedingIteration) { iteration.creationDate = Date() iteration.id = SnapshotIterationIdentifier(date: iteration.creationDate) - iteration.precedingIteration = cachedIteration?.id + iteration.precedingIteration = precedingIteration?.id try write(iteration: iteration) } @@ -228,6 +642,10 @@ extension Snapshot { /// Only write to the store if we changed the manifest for any reason if manifest != cachedManifest { try write(manifest: manifest) + + /// Add the latest iteration to the chain now that it's been written to disk for this snapshot. + iterationChain.prepend(iteration: iteration) + await invalidateIterationChainState() } return returnValue } @@ -250,15 +668,7 @@ extension Snapshot { 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 - if let cachedIteration, cachedIteration.id == manifest.currentIteration { - iteration = cachedIteration - } else if let iterationID = manifest.currentIteration { - iteration = try self.loadIteration(for: iterationID) - } else { - let date = Date() - iteration = SnapshotIteration(id: SnapshotIterationIdentifier(date: date), creationDate: date) - } + let iteration = try await self.loadIteration(for: manifest.currentIteration) ?? SnapshotIteration() /// Let the accessor do something with the manifest, storing the variable on the Task Local stack. return try await SnapshotTaskLocals.with(manifest: manifest, iteration: iteration, for: persistence) { @@ -290,6 +700,16 @@ private enum SnapshotTaskLocals { } } +enum SnapshotPruneMode { + case pruneRemoved + case pruneAdded +} + +enum SnapshotPruneOptions { + case pruneAndDelete + case pruneOnly +} + // MARK: - Datastore Management extension Snapshot { /// Load the datastore for the given key. @@ -369,3 +789,5 @@ extension Snapshot { } } } + +fileprivate struct CrawlingCouldNotStartError: Error {} diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotIteration.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotIteration.swift index 9e94b41..32cf52b 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotIteration.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotIteration.swift @@ -78,6 +78,12 @@ extension SnapshotIteration { } extension SnapshotIteration { + /// Initialize a snapshot iteration with a date + /// - Parameter date: The date to base the identifier and creation date off of. + init(date: Date = Date()) { + self.init(id: SnapshotIterationIdentifier(date: date), creationDate: date) + } + /// Internal method to check if an instance should be persisted based on iff it changed significantly from a previous iteration /// - Parameter existingInstance: The previous iteration to check /// - Returns: `true` if the iteration should be persisted, `false` if it represents the same data from `existingInstance`. @@ -87,4 +93,24 @@ extension SnapshotIteration { else { return true } return false } + + func datastoresToPrune(for mode: SnapshotPruneMode) -> Set { + switch mode { + case .pruneRemoved: removedDatastores + case .pruneAdded: addedDatastores + } + } + + func datastoreRootsToPrune( + for mode: SnapshotPruneMode, + options: SnapshotPruneOptions + ) -> Set { + switch (mode, options) { + case (.pruneRemoved, .pruneAndDelete): removedDatastoreRoots + case (.pruneAdded, .pruneAndDelete): addedDatastoreRoots + /// Flip the results when we aren't deleting, but only when removing from the bottom end. + case (.pruneRemoved, .pruneOnly): addedDatastoreRoots + case (.pruneAdded, .pruneOnly): [] + } + } } diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotRetentionPolicy.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotRetentionPolicy.swift new file mode 100644 index 0000000..61fcabf --- /dev/null +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotRetentionPolicy.swift @@ -0,0 +1,304 @@ +// +// SnapshotRetentionPolicy.swift +// https://github.com/mochidev/CodableDatastore +// +// Created by Dimitri Bouniol on 2024-09-02. +// Copyright © 2023-26 Mochi Development, Inc. All rights reserved. +// mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 +// + +public import Foundation + +/// A retention policy describing which snapshot iterations should be kept around on disk. +/// +/// Every write is made as a part of a top-level transaction that gets recorded atomically to disk as a snapshot iteration. These iterations can domtain edits to one or more datastores, and represent a complete view of all data at any one moment in time. Keeping iterations around allows you to rewind the datastores in a consistent and non-breaking way, though they take up disk space for all pages that are no longer current, ie. those containing deletions or older versions of records persisted to disk. +/// +/// A retention policy allows the disk persistence to automatically clean up these older iterations according to the policy you need for your app. The retention policy is only enforced when a write transaction completes, though the persistence may defer cleanup until later if write volumes are high. +public struct SnapshotRetentionPolicy: Sendable { + /// Internal predicate that tests if an iteration should be pruned. + /// + /// - Parameter creationDate: The creation date to check. + /// - Parameter distance: How far the iteration is from the current root. The current root is `0` away from itself, while the next oldest iteration has a distance of `1`. + /// - Returns: `true` if the iteration, all its ancestors, and all it's other decedents should be pruned, `false` if the next iteration should be checked. + typealias PrunePredicate = @Sendable (_ now: Date, _ creationDate: Date, _ distance: Int) -> Bool + + /// Internal marker indicating if the retention policy refers to the ``none`` policy. + let isNone: Bool + + /// Internal marker indicating if the retention policy refers to the ``indefinite`` policy. + let isIndefinite: Bool + + /// Internal predicate that tests if an iteration should be pruned. + /// + /// - Parameter iteration: The iteration to check. + /// - Parameter distance: How far the iteration is from the current root. The current root is `0` away from itself, while the next oldest iteration has a distance of `1`. + /// - Returns: `true` if the iteration, all its ancestors, and all it's other decedents should be pruned, `false` if the next iteration should be checked. + let shouldPrune: PrunePredicate + + /// Internal initializer for creating a retention policy from flags and a predicate. + /// - Parameters: + /// - isNone: Wether this represents a ``none`` policy. + /// - isIndefinite: Wether this represents an ``indefinite`` policy. + /// - shouldPrune: The predicate to use when testing retention. + init( + isNone: Bool = false, + isIndefinite: Bool = false, + shouldPrune: @escaping PrunePredicate + ) { + self.isNone = isNone + self.isIndefinite = isIndefinite + self.shouldPrune = shouldPrune + } + + /// A retention policy that only the most recent iteration should be kept around on disk, and all other iterations should be discarded. + /// + /// - Note: It will not be possible to rewind the datastore to a previous state using this policy, and other processes won't be able to read from a read-only datastore while the main one is writing to it. + public static let none = SnapshotRetentionPolicy(isNone: true) { _, _, _ in true } + + /// A retention policy that includes all iterations. + /// + /// - Note: This policy may incur a large amount of disc usage, especially on datastores with many writes. + public static let indefinite = SnapshotRetentionPolicy(isIndefinite: true) { _, _, _ in false } + + /// A retention policy that retains the specified number of transactions, including the most recent transaction. + /// + /// To retain only the most recent transaction, specify a count of `0`. To retain the last 10 transactions, in addition to the current one (leaving up to 11 on disk at once), specify a count of `10`. Specifying a negative number will assert at runtime if assertions are enabled. + /// + /// This is a useful way to ensure a minimum number of transactions will always be accessible on disk at once for other processes to read, though the exact number an app will need will depend on how often write transactions occur, and how much disk space each write transaction occupies. + /// + /// - Parameter count: The number of additional transactions to retain. + /// - Returns: A policy retaining at most `count` additional transactions. + public static func transactionCount(_ count: Int) -> Self { + assert(count >= 0, "Transaction count must be larger or equal to 0") + return SnapshotRetentionPolicy { _, _, distance in distance > count} + } + + /// A retention policy that retains transactions younger than a specified duration. + /// + /// A retention cutoff is calculated right at the moment the last write transaction takes place, subtracting the specified `timeInterval` from this moment in time. Note that this policy is sensitive to time changes on the host, as previous transactions record their creation date in a runtime agnostic way that relies on an absolute date and time. + /// + /// - Note: This policy may be more stable than ``transactionCount(_:)``, but may incur a non-constant amount of additional disk space depending on write volume. + /// - Parameter timeInterval: The time interval in seconds to indicate an acceptable retention window. + /// - Returns: A policy retaining transactions as old as the specified `timeInterval`. + public static func duration(_ timeInterval: TimeInterval) -> Self { + SnapshotRetentionPolicy { now, creationDate, _ in creationDate < now.addingTimeInterval(-timeInterval) } + } + + /// A retention policy that retains transactions younger than a specified duration. + /// + /// A retention cutoff is calculated right at the moment the last write transaction takes place, subtracting the specified `duration` from this moment in time. Note that this policy is sensitive to time changes on the host, as previous transactions record their creation date in a runtime agnostic way that relies on an absolute date and time. + /// + /// - Note: This policy may be more stable than ``transactionCount(_:)``, but may incur a non-constant amount of additional disk space depending on write volume. + /// - Parameter duration: The duration to indicate an acceptable retention window. + /// - Returns: A policy retaining transactions as old as the specified `duration`. + @_disfavoredOverload + @available(macOS 13.0, *) + public static func duration(_ duration: Duration) -> Self { + .duration(TimeInterval(duration.components.seconds)) + } + + /// A retention policy that retains transactions younger than a specified duration. + /// + /// A retention cutoff is calculated right at the moment the last write transaction takes place, subtracting the specified `duration` from this moment in time. Note that this policy is sensitive to time changes on the host, as previous transactions record their creation date in a runtime agnostic way that relies on an absolute date and time. + /// + /// - Note: This policy may be more stable than ``transactionCount(_:)``, but may incur a non-constant amount of additional disk space depending on write volume. + /// - Parameter duration: The duration in seconds to indicate an acceptable retention window. + /// - Returns: A policy retaining transactions as old as the specified `duration`. + public static func duration(_ duration: RetentionDuration) -> Self { + .duration(TimeInterval(duration.timeInterval)) + } + + /// A retention policy ensuring both specified policies are enforced before pruning a snapshot. + /// + /// This policy is useful to indicate that at least the specified number of transactions should be kept around, for at least a specified amount of time: + /// + /// persistence.retentionPolicy = .both(.transactionCount(10), and: .duration(.days(2))) + /// + /// As a result, this policy errs on the side of keeping transactions around when compared with ``either(_:or:)``. + /// + /// - Parameters: + /// - lhs: A policy to evaluate. + /// - rhs: Another policy to evaluate. + /// - Returns: A policy that ensures both `lhs` and `rhs` allow a transaction to be pruned before actually pruning it. + public static func both(_ lhs: SnapshotRetentionPolicy, and rhs: SnapshotRetentionPolicy) -> Self { + guard !lhs.isIndefinite, !rhs.isIndefinite else { return .indefinite } + if lhs.isNone { return rhs } + if rhs.isNone { return lhs } + return SnapshotRetentionPolicy { lhs.shouldIterationBePruned(now: $0, creationDate: $1, distance: $2) && rhs.shouldIterationBePruned(now: $0, creationDate: $1, distance: $2)} + } + + /// A retention policy ensuring either specified policies are enforced before pruning a snapshot. + /// + /// This policy is useful to indicate that at most the specified number of transactions should be kept around, for at most a specified amount of time: + /// + /// persistence.retentionPolicy = .either(.transactionCount(10), or: .duration(.days(2))) + /// + /// As a result, this policy errs on the side of removing transactions when compared with ``both(_:and:)``. + /// + /// - Parameters: + /// - lhs: A policy to evaluate. + /// - rhs: Another policy to evaluate. + /// - Returns: A policy that ensures either `lhs` or `rhs` allow a transaction to be pruned before actually pruning it. + public static func either(_ lhs: SnapshotRetentionPolicy, or rhs: SnapshotRetentionPolicy) -> Self { + guard !lhs.isNone, !rhs.isNone else { return .none } + if lhs.isIndefinite { return rhs } + if rhs.isIndefinite { return lhs } + return SnapshotRetentionPolicy { lhs.shouldIterationBePruned(now: $0, creationDate: $1, distance: $2) || rhs.shouldIterationBePruned(now: $0, creationDate: $1, distance: $2)} + } + + /// Internal method to check if an iteration should be pruned and removed from disk. + /// + /// - Parameter now: The current time to check against. + /// - Parameter creationDate: The creation date to check. + /// - Parameter distance: How far the iteration is from the current root. The current root is `0` away from itself, while the next oldest iteration has a distance of `1`. + /// - Returns: `true` if the iteration, all its ancestors, and all it's other decedents should be pruned, `false` if the next iteration should be checked. + func shouldIterationBePruned(now: Date = Date(), creationDate: Date, distance: Int) -> Bool { + shouldPrune(now, creationDate, distance) + } +} + +/// The duration in time snapshot iterations should be retained for. +public struct RetentionDuration: Hashable, Sendable { + /// Internal representation of a retention duration. + @usableFromInline + var timeInterval: TimeInterval + + /// Internal initializer for creating a retention duration from a time interval. + @usableFromInline + init(timeInterval: TimeInterval) { + self.timeInterval = timeInterval + } + + /// A retention duration in seconds. + @inlinable + public static func seconds(_ seconds: Int) -> Self { + RetentionDuration(timeInterval: TimeInterval(seconds)) + } + + /// A retention duration in seconds. + @inlinable + public static func seconds(_ seconds: Float) -> Self { + RetentionDuration(timeInterval: TimeInterval(seconds)) + } + + /// A retention duration in minutes. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with minutes when leap seconds are applied. + @inlinable + public static func minutes(_ minutes: Int) -> Self { + RetentionDuration(timeInterval: TimeInterval(minutes)*60) + } + + /// A retention duration in minutes. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with minutes when leap seconds are applied. + @inlinable + public static func minutes(_ minutes: Float) -> Self { + RetentionDuration(timeInterval: TimeInterval(minutes)*60) + } + + /// A retention duration in hours. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with hours on a calendar across events like seasonal time changes dependent on timezone. + @inlinable + public static func hours(_ hours: Int) -> Self { + RetentionDuration(timeInterval: TimeInterval(hours)*60*60) + } + + /// A retention duration in hours. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with hours on a calendar across events like seasonal time changes dependent on timezone. + @inlinable + public static func hours(_ hours: Float) -> Self { + RetentionDuration(timeInterval: TimeInterval(hours)*60*60) + } + + /// A retention duration in 24 hour days. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with days on a calendar across events like seasonal time changes dependent on timezone. + @inlinable + public static func days(_ days: Int) -> Self { + RetentionDuration(timeInterval: TimeInterval(days)*60*60*24) + } + + /// A retention duration in 24 hour days. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with days on a calendar across events like seasonal time changes dependent on timezone. + @inlinable + public static func days(_ days: Float) -> Self { + RetentionDuration(timeInterval: TimeInterval(days)*60*60*24) + } + + /// A retention duration in weeks, defined as seven 24 hour days. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with days on a calendar across events like seasonal time changes dependent on timezone. + @inlinable + public static func weeks(_ weeks: Int) -> Self { + RetentionDuration(timeInterval: TimeInterval(weeks)*60*60*24*7) + } + + /// A retention duration in weeks, defined as seven 24 hour days. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with days on a calendar across events like seasonal time changes dependent on timezone. + @inlinable + public static func weeks(_ weeks: Float) -> Self { + RetentionDuration(timeInterval: TimeInterval(weeks)*60*60*24*7) + } + + /// A retention duration in months, defined as thirty 24 hour days. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with days or even months on a calendar across events like seasonal time changes dependent on timezone, different length months, or leap days. + @inlinable + public static func months(_ months: Int) -> Self { + RetentionDuration(timeInterval: TimeInterval(months)*60*60*24*30) + } + + /// A retention duration in months, defined as thirty 24 hour days. + /// + /// - Warning: This duration does not take into account timezones or calendar dates, and strictly represents a duration of time. It therefore makes no guarantees to line up with days or even months on a calendar across events like seasonal time changes dependent on timezone, different length months, or leap days. + @inlinable + public static func months(_ months: Float) -> Self { + RetentionDuration(timeInterval: TimeInterval(months)*60*60*24*30) + } +} + +extension RetentionDuration: Comparable { + @inlinable + public static func < (lhs: Self, rhs: Self) -> Bool { + lhs.timeInterval < rhs.timeInterval + } +} + +extension RetentionDuration: AdditiveArithmetic { + public static let zero = RetentionDuration(timeInterval: 0) + + @inlinable + public prefix static func + (rhs: Self) -> Self { + rhs + } + + @inlinable + public prefix static func - (rhs: Self) -> Self { + RetentionDuration(timeInterval: -rhs.timeInterval) + } + + @inlinable + public static func + (lhs: Self, rhs: Self) -> Self { + RetentionDuration(timeInterval: lhs.timeInterval + rhs.timeInterval) + } + + @inlinable + public static func += (lhs: inout Self, rhs: Self) { + lhs.timeInterval += rhs.timeInterval + } + + @inlinable + public static func - (lhs: Self, rhs: Self) -> Self { + RetentionDuration(timeInterval: lhs.timeInterval - rhs.timeInterval) + } + + @inlinable + public static func -= (lhs: inout Self, rhs: Self) { + lhs.timeInterval -= rhs.timeInterval + } +} diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SparseIterationChain.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SparseIterationChain.swift new file mode 100644 index 0000000..2d562f8 --- /dev/null +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SparseIterationChain.swift @@ -0,0 +1,219 @@ +// +// SparseIterationChain.swift +// https://github.com/mochidev/CodableDatastore +// +// Created by Dimitri Bouniol on 2024-10-16. +// Copyright © 2023-26 Mochi Development, Inc. All rights reserved. +// mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 +// + +import Foundation + +/// An internal type for tracking all downstream iterations that need to be purged. +struct SparseIterationChain { + typealias IterationProxy = (iteration: SnapshotIteration.ID, creationDate: Date) + + struct Group { + var first: IterationProxy + var last: IterationProxy + var contents: [IterationProxy]? + var count: Int + + init(iteration: SnapshotIteration) { + first = (iteration.id, iteration.creationDate) + last = (iteration.id, iteration.creationDate) + contents = [(iteration.id, iteration.creationDate)] + count = 1 + } + + init(iteration: IterationProxy) { + first = iteration + last = iteration + contents = [iteration] + count = 1 + } + + mutating func prepend(iteration: IterationProxy) { + if contents != nil { + contents?.insert(iteration, at: 0) + } + first = iteration + count += 1 + } + + mutating func append(iteration: IterationProxy) { + if contents != nil { + contents?.append(iteration) + } + last = iteration + count += 1 + } + } + + var count: Int + var groups: [Group] + + init() { + self.count = 0 + self.groups = [] + } + + /// Add a snapshot iteration to the start of the chain, scheduled to be removed last. + mutating func prepend(iteration: SnapshotIteration) { + prepend(iteration: (iteration.id, iteration.creationDate)) + } + + /// Add a snapshot iteration to the start of the chain, scheduled to be removed last. + mutating func prepend(iteration: IterationProxy) { + count += 1 + if !groups.isEmpty { + /// If the group happens to have room, prepend to it and return. + guard groups[0].count >= chunkSize else { + groups[0].prepend(iteration: iteration) + return + } + /// Otherwise, empty out the subsequent group, and flow through to create a new one. + if groups.count >= 2 { + groups[0].contents = nil + } + } + groups.insert(Group(iteration: iteration), at: 0) + } + + /// Add a snapshot iteration to the end of the chain, scheduled to be removed first. + mutating func append(iteration: SnapshotIteration) { + append(iteration: (iteration.id, iteration.creationDate)) + } + + /// Add a snapshot iteration to the end of the chain, scheduled to be removed first. + mutating func append(iteration: IterationProxy) { + count += 1 + if !groups.isEmpty { + /// If the group happens to have room, append to it and return. + guard groups[groups.count-1].count >= chunkSize else { + groups[groups.count-1].append(iteration: iteration) + return + } + /// Otherwise, empty out the previous group, and flow through to create a new one. + if groups.count >= 2 { + groups[groups.count-1].contents = nil + } + } + groups.append(Group(iteration: iteration)) + } + + /// Remove and return groups from the end of the chain that should no longer be retained. If individual iterations should still be retained, they should be re-appended to the end. + mutating func removeIterations( + failing snapshotRetentionPolicy: SnapshotRetentionPolicy, + from iterationID: SnapshotIteration.ID, + now: Date + ) -> (distance: Int, removedGroups: [Group]) { + var totalDistance = 0 + var startingGroupIndex = 0 + + /// First, scan for the first iteration to anchor on. If we can't find the anchor, we can't make a reliable conclusion based on distance for this pruning operation, so artifitially pad it by setting a negative distance to be safe. + var foundAnchor = false + groupIterator: for group in groups { + if let contents = group.contents { + for proxy in contents { + if proxy.iteration == iterationID { + foundAnchor = true + break groupIterator + } + totalDistance -= 1 + } + } else { + /// We ran out of populated groups, so just check the first entry of the next one in case it happens to be the iteration we want before giving up. + if group.first.iteration == iterationID { + foundAnchor = true + } + break + } + } + /// If we still haven't found the anchor, scan backwards counting the iterations that are definitely safe to prune based on distance. + if !foundAnchor, totalDistance != -count { + totalDistance = -count + groupIterator: for group in groups.reversed() { + if let contents = group.contents { + for proxy in contents.reversed() { + totalDistance += 1 + if proxy.iteration == iterationID { + foundAnchor = true + break groupIterator + } + } + } else { + /// We ran out of populated groups, so just check the last entry of the previous one in case it happens to be the iteration we want before giving up. + totalDistance += 1 + if group.last.iteration == iterationID { + foundAnchor = true + } + break + } + } + } + + /// Determine the first group that fails the check by crawling them in reverse. + totalDistance += count + startingGroupIndex = groups.count + for group in groups.reversed() { + totalDistance -= group.count + count -= group.count + startingGroupIndex -= 1 + if !snapshotRetentionPolicy.shouldIterationBePruned(now: now, creationDate: group.first.creationDate, distance: totalDistance) { + break + } + } + + guard startingGroupIndex < groups.count + else { return (distance: totalDistance, removedGroups: []) } + + /// Collect the groups that will be returned, and remove them. + let groupsToRemove = Array(groups.suffix(groups.count - startingGroupIndex)) + groups.removeLast(groups.count - startingGroupIndex) + return (distance: totalDistance, removedGroups: groupsToRemove) + } + + var first: IterationProxy? { + groups.first?.first + } + + var last: IterationProxy? { + groups.last?.last + } + + /// The ideal size of a group before it is emptied and a new group is formed. + var chunkSize: Int { + /// Mapping count => group size => number of groups + /// `<1 (2^0)` => `1 << max(-9, 8)` = `256` => `1` group + /// `2 (2^1)` => `1 << max(-8, 8)` = `256` => `1` group + /// `4 (2^2)` => `1 << max(-7, 8)` = `256` => `1` group + /// `8 (2^3)` => `1 << max(-6, 8)` = `256` => `1` group + /// `16 (2^4)` => `1 << max(-5, 8)` = `256` => `1` group + /// `32 (2^5)` => `1 << max(-4, 8)` = `256` => `1` group + /// `64 (2^6)` => `1 << max(-3, 8)` = `256` => `1` group + /// `128 (2^7)` => `1 << max(-2, 8)` = `256` => `1` group + /// `256 (2^8)` => `1 << max(-1, 8)` = `256` => `1` group + /// `512 (2^9)` => `1 << max(0, 8)` = `256` => `2` groups + /// `1,024 (2^10)` => `1 << max(1, 8)` = `256` => `4` groups + /// `2,048 (2^11)` => `1 << max(2, 8)` = `256` => `8` groups + /// `4,096 (2^12)` => `1 << max(3, 8)` = `256` => `16` groups + /// `8,192 (2^13)` => `1 << max(4, 8)` = `256` => `32` groups + /// `16,384 (2^14)` => `1 << max(5, 8)` = `256` => `64` groups + /// `32,768 (2^15)` => `1 << max(6, 8)` = `256` => `128` groups + /// `65,536 (2^16)` => `1 << max(7, 8)` = `256` => `256` groups + /// `131,072 (2^17)` => `1 << max(8, 8)` = `256` => `512` groups + /// `262,144 (2^18)` => `1 << max(9, 8)` = `512` => `512` groups + /// `524,288 (2^19)` => `1 << max(10, 8)` = `1024` => `512` groups + /// `...` + 1 << max(Int.bitWidth - 1 - (max(count - 1, 0)).leadingZeroBitCount - 8, 8) + } +} + +extension SparseIterationChain { + enum State { + case forwardEditsOnly + case crawling(Task) + case complete + } +} diff --git a/Tests/CodableDatastoreTests/DiskPersistenceDatastoreRetentionTests.swift b/Tests/CodableDatastoreTests/DiskPersistenceDatastoreRetentionTests.swift new file mode 100644 index 0000000..9431be5 --- /dev/null +++ b/Tests/CodableDatastoreTests/DiskPersistenceDatastoreRetentionTests.swift @@ -0,0 +1,372 @@ +// +// DiskPersistenceDatastoreRetentionTests.swift +// https://github.com/mochidev/CodableDatastore +// +// Created by Dimitri Bouniol on 2024-09-09. +// Copyright © 2023-26 Mochi Development, Inc. All rights reserved. +// mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 +// + +#if !canImport(Darwin) +@preconcurrency import Foundation +#endif +import XCTest +@testable import CodableDatastore + +final class DiskPersistenceDatastoreRetentionTests: XCTestCase, @unchecked Sendable { + var temporaryStoreURL: URL = FileManager.default.temporaryDirectory + + override func setUp() async throws { + temporaryStoreURL = FileManager.default.temporaryDirectory.appendingPathComponent(ProcessInfo.processInfo.globallyUniqueString, isDirectory: true); + } + + override func tearDown() async throws { + try? FileManager.default.removeItem(at: temporaryStoreURL) + } + + func testTransactionCountPrunedDatastoreStillReadable() async throws { + struct TestFormat: DatastoreFormat { + enum Version: Int, CaseIterable { + case zero + } + + struct Instance: Codable, Identifiable { + var id: String + var value: String + var index: Int + var bucket: Int + } + + static let defaultKey: DatastoreKey = "test" + static let currentVersion = Version.zero + + let index = OneToOneIndex(\.index) + @Direct var bucket = Index(\.bucket) + } + + let max = 1000 + + do { + let persistence = try DiskPersistence(readWriteURL: temporaryStoreURL) + + let datastore = Datastore.JSONStore( + persistence: persistence, + format: TestFormat.self, + migrations: [ + .zero: { decoder, data in + try decoder.decode(TestFormat.Instance.self, from: data) + } + ] + ) + + await persistence.setTransactionRetentionPolicy(.transactionCount(0)) + try await persistence.createPersistenceIfNecessary() + + for index in 0..