File IO Improvements

Sources/CodableDatastore/Helpers/AsyncFileReader.swift

@@ -0,0 +1,195 @@
+//
+//  AsyncFileReader.swift
+//  https://github.com/mochidev/CodableDatastore
+//
+//  Created by Dimitri Bouniol on 2026-09-04.
+//  Copyright © 2023-26 Mochi Development, Inc. All rights reserved.
+//  mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325
+//
+
+import Bytes
+import Foundation
+import QuestionableConcurrency
+
+/// An asynchronous, low-latency file reader that consumes the file as fast as possible, but allows multiple readers to read it at their own pace.
+class AsyncFileReader: @unchecked Sendable {
+    static let chunkSize = 1024
+
+    let url: URL
+    var readerTask: Task<Void, Never>! = nil
+
+    var resultsLock = UnfairLock()
+
+    var writeHead: Int = 0
+    var byteChunks: [Bytes] = []
+    var finalResult: Result<Void, any Error>?
+    var continuations: [CheckedContinuation<Void, Never>] = []
+
+    init(contentsOf url: URL) {
+        self.url = url
+        self.readerTask = Task(name: "AsyncFileReader") { await self.startReading() }
+    }
+
+    func startReading() async {
+        do {
+            #if canImport(Darwin)
+            if #available(macOS 12.0, iOS 15, watchOS 8, tvOS 15, *) {
+                for try await byte in url.resourceBytes {
+                    resultsLock.withLock {
+                        let chunkIndex = writeHead/Self.chunkSize
+                        let indexInChunk = writeHead % Self.chunkSize
+
+                        /// If there isn't enough space, allocate a new chunk.
+                        if chunkIndex >= byteChunks.count {
+                            byteChunks.append(Bytes(repeating: 0, count: Self.chunkSize))
+                        }
+
+                        /// Write the byte and advance the head.
+                        byteChunks[chunkIndex][indexInChunk] = byte
+
+                        writeHead += 1
+
+                        /// If anyone is waiting on us, let them know.
+                        if !continuations.isEmpty {
+                            for continuation in continuations {
+                                continuation.resume()
+                            }
+                            continuations.removeAll(keepingCapacity: true)
+                        }
+                    }
+                }
+            } else {
+                let data = try Data(contentsOf: url)
+                for byte in data {
+                    resultsLock.withLock {
+                        let chunkIndex = writeHead/Self.chunkSize
+                        let indexInChunk = writeHead % Self.chunkSize
+
+                        /// If there isn't enough space, allocate a new chunk.
+                        if chunkIndex >= byteChunks.count {
+                            byteChunks.append(Bytes(repeating: 0, count: Self.chunkSize))
+                        }
+
+                        /// Write the byte and advance the head.
+                        byteChunks[chunkIndex][indexInChunk] = byte
+
+                        writeHead += 1
+
+                        /// If anyone is waiting on us, let them know.
+                        if !continuations.isEmpty {
+                            for continuation in continuations {
+                                continuation.resume()
+                            }
+                            continuations.removeAll(keepingCapacity: true)
+                        }
+                    }
+                }
+            }
+            #else
+            let data = try Data(contentsOf: url)
+            for byte in data {
+                resultsLock.withLock {
+                    let chunkIndex = writeHead/Self.chunkSize
+                    let indexInChunk = writeHead % Self.chunkSize
+
+                    /// If there isn't enough space, allocate a new chunk.
+                    if chunkIndex >= byteChunks.count {
+                        byteChunks.append(Bytes(repeating: 0, count: Self.chunkSize))
+                    }
+
+                    /// Write the byte and advance the head.
+                    byteChunks[chunkIndex][indexInChunk] = byte
+
+                    writeHead += 1
+
+                    /// If anyone is waiting on us, let them know.
+                    if !continuations.isEmpty {
+                        for continuation in continuations {
+                            continuation.resume()
+                        }
+                        continuations.removeAll(keepingCapacity: true)
+                    }
+                }
+            }
+            #endif
+
+            /// Let the reader know the sequence is done.
+            resultsLock.withLock {
+                finalResult = .success(())
+                /// If anyone is waiting on us, let them know.
+                if !continuations.isEmpty {
+                    for continuation in continuations {
+                        continuation.resume()
+                    }
+                    continuations.removeAll(keepingCapacity: true)
+                }
+            }
+        } catch {
+            /// Let the reader know the sequence failed.
+            resultsLock.withLock {
+                finalResult = .failure(error)
+                /// If anyone is waiting on us, let them know.
+                if !continuations.isEmpty {
+                    for continuation in continuations {
+                        continuation.resume()
+                    }
+                    continuations.removeAll(keepingCapacity: true)
+                }
+            }
+        }
+    }
+
+    func byte(for readHead: Int) async throws -> Byte? {
+        /// If we overtake the write head, check if we are done, otherwise suspend until there are more bytes
+        resultsLock.unsafeLock()
+        if readHead < writeHead {
+            resultsLock.unsafeUnlock()
+        } else {
+            if let finalResult {
+                resultsLock.unsafeUnlock()
+                /// Return nil or throw the error we encountered while reading.
+                try finalResult.get()
+                return nil
+            }
+
+            await withCheckedContinuation { continuation in
+                continuations.append(continuation)
+                resultsLock.unsafeUnlock()
+            }
+
+            /// Check one more time if we are actually at the end and an error was thrown or if we finished.
+            let (writeHead, finalResult) = resultsLock.withLock { (self.writeHead, self.finalResult) }
+            if readHead >= writeHead, let finalResult {
+                try finalResult.get()
+                return nil
+            }
+        }
+
+        let chunkIndex = readHead/Self.chunkSize
+        let indexInChunk = readHead % Self.chunkSize
+        return resultsLock.withLock { byteChunks[chunkIndex][indexInChunk] }
+    }
+}
+
+extension AsyncFileReader: AsyncSequence {
+    nonisolated func makeAsyncIterator() -> AsyncIterator {
+        AsyncIterator(fileReader: self)
+    }
+
+    struct AsyncIterator: AsyncIteratorProtocol {
+        let fileReader: AsyncFileReader
+        var readHead: Int = 0
+
+        mutating func next() async throws -> Byte? {
+            let readHead = readHead
+            self.readHead += 1
+            return try await fileReader.byte(for: readHead)
+        }
+
+        mutating func next(isolation actor: isolated (any Actor)?) async throws(any Error) -> Byte? {
+            let readHead = readHead
+            self.readHead += 1
+            return try await fileReader.byte(for: readHead)
+        }
+    }
+}

