Self-heal indexing so a stalled watcher can't silently stop it (#12)

* Self-heal indexing so a stalled watcher can't silently stop it

New screenshots are indexed only via the live FSEvents->OCR pipeline, and
rescanAll() fires solely from the manual Rescan Now button. When that live
path stalls - a long-lived FSEvent stream on an iCloud Drive folder quietly
stops delivering, or one file hangs OCR forever - indexing dies until the
app is relaunched. It stayed dead for 8 days in the wild.

Two complementary fixes:

- OCR can no longer hang. recognize(at:) now wraps the decode and Vision
  call in a withTimeout watchdog (60s default) that throws OCRError.timedOut,
  so a single bad file can't wedge the drain queue on an await that never
  resumes.

- The indexer runs a maintenance tick every 5 minutes that re-arms the
  FSEvents stream and rescans. It runs independently of the drain queue, so
  even a wedged queue recovers, and a dead watcher gets a fresh stream.

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: GitButler <gitbutler@gitbutler.com>

* Address PR review: make the OCR watchdog actually free the caller

The first cut of withTimeout raced the work against a sleeper inside a
withThrowingTaskGroup. A structured group can't return from its scope until
every child finishes, so when the OCR work hangs in a non-cancellable
Vision/ImageIO call the group hangs with it - the watchdog never fires and
the indexer can still wedge on the exact bad-file case it was meant to fix.
Codex and Copilot both flagged this.

Rework withTimeout to run the work in an unstructured task the caller never
awaits directly, suspending instead on a continuation that whichever of
{work, deadline} settles first resumes via a one-shot gate. A hung work task
is orphaned rather than awaited, so the caller is always freed at the
deadline. Added a regression test with a Thread.sleep operation that ignores
cooperative cancellation.

Also from review:
- Guard periodicMaintenance with isMaintenanceRunning so a scan that outruns
  the 5-minute interval can't stack a second concurrent scan.
- guard let self in the maintenance loop so the task exits if the actor is
  deallocated instead of looping forever.
- Fix a comment typo (worked -> worker task).

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: GitButler <gitbutler@gitbutler.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: GitButler <gitbutler@gitbutler.com>

Author: 3 people

Sources/VistaCore/Indexer.swift

@@ -50,6 +50,22 @@ public actor Indexer {
     private var pendingIndex: Int = 0
     private var workTask: Task<Void, Never>?
     private var eventTask: Task<Void, Never>?
+    private var maintenanceTask: Task<Void, Never>?
+    // Guards against overlapping maintenance ticks. `periodicMaintenance()`
+    // awaits `rescanAll()`, and the actor suspends across that await; if a scan
+    // runs longer than `maintenanceInterval` (large backlog, slow disk/iCloud),
+    // the next tick would otherwise start a second concurrent scan. The flag
+    // makes a still-running scan swallow the next tick instead.
+    private var isMaintenanceRunning = false
+
+    // How often the self-heal loop re-arms the watcher and rescans. New
+    // files normally index instantly via FSEvents; this is the safety net
+    // for when that live path silently stops — most notably long-lived
+    // FSEvent streams on iCloud Drive folders, which have been observed to
+    // stop delivering while the app keeps running. 5 minutes keeps the
+    // worst-case staleness small without meaningful cost (a rescan with a
+    // populated DB is just a fingerprint walk).
+    private static let maintenanceInterval: Duration = .seconds(300)
 
     private var progressContinuation: AsyncStream<Progress>.Continuation?
     // nonisolated because the stream is immutable after init and AsyncStream
@@ -98,16 +114,43 @@ public actor Indexer {
         }
 
         await emitWatchingProgress()
+
+        maintenanceTask = Task { [weak self] in
+            while !Task.isCancelled {
+                try? await Task.sleep(for: Self.maintenanceInterval)
+                // Bail if the actor was deallocated — `self?` would silently
+                // no-op the call while the loop kept sleeping forever, leaking
+                // the task.
+                guard let self, !Task.isCancelled else { return }
+                await self.periodicMaintenance()
+            }
+        }
     }
 
     public func stop() {
         watcher.stop()
         eventTask?.cancel()
         workTask?.cancel()
+        maintenanceTask?.cancel()
         progressContinuation?.finish()
         accessContinuation?.finish()
     }
 
