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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176 | import Foundation
import Testing
@testable import IrdP2PSignaling
import IrdP2PCrypto
@Suite struct RelaySessionTests {
func makeSession(ws: FakeWebSocketTransport, clock: TestClock, origin: String? = "https://app.example.test", pacing: Bool = false) -> RelaySession {
var cfg = RelaySession.Configuration(url: URL(string: "wss://app.example.test/ws")!, origin: origin, room: testRoom, peer: testPeer, limits: .defaults)
cfg.jitter = { 0.5 }
cfg.pacing = pacing
return RelaySession(configuration: cfg, transport: ws, clock: clock)
}
/// Runs `start()` concurrently and answers the join with the given roster.
func startAndJoin(_ s: RelaySession, ws: FakeWebSocketTransport, roster: [PeerId] = []) async throws -> FakeConnection {
let starter = Task { try await s.start() }
#expect(await eventually { ws.count == 1 && !(ws.latest?.sentFrames.isEmpty ?? true) })
let conn = ws.latest!
conn.push(#"{"t":"peers","peers":[\#(roster.map { "\"\($0.hex)\"" }.joined(separator: ","))]}"#)
try await starter.value
return conn
}
@Test func joinsWithOriginAndReportsRoster() async throws {
let ws = FakeWebSocketTransport(), clock = TestClock()
let s = makeSession(ws: ws, clock: clock)
var events = s.events.makeAsyncIterator()
let conn = try await startAndJoin(s, ws: ws, roster: [otherPeer])
#expect(conn.origin == "https://app.example.test")
#expect(conn.request.url?.absoluteString == "wss://app.example.test/ws")
#expect(conn.sentFrames.first == ClientFrame.join(room: testRoom, peer: testPeer).encoded)
#expect(await events.next() == .joined)
#expect(await events.next() == .roster([otherPeer]))
#expect(await s.isJoined)
conn.push(#"{"t":"peer-joined","peer":"\#(otherPeer.hex)"}"#)
conn.push(#"{"t":"signal","from":"\#(otherPeer.hex)","payload":{"v":1,"iv":"a","ct":"b"}}"#)
conn.push(#"{"t":"weird"}"#)
conn.push(#"{"t":"error","code":"rate-limited"}"#)
conn.push(#"{"t":"peer-left","peer":"\#(otherPeer.hex)"}"#)
#expect(await events.next() == .peerJoined(otherPeer))
#expect(await events.next() == .signal(from: otherPeer, payload: .object(["v": .int(1), "iv": .string("a"), "ct": .string("b")])))
#expect(await events.next() == .relayError(.rateLimited))
#expect(await events.next() == .peerLeft(otherPeer))
await s.leave()
#expect(conn.sentFrames.last == ClientFrame.leave.encoded)
#expect(conn.isClosedByClient)
#expect(await events.next() == .closed(reason: "left"))
#expect(await events.next() == nil)
}
@Test func noOriginWhenNotConfigured() async throws {
let ws = FakeWebSocketTransport(), clock = TestClock()
let s = makeSession(ws: ws, clock: clock, origin: nil)
let conn = try await startAndJoin(s, ws: ws)
#expect(conn.origin == nil)
await s.leave()
}
@Test func refusedUpgradeIsTerminal() async throws {
let ws = FakeWebSocketTransport(), clock = TestClock()
ws.script([.upgradeRejected(status: 403)])
let s = makeSession(ws: ws, clock: clock)
await #expect(throws: P2PError.upgradeRejected(status: 403)) { try await s.start() }
#expect(await s.state == .closed)
#expect(ws.count == 0)
}
@Test func roomFullOnJoinIsTerminal() async throws {
let ws = FakeWebSocketTransport(), clock = TestClock()
let s = makeSession(ws: ws, clock: clock)
let starter = Task { try await s.start() }
#expect(await eventually { ws.count == 1 })
ws.latest!.push(#"{"t":"error","code":"room-full"}"#)
await #expect(throws: P2PError.relay(.roomFull)) { try await starter.value }
#expect(await s.state == .closed)
}
@Test func transientFailuresReconnectWithSamePeerAndBackoff() async throws {
let ws = FakeWebSocketTransport(), clock = TestClock()
ws.script([.transport("dns")])
let s = makeSession(ws: ws, clock: clock)
var events = s.events.makeAsyncIterator()
let starter = Task { try await s.start() }
// first attempt fails: reconnecting after 500 ms × (0.5 + 0.5)
#expect(await events.next() == .reconnecting(attempt: 1, after: .milliseconds(500)))
#expect(await eventually { clock.sleeperCount == 1 })
clock.advance(by: .milliseconds(500))
#expect(await eventually { ws.count == 1 && !(ws.latest?.sentFrames.isEmpty ?? true) })
let c1 = ws.latest!
#expect(c1.sentFrames.first == ClientFrame.join(room: testRoom, peer: testPeer).encoded)
c1.push(#"{"t":"peers","peers":[]}"#)
try await starter.value
#expect(await events.next() == .joined)
#expect(await events.next() == .roster([]))
// server drops the socket: reconnect from attempt 0 again (backoff reset after a successful join)
c1.serverClose()
#expect(await events.next() == .reconnecting(attempt: 1, after: .milliseconds(500)))
#expect(await eventually { clock.sleeperCount == 1 }) // the keepalive sleeper was cancelled; only the backoff remains
clock.advance(by: .milliseconds(500))
#expect(await eventually { ws.count == 2 && !(ws.latest?.sentFrames.isEmpty ?? true) })
let c2 = ws.latest!
#expect(c2.sentFrames.first == ClientFrame.join(room: testRoom, peer: testPeer).encoded)
c2.push(#"{"t":"peers","peers":["\#(otherPeer.hex)"]}"#)
#expect(await events.next() == .joined)
#expect(await events.next() == .roster([otherPeer]))
await s.leave()
}
@Test func wakeSkipsTheBackoffSleep() async throws {
let ws = FakeWebSocketTransport(), clock = TestClock()
let s = makeSession(ws: ws, clock: clock)
var events = s.events.makeAsyncIterator()
let c1 = try await startAndJoin(s, ws: ws)
_ = await events.next(); _ = await events.next()
c1.serverClose()
#expect(await events.next() == .reconnecting(attempt: 1, after: .milliseconds(500)))
#expect(await eventually { clock.sleeperCount == 1 })
await s.wake()
#expect(await eventually { ws.count == 2 })
ws.latest!.push(#"{"t":"peers","peers":[]}"#)
#expect(await events.next() == .joined)
await s.leave()
}
@Test func keepalivePingsAndDropsHalfOpenConnections() async throws {
let ws = FakeWebSocketTransport(), clock = TestClock()
let s = makeSession(ws: ws, clock: clock)
var events = s.events.makeAsyncIterator()
let conn = try await startAndJoin(s, ws: ws)
_ = await events.next(); _ = await events.next()
#expect(await eventually { clock.sleeperCount == 1 }) // the keepalive timer
clock.advance(by: .seconds(20))
#expect(await eventually { conn.sentFrames.filter { $0 == ClientFrame.ping.encoded }.count == 1 })
conn.push(#"{"t":"pong"}"#)
#expect(await eventually { clock.sleeperCount == 1 })
clock.advance(by: .seconds(20))
#expect(await eventually { conn.sentFrames.filter { $0 == ClientFrame.ping.encoded }.count == 2 })
// no pong this time, nor the next: the client drops the socket and reconnects
#expect(await eventually { clock.sleeperCount == 1 })
clock.advance(by: .seconds(20))
#expect(await eventually { conn.sentFrames.filter { $0 == ClientFrame.ping.encoded }.count == 3 })
#expect(await eventually { clock.sleeperCount == 1 })
clock.advance(by: .seconds(20))
#expect(await eventually { conn.isClosedByClient })
#expect(await events.next() == .reconnecting(attempt: 1, after: .milliseconds(500)))
await s.leave()
}
@Test func sendRefusesOversizeAndUnjoined() async throws {
let ws = FakeWebSocketTransport(), clock = TestClock()
let s = makeSession(ws: ws, clock: clock)
await #expect(throws: P2PError.notJoined) { try await s.send(signal: .null, to: otherPeer) }
let conn = try await startAndJoin(s, ws: ws)
try await s.send(signal: .object(["v": .int(1)]), to: otherPeer)
#expect(conn.sentFrames.last == #"{"t":"signal","to":"\#(otherPeer.hex)","payload":{"v":1}}"#)
let big = JSONValue.string(String(repeating: "x", count: 65_536))
let frameBytes = ClientFrame.signal(to: otherPeer, payload: big).encoded.utf8.count
await #expect(throws: P2PError.frameTooLarge(bytes: frameBytes, limit: 65_536)) { try await s.send(signal: big, to: otherPeer) }
await s.leave()
}
@Test func pacingDelaysBurstsPastTheBudget() async throws {
let ws = FakeWebSocketTransport(), clock = TestClock()
let s = makeSession(ws: ws, clock: clock, pacing: true)
let conn = try await startAndJoin(s, ws: ws)
for _ in 0..<60 { try await s.send(signal: .int(1), to: otherPeer) } // the burst goes through untouched
#expect(conn.sentFrames.count == 61)
let sender = Task { try await s.send(signal: .int(2), to: otherPeer) }
#expect(await eventually { clock.sleeperCount >= 2 }) // keepalive + paced send
#expect(conn.sentFrames.count == 61)
clock.advance(by: .milliseconds(40))
try await sender.value
#expect(conn.sentFrames.count == 62)
await s.leave()
}
}
|