Sources/CodableDatastore/Helpers/Swift6.1+Compatibility.swift

@@ -0,0 +1,17 @@
+//
+//  Swift6.1+Compatibility.swift
+//  https://github.com/mochidev/CodableDatastore
+//
+//  Created by Dimitri Bouniol on 2026-08-31.
+//  Copyright © 2023-26 Mochi Development, Inc. All rights reserved.
+//  mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325
+//
+
+#if compiler(<6.2)
+extension Task where Failure == Never {
+    @discardableResult
+    init(name: String?, priority: TaskPriority? = nil, operation: sending @escaping @isolated(any) () async -> Success) {
+        self.init(priority: priority, operation: operation)
+    }
+}
+#endif

Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndex.swift

@@ -153,7 +153,7 @@ extension DiskPersistence.Datastore.Index {
 
         /// Encode the provided manifest, and write it to disk.
         let data = Data(manifest.bytes)
-        try data.write(to: manifestURL, options: .atomic)
+        try data.write(to: manifestURL, options: [])
         isPersisted = true
         await datastore.mark(identifier: id, asLoaded: true)
     }

Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreIndexManifest.swift

@@ -95,18 +95,10 @@ extension DatastoreIndexManifest {
 
 extension DatastoreIndexManifest {
     init(contentsOf url: URL, id: ID) async throws {
-#if canImport(Darwin)
-        if #available(macOS 12.0, iOS 15, watchOS 8, tvOS 15, *) {
-            try await self.init(sequence: AnyReadableSequence(url.resourceBytes), id: id)
-        } else {
-            try await self.init(sequence: AnyReadableSequence(try Data(contentsOf: url)), id: id)
-        }
-#else
-        try await self.init(sequence: AnyReadableSequence(try Data(contentsOf: url)), id: id)
-#endif
+        try await self.init(sequence: AsyncFileReader(contentsOf: url), id: id)
     }
 
-    init(sequence: AnyReadableSequence<Byte, any Error>, id: ID) async throws {
+    init(sequence: AsyncFileReader, id: ID) async throws {
         self.id = id
 
         var iterator = sequence.makeAsyncIterator()

Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastorePage.swift

@@ -92,21 +92,11 @@ extension DiskPersistence.Datastore.Page {
 // MARK: - Persistence
 
 extension DiskPersistence.Datastore.Page {
-    private var readableSequence: AnyReadableSequence<Byte, any Error> {
-        get throws {
-#if canImport(Darwin)
-            if #available(macOS 12.0, iOS 15, watchOS 8, tvOS 15, *) {
-                return AnyReadableSequence(pageURL.resourceBytes)
-            } else {
-                return AnyReadableSequence(try Data(contentsOf: pageURL))
-            }
-#else
-            return AnyReadableSequence(try Data(contentsOf: pageURL))
-#endif
-        }
+    private var readableSequence: AsyncFileReader {
+        AsyncFileReader(contentsOf: pageURL)
     }
 
-    private nonisolated func performRead(sequence: AnyReadableSequence<Byte, any Error>) async throws -> MultiplexedAsyncSequence<AnyReadableSequence<DatastorePageEntryBlock, any Error>> {
+    private nonisolated func performRead(sequence: AsyncFileReader) async throws -> MultiplexedAsyncSequence<AnyReadableSequence<DatastorePageEntryBlock, any Error>> {
         var iterator = sequence.makeBufferedIterator()
 
         try await iterator.check(Self.header)
@@ -146,7 +136,7 @@ extension DiskPersistence.Datastore.Page {
             }
 
             let readerTask = Task {
-                try await performRead(sequence: try readableSequence)
+                try await performRead(sequence: readableSequence)
             }
             isPersisted = true
             blocksReaderTask = readerTask
@@ -166,7 +156,7 @@ extension DiskPersistence.Datastore.Page {
         try FileManager.default.createDirectory(at: pageURL.deletingLastPathComponent(), withIntermediateDirectories: true)
 
         /// Write the bytes for the page to disk.
-        try Data(bytes).write(to: pageURL, options: .atomic)
+        try Data(bytes).write(to: pageURL, options: [])
         isPersisted = true
         await datastore.mark(identifier: id, asLoaded: true)
     }

Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/DatastoreRoot.swift

@@ -131,7 +131,7 @@ extension DiskPersistence.Datastore.RootObject {
 
         /// Encode the provided manifest, and write it to disk.
         let data = try JSONEncoder.shared.encode(rootObject)
-        try data.write(to: rootObjectURL, options: .atomic)
+        try data.write(to: rootObjectURL, options: [])
         isPersisted = true
         await datastore.mark(identifier: id, asLoaded: true)
     }

Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift

@@ -169,7 +169,7 @@ extension Snapshot {
 
         /// Encode the provided iteration, and write it to disk.
         let data = try JSONEncoder.shared.encode(iteration)
-        try data.write(to: iterationURL, options: .atomic)
+        try data.write(to: iterationURL, options: [])
 
         /// Update the cache since we know what it should be.
         cachedIteration = iteration