+    /// Self-heal tick: re-arm the FSEvents stream (recovers a stream that
+    /// silently stopped delivering) then rescan to catch anything the live
+    /// path missed. Re-arm first so we're listening for new events before
+    /// the rescan enumerates — anything that lands in the gap is still
+    /// caught by the rescan, so there's no window where a file is lost.
+    /// Skipped while paused, matching the intent of the pause control.
+    private func periodicMaintenance() async {
+        guard !isPaused, !isMaintenanceRunning else { return }
+        isMaintenanceRunning = true
+        defer { isMaintenanceRunning = false }
+        watcher.stop()
+        watcher.start(paths: watchedFolders)
+        await rescanAll()
+    }
+
     public func setPaused(_ paused: Bool) {
         isPaused = paused
         if !paused {

Sources/VistaCore/OCRRecognizer.swift

@@ -34,23 +34,38 @@ public final class OCRRecognizer: Sendable {
         self.languages = languages
     }
 
+    /// Upper bound on a single file's OCR. Vision has been observed to
+    /// accept an image and then never fire its completion handler (nor
+    /// throw), which would await forever; the ImageIO decode can likewise
+    /// stall on a malformed iCloud file. A per-file timeout turns either
+    /// hang into a thrown error the indexer can log and skip, so one bad
+    /// file can't wedge the whole queue. 60s is far above real OCR cost
+    /// (accurate mode on the largest screenshots is a few seconds) — it
+    /// only ever trips on a genuine infinite hang.
+    public static let defaultTimeout: Duration = .seconds(60)
+
     /// Recognises text in the image at `url`. Returns a single joined
     /// string (lines separated by `\n`). Empty when OCR is off or when
     /// Vision found no text.
     ///
-    /// Throws on unreadable files or Vision failures — callers can decide
-    /// whether to log and skip or surface to the user.
-    public func recognize(at url: URL) async throws -> String {
+    /// Throws on unreadable files, Vision failures, or when the work
+    /// exceeds `timeout` — callers can decide whether to log and skip or
+    /// surface to the user.
+    public func recognize(at url: URL, timeout: Duration = OCRRecognizer.defaultTimeout) async throws -> String {
         guard level != .off else { return "" }
 
-        // Load once via ImageIO — avoids pulling the full decoded bitmap
-        // into memory when Vision can stream from the CGImage source.
-        guard let source = CGImageSourceCreateWithURL(url as CFURL, nil),
-              let cgImage = CGImageSourceCreateImageAtIndex(source, 0, nil) else {
-            throw OCRError.unreadableImage(url)
+        // Both the decode and the Vision call run inside the timeout so a
+        // stall in either one is bounded. The decode moves off the caller's
+        // thread into the worker task as a side effect, which is harmless.
+        return try await withTimeout(timeout, onTimeout: OCRError.timedOut(url)) {
+            // Load once via ImageIO — avoids pulling the full decoded bitmap
+            // into memory when Vision can stream from the CGImage source.
+            guard let source = CGImageSourceCreateWithURL(url as CFURL, nil),
+                  let cgImage = CGImageSourceCreateImageAtIndex(source, 0, nil) else {
+                throw OCRError.unreadableImage(url)
+            }
+            return try await self.recognize(cgImage: cgImage)
         }
-
-        return try await recognize(cgImage: cgImage)
     }
 
     /// Recognises text in an already-decoded CGImage. Exposed so tests and
@@ -136,5 +151,65 @@ public final class OCRRecognizer: Sendable {
 
     public enum OCRError: Error, Equatable {
         case unreadableImage(URL)
+        case timedOut(URL)
+    }
+}
+
+/// Runs `operation`, throwing `onTimeout` if it hasn't finished within
+/// `timeout`. The caller is freed the instant either the work or the deadline
+/// settles, even when the work is stuck in a non-cancellable synchronous call
+/// (Vision's `perform`, an ImageIO decode) that ignores cooperative cancellation.
+///
+/// This deliberately does NOT use a `withThrowingTaskGroup`: a structured group
+/// cannot return from its scope until every child task has finished, so if the
+/// operation hangs the group hangs with it and the timeout never frees the
+/// caller — defeating the whole watchdog. Instead the work runs in an
+/// unstructured task that the caller never awaits directly. The caller suspends
+/// on a continuation that whichever of {work, deadline} finishes first resumes
+/// via a one-shot gate; a hung work task is simply orphaned (it finishes in the
+/// background and its late result is discarded). This bounds the *await*, not
+/// necessarily the CPU — the right trade for a watchdog.
+func withTimeout<T: Sendable>(
+    _ timeout: Duration,
+    onTimeout: @autoclosure @escaping @Sendable () -> Error,
+    operation: @escaping @Sendable () async throws -> T
+) async throws -> T {
+    return try await withCheckedThrowingContinuation { continuation in
+        let gate = TimeoutGate(continuation)
+        let deadline = Task {
+            do {
+                try await Task.sleep(for: timeout)
+                gate.resume(with: .failure(onTimeout()))
+            } catch {
+                // Sleep cancelled because the work settled first — nothing to do.
+            }
+        }
+        Task {
+            do { gate.resume(with: .success(try await operation())) }
+            catch { gate.resume(with: .failure(error)) }
+            // Stop the watchdog as soon as the work settles so the common fast
+            // path doesn't leave a task sleeping out the full timeout.
+            deadline.cancel()
+        }
+    }
+}
+
+/// One-shot resume gate for `withTimeout`: only the first of {work, deadline}
+/// resumes the caller; the loser's later attempt is dropped. The NSLock guards
+/// the single-resume invariant that `CheckedContinuation` asserts on. Declared
+/// at file scope because Swift can't nest a type inside a generic function.
+private final class TimeoutGate<T: Sendable>: @unchecked Sendable {
+    private let lock = NSLock()
+    private var resumed = false
+    private let continuation: CheckedContinuation<T, Error>
+
+    init(_ continuation: CheckedContinuation<T, Error>) { self.continuation = continuation }
+
+    func resume(with result: Result<T, Error>) {
+        lock.lock()
+        let firstTime = !resumed
+        resumed = true
+        lock.unlock()
+        if firstTime { continuation.resume(with: result) }
     }
 }

Tests/VistaCoreTests/OCRRecognizerTests.swift

@@ -30,6 +30,64 @@ final class OCRRecognizerTests: XCTestCase {
         XCTAssertEqual(text, "")
     }
 
+    // MARK: - Timeout watchdog
+
+    private struct Boom: Error, Equatable {}
+
+    func testWithTimeoutReturnsFastResult() async throws {
+        let value = try await withTimeout(.seconds(5), onTimeout: Boom()) {
+            "done"
+        }
+        XCTAssertEqual(value, "done")
+    }
+
+    func testWithTimeoutThrowsWhenWorkExceedsDeadline() async {
+        do {
+            _ = try await withTimeout(.milliseconds(20), onTimeout: Boom()) {
+                // Far longer than the deadline; the watchdog should fire first.
+                try await Task.sleep(for: .seconds(10))
+                return "should not reach here"
+            }
+            XCTFail("expected the timeout to throw")
+        } catch {
+            XCTAssertEqual(error as? Boom, Boom())
+        }
+    }
+
+    func testWithTimeoutFreesCallerWhenWorkIgnoresCancellation() async {
+        // The regression the codex/copilot review caught: a structured task
+        // group won't return until every child finishes, so a work task stuck
+        // in a non-cancellable synchronous call would hang the watchdog too.
+        // Here the operation blocks the thread with Thread.sleep (which ignores
+        // cooperative cancellation); the caller must still be freed at the
+        // deadline rather than waiting the full 2s.
+        let start = Date()
+        do {
+            _ = try await withTimeout(.milliseconds(50), onTimeout: Boom()) {
+                Thread.sleep(forTimeInterval: 2)
+                return "should not reach here"
+            }
+            XCTFail("expected the timeout to throw")
+        } catch {
+            XCTAssertEqual(error as? Boom, Boom())
+            XCTAssertLessThan(Date().timeIntervalSince(start), 1.5,
+                              "caller should be freed at the deadline, not after the blocking work")
+        }
+    }
+
+    func testWithTimeoutPropagatesOperationError() async {
+        do {
+            _ = try await withTimeout(.seconds(5), onTimeout: Boom()) {
+                throw OCRRecognizer.OCRError.unreadableImage(URL(fileURLWithPath: "/nope.png"))
+            }
+            XCTFail("expected the operation error to propagate")
+        } catch let error as OCRRecognizer.OCRError {
+            XCTAssertEqual(error, .unreadableImage(URL(fileURLWithPath: "/nope.png")))
+        } catch {
+            XCTFail("unexpected error: \(error)")
+        }
+    }
+
     // MARK: - Fixture rendering
 
     /// Draws black text on a white background. Large font so Vision has