diff --git a/Packages/CmuxFileWatch/Package.swift b/Packages/CmuxFileWatch/Package.swift new file mode 100644 index 000000000000..2f9e096f7413 --- /dev/null +++ b/Packages/CmuxFileWatch/Package.swift @@ -0,0 +1,35 @@ +// swift-tools-version: 6.0 + +import PackageDescription + +let package = Package( + name: "CmuxFileWatch", + platforms: [ + .macOS(.v14), + ], + products: [ + .library( + name: "CmuxFileWatch", + targets: ["CmuxFileWatch"] + ), + ], + targets: [ + .target( + name: "CmuxFileWatch", + swiftSettings: [ + .swiftLanguageMode(.v6), + .enableUpcomingFeature("ExistentialAny"), + .enableUpcomingFeature("InternalImportsByDefault"), + ] + ), + .testTarget( + name: "CmuxFileWatchTests", + dependencies: ["CmuxFileWatch"], + swiftSettings: [ + .swiftLanguageMode(.v6), + .enableUpcomingFeature("ExistentialAny"), + .enableUpcomingFeature("InternalImportsByDefault"), + ] + ), + ] +) diff --git a/Packages/CmuxFileWatch/README.md b/Packages/CmuxFileWatch/README.md new file mode 100644 index 000000000000..dbc8d7664c5b --- /dev/null +++ b/Packages/CmuxFileWatch/README.md @@ -0,0 +1,67 @@ +# CmuxFileWatch + +Generic recursive-path filesystem watching, as a Wave-2 infrastructure leaf of +the cmux modular refactor. + +`RecursivePathWatcher` is an `actor` that wraps an `FSEventStream` (hidden behind +`FileSystemEventStream`) and reports coalesced changes as an `AsyncStream`. +The caller decides which paths matter for its domain, constructs a watcher, and +reacts to each yielded element. It has no knowledge of what it is watching. + +A leading-edge throttle folds a burst of filesystem events into a single yield. +The throttle delay is driven through the `FileWatchClock` seam: production uses +`SystemFileWatchClock` (a real `Task.sleep`); tests inject a clock that suspends +until released, so coalescing is verified deterministically with no real +waiting. + +```swift +guard let watcher = RecursivePathWatcher(paths: paths) else { return } +let task = Task { @MainActor in + for await _ in watcher.events { reload() } +} +// teardown +task.cancel() +await watcher.stop() +``` + +`FSEventStream` is the only macOS primitive that watches a *set of paths +recursively* with one coalescing stream, so it is used here rather than a +per-descriptor `DispatchSource` file source. The first consumer is the workspace +git-metadata watcher in `TabManager` (the git path-resolution logic stays there, +bound for `CmuxSidebarGit`). `CmuxSettings`' `JSONConfigFileWatcher` is a +candidate to migrate onto this package as a second consumer. + +## Testing + +`RecursivePathWatcher` takes its throttle clock via `FileWatchClock`, so the +coalescing behavior is testable with no real waiting. Conform a test clock whose +`sleep(for:)` suspends until the test releases it, then drive synthetic events +through the throttle and assert the yielded `events`. The package ships such a +gate clock in `Tests/CmuxFileWatchTests`: + +```swift +private actor GateClock: FileWatchClock { + private var sleepers: [CheckedContinuation] = [] + func sleep(for duration: Duration) async throws { + await withCheckedContinuation { sleepers.append($0) } + } + func releaseOne() { if !sleepers.isEmpty { sleepers.removeFirst().resume() } } +} + +@Test func burstCoalescesIntoOneYield() async { + let clock = GateClock() + // The test-only initializer drives the throttle directly, with no FSEventStream. + let watcher = RecursivePathWatcher(testThrottleClock: clock) + var events = watcher.events.makeAsyncIterator() + + for _ in 0..<5 { await watcher.simulateFileSystemEventForTesting() } // one window + await clock.releaseOne() // flush + #expect(await events.next() != nil) // exactly one yield + + await watcher.stop() + #expect(await events.next() == nil) // stream finished +} +``` + +Production code constructs the watcher with the default `SystemFileWatchClock` +and a real path set; only tests use the `testThrottleClock` initializer. diff --git a/Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileSystemEventStream.swift b/Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileSystemEventStream.swift new file mode 100644 index 000000000000..08fde56a3b44 --- /dev/null +++ b/Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileSystemEventStream.swift @@ -0,0 +1,121 @@ +import CoreServices +import Foundation + +/// A thin owner of an `FSEventStream` that reports raw filesystem events through +/// a `@Sendable` sink. +/// +/// `FSEventStream` is a C API with no async-native replacement, and it is the +/// only macOS primitive that watches a *set of paths recursively* with a single +/// coalescing stream (a `DispatchSource` file source watches one descriptor and +/// does not recurse). It stays hidden behind this type; consumers +/// (``RecursivePathWatcher``) observe events only via the watcher's +/// `AsyncStream`. The stream is configured for file-level events. +/// +/// **Threading.** Every instance shares one serial dispatch queue (rather than +/// one queue per stream) to bound thread usage when many workspaces are tracked. +/// All mutable state is touched only on that queue, which is why the type is +/// `@unchecked Sendable`. ``onEvent`` fires on the shared queue, so it MUST be +/// non-blocking — a slow sink would serialize behind every other stream's +/// teardown. The production sink only spawns a `Task` and returns. +/// +/// **Context lifetime.** The stream is registered with FSEvents as its own +/// context `info` pointer, passed *unretained* (no `retain`/`release` callbacks). +/// That is safe because ``stop()`` invalidates the stream synchronously on the +/// shared queue before `deinit` returns: FSEvents delivers callbacks on that +/// same serial queue, so any in-flight or already-enqueued callback runs to +/// completion before the `queue.sync` teardown block, and none is delivered +/// after `FSEventStreamInvalidate`. No callback ever touches a freed instance, +/// so a separately retained context box is unnecessary. +final class FileSystemEventStream: @unchecked Sendable { + private static let queueSpecificKey = DispatchSpecificKey() + private static let queue: DispatchQueue = { + let queue = DispatchQueue(label: "com.cmux.recursive-path-watcher", qos: .utility) + queue.setSpecific(key: queueSpecificKey, value: 1) + return queue + }() + + /// The C trampoline `FSEventStreamCreate` requires. + /// + /// It must be a context-free `@convention(c)` function pointer, so it cannot + /// be an instance method (which would be curried over `self`). The owning + /// stream is recovered from the context's `info` pointer instead — passed + /// *unretained* (see the type's "Context lifetime" note), so this uses + /// `takeUnretainedValue()` and never adjusts the reference count. + private static let callback: FSEventStreamCallback = { _, info, _, _, _, _ in + guard let info else { return } + Unmanaged.fromOpaque(info).takeUnretainedValue().onEvent() + } + + /// The non-blocking sink invoked on the shared queue for each coalesced batch + /// of filesystem events. + private let onEvent: @Sendable () -> Void + private var stream: FSEventStreamRef? + + /// Creates and starts a stream for `paths`. + /// + /// - Parameters: + /// - paths: The files and directories to watch. Must be non-empty. + /// - latency: The FSEvents coalescing latency in seconds. + /// - onEvent: A non-blocking sink invoked on the shared queue for each + /// coalesced batch of filesystem events. + /// - Returns: `nil` if `paths` is empty or the underlying `FSEventStream` + /// could not be created or started. + init?(paths: [String], latency: TimeInterval, onEvent: @escaping @Sendable () -> Void) { + guard !paths.isEmpty else { return nil } + self.onEvent = onEvent + self.stream = nil + + var context = FSEventStreamContext( + version: 0, + info: Unmanaged.passUnretained(self).toOpaque(), + retain: nil, + release: nil, + copyDescription: nil + ) + let flags = FSEventStreamCreateFlags(kFSEventStreamCreateFlagFileEvents) + guard let stream = FSEventStreamCreate( + nil, + Self.callback, + &context, + paths as CFArray, + FSEventStreamEventId(kFSEventStreamEventIdSinceNow), + latency, + flags + ) else { + return nil + } + self.stream = stream + FSEventStreamSetDispatchQueue(stream, Self.queue) + guard FSEventStreamStart(stream) else { + stop() + return nil + } + } + + /// Stops and tears down the stream. Idempotent. + /// + /// Teardown runs synchronously on the shared queue so it completes before the + /// caller continues — critically, before `deinit` returns. An async hop could + /// let the instance deallocate before the stream is invalidated, leaking the + /// `FSEventStream`. The `getSpecific` check tears down inline when already on + /// the queue, avoiding a deadlock. + func stop() { + if DispatchQueue.getSpecific(key: Self.queueSpecificKey) != nil { + stopOnQueue() + } else { + Self.queue.sync { stopOnQueue() } + } + } + + private func stopOnQueue() { + guard let stream else { return } + FSEventStreamStop(stream) + FSEventStreamInvalidate(stream) + FSEventStreamRelease(stream) + self.stream = nil + } + + deinit { + stop() + } +} diff --git a/Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileWatchClock.swift b/Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileWatchClock.swift new file mode 100644 index 000000000000..9657524562f7 --- /dev/null +++ b/Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileWatchClock.swift @@ -0,0 +1,37 @@ +import Foundation + +/// An injectable clock for a watcher's leading-edge coalescing delay. +/// +/// A ``RecursivePathWatcher`` coalesces a burst of filesystem events into a +/// single yield by arming one bounded delay on the first event of a window and +/// ignoring further events until it fires. That delay is an *intended* rate +/// limit, never a poll or a "let things settle" hack, so it is driven through +/// this seam rather than a raw timer. +/// +/// Production uses ``SystemFileWatchClock`` (a real `Task.sleep`). Tests inject a +/// fake that suspends until released, so the coalescing behavior is verified +/// deterministically with no real waiting. +/// +/// ```swift +/// let watcher = RecursivePathWatcher(paths: paths, clock: SystemFileWatchClock()) +/// ``` +public protocol FileWatchClock: Sendable { + /// Suspends for `duration`, throwing `CancellationError` if the surrounding + /// task is cancelled first. + /// + /// - Parameter duration: How long to suspend before resuming. + func sleep(for duration: Duration) async throws +} + +/// The production ``FileWatchClock``, backed by `Task.sleep`. +public struct SystemFileWatchClock: FileWatchClock { + /// Creates a system clock. + public init() {} + + public func sleep(for duration: Duration) async throws { + // Bounded, cancellable, intended throttle delay behind the clock seam + // (modern-concurrency policy carve-out): coalesce a filesystem-event burst + // into one yield. Cancelled when the watcher stops. + try await Task.sleep(for: duration) + } +} diff --git a/Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileWatcher.swift b/Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileWatcher.swift new file mode 100644 index 000000000000..4eb5e829a29d --- /dev/null +++ b/Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileWatcher.swift @@ -0,0 +1,238 @@ +import Foundation + +/// Watches a single file or directory for change events and exposes them as an +/// `AsyncStream`. +/// +/// This is the single-path counterpart to ``RecursivePathWatcher``: it wraps +/// `DispatchSource.makeFileSystemObjectSource` (the kqueue-backed primitive for +/// per-descriptor file events on macOS, which has no async-native replacement) +/// rather than `FSEventStream`. Use it to react to changes to one path — a +/// config file, a settings file, a directory listing — where recursive subtree +/// watching is not needed. +/// +/// **Recovery.** Alongside the path itself the watcher observes the *nearest +/// existing ancestor directory*, so its stream recovers when the path is created, +/// replaced, atomically saved, or when an intervening directory is created or +/// removed after the watcher starts (the path frequently does not exist on first +/// use). As ancestors appear the directory source migrates closer to the path, +/// and the path-level source reattaches to the current inode. +/// +/// An optional leading-edge ``throttle`` coalesces a burst of events into a +/// single yield, driven through the injectable ``FileWatchClock`` (see +/// ``RecursivePathWatcher`` for the rationale and the deterministic test +/// pattern). With no throttle, every coalesced `DispatchSource` batch yields one +/// element. +/// +/// ```swift +/// let watcher = FileWatcher(path: configPath, throttle: .milliseconds(300)) +/// let task = Task { @MainActor in +/// for await _ in watcher.events { reloadFromDisk() } +/// } +/// // teardown +/// task.cancel() +/// await watcher.stop() +/// ``` +public actor FileWatcher { + /// Stream of change events. Yields when the watched path or a watched + /// ancestor directory changes. Finishes when ``stop()`` is called or the + /// watcher is deallocated. + public nonisolated let events: AsyncStream + + private let path: String + private let throttle: Duration? + private let clock: any FileWatchClock + // DispatchSource requires a queue; this is internal isolation only and never + // exposed. Sources are owned by the actor and torn down on stop/deinit. + private let queue: DispatchQueue + private let continuation: AsyncStream.Continuation + // The DispatchSource event handlers yield into this (a Sendable value), not + // the actor, so the sources can be attached synchronously in `init`. + private let rawContinuation: AsyncStream.Continuation + // File-descriptor lifetime is owned by each source's `setCancelHandler`, + // which calls `close(fd)` exactly once when the source's cancel completes. + private var fileSource: (any DispatchSourceFileSystemObject)? + private var directorySource: (any DispatchSourceFileSystemObject)? + private var watchedDirectory: String? + private var throttleTask: Task? + private var isStopped = false + + // OptionSet masks. Inlined (not stored as statics) because + // `DispatchSource.FileSystemEvent` is not `Sendable`. + private static var fileEventMask: DispatchSource.FileSystemEvent { + [.write, .delete, .rename, .extend, .attrib] + } + private static var directoryEventMask: DispatchSource.FileSystemEvent { + [.write, .rename, .delete] + } + + /// Creates and starts a watcher for `path`. + /// + /// - Parameters: + /// - path: The file or directory to watch. Need not exist yet; the nearest + /// existing ancestor is observed so the stream recovers when it appears. + /// - throttle: Optional leading-edge coalescing window. `nil` (default) + /// yields one element per `DispatchSource` event batch. + /// - clock: The clock driving the throttle. Defaults to + /// ``SystemFileWatchClock``. Ignored when `throttle` is `nil`. + public init( + path: String, + throttle: Duration? = nil, + clock: any FileWatchClock = SystemFileWatchClock() + ) { + self.path = path + self.throttle = throttle + self.clock = clock + self.queue = DispatchQueue(label: "com.cmux.file-watcher", qos: .utility) + let (events, eventsContinuation) = AsyncStream.makeStream() + self.events = events + self.continuation = eventsContinuation + let (rawEvents, rawContinuation) = AsyncStream.makeStream() + self.rawContinuation = rawContinuation + + // Attach the sources synchronously so the watcher is already listening + // when init returns. The handlers capture `rawContinuation` (Sendable), + // not `self`, so this does not escape the actor mid-init. + let directory = Self.nearestExistingDirectory(forPath: path) + self.watchedDirectory = directory + self.directorySource = Self.makeSource( + forPath: directory, + eventMask: Self.directoryEventMask, + queue: queue, + rawContinuation: rawContinuation + ) + self.fileSource = Self.makeSource( + forPath: path, + eventMask: Self.fileEventMask, + queue: queue, + rawContinuation: rawContinuation + ) + + // Drain raw events through the actor (reattach sources, then + // yield/throttle). Started last so init touches no isolated state after + // `self` escapes; ends when `rawEvents` finishes (stop/deinit). + Task { [weak self] in + for await _ in rawEvents { + await self?.handleRawEvent() + } + } + } + + /// Stops the watcher, tears down its sources, and finishes ``events``. + /// Idempotent. + public func stop() { + isStopped = true + throttleTask?.cancel() + throttleTask = nil + fileSource?.cancel() + fileSource = nil + directorySource?.cancel() + directorySource = nil + rawContinuation.finish() + continuation.finish() + } + + deinit { + // DispatchSource.cancel() is thread-safe; safe from deinit. + fileSource?.cancel() + directorySource?.cancel() + throttleTask?.cancel() + rawContinuation.finish() + continuation.finish() + } + + // MARK: - Private + + /// Walks up from `path`'s parent to the first existing directory, so the + /// watcher can observe an ancestor while the path itself does not exist. + /// Falls back to the current directory. + private static func nearestExistingDirectory(forPath path: String) -> String { + let fileManager = FileManager.default + var current = (path as NSString).deletingLastPathComponent + var seen = Set() + while !current.isEmpty { + let standardized = (current as NSString).standardizingPath + guard seen.insert(standardized).inserted else { break } + var isDirectory: ObjCBool = false + if fileManager.fileExists(atPath: standardized, isDirectory: &isDirectory), + isDirectory.boolValue { + return standardized + } + let parent = (standardized as NSString).deletingLastPathComponent + if parent == standardized || parent.isEmpty { break } + current = parent + } + return fileManager.currentDirectoryPath + } + + /// Builds, resumes, and returns a source for `path`, or `nil` if it cannot + /// be opened. The event handler captures only `rawContinuation`. + private static func makeSource( + forPath path: String, + eventMask: DispatchSource.FileSystemEvent, + queue: DispatchQueue, + rawContinuation: AsyncStream.Continuation + ) -> (any DispatchSourceFileSystemObject)? { + let fd = open(path, O_EVTONLY) + guard fd >= 0 else { return nil } + let source = DispatchSource.makeFileSystemObjectSource( + fileDescriptor: fd, + eventMask: eventMask, + queue: queue + ) + source.setEventHandler { rawContinuation.yield(()) } + source.setCancelHandler { close(fd) } + source.resume() + return source + } + + /// Reacts to a raw source event: re-evaluate which ancestor directory to + /// watch and reattach the path source to the current inode, then signal a + /// change (throttled if configured). + private func handleRawEvent() { + guard !isStopped else { return } + reattachSources() + guard let throttle else { + continuation.yield(()) + return + } + // Leading-edge throttle: the first event arms one bounded delay; events + // during the window coalesce (the `throttleTask == nil` guard). + guard throttleTask == nil else { return } + let clock = self.clock + throttleTask = Task { [weak self] in + try? await clock.sleep(for: throttle) + await self?.flushThrottle() + } + } + + private func flushThrottle() { + throttleTask = nil + guard !isStopped else { return } + continuation.yield(()) + } + + /// Moves the directory source to the current nearest existing ancestor (if it + /// changed) and reattaches the path source to the current inode. Each previous + /// source's `setCancelHandler` closes its own fd. + private func reattachSources() { + let directory = Self.nearestExistingDirectory(forPath: path) + if directory != watchedDirectory { + directorySource?.cancel() + directorySource = Self.makeSource( + forPath: directory, + eventMask: Self.directoryEventMask, + queue: queue, + rawContinuation: rawContinuation + ) + watchedDirectory = directory + } + let newFileSource = Self.makeSource( + forPath: path, + eventMask: Self.fileEventMask, + queue: queue, + rawContinuation: rawContinuation + ) + fileSource?.cancel() + fileSource = newFileSource + } +} diff --git a/Packages/CmuxFileWatch/Sources/CmuxFileWatch/RecursivePathWatcher.swift b/Packages/CmuxFileWatch/Sources/CmuxFileWatch/RecursivePathWatcher.swift new file mode 100644 index 000000000000..3cc657ed3e37 --- /dev/null +++ b/Packages/CmuxFileWatch/Sources/CmuxFileWatch/RecursivePathWatcher.swift @@ -0,0 +1,168 @@ +import Foundation + +/// Watches a set of filesystem paths recursively and reports changes as a +/// coalesced `AsyncStream`. +/// +/// Construct one with the paths to watch (the caller resolves which paths matter +/// for its domain) and consume ``events`` to react to changes: +/// +/// ```swift +/// guard let watcher = RecursivePathWatcher(paths: paths) else { return } +/// let task = Task { @MainActor in +/// for await _ in watcher.events { reload() } +/// } +/// // later: task.cancel(); await watcher.stop() +/// ``` +/// +/// **Coalescing.** A leading-edge throttle folds a burst into one yield: the +/// first event in a window arms a single bounded delay (``FileWatchClock``) and +/// events arriving while that delay is pending are folded into it. Combined with +/// the underlying `FSEventStream` latency, the worst-case delay from a change to +/// an ``events`` element is roughly twice the throttle interval. During a +/// sustained storm an element is yielded at most once per window — it does *not* +/// wait for changes to stop, which keeps reactions responsive without per-event +/// churn. +/// +/// **Construction.** The `FSEventStream` is created synchronously in ``init``, +/// so the watcher is already listening when it returns (nothing is missed in the +/// gap a deferred start would open) and ``init`` fails (`nil`) if the stream +/// cannot be created. The stream's `@Sendable` sink forwards into a private +/// raw-event `AsyncStream` rather than capturing the actor, which is what lets +/// creation happen in-`init`; a single actor-isolated pump drains that raw stream +/// and applies the throttle. The pump's lifetime is the raw stream's: ``stop()`` +/// and `deinit` finish it. +public actor RecursivePathWatcher { + /// The paths this watcher observes, as passed to ``init(paths:clock:)``. + /// + /// Exposed so callers can compare against a freshly resolved set and skip + /// recreating an equivalent watcher. + public nonisolated let watchedPaths: [String] + + /// Stream of coalesced change events. Yields one element per throttle window + /// in which at least one filesystem event affected a watched path. Finishes + /// when ``stop()`` is called or the watcher is deallocated. + public nonisolated let events: AsyncStream + + private let continuation: AsyncStream.Continuation + private let clock: any FileWatchClock + // nil only for the test-throttle initializer, which drives the throttle + // directly without a real FSEventStream. + private let eventStream: FileSystemEventStream? + // Finishing this ends the pump task (see init); raw FS events flow through it. + private let rawContinuation: AsyncStream.Continuation + private var throttleTask: Task? + private var isStopped = false + + /// The `FSEventStream` coalescing latency, in seconds. + private static let streamLatency = 0.25 + /// The leading-edge throttle window. Combined with ``streamLatency`` the + /// worst-case change-to-yield delay is roughly twice this. + private static let throttleInterval: Duration = .milliseconds(250) + + /// Creates and starts a watcher for `paths`. + /// + /// - Parameters: + /// - paths: The files and directories to watch. Must be non-empty. + /// - clock: The clock driving the coalescing throttle. Defaults to + /// ``SystemFileWatchClock``. + /// - Returns: `nil` if `paths` is empty or the underlying `FSEventStream` + /// could not be created or started. On success the stream is already + /// listening. + public init?( + paths: [String], + clock: any FileWatchClock = SystemFileWatchClock() + ) { + guard !paths.isEmpty else { return nil } + self.watchedPaths = paths + self.clock = clock + let (events, eventsContinuation) = AsyncStream.makeStream() + self.events = events + self.continuation = eventsContinuation + let (rawEvents, rawContinuation) = AsyncStream.makeStream() + self.rawContinuation = rawContinuation + + // The sink captures `rawContinuation` (a Sendable value), not `self`, so + // the stream can be created synchronously here without escaping the + // actor mid-init. + guard let eventStream = FileSystemEventStream( + paths: paths, + latency: Self.streamLatency, + onEvent: { rawContinuation.yield(()) } + ) else { + eventsContinuation.finish() + rawContinuation.finish() + return nil + } + self.eventStream = eventStream + + // Drain raw FS events through the actor-isolated throttle. Started last so + // init touches no isolated state after `self` escapes into the task; it + // holds `self` weakly and ends when `rawEvents` finishes (stop/deinit). + Task { [weak self] in + for await _ in rawEvents { + await self?.handleRawEvent() + } + } + } + + /// Creates a watcher with no underlying `FSEventStream`, driven only by + /// ``simulateFileSystemEventForTesting()``. + /// + /// Used by the package tests to exercise the coalescing throttle in isolation + /// with an injected clock and no real filesystem dependency. + init(testThrottleClock clock: any FileWatchClock) { + self.watchedPaths = [] + self.clock = clock + let (events, eventsContinuation) = AsyncStream.makeStream() + self.events = events + self.continuation = eventsContinuation + let (_, rawContinuation) = AsyncStream.makeStream() + self.rawContinuation = rawContinuation + self.eventStream = nil + } + + /// Stops the watcher, tears down the underlying stream, and finishes + /// ``events``. Idempotent. + public func stop() { + isStopped = true + throttleTask?.cancel() + throttleTask = nil + eventStream?.stop() + rawContinuation.finish() + continuation.finish() + } + + deinit { + // FSEventStream teardown is synchronous and thread-safe; finishing the + // continuations ends the pump and any consumer. + eventStream?.stop() + throttleTask?.cancel() + rawContinuation.finish() + continuation.finish() + } + + /// Leading-edge throttle entry point. The first event of a window arms one + /// delay; events arriving while it is pending are no-ops (the `throttleTask + /// == nil` guard), so a burst yields a single ``events`` element. + private func handleRawEvent() { + guard !isStopped, throttleTask == nil else { return } + let clock = self.clock + let interval = Self.throttleInterval + throttleTask = Task { [weak self] in + try? await clock.sleep(for: interval) + await self?.flushThrottle() + } + } + + private func flushThrottle() { + throttleTask = nil + guard !isStopped else { return } + continuation.yield(()) + } + + /// Feeds a synthetic filesystem event into the throttle. Test-only seam used + /// by ``init(testThrottleClock:)``-constructed watchers. + func simulateFileSystemEventForTesting() { + handleRawEvent() + } +} diff --git a/Packages/CmuxFileWatch/Tests/CmuxFileWatchTests/FileWatcherTests.swift b/Packages/CmuxFileWatch/Tests/CmuxFileWatchTests/FileWatcherTests.swift new file mode 100644 index 000000000000..ccbe9b0f6220 --- /dev/null +++ b/Packages/CmuxFileWatch/Tests/CmuxFileWatchTests/FileWatcherTests.swift @@ -0,0 +1,146 @@ +import Foundation +import Testing + +@testable import CmuxFileWatch + +@Suite struct FileWatcherTests { + /// Awaits the watcher's first event, bounded so a broken watcher fails the + /// test instead of hanging CI. + private func firstEvent(_ watcher: FileWatcher, within seconds: Double) async -> Bool { + await withTaskGroup(of: Bool.self) { group in + group.addTask { + var iterator = watcher.events.makeAsyncIterator() + return await iterator.next() != nil + } + group.addTask { + try? await Task.sleep(for: .seconds(seconds)) + return false + } + let result = await group.next() ?? false + group.cancelAll() + return result + } + } + + @Test func fileWriteYieldsEvent() async throws { + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("cmux-file-watcher-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: directory) } + let file = directory.appendingPathComponent("watched.txt") + try "initial".write(to: file, atomically: true, encoding: .utf8) + + let watcher = FileWatcher(path: file.path) + defer { Task { await watcher.stop() } } + + // Mutate after the watcher is listening (sources attach synchronously in init). + try "changed".write(to: file, atomically: false, encoding: .utf8) + + #expect(await firstEvent(watcher, within: 5)) + } + + @Test func fileCreatedAfterStartYieldsEvent() async throws { + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("cmux-file-watcher-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: directory) } + let file = directory.appendingPathComponent("not-yet.txt") + + // Watch a path that does not exist yet; the parent-directory source must + // recover when it is created. + let watcher = FileWatcher(path: file.path) + defer { Task { await watcher.stop() } } + + try "created".write(to: file, atomically: true, encoding: .utf8) + + #expect(await firstEvent(watcher, within: 5)) + } + + @Test func fileInNonexistentSubdirectoryRecoversViaAncestor() async throws { + let root = FileManager.default.temporaryDirectory + .appendingPathComponent("cmux-file-watcher-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: root) } + // Neither the intermediate directory nor the file exists yet; the watcher + // must observe `root` (the nearest existing ancestor) and recover. + let nested = root.appendingPathComponent("a/b", isDirectory: true) + let file = nested.appendingPathComponent("watched.txt") + + let watcher = FileWatcher(path: file.path) + defer { Task { await watcher.stop() } } + + try FileManager.default.createDirectory(at: nested, withIntermediateDirectories: true) + try "created".write(to: file, atomically: true, encoding: .utf8) + + #expect(await firstEvent(watcher, within: 5)) + } + + @Test func directoryTargetYieldsOnChildChange() async throws { + // The FileExplorer path: the watched path is itself a directory, and a + // change to its contents must yield. + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("cmux-file-watcher-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: directory) } + + let watcher = FileWatcher(path: directory.path) + defer { Task { await watcher.stop() } } + + try "child".write(to: directory.appendingPathComponent("child.txt"), atomically: true, encoding: .utf8) + + #expect(await firstEvent(watcher, within: 5)) + } + + @Test func atomicReplaceYieldsEvent() async throws { + // The MarkdownPanel / JSONConfigStore save path: an atomic write replaces + // the inode (temp file + rename), so the watcher must reattach via the + // directory source and still yield. + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("cmux-file-watcher-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: directory) } + let file = directory.appendingPathComponent("watched.txt") + try "initial".write(to: file, atomically: true, encoding: .utf8) + + let watcher = FileWatcher(path: file.path) + defer { Task { await watcher.stop() } } + + // `.atomic` writes to a sibling temp file then renames over the original, + // replacing the inode the file source was attached to. + try "replaced".write(to: file, atomically: true, encoding: .utf8) + + #expect(await firstEvent(watcher, within: 5)) + } + + @Test func throttledWatcherYieldsAfterChange() async throws { + // A throttled watcher (the FileExplorer/MarkdownPanel coalescing config) + // still delivers an event for a real change. + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("cmux-file-watcher-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: directory) } + let file = directory.appendingPathComponent("watched.txt") + try "initial".write(to: file, atomically: true, encoding: .utf8) + + let watcher = FileWatcher(path: file.path, throttle: .milliseconds(50)) + defer { Task { await watcher.stop() } } + + try "changed".write(to: file, atomically: false, encoding: .utf8) + + #expect(await firstEvent(watcher, within: 5)) + } + + @Test func stopFinishesStream() async { + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("cmux-file-watcher-\(UUID().uuidString)", isDirectory: true) + try? FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: directory) } + let file = directory.appendingPathComponent("watched.txt") + + let watcher = FileWatcher(path: file.path) + var iterator = watcher.events.makeAsyncIterator() + await watcher.stop() + let next: Void? = await iterator.next() + #expect(next == nil) + } +} diff --git a/Packages/CmuxFileWatch/Tests/CmuxFileWatchTests/RecursivePathWatcherTests.swift b/Packages/CmuxFileWatch/Tests/CmuxFileWatchTests/RecursivePathWatcherTests.swift new file mode 100644 index 000000000000..f1bd555cf4f5 --- /dev/null +++ b/Packages/CmuxFileWatch/Tests/CmuxFileWatchTests/RecursivePathWatcherTests.swift @@ -0,0 +1,106 @@ +import Foundation +import Testing + +@testable import CmuxFileWatch + +/// A clock whose `sleep(for:)` suspends until the test releases it, so the +/// watcher's coalescing throttle can be advanced with no real waiting. +private actor GateClock: FileWatchClock { + private var sleepers: [CheckedContinuation] = [] + private var arrivalWaiters: [CheckedContinuation] = [] + + func sleep(for duration: Duration) async throws { + await withCheckedContinuation { (continuation: CheckedContinuation) in + sleepers.append(continuation) + let waiters = arrivalWaiters + arrivalWaiters.removeAll() + for waiter in waiters { waiter.resume() } + } + } + + /// Number of throttle delays currently parked on the clock. + var sleeperCount: Int { sleepers.count } + + /// Suspends until at least one sleeper has registered. + func waitForSleeper() async { + if !sleepers.isEmpty { return } + await withCheckedContinuation { (continuation: CheckedContinuation) in + arrivalWaiters.append(continuation) + } + } + + /// Releases the oldest parked throttle delay. + func releaseOne() { + guard !sleepers.isEmpty else { return } + sleepers.removeFirst().resume() + } +} + +@Suite struct RecursivePathWatcherTests { + @Test func emptyPathsFailsInitialization() { + let watcher = RecursivePathWatcher(paths: []) + #expect(watcher == nil) + } + + @Test func realDirectoryStartsAndStops() async { + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("cmux-file-watch-\(UUID().uuidString)", isDirectory: true) + try? FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: directory) } + + let watcher = RecursivePathWatcher(paths: [directory.path]) + #expect(watcher != nil) + #expect(watcher?.watchedPaths == [directory.path]) + await watcher?.stop() + } + + /// A burst of events inside one throttle window coalesces into a single + /// yield, a fresh event re-arms the throttle, and `stop()` finishes the + /// stream. This is the leading-edge behavior the watcher provides: react once + /// per window during a storm, never once per event and never only after + /// changes stop. + @Test func burstCoalescesAndThrottleRearms() async { + let clock = GateClock() + let watcher = RecursivePathWatcher(testThrottleClock: clock) + var iterator = watcher.events.makeAsyncIterator() + + // Window 1: five events, but only the first arms the throttle. + for _ in 0..<5 { + await watcher.simulateFileSystemEventForTesting() + } + await clock.waitForSleeper() + #expect(await clock.sleeperCount == 1) + + await clock.releaseOne() + let first: Void? = await iterator.next() + #expect(first != nil) + + // Window 2: the throttle re-arms after the previous flush. + for _ in 0..<3 { + await watcher.simulateFileSystemEventForTesting() + } + await clock.waitForSleeper() + #expect(await clock.sleeperCount == 1) + + await clock.releaseOne() + let second: Void? = await iterator.next() + #expect(second != nil) + + await watcher.stop() + let afterStop: Void? = await iterator.next() + #expect(afterStop == nil) + } + + /// Events delivered after `stop()` produce no further yields. + @Test func eventsAfterStopAreIgnored() async { + let clock = GateClock() + let watcher = RecursivePathWatcher(testThrottleClock: clock) + var iterator = watcher.events.makeAsyncIterator() + + await watcher.stop() + await watcher.simulateFileSystemEventForTesting() + let next: Void? = await iterator.next() + #expect(next == nil) + #expect(await clock.sleeperCount == 0) + } +} diff --git a/Packages/CmuxSettings/Package.swift b/Packages/CmuxSettings/Package.swift index 679c7612d13e..aa0f75ff0255 100644 --- a/Packages/CmuxSettings/Package.swift +++ b/Packages/CmuxSettings/Package.swift @@ -13,9 +13,15 @@ let package = Package( targets: ["CmuxSettings"] ), ], + dependencies: [ + .package(path: "../CmuxFileWatch"), + ], targets: [ .target( - name: "CmuxSettings" + name: "CmuxSettings", + dependencies: [ + .product(name: "CmuxFileWatch", package: "CmuxFileWatch"), + ] ), .testTarget( name: "CmuxSettingsTests", diff --git a/Packages/CmuxSettings/README.md b/Packages/CmuxSettings/README.md index f00a627117b3..9429138afd0b 100644 --- a/Packages/CmuxSettings/README.md +++ b/Packages/CmuxSettings/README.md @@ -169,11 +169,9 @@ Reflection picks it up recursively; `catalog.all` includes every leaf. - **`UserDefaultsSettingsStore`** — `actor`. Async reads/writes/observe. Observation uses `NotificationCenter.default.notifications(named:)`. - **`JSONConfigStore`** — `actor`. Async reads/writes/observe. Owns one - `JSONConfigFileWatcher` and fans out file-change events to per-subscriber - bounded signal streams (no `N × parse` work under burst changes). -- **`JSONConfigFileWatcher`** — `actor` wrapping `DispatchSource` - file-system events behind an `AsyncStream`. The only public surface - is the typed event stream; the underlying dispatch sources never leak. + `CmuxFileWatch.FileWatcher` and fans out file-change events to per-subscriber + bounded signal streams (no `N × parse` work under burst changes). File + watching itself lives in the `CmuxFileWatch` package. ## Testing diff --git a/Packages/CmuxSettings/Sources/CmuxSettings/CmuxSettings.docc/CmuxSettings.md b/Packages/CmuxSettings/Sources/CmuxSettings/CmuxSettings.docc/CmuxSettings.md index 29a8664ec92e..6db0bb10e903 100644 --- a/Packages/CmuxSettings/Sources/CmuxSettings/CmuxSettings.docc/CmuxSettings.md +++ b/Packages/CmuxSettings/Sources/CmuxSettings/CmuxSettings.docc/CmuxSettings.md @@ -46,7 +46,6 @@ principles: ### JSON config internals -- ``JSONConfigFileWatcher`` - ``JSONCSanitizer`` - ``JSONPath`` - ``CmuxConfigLocation`` diff --git a/Packages/CmuxSettings/Sources/CmuxSettings/Stores/JSONConfigFileWatcher.swift b/Packages/CmuxSettings/Sources/CmuxSettings/Stores/JSONConfigFileWatcher.swift deleted file mode 100644 index 7d0dfcf39985..000000000000 --- a/Packages/CmuxSettings/Sources/CmuxSettings/Stores/JSONConfigFileWatcher.swift +++ /dev/null @@ -1,140 +0,0 @@ -import Foundation - -/// Watches a single file for change events and exposes them as an -/// `AsyncStream`. -/// -/// Wraps `DispatchSource.makeFileSystemObjectSource` — the kqueue-backed -/// Foundation primitive for file events on macOS — because there is no -/// async-native replacement. The dispatch sources are owned by this actor -/// and stay isolated to it. Consumers see only the actor's ``events`` -/// stream; `DispatchSource` itself is never exposed. -/// -/// Construct one per file you want to watch and inject it into the -/// consumer (e.g. ``JSONConfigStore``). The watcher also observes the -/// parent directory so its stream recovers when the file is created or -/// replaced after the watcher starts — the standard cmux config file -/// frequently does not exist on first launch. -/// -/// **Known race window.** ``init(fileURL:)`` schedules `Task { await -/// self.start() }` to attach the dispatch sources on the actor's -/// executor. Between init returning and that Task running, kqueue is not -/// yet listening, so file events in that window are missed. The window -/// is short (one actor-executor hop) and pragmatic for now; a future -/// revisit may move construction to an `async make()` factory so source -/// arming completes before the caller gets a watcher back. -/// -/// ```swift -/// let watcher = JSONConfigFileWatcher(fileURL: locations.userConfigFile) -/// for await _ in watcher.events { -/// await store.reloadFromDisk() -/// } -/// ``` -public actor JSONConfigFileWatcher { - /// Stream of change events. Yields one element per file-system event - /// affecting the watched file or its parent directory. - public nonisolated let events: AsyncStream - - private let fileURL: URL - // DispatchSource.makeFileSystemObjectSource requires a DispatchQueue. - // No async-native Foundation API exists for kqueue file events; this - // queue is internal isolation only and never exposed. - private let queue: DispatchQueue - private let continuation: AsyncStream.Continuation - // File-descriptor lifetime is owned exclusively by each source's - // `setCancelHandler`, which calls `close(fd)` exactly once when the - // source's cancel completes. We never close fds manually. - private var fileSource: DispatchSourceFileSystemObject? - private var directorySource: DispatchSourceFileSystemObject? - - /// Creates and starts a watcher for ``fileURL``. - /// - /// - Parameter fileURL: The file to watch. The parent directory is - /// also observed so the watcher recovers when the file is created - /// or replaced after the watcher starts. - public init(fileURL: URL) { - self.fileURL = fileURL - self.queue = DispatchQueue(label: "com.cmux.json-config-file-watcher") - let (stream, continuation) = AsyncStream.makeStream() - self.events = stream - self.continuation = continuation - Task { await self.start() } - } - - deinit { - // DispatchSource.cancel() is thread-safe; safe to call from deinit. - fileSource?.cancel() - directorySource?.cancel() - continuation.finish() - } - - /// Stops the watcher and finishes its ``events`` stream. - /// - /// Subsequent dispatch events are ignored. Idempotent. - public func stop() { - fileSource?.cancel() - fileSource = nil - directorySource?.cancel() - directorySource = nil - continuation.finish() - } - - // MARK: - Private - - private func start() { - attachDirectorySource() - tryAttachFileSource() - } - - /// Reacts to a parent-directory change. The file may have just been - /// created or replaced; reattach the file-level source so we observe - /// the new inode, then yield an event so consumers reread. - private func handleDirectoryEvent() { - tryAttachFileSource() - continuation.yield(()) - } - - /// Reacts to a file-level change. Yields; consumers reread. - private func handleFileEvent() { - continuation.yield(()) - } - - private func attachDirectorySource() { - let parentFD = open(fileURL.deletingLastPathComponent().path, O_EVTONLY) - guard parentFD >= 0 else { return } - let source = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: parentFD, - eventMask: [.write, .rename, .delete], - queue: queue - ) - source.setEventHandler { [weak self] in - guard let self else { return } - Task { await self.handleDirectoryEvent() } - } - source.setCancelHandler { close(parentFD) } - source.resume() - directorySource = source - } - - /// Re-attaches the file-level source after a directory event. - /// - /// When we replace ``fileSource``, the previous source's - /// `setCancelHandler` is responsible for closing its own captured fd. - /// We never close fds manually here. - private func tryAttachFileSource() { - let fd = open(fileURL.path, O_EVTONLY) - guard fd >= 0 else { return } - fileSource?.cancel() - let newSource = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: fd, - eventMask: [.write, .delete, .rename, .attrib], - queue: queue - ) - newSource.setEventHandler { [weak self] in - guard let self else { return } - Task { await self.handleFileEvent() } - } - newSource.setCancelHandler { close(fd) } - newSource.resume() - fileSource = newSource - } -} diff --git a/Packages/CmuxSettings/Sources/CmuxSettings/Stores/JSONConfigStore.swift b/Packages/CmuxSettings/Sources/CmuxSettings/Stores/JSONConfigStore.swift index 61cf79133b51..fc6042554849 100644 --- a/Packages/CmuxSettings/Sources/CmuxSettings/Stores/JSONConfigStore.swift +++ b/Packages/CmuxSettings/Sources/CmuxSettings/Stores/JSONConfigStore.swift @@ -1,3 +1,4 @@ +import CmuxFileWatch import Foundation /// Typed read/write/observe access to settings persisted in the cmux JSON config file. @@ -12,7 +13,7 @@ import Foundation /// `JSONSerialization` with sorted, pretty-printed output; comment-preserving /// edits are a follow-up. /// -/// Observation uses a single ``JSONConfigFileWatcher`` owned by the store +/// Observation uses a single ``CmuxFileWatch/FileWatcher`` owned by the store /// and fans out file-change events to per-subscriber `AsyncStream` /// signals. One file event causes exactly one cache invalidation and one /// notification per active subscriber, regardless of how many keys are being @@ -32,7 +33,7 @@ public actor JSONConfigStore { public nonisolated let fileURL: URL private let sanitizer: JSONCSanitizer - private let watcher: JSONConfigFileWatcher + private let watcher: FileWatcher private var cachedRoot: [String: Any] = [:] private var cacheValid = false @@ -52,7 +53,7 @@ public actor JSONConfigStore { public init(fileURL: URL, sanitizer: JSONCSanitizer = JSONCSanitizer()) { self.fileURL = fileURL self.sanitizer = sanitizer - self.watcher = JSONConfigFileWatcher(fileURL: fileURL) + self.watcher = FileWatcher(path: fileURL.path) } deinit { @@ -152,7 +153,7 @@ public actor JSONConfigStore { } /// Spawns the watcher-consumer task on the first subscribe. The task - /// drains ``JSONConfigFileWatcher/events`` and fans out to every + /// drains the ``CmuxFileWatch/FileWatcher`` events and fans out to every /// registered subscriber after invalidating the cache. private func ensureWatcherTask() { guard watcherTask == nil else { return } diff --git a/Sources/CmuxConfig.swift b/Sources/CmuxConfig.swift index b676277d228d..25673538991e 100644 --- a/Sources/CmuxConfig.swift +++ b/Sources/CmuxConfig.swift @@ -1,4 +1,5 @@ import Bonsplit +import CmuxFileWatch import Combine import CryptoKit import Foundation @@ -1964,15 +1965,18 @@ final class CmuxConfigStore: ObservableObject { private var parsedConfigCache: [String: ParsedConfigCacheEntry] = [:] private var lifetimeCancellables = Set() private var trackingCancellables = Set() + // The local config still uses a bespoke DispatchSource watcher because it + // performs search-directory *path re-resolution* (not just reload-on-change). + // The global config and hook files use CmuxFileWatch.FileWatcher. private var localFileWatchSource: DispatchSourceFileSystemObject? private var localFileDescriptor: Int32 = -1 private var localConfigSearchDirectory: String? - private var localHookFileWatchSources: [String: DispatchSourceFileSystemObject] = [:] - private var localHookFileDescriptors: [String: Int32] = [:] + private var hookWatchers: [String: FileWatcher] = [:] + private var hookWatchTasks: [String: Task] = [:] private var localFallbackDirectoryWatchSource: DispatchSourceFileSystemObject? private var localFallbackDirectoryDescriptor: Int32 = -1 - private var globalFileWatchSource: DispatchSourceFileSystemObject? - private var globalFileDescriptor: Int32 = -1 + private var globalWatcher: FileWatcher? + private var globalWatchTask: Task? private let watchQueue = DispatchQueue(label: "com.cmux.config-file-watch") private static let maxReattachAttempts = 5 @@ -2010,17 +2014,15 @@ final class CmuxConfigStore: ObservableObject { if localConfigPath != nil { startLocalFileWatcher() } - startGlobalFileWatcher() + startGlobalWatching() } } deinit { localFileWatchSource?.cancel() - for source in localHookFileWatchSources.values { - source.cancel() - } localFallbackDirectoryWatchSource?.cancel() - globalFileWatchSource?.cancel() + hookWatchTasks.values.forEach { $0.cancel() } + globalWatchTask?.cancel() } // MARK: - Public API @@ -3215,51 +3217,35 @@ final class CmuxConfigStore: ObservableObject { let desiredPaths = Set(paths.filter { path in path != primaryLocalPath && FileManager.default.fileExists(atPath: path) }) - for path in Array(localHookFileWatchSources.keys) where !desiredPaths.contains(path) { + for path in Array(hookWatchers.keys) where !desiredPaths.contains(path) { stopLocalHookFileWatcher(at: path) } - for path in desiredPaths where localHookFileWatchSources[path] == nil { + for path in desiredPaths where hookWatchers[path] == nil { startLocalHookFileWatcher(at: path) } } private func startLocalHookFileWatcher(at path: String) { - let fd = open(path, O_EVTONLY) - guard fd >= 0 else { return } - let source = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: fd, - eventMask: [.write, .delete, .rename, .extend], - queue: watchQueue - ) - source.setEventHandler { [weak self] in - guard let self else { return } - let flags = source.data - DispatchQueue.main.async { - if flags.contains(.delete) || flags.contains(.rename) { - self.stopLocalHookFileWatcher(at: path) - } + let watcher = FileWatcher(path: path) + hookWatchers[path] = watcher + let events = watcher.events + hookWatchTasks[path] = Task { @MainActor [weak self] in + for await _ in events { + guard let self else { break } self.loadAll() } } - source.setCancelHandler { - Darwin.close(fd) - } - source.resume() - localHookFileWatchSources[path] = source - localHookFileDescriptors[path] = fd } private func stopLocalHookFileWatcher(at path: String) { - localHookFileWatchSources.removeValue(forKey: path)?.cancel() - localHookFileDescriptors.removeValue(forKey: path) + hookWatchTasks.removeValue(forKey: path)?.cancel() + hookWatchers.removeValue(forKey: path) } private func stopLocalHookFileWatchers() { - for source in localHookFileWatchSources.values { - source.cancel() - } - localHookFileWatchSources.removeAll() - localHookFileDescriptors.removeAll() + hookWatchTasks.values.forEach { $0.cancel() } + hookWatchTasks.removeAll() + hookWatchers.removeAll() } private func startLocalDirectoryWatcher() { @@ -3364,102 +3350,31 @@ final class CmuxConfigStore: ObservableObject { // MARK: - File watching (global) - private func startGlobalFileWatcher() { - let fd = open(globalConfigPath, O_EVTONLY) - guard fd >= 0 else { - startGlobalDirectoryWatcher() - return - } - globalFileDescriptor = fd - - let source = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: fd, - eventMask: [.write, .delete, .rename, .extend], - queue: watchQueue - ) - - source.setEventHandler { [weak self] in - guard let self else { return } - let flags = source.data - if flags.contains(.delete) || flags.contains(.rename) { - DispatchQueue.main.async { - self.stopGlobalFileWatcher() - self.loadAll() - self.scheduleGlobalReattach(attempt: 1) - } - } else { - DispatchQueue.main.async { - self.loadAll() - } - } - } - - source.setCancelHandler { - Darwin.close(fd) - } - - source.resume() - globalFileWatchSource = source - } - - private func scheduleGlobalReattach(attempt: Int) { - guard attempt <= Self.maxReattachAttempts else { - startGlobalDirectoryWatcher() - return - } - watchQueue.asyncAfter(deadline: .now() + Self.reattachDelay) { [weak self] in - guard let self else { return } - DispatchQueue.main.async { - if FileManager.default.fileExists(atPath: self.globalConfigPath) { - self.loadAll() - self.startGlobalFileWatcher() - } else { - self.scheduleGlobalReattach(attempt: attempt + 1) - } - } - } - } - - private func startGlobalDirectoryWatcher() { + /// Watches the global config via ``CmuxFileWatch/FileWatcher``, which handles + /// inode reattachment and nearest-existing-ancestor recovery internally; each + /// change reloads. Ensures the config directory exists first (the previous + /// directory-watcher created it). + private func startGlobalWatching() { + stopGlobalWatching() let dirPath = (globalConfigPath as NSString).deletingLastPathComponent - let fm = FileManager.default - if !fm.fileExists(atPath: dirPath) { - try? fm.createDirectory(atPath: dirPath, withIntermediateDirectories: true) - } - let fd = open(dirPath, O_EVTONLY) - guard fd >= 0 else { return } - globalFileDescriptor = fd - - let source = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: fd, - eventMask: [.write, .link, .rename], - queue: watchQueue - ) - - source.setEventHandler { [weak self] in - guard let self else { return } - DispatchQueue.main.async { - guard FileManager.default.fileExists(atPath: self.globalConfigPath) else { return } - self.stopGlobalFileWatcher() + if !FileManager.default.fileExists(atPath: dirPath) { + try? FileManager.default.createDirectory(atPath: dirPath, withIntermediateDirectories: true) + } + let watcher = FileWatcher(path: globalConfigPath) + globalWatcher = watcher + let events = watcher.events + globalWatchTask = Task { @MainActor [weak self] in + for await _ in events { + guard let self else { break } self.loadAll() - self.startGlobalFileWatcher() } } - - source.setCancelHandler { - Darwin.close(fd) - } - - source.resume() - globalFileWatchSource = source } - private func stopGlobalFileWatcher() { - if let source = globalFileWatchSource { - source.cancel() - globalFileWatchSource = nil - } - globalFileDescriptor = -1 + private func stopGlobalWatching() { + globalWatchTask?.cancel() + globalWatchTask = nil + globalWatcher = nil } } diff --git a/Sources/FileExplorerStore.swift b/Sources/FileExplorerStore.swift index 2258b7a4bb4b..63852f43c7d5 100644 --- a/Sources/FileExplorerStore.swift +++ b/Sources/FileExplorerStore.swift @@ -1,4 +1,5 @@ import AppKit +import CmuxFileWatch import Combine import Foundation import QuartzCore @@ -620,7 +621,9 @@ final class FileExplorerStore: ObservableObject { var showHiddenFiles: Bool = false /// Watches the root directory for filesystem changes (local only). - private var directoryWatcher: FileExplorerDirectoryWatcher? + private var directoryWatcher: FileWatcher? + private var directoryWatchTask: Task? + private var directoryWatchPath: String? /// Paths that are logically expanded (persisted across provider changes) private(set) var expandedPaths: Set = [] @@ -748,18 +751,34 @@ final class FileExplorerStore: ObservableObject { private func updateDirectoryWatcher() { if provider is LocalFileExplorerProvider, !rootPath.isEmpty { - if directoryWatcher == nil { - directoryWatcher = FileExplorerDirectoryWatcher { [weak self] in - self?.reload() - self?.refreshGitStatus() + guard directoryWatchPath != rootPath || directoryWatcher == nil else { return } + stopDirectoryWatcher() + // Preserve the previous 0.3s coalescing as a leading-edge throttle. + let watcher = FileWatcher(path: rootPath, throttle: .milliseconds(300)) + directoryWatcher = watcher + directoryWatchPath = rootPath + let events = watcher.events + directoryWatchTask = Task { @MainActor [weak self] in + for await _ in events { + guard let self else { break } + self.reload() + self.refreshGitStatus() } } - directoryWatcher?.watch(path: rootPath) } else { - directoryWatcher?.stop() + stopDirectoryWatcher() } } + /// Cancels the directory-watch consumer and drops the watcher; the watcher's + /// deinit cancels its `DispatchSource`s synchronously. + private func stopDirectoryWatcher() { + directoryWatchTask?.cancel() + directoryWatchTask = nil + directoryWatcher = nil + directoryWatchPath = nil + } + private func setProvider(_ newProvider: FileExplorerProvider?, reloadIfAvailable: Bool = true) { #if DEBUG NSLog("[FileExplorer] setProvider: \(type(of: newProvider).self) available=\(newProvider?.isAvailable ?? false)") @@ -1108,69 +1127,7 @@ final class FileExplorerStore: ObservableObject { deinit { cancelRemoteHomeResolution() - } -} - -// MARK: - Directory Watcher - -/// Watches a local directory for filesystem changes and calls back on the main thread. -/// Debounces events to avoid rapid-fire reloads during bulk operations (e.g., git checkout). -final class FileExplorerDirectoryWatcher { - private var fileDescriptor: Int32 = -1 - private var watchSource: DispatchSourceFileSystemObject? - private let watchQueue = DispatchQueue(label: "com.cmux.fileExplorerWatcher", qos: .utility) - private var debounceWorkItem: DispatchWorkItem? - private let onChange: () -> Void - - init(onChange: @escaping () -> Void) { - self.onChange = onChange - } - - func watch(path: String) { - stop() - let fd = open(path, O_EVTONLY) - guard fd >= 0 else { return } - fileDescriptor = fd - - let source = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: fd, - eventMask: [.write, .link, .rename, .delete], - queue: watchQueue - ) - - source.setEventHandler { [weak self] in - self?.scheduleReload() - } - - source.setCancelHandler { - Darwin.close(fd) - } - - source.resume() - watchSource = source - } - - func stop() { - debounceWorkItem?.cancel() - debounceWorkItem = nil - watchSource?.cancel() - watchSource = nil - fileDescriptor = -1 - } - - private func scheduleReload() { - debounceWorkItem?.cancel() - let work = DispatchWorkItem { [weak self] in - DispatchQueue.main.async { - self?.onChange() - } - } - debounceWorkItem = work - watchQueue.asyncAfter(deadline: .now() + 0.3, execute: work) - } - - deinit { - stop() + directoryWatchTask?.cancel() } } diff --git a/Sources/KeyboardShortcutSettingsFileStore.swift b/Sources/KeyboardShortcutSettingsFileStore.swift index 66ac16a86472..06cffde97d02 100644 --- a/Sources/KeyboardShortcutSettingsFileStore.swift +++ b/Sources/KeyboardShortcutSettingsFileStore.swift @@ -1,4 +1,5 @@ import Combine +import CmuxFileWatch import CmuxSocketControl import Foundation import os @@ -62,8 +63,8 @@ final class CmuxSettingsFileStore { private let appearanceEnvironment: AppearanceSettings.LiveApplyEnvironment private let stateLock = NSLock() - private var primaryWatcher: ShortcutSettingsFileWatcher? - private var fallbackWatchers: [ShortcutSettingsFileWatcher] = [] + private var watchers: [FileWatcher] = [] + private var watchTasks: [Task] = [] private var defaultsCancellable: AnyCancellable? private var socketPasswordObserver: NSObjectProtocol? @@ -105,15 +106,13 @@ final class CmuxSettingsFileStore { ) guard startWatching else { return } - primaryWatcher = ShortcutSettingsFileWatcher(path: primaryPath, fileManager: fileManager) { [weak self] in - DispatchQueue.main.async { - self?.reload() - } - } - fallbackWatchers = fallbackPaths.map { fallbackPath in - ShortcutSettingsFileWatcher(path: fallbackPath, fileManager: fileManager) { [weak self] in - DispatchQueue.main.async { - self?.reload() + watchers = ([primaryPath] + fallbackPaths).map { FileWatcher(path: $0) } + watchTasks = watchers.map { watcher in + let events = watcher.events + return Task { @MainActor [weak self] in + for await _ in events { + guard let self else { break } + self.reload() } } } @@ -125,8 +124,9 @@ final class CmuxSettingsFileStore { } deinit { - primaryWatcher?.stop() - fallbackWatchers.forEach { $0.stop() } + watchTasks.forEach { $0.cancel() } + // Dropping the watchers runs each deinit, cancelling its DispatchSources. + watchers.removeAll() defaultsCancellable?.cancel() if let socketPasswordObserver { notificationCenter.removeObserver(socketPasswordObserver) @@ -1925,92 +1925,3 @@ private enum BackupValue: Codable, Equatable { } } } - -private final class ShortcutSettingsFileWatcher { - private let path: String - private let fileManager: FileManager - private let onChange: () -> Void - private let watchQueue = DispatchQueue(label: "com.cmux.shortcut-settings-file-watch") - - private var source: DispatchSourceFileSystemObject? - - init(path: String, fileManager: FileManager = .default, onChange: @escaping () -> Void) { - self.path = path - self.fileManager = fileManager - self.onChange = onChange - start() - } - - func stop() { - source?.cancel() - source = nil - } - - private func start() { - stop() - - if fileManager.fileExists(atPath: path) { - startFileWatcher() - } else { - startDirectoryWatcher() - } - } - - private func startFileWatcher() { - let fd = open(path, O_EVTONLY) - guard fd >= 0 else { - startDirectoryWatcher() - return - } - - let source = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: fd, - eventMask: [.write, .delete, .rename, .extend], - queue: watchQueue - ) - - source.setEventHandler { [weak self] in - guard let self else { return } - let flags = source.data - if flags.contains(.delete) || flags.contains(.rename) { - self.start() - } - self.onChange() - } - - source.setCancelHandler { - close(fd) - } - - self.source = source - source.resume() - } - - private func startDirectoryWatcher() { - let directoryPath = (path as NSString).deletingLastPathComponent - let fd = open(directoryPath, O_EVTONLY) - guard fd >= 0 else { return } - - let source = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: fd, - eventMask: [.write, .rename, .delete], - queue: watchQueue - ) - - source.setEventHandler { [weak self] in - guard let self else { return } - if self.fileManager.fileExists(atPath: self.path) { - self.start() - } else { - self.onChange() - } - } - - source.setCancelHandler { - close(fd) - } - - self.source = source - source.resume() - } -} diff --git a/Sources/Panels/MarkdownPanel.swift b/Sources/Panels/MarkdownPanel.swift index 27f35c2af87c..caa05ba7b197 100644 --- a/Sources/Panels/MarkdownPanel.swift +++ b/Sources/Panels/MarkdownPanel.swift @@ -1,4 +1,5 @@ import AppKit +import CmuxFileWatch import Combine import Foundation @@ -73,11 +74,9 @@ final class MarkdownPanel: Panel, ObservableObject, FilePreviewTextEditingPanel // MARK: - File watching - // nonisolated(unsafe) because deinit is not guaranteed to run on the - // main actor, but DispatchSource.cancel() is thread-safe. - private nonisolated(unsafe) var fileWatchSource: DispatchSourceFileSystemObject? - private nonisolated(unsafe) var directoryWatchSource: DispatchSourceFileSystemObject? - private var directoryWatchPath: String? + // Watches `filePath` (file + ancestor-directory recovery) via CmuxFileWatch. + private var fileWatcher: FileWatcher? + private var fileWatchTask: Task? private var originalTextContent: String = "" private var textEncoding: String.Encoding = .utf8 private var saveGeneration: Int = 0 @@ -85,7 +84,6 @@ final class MarkdownPanel: Panel, ObservableObject, FilePreviewTextEditingPanel private var pendingSearchNeedle: String? private weak var textView: NSTextView? private var isClosed: Bool = false - private let watchQueue = DispatchQueue(label: "com.cmux.markdown-file-watch", qos: .utility) // NotificationCenter token; removal is thread-safe so deinit can drop it. private nonisolated(unsafe) var typographyDefaultsObserver: NSObjectProtocol? // The typography default this viewer is currently tracking. While the panel @@ -116,7 +114,7 @@ final class MarkdownPanel: Panel, ObservableObject, FilePreviewTextEditingPanel self.displayTitle = (filePath as NSString).lastPathComponent loadFileContent() - startFileWatcher() + startWatching() observeTypographyDefaults() } @@ -420,165 +418,33 @@ final class MarkdownPanel: Panel, ObservableObject, FilePreviewTextEditingPanel pendingSearchNeedle = nil } - // MARK: - File watcher via DispatchSource + // MARK: - File watcher - private func startFileWatcher() { - stopFileWatcher() - - let fd = open(filePath, O_EVTONLY) - guard fd >= 0 else { - startDirectoryWatcher() - return - } - - stopDirectoryWatcher() - - let source = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: fd, - eventMask: [.write, .delete, .rename, .extend], - queue: watchQueue - ) - - source.setEventHandler { [weak self] in - guard let self else { return } - let flags = source.data - if flags.contains(.delete) || flags.contains(.rename) { - // File was deleted or renamed. The old file descriptor points to - // a stale inode, so we must always stop and reattach the watcher - // even if the new file is already readable (atomic save case). - DispatchQueue.main.async { - guard !self.isClosed else { return } - self.stopFileWatcher() - self.loadFileContent(replacingDirtyContent: false) - // Reattach to the replacement inode when atomic-save - // already created it; otherwise watch the directory until - // the file comes back. - self.startFileWatcher() - } - } else { - // Content changed — reload. - DispatchQueue.main.async { - guard !self.isClosed else { return } - self.loadFileContent(replacingDirtyContent: false) - } - } - } - - source.setCancelHandler { - Darwin.close(fd) - } - - source.resume() - fileWatchSource = source - } - - private func startDirectoryWatcher() { - for directoryPath in existingDirectoryCandidatesForWatcher() { - if directoryWatchPath == directoryPath, directoryWatchSource != nil { - return - } - - let fd = open(directoryPath, O_EVTONLY) - guard fd >= 0 else { continue } - - stopDirectoryWatcher() - - installDirectoryWatcher(fileDescriptor: fd, directoryPath: directoryPath) - return - } - } - - private func installDirectoryWatcher(fileDescriptor fd: Int32, directoryPath: String) { - let source = DispatchSource.makeFileSystemObjectSource( - fileDescriptor: fd, - eventMask: [.write, .delete, .rename, .extend], - queue: watchQueue - ) - - source.setEventHandler { [weak self] in - guard let self else { return } - let flags = source.data - DispatchQueue.main.async { - guard !self.isClosed else { return } - if flags.contains(.delete) || flags.contains(.rename) { - // The watched directory inode changed. Drop the stale file - // descriptor before reattaching, even if the replacement is - // created at the same path string. - self.stopDirectoryWatcher() - } + /// Watches ``filePath`` for changes via ``CmuxFileWatch/FileWatcher``, which + /// handles inode reattachment and nearest-existing-ancestor recovery + /// internally; each change reloads the content. + private func startWatching() { + stopWatching() + let watcher = FileWatcher(path: filePath) + fileWatcher = watcher + let events = watcher.events + fileWatchTask = Task { @MainActor [weak self] in + for await _ in events { + guard let self, !self.isClosed else { break } self.loadFileContent(replacingDirtyContent: false) - if !self.isFileUnavailable { - self.startFileWatcher() - } else { - // If we were watching an ancestor, a child directory may - // have been recreated. Move the watcher as close to the - // target file as possible. - self.startDirectoryWatcher() - } - } - } - - source.setCancelHandler { - Darwin.close(fd) - } - - source.resume() - directoryWatchSource = source - directoryWatchPath = directoryPath - } - - private func existingDirectoryCandidatesForWatcher() -> [String] { - let fileManager = FileManager.default - var current = (filePath as NSString).deletingLastPathComponent - if current.isEmpty { - current = fileManager.currentDirectoryPath - } - - var candidates: [String] = [] - var seen = Set() - while !current.isEmpty { - let standardized = (current as NSString).standardizingPath - guard seen.insert(standardized).inserted else { break } - - var isDirectory: ObjCBool = false - if fileManager.fileExists(atPath: standardized, isDirectory: &isDirectory), - isDirectory.boolValue { - candidates.append(standardized) } - - let parent = (standardized as NSString).deletingLastPathComponent - if parent == standardized || parent.isEmpty { - break - } - current = parent - } - return candidates - } - - private func stopFileWatcher() { - if let source = fileWatchSource { - source.cancel() - fileWatchSource = nil - } - } - - private func stopDirectoryWatcher() { - if let source = directoryWatchSource { - source.cancel() - directoryWatchSource = nil } - directoryWatchPath = nil } private func stopWatching() { - stopFileWatcher() - stopDirectoryWatcher() + fileWatchTask?.cancel() + fileWatchTask = nil + // Dropping the watcher runs its deinit, cancelling the DispatchSources. + fileWatcher = nil } deinit { - // DispatchSource cancel and removeObserver are safe from any thread. - fileWatchSource?.cancel() - directoryWatchSource?.cancel() + fileWatchTask?.cancel() if let typographyDefaultsObserver { NotificationCenter.default.removeObserver(typographyDefaultsObserver) } diff --git a/Sources/TabManager.swift b/Sources/TabManager.swift index 2d9863e5b6f5..d1f1b0c390dc 100644 --- a/Sources/TabManager.swift +++ b/Sources/TabManager.swift @@ -2,6 +2,7 @@ import AppKit import SwiftUI import Foundation import Bonsplit +import CmuxFileWatch import CmuxProcess import CoreVideo import Combine @@ -15,217 +16,6 @@ typealias Tab = Workspace private let tabManagerLogger = Logger(subsystem: "com.cmuxterm.app", category: "TabManager") -private final class WorkspaceGitMetadataWatcherCallbackBox: @unchecked Sendable { - let handleEvent: @Sendable () -> Void - - init(handleEvent: @escaping @Sendable () -> Void) { - self.handleEvent = handleEvent - } -} - -#if DEBUG -// @unchecked Sendable: every mutable field is guarded by `lock`. This is -// DEBUG-only test scaffolding, touched synchronously from the FSEvents callback, -// the synthetic-event producer thread, and the measuring helper — a synchronous -// compare-and-set guard, not ongoing domain state, so a lock is the right shape. -private final class WorkspaceGitMetadataWatcherRefreshRecorder: @unchecked Sendable { - private let lock = NSLock() - private var startedAt: Date? - private var firstRefreshDelay: TimeInterval? - private var cancelled = false - - /// Starts the delay clock. Idempotent; only the first call takes effect. - /// Called once the watcher exists and the storm is about to begin so the - /// measured delay excludes watcher setup. - func start() { - lock.lock() - defer { lock.unlock() } - if startedAt == nil { startedAt = Date() } - } - - /// Records the first refresh delay relative to `start()`. No-op before the - /// clock is started or after the first refresh has already been recorded. - func recordFirstRefresh() { - lock.lock() - defer { lock.unlock() } - guard firstRefreshDelay == nil, let startedAt else { return } - firstRefreshDelay = Date().timeIntervalSince(startedAt) - } - - /// Signals the synthetic-event producer to stop emitting events. - func cancel() { - lock.lock() - defer { lock.unlock() } - cancelled = true - } - - /// Whether `cancel()` has been called; polled by the producer loop. - var isCancelled: Bool { - lock.lock() - defer { lock.unlock() } - return cancelled - } - - var delay: TimeInterval? { - lock.lock() - defer { lock.unlock() } - return firstRefreshDelay - } -} -#endif - -private let workspaceGitMetadataWatcherContextRetain: CFAllocatorRetainCallBack = { info in - guard let info else { return nil } - _ = Unmanaged.fromOpaque(info).retain() - return info -} - -private let workspaceGitMetadataWatcherContextRelease: CFAllocatorReleaseCallBack = { info in - guard let info else { return } - Unmanaged.fromOpaque(info).release() -} - -private let workspaceGitMetadataWatcherCallback: FSEventStreamCallback = { _, info, _, _, _, _ in - guard let info else { return } - let box = Unmanaged.fromOpaque(info).takeUnretainedValue() - box.handleEvent() -} - -private final class WorkspaceGitMetadataWatcher: @unchecked Sendable { - private static let queueSpecificKey = DispatchSpecificKey() - // A single serial queue is shared by every watcher instance (rather than one - // queue per watcher) to bound thread usage when many workspaces are tracked. - // Contract: because `stop()` synchronizes on this queue, the `onChange` - // closure MUST stay non-blocking — a slow `onChange` on any watcher would - // serialize behind it and stall every other watcher's `stop()` on the main - // actor. The production closure only spawns a `@MainActor` Task and returns. - private static let queue: DispatchQueue = { - let queue = DispatchQueue(label: "com.cmux.workspace-git-metadata-watcher", qos: .utility) - queue.setSpecific(key: queueSpecificKey, value: 1) - return queue - }() - - struct Descriptor: Equatable, Sendable { - let directory: String - let watchedPaths: [String] - } - - let descriptor: Descriptor - - private var callbackBox: WorkspaceGitMetadataWatcherCallbackBox? - private let onChange: @Sendable () -> Void - private var stream: FSEventStreamRef? - private var debounceWorkItem: DispatchWorkItem? - - init?(descriptor: Descriptor, onChange: @escaping @Sendable () -> Void) { - guard !descriptor.watchedPaths.isEmpty else { return nil } - self.descriptor = descriptor - self.onChange = onChange - let callbackBox = WorkspaceGitMetadataWatcherCallbackBox { [weak self] in - self?.scheduleChange() - } - self.callbackBox = callbackBox - - var context = FSEventStreamContext( - version: 0, - info: Unmanaged.passUnretained(callbackBox).toOpaque(), - retain: workspaceGitMetadataWatcherContextRetain, - release: workspaceGitMetadataWatcherContextRelease, - copyDescription: nil - ) - let flags = FSEventStreamCreateFlags( - kFSEventStreamCreateFlagFileEvents - ) - guard let stream = FSEventStreamCreate( - nil, - workspaceGitMetadataWatcherCallback, - &context, - descriptor.watchedPaths as CFArray, - FSEventStreamEventId(kFSEventStreamEventIdSinceNow), - 0.25, - flags - ) else { - return nil - } - self.stream = stream - FSEventStreamSetDispatchQueue(stream, Self.queue) - guard FSEventStreamStart(stream) else { - stop() - return nil - } - } - - func stop() { - // Teardown runs synchronously on the shared queue so it completes before - // the caller continues — critically, before `deinit` returns. An async hop - // could let the watcher deallocate (and its `[weak self]` work no-op) - // before the stream is invalidated, leaking the FSEventStream. The - // `getSpecific` check tears down inline (no `sync`) when already on the - // queue, avoiding a deadlock. - if DispatchQueue.getSpecific(key: Self.queueSpecificKey) != nil { - stopOnQueue() - } else { - Self.queue.sync { - stopOnQueue() - } - } - } - - private func stopOnQueue() { - debounceWorkItem?.cancel() - debounceWorkItem = nil - guard let stream else { return } - FSEventStreamStop(stream) - FSEventStreamInvalidate(stream) - FSEventStreamRelease(stream) - self.stream = nil - } - - private func scheduleChange() { - if DispatchQueue.getSpecific(key: Self.queueSpecificKey) != nil { - scheduleChangeOnQueue() - } else { - Self.queue.async { [weak self] in - self?.scheduleChangeOnQueue() - } - } - } - - // Leading-edge throttle: the first event arms a single flush 0.25s later; - // events arriving while that flush is pending are coalesced into it (the - // `debounceWorkItem == nil` guard makes them no-ops). Combined with the - // stream's 0.25s FSEvents latency, the worst-case delay from a filesystem - // change to `onChange` is ~0.5s. During a sustained storm `onChange` fires at - // most once per ~0.25s window — it intentionally does NOT wait for the repo to - // go quiet. The previous trailing-edge debounce (cancel-and-reschedule on every - // event) fired once after quiet, which produced the allocate-and-cancel churn - // this watcher was hardened against. - private func scheduleChangeOnQueue() { - guard stream != nil, debounceWorkItem == nil else { return } - let workItem = DispatchWorkItem { [weak self] in - self?.flushScheduledChangeOnQueue() - } - debounceWorkItem = workItem - Self.queue.asyncAfter(deadline: .now() + 0.25, execute: workItem) - } - - private func flushScheduledChangeOnQueue() { - debounceWorkItem = nil - guard stream != nil else { return } - onChange() - } - -#if DEBUG - func simulateEventForTesting() { - scheduleChange() - } -#endif - - deinit { - stop() - } -} - enum NewWorkspacePlacement: String, CaseIterable, Identifiable { case top case afterCurrent @@ -1508,7 +1298,8 @@ class TabManager: ObservableObject { private var workspaceGitCleanIndexSignatureByKey: [WorkspaceGitProbeKey: String] = [:] private var workspaceGitCleanIndexContentSignatureByKey: [WorkspaceGitProbeKey: String] = [:] private var workspaceGitHeadSignatureByKey: [WorkspaceGitProbeKey: String] = [:] - private var workspaceGitMetadataWatchersByKey: [WorkspaceGitProbeKey: WorkspaceGitMetadataWatcher] = [:] + private var workspaceGitMetadataWatchersByKey: [WorkspaceGitProbeKey: RecursivePathWatcher] = [:] + private var workspaceGitMetadataWatcherRefreshTasksByKey: [WorkspaceGitProbeKey: Task] = [:] private var workspaceGitMetadataWatcherSourceDirectoryByKey: [WorkspaceGitProbeKey: String] = [:] private var workspaceGitMetadataWatcherDescriptorRequestsByKey: [WorkspaceGitProbeKey: WorkspaceGitMetadataWatcherDescriptorRequest] = [:] private var workspaceGitMetadataWatcherDescriptorGeneration: UInt64 = 0 @@ -1861,12 +1652,12 @@ class TabManager: ObservableObject { workspaceGitMetadataWatcherDescriptorRequestsByKey[key] = request Task { [weak self] in - let descriptor = await Task.detached(priority: .utility) { - Self.workspaceGitMetadataWatcherDescriptor(for: directory) + let watchedPaths = await Task.detached(priority: .utility) { + Self.workspaceGitMetadataWatchedPaths(for: directory) }.value await MainActor.run { [weak self] in self?.applyWorkspaceGitMetadataWatcherDescriptor( - descriptor, + watchedPaths, for: key, request: request ) @@ -1875,7 +1666,7 @@ class TabManager: ObservableObject { } private func applyWorkspaceGitMetadataWatcherDescriptor( - _ descriptor: WorkspaceGitMetadataWatcher.Descriptor?, + _ watchedPaths: [String]?, for key: WorkspaceGitProbeKey, request: WorkspaceGitMetadataWatcherDescriptorRequest ) { @@ -1886,27 +1677,29 @@ class TabManager: ObservableObject { guard sidebarGitMetadataWatchEnabled, workspaceGitTrackedDirectoryByKey[key] == request.directory, - let descriptor else { + let watchedPaths else { stopWorkspaceGitMetadataWatcher(for: key) return } - if workspaceGitMetadataWatchersByKey[key]?.descriptor == descriptor { + if workspaceGitMetadataWatchersByKey[key]?.watchedPaths == watchedPaths { workspaceGitMetadataWatcherSourceDirectoryByKey[key] = request.directory return } stopWorkspaceGitMetadataWatcher(for: key) - workspaceGitMetadataWatchersByKey[key] = WorkspaceGitMetadataWatcher( - descriptor: descriptor - ) { [weak self] in - Task { @MainActor [weak self] in - guard let self else { return } - self.scheduleWorkspaceGitMetadataRefreshIfPossible( - workspaceId: key.workspaceId, - panelId: key.panelId, - reason: "filesystemEvent" - ) + if let watcher = RecursivePathWatcher(paths: watchedPaths) { + workspaceGitMetadataWatchersByKey[key] = watcher + let events = watcher.events + workspaceGitMetadataWatcherRefreshTasksByKey[key] = Task { @MainActor [weak self] in + for await _ in events { + guard let self else { break } + self.scheduleWorkspaceGitMetadataRefreshIfPossible( + workspaceId: key.workspaceId, + panelId: key.panelId, + reason: "filesystemEvent" + ) + } } } workspaceGitMetadataWatcherSourceDirectoryByKey[key] = request.directory @@ -1915,7 +1708,12 @@ class TabManager: ObservableObject { private func stopWorkspaceGitMetadataWatcher(for key: WorkspaceGitProbeKey) { workspaceGitMetadataWatcherDescriptorRequestsByKey.removeValue(forKey: key) workspaceGitMetadataWatcherSourceDirectoryByKey.removeValue(forKey: key) - workspaceGitMetadataWatchersByKey.removeValue(forKey: key)?.stop() + workspaceGitMetadataWatcherRefreshTasksByKey.removeValue(forKey: key)?.cancel() + // Dropping the last reference runs the watcher's deinit synchronously, + // which invalidates the FSEventStream on its shared queue before this + // returns. The consumer task captures the events stream (not the watcher), + // so removal here is the last reference. + workspaceGitMetadataWatchersByKey.removeValue(forKey: key) } private func stopWorkspaceGitMetadataWatchers(workspaceId: UUID) { @@ -1926,9 +1724,12 @@ class TabManager: ObservableObject { } private func stopAllWorkspaceGitMetadataWatchers() { - for watcher in workspaceGitMetadataWatchersByKey.values { - watcher.stop() + for task in workspaceGitMetadataWatcherRefreshTasksByKey.values { + task.cancel() } + workspaceGitMetadataWatcherRefreshTasksByKey.removeAll() + // Dropping the references runs each watcher's deinit synchronously, + // invalidating its FSEventStream. workspaceGitMetadataWatchersByKey.removeAll() workspaceGitMetadataWatcherSourceDirectoryByKey.removeAll() workspaceGitMetadataWatcherDescriptorRequestsByKey.removeAll() @@ -3564,9 +3365,9 @@ class TabManager: ObservableObject { .path } - private nonisolated static func workspaceGitMetadataWatcherDescriptor( + private nonisolated static func workspaceGitMetadataWatchedPaths( for directory: String - ) -> WorkspaceGitMetadataWatcher.Descriptor? { + ) -> [String]? { guard let repository = resolveGitRepository(containing: directory) else { return nil } @@ -3587,10 +3388,7 @@ class TabManager: ObservableObject { watchedPaths.append(normalized) } - return WorkspaceGitMetadataWatcher.Descriptor( - directory: repository.workTreeRoot, - watchedPaths: watchedPaths.sorted() - ) + return watchedPaths.sorted() } private nonisolated static func gitRepositoryMetadataWatchPaths( @@ -5088,63 +4886,7 @@ class TabManager: ObservableObject { #if DEBUG nonisolated static func workspaceGitMetadataWatchedPathsForTesting(directory: String) -> [String] { - workspaceGitMetadataWatcherDescriptor(for: directory)?.watchedPaths ?? [] - } - - nonisolated static func workspaceGitMetadataWatcherFirstRefreshDelayDuringStormForTesting( - eventCount: Int, - eventInterval: TimeInterval, - waitTimeout: TimeInterval - ) -> TimeInterval? { - let semaphore = DispatchSemaphore(value: 0) - let recorder = WorkspaceGitMetadataWatcherRefreshRecorder() - - // Watch a unique, empty temp directory so only simulated events reach the - // watcher. Pointing it at the shared NSTemporaryDirectory() would let - // unrelated temp-dir traffic from other processes or tests trip the watcher - // and skew the measured first-refresh delay. - let temporaryDirectoryURL = FileManager.default.temporaryDirectory - .appendingPathComponent(UUID().uuidString, isDirectory: true) - try? FileManager.default.createDirectory(at: temporaryDirectoryURL, withIntermediateDirectories: true) - defer { try? FileManager.default.removeItem(at: temporaryDirectoryURL) } - - let createdWatcher = WorkspaceGitMetadataWatcher( - descriptor: WorkspaceGitMetadataWatcher.Descriptor( - directory: temporaryDirectoryURL.path, - watchedPaths: [temporaryDirectoryURL.path] - ) - ) { - recorder.recordFirstRefresh() - semaphore.signal() - } - guard let watcher = createdWatcher else { - return nil - } - // Cancel the producer before tearing down the watcher so a timeout doesn't - // leave synthetic events landing on the shared watcher queue after this - // helper returns (which would bleed into later tests). Deferred blocks run - // last-in-first-out, so this runs before the temp-directory cleanup above. - defer { - recorder.cancel() - watcher.stop() - } - - DispatchQueue.global(qos: .utility).async { - // Start the clock only once the storm actually begins; the watcher is - // already constructed here, so setup time isn't counted as delay. - recorder.start() - for _ in 0.. [String] { diff --git a/cmux.xcodeproj/project.pbxproj b/cmux.xcodeproj/project.pbxproj index 5accfc93b1d4..cf45fb7f6a0e 100644 --- a/cmux.xcodeproj/project.pbxproj +++ b/cmux.xcodeproj/project.pbxproj @@ -140,6 +140,8 @@ C0DE46000000000000000001 /* CMUXExtensionHostSupport in Frameworks */ = {isa = PBXBuildFile; productRef = C0DE46000000000000000002 /* CMUXExtensionHostSupport */; }; C0DE45000000000000000001 /* CmuxExtensionKit in Frameworks */ = {isa = PBXBuildFile; productRef = C0DE45000000000000000002 /* CmuxExtensionKit */; }; C0DE49000000000000000001 /* CmuxExtensionSidebarExamples in Frameworks */ = {isa = PBXBuildFile; productRef = C0DE49000000000000000002 /* CmuxExtensionSidebarExamples */; }; + C8000313C8000313C8000313 /* CmuxFileWatch in Frameworks */ = {isa = PBXBuildFile; productRef = C8000312C8000312C8000312 /* CmuxFileWatch */; }; + C8000314C8000314C8000314 /* CmuxFileWatch in Frameworks */ = {isa = PBXBuildFile; productRef = C8000312C8000312C8000312 /* CmuxFileWatch */; }; CF00F00000000000000000A3 /* CmuxFoundation in Frameworks */ = {isa = PBXBuildFile; productRef = CF00F00000000000000000A2 /* CmuxFoundation */; }; C0DE34020000000000000001 /* CmuxHelpCommands.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0DE34020000000000000003 /* CmuxHelpCommands.swift */; }; C0DE34020000000000000002 /* CmuxHelpResource.swift in Sources */ = {isa = PBXBuildFile; fileRef = C0DE34020000000000000004 /* CmuxHelpResource.swift */; }; @@ -1205,6 +1207,7 @@ C0DE46000000000000000001 /* CMUXExtensionHostSupport in Frameworks */, C0DE45000000000000000001 /* CmuxExtensionKit in Frameworks */, C0DE49000000000000000001 /* CmuxExtensionSidebarExamples in Frameworks */, + C8000313C8000313C8000313 /* CmuxFileWatch in Frameworks */, CF00F00000000000000000A3 /* CmuxFoundation in Frameworks */, 3069F1D10000000000000005 /* CMUXPasteboardFidelity in Frameworks */, C8000303C8000303C8000303 /* CmuxProcess in Frameworks */, @@ -1237,6 +1240,7 @@ buildActionMask = 2147483647; files = ( A5B00004A1B2C3D4E5F60718 /* CMUXAgentLaunch in Frameworks */, + C8000314C8000314C8000314 /* CmuxFileWatch in Frameworks */, C8000304C8000304C8000304 /* CmuxProcess in Frameworks */, B900004CA1B2C3D4E5F60719 /* CmuxSocketControl in Frameworks */, B9000024A1B2C3D4E5F60719 /* Sentry in Frameworks */, @@ -1926,6 +1930,7 @@ A5B00002A1B2C3D4E5F60718 /* CMUXAgentLaunch */, A5354305A5354305A5354305 /* CmuxSocketControl */, C8000302C8000302C8000302 /* CmuxProcess */, + C8000312C8000312C8000312 /* CmuxFileWatch */, A500D013A1B2C3D4E5F60718 /* CMUXDebugLog */, CD0CFE5200000000CD0CFE52 /* CmuxSettings */, CD0CFE5500000000CD0CFE55 /* CmuxSettingsUI */, @@ -1955,6 +1960,7 @@ A5B00002A1B2C3D4E5F60718 /* CMUXAgentLaunch */, A5354305A5354305A5354305 /* CmuxSocketControl */, C8000302C8000302C8000302 /* CmuxProcess */, + C8000312C8000312C8000312 /* CmuxFileWatch */, ); productName = cmux; productReference = B9000004A1B2C3D4E5F60719 /* cmux */; @@ -2071,6 +2077,7 @@ A500D012A1B2C3D4E5F60718 /* XCLocalSwiftPackageReference "CMUXDebugLog" */, A5354304A5354304A5354304 /* XCLocalSwiftPackageReference "CmuxSocketControl" */, C8000301C8000301C8000301 /* XCLocalSwiftPackageReference "CmuxProcess" */, + C8000311C8000311C8000311 /* XCLocalSwiftPackageReference "CmuxFileWatch" */, CD0CFE5100000000CD0CFE51 /* XCLocalSwiftPackageReference "CmuxSettings" */, CD0CFE5400000000CD0CFE54 /* XCLocalSwiftPackageReference "CmuxSettingsUI" */, CDFEED0100000000CDFEED01 /* XCLocalSwiftPackageReference "CmuxUpdater" */, @@ -3226,6 +3233,10 @@ isa = XCLocalSwiftPackageReference; relativePath = Packages/CmuxProcess; }; + C8000311C8000311C8000311 /* XCLocalSwiftPackageReference "CmuxFileWatch" */ = { + isa = XCLocalSwiftPackageReference; + relativePath = Packages/CmuxFileWatch; + }; CD0CFE5100000000CD0CFE51 /* XCLocalSwiftPackageReference "CmuxSettings" */ = { isa = XCLocalSwiftPackageReference; relativePath = Packages/CmuxSettings; @@ -3348,6 +3359,11 @@ package = A5354304A5354304A5354304 /* XCLocalSwiftPackageReference "CmuxSocketControl" */; productName = CmuxSocketControl; }; + C8000312C8000312C8000312 /* CmuxFileWatch */ = { + isa = XCSwiftPackageProductDependency; + package = C8000311C8000311C8000311 /* XCLocalSwiftPackageReference "CmuxFileWatch" */; + productName = CmuxFileWatch; + }; C8000302C8000302C8000302 /* CmuxProcess */ = { isa = XCSwiftPackageProductDependency; package = C8000301C8000301C8000301 /* XCLocalSwiftPackageReference "CmuxProcess" */; diff --git a/cmuxTests/WorkspacePullRequestSidebarTests.swift b/cmuxTests/WorkspacePullRequestSidebarTests.swift index 2da47d2cd285..9a8ec94e2bfe 100644 --- a/cmuxTests/WorkspacePullRequestSidebarTests.swift +++ b/cmuxTests/WorkspacePullRequestSidebarTests.swift @@ -1632,28 +1632,12 @@ final class WorkspacePullRequestSidebarTests: XCTestCase { ) } - func testGitMetadataWatcherRefreshesDuringSustainedEventStorm() throws { - // waitTimeout (2.0) sits well above the asserted bound (1.0) so the - // XCTUnwrap and comparison have headroom against CI timing jitter: a - // healthy in-burst refresh fires ~0.25s in and passes comfortably, while a - // "wait for quiet" regression (~1.85s for this 1.6s burst) still returns a - // delay that trips the < 1.0 assertion instead of timing out on the unwrap. - let firstRefreshDelay = TabManager.workspaceGitMetadataWatcherFirstRefreshDelayDuringStormForTesting( - eventCount: 80, - eventInterval: 0.02, - waitTimeout: 2.0 - ) - - let delay = try XCTUnwrap( - firstRefreshDelay, - "A sustained FSEvents burst must not keep canceling and reallocating the pending refresh until the repo becomes quiet." - ) - XCTAssertLessThan( - delay, - 1.0, - "The git metadata watcher should refresh during the burst instead of waiting for a quiet period after every event." - ) - } + // The watcher's leading-edge coalescing (a sustained FSEvents burst yields + // one refresh per window rather than waiting for the repo to go quiet) is now + // verified deterministically in CmuxWorkspaceGit's package tests + // (WorkspaceGitMetadataWatcherTests.burstCoalescesAndThrottleRearms), with an + // injected clock and no real waiting, instead of the previous timing-based + // semaphore storm helper. func testModeOnlyTrackedChangesMarkSidebarDirty() throws { let defaults = UserDefaults.standard