accidental.cc
podcast manager
simple reconnect loop
+28
-6
src/common/async/sleep.ts
+28
-6
src/common/async/sleep.ts
···
3
* @param signal - an aptional abort signal, to cancel the sleep
4
* @returns a promise that resolves after given amount of time, and is interruptable with an abort signal.
5
*/
6
-
export function sleep(ms: number, signal?: AbortSignal): Promise<void> {
7
signal?.throwIfAborted()
8
-
9
-
// not sure why this error is coming up
10
11
const {resolve, reject, promise} = Promise.withResolvers<void>()
12
const timeout = setTimeout(resolve, ms)
13
14
if (!signal) return promise
15
16
const abortHandler = () => {
···
18
reject(signal.reason)
19
}
20
21
-
signal.addEventListener('abort', abortHandler)
22
-
return promise.finally(() => {
23
signal.removeEventListener('abort', abortHandler)
24
-
})
25
}
···
3
* @param signal - an aptional abort signal, to cancel the sleep
4
* @returns a promise that resolves after given amount of time, and is interruptable with an abort signal.
5
*/
6
+
export async function sleep(ms: number, signal?: AbortSignal): Promise<void> {
7
signal?.throwIfAborted()
8
9
const {resolve, reject, promise} = Promise.withResolvers<void>()
10
const timeout = setTimeout(resolve, ms)
11
12
+
// without the signal, we can't cancel it, so just let the timeout resolve
13
if (!signal) return promise
14
15
const abortHandler = () => {
···
17
reject(signal.reason)
18
}
19
20
+
try {
21
+
signal.addEventListener('abort', abortHandler)
22
+
await promise
23
+
return
24
+
} finally {
25
signal.removeEventListener('abort', abortHandler)
26
+
}
27
+
}
28
+
29
+
export function backoff(options?: {maxAttempts?: number; baseDelay?: number; maxDelay?: number}) {
30
+
const maxAttempts = options?.maxAttempts ?? 10
31
+
const baseDelay = options?.baseDelay ?? 1000
32
+
const maxDelay = options?.maxDelay ?? 30_000
33
+
34
+
let attempts = 0
35
+
const nextDelay = () => {
36
+
return attempts === 0
37
+
? attempts++ // immediate at 0
38
+
: Math.min(baseDelay * Math.pow(2, attempts++ - 1), maxDelay)
39
+
}
40
+
41
+
return async (signal?: AbortSignal) => {
42
+
signal?.throwIfAborted()
43
+
if (attempts > maxAttempts) throw new Error('exceeded max attempts!')
44
+
45
+
await sleep(nextDelay(), signal)
46
+
}
47
}
+89
-17
src/realm/client/service-connection.ts
+89
-17
src/realm/client/service-connection.ts
···
3
import {z} from 'zod/v4'
4
5
import {timeoutSignal} from '#common/async/aborts'
6
-
import {sleep} from '#common/async/sleep'
7
import {generateSignableJwt, jwkImport} from '#common/crypto/jwks'
8
import {jwtPayload, verifyJwtToken} from '#common/crypto/jwts'
9
import {normalizeError, normalizeProtocolError, ProtocolError} from '#common/errors'
···
34
35
/** manages websocket and webrtc connections for a realm */
36
export class RealmConnection extends EventTarget {
37
#socket: WebSocket
38
#connectopts: ConnectionOptions
39
#abort: AbortController
40
41
#identities: Map<IdentID, CryptoKey>
···
51
52
constructor(url: string, db: Database, identity: RealmIdentity, options: ConnectionOptions) {
53
super()
54
this.#abort = new AbortController()
55
this.#identity = identity
56
this.#connectopts = options
···
131
132
destroy() {
133
console.debug('realm connection destroy!')
134
-
if (this.connected) {
135
-
this.#socket.close()
136
-
}
137
138
-
// disconnect from peers
139
for (const peer of this.#peers.values()) {
140
peer.destroy()
141
}
142
143
-
// shutdown loops
144
-
this.#abort.abort()
145
-
146
this.#peers.clear()
147
this.#nonces.clear()
148
}
149
150
#dispatchCustomEvent(type: string, detail?: object) {
151
this.dispatchEvent(new CustomEvent(type, {bubbles: true, detail}))
152
}
···
178
for await (const data of streamSocketJson(this.#socket)) {
179
await this.#handleOpenMessage(data)
180
}
181
-
} catch (exc) {
182
const err = normalizeProtocolError(exc)
183
-
184
console.error('realm connection, socket loop error', err)
185
this.#dispatchCustomEvent('wserror', {error: err})
186
} finally {
187
-
this.destroy()
188
}
189
}
190
···
335
}
336
}
337
338
-
#handleSocketError: WebSocket['onerror'] = (exc) => {
339
-
this.#dispatchCustomEvent('wserror', {error: normalizeProtocolError(exc)})
340
-
this.destroy()
341
}
342
343
#handleSocketClose: WebSocket['onclose'] = () => {
344
this.#dispatchCustomEvent('wsclose')
345
-
this.destroy()
346
}
347
348
#socketLoop = async () => {
349
this.#abort.signal.throwIfAborted()
350
351
-
await sleep(250)
352
await this.#pingSocket()
353
354
while (!this.#abort.signal.aborted) {
355
-
await sleep(30_000)
356
await this.#pingSocket()
357
}
358
}
···
3
import {z} from 'zod/v4'
4
5
import {timeoutSignal} from '#common/async/aborts'
6
+
import {backoff, sleep} from '#common/async/sleep'
7
import {generateSignableJwt, jwkImport} from '#common/crypto/jwks'
8
import {jwtPayload, verifyJwtToken} from '#common/crypto/jwts'
9
import {normalizeError, normalizeProtocolError, ProtocolError} from '#common/errors'
···
34
35
/** manages websocket and webrtc connections for a realm */
36
export class RealmConnection extends EventTarget {
37
+
#url: string
38
#socket: WebSocket
39
#connectopts: ConnectionOptions
40
+
#reconnecting = false
41
#abort: AbortController
42
43
#identities: Map<IdentID, CryptoKey>
···
53
54
constructor(url: string, db: Database, identity: RealmIdentity, options: ConnectionOptions) {
55
super()
56
+
this.#url = url
57
this.#abort = new AbortController()
58
this.#identity = identity
59
this.#connectopts = options
···
134
135
destroy() {
136
console.debug('realm connection destroy!')
137
+
this.#reconnecting = false
138
+
139
+
// shutdown the loops
140
+
this.#abort.abort()
141
142
+
// close ports
143
+
this.#closeWebSocket()
144
for (const peer of this.#peers.values()) {
145
peer.destroy()
146
}
147
148
+
// empty state
149
this.#peers.clear()
150
this.#nonces.clear()
151
}
152
153
+
#closeWebSocket() {
154
+
if (this.connected) {
155
+
this.#socket.close()
156
+
}
157
+
}
158
+
159
+
async #attemptReconnect() {
160
+
if (this.#reconnecting || this.#abort.signal.aborted) return
161
+
162
+
// backoff sleeps for exponentially longer amounts of time
163
+
const gate = backoff({maxAttempts: 100_000})
164
+
try {
165
+
this.#reconnecting = true
166
+
while (true) {
167
+
await gate(this.#abort.signal)
168
+
169
+
console.debug('attempting WebSocket reconnection')
170
+
this.#dispatchCustomEvent('wsreconnect.start')
171
+
172
+
try {
173
+
this.#socket = new WebSocket(this.#url)
174
+
this.#socket.onopen = this.#handleSocketOpen
175
+
this.#socket.onclose = this.#handleSocketClose
176
+
this.#socket.onerror = this.#handleSocketError
177
+
178
+
// wait a bit to see if connection succeeds
179
+
await sleep(500)
180
+
181
+
if (this.connected) {
182
+
console.debug('WebSocket reconnection successful')
183
+
this.#reconnecting = false
184
+
return
185
+
}
186
+
} catch (err: unknown) {
187
+
console.warn('WebSocket reconnection attempt failed:', err)
188
+
this.#dispatchCustomEvent('wsreconnect.failed')
189
+
}
190
+
}
191
+
} catch (err: unknown) {
192
+
// why is typescript saying signal.aborted: false, not boolean?
193
+
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
194
+
if (this.#abort.signal.aborted) {
195
+
console.warn('WebSocket reconnection aborted')
196
+
this.#dispatchCustomEvent('wsreconnect.abandoned')
197
+
} else {
198
+
console.error('WebSocket reconnection failed after max attempts', err)
199
+
this.#dispatchCustomEvent('wsreconnect.abandoned')
200
+
}
201
+
202
+
this.#reconnecting = false
203
+
}
204
+
}
205
+
206
#dispatchCustomEvent(type: string, detail?: object) {
207
this.dispatchEvent(new CustomEvent(type, {bubbles: true, detail}))
208
}
···
234
for await (const data of streamSocketJson(this.#socket)) {
235
await this.#handleOpenMessage(data)
236
}
237
+
} catch (exc: unknown) {
238
const err = normalizeProtocolError(exc)
239
console.error('realm connection, socket loop error', err)
240
this.#dispatchCustomEvent('wserror', {error: err})
241
} finally {
242
+
this.#closeWebSocket()
243
+
this.#attemptReconnect().catch((err: unknown) => {
244
+
console.error('realm connection, reconnect loop error:', err)
245
+
this.#dispatchCustomEvent('wserror', {error: err})
246
+
})
247
}
248
}
249
···
394
}
395
}
396
397
+
#handleSocketError: WebSocket['onerror'] = (exc: unknown) => {
398
+
const error = normalizeProtocolError(exc)
399
+
400
+
console.error('socket error', error)
401
+
this.#dispatchCustomEvent('wserror', {error})
402
}
403
404
#handleSocketClose: WebSocket['onclose'] = () => {
405
this.#dispatchCustomEvent('wsclose')
406
+
407
+
if (!this.#reconnecting && !this.#abort.signal.aborted) {
408
+
this.#attemptReconnect().catch((exc: unknown) => {
409
+
const error = normalizeProtocolError(exc)
410
+
411
+
console.error('socket error', error)
412
+
this.#dispatchCustomEvent('wserror', {error})
413
+
})
414
+
}
415
}
416
417
+
// loop that handles _realm_ level socket comms
418
#socketLoop = async () => {
419
this.#abort.signal.throwIfAborted()
420
421
+
// why does typescript say #abort.signal: false, not : boolean?
422
+
423
+
await sleep(250, this.#abort.signal)
424
await this.#pingSocket()
425
426
while (!this.#abort.signal.aborted) {
427
+
await sleep(30_000, this.#abort.signal)
428
await this.#pingSocket()
429
}
430
}
+2
-3
src/realm/server/state-storage.ts
+2
-3
src/realm/server/state-storage.ts