amadan

rastrillo / native Public

Clone 
git clone https://amadan.net/rastrillo/native

Plain git — no account needed to clone.

Download

Download this file

1import Foundation
2import RastrilloNative
3import RastrilloTestSupport
4import XCTest
5
6private func bytes(_ text: String) -> Data { Data(text.utf8) }
7
8/// Answers `.corrupt` for chosen paths, removing them, as the sealed store does.
9private final class CorruptingStore: BlobStore, @unchecked Sendable {
10 let base: MemoryBlobStore
11 let corrupt: Set<String>
12 init(_ base: MemoryBlobStore, corrupt: Set<String>) { self.base = base; self.corrupt = corrupt }
13 func write(_ data: Data, to path: String) throws { try base.write(data, to: path) }
14 func read(_ path: String) throws -> Data? {
15 if corrupt.contains(path) { try base.remove(path); throw BlobStoreError.corrupt(path) }
16 return try base.read(path)
17 }
18 func list(_ directory: String) throws -> [String] { try base.list(directory) }
19 func remove(_ path: String) throws { try base.remove(path) }
20}
21
22/// Answers `.keyLost` on the first read and removes everything, as the
23/// sealed store does when the Keychain has lost the key.
24private final class KeyLosingStore: BlobStore, @unchecked Sendable {
25 let base: MemoryBlobStore
26 init(_ base: MemoryBlobStore) { self.base = base }
27 func write(_ data: Data, to path: String) throws { try base.write(data, to: path) }
28 func read(_ path: String) throws -> Data? {
29 guard try base.read(path) != nil else { return nil }
30 try base.remove("")
31 throw BlobStoreError.keyLost
32 }
33 func list(_ directory: String) throws -> [String] { try base.list(directory) }
34 func remove(_ path: String) throws { try base.remove(path) }
35}
36
37final class OutboxTests: XCTestCase {
38 func testEnqueueIsDurableOnReturnAndSeqResumesAfterTheHighest() async throws {
39 let disk = MemoryBlobStore()
40 let outbox = Outbox(store: disk, account: "a")
41 let first = try await outbox.enqueue(lane: "doc:1", kind: "update", payload: bytes("one"))
42 let second = try await outbox.enqueue(lane: "doc:1", kind: "update", payload: bytes("two"))
43 XCTAssertEqual(disk.contents.keys.sorted(), ["outbox/0000000000000001.entry", "outbox/0000000000000002.entry"])
44
45 let relaunched = Outbox(store: MemoryBlobStore(disk.contents), account: "a")
46 let entries = await relaunched.entries(lane: "doc:1")
47 XCTAssertEqual(entries.map(\.id), [first.id, second.id])
48 XCTAssertEqual(entries.map(\.payload), [bytes("one"), bytes("two")])
49 let third = try await relaunched.enqueue(lane: "doc:1", kind: "update", payload: bytes("three"))
50 XCTAssertEqual(third.seq, 3, "seq is recovered as max + 1")
51 }
52
53 func testLanesAreFIFOAndIndependentAndBatchesAreCutByBytes() async throws {
54 let outbox = Outbox(store: MemoryBlobStore(), account: "a")
55 for index in 0..<4 { try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data(repeating: UInt8(index), count: 100)) }
56 try await outbox.enqueue(lane: "doc:b", kind: "u", payload: bytes("b"))
57 let batch = await outbox.sendable(lane: "doc:a", maxBytes: 250)
58 XCTAssertEqual(batch.map(\.seq), [1, 2], "Cut by bytes, in seq order")
59 let oversized = await outbox.sendable(lane: "doc:a", maxBytes: 10)
60 XCTAssertEqual(oversized.map(\.seq), [1], "A batch always carries at least one entry")
61 let other = await outbox.sendable(lane: "doc:b", maxBytes: nil)
62 XCTAssertEqual(other.map(\.payload), [bytes("b")])
63 }
64
65 func testAHoldStopsItsOwnLaneOnly() async throws {
66 let outbox = Outbox(store: MemoryBlobStore(), account: "a")
67 let first = try await outbox.enqueue(lane: "doc:a", kind: "u", payload: bytes("1"))
68 try await outbox.enqueue(lane: "doc:a", kind: "u", payload: bytes("2"))
69 try await outbox.enqueue(lane: "doc:b", kind: "u", payload: bytes("3"))
70 try await outbox.hold([first.id], Hold(.accessLost))
71 let held = await outbox.sendable(lane: "doc:a", maxBytes: nil)
72 XCTAssertEqual(held, [], "Nothing passes a hold: Docs updates depend on the ones before them")
73 let other = await outbox.sendable(lane: "doc:b", maxBytes: nil)
74 XCTAssertEqual(other.count, 1)
75 let summary = await outbox.summary
76 XCTAssertEqual(summary.lanes["doc:a"]?.held, 1)
77 XCTAssertEqual(summary.lanes["doc:a"]?.pending, 1)
78 XCTAssertEqual(summary.lanes["doc:a"]?.firstHold, .accessLost)
79 try await outbox.release(first.id)
80 let released = await outbox.sendable(lane: "doc:a", maxBytes: nil)
81 XCTAssertEqual(released.count, 2)
82 }
83
84 func testReplaceKeepsTheOldestSeqAndTakesANewID() async throws {
85 let disk = MemoryBlobStore()
86 let outbox = Outbox(store: disk, account: "a")
87 var ids: [UUID] = []
88 for text in ["a", "b", "c"] { ids.append(try await outbox.enqueue(lane: "doc:1", kind: "update", payload: bytes(text)).id) }
89 let after = try await outbox.enqueue(lane: "doc:1", kind: "update", payload: bytes("d"))
90 let merged = try await outbox.replace(Array(ids[0...2]), with: bytes("abc"))
91 XCTAssertEqual(merged.seq, 1)
92 XCTAssertFalse(ids.contains(merged.id), "The merged entry gets a new id")
93 XCTAssertEqual(merged.lane, "doc:1")
94 XCTAssertEqual(merged.kind, "update")
95 let lane = await outbox.entries(lane: "doc:1")
96 XCTAssertEqual(lane.map(\.payload), [bytes("abc"), bytes("d")], "Lane order is unchanged")
97 XCTAssertEqual(lane.last?.id, after.id)
98 XCTAssertEqual(disk.contents.keys.sorted(), ["outbox/0000000000000001.entry", "outbox/0000000000000004.entry"])
99 }
100
101 func testReplaceRefusesInFlightHeldGappedAndCrossLaneRuns() async throws {
102 let outbox = Outbox(store: MemoryBlobStore(), account: "a")
103 var a: [UUID] = []
104 for index in 0..<4 {
105 a.append(try await outbox.enqueue(lane: "doc:a", kind: "u", payload: Data(repeating: 1, count: 10)).id)
106 // Another lane interleaves in seq; that does not break a run of doc:a.
107 try await outbox.enqueue(lane: "doc:b", kind: "u", payload: bytes("b\(index)"))
108 }
109 let b = await outbox.entries(lane: "doc:b").map(\.id)
110 guard b.count == 4 else { return XCTFail("Expected four durable entries in lane b") }
111 let inFlight = try await outbox.beginSend(lane: "doc:a", maxBytes: 15)
112 XCTAssertEqual(inFlight.map(\.id), [a[0]])
113
114 await assertThrows(OutboxError.inFlight) { try await outbox.replace([a[0], a[1]], with: Data()) }
115 await assertThrows(OutboxError.notContiguous) { try await outbox.replace([a[1], a[3]], with: Data()) }
116 await assertThrows(OutboxError.notContiguous) { try await outbox.replace([a[1], b[1]], with: Data()) }
117 try await outbox.hold([a[3]], Hold(.tooLarge))
118 await assertThrows(OutboxError.held) { try await outbox.replace([a[2], a[3]], with: Data()) }
119
120 let merged = try await outbox.replace([a[1], a[2]], with: bytes("merged"))
121 XCTAssertEqual(merged.seq, 3, "A run of doc:a skips over doc:b's seqs")
122 try await outbox.ack([a[0]])
123 let lane = await outbox.entries(lane: "doc:a")
124 XCTAssertEqual(lane.map(\.id), [merged.id, a[3]], "Acking the in-flight head never touches the merged entry")
125 }
126
127 func testACrashBetweenReplaceStepsLeavesDuplicatesNeverAGap() async throws {
128 let base = MemoryBlobStore()
129 let setup = Outbox(store: base, account: "a")
130 var ids: [UUID] = []
131 for text in ["a", "b", "c", "d"] { ids.append(try await setup.enqueue(lane: "doc:1", kind: "u", payload: bytes(text)).id) }
132
133 let faulting = FaultingBlobStore(MemoryBlobStore(base.contents))
134 let outbox = Outbox(store: faulting, account: "a")
135 try await outbox.load()
136 faulting.crash(after: 1) // step 1 (the write) happens; the first remove does not
137 do {
138 try await outbox.replace(Array(ids[0...2]), with: bytes("abc"))
139 XCTFail("Expected the simulated crash")
140 } catch is SimulatedCrash {}
141
142 let relaunched = Outbox(store: faulting.committed, account: "a")
143 let lane = await relaunched.entries(lane: "doc:1")
144 XCTAssertEqual(lane.map(\.payload), [bytes("abc"), bytes("b"), bytes("c"), bytes("d")],
145 "Duplicates only: the merged head, then what it merged, then the rest in order")
146 }
147
148 func testAckDiscardAndPurgeRemoveFiles() async throws {
149 let disk = MemoryBlobStore()
150 let outbox = Outbox(store: disk, account: "a")
151 let one = try await outbox.enqueue(lane: "doc:a", kind: "u", payload: bytes("1"))
152 try await outbox.enqueue(lane: "doc:b", kind: "u", payload: bytes("2"))
153 try await outbox.enqueue(lane: "doc:c", kind: "u", payload: bytes("3"))
154 try await outbox.ack([one.id, UUID()])
155 try await outbox.discard(lane: "doc:b")
156 XCTAssertEqual(disk.contents.count, 1)
157 try await outbox.purge()
158 XCTAssertEqual(disk.contents.count, 0)
159 let summary = await outbox.summary
160 XCTAssertTrue(summary.isEmpty)
161 }
162
163 func testAnUnavailableStoreIsNeverAnEmptyStore() async throws {
164 let base = MemoryBlobStore()
165 try await Outbox(store: base, account: "a").enqueue(lane: "doc:1", kind: "u", payload: bytes("kept"))
166 let faulting = FaultingBlobStore(base)
167 faulting.unavailable = true
168 let outbox = Outbox(store: faulting, account: "a")
169 await assertThrows(BlobStoreError.unavailable) { try await outbox.load() }
170 await assertThrows(BlobStoreError.unavailable) { try await outbox.enqueue(lane: "doc:1", kind: "u", payload: Data()) }
171 let summary = await outbox.summary
172 XCTAssertTrue(summary.unavailable)
173 XCTAssertEqual(base.contents.count, 1, "Nothing was deleted while locked")
174 XCTAssertFalse(faulting.operations.contains { if case .remove = $0 { return true } else { return false } })
175
176 faulting.unavailable = false
177 try await outbox.load()
178 let lane = await outbox.entries(lane: "doc:1")
179 XCTAssertEqual(lane.map(\.payload), [bytes("kept")])
180 let after = await outbox.summary
181 XCTAssertFalse(after.unavailable)
182 }
183
184 func testOneCorruptEntryIsRemovedAloneAndReported() async throws {
185 let base = MemoryBlobStore()
186 let setup = Outbox(store: base, account: "a")
187 for text in ["1", "2", "3"] { try await setup.enqueue(lane: "doc:1", kind: "u", payload: bytes(text)) }
188 let outbox = Outbox(store: CorruptingStore(base, corrupt: ["outbox/0000000000000002.entry"]), account: "a")
189 let lane = await outbox.entries(lane: "doc:1")
190 XCTAssertEqual(lane.map(\.payload), [bytes("1"), bytes("3")])
191 XCTAssertEqual(base.contents.count, 2, "Only the corrupt file went")
192 var summary = await outbox.summary
193 XCTAssertTrue(summary.lostUnsent)
194 await outbox.acknowledgeLoss()
195 summary = await outbox.summary
196 XCTAssertFalse(summary.lostUnsent, "Reported once")
197 }
198
199 func testALostKeyEmptiesTheOutboxAndSaysSoOnce() async throws {
200 let base = MemoryBlobStore()
201 try await Outbox(store: base, account: "a").enqueue(lane: "doc:1", kind: "u", payload: bytes("gone"))
202 let outbox = Outbox(store: KeyLosingStore(base), account: "a")
203 try await outbox.load()
204 let summary = await outbox.summary
205 XCTAssertTrue(summary.isEmpty)
206 XCTAssertTrue(summary.lostUnsent)
207 try await outbox.enqueue(lane: "doc:1", kind: "u", payload: bytes("new"))
208 let lane = await outbox.entries(lane: "doc:1")
209 XCTAssertEqual(lane.map(\.payload), [bytes("new")])
210 }
211
212 func testChangesDeliverTheCurrentSummaryThenEachChange() async throws {
213 let outbox = Outbox(store: MemoryBlobStore(), account: "a")
214 let changes = await outbox.changes
215 var iterator = changes.makeAsyncIterator()
216 let first = await iterator.next()
217 XCTAssertEqual(first?.pending, 0)
218 try await outbox.enqueue(lane: "doc:1", kind: "u", payload: bytes("x"))
219 let second = await iterator.next()
220 XCTAssertEqual(second?.pending, 1)
221 }
222
223 /// Random enqueues, sends, acks, compactions and crashes. After every
224 /// relaunch: every entry whose enqueue returned and was never acked is
225 /// still there, in order, and each lane's content is a
226 /// duplicates-allowed replay of what was enqueued.
227 func testRandomCrashesNeverLoseOrReorderAnEntry() async throws {
228 var generator = SeededGenerator(seed: 0x6b33)
229 for _ in 0..<150 {
230 var disk = MemoryBlobStore()
231 var faulting = FaultingBlobStore(disk)
232 var outbox = Outbox(store: faulting, account: "a")
233 var expected: [String: [String]] = [:] // lane -> symbols not yet acked, in order
234 var acked: Set<String> = []
235 var symbol = 0
236 for _ in 0..<40 {
237 let lane = ["doc:a", "doc:b"][Int.random(in: 0...1, using: &generator)]
238 if Int.random(in: 0..<12, using: &generator) == 0 { faulting.crash(after: Int.random(in: 0...2, using: &generator)) }
239 do {
240 switch Int.random(in: 0..<10, using: &generator) {
241 case 0..<5:
242 symbol += 1
243 let text = "s\(symbol)"
244 try await outbox.enqueue(lane: lane, kind: "u", payload: bytes(text))
245 expected[lane, default: []].append(text)
246 case 5..<7:
247 let batch = try await outbox.beginSend(lane: lane, maxBytes: 8)
248 guard !batch.isEmpty else { continue }
249 // The server has the batch before the ack starts: a crash part
250 // way through the ack loses nothing the server lacks.
251 for entry in batch {
252 let text = String(decoding: entry.payload, as: UTF8.self)
253 for part in text.split(separator: "+") { acked.insert(String(part)) }
254 }
255 expected[lane] = expected[lane, default: []].filter { !acked.contains($0) }
256 try await outbox.ack(batch.map(\.id))
257 case 7..<9:
258 let run = await outbox.sendable(lane: lane, maxBytes: nil)
259 guard run.count >= 2 else { continue }
260 let take = Array(run.prefix(Int.random(in: 2...run.count, using: &generator)))
261 let joined = take.map { String(decoding: $0.payload, as: UTF8.self) }.joined(separator: "+")
262 try await outbox.replace(take.map(\.id), with: bytes(joined))
263 default:
264 let batch = try await outbox.beginSend(lane: lane, maxBytes: nil)
265 await outbox.endSend(batch.map(\.id))
266 }
267 } catch is SimulatedCrash {
268 disk = faulting.committed
269 faulting = FaultingBlobStore(disk)
270 outbox = Outbox(store: faulting, account: "a")
271 }
272 // The invariant, checked against a fresh reading of the disk.
273 let reread = Outbox(store: MemoryBlobStore(faulting.committed.contents), account: "a")
274 for (lane, symbols) in expected {
275 let present = await reread.entries(lane: lane).flatMap {
276 String(decoding: $0.payload, as: UTF8.self).split(separator: "+").map(String.init)
277 }
278 let unacked = present.filter { !acked.contains($0) }
279 var seen: [String] = []
280 for item in unacked where seen.last != item && !seen.contains(item) { seen.append(item) }
281 XCTAssertEqual(seen, symbols, "lane \(lane): lost or reordered an entry")
282 }
283 }
284 }
285 }
286}
287
288struct SeededGenerator: RandomNumberGenerator {
289 private var state: UInt64
290 init(seed: UInt64) { state = seed }
291 mutating func next() -> UInt64 {
292 state &+= 0x9E3779B97F4A7C15
293 var z = state
294 z = (z ^ (z >> 30)) &* 0xBF58476D1CE4E5B9
295 z = (z ^ (z >> 27)) &* 0x94D049BB133111EB
296 return z ^ (z >> 31)
297 }
298}
299
300func assertThrows<E: Error & Equatable>(_ expected: E, file: StaticString = #filePath, line: UInt = #line,
301 _ body: () async throws -> some Any) async {
302 do {
303 _ = try await body()
304 XCTFail("Expected \(expected)", file: file, line: line)
305 } catch let error as E {
306 XCTAssertEqual(error, expected, file: file, line: line)
307 } catch {
308 XCTFail("Expected \(expected), got \(error)", file: file, line: line)
309 }
310}
311