proxy: WatchFolder — size-stability partial-write guard, extension filter, cross-run skip list — 4 tests

Sources/ForgeProxy/WatchFolder.swift

@@ -0,0 +1,101 @@
+import Foundation
+
+/// Tracks files already handed off — shared across watcher restarts.
+public actor ProcessedSkipList {
+    private var processed = Set<String>()
+
+    public init() {}
+
+    public func contains(_ path: String) -> Bool {
+        processed.contains(path)
+    }
+
+    public func mark(_ path: String) {
+        processed.insert(path)
+    }
+}
+
+/// Polling watch folder. Emits file paths once they are:
+/// - matching extension
+/// - size-stable for `stabilityChecks` consecutive polls (partial-write guard)
+/// - not in the skip list
+/// Polling (not inotify): cheap at DIT-cart scale, identical behavior on
+/// network volumes where inotify is unreliable.
+public actor WatchFolder {
+    private let path: String
+    private let extensions: Set<String>
+    private let pollInterval: Duration
+    private let stabilityChecks: Int
+    private let skipList: ProcessedSkipList
+
+    private var sizeHistory: [String: (size: UInt64, stableCount: Int)] = [:]
+    private var task: Task<Void, Never>?
+    private var continuation: AsyncStream<String>.Continuation?
+
+    public init(
+        path: String,
+        extensions: [String],
+        pollInterval: Duration = .seconds(2),
+        stabilityChecks: Int = 2,
+        skipList: ProcessedSkipList = ProcessedSkipList()
+    ) {
+        self.path = path
+        self.extensions = Set(extensions.map { $0.lowercased() })
+        self.pollInterval = pollInterval
+        self.stabilityChecks = stabilityChecks
+        self.skipList = skipList
+    }
+
+    /// Start watching; returns stream of ready file paths.
+    public func start() -> AsyncStream<String> {
+        let stream = AsyncStream<String> { continuation in
+            self.continuation = continuation
+        }
+        task = Task {
+            while !Task.isCancelled {
+                await pollOnce()
+                try? await Task.sleep(for: pollInterval)
+            }
+        }
+        return stream
+    }
+
+    public func stop() {
+        task?.cancel()
+        task = nil
+        continuation?.finish()
+        continuation = nil
+    }
+
+    private func pollOnce() async {
+        let fm = FileManager.default
+        guard let entries = try? fm.contentsOfDirectory(atPath: path) else { return }
+
+        for entry in entries {
+            let ext = (entry as NSString).pathExtension.lowercased()
+            guard extensions.contains(ext) else { continue }
+            let full = path + "/" + entry
+            if await skipList.contains(full) { continue }
+
+            guard let attrs = try? fm.attributesOfItem(atPath: full),
+                  let size = attrs[.size] as? UInt64 else { continue }
+
+            if let prev = sizeHistory[full] {
+                if prev.size == size {
+                    let stable = prev.stableCount + 1
+                    if stable >= stabilityChecks {
+                        await skipList.mark(full)
+                        sizeHistory.removeValue(forKey: full)
+                        continuation?.yield(full)
+                    } else {
+                        sizeHistory[full] = (size, stable)
+                    }
+                } else {
+                    sizeHistory[full] = (size, 0) // still growing
+                }
+            } else {
+                sizeHistory[full] = (size, 0)
+            }
+        }
+    }
+}

Tests/ForgeProxyTests/WatchFolderTests.swift

