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
177
178
179
180 | (ns ird.p2p.transport-test
"The frame codec and the `/ws` session over the in-process fake relay:
join, roster, signal forwarding with relay-stamped `from`, ping/pong,
frame-size and pacing checks, terminal refusals, reconnection with the
same peer id."
(:require [cljs.test :refer [deftest is testing]]
[cljs.core.async :as async :refer [<!]]
[ird.p2p.async :as a]
[ird.p2p.descriptor :as desc]
[ird.p2p.errors :as err]
[ird.p2p.fake-relay :as fake]
[ird.p2p.test-util :as tu :refer [deftest-async]]
[ird.p2p.transport :as t]
[ird.p2p.transport.ws :as ws]))
(def room "Xgt16oloGeosHfjLPLb9pPNh0BYNdIknBbtc2ZJm88Q")
(def low "00000000000000000000000000000001")
(def high "0f1e2d3c4b5a69788796a5b4c3d2e1f0")
(deftest frame-codec
(is (= "{\"t\":\"join\",\"room\":\"r\",\"peer\":\"p\"}" (t/encode-frame {:t :join :room "r" :peer "p"})))
(is (= "{\"t\":\"signal\",\"to\":\"p\",\"payload\":{\"v\":1,\"iv\":\"a\",\"ct\":\"b\"}}"
(t/encode-frame {:t :signal :to "p" :payload {:v 1 :iv "a" :ct "b"}})))
(is (= "{\"t\":\"leave\"}" (t/encode-frame {:t :leave})))
(is (= "{\"t\":\"ping\"}" (t/encode-frame {:t :ping})))
(is (= {:t :peers :peers [low high]} (t/decode-text (str "{\"t\":\"peers\",\"peers\":[\"" low "\",\"" high "\"]}"))))
(is (= {:t :peers :peers []} (t/decode-text "{\"t\":\"peers\"}")))
(is (= {:t :peer-joined :peer low} (t/decode-text (str "{\"t\":\"peer-joined\",\"peer\":\"" low "\"}"))))
(is (= {:t :peer-left :peer low} (t/decode-text (str "{\"t\":\"peer-left\",\"peer\":\"" low "\"}"))))
(is (= {:t :signal :from low :payload {:v 1}} (t/decode-text (str "{\"t\":\"signal\",\"from\":\"" low "\",\"payload\":{\"v\":1}}"))))
(is (nil? (t/decode-text "{\"t\":\"signal\",\"from\":\"nope\",\"payload\":1}")) "a bad peer id is undecodable")
(is (nil? (t/decode-text (str "{\"t\":\"signal\",\"from\":\"" low "\"}"))) "a signal without payload is undecodable")
(is (= {:t :pong} (t/decode-text "{\"t\":\"pong\"}")))
(is (= {:t :error :code "room-full"} (t/decode-text "{\"t\":\"error\",\"code\":\"room-full\"}")))
(is (= {:t :error :code ""} (t/decode-text "{\"t\":\"error\"}")))
(is (= {:t :future :x 1} (t/decode-text "{\"t\":\"future\",\"x\":1}")) "unknown types are kept")
(is (nil? (t/decode-text "[1,2]")))
(is (nil? (t/decode-text "{\"x\":1}"))))
(deftest backoff-and-pacing
(is (= 250 (ws/reconnect-delay 0 0)))
(is (= 750 (ws/reconnect-delay 0 1)))
(is (= 1000 (ws/reconnect-delay 1 0.5)))
(is (= 15000 (ws/reconnect-delay 10 0.5)) "capped at 15 s")
(let [p (ws/make-pacer desc/default-limits 0)
[p w1] (ws/pacer-take p 100 0)
[p _] (reduce (fn [[p _] _] (ws/pacer-take p 100 0)) [p 0] (range 59))
[p w61] (ws/pacer-take p 100 0)]
(is (zero? w1))
(is (= 34 w61) "the 61st message in the same instant waits ~1/30 s")
(let [[_ w] (ws/pacer-take p 100 2000)]
(is (zero? w) "refilled after two seconds")))
(let [p (ws/make-pacer desc/default-limits 0)
[_ w] (ws/pacer-take p (* 5 262144) 0)]
(is (= 1000 w) "a byte burst of 5× the rate waits a second"))
(let [p (ws/pacer-drain (ws/make-pacer desc/default-limits 0) 0)
[_ w] (ws/pacer-take p 10 0)]
(is (pos? w) "after rate-limited the next send waits")))
(defn- harness []
(let [relay (fake/make-relay)
clock (tu/test-clock)]
{:relay relay :clock clock
:transport (ws/transport {:socket-fn (fake/socket-fn relay) :sleep-fn (:sleep-fn clock)
:now-fn (:now-fn clock) :jitter-fn (constantly 0.5)})}))
(defn- connect [h peer & [opts]]
(t/connect (:transport h) (merge {:url "wss://x/ws" :room room :peer peer :limits desc/default-limits :pacing? false} opts)))
(deftest-async join-roster-signal-and-leave
(let [h (harness)
a (<! (connect h low))
_ (is (not (tu/error? a)))
ea (tu/collector (t/inbound a))
_ (is (= :joined (t/state a)))
b (<! (connect h high))
eb (tu/collector (t/inbound b))]
(is (some? (<! (tu/wait-for ea #(= {:t :joined} %)))))
(is (some? (<! (tu/wait-for ea #(= {:t :peers :peers []} %)))))
(is (some? (<! (tu/wait-for eb #(= {:t :peers :peers [low]} %)))))
(is (some? (<! (tu/wait-for ea #(= {:t :peer-joined :peer high} %)))))
(is (true? (<! (t/send! a {:t :signal :to high :payload {:v 1 :iv "a" :ct "b"}}))))
(is (some? (<! (tu/wait-for eb #(= {:t :signal :from low :payload {:v 1 :iv "a" :ct "b"}} %)))) "from is relay-stamped")
(is (= [low high] (fake/members (:relay h) room)))
(<! (t/close! b))
(is (some? (<! (tu/wait-for ea #(= {:t :peer-left :peer high} %)))))
(is (some? (<! (tu/wait-for eb #(= {:t :closed :reason "left"} %)))))
(is (= :closed (t/state b)))
(is (err/closed? (<! (t/send! b {:t :ping}))))
(<! (t/close! a))
(is (nil? (<! (t/inbound a))) "the inbound channel closes after :closed")))
(deftest-async frame-too-large-is-refused-before-sending
(let [h (harness)
a (<! (connect h low {:limits (assoc desc/default-limits :max_frame_bytes 200)}))
e (<! (t/send! a {:t :signal :to high :payload (apply str (repeat 300 "x"))}))]
(is (err/frame-too-large? e))
(is (= 200 (:limit (ex-data e))))
(is (empty? (fake/signal-frames (:relay h))))
(<! (t/close! a))))
(deftest-async terminal-refusals
(testing "403 and 404 upgrades end the session for good"
(let [relay (fake/make-relay {:reject-with 403})
tr (ws/transport {:socket-fn (fake/socket-fn relay) :sleep-fn tu/no-sleep :now-fn (constantly 0)})
e (<! (t/connect tr {:url "wss://x/ws" :room room :peer low :limits desc/default-limits}))]
(is (err/upgrade-rejected? e))
(is (= 403 (:status (ex-data e))))))
(testing "room-full is terminal"
(let [relay (fake/make-relay {:max-room-peers 1})
clock (tu/test-clock)
tr (ws/transport {:socket-fn (fake/socket-fn relay) :sleep-fn (:sleep-fn clock) :now-fn (:now-fn clock)})
a (<! (t/connect tr {:url "wss://x/ws" :room room :peer low :limits desc/default-limits}))
e (<! (t/connect tr {:url "wss://x/ws" :room room :peer high :limits desc/default-limits}))]
(is (not (tu/error? a)))
(is (err/relay-error? e))
(is (= "room-full" (:code (ex-data e))))
(<! (t/close! a))))
(testing "a 429 keeps retrying in the background and delivers the session early"
(let [relay (fake/make-relay {:reject-with 429})
clock (tu/test-clock)
tr (ws/transport {:socket-fn (fake/socket-fn relay) :sleep-fn (:sleep-fn clock) :now-fn (:now-fn clock) :jitter-fn (constantly 0.5)})
s (<! (t/connect tr {:url "wss://x/ws" :room room :peer low :limits desc/default-limits}))
es (tu/collector (t/inbound s))]
(is (not (tu/error? s)))
(is (= :reconnecting (t/state s)))
(is (some? (<! (tu/wait-for es #(= {:t :reconnecting :attempt 1 :after-ms 500} %)))))
(reset! (:reject-with relay) nil)
((:advance! clock) 500)
(is (some? (<! (tu/wait-for es #(= {:t :joined} %)))))
(is (= :joined (t/state s)))
(<! (t/close! s)))))
(deftest-async keepalive-and-reconnect-with-the-same-peer-id
(let [h (harness)
{:keys [relay clock]} h
a (<! (connect h low))
ea (tu/collector (t/inbound a))
b (<! (connect h high))
eb (tu/collector (t/inbound b))]
(is (some? (<! (tu/wait-for eb #(= {:t :peers :peers [low]} %)))))
;; keepalive: after 20 s a ping goes out and the relay answers pong
(is (true? (<! (tu/eventually #(>= ((:sleepers clock)) 2)))))
((:advance! clock) 20000)
(is (true? (<! (tu/eventually #(some (fn [f] (= "{\"t\":\"ping\"}" (:text f))) @(:frames relay))))))
;; network loss: b's socket dies without a leave
(fake/drop-socket! relay high)
(is (some? (<! (tu/wait-for ea #(= {:t :peer-left :peer high} %)))))
(is (some? (<! (tu/wait-for eb #(= :reconnecting (:t %))))))
(is (= :reconnecting (t/state b)))
(is (err/not-joined? (<! (t/send! b {:t :signal :to low :payload 1}))))
;; b reconnects after the backoff with the same peer id
(is (true? (<! (tu/eventually #(>= ((:sleepers clock)) 1)))))
((:advance! clock) 15000)
(is (true? (<! (tu/eventually #(= 2 (tu/count-of eb (fn [e] (= {:t :joined} e))))))))
(is (some? (<! (tu/wait-for ea #(= {:t :peer-joined :peer high} %)))))
(is (= [low high] (fake/members relay room)))
;; wake! cuts a pending backoff short
(fake/drop-socket! relay high)
(is (true? (<! (tu/eventually #(= 2 (tu/count-of eb (fn [e] (= :reconnecting (:t e)))))))))
(t/wake! b)
(is (true? (<! (tu/eventually #(= 3 (tu/count-of eb (fn [e] (= {:t :joined} e))))))))
(<! (t/close! a))
(<! (t/close! b))))
(deftest-async pacing-waits-through-the-clock
(let [h (harness)
clock (:clock h)
a (<! (connect h low {:pacing? true :limits (assoc desc/default-limits :message_burst 2 :messages_per_sec 1)}))
b (<! (connect h high {:pacing? true}))
eb (tu/collector (t/inbound b))
_ (<! (t/send! a {:t :signal :to high :payload 1}))
_ (<! (t/send! a {:t :signal :to high :payload 2}))
third (t/send! a {:t :signal :to high :payload 3})]
(is (some? (<! (tu/wait-for eb #(= 2 (:payload %))))))
(is (nil? (async/poll! third)) "the third frame waits for the budget")
((:advance! clock) 1000)
(is (true? (<! third)))
(is (some? (<! (tu/wait-for eb #(= 3 (:payload %))))))
(<! (t/close! a)) (<! (t/close! b))))
|