Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions Packages/CmuxFileWatch/Package.swift
Original file line number Diff line number Diff line change
@@ -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"),
]
),
]
)
67 changes: 67 additions & 0 deletions Packages/CmuxFileWatch/README.md
Original file line number Diff line number Diff line change
@@ -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<Void>`.
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()
```
Comment thread
coderabbitai[bot] marked this conversation as resolved.

`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<Void, Never>] = []
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.
Original file line number Diff line number Diff line change
@@ -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<UInt8>()
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<FileSystemEventStream>.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()
}
}
37 changes: 37 additions & 0 deletions Packages/CmuxFileWatch/Sources/CmuxFileWatch/FileWatchClock.swift
Original file line number Diff line number Diff line change
@@ -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)
}
}
Loading
Loading