1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155 | import Foundation
import Testing
@testable import IrdP2P
@testable import IrdP2PSignaling
import IrdP2PCrypto
@Suite struct ChunkerTests {
static func payload(_ n: Int, seed: Int = 0) -> Data { Data((0..<n).map { UInt8(($0 &* 31 &+ seed) % 251) }) }
/// Unwraps `.chunk` inner messages.
static func parts(_ messages: [InnerMessage]) -> [(id: String, index: Int, count: Int, encoding: ChunkEncoding, data: Data)] {
messages.compactMap { if case .chunk(let id, let i, let n, let enc, let d) = $0 { return (id, i, n, enc, d) } else { return nil } }
}
@Test func chunkIdIsDerivedFromThePayload() throws {
let data = Self.payload(20_000)
let first = Self.parts(Chunker.split(data, encoding: .bin, sliceSize: 8_192))
let second = Self.parts(Chunker.split(data, encoding: .bin, sliceSize: 8_192))
#expect(first.count == 3 && Set(first.map(\.id)).count == 1)
#expect(first.map(\.id) == second.map(\.id)) // same payload ⇒ same id
// the documented formula: b64url(SHA-256(UTF8(enc) ‖ UInt32BE(n) ‖ data)[0..<8])
let expected = Bytes.base64URL(Bytes.sha256(Data("bin".utf8) + Data([0, 0, 0, 3]) + data).prefix(8))
#expect(first[0].id == expected)
#expect(Bytes.fromBase64URL(expected)?.count == 8)
// anything that changes the assembly changes the id
#expect(Chunker.contentId(data, encoding: .json, count: 3) != expected)
#expect(Chunker.contentId(data, encoding: .bin, count: 2) != expected)
#expect(Chunker.contentId(Self.payload(20_000, seed: 1), encoding: .bin, count: 3) != expected)
}
@Test func aRetransmitResumesTheSameAssembly() {
let data = Self.payload(30_000)
var asm = ChunkAssembler(maxBytes: 1 << 20)
// the first attempt loses its tail…
for c in Self.parts(Chunker.split(data, encoding: .bin, sliceSize: 8_192)).prefix(2) {
#expect(asm.add(id: c.id, index: c.index, count: c.count, encoding: c.encoding, data: c.data, now: .zero) == .incomplete)
}
// …and the retry, a fresh split of the same payload, fills the same assembly instead of a rival one
var results: [ChunkAssembler.Result] = []
for c in Self.parts(Chunker.split(data, encoding: .bin, sliceSize: 8_192)) {
results.append(asm.add(id: c.id, index: c.index, count: c.count, encoding: c.encoding, data: c.data, now: .seconds(1)))
#expect(asm.partials.count <= 1)
}
#expect(results == [.incomplete, .incomplete, .incomplete, .complete(encoding: .bin, data: data)])
#expect(asm.partials.isEmpty)
}
@Test func dropsSayWhy() {
let d = Data([1, 2, 3, 4])
var asm = ChunkAssembler(maxBytes: 10)
#expect(asm.add(id: "a", index: 0, count: 3, encoding: .bin, data: d, now: .zero) == .incomplete)
#expect(asm.add(id: "a", index: 0, count: 3, encoding: .bin, data: d, now: .zero) == .incomplete) // identical repeat: ignored
#expect(asm.add(id: "a", index: 0, count: 3, encoding: .bin, data: Data([9]), now: .zero) == .dropped(reason: .duplicateIndex))
#expect(asm.partials.isEmpty)
#expect(asm.add(id: "b", index: 0, count: 3, encoding: .bin, data: d, now: .zero) == .incomplete)
#expect(asm.add(id: "b", index: 1, count: 4, encoding: .bin, data: d, now: .zero) == .dropped(reason: .mismatched))
#expect(asm.add(id: "c", index: 0, count: 3, encoding: .bin, data: d, now: .zero) == .incomplete)
#expect(asm.add(id: "c", index: 1, count: 3, encoding: .json, data: d, now: .zero) == .dropped(reason: .mismatched))
#expect(asm.add(id: "d", index: 3, count: 3, encoding: .bin, data: d, now: .zero) == .dropped(reason: .mismatched))
#expect(asm.add(id: "e", index: 0, count: 3, encoding: .bin, data: Data(count: 8), now: .zero) == .incomplete)
#expect(asm.add(id: "e", index: 1, count: 3, encoding: .bin, data: d, now: .zero) == .dropped(reason: .tooManyBytes))
#expect(asm.partials.isEmpty)
// the freed bytes are available again, and completing is not a drop
#expect(asm.add(id: "f", index: 0, count: 2, encoding: .bin, data: d, now: .zero) == .incomplete)
#expect(asm.add(id: "f", index: 1, count: 2, encoding: .bin, data: d, now: .zero) == .complete(encoding: .bin, data: d + d))
#expect(asm.sweep(now: .seconds(3600)).isEmpty)
}
@Test func aNinthConcurrentAssemblyIsRefused() {
var asm = ChunkAssembler(maxBytes: 1 << 20)
for i in 0..<ChunkAssembler.maxConcurrent {
#expect(asm.add(id: "id\(i)", index: 0, count: 2, encoding: .bin, data: Data([1]), now: .zero) == .incomplete)
}
#expect(asm.add(id: "late", index: 0, count: 2, encoding: .bin, data: Data([1]), now: .zero) == .dropped(reason: .tooManyConcurrent))
#expect(asm.partials.count == ChunkAssembler.maxConcurrent) // the ones in flight are untouched
#expect(asm.add(id: "id0", index: 1, count: 2, encoding: .bin, data: Data([2]), now: .zero) == .complete(encoding: .bin, data: Data([1, 2])))
}
@Test func sweepReportsWhatExpired() {
#expect(ChunkAssembler(maxBytes: 1).maxAge == .seconds(60))
#expect(RoomOptions().maxChunkAge == .seconds(60))
var asm = ChunkAssembler(maxBytes: 1 << 20, maxAge: .seconds(5))
_ = asm.add(id: "old", index: 0, count: 2, encoding: .bin, data: Data([1]), now: .zero)
_ = asm.add(id: "young", index: 0, count: 2, encoding: .bin, data: Data([1]), now: .seconds(4))
#expect(asm.sweep(now: .seconds(5)).isEmpty) // exactly maxAge is still alive
#expect(asm.sweep(now: .seconds(6)) == [ChunkAssembler.Drop(id: "old", reason: .ageExpired)])
#expect(asm.sweep(now: .seconds(6)).isEmpty)
#expect(asm.add(id: "young", index: 1, count: 2, encoding: .bin, data: Data([2]), now: .seconds(6)) == .complete(encoding: .bin, data: Data([1, 2])))
// the default outlives that
var slow = ChunkAssembler(maxBytes: 1 << 20)
_ = slow.add(id: "x", index: 0, count: 2, encoding: .bin, data: Data([1]), now: .zero)
#expect(slow.sweep(now: .seconds(60)).isEmpty)
#expect(slow.sweep(now: .seconds(61)) == [ChunkAssembler.Drop(id: "x", reason: .ageExpired)])
}
// MARK: through a room (relayed: 32 768-byte slices)
func pair(_ h: Harness, receiver: RoomOptions) async throws -> (Room, Room, Collector) {
let a = try await h.app().join(secret: h.secret, peerId: lowPeer, options: h.options(preferDataChannel: false))
let b = try await h.app().join(secret: h.secret, peerId: highPeer, options: receiver)
let cb = Collector(b)
#expect(await eventuallyTrue { await a.members == [highPeer] })
#expect(await eventuallyTrue { await b.members == [lowPeer] })
return (a, b, cb)
}
@Test func anExpiredAssemblyIsReportedAfterMaxChunkAge() async throws {
let h = Harness()
var options = h.options(preferDataChannel: false)
options.maxChunkAge = .seconds(5)
let (a, b, cb) = try await pair(h, receiver: options)
let lost = Self.payload(70_000), whole = Self.payload(70_000, seed: 7)
h.relay.dropNextSignals = 1 // the first of three slices never arrives
try await a.send(to: highPeer, .data(lost))
try await a.send(to: highPeer, .json(.string("marker"))) // in order behind the slices: they were processed
#expect(await cb.waitFor(.message(from: lowPeer, .json(.string("marker")))))
h.clock.advance(by: .seconds(6))
try await a.send(to: highPeer, .data(whole))
#expect(await cb.waitFor(.message(from: lowPeer, .data(whole))))
#expect(await cb.waitFor(.assemblyDropped(peer: lowPeer, id: Chunker.contentId(lost, encoding: .bin, count: 3), reason: .ageExpired)))
#expect(!cb.contains(.message(from: lowPeer, .data(lost))))
#expect(cb.all.filter { if case .assemblyDropped = $0 { return true }; return false }.count == 1) // completing is not a drop
await a.leave(); await b.leave()
}
@Test func aYoungAssemblyIsNotReported() async throws {
let h = Harness()
let (a, b, cb) = try await pair(h, receiver: h.options(preferDataChannel: false)) // default: 60 s
let lost = Self.payload(70_000), whole = Self.payload(70_000, seed: 7)
h.relay.dropNextSignals = 1
try await a.send(to: highPeer, .data(lost))
try await a.send(to: highPeer, .json(.string("marker"))) // in order behind the slices: they were processed
#expect(await cb.waitFor(.message(from: lowPeer, .json(.string("marker")))))
h.clock.advance(by: .seconds(6))
try await a.send(to: highPeer, .data(whole))
#expect(await cb.waitFor(.message(from: lowPeer, .data(whole))))
// the retry resumes what the first attempt left behind: only the lost slice was missing
try await a.send(to: highPeer, .data(lost))
#expect(await cb.waitFor(.message(from: lowPeer, .data(lost))))
#expect(!cb.all.contains { if case .assemblyDropped = $0 { return true }; return false })
await a.leave(); await b.leave()
}
@Test func exceedingTheByteBudgetIsReported() async throws {
let h = Harness()
var options = h.options(preferDataChannel: false)
options.maxInflightChunkBytes = 40_000
let (a, b, cb) = try await pair(h, receiver: options)
let big = Self.payload(70_000)
try await a.send(to: highPeer, .data(big))
#expect(await cb.waitFor(.assemblyDropped(peer: lowPeer, id: Chunker.contentId(big, encoding: .bin, count: 3), reason: .tooManyBytes)))
#expect(!cb.contains(.message(from: lowPeer, .data(big))))
await a.leave(); await b.leave()
}
}
|