Merge pull request #195 from straff2002/claude/offline-queue-durability

Offline-queue durability: restart recovery, launch flush, bounded drain (BM P1)

Author: straff2002

OpenGlasses/Sources/App/OpenGlassesApp.swift

@@ -792,6 +792,17 @@ class AppState: ObservableObject, AppStateProtocol {
         syncEngine.onConflict = { [weak self] _, reason in
             Task { @MainActor in await self?.speechService.speak("Heads up — \(reason).") }
         }
+        // Launch-time flush (Plan BM P1): a relaunch that comes up already-online produces no
+        // reachability edge, so bind() alone would never sync work captured before the last quit.
+        // OfflineQueue.init already re-armed any inFlight strands. Reclaim space either way.
+        Task { @MainActor [weak self] in
+            guard let self else { return }
+            if self.reachability.isOnline, self.offlineQueue.pendingCount > 0 {
+                await self.syncEngine.flush()      // drains (in pages) and maintains
+            } else {
+                self.syncEngine.runMaintenance()
+            }
+        }
 
         // Register live translation tool with its service reference
         var translationTool = LiveTranslationTool()

OpenGlasses/Sources/Services/Offline/OfflineQueue.swift

@@ -32,6 +32,18 @@ final class OfflineQueue {
         )
         """)
         exec("CREATE INDEX IF NOT EXISTS idx_ops_state ON ops(state)")
+        recoverInFlight()
+    }
+
+    // MARK: - Startup recovery
+
+    /// Re-arm ops stranded `inFlight` by a process killed mid-delivery. Only one process ever owns
+    /// this file, so any `inFlight` row seen at open time is a strand — `pending()` would never
+    /// surface it again, silently losing the work. Runs once on open; returns rows recovered.
+    @discardableResult
+    func recoverInFlight() -> Int {
+        exec("UPDATE ops SET state = 'pending' WHERE state = 'inFlight'")
+        return Int(sqlite3_changes(db))
     }
 
     deinit {
@@ -78,9 +90,11 @@ final class OfflineQueue {
         _ = sqlite3_step(stmt)
     }
 
-    /// Delete delivered (tombstone) ops to reclaim space.
+    /// Delete delivered (tombstone) ops to reclaim DB space. Excludes `photoUpload` — those keep
+    /// their delivered tombstone so `prunePhotoEvidence` can bound the on-disk photo cache and
+    /// evict the actual files; dropping their rows here would orphan the images on disk.
     func purgeDone() {
-        exec("DELETE FROM ops WHERE state = 'done'")
+        exec("DELETE FROM ops WHERE state = 'done' AND kind != 'photoUpload'")
     }
 
     /// Delete a single op by id.

OpenGlasses/Sources/Services/Offline/SyncEngine.swift

@@ -47,6 +47,11 @@ final class SyncEngine: ObservableObject {
     var onConflict: ((QueuedOp, String) -> Void)?
     /// Max delivery attempts before an op is marked `failed`.
     var maxAttempts = 6
+    /// Ops fetched per drain page. A backlog larger than this is drained across loop iterations.
+    var batchSize = 500
+    /// Disk cap for delivered photo evidence; pruned after every flush and on launch so the queue
+    /// can't grow without bound. `nil` disables photo pruning.
+    var photoEvidenceCapBytes: Int? = 256 * 1024 * 1024   // 256 MB
 
     private var reachabilityChange: ((Bool) -> Void)?
 
@@ -65,7 +70,9 @@ final class SyncEngine: ObservableObject {
     }
 
     /// Deliver all pending ops in FIFO order. Returns the number delivered this pass. Re-entrant
-    /// calls are coalesced (a flush already running wins).
+    /// calls are coalesced (a flush already running wins). Loops in `batchSize` pages until the
+    /// whole backlog is drained — a >`batchSize` queue no longer needs multiple reconnects — while
+    /// tracking handled ids so a transient op re-queued this flush isn't reprocessed in a spin.
     @discardableResult
     func flush() async -> Int {
         guard !isFlushing else { return 0 }
@@ -74,32 +81,50 @@ final class SyncEngine: ObservableObject {
 
         var delivered = 0
         var conflicts = 0
-        for op in queue.pending() {
-            queue.mark(op.id, state: .inFlight)
-            switch await sink.deliver(op) {
-            case .done:
-                queue.mark(op.id, state: .done)
-                delivered += 1
-            case .conflict(let reason):
-                queue.mark(op.id, state: .conflict)
-                conflicts += 1
-                onConflict?(op, reason)
-            case .transient(let reason):
-                let attempts = op.attempts + 1
-                if attempts >= maxAttempts {
-                    queue.mark(op.id, state: .failed, attempts: attempts)
-                    NSLog("[SyncEngine] op %@ failed after %d attempts: %@", op.id, attempts, reason)
-                } else {
-                    queue.mark(op.id, state: .pending, attempts: attempts)   // retained for a later flush
+        // Transient failures are re-armed to `pending` only *after* the drain completes. During the
+        // loop they stay `inFlight` (out of the pending set) so `pending()` strictly shrinks each
+        // page and the loop can't spin on a retained op — the whole backlog drains in one flush.
+        var retry: [(id: String, attempts: Int)] = []
+        while true {
+            let batch = queue.pending(limit: batchSize)
+            if batch.isEmpty { break }
+            for op in batch {
+                queue.mark(op.id, state: .inFlight)
+                switch await sink.deliver(op) {
+                case .done:
+                    queue.mark(op.id, state: .done)
+                    delivered += 1
+                case .conflict(let reason):
+                    queue.mark(op.id, state: .conflict)
+                    conflicts += 1
+                    onConflict?(op, reason)
+                case .transient(let reason):
+                    let attempts = op.attempts + 1
+                    if attempts >= maxAttempts {
+                        queue.mark(op.id, state: .failed, attempts: attempts)
+                        NSLog("[SyncEngine] op %@ failed after %d attempts: %@", op.id, attempts, reason)
+                    } else {
+                        retry.append((op.id, attempts))   // re-armed below, once the drain is done
+                    }
+                case .permanent(let reason):
+                    queue.mark(op.id, state: .failed)
+                    NSLog("[SyncEngine] op %@ permanently failed: %@", op.id, reason)
                 }
-            case .permanent(let reason):
-                queue.mark(op.id, state: .failed)
-                NSLog("[SyncEngine] op %@ permanently failed: %@", op.id, reason)
             }
         }
+        for r in retry { queue.mark(r.id, state: .pending, attempts: r.attempts) }
 
         lastSyncedCount = delivered
         lastConflictCount = conflicts
+        runMaintenance()
         return delivered
     }
+
+    /// Reclaim queue space: drop delivered non-photo tombstones and evict oldest delivered photo
+    /// evidence past `photoEvidenceCapBytes`. Runs at the tail of every flush and can be called
+    /// standalone on launch so space is reclaimed even when there's nothing to send.
+    func runMaintenance() {
+        if let cap = photoEvidenceCapBytes { queue.prunePhotoEvidence(maxBytes: cap) }
+        queue.purgeDone()
+    }
 }

OpenGlassesTests/OfflineQueueTests.swift

@@ -8,17 +8,28 @@ import XCTest
 final class OfflineQueueTests: XCTestCase {
 
     private var tempFiles: [URL] = []
+    private var tempAux: [URL] = []
 
     override func tearDown() {
         for url in tempFiles {
             for suffix in ["", "-wal", "-shm"] {
                 try? FileManager.default.removeItem(at: URL(fileURLWithPath: url.path + suffix))
             }
         }
+        for url in tempAux { try? FileManager.default.removeItem(at: url) }
         tempFiles.removeAll()
+        tempAux.removeAll()
         super.tearDown()
     }
 
+    /// Create a throwaway file of `bytes` and return its path (tracked for cleanup).
+    private func makePhotoFile(bytes: Int) -> String {
+        let url = FileManager.default.temporaryDirectory.appendingPathComponent("photo_\(UUID().uuidString).jpg")
+        try? Data(repeating: 0xAB, count: bytes).write(to: url)
+        tempAux.append(url)
+        return url.path
+    }
+
     private func tempPath() -> URL {
         let url = FileManager.default.temporaryDirectory.appendingPathComponent("oq_\(UUID().uuidString).sqlite")
         tempFiles.append(url)
@@ -170,6 +181,92 @@ final class OfflineQueueTests: XCTestCase {
         XCTAssertEqual(q.pendingCount, 0)
     }
 
+    // MARK: - Plan BM P1: durability
+
+    func testInFlightStrandRecoveredOnReopen() {
+        let path = tempPath()
+        let id: String
+        do {
+            let q = OfflineQueue(path: path)
+            let a = op("s", at: 100)
+            id = a.id
+            q.enqueue(a)
+            q.mark(a.id, state: .inFlight)          // killed mid-delivery
+            XCTAssertTrue(q.pending().isEmpty)       // inFlight is invisible to pending()
+        }
+        let reopened = OfflineQueue(path: path)      // startup recovery re-arms it
+        XCTAssertEqual(reopened.pendingCount, 1)
+        XCTAssertEqual(reopened.pending().first?.id, id)
+    }
+
+    func testRecoverInFlightReturnsCount() {
+        let q = OfflineQueue(path: tempPath())
+        let a = op("s", at: 100), b = op("s", at: 200)
+        q.enqueue(a); q.enqueue(b)
+        q.mark(a.id, state: .inFlight)
+        q.mark(b.id, state: .inFlight)
+        XCTAssertEqual(q.recoverInFlight(), 2)
+        XCTAssertEqual(q.pendingCount, 2)
+        XCTAssertEqual(q.recoverInFlight(), 0)       // nothing left to recover
+    }
+
+    func testFlushDrainsBacklogBeyondBatchSize() async {
+        let q = OfflineQueue(path: tempPath())
+        for i in 0..<5 { q.enqueue(op("s", at: TimeInterval(i))) }
+        let engine = SyncEngine(queue: q, sink: LocalSyncSink())
+        engine.batchSize = 2                          // 5 ops, 2 per page → must loop
+        let delivered = await engine.flush()
+        XCTAssertEqual(delivered, 5)
+        XCTAssertEqual(q.pendingCount, 0)
+    }
+
+    func testAllTransientBacklogTerminatesAndRetainsOncePerOp() async {
+        let q = OfflineQueue(path: tempPath())
+        for i in 0..<4 { q.enqueue(op("s", at: TimeInterval(i))) }
+        let sink = FakeSink(); sink.defaultOutcome = .transient(reason: "no signal")
+        let engine = SyncEngine(queue: q, sink: sink)
+        engine.batchSize = 1                          // small page must not spin or skip ops
+        _ = await engine.flush()
+        XCTAssertEqual(q.pendingCount, 4)             // all retained, none lost
+        XCTAssertEqual(Set(q.pending().map(\.attempts)), [1])  // each attempted exactly once
+        XCTAssertEqual(sink.deliveredIds.count, 4)    // every op tried, no spin
+    }
+
+    func testFlushMaintenancePurgesDeliveredNonPhotoTombstones() async {
+        let q = OfflineQueue(path: tempPath())
+        q.enqueue(op("s", at: 100)); q.enqueue(op("s", at: 200))
+        let engine = SyncEngine(queue: q, sink: LocalSyncSink())
+        _ = await engine.flush()                      // delivers → done → maintenance purges
+        XCTAssertTrue(q.all().isEmpty, "delivered log tombstones are reclaimed post-flush")
+    }
+
+    func testPurgeDoneKeepsPhotoTombstones() {
+        let q = OfflineQueue(path: tempPath())
+        let photo = QueuedOp.make(kind: .photoUpload, sessionId: "s", json: ["path": "/tmp/x.jpg"])
+        let log = op("s", at: 100)
+        q.enqueue(photo); q.enqueue(log)
+        q.mark(photo.id, state: .done)
+        q.mark(log.id, state: .done)
+        q.purgeDone()
+        // Photo tombstone survives (so prunePhotoEvidence can still manage its file); log is gone.
+        XCTAssertEqual(q.all().map(\.id), [photo.id])
+    }
+
+    func testFlushPrunesDeliveredPhotoEvidencePastCap() async {
+        let q = OfflineQueue(path: tempPath())
+        let paths = (0..<3).map { _ in makePhotoFile(bytes: 100) }
+        for p in paths {
+            q.enqueue(QueuedOp.make(kind: .photoUpload, sessionId: "s", json: ["path": p]))
+        }
+        let engine = SyncEngine(queue: q, sink: LocalSyncSink())
+        engine.photoEvidenceCapBytes = 150            // 300 bytes on disk, cap 150 → evict 2 oldest
+        _ = await engine.flush()
+
+        let onDisk = paths.filter { FileManager.default.fileExists(atPath: $0) }
+        XCTAssertEqual(onDisk.count, 1, "only the newest delivered photo stays under the cap")
+        XCTAssertEqual(q.all(limit: 500).filter { $0.kind == .photoUpload }.count, 1)
+    }
+
     // MARK: - ConflictResolver
 
     func testConflictResolverDetectsServerAdvance() {