@@ -0,0 +1,132 @@
+import XCTest
+import Foundation
+@testable import ForgeProxy
+
+final class WatchFolderTests: XCTestCase {
+
+    var dir: String!
+
+    override func setUpWithError() throws {
+        dir = NSTemporaryDirectory() + "forge-watch-\(UUID().uuidString)"
+        try FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true)
+    }
+
+    override func tearDown() {
+        try? FileManager.default.removeItem(atPath: dir)
+    }
+
+    // New stable file enqueued exactly once.
+    func testNewFileEnqueuedOnce() async throws {
+        let watcher = WatchFolder(
+            path: dir,
+            extensions: ["mov"],
+            pollInterval: .milliseconds(30),
+            stabilityChecks: 2)
+        var enqueued = [String]()
+        let stream = await watcher.start()
+
+        try Data("clip".utf8).write(to: URL(fileURLWithPath: dir + "/C001.mov"))
+
+        let collector = Task {
+            for await path in stream {
+                enqueued.append(path)
+            }
+        }
+        try await Task.sleep(for: .milliseconds(400))
+        await watcher.stop()
+        _ = await collector.result
+
+        XCTAssertEqual(enqueued.count, 1)
+        XCTAssertTrue(enqueued[0].hasSuffix("C001.mov"))
+    }
+
+    // Growing file (partial write) not enqueued until size stable.
+    func testGrowingFileWaitsForStability() async throws {
+        let watcher = WatchFolder(
+            path: dir,
+            extensions: ["mov"],
+            pollInterval: .milliseconds(40),
+            stabilityChecks: 2)
+        let path = dir + "/GROW.mov"
+        try Data("start".utf8).write(to: URL(fileURLWithPath: path))
+
+        let stream = await watcher.start()
+        actor Seen {
+            var paths = [String]()
+            var at: [ContinuousClock.Instant] = []
+            func add(_ p: String) {
+                paths.append(p)
+                at.append(.now)
+            }
+        }
+        let seen = Seen()
+        let collector = Task {
+            for await p in stream {
+                await seen.add(p)
+            }
+        }
+
+        // Keep growing for ~200ms.
+        for i in 0..<5 {
+            try await Task.sleep(for: .milliseconds(40))
+            let h = FileHandle(forWritingAtPath: path)!
+            h.seekToEndOfFile()
+            h.write(Data("more\(i)".utf8))
+            try h.close()
+        }
+        let growEnd = ContinuousClock.now
+        try await Task.sleep(for: .milliseconds(400))
+        await watcher.stop()
+        _ = await collector.result
+
+        let paths = await seen.paths
+        let times = await seen.at
+        XCTAssertEqual(paths.count, 1)
+        // Enqueue happened after growth stopped.
+        XCTAssertGreaterThan(times[0], growEnd)
+    }
+
+    // Extension filter.
+    func testExtensionFilter() async throws {
+        let watcher = WatchFolder(path: dir, extensions: ["mov"], pollInterval: .milliseconds(30), stabilityChecks: 1)
+        try Data("x".utf8).write(to: URL(fileURLWithPath: dir + "/note.txt"))
+        try Data("y".utf8).write(to: URL(fileURLWithPath: dir + "/C002.mov"))
+
+        let stream = await watcher.start()
+        var enqueued = [String]()
+        let collector = Task {
+            for await p in stream { enqueued.append(p) }
+        }
+        try await Task.sleep(for: .milliseconds(300))
+        await watcher.stop()
+        _ = await collector.result
+
+        XCTAssertEqual(enqueued.count, 1)
+        XCTAssertTrue(enqueued[0].hasSuffix(".mov"))
+    }
+
+    // Already-processed file skipped across watcher runs via skip list.
+    func testSkipListPersistsAcrossRuns() async throws {
+        try Data("x".utf8).write(to: URL(fileURLWithPath: dir + "/C003.mov"))
+
+        let skipList = ProcessedSkipList()
+        let w1 = WatchFolder(path: dir, extensions: ["mov"], pollInterval: .milliseconds(30), stabilityChecks: 1, skipList: skipList)
+        let s1 = await w1.start()
+        var first = [String]()
+        let c1 = Task { for await p in s1 { first.append(p) } }
+        try await Task.sleep(for: .milliseconds(250))
+        await w1.stop()
+        _ = await c1.result
+        XCTAssertEqual(first.count, 1)
+
+        // Second watcher with same skip list: nothing new.
+        let w2 = WatchFolder(path: dir, extensions: ["mov"], pollInterval: .milliseconds(30), stabilityChecks: 1, skipList: skipList)
+        let s2 = await w2.start()
+        var second = [String]()
+        let c2 = Task { for await p in s2 { second.append(p) } }
+        try await Task.sleep(for: .milliseconds(250))
+        await w2.stop()
+        _ = await c2.result
+        XCTAssertTrue(second.isEmpty)
+    }
+}