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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250 | (ns ird.p2p.transport.ws
"The default transport: one `/ws` connection to a room with join,
keepalive (ping/pong), reconnection with backoff — reusing the same peer
id so the relay replaces the stale socket — pacing against the node's
budgets and frame-size checks. The socket itself comes from an injectable
:socket-fn (default ird.p2p.impl.ws/open over the `ws` package), which is
how the offline tests run against an in-process relay.
A socket is {:in <channel of {:type :text/:binary :data ...}, closes on
close> :send! (fn [text]) :close! (fn [code reason])}; the `/ws` protocol
is JSON text frames, binary frames are ignored."
(:require [clojure.string :as str]
[cljs.core.async :as async :refer [go go-loop <! >! alts!]]
[ird.p2p.async :as a :refer [go-try <?]]
[ird.p2p.descriptor :as descriptor]
[ird.p2p.errors :as err]
[ird.p2p.impl.bytes :as bytes]
[ird.p2p.impl.node :as node]
[ird.p2p.impl.ws :as impl-ws]
[ird.p2p.transport :as t]))
;; ---- backoff -------------------------------------------------------------------------
(defn reconnect-delay
"Relay reconnect: min(15000, 500·2^attempt) × (0.5 + jitter) milliseconds."
[attempt jitter]
(let [base (min 15000 (* 500 (apply * (repeat (max 0 attempt) 2))))]
(long (* base (+ 0.5 jitter)))))
;; ---- pacing --------------------------------------------------------------------------
(defn make-bucket
"A token bucket: {:capacity :refill-per-sec :tokens :last}."
[capacity refill-per-sec now]
{:capacity capacity :refill-per-sec refill-per-sec :tokens capacity :last now})
(defn- ceil-ms [v]
(let [t (long v)] (if (< t v) (inc t) t)))
(defn- refill [{:keys [capacity refill-per-sec tokens last] :as b} now]
(let [elapsed (max 0 (/ (- now last) 1000))]
(assoc b :tokens (min capacity (+ tokens (* elapsed refill-per-sec))) :last now)))
(defn bucket-take
"[bucket' wait-ms]: takes COST tokens; the bucket goes into debt, so
concurrent callers queue behind each other."
[b cost now]
(let [b (refill b now)
tokens (- (:tokens b) cost)
b (assoc b :tokens tokens)]
[b (if (>= tokens 0) 0 (ceil-ms (* 1000 (/ (- tokens) (:refill-per-sec b)))))]))
(defn make-pacer
"The two relay budgets: messages per second (with burst) and bytes per second."
[limits now]
{:messages (make-bucket (:message_burst limits) (:messages_per_sec limits) now)
:bytes (make-bucket (max (* 4 (:max_bytes_per_sec limits)) (:max_frame_bytes limits)) (:max_bytes_per_sec limits) now)})
(defn pacer-take
"[pacer' wait-ms] before a frame of BYTE-COUNT bytes may be sent."
[pacer byte-count now]
(let [[m a] (bucket-take (:messages pacer) 1 now)
[b c] (bucket-take (:bytes pacer) byte-count now)]
[{:messages m :bytes b} (max a c)]))
(defn pacer-drain
"After a rate-limited frame: empty both buckets so the next sends wait a full refill."
[pacer now]
(let [[m _] (bucket-take (:messages pacer) (get-in pacer [:messages :capacity]) now)
[b _] (bucket-take (:bytes pacer) (get-in pacer [:bytes :capacity]) now)]
{:messages m :bytes b}))
;; ---- session --------------------------------------------------------------------------
(defn- terminal-upgrade? [e]
(and (err/upgrade-rejected? e) (contains? #{403 404} (:status (ex-data e)))))
(defn- terminal-error? [e]
(or (terminal-upgrade? e)
(and (err/relay-error? e) (contains? t/terminal-relay-codes (:code (ex-data e))))))
(defn- socket-frames
"Decode text messages of a socket into server frames; nil for anything else."
[{:keys [type data]}]
(when (= :text type) (t/decode-text data)))
(defn- await-join
"Read frames until the roster (join accepted) or a join error. Returns
:joined, {:error e} for a terminal refusal, or :retry when the socket died."
[conn]
(go-loop []
(let [m (<! (:in conn))]
(if (nil? m)
:retry
(let [f (socket-frames m)]
(case (:t f)
:peers {:peers f}
:error (let [e (err/relay-error (:code f))]
(if (terminal-error? e) {:error e} (recur)))
(recur)))))))
(defn- run
"The connection loop of one session. Owns the socket; delivers the session
on CONNECT-CH the first time the room is joined (or the terminal error)."
[{:keys [socket-fn sleep-fn now-fn jitter-fn] :as _transport}
{:keys [url origin room peer keepalive-ms missed-pongs connect-timeout-ms] :as _opts}
{:keys [st sock emit close-out wake-ch pacer pongs session connect-ch] :as s}]
(let [keepalive-ms (or keepalive-ms 20000)
missed-pongs (or missed-pongs 2)
deliver-once! (fn [v] (when (compare-and-set! (:delivered s) false true) (a/deliver! connect-ch v)))
finish! (fn [reason]
(reset! st :closed)
(emit {:t :closed :reason reason})
(close-out))
backoff! (fn [attempt]
(go
(let [delay (reconnect-delay attempt (jitter-fn))]
(reset! st :reconnecting)
(emit {:t :reconnecting :attempt (inc attempt) :after-ms delay})
(alts! [(sleep-fn delay) @wake-ch])
nil)))]
(go-loop [attempt 0]
(when (not= :closed @st)
(reset! st (if (zero? attempt) :connecting :reconnecting))
(let [conn (<! (socket-fn {:url url :origin origin :timeout-ms (or connect-timeout-ms 15000)}))
outcome
(cond
(a/error? conn)
(if (terminal-error? conn)
(do (deliver-once! conn) (finish! (ex-message conn)) :done)
(do (reset! st :reconnecting)
(when (err/upgrade-rejected? conn) (deliver-once! session))
:retry))
(= :closed @st)
(do ((:close! conn) 1000 nil) :done)
:else
(do
(reset! sock conn)
(if (some? ((:send! conn) (t/encode-frame {:t :join :room room :peer peer})))
(do ((:close! conn) 1000 nil) (reset! sock nil) :retry)
(let [r (<! (await-join conn))]
(cond
(= :retry r) (do (reset! sock nil) :retry)
(:error r) (do ((:close! conn) 1000 nil) (reset! sock nil)
(deliver-once! (:error r)) (finish! (ex-message (:error r))) :done)
:else (do (reset! st :joined)
(deliver-once! session)
(emit {:t :joined})
(emit (:peers r))
:joined))))))]
(case outcome
:done nil
:joined
(let [conn @sock
gen (swap! (:keepalive-gen s) inc)]
(reset! pongs 0)
;; keepalive: ping every keepalive-ms; drop a half-open socket after missed pongs
(go-loop []
(<! (sleep-fn keepalive-ms))
(when (and (= gen @(:keepalive-gen s)) (= :joined @st) (identical? conn @sock))
(if (>= @pongs missed-pongs)
((:close! conn) 1001 "no pong")
(do (swap! pongs inc)
(when (nil? ((:send! conn) (t/encode-frame {:t :ping})))
(recur))))))
;; receive loop until the socket closes
(loop []
(when-let [m (<! (:in conn))]
(when-let [f (socket-frames m)]
(case (:t f)
:pong (reset! pongs 0)
:error (do (when (= "rate-limited" (:code f)) (swap! pacer pacer-drain (now-fn)))
(emit f))
(emit f)))
(recur)))
(swap! (:keepalive-gen s) inc)
(reset! sock nil)
(when (not= :closed @st)
(<! (backoff! 0))
(recur 1)))
:retry
(when (not= :closed @st)
(<! (backoff! attempt))
(recur (inc attempt)))))))))
(defn- make-session
[{:keys [sleep-fn now-fn] :as transport} {:keys [room peer limits pacing?] :as opts}]
(let [{:keys [out put! close!]} (a/unbounded-chan)
limits (or limits descriptor/default-limits)
st (atom :connecting)
sock (atom nil)
pacer (atom (make-pacer limits (now-fn)))
wake-ch (atom (async/chan))
s {:st st :sock sock :emit put! :close-out close! :pacer pacer :pongs (atom 0)
:keepalive-gen (atom 0) :delivered (atom false) :connect-ch (async/promise-chan)}
max-frame (:max_frame_bytes limits)
session
(reify
t/Session
(send! [_ frame]
(go-try
(let [text (t/encode-frame frame)
size (bytes/utf8-byte-length text)]
(when (> size max-frame) (throw (err/frame-too-large size max-frame)))
(when (not= false pacing?)
(let [wait (atom 0)]
(swap! pacer (fn [p] (let [[p' w] (pacer-take p size (now-fn))] (reset! wait w) p')))
(when (pos? @wait) (<! (sleep-fn @wait)))))
(when (= :closed @st) (throw (err/closed)))
(when-not (and (= :joined @st) @sock) (throw (err/not-joined)))
(let [e ((:send! @sock) text)]
(when (some? e) (throw e)))
true)))
(inbound [_] out)
(state [_] @st)
(wake! [_]
(let [old @wake-ch]
(reset! wake-ch (async/chan))
(async/close! old))
nil)
(close! [_]
(go
(when (not= :closed @st)
(reset! st :closed)
(when-let [c @sock]
((:send! c) (t/encode-frame {:t :leave}))
((:close! c) 1000 "leave")
(reset! sock nil))
(async/close! @wake-ch)
(put! {:t :closed :reason "left"})
(close!))
nil)))]
(assoc s :session session :wake-ch wake-ch)))
(defn transport
"A `/ws` relay transport. OPTS: :socket-fn (default ird.p2p.impl.ws/open),
:sleep-fn (ms → channel), :now-fn (→ ms), :jitter-fn (→ [0,1))."
([] (transport nil))
([{:keys [socket-fn sleep-fn now-fn jitter-fn]}]
(let [tr {:socket-fn (or socket-fn impl-ws/open)
:sleep-fn (or sleep-fn a/sleep)
:now-fn (or now-fn node/now-ms)
:jitter-fn (or jitter-fn rand)}]
(reify t/Transport
(connect [_ opts]
(let [s (make-session tr opts)]
(run tr opts s)
(:connect-ch s)))))))
|