diff --git a/Packages/SpaceTraceKit/Package.swift b/Packages/SpaceTraceKit/Package.swift index 3813dc3..d1ecedf 100644 --- a/Packages/SpaceTraceKit/Package.swift +++ b/Packages/SpaceTraceKit/Package.swift @@ -51,6 +51,7 @@ let package = Package( name: "SpaceTracePersistenceTests", dependencies: [ "SpaceTraceApplication", + "SpaceTraceFileSystem", "SpaceTracePersistence", ] ), diff --git a/Packages/SpaceTraceKit/README.md b/Packages/SpaceTraceKit/README.md index 87ff341..b1f13ae 100644 --- a/Packages/SpaceTraceKit/README.md +++ b/Packages/SpaceTraceKit/README.md @@ -7,11 +7,11 @@ | Module | Responsibility | Platform coupling | | --- | --- | --- | | `SpaceTraceDomain` | Validated storage quantities, observation identities, coverage, and deltas | None | -| `SpaceTraceApplication` | Application-owned ports and crash-consistency contracts | None | -| `SpaceTraceFileSystem` | FSEvents callback bridge and conservative flag interpretation | CoreServices | -| `SpaceTracePersistence` | Actor-isolated raw SQLite prototype for the event journal | SQLite3 | +| `SpaceTraceApplication` | Dirty-region planning, calibration orchestration, ports, and crash-consistency contracts | Foundation | +| `SpaceTraceFileSystem` | FSEvents callback bridge, conservative flag interpretation, and application-semantic mapping | CoreServices | +| `SpaceTracePersistence` | Actor-isolated raw SQLite prototype for cursor, dirty work, and conditional finalization | SQLite3 | -The filesystem and persistence modules are architecture-spike implementations for proposed ADR-003 and ADR-004. They are not yet accepted production decisions and are not connected to a user-visible workflow. +The filesystem, application pipeline, and persistence modules are architecture-spike implementations for proposed ADR-003 and ADR-004. The pipeline is validated with an injected scanner; a production metadata enumerator and user-visible workflow are not connected yet. ## Verify diff --git a/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/DirtyRegionPlanner.swift b/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/DirtyRegionPlanner.swift new file mode 100644 index 0000000..89a04ba --- /dev/null +++ b/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/DirtyRegionPlanner.swift @@ -0,0 +1,141 @@ +import Foundation + +public struct DirtyRegionPlanner: Sendable { + public init() {} + + /// Converts invalidation hints into the narrowest safe directory work. + /// Replay duplicates are ignored unless they carry a calibration signal. + public func plan( + watchRoot: DirtyRegionPath, + invalidations: [FileSystemInvalidation], + after storedCheckpoint: EventJournalCursor? + ) throws -> DirtyRegionPlan { + var journaled: [DirtyRegion] = [] + var outOfBand: [DirtyRegion] = [] + + for invalidation in invalidations { + let isCalibrationSignal = invalidation.reasons.contains(.requiresCalibration) + let requiresScopeCalibration = invalidation.reasons.isDisjoint( + with: [.droppedEvents, .rootChanged, .mountChanged] + ) == false + let target = targetDirectory( + for: invalidation, + watchRoot: watchRoot, + forceRoot: requiresScopeCalibration + ) + var reasons = invalidation.reasons + if target.fellBackToRoot { + reasons.insert(.requiresCalibration) + } + + if let cursor = invalidation.cursor, + storedCheckpoint.map({ cursor > $0 }) ?? true { + journaled.append( + try DirtyRegion(path: target.path, reasons: reasons, maximumCursor: cursor) + ) + } else if isCalibrationSignal || invalidation.cursor == nil { + outOfBand.append( + try DirtyRegion(path: target.path, reasons: reasons, maximumCursor: nil) + ) + } + } + + let coalescedJournaled = try coalesce(journaled) + let coalescedOutOfBand = try coalesce(outOfBand) + return DirtyRegionPlan( + checkpoint: coalescedJournaled.compactMap(\.maximumCursor).max(), + journaledRegions: coalescedJournaled, + outOfBandRegions: coalescedOutOfBand + ) + } + + private func targetDirectory( + for invalidation: FileSystemInvalidation, + watchRoot: DirtyRegionPath, + forceRoot: Bool + ) -> (path: DirtyRegionPath, fellBackToRoot: Bool) { + guard forceRoot == false, + let rawPath = invalidation.path, + let normalizedPath = normalize(rawPath), + contains(normalizedPath, in: watchRoot.rawValue) else { + return (watchRoot, true) + } + + let directory: String + switch invalidation.itemKind { + case .directory: + directory = normalizedPath + case .file, .symbolicLink: + directory = deletingLastComponent(of: normalizedPath) + case .unknown: + return (watchRoot, true) + } + + guard contains(directory, in: watchRoot.rawValue), + let path = try? DirtyRegionPath(directory) else { + return (watchRoot, true) + } + return (path, false) + } + + private func normalize(_ rawPath: String) -> String? { + guard rawPath.first == "/", rawPath.utf8.contains(0) == false else { + return nil + } + let normalized = URL(fileURLWithPath: rawPath).standardizedFileURL.path + return normalized.isEmpty ? nil : normalized + } + + private func deletingLastComponent(of path: String) -> String { + let result = (path as NSString).deletingLastPathComponent + return result.isEmpty ? "/" : result + } + + private func contains(_ candidate: String, in root: String) -> Bool { + root == "/" || candidate == root || candidate.hasPrefix(root + "/") + } + + private func coalesce(_ regions: [DirtyRegion]) throws -> [DirtyRegion] { + var result: [DirtyRegion] = [] + for region in regions.sorted(by: regionOrder) { + if let ancestorIndex = result.firstIndex(where: { + contains(region.path.rawValue, in: $0.path.rawValue) + }) { + let ancestor = result[ancestorIndex] + result[ancestorIndex] = try merged(ancestor, region, at: ancestor.path) + continue + } + + let descendants = result.filter { + contains($0.path.rawValue, in: region.path.rawValue) + } + var mergedRegion = region + for descendant in descendants { + mergedRegion = try merged(mergedRegion, descendant, at: region.path) + } + result.removeAll { descendants.contains($0) } + result.append(mergedRegion) + } + return result.sorted { $0.path.rawValue < $1.path.rawValue } + } + + private func regionOrder(_ lhs: DirtyRegion, _ rhs: DirtyRegion) -> Bool { + let lhsDepth = lhs.path.rawValue.split(separator: "/").count + let rhsDepth = rhs.path.rawValue.split(separator: "/").count + return lhsDepth == rhsDepth + ? lhs.path.rawValue < rhs.path.rawValue + : lhsDepth < rhsDepth + } + + private func merged( + _ lhs: DirtyRegion, + _ rhs: DirtyRegion, + at path: DirtyRegionPath + ) throws -> DirtyRegion { + try DirtyRegion( + path: path, + reasons: lhs.reasons.union(rhs.reasons), + maximumCursor: [lhs.maximumCursor, rhs.maximumCursor].compactMap { $0 }.max() + ) + } +} diff --git a/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/FileSystemCalibrationPipeline.swift b/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/FileSystemCalibrationPipeline.swift new file mode 100644 index 0000000..06a148f --- /dev/null +++ b/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/FileSystemCalibrationPipeline.swift @@ -0,0 +1,90 @@ +public struct CalibrationRequest: Sendable, Equatable { + public let streamID: EventStreamID + public let workItem: DirtyRegionWorkItem + + public init(streamID: EventStreamID, workItem: DirtyRegionWorkItem) { + self.streamID = streamID + self.workItem = workItem + } +} + +public enum CalibrationCoverage: Sendable, Equatable { + /// Every descendant in the requested region was enumerated successfully. + case complete + /// Cancellation, permission denial, or an enumeration failure left gaps. + case partial +} + +public protocol CalibrationScanner: Sendable { + func scan(_ request: CalibrationRequest) async throws -> CalibrationCoverage +} + +public actor FileSystemCalibrationPipeline { + private let streamID: EventStreamID + private let watchRoot: DirtyRegionPath + private let repository: any EventJournalRepository + private let scanner: any CalibrationScanner + private let planner: DirtyRegionPlanner + + public init( + streamID: EventStreamID, + watchRoot: DirtyRegionPath, + repository: any EventJournalRepository, + scanner: any CalibrationScanner, + planner: DirtyRegionPlanner = DirtyRegionPlanner() + ) { + self.streamID = streamID + self.watchRoot = watchRoot + self.repository = repository + self.scanner = scanner + self.planner = planner + } + + public func ingest(_ invalidations: [FileSystemInvalidation]) async throws { + guard invalidations.isEmpty == false else { return } + + let checkpoint = try await repository.checkpoint(for: streamID) + let plan = try planner.plan( + watchRoot: watchRoot, + invalidations: invalidations, + after: checkpoint + ) + + if plan.outOfBandRegions.isEmpty == false { + try await repository.markDirty( + streamID: streamID, + regions: plan.outOfBandRegions + ) + } + if let checkpoint = plan.checkpoint, plan.journaledRegions.isEmpty == false { + try await repository.commit( + EventJournalBatch( + streamID: streamID, + checkpoint: checkpoint, + dirtyRegions: plan.journaledRegions + ) + ) + } + } + + /// Processes a bounded snapshot. New events can safely arrive during a + /// scan because `resolve` compares the revision captured before scanning. + @discardableResult + public func calibratePending(limit: Int) async throws -> Int { + guard limit > 0 else { return 0 } + let workItems = try await repository.pendingDirtyWork(for: streamID, limit: limit) + var resolvedCount = 0 + + for workItem in workItems { + try Task.checkCancellation() + let coverage = try await scanner.scan( + CalibrationRequest(streamID: streamID, workItem: workItem) + ) + guard coverage == .complete else { continue } + if try await repository.resolve(workItem, for: streamID) { + resolvedCount += 1 + } + } + return resolvedCount + } +} diff --git a/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/FileSystemInvalidation.swift b/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/FileSystemInvalidation.swift new file mode 100644 index 0000000..0ceb622 --- /dev/null +++ b/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Calibration/FileSystemInvalidation.swift @@ -0,0 +1,50 @@ +/// Application-level item classification. It intentionally contains no +/// FSEvents or Foundation types so other observation adapters can reuse it. +public enum FileSystemItemKind: Sendable, Equatable, Hashable, Codable { + case file + case directory + case symbolicLink + case unknown +} + +/// One lossy filesystem invalidation hint after adapter-specific flags have +/// been translated into application semantics. +public struct FileSystemInvalidation: Sendable, Equatable, Hashable, Codable { + public let path: String? + public let cursor: EventJournalCursor? + public let reasons: DirtyRegionReason + public let itemKind: FileSystemItemKind + + public init( + path: String?, + cursor: EventJournalCursor?, + reasons: DirtyRegionReason, + itemKind: FileSystemItemKind = .unknown + ) throws(EventJournalModelError) { + guard reasons.isEmpty == false else { + throw .emptyDirtyRegionReasons + } + + self.path = path + self.cursor = cursor + self.reasons = reasons + self.itemKind = itemKind + } +} + +/// Durable work split by whether it can safely advance the journal cursor. +public struct DirtyRegionPlan: Sendable, Equatable { + public let checkpoint: EventJournalCursor? + public let journaledRegions: [DirtyRegion] + public let outOfBandRegions: [DirtyRegion] + + public init( + checkpoint: EventJournalCursor?, + journaledRegions: [DirtyRegion], + outOfBandRegions: [DirtyRegion] + ) { + self.checkpoint = checkpoint + self.journaledRegions = journaledRegions + self.outOfBandRegions = outOfBandRegions + } +} diff --git a/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Persistence/EventJournalRepository.swift b/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Persistence/EventJournalRepository.swift index f560e9c..a9ad522 100644 --- a/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Persistence/EventJournalRepository.swift +++ b/Packages/SpaceTraceKit/Sources/SpaceTraceApplication/Persistence/EventJournalRepository.swift @@ -123,12 +123,15 @@ public struct DirtyRegionReason: OptionSet, Sendable, Equatable, Hashable, Codab public struct DirtyRegion: Sendable, Equatable, Hashable, Codable { public let path: DirtyRegionPath public let reasons: DirtyRegionReason - public let maximumCursor: EventJournalCursor + /// The newest journal cursor represented by this work, when one exists. + /// Out-of-band continuity failures such as `RootChanged` may create + /// durable calibration work without advancing or inventing a cursor. + public let maximumCursor: EventJournalCursor? public init( path: DirtyRegionPath, reasons: DirtyRegionReason, - maximumCursor: EventJournalCursor + maximumCursor: EventJournalCursor? ) throws(EventJournalModelError) { guard reasons.isEmpty == false else { throw .emptyDirtyRegionReasons @@ -143,7 +146,10 @@ public struct DirtyRegion: Sendable, Equatable, Hashable, Codable { let container = try decoder.container(keyedBy: CodingKeys.self) let path = try container.decode(DirtyRegionPath.self, forKey: .path) let reasons = try container.decode(DirtyRegionReason.self, forKey: .reasons) - let maximumCursor = try container.decode(EventJournalCursor.self, forKey: .maximumCursor) + let maximumCursor = try container.decodeIfPresent( + EventJournalCursor.self, + forKey: .maximumCursor + ) do { try self.init(path: path, reasons: reasons, maximumCursor: maximumCursor) @@ -179,8 +185,13 @@ public struct EventJournalBatch: Sendable, Equatable, Codable { guard dirtyRegions.isEmpty == false else { throw .emptyBatch } - guard dirtyRegions.allSatisfy({ $0.maximumCursor <= checkpoint }) else { - throw .dirtyRegionCursorExceedsCheckpoint + for region in dirtyRegions { + guard let maximumCursor = region.maximumCursor else { + throw .eventBatchRequiresCursors + } + guard maximumCursor <= checkpoint else { + throw .dirtyRegionCursorExceedsCheckpoint + } } self.streamID = streamID @@ -221,7 +232,54 @@ public enum EventJournalModelError: Error, Sendable, Equatable { case invalidDirtyRegionPath(String) case emptyDirtyRegionReasons case emptyBatch + case eventBatchRequiresCursors case dirtyRegionCursorExceedsCheckpoint + case invalidDirtyRegionRevision(UInt64) +} + +/// A monotonic row token used to prevent an older calibration result from +/// clearing dirty work that changed while the scan was in flight. +public struct DirtyRegionRevision: Sendable, Equatable, Hashable, Comparable, Codable { + public let rawValue: UInt64 + + public init(_ rawValue: UInt64) throws(EventJournalModelError) { + guard rawValue > 0 else { + throw .invalidDirtyRegionRevision(rawValue) + } + self.rawValue = rawValue + } + + public init(from decoder: any Decoder) throws { + let container = try decoder.singleValueContainer() + let rawValue = try container.decode(UInt64.self) + do { + try self.init(rawValue) + } catch { + throw DecodingError.dataCorruptedError( + in: container, + debugDescription: "DirtyRegionRevision must be greater than zero." + ) + } + } + + public func encode(to encoder: any Encoder) throws { + var container = encoder.singleValueContainer() + try container.encode(rawValue) + } + + public static func < (lhs: DirtyRegionRevision, rhs: DirtyRegionRevision) -> Bool { + lhs.rawValue < rhs.rawValue + } +} + +public struct DirtyRegionWorkItem: Sendable, Equatable, Hashable, Codable { + public let region: DirtyRegion + public let revision: DirtyRegionRevision + + public init(region: DirtyRegion, revision: DirtyRegionRevision) { + self.region = region + self.revision = revision + } } /// Application-owned persistence port. Implementations may use SQLite, an @@ -229,6 +287,15 @@ public enum EventJournalModelError: Error, Sendable, Equatable { /// into filesystem or application logic. public protocol EventJournalRepository: Sendable { func commit(_ batch: EventJournalBatch) async throws + func markDirty(streamID: EventStreamID, regions: [DirtyRegion]) async throws func checkpoint(for streamID: EventStreamID) async throws -> EventJournalCursor? func dirtyRegions(for streamID: EventStreamID) async throws -> [DirtyRegion] + func pendingDirtyWork( + for streamID: EventStreamID, + limit: Int + ) async throws -> [DirtyRegionWorkItem] + func resolve( + _ workItem: DirtyRegionWorkItem, + for streamID: EventStreamID + ) async throws -> Bool } diff --git a/Packages/SpaceTraceKit/Sources/SpaceTraceFileSystem/FSEventInvalidationMapper.swift b/Packages/SpaceTraceKit/Sources/SpaceTraceFileSystem/FSEventInvalidationMapper.swift new file mode 100644 index 0000000..aaa43db --- /dev/null +++ b/Packages/SpaceTraceKit/Sources/SpaceTraceFileSystem/FSEventInvalidationMapper.swift @@ -0,0 +1,78 @@ +import SpaceTraceApplication + +/// The anti-corruption layer between native FSEvents semantics and the +/// application-owned calibration pipeline. +public struct FSEventInvalidationMapper: Sendable { + public init() {} + + public func map(_ observation: FSEventObservation) throws -> FileSystemInvalidation? { + if observation.reasons == [.historicalReplayCompleted] { + return nil + } + + var reasons: DirtyRegionReason = [] + for reason in observation.reasons { + reasons.formUnion(map(reason)) + } + guard reasons.isEmpty == false else { + return nil + } + + let cursor: EventJournalCursor? + if observation.reasons.contains(.watchedRootChanged), + observation.eventID?.rawValue == 0 { + cursor = nil + } else { + cursor = observation.eventID.map { EventJournalCursor($0.rawValue) } + } + + return try FileSystemInvalidation( + path: observation.path, + cursor: cursor, + reasons: reasons, + itemKind: itemKind(from: observation.reasons) + ) + } + + private func itemKind(from reasons: Set) -> FileSystemItemKind { + let kinds: [(FSEventReason, FileSystemItemKind)] = [ + (.itemIsFile, .file), + (.itemIsDirectory, .directory), + (.itemIsSymbolicLink, .symbolicLink), + ] + let matches = kinds.compactMap { reasons.contains($0.0) ? $0.1 : nil } + return matches.count == 1 ? matches[0] : .unknown + } + + private func map(_ reason: FSEventReason) -> DirtyRegionReason { + switch reason { + case .pathChanged: + return .metadataChanged + case .descendantsMustBeScanned: + return [.mustScanSubdirectories, .requiresCalibration] + case .eventsDroppedByUserSpace, .eventsDroppedByKernel, + .eventIdentifiersWrapped, .callbackBridgeOverflow, + .unrecognizedFlags: + return [.droppedEvents, .requiresCalibration] + case .watchedRootChanged: + return [.rootChanged, .requiresCalibration] + case .volumeMounted, .volumeUnmounted: + return [.mountChanged, .requiresCalibration] + case .itemCreated: + return .created + case .itemRemoved: + return .removed + case .itemRenamed: + return .renamed + case .itemContentModified, .itemCloned: + return .contentModified + case .itemMetadataModified, .itemFinderInfoModified, + .itemOwnershipChanged, .itemExtendedAttributesModified, + .itemIsHardLink, .itemWasLastHardLink: + return .metadataChanged + case .historicalReplayCompleted, .itemIsFile, .itemIsDirectory, + .itemIsSymbolicLink, .eventOriginatedFromThisProcess: + return [] + } + } +} diff --git a/Packages/SpaceTraceKit/Sources/SpaceTracePersistence/SQLiteEventJournalRepository.swift b/Packages/SpaceTraceKit/Sources/SpaceTracePersistence/SQLiteEventJournalRepository.swift index c6e3fb9..66fd5fb 100644 --- a/Packages/SpaceTraceKit/Sources/SpaceTracePersistence/SQLiteEventJournalRepository.swift +++ b/Packages/SpaceTraceKit/Sources/SpaceTracePersistence/SQLiteEventJournalRepository.swift @@ -6,7 +6,7 @@ import Synchronization /// A dependency-free SQLite prototype for ADR-004. The actor is the sole /// owner of the connection and serializes every transaction and query. public actor SQLiteEventJournalRepository: EventJournalRepository { - private static let schemaVersion: Int32 = 1 + private static let schemaVersion: Int32 = 2 /// `Mutex` makes the non-Sendable C handle safe to release from the /// actor's nonisolated deinitializer. All operational access remains @@ -72,9 +72,7 @@ public actor SQLiteEventJournalRepository: EventJournalRepository { ) } - for region in batch.dirtyRegions { - try upsert(region: region, streamID: batch.streamID) - } + try merge(regions: batch.dirtyRegions, streamID: batch.streamID) try failIfRequested(at: .afterDirtyRegionsBeforeCheckpoint) try upsert(checkpoint: batch.checkpoint, streamID: batch.streamID) @@ -84,6 +82,19 @@ public actor SQLiteEventJournalRepository: EventJournalRepository { } } + /// Persists calibration work without moving the journal checkpoint. This + /// is required for stream-wide sentinels that do not carry a usable ID. + public func markDirty(streamID: EventStreamID, regions: [DirtyRegion]) throws { + guard regions.isEmpty == false else { return } + try execute("BEGIN IMMEDIATE TRANSACTION", operation: "begin dirty-region transaction") + do { + try merge(regions: regions, streamID: streamID) + try execute("COMMIT TRANSACTION", operation: "commit dirty-region transaction") + } catch { + try rollback(after: error) + } + } + public func checkpoint(for streamID: EventStreamID) throws -> EventJournalCursor? { try readCheckpoint(for: streamID) } @@ -116,7 +127,7 @@ public actor SQLiteEventJournalRepository: EventJournalRepository { let reasonsRawValue = UInt64( bitPattern: sqlite3_column_int64(statement, 1) ) - let cursor = try readCursor( + let cursor = try readOptionalCursor( from: statement, column: 2, field: "dirty_region.maximum_cursor_be" @@ -147,6 +158,120 @@ public actor SQLiteEventJournalRepository: EventJournalRepository { } } + public func pendingDirtyWork( + for streamID: EventStreamID, + limit: Int + ) throws -> [DirtyRegionWorkItem] { + guard limit > 0 else { return [] } + let sql = """ + SELECT path, reasons, maximum_cursor_be, revision_be + FROM dirty_region + WHERE stream_id = ?1 + ORDER BY path ASC + LIMIT ?2 + """ + return try withStatement(sql, operation: "read pending dirty work") { statement in + try streamID.rawValue.withCString { streamCString in + try check( + sqlite3_bind_text(statement, 1, streamCString, -1, nil), + operation: "bind pending-work stream ID" + ) + try check( + sqlite3_bind_int64(statement, 2, Int64(limit)), + operation: "bind pending-work limit" + ) + var work: [DirtyRegionWorkItem] = [] + while true { + let result = sqlite3_step(statement) + switch result { + case SQLITE_ROW: + let path = try readText( + from: statement, + column: 0, + field: "dirty_region.path" + ) + let reasons = DirtyRegionReason( + rawValue: UInt64(bitPattern: sqlite3_column_int64(statement, 1)) + ) + let cursor = try readOptionalCursor( + from: statement, + column: 2, + field: "dirty_region.maximum_cursor_be" + ) + let revisionValue = try readUInt64( + from: statement, + column: 3, + field: "dirty_region.revision_be" + ) + do { + let region = try DirtyRegion( + path: DirtyRegionPath(path), + reasons: reasons, + maximumCursor: cursor + ) + work.append( + DirtyRegionWorkItem( + region: region, + revision: try DirtyRegionRevision(revisionValue) + ) + ) + } catch { + throw SQLiteEventJournalError.corruptStoredValue( + field: "dirty_region" + ) + } + case SQLITE_DONE: + return work + default: + throw sqliteFailure( + operation: "step pending-work query", + code: result + ) + } + } + } + } + } + + public func resolve( + _ workItem: DirtyRegionWorkItem, + for streamID: EventStreamID + ) throws -> Bool { + let sql = """ + DELETE FROM dirty_region + WHERE stream_id = ?1 AND path = ?2 AND revision_be = ?3 + """ + let revisionBytes = Self.encode(workItem.revision.rawValue) + return try withStatement(sql, operation: "resolve dirty work") { statement in + try streamID.rawValue.withCString { streamCString in + try workItem.region.path.rawValue.withCString { pathCString in + try revisionBytes.withUnsafeBytes { revisionBuffer in + try check( + sqlite3_bind_text(statement, 1, streamCString, -1, nil), + operation: "bind resolve stream ID" + ) + try check( + sqlite3_bind_text(statement, 2, pathCString, -1, nil), + operation: "bind resolve path" + ) + try check( + sqlite3_bind_blob( + statement, + 3, + revisionBuffer.baseAddress, + Int32(revisionBuffer.count), + nil + ), + operation: "bind resolve revision" + ) + try stepExpectingDone(statement, operation: "delete resolved work") + return sqlite3_changes(try databaseHandle()) == 1 + } + } + } + } + } + /// Explicitly releases SQLite resources. Calling close repeatedly is safe; /// repository operations after closing report `databaseClosed`. public func close() throws { @@ -193,11 +318,67 @@ public actor SQLiteEventJournalRepository: EventJournalRepository { return case 0: try migrateToVersionOne(database) + try migrateToVersionTwo(database) + case 1: + try migrateToVersionTwo(database) default: throw SQLiteEventJournalError.unsupportedSchemaVersion(currentVersion) } } + private static func migrateToVersionTwo(_ database: OpaquePointer) throws { + try execute(on: database, "BEGIN IMMEDIATE TRANSACTION", operation: "begin schema migration v2") + do { + try execute( + on: database, + """ + ALTER TABLE dirty_region RENAME TO dirty_region_v1; + + CREATE TABLE dirty_region ( + stream_id TEXT NOT NULL + CHECK(length(stream_id) > 0 AND instr(stream_id, char(0)) = 0), + path TEXT NOT NULL + CHECK(length(path) > 0 AND substr(path, 1, 1) = '/' + AND instr(path, char(0)) = 0), + reasons INTEGER NOT NULL CHECK(reasons != 0), + maximum_cursor_be BLOB CHECK( + maximum_cursor_be IS NULL OR length(maximum_cursor_be) = 8 + ), + revision_be BLOB NOT NULL CHECK(length(revision_be) = 8), + PRIMARY KEY(stream_id, path) + ) WITHOUT ROWID; + + INSERT INTO dirty_region( + stream_id, path, reasons, maximum_cursor_be, revision_be + ) + SELECT stream_id, path, reasons, maximum_cursor_be, + X'0000000000000001' + FROM dirty_region_v1; + + DROP TABLE dirty_region_v1; + + INSERT INTO schema_migration(version, applied_at_ms, checksum) + VALUES(2, CAST(strftime('%s', 'now') AS INTEGER) * 1000, + 'dirty-region-v2-optional-cursor-revision'); + + PRAGMA user_version = 2; + """, + operation: "apply schema migration version 2" + ) + try execute(on: database, "COMMIT TRANSACTION", operation: "commit schema migration v2") + } catch let migrationError { + do { + try execute(on: database, "ROLLBACK TRANSACTION", operation: "roll back schema migration v2") + } catch let rollbackError { + throw SQLiteEventJournalError.rollbackFailed( + original: String(describing: migrationError), + rollback: String(describing: rollbackError) + ) + } + throw migrationError + } + } + private static func migrateToVersionOne(_ database: OpaquePointer) throws { try execute( on: database, @@ -322,24 +503,64 @@ public actor SQLiteEventJournalRepository: EventJournalRepository { } } - private func upsert(region: DirtyRegion, streamID: EventStreamID) throws { + private func merge(regions: [DirtyRegion], streamID: EventStreamID) throws { + var current = try pendingDirtyWork(for: streamID, limit: Int.max) + for incoming in regions { + let overlapping = current.filter { + Self.contains($0.region.path.rawValue, in: incoming.path.rawValue) + || Self.contains(incoming.path.rawValue, in: $0.region.path.rawValue) + } + let canonicalPath = ([incoming.path] + overlapping.map(\.region.path)).min { + let lhsDepth = $0.rawValue.split(separator: "/").count + let rhsDepth = $1.rawValue.split(separator: "/").count + return lhsDepth == rhsDepth + ? $0.rawValue < $1.rawValue + : lhsDepth < rhsDepth + } ?? incoming.path + + let reasons = overlapping.reduce(incoming.reasons) { + $0.union($1.region.reasons) + } + let maximumCursor = ([incoming.maximumCursor] + overlapping.map(\.region.maximumCursor)) + .compactMap { $0 } + .max() + let maximumRevision = overlapping.map(\.revision.rawValue).max() ?? 0 + guard maximumRevision < UInt64.max else { + throw SQLiteEventJournalError.revisionOverflow(path: canonicalPath.rawValue) + } + let revision = try DirtyRegionRevision(maximumRevision + 1) + for item in overlapping { + try delete(path: item.region.path, streamID: streamID) + } + let mergedRegion = try DirtyRegion( + path: canonicalPath, + reasons: reasons, + maximumCursor: maximumCursor + ) + try insert(region: mergedRegion, revision: revision, streamID: streamID) + current.removeAll { overlapping.contains($0) } + current.append(DirtyRegionWorkItem(region: mergedRegion, revision: revision)) + } + } + + private func insert( + region: DirtyRegion, + revision: DirtyRegionRevision, + streamID: EventStreamID + ) throws { let sql = """ - INSERT INTO dirty_region(stream_id, path, reasons, maximum_cursor_be) - VALUES(?1, ?2, ?3, ?4) - ON CONFLICT(stream_id, path) DO UPDATE SET - reasons = dirty_region.reasons | excluded.reasons, - maximum_cursor_be = CASE - WHEN dirty_region.maximum_cursor_be < excluded.maximum_cursor_be - THEN excluded.maximum_cursor_be - ELSE dirty_region.maximum_cursor_be - END + INSERT INTO dirty_region( + stream_id, path, reasons, maximum_cursor_be, revision_be + ) + VALUES(?1, ?2, ?3, ?4, ?5) """ - let cursorBytes = Self.encode(region.maximumCursor) + let cursorBytes = region.maximumCursor.map(Self.encode) + let revisionBytes = Self.encode(revision.rawValue) - try withStatement(sql, operation: "upsert dirty region") { statement in + try withStatement(sql, operation: "insert dirty region") { statement in try streamID.rawValue.withCString { streamCString in try region.path.rawValue.withCString { pathCString in - try cursorBytes.withUnsafeBytes { cursorBuffer in + try revisionBytes.withUnsafeBytes { revisionBuffer in try check( sqlite3_bind_text(statement, 1, streamCString, -1, nil), operation: "bind dirty-region stream ID" @@ -356,15 +577,34 @@ public actor SQLiteEventJournalRepository: EventJournalRepository { ), operation: "bind dirty-region reasons" ) + if let cursorBytes { + try cursorBytes.withUnsafeBytes { cursorBuffer in + try check( + sqlite3_bind_blob( + statement, + 4, + cursorBuffer.baseAddress, + Int32(cursorBuffer.count), + nil + ), + operation: "bind dirty-region cursor" + ) + } + } else { + try check( + sqlite3_bind_null(statement, 4), + operation: "bind null dirty-region cursor" + ) + } try check( sqlite3_bind_blob( statement, - 4, - cursorBuffer.baseAddress, - Int32(cursorBuffer.count), + 5, + revisionBuffer.baseAddress, + Int32(revisionBuffer.count), nil ), - operation: "bind dirty-region cursor" + operation: "bind dirty-region revision" ) try stepExpectingDone(statement, operation: "write dirty region") } @@ -373,6 +613,23 @@ public actor SQLiteEventJournalRepository: EventJournalRepository { } } + private func delete(path: DirtyRegionPath, streamID: EventStreamID) throws { + let sql = "DELETE FROM dirty_region WHERE stream_id = ?1 AND path = ?2" + try withStatement(sql, operation: "delete overlapping dirty region") { statement in + try streamID.rawValue.withCString { streamCString in + try path.rawValue.withCString { pathCString in + try check(sqlite3_bind_text(statement, 1, streamCString, -1, nil), operation: "bind delete stream ID") + try check(sqlite3_bind_text(statement, 2, pathCString, -1, nil), operation: "bind delete path") + try stepExpectingDone(statement, operation: "delete overlapping dirty region") + } + } + } + } + + private static func contains(_ candidate: String, in root: String) -> Bool { + root == "/" || candidate == root || candidate.hasPrefix(root + "/") + } + private func upsert(checkpoint: EventJournalCursor, streamID: EventStreamID) throws { let sql = """ INSERT INTO event_checkpoint(stream_id, cursor_be) @@ -545,10 +802,33 @@ public actor SQLiteEventJournalRepository: EventJournalRepository { return EventJournalCursor(rawValue) } + private func readOptionalCursor( + from statement: OpaquePointer, + column: Int32, + field: String + ) throws -> EventJournalCursor? { + if sqlite3_column_type(statement, column) == SQLITE_NULL { + return nil + } + return try readCursor(from: statement, column: column, field: field) + } + + private func readUInt64( + from statement: OpaquePointer, + column: Int32, + field: String + ) throws -> UInt64 { + try readCursor(from: statement, column: column, field: field).rawValue + } + private static func encode(_ cursor: EventJournalCursor) -> [UInt8] { + encode(cursor.rawValue) + } + + private static func encode(_ rawValue: UInt64) -> [UInt8] { (0..<8).map { offset in let shift = UInt64((7 - offset) * 8) - return UInt8(truncatingIfNeeded: cursor.rawValue >> shift) + return UInt8(truncatingIfNeeded: rawValue >> shift) } } @@ -570,6 +850,7 @@ public enum SQLiteEventJournalError: Error, Sendable, Equatable { case databaseClosed case unsupportedSchemaVersion(Int32) case cursorRegression(stored: UInt64, attempted: UInt64) + case revisionOverflow(path: String) case invalidCursorEncoding(field: String, actualByteCount: Int) case corruptStoredValue(field: String) case sqliteFailure(operation: String, code: Int32, message: String) diff --git a/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/DirtyRegionPlannerTests.swift b/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/DirtyRegionPlannerTests.swift new file mode 100644 index 0000000..1b3ae9a --- /dev/null +++ b/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/DirtyRegionPlannerTests.swift @@ -0,0 +1,105 @@ +import Testing +@testable import SpaceTraceApplication + +struct DirtyRegionPlannerTests { + private let planner = DirtyRegionPlanner() + + @Test("File events collapse to their parent and union reasons") + func coalescesFileEvents() throws { + let plan = try planner.plan( + watchRoot: DirtyRegionPath("/Users/example"), + invalidations: [ + invalidation("/Users/example/Documents/a.txt", 11, [.created], .file), + invalidation("/Users/example/Documents/b.txt", 12, [.contentModified], .file), + ], + after: EventJournalCursor(10) + ) + + let region = try #require(plan.journaledRegions.first) + #expect(plan.checkpoint == EventJournalCursor(12)) + #expect(plan.journaledRegions.count == 1) + #expect(region.path.rawValue == "/Users/example/Documents") + #expect(region.reasons == [.created, .contentModified]) + #expect(region.maximumCursor == EventJournalCursor(12)) + } + + @Test("Ordinary overlap replay at or before the checkpoint is ignored") + func ignoresReplayOverlap() throws { + let plan = try planner.plan( + watchRoot: DirtyRegionPath("/Users/example"), + invalidations: [ + invalidation("/Users/example/a.txt", 10, [.contentModified], .file), + ], + after: EventJournalCursor(10) + ) + #expect(plan.journaledRegions.isEmpty) + #expect(plan.outOfBandRegions.isEmpty) + #expect(plan.checkpoint == nil) + } + + @Test("A replayed continuity gap still schedules root calibration without moving the cursor") + func preservesReplayedGap() throws { + let plan = try planner.plan( + watchRoot: DirtyRegionPath("/Users/example"), + invalidations: [ + invalidation( + "/Users/example/Documents", + 9, + [.droppedEvents, .requiresCalibration], + .directory + ), + ], + after: EventJournalCursor(10) + ) + let region = try #require(plan.outOfBandRegions.first) + #expect(plan.checkpoint == nil) + #expect(region.path.rawValue == "/Users/example") + #expect(region.maximumCursor == nil) + } + + @Test("MustScanSubdirectories retains a known safe directory boundary") + func retainsNarrowRecursiveRegion() throws { + let plan = try planner.plan( + watchRoot: DirtyRegionPath("/Users/example"), + invalidations: [ + invalidation( + "/Users/example/Documents/Project", + 22, + [.mustScanSubdirectories, .requiresCalibration], + .directory + ), + ], + after: nil + ) + let region = try #require(plan.journaledRegions.first) + #expect(region.path.rawValue == "/Users/example/Documents/Project") + } + + @Test("Unknown and outside paths fall back to the watched root") + func safelyFallsBackToRoot() throws { + let plan = try planner.plan( + watchRoot: DirtyRegionPath("/Users/example"), + invalidations: [ + invalidation("/tmp/other", 21, [.created], .unknown), + ], + after: nil + ) + let region = try #require(plan.journaledRegions.first) + #expect(region.path.rawValue == "/Users/example") + #expect(region.reasons.contains(.requiresCalibration)) + } + + private func invalidation( + _ path: String?, + _ cursor: UInt64?, + _ reasons: DirtyRegionReason, + _ kind: FileSystemItemKind + ) throws -> FileSystemInvalidation { + try FileSystemInvalidation( + path: path, + cursor: cursor.map(EventJournalCursor.init), + reasons: reasons, + itemKind: kind + ) + } +} diff --git a/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/EventJournalModelTests.swift b/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/EventJournalModelTests.swift index daf8082..d095fad 100644 --- a/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/EventJournalModelTests.swift +++ b/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/EventJournalModelTests.swift @@ -55,4 +55,29 @@ struct EventJournalModelTests { ) } } + + @Test("Cursor-free work cannot be used to advance a checkpoint") + func rejectsCursorFreeBatch() throws { + let streamID = try EventStreamID("volume-a:generation-1") + let region = try DirtyRegion( + path: DirtyRegionPath("/Users/example"), + reasons: [.rootChanged, .requiresCalibration], + maximumCursor: nil + ) + + #expect(throws: EventJournalModelError.eventBatchRequiresCursors) { + try EventJournalBatch( + streamID: streamID, + checkpoint: EventJournalCursor(42), + dirtyRegions: [region] + ) + } + } + + @Test("Dirty row revisions start above zero") + func rejectsZeroRevision() { + #expect(throws: EventJournalModelError.invalidDirtyRegionRevision(0)) { + try DirtyRegionRevision(0) + } + } } diff --git a/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/FileSystemCalibrationPipelineTests.swift b/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/FileSystemCalibrationPipelineTests.swift new file mode 100644 index 0000000..c3686a3 --- /dev/null +++ b/Packages/SpaceTraceKit/Tests/SpaceTraceApplicationTests/FileSystemCalibrationPipelineTests.swift @@ -0,0 +1,156 @@ +import Testing +@testable import SpaceTraceApplication + +struct FileSystemCalibrationPipelineTests { + @Test("Complete calibration conditionally resolves durable work") + func resolvesCompleteCalibration() async throws { + let repository = InMemoryEventJournalRepository() + let scanner = FakeCalibrationScanner(coverage: .complete) + let pipeline = try makePipeline(repository: repository, scanner: scanner) + + try await pipeline.ingest([try fileInvalidation(cursor: 1)]) + #expect(try await pipeline.calibratePending(limit: 10) == 1) + #expect(try await repository.dirtyRegions(for: streamID()).isEmpty) + #expect(await scanner.requestCount == 1) + } + + @Test("Partial calibration never clears work") + func preservesPartialCalibration() async throws { + let repository = InMemoryEventJournalRepository() + let pipeline = try makePipeline( + repository: repository, + scanner: FakeCalibrationScanner(coverage: .partial) + ) + + try await pipeline.ingest([try fileInvalidation(cursor: 1)]) + #expect(try await pipeline.calibratePending(limit: 10) == 0) + #expect(try await repository.dirtyRegions(for: streamID()).count == 1) + } + + @Test("A newer event arriving during a scan defeats the stale resolution token") + func newerEventWinsRace() async throws { + let repository = InMemoryEventJournalRepository() + let streamID = try streamID() + let scanner = FakeCalibrationScanner(coverage: .complete) { request in + let newer = try DirtyRegion( + path: request.workItem.region.path, + reasons: [.removed], + maximumCursor: nil + ) + try await repository.markDirty(streamID: streamID, regions: [newer]) + } + let pipeline = try makePipeline(repository: repository, scanner: scanner) + + try await pipeline.ingest([try fileInvalidation(cursor: 1)]) + #expect(try await pipeline.calibratePending(limit: 1) == 0) + let regions = await repository.dirtyRegions(for: streamID) + let remaining = try #require(regions.first) + #expect(remaining.reasons == [.contentModified, .removed]) + } + + private func makePipeline( + repository: InMemoryEventJournalRepository, + scanner: FakeCalibrationScanner + ) throws -> FileSystemCalibrationPipeline { + try FileSystemCalibrationPipeline( + streamID: streamID(), + watchRoot: DirtyRegionPath("/Users/example"), + repository: repository, + scanner: scanner + ) + } + + private func fileInvalidation(cursor: UInt64) throws -> FileSystemInvalidation { + try FileSystemInvalidation( + path: "/Users/example/Documents/file.txt", + cursor: EventJournalCursor(cursor), + reasons: [.contentModified], + itemKind: .file + ) + } + + private func streamID() throws -> EventStreamID { + try EventStreamID("volume-a:generation-1") + } +} + +private actor FakeCalibrationScanner: CalibrationScanner { + private let coverage: CalibrationCoverage + private let beforeReturn: @Sendable (CalibrationRequest) async throws -> Void + private(set) var requestCount = 0 + + init( + coverage: CalibrationCoverage, + beforeReturn: @escaping @Sendable (CalibrationRequest) async throws -> Void = { _ in } + ) { + self.coverage = coverage + self.beforeReturn = beforeReturn + } + + func scan(_ request: CalibrationRequest) async throws -> CalibrationCoverage { + requestCount += 1 + try await beforeReturn(request) + return coverage + } +} + +private actor InMemoryEventJournalRepository: EventJournalRepository { + private var savedCheckpoint: EventJournalCursor? + private var work: [DirtyRegionPath: DirtyRegionWorkItem] = [:] + + func commit(_ batch: EventJournalBatch) throws { + if let savedCheckpoint, batch.checkpoint < savedCheckpoint { + return + } + try merge(batch.dirtyRegions) + savedCheckpoint = batch.checkpoint + } + + func markDirty(streamID: EventStreamID, regions: [DirtyRegion]) throws { + try merge(regions) + } + + func checkpoint(for streamID: EventStreamID) -> EventJournalCursor? { + savedCheckpoint + } + + func dirtyRegions(for streamID: EventStreamID) -> [DirtyRegion] { + work.values.map(\.region).sorted { $0.path.rawValue < $1.path.rawValue } + } + + func pendingDirtyWork( + for streamID: EventStreamID, + limit: Int + ) -> [DirtyRegionWorkItem] { + Array( + work.values.sorted { $0.region.path.rawValue < $1.region.path.rawValue } + .prefix(max(0, limit)) + ) + } + + func resolve(_ workItem: DirtyRegionWorkItem, for streamID: EventStreamID) -> Bool { + guard work[workItem.region.path]?.revision == workItem.revision else { + return false + } + work[workItem.region.path] = nil + return true + } + + private func merge(_ regions: [DirtyRegion]) throws { + for region in regions { + let existing = work[region.path] + let nextRevision = (existing?.revision.rawValue ?? 0) + 1 + let merged = try DirtyRegion( + path: region.path, + reasons: region.reasons.union(existing?.region.reasons ?? []), + maximumCursor: [region.maximumCursor, existing?.region.maximumCursor] + .compactMap { $0 } + .max() + ) + work[region.path] = DirtyRegionWorkItem( + region: merged, + revision: try DirtyRegionRevision(nextRevision) + ) + } + } +} diff --git a/Packages/SpaceTraceKit/Tests/SpaceTraceFileSystemTests/FSEventInvalidationMapperTests.swift b/Packages/SpaceTraceKit/Tests/SpaceTraceFileSystemTests/FSEventInvalidationMapperTests.swift new file mode 100644 index 0000000..e39cfca --- /dev/null +++ b/Packages/SpaceTraceKit/Tests/SpaceTraceFileSystemTests/FSEventInvalidationMapperTests.swift @@ -0,0 +1,51 @@ +import Testing +import SpaceTraceApplication +@testable import SpaceTraceFileSystem + +struct FSEventInvalidationMapperTests { + private let mapper = FSEventInvalidationMapper() + + @Test("A file modification retains its cursor and item classification") + func mapsFileModification() throws { + let mapped = try mapper.map( + FSEventObservation( + path: "/Users/example/file.txt", + eventID: FSEventID(rawValue: 42), + reasons: [.itemContentModified, .itemIsFile], + rawFlags: 0 + ) + ) + let result = try #require(mapped) + #expect(result.cursor == EventJournalCursor(42)) + #expect(result.reasons == [.contentModified]) + #expect(result.itemKind == .file) + } + + @Test("RootChanged zero is a sentinel and schedules cursor-free calibration") + func suppressesRootChangedSentinel() throws { + let mapped = try mapper.map( + FSEventObservation( + path: "/Users/example", + eventID: FSEventID(rawValue: 0), + reasons: [.watchedRootChanged], + rawFlags: 0 + ) + ) + let result = try #require(mapped) + #expect(result.cursor == nil) + #expect(result.reasons == [.rootChanged, .requiresCalibration]) + } + + @Test("History completion alone does not manufacture dirty work") + func ignoresHistoryDone() throws { + let result = try mapper.map( + FSEventObservation( + path: nil, + eventID: FSEventID(rawValue: 100), + reasons: [.historicalReplayCompleted], + rawFlags: 0 + ) + ) + #expect(result == nil) + } +} diff --git a/Packages/SpaceTraceKit/Tests/SpaceTracePersistenceTests/CalibrationPipelineIntegrationTests.swift b/Packages/SpaceTraceKit/Tests/SpaceTracePersistenceTests/CalibrationPipelineIntegrationTests.swift new file mode 100644 index 0000000..ff7e48f --- /dev/null +++ b/Packages/SpaceTraceKit/Tests/SpaceTracePersistenceTests/CalibrationPipelineIntegrationTests.swift @@ -0,0 +1,73 @@ +import Foundation +import Testing +import SpaceTraceApplication +import SpaceTraceFileSystem +@testable import SpaceTracePersistence + +struct CalibrationPipelineIntegrationTests { + @Test("FSEvents mapping flows through durable dirty work into calibration") + func endToEndPipeline() async throws { + let fixture = try PipelineTemporaryDatabase() + let repository = try SQLiteEventJournalRepository(databaseURL: fixture.databaseURL) + defer { fixture.remove() } + + let streamID = try EventStreamID("volume-integration:generation-1") + let scanner = IntegrationScanner() + let pipeline = try FileSystemCalibrationPipeline( + streamID: streamID, + watchRoot: DirtyRegionPath("/Users/example"), + repository: repository, + scanner: scanner + ) + let observation = FSEventObservation( + path: "/Users/example/Documents/report.pdf", + eventID: FSEventID(rawValue: 88), + reasons: [.itemContentModified, .itemIsFile], + rawFlags: 0 + ) + let mapped = try FSEventInvalidationMapper().map(observation) + let invalidation = try #require(mapped) + + try await pipeline.ingest([invalidation]) + #expect(try await repository.checkpoint(for: streamID) == EventJournalCursor(88)) + let durable = try #require( + try await repository.dirtyRegions(for: streamID).first + ) + #expect(durable.path.rawValue == "/Users/example/Documents") + + #expect(try await pipeline.calibratePending(limit: 1) == 1) + #expect(try await repository.dirtyRegions(for: streamID).isEmpty) + let request = try #require(await scanner.requests.first) + #expect(request.workItem.region.path.rawValue == "/Users/example/Documents") + try await repository.close() + } +} + +private actor IntegrationScanner: CalibrationScanner { + private(set) var requests: [CalibrationRequest] = [] + + func scan(_ request: CalibrationRequest) -> CalibrationCoverage { + requests.append(request) + return .complete + } +} + +private struct PipelineTemporaryDatabase { + let directoryURL: URL + let databaseURL: URL + + init() throws { + directoryURL = FileManager.default.temporaryDirectory + .appendingPathComponent("SpaceTracePipelineTests-\(UUID().uuidString)", isDirectory: true) + databaseURL = directoryURL.appendingPathComponent("SpaceTrace.sqlite") + try FileManager.default.createDirectory( + at: directoryURL, + withIntermediateDirectories: true, + attributes: [.posixPermissions: 0o700] + ) + } + + func remove() { + try? FileManager.default.removeItem(at: directoryURL) + } +} diff --git a/Packages/SpaceTraceKit/Tests/SpaceTracePersistenceTests/SQLiteEventJournalRepositoryTests.swift b/Packages/SpaceTraceKit/Tests/SpaceTracePersistenceTests/SQLiteEventJournalRepositoryTests.swift index cd4b46e..20907e6 100644 --- a/Packages/SpaceTraceKit/Tests/SpaceTracePersistenceTests/SQLiteEventJournalRepositoryTests.swift +++ b/Packages/SpaceTraceKit/Tests/SpaceTracePersistenceTests/SQLiteEventJournalRepositoryTests.swift @@ -1,6 +1,7 @@ import Foundation import Testing import SpaceTraceApplication +import SQLite3 @testable import SpaceTracePersistence struct SQLiteEventJournalRepositoryTests { @@ -14,6 +15,26 @@ struct SQLiteEventJournalRepositoryTests { } } + @Test("Schema version one journal state migrates to revisioned work") + func migratesVersionOne() async throws { + let fixture = try TemporaryDatabase() + try createVersionOneFixture(at: fixture.databaseURL) + let repository = try SQLiteEventJournalRepository(databaseURL: fixture.databaseURL) + let streamID = try EventStreamID("volume-v1:generation-1") + + #expect(try await repository.checkpoint(for: streamID) == EventJournalCursor(42)) + let workItem = try #require( + try await repository.pendingDirtyWork(for: streamID, limit: 1).first + ) + let initialRevision = try DirtyRegionRevision(1) + #expect(workItem.region.path.rawValue == "/Users/example/Documents") + #expect(workItem.region.maximumCursor == EventJournalCursor(42)) + #expect(workItem.revision == initialRevision) + + try await repository.close() + fixture.remove() + } + @Test("A batch makes dirty work and its checkpoint visible atomically") func atomicCommit() async throws { try await withRepository { repository in @@ -69,6 +90,90 @@ struct SQLiteEventJournalRepositoryTests { } } + @Test("Cursor-free calibration work does not create or advance a checkpoint") + func marksDirtyWithoutCursor() async throws { + try await withRepository { repository in + let streamID = try EventStreamID("volume-root:generation-1") + let region = try DirtyRegion( + path: DirtyRegionPath("/Users/example"), + reasons: [.rootChanged, .requiresCalibration], + maximumCursor: nil + ) + try await repository.markDirty(streamID: streamID, regions: [region]) + + #expect(try await repository.checkpoint(for: streamID) == nil) + #expect(try await repository.dirtyRegions(for: streamID) == [region]) + } + } + + @Test("An ancestor region absorbs durable descendant work") + func coalescesOverlappingPaths() async throws { + try await withRepository { repository in + let streamID = try EventStreamID("volume-overlap:generation-1") + try await repository.commit( + makeBatch( + streamID: streamID, + path: "/Users/example/Documents/Project", + reasons: [.created], + cursor: 3 + ) + ) + try await repository.commit( + makeBatch( + streamID: streamID, + path: "/Users/example/Documents", + reasons: [.mustScanSubdirectories], + cursor: 4 + ) + ) + + let region = try #require( + try await repository.dirtyRegions(for: streamID).first + ) + #expect(try await repository.dirtyRegions(for: streamID).count == 1) + #expect(region.path.rawValue == "/Users/example/Documents") + #expect(region.reasons == [.created, .mustScanSubdirectories]) + #expect(region.maximumCursor == EventJournalCursor(4)) + } + } + + @Test("A stale scan token cannot clear work updated during calibration") + func conditionalResolutionPreservesNewerWork() async throws { + try await withRepository { repository in + let streamID = try EventStreamID("volume-race:generation-1") + try await repository.commit( + makeBatch( + streamID: streamID, + path: "/Users/example/Documents", + reasons: [.contentModified], + cursor: 1 + ) + ) + let stale = try #require( + try await repository.pendingDirtyWork(for: streamID, limit: 1).first + ) + try await repository.markDirty( + streamID: streamID, + regions: [ + try DirtyRegion( + path: stale.region.path, + reasons: [.removed], + maximumCursor: nil + ), + ] + ) + + #expect(try await repository.resolve(stale, for: streamID) == false) + let current = try #require( + try await repository.pendingDirtyWork(for: streamID, limit: 1).first + ) + #expect(current.revision > stale.revision) + #expect(current.region.reasons == [.contentModified, .removed]) + #expect(try await repository.resolve(current, for: streamID)) + #expect(try await repository.dirtyRegions(for: streamID).isEmpty) + } + } + @Test("A checkpoint regression is rejected without persisting its dirty work") func rejectsCursorRegression() async throws { try await withRepository { repository in @@ -278,6 +383,50 @@ private func makeBatch( ) } +private func createVersionOneFixture(at databaseURL: URL) throws { + var database: OpaquePointer? + guard sqlite3_open(databaseURL.path, &database) == SQLITE_OK, let database else { + throw MigrationFixtureError.openFailed + } + defer { sqlite3_close(database) } + + let sql = """ + CREATE TABLE schema_migration ( + version INTEGER PRIMARY KEY, + applied_at_ms INTEGER NOT NULL, + checksum TEXT NOT NULL + ); + CREATE TABLE event_checkpoint ( + stream_id TEXT PRIMARY KEY NOT NULL, + cursor_be BLOB NOT NULL CHECK(length(cursor_be) = 8) + ) WITHOUT ROWID; + CREATE TABLE dirty_region ( + stream_id TEXT NOT NULL, + path TEXT NOT NULL, + reasons INTEGER NOT NULL CHECK(reasons != 0), + maximum_cursor_be BLOB NOT NULL CHECK(length(maximum_cursor_be) = 8), + PRIMARY KEY(stream_id, path), + FOREIGN KEY(stream_id) REFERENCES event_checkpoint(stream_id) + ) WITHOUT ROWID; + INSERT INTO schema_migration(version, applied_at_ms, checksum) + VALUES(1, 0, 'event-journal-v1-big-endian-cursor'); + INSERT INTO event_checkpoint(stream_id, cursor_be) + VALUES('volume-v1:generation-1', X'000000000000002A'); + INSERT INTO dirty_region(stream_id, path, reasons, maximum_cursor_be) + VALUES('volume-v1:generation-1', '/Users/example/Documents', 8, + X'000000000000002A'); + PRAGMA user_version = 1; + """ + guard sqlite3_exec(database, sql, nil, nil, nil) == SQLITE_OK else { + throw MigrationFixtureError.createFailed + } +} + +private enum MigrationFixtureError: Error { + case openFailed + case createFailed +} + private func makeBatch( streamID: EventStreamID, path: String, diff --git a/README.md b/README.md index 984c4d3..c134879 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ The MVP does **not** promise exact process attribution, reproduce Apple's System - Supported baseline: macOS 15.6+, Apple Silicon first - Minimum-version qualification: build/test configuration is aligned; a macOS 15.6 runtime matrix remains required before Public Beta - Proposed implementation: Swift, SwiftUI with targeted AppKit integration -- Implemented foundation: local `SpaceTraceKit` modules for domain observations, application persistence ports, a bounded FSEvents bridge, and an actor-isolated SQLite journal prototype +- Implemented foundation: local `SpaceTraceKit` modules for domain observations, a bounded FSEvents bridge, conservative dirty-region planning, actor-isolated calibration orchestration, and a revision-safe SQLite journal prototype - Proposed license: MIT (**TBD until explicitly approved**) - Distribution and signing: Developer ID, notarization, and update strategy remain release-blocking decisions diff --git a/docs/engineering/implementation-status.md b/docs/engineering/implementation-status.md index 9619288..d52a57b 100644 --- a/docs/engineering/implementation-status.md +++ b/docs/engineering/implementation-status.md @@ -12,16 +12,19 @@ This document records what the first implementation slice proves and, equally im | --- | --- | --- | | Domain observations | Checked byte quantities, typed identities, coverage states, comparable observations, and metric-preserving deltas | Unknown or partial evidence cannot be presented as a complete zero-byte observation | | FSEvents bridge | Public flag interpretation, restart-safe full-history replay, root-change sentinel handling, and a single-consumer bounded `AsyncThrowingStream` adapter | Events are invalidation hints only; callback-buffer loss becomes an explicit calibration gap and sentinel ID zero never becomes a durable cursor | -| Event journal port | Application-owned stream, cursor, dirty-region, reason, and batch contracts | A checkpoint cannot be accepted without durable dirty work in the same batch | -| SQLite prototype | Actor-owned SQLite3 connection, WAL, foreign keys, schema v1, big-endian `UInt64` cursors, atomic upsert/checkpoint transaction, and rollback fault seam | Durable cursor advancement is atomic with dirty-region persistence and cannot regress | +| Invalidation mapping | Adapter-to-application semantic mapping, lexical scope validation, file-to-parent projection, replay-overlap filtering, and ancestor coalescing | Ambiguous paths and continuity gaps sacrifice precision by falling back to scope calibration; they never manufacture narrow certainty | +| Calibration pipeline | Actor-isolated ingestion and bounded reconciliation orchestration behind an application-owned scanner port | Partial scans retain dirty work; a completed stale scan cannot clear work updated while it was running | +| Event journal port | Application-owned stream, cursor, dirty-region, row-revision, reason, batch, and conditional-resolution contracts | A checkpoint cannot be accepted without durable dirty work in the same batch; cursor-free sentinels cannot move it | +| SQLite prototype | Actor-owned SQLite3 connection, WAL, schema v2, big-endian `UInt64` cursors and revisions, atomic coalescing/checkpoint transaction, cursor-free calibration markers, and rollback fault seam | Durable cursor advancement is atomic with dirty-region persistence, cannot regress, and scan finalization is revision-conditional | | Build integration | Local `SpaceTraceKit` package linked to the macOS application target | The application composition root can depend on modular non-UI targets without source duplication | ## Verification evidence The following gates passed on 2026-07-18 with Swift 6.2.1 and Xcode 26.1.1: -- `swift test --package-path Packages/SpaceTraceKit`: 45 tests in 7 suites; +- `swift test --package-path Packages/SpaceTraceKit`: 63 tests in 11 suites; - the same package tests with complete strict-concurrency diagnostics and compiler warnings treated as errors; +- an adapter-to-application-to-real-SQLite integration test with an injected calibration scanner; - `make verify`, including architecture checks, package tests, Xcode scheme discovery, Debug build, application unit tests, and Release build; - Xcode compile and link target `arm64-apple-macos15.6` with the local package resolved from this repository. @@ -30,6 +33,7 @@ The strict-concurrency run is an audit for newly introduced package code. The ap ## Deliberately not claimed - No user-facing scan, history, explanation, menu-bar, permission, or export workflow is implemented. +- The application pipeline owns a scanner port, but the production metadata enumerator and atomic observation staging/finalization are not implemented yet. - No exact byte delta or process attribution is inferred from FSEvents. - The native FSEvents bridge has not yet passed an isolated APFS-volume lifecycle/integration suite. - The SQLite adapter is a dependency-free architecture prototype. ADR-004's GRDB, migration-fixture, retention, disk-full, corruption, and benchmark decisions are still open. @@ -38,9 +42,9 @@ The strict-concurrency run is an audit for newly introduced package code. The ap ## Next acceptance gates -1. Create deterministic application-layer event-to-dirty-region mapping and ancestor coalescing tests, including `FullHistory` overlap events below the saved cursor. +1. Implement the bounded metadata enumerator and staged observation finalization behind `CalibrationScanner`, with permission and cancellation coverage. 2. Run FSEvents create/start/replay/cancel/drop integration tests on a disposable APFS scope without scanning a developer home directory. 3. Decide per-device stream identity and volume-generation behavior before persisting production cursors. 4. Complete the ADR-004 GRDB-versus-raw-SQLite review, including license, build, migration, and notarization evidence. -5. Add schema golden fixtures, disk-full/corruption tests, retention behavior, and oldest-supported-OS qualification. +5. Expand the schema migration fixture matrix and add disk-full/corruption tests, retention behavior, and oldest-supported-OS qualification. 6. Keep all spike code unreachable from user-visible workflows until its corresponding ADR is accepted.