fix(mac): make Canvas file watcher reliable
This commit is contained in:
@@ -1,11 +1,10 @@
|
||||
import Foundation
|
||||
import Darwin
|
||||
import CoreServices
|
||||
|
||||
final class CanvasFileWatcher: @unchecked Sendable {
|
||||
private let url: URL
|
||||
private let queue: DispatchQueue
|
||||
private var source: DispatchSourceFileSystemObject?
|
||||
private var fd: Int32 = -1
|
||||
private var stream: FSEventStreamRef?
|
||||
private var pending = false
|
||||
private let onChange: () -> Void
|
||||
|
||||
@@ -20,42 +19,76 @@ final class CanvasFileWatcher: @unchecked Sendable {
|
||||
}
|
||||
|
||||
func start() {
|
||||
guard self.source == nil else { return }
|
||||
let path = (self.url as NSURL).fileSystemRepresentation
|
||||
let fd = open(path, O_EVTONLY)
|
||||
guard fd >= 0 else { return }
|
||||
self.fd = fd
|
||||
guard self.stream == nil else { return }
|
||||
|
||||
let source = DispatchSource.makeFileSystemObjectSource(
|
||||
fileDescriptor: fd,
|
||||
eventMask: [.write, .delete, .rename, .attrib, .extend, .link, .revoke],
|
||||
queue: self.queue)
|
||||
let retainedSelf = Unmanaged.passRetained(self)
|
||||
var context = FSEventStreamContext(
|
||||
version: 0,
|
||||
info: retainedSelf.toOpaque(),
|
||||
retain: nil,
|
||||
release: { pointer in
|
||||
guard let pointer else { return }
|
||||
Unmanaged<CanvasFileWatcher>.fromOpaque(pointer).release()
|
||||
},
|
||||
copyDescription: nil)
|
||||
|
||||
source.setEventHandler { [weak self] in
|
||||
guard let self else { return }
|
||||
if self.pending { return }
|
||||
self.pending = true
|
||||
self.queue.asyncAfter(deadline: .now() + 0.12) { [weak self] in
|
||||
guard let self else { return }
|
||||
self.pending = false
|
||||
self.onChange()
|
||||
}
|
||||
let paths = [self.url.path] as CFArray
|
||||
let flags = FSEventStreamCreateFlags(
|
||||
kFSEventStreamCreateFlagFileEvents |
|
||||
kFSEventStreamCreateFlagUseCFTypes |
|
||||
kFSEventStreamCreateFlagNoDefer)
|
||||
|
||||
guard let stream = FSEventStreamCreate(
|
||||
kCFAllocatorDefault,
|
||||
Self.callback,
|
||||
&context,
|
||||
paths,
|
||||
FSEventStreamEventId(kFSEventStreamEventIdSinceNow),
|
||||
0.05,
|
||||
flags)
|
||||
else {
|
||||
retainedSelf.release()
|
||||
return
|
||||
}
|
||||
|
||||
source.setCancelHandler { [weak self] in
|
||||
guard let self else { return }
|
||||
if self.fd >= 0 {
|
||||
close(self.fd)
|
||||
self.fd = -1
|
||||
}
|
||||
self.stream = stream
|
||||
FSEventStreamSetDispatchQueue(stream, self.queue)
|
||||
if FSEventStreamStart(stream) == false {
|
||||
self.stream = nil
|
||||
FSEventStreamSetDispatchQueue(stream, nil)
|
||||
FSEventStreamInvalidate(stream)
|
||||
FSEventStreamRelease(stream)
|
||||
}
|
||||
|
||||
self.source = source
|
||||
source.resume()
|
||||
}
|
||||
|
||||
func stop() {
|
||||
self.source?.cancel()
|
||||
self.source = nil
|
||||
guard let stream = self.stream else { return }
|
||||
self.stream = nil
|
||||
FSEventStreamStop(stream)
|
||||
FSEventStreamSetDispatchQueue(stream, nil)
|
||||
FSEventStreamInvalidate(stream)
|
||||
FSEventStreamRelease(stream)
|
||||
}
|
||||
}
|
||||
|
||||
extension CanvasFileWatcher {
|
||||
private static let callback: FSEventStreamCallback = { _, info, numEvents, _, eventFlags, _ in
|
||||
guard let info else { return }
|
||||
let watcher = Unmanaged<CanvasFileWatcher>.fromOpaque(info).takeUnretainedValue()
|
||||
watcher.handleEvents(numEvents: numEvents, eventFlags: eventFlags)
|
||||
}
|
||||
|
||||
private func handleEvents(numEvents: Int, eventFlags: UnsafePointer<FSEventStreamEventFlags>?) {
|
||||
guard numEvents > 0 else { return }
|
||||
guard eventFlags != nil else { return }
|
||||
|
||||
// Coalesce rapid changes (common during builds/atomic saves).
|
||||
if self.pending { return }
|
||||
self.pending = true
|
||||
self.queue.asyncAfter(deadline: .now() + 0.12) { [weak self] in
|
||||
guard let self else { return }
|
||||
self.pending = false
|
||||
self.onChange()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
import Foundation
|
||||
import os
|
||||
import Testing
|
||||
@testable import Clawdis
|
||||
|
||||
@Suite(.serialized) struct CanvasFileWatcherTests {
|
||||
private func makeTempDir() throws -> URL {
|
||||
let base = URL(fileURLWithPath: NSTemporaryDirectory(), isDirectory: true)
|
||||
let dir = base.appendingPathComponent("clawdis-canvaswatch-\(UUID().uuidString)", isDirectory: true)
|
||||
try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true)
|
||||
return dir
|
||||
}
|
||||
|
||||
@Test func detectsInPlaceFileWrites() async throws {
|
||||
let dir = try self.makeTempDir()
|
||||
defer { try? FileManager.default.removeItem(at: dir) }
|
||||
|
||||
let file = dir.appendingPathComponent("index.html")
|
||||
try "hello".write(to: file, atomically: false, encoding: .utf8)
|
||||
|
||||
let fired = OSAllocatedUnfairLock(initialState: false)
|
||||
let waitState = OSAllocatedUnfairLock<(fired: Bool, cont: CheckedContinuation<Void, Never>?)>(
|
||||
initialState: (false, nil))
|
||||
|
||||
func waitForFire(timeoutNs: UInt64) async -> Bool {
|
||||
await withTaskGroup(of: Bool.self) { group in
|
||||
group.addTask {
|
||||
await withCheckedContinuation { cont in
|
||||
let resumeImmediately = waitState.withLock { state in
|
||||
if state.fired { return true }
|
||||
state.cont = cont
|
||||
return false
|
||||
}
|
||||
if resumeImmediately {
|
||||
cont.resume()
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
group.addTask {
|
||||
try? await Task.sleep(nanoseconds: timeoutNs)
|
||||
return false
|
||||
}
|
||||
|
||||
let result = await group.next() ?? false
|
||||
group.cancelAll()
|
||||
return result
|
||||
}
|
||||
}
|
||||
|
||||
let watcher = CanvasFileWatcher(url: dir) {
|
||||
fired.withLock { $0 = true }
|
||||
let cont = waitState.withLock { state in
|
||||
state.fired = true
|
||||
let cont = state.cont
|
||||
state.cont = nil
|
||||
return cont
|
||||
}
|
||||
cont?.resume()
|
||||
}
|
||||
watcher.start()
|
||||
defer { watcher.stop() }
|
||||
|
||||
// Give the stream a moment to start.
|
||||
try await Task.sleep(nanoseconds: 150 * 1_000_000)
|
||||
|
||||
// Modify the file in-place (no rename). This used to be missed when only watching the directory vnode.
|
||||
let handle = try FileHandle(forUpdating: file)
|
||||
try handle.seekToEnd()
|
||||
try handle.write(contentsOf: Data(" world".utf8))
|
||||
try handle.close()
|
||||
|
||||
let ok = await waitForFire(timeoutNs: 2_000_000_000)
|
||||
#expect(ok == true)
|
||||
#expect(fired.withLock { $0 } == true)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user