| 1 | import Foundation |
| 2 | import RastrilloNative |
| 3 | import RastrilloTestSupport |
| 4 | import XCTest |
| 5 | @testable import RastrilloApple |
| 6 | |
| 7 | @MainActor private func settle(until condition: () async -> Bool, seconds: Double = 3) async { |
| 8 | let deadline = Date().addingTimeInterval(seconds) |
| 9 | while !(await condition()) && Date() < deadline { try? await Task.sleep(nanoseconds: 5_000_000) } |
| 10 | } |
| 11 | |
| 12 | /// A send that POSTs each entry's payload, in order, as one request per batch. |
| 13 | @MainActor private func poster(_ calls: @escaping @MainActor (String, [OutboxEntry]) -> Void = { _, _ in }) -> OutboxDrainer.Send { |
| 14 | { lane, entries, transport in |
| 15 | calls(lane, entries) |
| 16 | let body = entries.map { String(decoding: $0.payload, as: UTF8.self) }.joined(separator: ",") |
| 17 | let reply = try await transport.responseDetails("/\(lane)", method: "POST", body: Data(body.utf8)) |
| 18 | try reply.requireSuccess() |
| 19 | return Dictionary(uniqueKeysWithValues: entries.map { ($0.id, .sent) }) |
| 20 | } |
| 21 | } |
| 22 | |
| 23 | @MainActor final class OutboxDrainerTests: XCTestCase { |
| 24 | private func connectivity(_ status: NativePath.Status = .satisfied) async -> (NativeConnectivity, ScriptedPathSource) { |
| 25 | let source = ScriptedPathSource(status) |
| 26 | let connectivity = NativeConnectivity(source: source) |
| 27 | await settle { connectivity.path?.status == status } |
| 28 | return (connectivity, source) |
| 29 | } |
| 30 | |
| 31 | private func outbox(_ lanes: [(String, String)]) async throws -> Outbox { |
| 32 | let outbox = Outbox(store: MemoryBlobStore(), account: "a") |
| 33 | for (lane, text) in lanes { try await outbox.enqueue(lane: lane, kind: "u", payload: Data(text.utf8)) } |
| 34 | return outbox |
| 35 | } |
| 36 | |
| 37 | private func bodies(_ transport: ScriptedTransport) -> [String] { |
| 38 | transport.committed.map { String(decoding: $0.body ?? Data(), as: UTF8.self) } |
| 39 | } |
| 40 | |
| 41 | func testThrowingLaneDoesNotStarveOtherLanesAndEventuallyHolds() async throws { |
| 42 | struct Broken: Error {} |
| 43 | let (connectivity, _) = await connectivity() |
| 44 | let outbox = try await outbox([("a", "broken"), ("b", "good"), ("c", "good")]) |
| 45 | let transport = ScriptedTransport([]) |
| 46 | var calls: [String] = [] |
| 47 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity) { lane, entries, _ in |
| 48 | calls.append(lane) |
| 49 | if lane == "a" { throw Broken() } |
| 50 | return Dictionary(uniqueKeysWithValues: entries.map { ($0.id, .sent) }) |
| 51 | } |
| 52 | await drainer.drain() |
| 53 | XCTAssertEqual(calls, ["a", "b", "c"]) |
| 54 | let remaining = await outbox.summary |
| 55 | XCTAssertEqual(remaining.lanes.count, 1) |
| 56 | await drainer.drain(); await drainer.drain() |
| 57 | let held = await outbox.entries(lane: "a") |
| 58 | XCTAssertEqual(held.first?.hold, Hold(.rejected("send"))) |
| 59 | await drainer.drain() |
| 60 | XCTAssertEqual(calls.filter { $0 == "a" }.count, 3, "Three consecutive throws hold the lane instead of spinning") |
| 61 | } |
| 62 | |
| 63 | func testConcurrentDrainWaitsForTheReservedBatch() async throws { |
| 64 | let (connectivity, _) = await connectivity() |
| 65 | let outbox = try await outbox([("doc:a", "1")]) |
| 66 | let transport = ScriptedTransport([]) |
| 67 | var release: CheckedContinuation<Void, Never>? |
| 68 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity) { _, entries, _ in |
| 69 | await withCheckedContinuation { release = $0 } |
| 70 | return Dictionary(uniqueKeysWithValues: entries.map { ($0.id, .sent) }) |
| 71 | } |
| 72 | let first = Task { await drainer.drain() } |
| 73 | await settle { release != nil } |
| 74 | var finished = false |
| 75 | let second = Task { let outcome = await drainer.drain(); finished = true; return outcome } |
| 76 | try await Task.sleep(for: .milliseconds(30)) |
| 77 | XCTAssertFalse(finished, "A concurrent close-save cannot report idle before acknowledgement") |
| 78 | release?.resume() |
| 79 | let outcomes = await (first.value, second.value) |
| 80 | XCTAssertEqual(outcomes.0, .idle); XCTAssertEqual(outcomes.1, .idle) |
| 81 | let summary = await outbox.summary |
| 82 | XCTAssertTrue(summary.isEmpty) |
| 83 | } |
| 84 | |
| 85 | func testUnavailableStoreKeepsTheLastKnownStatusUntilUnlock() async throws { |
| 86 | let (connectivity, _) = await connectivity(.unsatisfied) |
| 87 | let disk = FaultingBlobStore() |
| 88 | let outbox = Outbox(store: disk, account: "a") |
| 89 | try await outbox.enqueue(lane: "doc:a", kind: "update", payload: Data([1])) |
| 90 | let drainer = OutboxDrainer(outbox: outbox, transport: { ScriptedTransport([]) }, connectivity: connectivity, send: poster()) |
| 91 | await drainer.drain() |
| 92 | XCTAssertEqual(drainer.status(), .offline(pending: 1)) |
| 93 | disk.unavailable = true |
| 94 | // Force an I/O operation even after a successful initial load. |
| 95 | do { try await outbox.enqueue(lane: "doc:a", kind: "update", payload: Data([2])); XCTFail("Locked store must refuse enqueue") } |
| 96 | catch BlobStoreError.unavailable { } |
| 97 | await drainer.drain() |
| 98 | XCTAssertEqual(drainer.status(), .offline(pending: 1), "Unavailable keeps the known status; it is not lost") |
| 99 | XCTAssertEqual(drainer.status(localFailure: .cannotSaveOnDevice), .failed(pending: 1, reason: .cannotSaveOnDevice), "The app still reports undurable input explicitly") |
| 100 | disk.unavailable = false |
| 101 | await drainer.drain() |
| 102 | XCTAssertEqual(drainer.status(), .offline(pending: 1)) |
| 103 | XCTAssertEqual(disk.committed.contents.count, 1) |
| 104 | } |
| 105 | |
| 106 | func testDrainsEachLaneInOrderAndAcknowledges() async throws { |
| 107 | let (connectivity, _) = await connectivity() |
| 108 | let transport = ScriptedTransport([.respond(status: 204), .respond(status: 204)]) |
| 109 | let outbox = try await outbox([("doc:a", "1"), ("doc:b", "x"), ("doc:a", "2")]) |
| 110 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 111 | let outcome = await drainer.drain() |
| 112 | XCTAssertEqual(outcome, .idle) |
| 113 | XCTAssertEqual(transport.committed.map(\.path), ["/doc:a", "/doc:b"]) |
| 114 | XCTAssertEqual(bodies(transport), ["1,2", "x"], "One batch per lane, FIFO") |
| 115 | let summary = await outbox.summary |
| 116 | XCTAssertTrue(summary.isEmpty) |
| 117 | XCTAssertEqual(drainer.status(), .saved) |
| 118 | } |
| 119 | |
| 120 | func testBatchesAreCutByBytes() async throws { |
| 121 | let (connectivity, _) = await connectivity() |
| 122 | let transport = ScriptedTransport([.respond(status: 204), .respond(status: 204)]) |
| 123 | let outbox = try await outbox([("doc:a", "aaaa"), ("doc:a", "bbbb"), ("doc:a", "cccc")]) |
| 124 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, maxBatchBytes: 8, send: poster()) |
| 125 | await drainer.drain() |
| 126 | XCTAssertEqual(bodies(transport), ["aaaa,bbbb", "cccc"]) |
| 127 | } |
| 128 | |
| 129 | func testAPathDownMakesNoAttemptAndWaits() async throws { |
| 130 | let (connectivity, _) = await connectivity(.unsatisfied) |
| 131 | let transport = ScriptedTransport([]) |
| 132 | let outbox = try await outbox([("doc:a", "1")]) |
| 133 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 134 | let outcome = await drainer.drain() |
| 135 | XCTAssertEqual(outcome, .waitingForNetwork) |
| 136 | XCTAssertEqual(transport.calls.count, 0, "An offline phone makes no attempts") |
| 137 | XCTAssertEqual(drainer.status(), .offline(pending: 1)) |
| 138 | } |
| 139 | |
| 140 | func testOfflineMidSendWithThePathDownWaitsForTheNetwork() async throws { |
| 141 | let (connectivity, source) = await connectivity() |
| 142 | let outbox = try await outbox([("doc:a", "1")]) |
| 143 | let transport = ScriptedTransport(responder: { _ in |
| 144 | source.send(.unsatisfied) |
| 145 | return .offline(.notConnectedToInternet) |
| 146 | }) |
| 147 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 148 | await settle { connectivity.isPathSatisfied } |
| 149 | let first = await drainer.drain() |
| 150 | await settle { !connectivity.isPathSatisfied } |
| 151 | // Whether the path update landed before or after the classification, |
| 152 | // the next drain must not try again while the path is down. |
| 153 | let second = await drainer.drain() |
| 154 | XCTAssertEqual(second, .waitingForNetwork, "first was \(first)") |
| 155 | XCTAssertEqual(transport.calls.count, 1) |
| 156 | } |
| 157 | |
| 158 | /// M2: a lost response or timeout while the path stays up backs off and |
| 159 | /// retries the same batch; it does not wait for a path change. |
| 160 | func testALostResponseOnAWorkingPathBacksOffAndResendsTheSameBatch() async throws { |
| 161 | let (connectivity, _) = await connectivity() |
| 162 | let transport = ScriptedTransport([.commitThenLoseResponse(), .respond(status: 204)]) |
| 163 | let outbox = try await outbox([("doc:a", "1"), ("doc:a", "2")]) |
| 164 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 165 | let first = await drainer.drain() |
| 166 | guard case .backingOff(let delay) = first else { return XCTFail("Expected a backoff, got \(first)") } |
| 167 | XCTAssertGreaterThan(delay, 0) |
| 168 | XCTAssertEqual(drainer.phase, .backingOff) |
| 169 | XCTAssertEqual(drainer.status(), .retrying(pending: 1)) |
| 170 | let kept = await outbox.entries(lane: "doc:a") |
| 171 | XCTAssertEqual(kept.count, 2, "Nothing is acknowledged that the drainer did not see acknowledged") |
| 172 | let outcome1 = await drainer.drain() |
| 173 | XCTAssertEqual(outcome1, .idle) |
| 174 | XCTAssertEqual(bodies(transport), ["1,2", "1,2"], "The same batch, sent again") |
| 175 | } |
| 176 | |
| 177 | func testAnEndedSessionStopsEveryLaneAndHoldsNothing() async throws { |
| 178 | let (connectivity, _) = await connectivity() |
| 179 | let transport = ScriptedTransport([.respond(status: 401), .respond(status: 204), .respond(status: 204)]) |
| 180 | let outbox = try await outbox([("doc:a", "1"), ("doc:b", "2")]) |
| 181 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 182 | let outcome2 = await drainer.drain() |
| 183 | XCTAssertEqual(outcome2, .sessionEnded) |
| 184 | XCTAssertEqual(transport.calls.count, 1, "Every lane stops") |
| 185 | let summary = await outbox.summary |
| 186 | XCTAssertEqual(summary.held, 0, "Never hold or drop on 401") |
| 187 | XCTAssertEqual(summary.pending, 2) |
| 188 | XCTAssertEqual(drainer.status(), .needsSignIn(pending: 2)) |
| 189 | let outcome3 = await drainer.drain() |
| 190 | XCTAssertEqual(outcome3, .sessionEnded, "It stays stopped until signed in again") |
| 191 | drainer.sessionRestored() |
| 192 | let outcome4 = await drainer.drain() |
| 193 | XCTAssertEqual(outcome4, .idle) |
| 194 | XCTAssertEqual(transport.calls.count, 3) |
| 195 | } |
| 196 | |
| 197 | func testNoSessionIsAnEndedSession() async throws { |
| 198 | let (connectivity, _) = await connectivity() |
| 199 | let outbox = try await outbox([("doc:a", "1")]) |
| 200 | let drainer = OutboxDrainer(outbox: outbox, transport: { nil }, connectivity: connectivity, send: poster()) |
| 201 | let outcome5 = await drainer.drain() |
| 202 | XCTAssertEqual(outcome5, .sessionEnded) |
| 203 | } |
| 204 | |
| 205 | func testRefusalsHoldTheirLaneAndOtherLanesKeepDraining() async throws { |
| 206 | let (connectivity, _) = await connectivity() |
| 207 | let transport = ScriptedTransport { call in |
| 208 | switch call.path { |
| 209 | case "/doc:gone": return .respond(status: 403) |
| 210 | case "/doc:big": return .respond(status: 413) |
| 211 | case "/doc:bad": return .respond(status: 400) |
| 212 | default: return .respond(status: 204) |
| 213 | } |
| 214 | } |
| 215 | let outbox = try await outbox([("doc:gone", "1"), ("doc:big", "2"), ("doc:bad", "3"), ("doc:ok", "4")]) |
| 216 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 217 | let outcome6 = await drainer.drain() |
| 218 | XCTAssertEqual(outcome6, .idle) |
| 219 | let summary = await outbox.summary |
| 220 | XCTAssertEqual(summary.lanes["doc:gone"]?.firstHold, .accessLost) |
| 221 | XCTAssertEqual(summary.lanes["doc:big"]?.firstHold, .tooLarge) |
| 222 | XCTAssertEqual(summary.lanes["doc:bad"]?.firstHold, .rejected("400")) |
| 223 | XCTAssertNil(summary.lanes["doc:ok"], "Other lanes are not stopped by a hold") |
| 224 | XCTAssertEqual(transport.calls.count, 4, "A held lane is not retried") |
| 225 | XCTAssertEqual(drainer.status(lane: "doc:gone"), .failed(pending: 1, reason: .held(.accessLost))) |
| 226 | XCTAssertEqual(drainer.status(lane: "doc:ok"), .saved) |
| 227 | } |
| 228 | |
| 229 | func testRateLimitingIsAFloorOnTheDelay() async throws { |
| 230 | let (connectivity, _) = await connectivity() |
| 231 | let transport = ScriptedTransport([.respond(status: 429, headers: ["retry-after": "30"])]) |
| 232 | let outbox = try await outbox([("doc:a", "1")]) |
| 233 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 234 | let outcome = await drainer.drain() |
| 235 | guard case .rateLimited(let delay) = outcome else { return XCTFail("Expected rate limiting, got \(outcome)") } |
| 236 | XCTAssertGreaterThanOrEqual(delay, 30) |
| 237 | } |
| 238 | |
| 239 | func testServerErrorsBackOff() async throws { |
| 240 | let (connectivity, _) = await connectivity() |
| 241 | let transport = ScriptedTransport([.respond(status: 502)]) |
| 242 | let outbox = try await outbox([("doc:a", "1")]) |
| 243 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 244 | let backedOff = await drainer.drain() |
| 245 | guard case .backingOff = backedOff else { return XCTFail("Expected a backoff") } |
| 246 | let summary = await outbox.summary |
| 247 | XCTAssertEqual(summary.pending, 1) |
| 248 | XCTAssertEqual(summary.held, 0) |
| 249 | } |
| 250 | |
| 251 | /// H4: the send went through but the store locked before the ack. The |
| 252 | /// batch stays queued and goes again; it is never lost. |
| 253 | func testALockedStoreDuringTheAckIsARetry() async throws { |
| 254 | let (connectivity, _) = await connectivity() |
| 255 | let disk = FaultingBlobStore() |
| 256 | let outbox = Outbox(store: disk, account: "a") |
| 257 | try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data("1".utf8)) |
| 258 | let transport = ScriptedTransport([.respond(status: 204), .respond(status: 204)]) |
| 259 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, |
| 260 | send: poster { _, _ in disk.unavailable = true }) |
| 261 | let backedOff = await drainer.drain() |
| 262 | guard case .backingOff = backedOff else { return XCTFail("Expected a retry") } |
| 263 | disk.unavailable = false |
| 264 | let kept = await outbox.entries(lane: "doc:a") |
| 265 | XCTAssertEqual(kept.count, 1) |
| 266 | let again = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 267 | let outcome7 = await again.drain() |
| 268 | XCTAssertEqual(outcome7, .idle) |
| 269 | XCTAssertEqual(transport.committed.count, 2) |
| 270 | } |
| 271 | |
| 272 | func testAnUnavailableStoreWaitsAndDeletesNothing() async throws { |
| 273 | let (connectivity, _) = await connectivity() |
| 274 | let disk = FaultingBlobStore() |
| 275 | let setup = Outbox(store: disk, account: "a") |
| 276 | try await setup.enqueue(lane: "doc:a", kind: "u", payload: Data("1".utf8)) |
| 277 | disk.unavailable = true |
| 278 | let outbox = Outbox(store: disk, account: "a") |
| 279 | let transport = ScriptedTransport([]) |
| 280 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 281 | let outcome = await drainer.drain() |
| 282 | XCTAssertTrue(outcome == .storeUnavailable || outcome == .idle, "\(outcome)") |
| 283 | XCTAssertEqual(transport.calls.count, 0) |
| 284 | disk.unavailable = false |
| 285 | XCTAssertEqual(disk.committed.contents.count, 1) |
| 286 | } |
| 287 | |
| 288 | func testAppDispositionsRetryHoldAndSkip() async throws { |
| 289 | let (connectivity, _) = await connectivity() |
| 290 | let outbox = try await outbox([("doc:a", "1"), ("doc:a", "2"), ("doc:a", "3")]) |
| 291 | let transport = ScriptedTransport([]) |
| 292 | var round = 0 |
| 293 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity) { _, entries, _ in |
| 294 | round += 1 |
| 295 | // Round 1: only the first could be sent (a merge fallback); the rest |
| 296 | // are left out and go back without penalty. |
| 297 | if round == 1 { return [entries[0].id: .sent] } |
| 298 | return [entries[0].id: .hold(Hold(.rejected("no")))] |
| 299 | } |
| 300 | let outcome8 = await drainer.drain() |
| 301 | XCTAssertEqual(outcome8, .idle) |
| 302 | XCTAssertEqual(round, 2) |
| 303 | let lane = await outbox.entries(lane: "doc:a") |
| 304 | XCTAssertEqual(lane.map { String(decoding: $0.payload, as: UTF8.self) }, ["2", "3"]) |
| 305 | XCTAssertEqual(lane.first?.hold, Hold(.rejected("no"))) |
| 306 | XCTAssertEqual(lane.last?.attempts, 0, "A left-out entry is not a failed attempt") |
| 307 | } |
| 308 | |
| 309 | /// The background loop: an enqueue kicks it, and a restored path wakes it |
| 310 | /// with no timer running in between. |
| 311 | func testTheLoopSendsOnKickAndOnReconnect() async throws { |
| 312 | let (connectivity, source) = await connectivity() |
| 313 | let transport = ScriptedTransport { _ in .respond(status: 204) } |
| 314 | let outbox = Outbox(store: MemoryBlobStore(), account: "a") |
| 315 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity, send: poster()) |
| 316 | drainer.start() |
| 317 | defer { drainer.stop() } |
| 318 | try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data("1".utf8)) |
| 319 | drainer.kick() |
| 320 | await settle { transport.committed.count == 1 } |
| 321 | XCTAssertEqual(transport.committed.count, 1) |
| 322 | |
| 323 | source.send(.unsatisfied) |
| 324 | await settle { !connectivity.isPathSatisfied } |
| 325 | try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data("2".utf8)) |
| 326 | drainer.kick() |
| 327 | try await Task.sleep(nanoseconds: 100_000_000) |
| 328 | XCTAssertEqual(transport.calls.count, 1, "Nothing is attempted while the path is down") |
| 329 | await settle { drainer.status() == .offline(pending: 1) } |
| 330 | XCTAssertEqual(drainer.status(), .offline(pending: 1)) |
| 331 | source.send(.satisfied) |
| 332 | await settle { transport.committed.count == 2 } |
| 333 | XCTAssertEqual(bodies(transport), ["1", "2"]) |
| 334 | await settle { drainer.summary.isEmpty } |
| 335 | XCTAssertEqual(drainer.status(), .saved) |
| 336 | } |
| 337 | func testAnAppRetryDoesNotStarveOtherLanes() async throws { |
| 338 | let (connectivity, _) = await connectivity() |
| 339 | let outbox = try await outbox([("doc:a-draft", "1"), ("doc:z-ready", "2")]) |
| 340 | let transport = ScriptedTransport([]) |
| 341 | var sent: [String] = [] |
| 342 | let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity) { lane, entries, _ in |
| 343 | sent.append(lane) |
| 344 | return Dictionary(uniqueKeysWithValues: entries.map { ($0.id, lane == "doc:a-draft" ? .retry : .sent) }) |
| 345 | } |
| 346 | let result = await drainer.drain() |
| 347 | guard case .backingOff = result else { return XCTFail("The draft still needs a retry") } |
| 348 | XCTAssertEqual(sent, ["doc:a-draft", "doc:z-ready"]) |
| 349 | let remaining = await outbox.entries(lane: "doc:z-ready") |
| 350 | XCTAssertTrue(remaining.isEmpty, "An unattached draft must not block another document") |
| 351 | } |
| 352 | |
| 353 | } |
| 354 | |
amadan