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/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift index ffddbd4..2412473 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift @@ -87,6 +87,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 d3030d3..de7d6c8 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift @@ -155,6 +155,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 URLError.fileDoesNotExist, CocoaError.fileReadNoSuchFile, CocoaError.fileNoSuchFile, POSIXError.ENOENT { + 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) + try? fileManager.removeDirectoryIfEmpty(url: pageURL.deletingLastPathComponent(), recursivelyRemoveParents: true) + } + + 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 URLError.fileDoesNotExist, CocoaError.fileReadNoSuchFile, CocoaError.fileNoSuchFile, POSIXError.ENOENT { + 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) + try? fileManager.removeDirectoryIfEmpty(url: pageURL.deletingLastPathComponent(), recursivelyRemoveParents: true) + } + } + + /// 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) + try? fileManager.removeDirectoryIfEmpty(url: rootURL.deletingLastPathComponent(), recursivelyRemoveParents: true) + } + } + func index(for identifier: Index.ID) -> Index { if let index = trackedIndexes[identifier]?.value { return index @@ -216,14 +319,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 3da6c28..ff9ab03 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift @@ -28,6 +28,12 @@ public actor DiskPersistence: Persistence { var lastMutatingTransaction: Transaction? var rootTransactionStream = TransactionStream() + var _transactionRetentionPolicy: SnapshotRetentionPolicy = .indefinite + + var nextSnapshotIterationCandidateToEnforce: (snapshot: Snapshot, iteration: SnapshotIteration, taskPriority: TaskPriority)? + var snapshotIterationPruningTask: Task? + var cachedSnapshotIterationChain: [SnapshotIdentifier: [(id: SnapshotIteration.ID, creationDate: Date)]] = [:] + /// Shared caches across all snapshots and datastores. var rollingRootObjectCacheIndex = 0 var rollingRootObjectCache: [Datastore.RootObject] = [] @@ -61,6 +67,10 @@ public actor DiskPersistence: Persistence { storeURL = readOnlyURL } + deinit { + snapshotIterationPruningTask?.cancel() + } + /// The default URL to use for disk persistences. static var defaultURL: URL { // TODO: Make non-throwing: https://github.com/mochidev/CodableDatastore/issues/15 @@ -252,7 +262,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 @@ -570,7 +580,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 @@ -582,8 +592,182 @@ extension DiskPersistence { root: root.id ) } + return (snapshot, iteration) + } + } + + enforceRetentionPolicy(snapshot: currentSnapshot, fromIteration: persistedIteration) + } +} + +// 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. + /// + /// - Parameter policy: The new policy to enforce on write. + /// + /// - SeeAlso: ``SnapshotRetentionPolicy``. + public func setTransactionRetentionPolicy(_ policy: SnapshotRetentionPolicy) async { + _transactionRetentionPolicy = policy + /// Configure snapshots to start caching iterations to speed up pruning. + for (_, snapshot) in snapshots { + await snapshot.setExtendedIterationCacheEnabled(!_transactionRetentionPolicy.isIndefinite) + } + /// Cancel any in-progress pruning tasks + snapshotIterationPruningTask?.cancel() + } + + /// Enforce the retention policy on the persistence immediately. + /// + /// - Note: Transaction retention policies are enforced after ever 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 { + // TODO: Don't create any snapshots if they don't exist yet + let info = try? await self.readingCurrentSnapshot { snapshot in + try await snapshot.readingManifest { manifest, iteration in + (snapshot: snapshot, iteration: iteration) + } + } + + if let (snapshot, iteration) = info { + enforceRetentionPolicy(snapshot: snapshot, fromIteration: iteration, taskPriority: Task.currentPriority) + } + + await finishTransactionCleanup() + } +} + +extension DiskPersistence { + /// Internal method to envorce the retention policy after a transaction is written. + private func enforceRetentionPolicy( + snapshot: Snapshot, + fromIteration iteration: SnapshotIteration, + taskPriority: TaskPriority = .background + ) { +// print("Scheduling pruning starting at \(iteration.id)") + nextSnapshotIterationCandidateToEnforce = (snapshot, iteration, taskPriority) + + /// If a task is on-going, just set the next iteration and stop here — the task should pick up the latest scheduled iteration automatically. + guard snapshotIterationPruningTask == nil else { return } + + /// Update the next snapshot iteration we should be checking, and enqueue a task since we know one isn't currently running. + checkNextSnapshotIterationCandidateForPruning() + } + + /// Private method to check the next candidate for pruning. + /// + /// First, this method walks down the linked list defining the iteration chain, from newest to oldest, and collects the iterations that should be pruned. Then, it iterates that list in reverse (from oldest to newest) actually removing the iterations as they are encountered. + /// - Note: This method should only ever be called when it is known that no `snapshotIterationPruningTask` is ongoing (it is nil), or when one just finishes. + @discardableResult + private func checkNextSnapshotIterationCandidateForPruning() -> Task? { + print("Checking for pruning work.") + let transactionRetentionPolicy = _transactionRetentionPolicy + let iterationCandidate = nextSnapshotIterationCandidateToEnforce + + snapshotIterationPruningTask = nil + nextSnapshotIterationCandidateToEnforce = nil + + guard + let (snapshot, iteration, taskPriority) = iterationCandidate, + !transactionRetentionPolicy.isIndefinite + else { + print("No more iterations to prune, stopping here.") + return nil + } + + snapshotIterationPruningTask = Task.detached(priority: taskPriority) { + print("Pruning started for \(iteration.id).") + await snapshot.setExtendedIterationCacheEnabled(true) + do { + var iterations: [SnapshotIteration.ID] = [] + var distance = 1 + var mainlineSuccessorIteration = iteration + var currentIteration = iteration + + + + /// First, walk the preceding iteration chain to the oldest iteration we can open, collecting the ones that should be pruned. + while let precedingIterationID = currentIteration.precedingIteration, let precedingIteration = try? await snapshot.loadIteration(for: precedingIterationID) { + try Task.checkCancellation() + + if !iterations.isEmpty || transactionRetentionPolicy.shouldIterationBePruned(creationDate: precedingIteration.creationDate, distance: distance) { + iterations.append(precedingIteration.id) + } else { + mainlineSuccessorIteration = precedingIteration + } + currentIteration = precedingIteration + + distance += 1 + + if distance % 5000 == 0 { + print("Found \(iterations.count) iterations to prune. Keeping \(distance - iterations.count) iterations.") + } + +// await Task.yield() + } + + print("Will prune \(iterations.count) iterations. Keeping \(distance - iterations.count) iterations.") + + /// Prune iterations from oldest to newest. + while let iterationID = iterations.popLast(), let iteration = try await snapshot.loadIteration(for: iterationID) { + let index = iterations.count /// The current index, since we just removed the last element + let mainlineSuccessorIterationID = index > 0 ? iterations[index-1] : mainlineSuccessorIteration.id + + if index % 1000 == 0 { + print("\(index) iterations left to delete.") + } + + var iterationsToPrune: [SnapshotIteration] = [] + var successorCandidatesToCheck = iteration.successiveIterations + successorCandidatesToCheck.removeAll { $0 == mainlineSuccessorIterationID } + + /// Walk the 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 snapshot.loadIteration(for: successorCandidateID) + else { continue } + + iterationsToPrune.append(successorIteration) + successorCandidatesToCheck.insert(contentsOf: successorIteration.successiveIterations, at: 0) + await Task.yield() + } + + /// First, remove the branch of iterations based on the one we are removing, but representing a history that was previously reverted. + /// 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 snapshot.pruneIteration(iteration, mode: .pruneAdded, shouldDelete: true) + } + + /// Finally, prune the iteration itself. + try await snapshot.pruneIteration(iteration, mode: .pruneRemoved, shouldDelete: true) + } + + try await snapshot.pruneIteration(mainlineSuccessorIteration, mode: .pruneRemoved, shouldDelete: false) + try await snapshot.drainPrunedIterations() + print("Pruning complete!") + } catch { + try? await snapshot.drainPrunedIterations() + print("Pruning stopped: \(error)") } + + await self.checkNextSnapshotIterationCandidateForPruning()?.value } + + return snapshotIterationPruningTask + } + + /// 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 in cases when disk resources must be cleared before progressing to another step. + public func finishTransactionCleanup() async { + await snapshotIterationPruningTask?.value } } diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/FileManager+Helpers.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/FileManager+Helpers.swift new file mode 100644 index 0000000..6906221 --- /dev/null +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/FileManager+Helpers.swift @@ -0,0 +1,30 @@ +// +// FileManager+Helpers.swift +// CodableDatastore +// +// Created by Dimitri Bouniol on 2024-09-08. +// Copyright © 2023-24 Mochi Development, Inc. All rights reserved. +// + +import Foundation + +enum DirectoryRemovalError: Error { + case missingEnumerator +} + +extension FileManager { + func removeDirectoryIfEmpty(url: URL, recursivelyRemoveParents: Bool) throws { + 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 + } + + try self.removeItem(at: url) + + guard recursivelyRemoveParents else { return } + try self.removeDirectoryIfEmpty(url: url.deletingLastPathComponent(), recursivelyRemoveParents: recursivelyRemoveParents) + } +} diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift index bdd0839..f08ce61 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift @@ -32,8 +32,9 @@ 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 /// A transaction stream for manifest updates, so reads and writes can be serialized in request order. var manifestTransactionStream = TransactionStream() @@ -41,14 +42,29 @@ actor Snapshot { /// The loaded datastores. var datastores: [DatastoreIdentifier: DiskPersistence.Datastore] = [:] + /// The chain of iterations + var iterationChain: SparseIterationChain + var iterationChainState: IterationChainState + +// private let (pruningTasksStream, pruningTasksStreamProvider) = AsyncStream.makeStream(of: Task.self) +// private var pruningTasksIterator: AsyncStream>.AsyncIterator + 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 +// prunedIterationIterator = prunedIterationStream.makeAsyncIterator() } } @@ -126,20 +142,224 @@ extension Snapshot { } } + func setExtendedIterationCacheEnabled(_ isEnabled: Bool) async { + isExtendedIterationCacheEnabled = isEnabled + + await invalidateIterationChainState() + } + + private func invalidateIterationChainState() async { + switch iterationChainState { + case .forwardEditsOnly: + /// If we are currently only collecting forward edits, and the extended cache was just enabled, start the crawling process. + if isExtendedIterationCacheEnabled { + iterationChainState = .crawling(Task { + do { + try await crawlIterations() + iterationChainState = .complete + } catch { + print("Error crawling iterations: \(error)") + iterationChainState = .forwardEditsOnly + } + }) + } + 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 + } + } + + func crawlIterations() async throws { + var currentIteration: SnapshotIteration + if let currentIterationID = iterationChain.last?.iteration { + currentIteration = try self.loadIterationNoCache(for: currentIterationID) + } else { + // TODO: We need to do this earlier, otheriwse we may end up duplicating an entry or skipping others, as the cached manifest may have changed since the task was initially started + /// Load the manifest so we have a fresh copy, unless we have a cached copy already. + var manifest = try cachedManifest ?? self.loadManifest() + + /// If there is no ID here, we are basically done, as we have nothing to crawl + guard let currentIterationID = manifest.currentIteration + else { return } + + /// Make sure not to await adding this first entry? + currentIteration = try self.loadIterationNoCache(for: currentIterationID) + iterationChain.append(iteration: currentIteration) + } + + /// Walk the preceding iteration chain to the oldest iteration we can open, collecting the ones that should be pruned. + while let precedingIterationID = currentIteration.precedingIteration, let precedingIteration = try? await loadIteration(for: precedingIterationID) { + try Task.checkCancellation() + + if !iterations.isEmpty || transactionRetentionPolicy.shouldIterationBePruned(creationDate: precedingIteration.creationDate, distance: distance) { + iterations.append(precedingIteration.id) + } else { + mainlineSuccessorIteration = precedingIteration + } + currentIteration = precedingIteration + + distance += 1 + + if distance % 5000 == 0 { + print("Found \(iterations.count) iterations to prune. Keeping \(distance - iterations.count) iterations.") + } + +// await Task.yield() + } + } + /// 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 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 + if cachedIterations.count >= 256, let firstKey = cachedIterations.keys.first { + cachedIterations.removeValue(forKey: firstKey) + } + cachedIterations[iteration.id] = iteration return iteration } catch { throw error } } + /// Load an iteration from 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) + } + + func pruneIteration(_ iteration: SnapshotIteration, mode: SnapshotPruneMode, shouldDelete: Bool) async throws { + let pruneTask = Task { + try await pruneIteration(iteration, mode: mode) + return iteration + } + lastPruningTask = Task { [lastPruningTask] in + try await lastPruningTask?.value + let iteration = try await pruneTask.value + if shouldDelete { + deleteIteration(iteration) + } + } + 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() + pruningWatermark = 0 + try await lastPruningTask?.value + await Task.yield() + } + } + + 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 therefor 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 URLError.fileDoesNotExist, CocoaError.fileReadNoSuchFile, CocoaError.fileNoSuchFile, POSIXError.ENOENT { + /// 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 URLError.fileDoesNotExist, CocoaError.fileReadNoSuchFile, CocoaError.fileNoSuchFile, POSIXError.ENOENT { + /// This datastore root is already gone. + } catch { + print("Could not prune datastore root \(datastoreRoot): \(error)") + throw error + } + datastoreRootsToPrune.remove(datastoreRoot) + } + /// If any regerences 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 URLError.fileDoesNotExist, CocoaError.fileReadNoSuchFile, CocoaError.fileNoSuchFile, POSIXError.ENOENT { + /// 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 URLError.fileDoesNotExist, CocoaError.fileReadNoSuchFile, CocoaError.fileNoSuchFile, POSIXError.ENOENT { + /// 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) + try? FileManager.default.removeDirectoryIfEmpty(url: iterationURL.deletingLastPathComponent(), recursivelyRemoveParents: true) + } + /// 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. @@ -157,7 +377,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. @@ -168,7 +388,10 @@ extension Snapshot { try data.write(to: iterationURL, options: .atomic) /// 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. @@ -195,15 +418,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) { @@ -211,10 +427,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) } @@ -225,6 +441,10 @@ extension Snapshot { 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) + // TODO: Update the target pruning snapshotID return returnValue } } @@ -246,15 +466,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) { @@ -286,6 +498,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. @@ -365,3 +587,114 @@ extension Snapshot { } } } + +struct SparseIterationChain { + var count: Int + var groups: [Group] + + init() { + self.count = 0 + self.groups = [] + } + + struct Group { + var first: (iteration: SnapshotIteration.ID, creationDate: Date) + var last: (iteration: SnapshotIteration.ID, creationDate: Date) + var contents: [(iteration: SnapshotIteration.ID, creationDate: Date)]? + var count: Int + + init(iteration: SnapshotIteration) { + first = (iteration.id, iteration.creationDate) + last = (iteration.id, iteration.creationDate) + contents = [(iteration.id, iteration.creationDate)] + count = 1 + } + + mutating func prepend(iteration: SnapshotIteration) { + if contents != nil { + contents?.insert((iteration.id, iteration.creationDate), at: 0) + } + first = (iteration.id, iteration.creationDate) + count += 1 + } + + mutating func append(iteration: SnapshotIteration) { + if contents != nil { + contents?.append((iteration.id, iteration.creationDate)) + } + last = (iteration.id, iteration.creationDate) + count += 1 + } + } + + mutating func prepend(iteration: SnapshotIteration) { + 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 + groups[0].contents = nil + } + groups.insert(Group(iteration: iteration), at: 0) + } + + mutating func append(iteration: SnapshotIteration) { + 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 + groups[groups.count-1].contents = nil + } + groups.append(Group(iteration: iteration)) + } + + mutating func removeIterations(failing snapshotRetentionPolicy: SnapshotRetentionPolicy) -> [Group] { + var groupsToRemove: [Group] = [] + + return groupsToRemove + } + + var first: (iteration: SnapshotIteration.ID, creationDate: Date)? { + groups.first?.first + } + + var last: (iteration: SnapshotIteration.ID, creationDate: Date)? { + groups.last?.last + } + + var chunkSize: Int { + 1 << max(Int.bitWidth - 1 - count.leadingZeroBitCount - 8, 8) + } +} + +enum IterationChainState { + case forwardEditsOnly + case crawling(Task) + case complete +} + + +/* + + - enforceRetentionPolicy enabled + 1. iteration chain created + 2. each new iteration added to the front of it + 3. iterations are crawled adding to the back of it + 4. only the first and last chunks are hydrated + 5. iteration chain owned by an actor so access is safe + 6. separate flag controls if iterations have been fully crawled + 7. if iterations are still being crawled, don't do anything + 8. if iterations completed crawling, scan for first one to delete + 9. delete from oldest to first item + 10. if final group is sparse, re-hydrate it + 11. if deleting, set new final iteration + 12. if not deleting + + */ diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotIteration.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotIteration.swift index e40e32b..d0d8c8c 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotIteration.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotIteration.swift @@ -74,6 +74,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`. @@ -83,4 +89,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..d2cffb4 --- /dev/null +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/SnapshotRetentionPolicy.swift @@ -0,0 +1,294 @@ +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 (_ 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 { creationDate, _ in creationDate < Date(timeIntervalSinceNow: -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(creationDate: $0, distance: $1) && rhs.shouldIterationBePruned(creationDate: $0, distance: $1)} + } + + /// 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(creationDate: $0, distance: $1) || rhs.shouldIterationBePruned(creationDate: $0, distance: $1)} + } + + /// Internal method to check if an iteration should be pruned and removed from disk. + /// + /// - 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(creationDate: Date, distance: Int) -> Bool { + shouldPrune(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/Tests/CodableDatastoreTests/DiskPersistenceDatastoreRetentionTests.swift b/Tests/CodableDatastoreTests/DiskPersistenceDatastoreRetentionTests.swift new file mode 100644 index 0000000..7b2ca5c --- /dev/null +++ b/Tests/CodableDatastoreTests/DiskPersistenceDatastoreRetentionTests.swift @@ -0,0 +1,307 @@ +// +// DiskPersistenceDatastoreRetentionTests.swift +// CodableDatastore +// +// Created by Dimitri Bouniol on 2024-09-09. +// Copyright © 2023-24 Mochi Development, Inc. All rights reserved. +// + +#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: { data, decoder in + try decoder.decode(TestFormat.Instance.self, from: data) + } + ] + ) + + await persistence.setTransactionRetentionPolicy(.transactionCount(0)) + try await persistence.createPersistenceIfNecessary() + + for index in 0..