| 1 | import Foundation |
| 2 | import Observation |
| 3 | import RastrilloNative |
| 4 | |
| 5 | /// Sends one account's outbox. The app supplies `send`, which turns a batch |
| 6 | /// of one lane into requests; the drainer decides when to call it, and what |
| 7 | /// an error it throws means, so every app reacts the same way: |
| 8 | /// |
| 9 | /// | Thrown | Reaction | |
| 10 | /// |---|---| |
| 11 | /// | `.offline`, path down | stop; wait for a connectivity change, with no timer | |
| 12 | /// | `.offline`, path up (a timeout, a lost response) | back off, retry the same batch | |
| 13 | /// | `.unauthorized` | session ended: stop every lane; never hold or drop | |
| 14 | /// | `.rateLimited(after)` | back off at least `after` | |
| 15 | /// | `.server`, `.transport`, `.invalidResponse` | back off, retry the same batch | |
| 16 | /// | `BlobStoreError.unavailable` | when thrown by send, retry this lane; hold after three consecutive throws | |
| 17 | /// | `.forbidden`, `.notFound`, `.conflict`, `.tooLarge`, `.rejected` | hold the batch rather than spin | |
| 18 | /// |
| 19 | /// A failure to acknowledge or hold after a send is a `.retry`: the server |
| 20 | /// may already have the batch, and each app's server-side idempotency makes |
| 21 | /// sending it again harmless. |
| 22 | /// |
| 23 | /// At most one batch is in flight, across all lanes: the simplest correct |
| 24 | /// ordering against one server. |
| 25 | @MainActor @Observable public final class OutboxDrainer { |
| 26 | public enum Disposition: Sendable, Equatable { |
| 27 | case sent |
| 28 | case retry |
| 29 | case hold(Hold) |
| 30 | } |
| 31 | |
| 32 | /// Where the drainer is. `SyncStatus` reads it. |
| 33 | public enum Phase: Sendable, Equatable { |
| 34 | case idle |
| 35 | case sending |
| 36 | case waitingForNetwork |
| 37 | case backingOff |
| 38 | case sessionEnded |
| 39 | case storeUnavailable |
| 40 | } |
| 41 | |
| 42 | /// What one `drain()` ended on. |
| 43 | public enum Outcome: Sendable, Equatable { |
| 44 | /// Nothing left that can be sent. |
| 45 | case idle |
| 46 | case waitingForNetwork |
| 47 | /// Retry after this delay (a connectivity change may cut it short). |
| 48 | case backingOff(TimeInterval) |
| 49 | /// The server’s Retry-After: neither connectivity nor "Try again" crosses it. |
| 50 | case rateLimited(TimeInterval) |
| 51 | case sessionEnded |
| 52 | case storeUnavailable |
| 53 | } |
| 54 | |
| 55 | /// Returns a disposition per entry. An entry it leaves out was not sent |
| 56 | /// this time (for example, a merge that had to fall back to the first |
| 57 | /// entry): it goes back to sendable with no backoff. |
| 58 | public typealias Send = @MainActor (_ lane: String, _ entries: [OutboxEntry], _ transport: NativeTransport) async throws -> [UUID: Disposition] |
| 59 | |
| 60 | public let outbox: Outbox |
| 61 | public private(set) var phase: Phase = .idle |
| 62 | public private(set) var summary = OutboxSummary() |
| 63 | public private(set) var sessionEnded = false |
| 64 | private let transport: @MainActor () -> NativeTransport? |
| 65 | private let connectivity: NativeConnectivity |
| 66 | private let send: Send |
| 67 | private let maxBatchBytes: Int? |
| 68 | private var backoff: Backoff |
| 69 | @ObservationIgnored private var knownStatuses: [String: SyncStatus] = [:] |
| 70 | private var consecutiveThrows: [String: Int] = [:] |
| 71 | private var draining = false |
| 72 | private var drainWaiters: [CheckedContinuation<Outcome, Never>] = [] |
| 73 | private var worker: Task<Void, Never>? |
| 74 | private var observation: Task<Void, Never>? |
| 75 | private var networkObservation: Task<Void, Never>? |
| 76 | private var timer: Task<Void, Never>? |
| 77 | private var wake: AsyncStream<Void>.Continuation? |
| 78 | private var retryDeadline: TimeInterval? |
| 79 | private var serverDeadline: TimeInterval? |
| 80 | private var now: TimeInterval { ProcessInfo.processInfo.systemUptime } |
| 81 | |
| 82 | public init(outbox: Outbox, transport: @escaping @MainActor () -> NativeTransport?, connectivity: NativeConnectivity, |
| 83 | maxBatchBytes: Int? = nil, backoff: Backoff = Backoff(), send: @escaping Send) { |
| 84 | self.outbox = outbox; self.transport = transport; self.connectivity = connectivity |
| 85 | self.maxBatchBytes = maxBatchBytes; self.backoff = backoff; self.send = send |
| 86 | } |
| 87 | public func start() { |
| 88 | guard worker == nil else { return } |
| 89 | let (events, continuation) = AsyncStream<Void>.makeStream(bufferingPolicy: .bufferingNewest(1)) |
| 90 | wake = continuation |
| 91 | observation = Task { [weak self, outbox] in |
| 92 | for await summary in await outbox.changes { |
| 93 | guard let self else { return } |
| 94 | if self.summary != summary { self.summary = summary; self.kick() } |
| 95 | } |
| 96 | } |
| 97 | let changes = connectivity.changes |
| 98 | networkObservation = Task { [weak self] in |
| 99 | for await _ in changes { |
| 100 | guard let self else { return } |
| 101 | self.backoff.reset(); self.retryDeadline = nil; self.kick() |
| 102 | } |
| 103 | } |
| 104 | worker = Task { [weak self] in |
| 105 | for await _ in events { |
| 106 | guard let self, !Task.isCancelled else { return } |
| 107 | if !self.connectivity.isPathSatisfied { |
| 108 | self.timer?.cancel(); self.timer = nil |
| 109 | self.phase = .waitingForNetwork; continue |
| 110 | } |
| 111 | let deadline = max(self.retryDeadline ?? 0, self.serverDeadline ?? 0) |
| 112 | if deadline > self.now { self.schedule(deadline - self.now); continue } |
| 113 | switch await self.drain() { |
| 114 | case .backingOff(let delay): self.retryDeadline = self.now + delay; self.schedule(delay) |
| 115 | case .rateLimited(let delay): self.schedule(delay) |
| 116 | default: break |
| 117 | } |
| 118 | } |
| 119 | } |
| 120 | kick() |
| 121 | } |
| 122 | private func schedule(_ delay: TimeInterval) { |
| 123 | timer?.cancel() |
| 124 | timer = Task { [weak self] in |
| 125 | do { try await Task.sleep(for: .seconds(delay)) } catch { return } |
| 126 | self?.kick() |
| 127 | } |
| 128 | } |
| 129 | public func stop() { |
| 130 | worker?.cancel(); observation?.cancel(); networkObservation?.cancel(); timer?.cancel(); wake?.finish() |
| 131 | worker = nil; observation = nil; networkObservation = nil; timer = nil; wake = nil |
| 132 | } |
| 133 | public func kick() { wake?.yield() } |
| 134 | public func retryNow() { retryDeadline = nil; backoff.reset(); kick() } |
| 135 | public func sessionRestored() { sessionEnded = false; retryNow() } |
| 136 | private func retry() -> Outcome { phase = .backingOff; return .backingOff(backoff.next()) } |
| 137 | |
| 138 | @discardableResult public func drain() async -> Outcome { |
| 139 | if draining { return await withCheckedContinuation { drainWaiters.append($0) } } |
| 140 | draining = true |
| 141 | let outcome = await drainOnce() |
| 142 | draining = false |
| 143 | let waiters = drainWaiters; drainWaiters.removeAll() |
| 144 | for waiter in waiters { waiter.resume(returning: outcome) } |
| 145 | return outcome |
| 146 | } |
| 147 | private func drainOnce() async -> Outcome { |
| 148 | do { try await outbox.load() } |
| 149 | catch { summary = await outbox.summary; phase = .storeUnavailable; return .storeUnavailable } |
| 150 | summary = await outbox.summary |
| 151 | if sessionEnded { phase = .sessionEnded; return .sessionEnded } |
| 152 | if !connectivity.isPathSatisfied { phase = .waitingForNetwork; return .waitingForNetwork } |
| 153 | if let deadline = serverDeadline, deadline > now { phase = .backingOff; return .rateLimited(deadline - now) } |
| 154 | serverDeadline = nil |
| 155 | var retryLanes: Set<String> = [] |
| 156 | while !Task.isCancelled { |
| 157 | summary = await outbox.summary |
| 158 | var batch: [OutboxEntry] = [] |
| 159 | do { |
| 160 | for lane in summary.lanes.keys.sorted() where !retryLanes.contains(lane) { |
| 161 | batch = try await outbox.beginSend(lane: lane, maxBytes: maxBatchBytes) |
| 162 | if !batch.isEmpty { break } |
| 163 | } |
| 164 | } catch { phase = .storeUnavailable; return .storeUnavailable } |
| 165 | guard let first = batch.first else { |
| 166 | if !retryLanes.isEmpty { return retry() } |
| 167 | phase = .idle; return .idle |
| 168 | } |
| 169 | let ids = batch.map(\.id) |
| 170 | guard let transport = transport() else { |
| 171 | await outbox.endSend(ids, failed: false); sessionEnded = true; phase = .sessionEnded; return .sessionEnded |
| 172 | } |
| 173 | phase = .sending |
| 174 | do { |
| 175 | let dispositions: [UUID: Disposition] |
| 176 | do { |
| 177 | dispositions = try await send(first.lane, batch, transport) |
| 178 | consecutiveThrows[first.lane] = nil |
| 179 | } |
| 180 | catch let error as NativeClientError { |
| 181 | consecutiveThrows[first.lane] = nil |
| 182 | let hold: Hold? |
| 183 | switch error { |
| 184 | case .forbidden, .notFound: hold = Hold(.accessLost) |
| 185 | case .conflict(let data): hold = Hold(.conflict(data)) |
| 186 | case .tooLarge: hold = Hold(.tooLarge) |
| 187 | case .rejected(let status, _): hold = Hold(.rejected(String(status))) |
| 188 | default: hold = nil |
| 189 | } |
| 190 | guard let hold else { throw error } |
| 191 | dispositions = Dictionary(uniqueKeysWithValues: ids.map { ($0, .hold(hold)) }) |
| 192 | } |
| 193 | catch { |
| 194 | guard !Task.isCancelled else { throw error } |
| 195 | let count = (consecutiveThrows[first.lane] ?? 0) + 1 |
| 196 | consecutiveThrows[first.lane] = count |
| 197 | // An app decoder/programming failure belongs to this lane, not the |
| 198 | // account. Three consecutive throws require an explicit resolution. |
| 199 | let disposition: Disposition = count >= 3 ? .hold(Hold(.rejected("send"))) : .retry |
| 200 | dispositions = Dictionary(uniqueKeysWithValues: ids.map { ($0, disposition) }) |
| 201 | if count >= 3 { consecutiveThrows[first.lane] = nil } |
| 202 | } |
| 203 | var mustRetry = false; var progressed = false |
| 204 | for entry in batch { |
| 205 | switch dispositions[entry.id] { |
| 206 | case .sent: try await outbox.ack([entry.id]); progressed = true |
| 207 | case .hold(let hold): try await outbox.hold([entry.id], hold); progressed = true |
| 208 | case .retry: await outbox.endSend([entry.id]); mustRetry = true |
| 209 | case nil: break |
| 210 | } |
| 211 | } |
| 212 | await outbox.endSend(ids, failed: false) |
| 213 | summary = await outbox.summary |
| 214 | if mustRetry || !progressed { retryLanes.insert(first.lane); continue } |
| 215 | backoff.reset(); retryDeadline = nil |
| 216 | } catch { |
| 217 | await outbox.endSend(ids); summary = await outbox.summary |
| 218 | switch error { |
| 219 | case NativeClientError.unauthorized: |
| 220 | sessionEnded = true; phase = .sessionEnded; return .sessionEnded |
| 221 | case NativeClientError.offline where !connectivity.isPathSatisfied: |
| 222 | phase = .waitingForNetwork; return .waitingForNetwork |
| 223 | case NativeClientError.rateLimited(let after): |
| 224 | let delay = backoff.next(retryAfter: after); serverDeadline = now + delay |
| 225 | phase = .backingOff; return .rateLimited(delay) |
| 226 | default: return retry() |
| 227 | } |
| 228 | } |
| 229 | if !connectivity.isPathSatisfied { phase = .waitingForNetwork; return .waitingForNetwork } |
| 230 | } |
| 231 | phase = .idle; return .idle |
| 232 | } |
| 233 | public func status(lane: String? = nil, localFailure: SyncFailure? = nil, |
| 234 | count: (OutboxSummary) -> Int = { $0.pendingLanes }) -> SyncStatus { |
| 235 | let state = lane.map { summary.only($0) } ?? summary |
| 236 | let offline = transport().map { connectivity.isOffline($0.origin) } ?? !connectivity.isPathSatisfied |
| 237 | let key = lane.map { "lane:" + $0 } ?? "account" |
| 238 | if state.unavailable, localFailure == nil, let known = knownStatuses[key] { return known } |
| 239 | let status = SyncStatus(summary: state, sending: state.inFlight > 0, sessionEnded: sessionEnded, |
| 240 | offline: offline, backingOff: phase == .backingOff, localFailure: localFailure, count: count) |
| 241 | // A locked store retains the last observable status. An app's explicit |
| 242 | // enqueue failure still reports input that has not reached durable storage. |
| 243 | if !state.unavailable, localFailure == nil { knownStatuses[key] = status } |
| 244 | return status |
| 245 | } |
| 246 | } |
| 247 | |
amadan