rickub
nandi/gleanpublic Fork
⑂ 5eef251
Branches
claude/modal-container-min-76527a main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/glean.git
git clone ssh://git@rickub.com/nandi/glean.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

Download ZIP

Keep a Jetstream subscription alive across drops

Adds a supervisor over the consumer: resume from a rewound cursor, rotate
connections on a timer, and back off with jitter.

The rewind matters more than the reconnect. Cursors are persisted in
batches, so the last one written always trails the last one seen;
reconnecting rewinds past that gap and accepts re-delivery, because
duplicates are cheap when handling is idempotent and a hole is not.
Rotation exists because server-side filters are fixed per connection, so a
subscription that never reconnects never learns about newly-known DIDs --
robustness is not the reason for it.

Tested against a local server that hangs up after three frames, since a
healthy server proves nothing about the case this module is for: 12
connections and 36 events in two seconds. The probe also checks the cursor
reaches the server rather than only advancing locally -- a consumer tracking
a cursor it never sends looks identical from the inside and loses the gap on
every reconnect.

One bug found: delayFor short-circuited attempt <= 0 to the bare initial
delay, skipping jitter on the first retry. That is the exact moment jitter
is for, since one server restart drops every consumer at once.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
nandithebull committed 2026-09-20T23:31:13-07:00 Browse files
5eef251 parent: 38b6da7
modified spike/nim/.gitignore +1 -0
@@ -5,3 +5,4 @@ dpop_probe
55 oauth_probe
66 xrpc_probe
77 jetstream_probe
8+reconnect_probe
@@ -5,3 +5,4 @@ dpop_probe
5 oauth_probe5 oauth_probe
6 xrpc_probe6 xrpc_probe
7 jetstream_probe7 jetstream_probe
8+reconnect_probe
added spike/nim/atproto/jetstream_run.nim +167 -0
new file mode 100644
@@ -0,0 +1,167 @@
1+## Keeping a Jetstream subscription alive.
2+##
3+## `jetstream.nim` reads a stream; this survives it ending. Three concerns,
4+## all of which are about not losing events rather than about uptime:
5+##
6+## 1. **Resume from a cursor, rewound.** Cursors are persisted in batches, so
7+## the last one written is always slightly behind the last one *seen*. On
8+## reconnect the consumer rewinds past that gap and accepts re-delivery --
9+## duplicate events are cheap when handling is idempotent, and a hole in
10+## the record is not recoverable.
11+##
12+## 2. **Rotate connections on a timer.** Server-side filters are fixed for the
13+## life of a connection, so a subscription that never reconnects never
14+## learns about newly-known DIDs. Glean's Go consumer rotates every 15
15+## minutes for exactly this reason.
16+##
17+## 3. **Back off, with jitter.** The Go consumer waits a flat 5s between
18+## attempts, which is fine against one healthy server and unkind to one
19+## that is down. This grows the delay to a cap, and jitters it so a fleet
20+## of consumers does not reconnect in lockstep.
21+
22+import std/[math, random]
23+import pkg/chronicles
24+import pkg/chronos
25+import pkg/websock/websock
26+import ./jetstream
27+
28+type
29+ BackoffConfig* = object
30+ initial*: Duration
31+ maxDelay*: Duration
32+ factor*: float
33+ jitter*: float ## 0.0-1.0 fraction of the delay to randomise
34+
35+ SupervisorConfig* = object
36+ jetstream*: JetstreamConfig
37+ backoff*: BackoffConfig
38+ rewind*: Duration
39+ ## How far behind the persisted cursor to resume. Covers the window
40+ ## between the last cursor flush and the disconnection.
41+ rotateEvery*: Duration
42+ ## Force a reconnect this often so filter changes take effect. Zero
43+ ## disables rotation.
44+
45+ CursorStore* = ref object of RootObj ## Override to persist across restarts.
46+ cursor*: int64
47+
48+ Supervisor* = ref object
49+ cfg*: SupervisorConfig
50+ store*: CursorStore
51+ handler*: EventHandler
52+ running*: bool
53+ reconnects*: int ## observable for tests and metrics
54+
55+# Both are `raises: []` on purpose. Saving a cursor happens on the event path,
56+# and a persistent store that throws -- a locked database, a full disk -- must
57+# not tear down the subscription. An implementation that can fail is expected
58+# to swallow and log, and to lose at most the rewind window.
59+method load*(s: CursorStore): int64 {.base, gcsafe, raises: [].} =
60+ s.cursor
61+method save*(s: CursorStore, cursor: int64) {.base, gcsafe, raises: [].} =
62+ s.cursor = cursor
63+
64+const
65+ DefaultBackoff* = BackoffConfig(
66+ initial: 1.seconds, maxDelay: 60.seconds, factor: 2.0, jitter: 0.2)
67+ DefaultRewind* = 5.seconds
68+ DefaultRotateEvery* = 15.minutes
69+
70+proc newSupervisorConfig*(js: JetstreamConfig): SupervisorConfig =
71+ SupervisorConfig(
72+ jetstream: js,
73+ backoff: DefaultBackoff,
74+ rewind: DefaultRewind,
75+ rotateEvery: DefaultRotateEvery,
76+ )
77+
78+proc delayFor*(b: BackoffConfig, attempt: int, rand01: float = -1.0): Duration =
79+ ## Delay before attempt `attempt` (0 = first retry).
80+ ##
81+ ## `rand01` is injectable so the jitter can be tested deterministically;
82+ ## left at -1 it draws from the global RNG.
83+ ##
84+ ## The first retry is jittered like every other. Short-circuiting it to the
85+ ## bare initial delay would leave the one moment a herd is most likely --
86+ ## every consumer dropped at once by the same server restart -- as the one
87+ ## moment they all wake together.
88+ let
89+ exp = max(attempt, 0).float
90+ grown = b.initial.milliseconds.float * pow(b.factor, exp)
91+ capped = min(grown, b.maxDelay.milliseconds.float)
92+ if b.jitter <= 0:
93+ return milliseconds(capped.int64)
94+ # Jitter downward only, so the cap stays a real ceiling.
95+ let r = if rand01 >= 0: rand01 else: rand(1.0)
96+ let jittered = capped * (1.0 - b.jitter * r)
97+ milliseconds(max(jittered.int64, 1'i64))
98+
99+proc rewoundCursor*(store: CursorStore, rewind: Duration): int64 =
100+ ## The persisted cursor, moved back to cover the unflushed window. Zero
101+ ## (start live) stays zero -- there is nothing to rewind to.
102+ let c = store.load()
103+ if c <= 0: return 0
104+ max(c - rewind.microseconds, 0'i64)
105+
106+proc newSupervisor*(cfg: SupervisorConfig, handler: EventHandler,
107+ store: CursorStore = nil): Supervisor =
108+ Supervisor(
109+ cfg: cfg,
110+ store: if store != nil: store else: CursorStore(),
111+ handler: handler,
112+ )
113+
114+proc runOnce(sup: Supervisor) {.async.} =
115+ ## One connection, read until it ends or the rotation timer fires.
116+ var cfg = sup.cfg.jetstream
117+ cfg.cursor = rewoundCursor(sup.store, sup.cfg.rewind)
118+
119+ let ws = await connect(cfg)
120+ defer:
121+ try: await ws.close()
122+ except CatchableError: discard
123+
124+ # Record every event's cursor, so a drop resumes near where it stopped.
125+ let store = sup.store
126+ let inner = sup.handler
127+ proc onEvent(ev: Event) {.gcsafe, raises: [].} =
128+ if ev.timeUs > 0:
129+ store.save(ev.timeUs)
130+ inner(ev)
131+
132+ let pump = readEvents(ws, onEvent)
133+ if sup.cfg.rotateEvery > ZeroDuration:
134+ if not await pump.withTimeout(sup.cfg.rotateEvery):
135+ # Rotation is a normal end to a connection, not a failure.
136+ await pump.cancelAndWait()
137+ else:
138+ await pump
139+
140+proc run*(sup: Supervisor) {.async.} =
141+ ## Reconnect forever. Stop by setting `running` to false.
142+ sup.running = true
143+ var attempt = 0
144+ while sup.running:
145+ try:
146+ await sup.runOnce()
147+ # A clean end (rotation, or the server closing) is not a failure, so
148+ # the next attempt starts from the bottom of the backoff curve.
149+ attempt = 0
150+ except CancelledError as e:
151+ raise e
152+ except CatchableError as e:
153+ inc sup.reconnects
154+ warn "jetstream connection lost", err = e.msg, attempt = attempt
155+ let delay = delayFor(sup.cfg.backoff, attempt)
156+ inc attempt
157+ await sleepAsync(delay)
158+ continue
159+
160+ if not sup.running:
161+ break
162+ # Even a clean rotation waits a moment, so a server that closes
163+ # immediately cannot spin this loop.
164+ await sleepAsync(sup.cfg.backoff.initial)
165+
166+proc stop*(sup: Supervisor) =
167+ sup.running = false
new file mode 100644
@@ -0,0 +1,167 @@
1+## Keeping a Jetstream subscription alive.
2+##
3+## `jetstream.nim` reads a stream; this survives it ending. Three concerns,
4+## all of which are about not losing events rather than about uptime:
5+##
6+## 1. **Resume from a cursor, rewound.** Cursors are persisted in batches, so
7+## the last one written is always slightly behind the last one *seen*. On
8+## reconnect the consumer rewinds past that gap and accepts re-delivery --
9+## duplicate events are cheap when handling is idempotent, and a hole in
10+## the record is not recoverable.
11+##
12+## 2. **Rotate connections on a timer.** Server-side filters are fixed for the
13+## life of a connection, so a subscription that never reconnects never
14+## learns about newly-known DIDs. Glean's Go consumer rotates every 15
15+## minutes for exactly this reason.
16+##
17+## 3. **Back off, with jitter.** The Go consumer waits a flat 5s between
18+## attempts, which is fine against one healthy server and unkind to one
19+## that is down. This grows the delay to a cap, and jitters it so a fleet
20+## of consumers does not reconnect in lockstep.
21+
22+import std/[math, random]
23+import pkg/chronicles
24+import pkg/chronos
25+import pkg/websock/websock
26+import ./jetstream
27+
28+type
29+ BackoffConfig* = object
30+ initial*: Duration
31+ maxDelay*: Duration
32+ factor*: float
33+ jitter*: float ## 0.0-1.0 fraction of the delay to randomise
34+
35+ SupervisorConfig* = object
36+ jetstream*: JetstreamConfig
37+ backoff*: BackoffConfig
38+ rewind*: Duration
39+ ## How far behind the persisted cursor to resume. Covers the window
40+ ## between the last cursor flush and the disconnection.
41+ rotateEvery*: Duration
42+ ## Force a reconnect this often so filter changes take effect. Zero
43+ ## disables rotation.
44+
45+ CursorStore* = ref object of RootObj ## Override to persist across restarts.
46+ cursor*: int64
47+
48+ Supervisor* = ref object
49+ cfg*: SupervisorConfig
50+ store*: CursorStore
51+ handler*: EventHandler
52+ running*: bool
53+ reconnects*: int ## observable for tests and metrics
54+
55+# Both are `raises: []` on purpose. Saving a cursor happens on the event path,
56+# and a persistent store that throws -- a locked database, a full disk -- must
57+# not tear down the subscription. An implementation that can fail is expected
58+# to swallow and log, and to lose at most the rewind window.
59+method load*(s: CursorStore): int64 {.base, gcsafe, raises: [].} =
60+ s.cursor
61+method save*(s: CursorStore, cursor: int64) {.base, gcsafe, raises: [].} =
62+ s.cursor = cursor
63+
64+const
65+ DefaultBackoff* = BackoffConfig(
66+ initial: 1.seconds, maxDelay: 60.seconds, factor: 2.0, jitter: 0.2)
67+ DefaultRewind* = 5.seconds
68+ DefaultRotateEvery* = 15.minutes
69+
70+proc newSupervisorConfig*(js: JetstreamConfig): SupervisorConfig =
71+ SupervisorConfig(
72+ jetstream: js,
73+ backoff: DefaultBackoff,
74+ rewind: DefaultRewind,
75+ rotateEvery: DefaultRotateEvery,
76+ )
77+
78+proc delayFor*(b: BackoffConfig, attempt: int, rand01: float = -1.0): Duration =
79+ ## Delay before attempt `attempt` (0 = first retry).
80+ ##
81+ ## `rand01` is injectable so the jitter can be tested deterministically;
82+ ## left at -1 it draws from the global RNG.
83+ ##
84+ ## The first retry is jittered like every other. Short-circuiting it to the
85+ ## bare initial delay would leave the one moment a herd is most likely --
86+ ## every consumer dropped at once by the same server restart -- as the one
87+ ## moment they all wake together.
88+ let
89+ exp = max(attempt, 0).float
90+ grown = b.initial.milliseconds.float * pow(b.factor, exp)
91+ capped = min(grown, b.maxDelay.milliseconds.float)
92+ if b.jitter <= 0:
93+ return milliseconds(capped.int64)
94+ # Jitter downward only, so the cap stays a real ceiling.
95+ let r = if rand01 >= 0: rand01 else: rand(1.0)
96+ let jittered = capped * (1.0 - b.jitter * r)
97+ milliseconds(max(jittered.int64, 1'i64))
98+
99+proc rewoundCursor*(store: CursorStore, rewind: Duration): int64 =
100+ ## The persisted cursor, moved back to cover the unflushed window. Zero
101+ ## (start live) stays zero -- there is nothing to rewind to.
102+ let c = store.load()
103+ if c <= 0: return 0
104+ max(c - rewind.microseconds, 0'i64)
105+
106+proc newSupervisor*(cfg: SupervisorConfig, handler: EventHandler,
107+ store: CursorStore = nil): Supervisor =
108+ Supervisor(
109+ cfg: cfg,
110+ store: if store != nil: store else: CursorStore(),
111+ handler: handler,
112+ )
113+
114+proc runOnce(sup: Supervisor) {.async.} =
115+ ## One connection, read until it ends or the rotation timer fires.
116+ var cfg = sup.cfg.jetstream
117+ cfg.cursor = rewoundCursor(sup.store, sup.cfg.rewind)
118+
119+ let ws = await connect(cfg)
120+ defer:
121+ try: await ws.close()
122+ except CatchableError: discard
123+
124+ # Record every event's cursor, so a drop resumes near where it stopped.
125+ let store = sup.store
126+ let inner = sup.handler
127+ proc onEvent(ev: Event) {.gcsafe, raises: [].} =
128+ if ev.timeUs > 0:
129+ store.save(ev.timeUs)
130+ inner(ev)
131+
132+ let pump = readEvents(ws, onEvent)
133+ if sup.cfg.rotateEvery > ZeroDuration:
134+ if not await pump.withTimeout(sup.cfg.rotateEvery):
135+ # Rotation is a normal end to a connection, not a failure.
136+ await pump.cancelAndWait()
137+ else:
138+ await pump
139+
140+proc run*(sup: Supervisor) {.async.} =
141+ ## Reconnect forever. Stop by setting `running` to false.
142+ sup.running = true
143+ var attempt = 0
144+ while sup.running:
145+ try:
146+ await sup.runOnce()
147+ # A clean end (rotation, or the server closing) is not a failure, so
148+ # the next attempt starts from the bottom of the backoff curve.
149+ attempt = 0
150+ except CancelledError as e:
151+ raise e
152+ except CatchableError as e:
153+ inc sup.reconnects
154+ warn "jetstream connection lost", err = e.msg, attempt = attempt
155+ let delay = delayFor(sup.cfg.backoff, attempt)
156+ inc attempt
157+ await sleepAsync(delay)
158+ continue
159+
160+ if not sup.running:
161+ break
162+ # Even a clean rotation waits a moment, so a server that closes
163+ # immediately cannot spin this loop.
164+ await sleepAsync(sup.cfg.backoff.initial)
165+
166+proc stop*(sup: Supervisor) =
167+ sup.running = false
modified spike/nim/readme.md +32 -2
@@ -14,6 +14,7 @@ nim c -r dpop_probe.nim && ./dpop_probe | (cd verify && go run .) # crypto
1414 nim c -r -d:ssl oauth_probe.nim [handle] # the flow
1515 nim c -r -d:ssl xrpc_probe.nim # signing
1616 nim c -r -d:ssl jetstream_probe.nim # firehose
17+nim c -r -d:ssl reconnect_probe.nim # supervisor
1718 ```
1819
1920 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.
@@ -174,6 +175,37 @@ is configured and resolves it in the *importing* module's scope — hence the
174175 `import chronicles` in `jetstream.nim` that nothing appears to use, and the
175176 `chronicles_enabled=off` in `nim.cfg`.
176177
178+## 6. Reconnect and backoff — works
179+
180+`atproto/jetstream_run.nim` keeps a subscription alive. Three concerns, all
181+about not losing events rather than about uptime:
182+
183+- **Resume from a rewound cursor.** Cursors are persisted in batches, so the
184+ last one written always trails the last one seen. Reconnecting rewinds past
185+ that gap and accepts re-delivery: duplicates are cheap when handling is
186+ idempotent, a hole in the record is not.
187+- **Rotate connections on a timer.** Server-side filters are fixed for the
188+ life of a connection, so a subscription that never reconnects never learns
189+ about newly-known DIDs. Glean's Go consumer rotates every 15 minutes for
190+ this reason, not for robustness.
191+- **Back off with jitter.** The Go consumer waits a flat 5s, which is fine
192+ against one healthy server and unkind to one that is down.
193+
194+A healthy server proves nothing here, so `reconnect_probe.nim` runs a local
195+websocket server that hangs up after three frames. Over two seconds the
196+supervisor made 12 connections and received 36 events across them.
197+
198+The probe checks that the cursor **reaches the server**, not merely that it
199+advances locally: a consumer that tracks a cursor it never sends looks
200+identical from the inside and silently loses the gap on every reconnect. 11
201+of 12 connections carried one, and the first correctly did not — it starts
202+live.
203+
204+One bug worth recording. `delayFor` short-circuited `attempt <= 0` straight
205+to the initial delay, skipping jitter on the first retry. That is precisely
206+the moment jitter exists for: one server restart drops every consumer at
207+once, and un-jittered first retries bring them all back simultaneously.
208+
177209 ## What this does and does not prove
178210
179211 Proven: the cryptography, the storage layer, and the OAuth flow up to user
@@ -186,8 +218,6 @@ Not proven, and still ahead:
186218 and it also covers the `invalid_token` refresh-and-retry branch of
187219 `xrpc.request`, which is currently unexercised.
188220 - **Session persistence** across restarts.
189-- **Reconnect and backoff** for a long-lived Jetstream subscription; the
190- consumer reads a stream but does not yet survive the connection dropping.
191221
192222 ### Not needed: DAG-CBOR, CAR and the MST
193223
@@ -14,6 +14,7 @@ nim c -r dpop_probe.nim && ./dpop_probe | (cd verify && go run .) # crypto
14 nim c -r -d:ssl oauth_probe.nim [handle] # the flow14 nim c -r -d:ssl oauth_probe.nim [handle] # the flow
15 nim c -r -d:ssl xrpc_probe.nim # signing15 nim c -r -d:ssl xrpc_probe.nim # signing
16 nim c -r -d:ssl jetstream_probe.nim # firehose16 nim c -r -d:ssl jetstream_probe.nim # firehose
17+nim c -r -d:ssl reconnect_probe.nim # supervisor
17 ```18 ```
18 19
19 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.20 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.
@@ -174,6 +175,37 @@ is configured and resolves it in the *importing* module's scope — hence the
174 `import chronicles` in `jetstream.nim` that nothing appears to use, and the175 `import chronicles` in `jetstream.nim` that nothing appears to use, and the
175 `chronicles_enabled=off` in `nim.cfg`.176 `chronicles_enabled=off` in `nim.cfg`.
176 177
178+## 6. Reconnect and backoff — works
179+
180+`atproto/jetstream_run.nim` keeps a subscription alive. Three concerns, all
181+about not losing events rather than about uptime:
182+
183+- **Resume from a rewound cursor.** Cursors are persisted in batches, so the
184+ last one written always trails the last one seen. Reconnecting rewinds past
185+ that gap and accepts re-delivery: duplicates are cheap when handling is
186+ idempotent, a hole in the record is not.
187+- **Rotate connections on a timer.** Server-side filters are fixed for the
188+ life of a connection, so a subscription that never reconnects never learns
189+ about newly-known DIDs. Glean's Go consumer rotates every 15 minutes for
190+ this reason, not for robustness.
191+- **Back off with jitter.** The Go consumer waits a flat 5s, which is fine
192+ against one healthy server and unkind to one that is down.
193+
194+A healthy server proves nothing here, so `reconnect_probe.nim` runs a local
195+websocket server that hangs up after three frames. Over two seconds the
196+supervisor made 12 connections and received 36 events across them.
197+
198+The probe checks that the cursor **reaches the server**, not merely that it
199+advances locally: a consumer that tracks a cursor it never sends looks
200+identical from the inside and silently loses the gap on every reconnect. 11
201+of 12 connections carried one, and the first correctly did not — it starts
202+live.
203+
204+One bug worth recording. `delayFor` short-circuited `attempt <= 0` straight
205+to the initial delay, skipping jitter on the first retry. That is precisely
206+the moment jitter exists for: one server restart drops every consumer at
207+once, and un-jittered first retries bring them all back simultaneously.
208+
177 ## What this does and does not prove209 ## What this does and does not prove
178 210
179 Proven: the cryptography, the storage layer, and the OAuth flow up to user211 Proven: the cryptography, the storage layer, and the OAuth flow up to user
@@ -186,8 +218,6 @@ Not proven, and still ahead:
186 and it also covers the `invalid_token` refresh-and-retry branch of218 and it also covers the `invalid_token` refresh-and-retry branch of
187 `xrpc.request`, which is currently unexercised.219 `xrpc.request`, which is currently unexercised.
188 - **Session persistence** across restarts.220 - **Session persistence** across restarts.
189-- **Reconnect and backoff** for a long-lived Jetstream subscription; the
190- consumer reads a stream but does not yet survive the connection dropping.
191 221
192 ### Not needed: DAG-CBOR, CAR and the MST222 ### Not needed: DAG-CBOR, CAR and the MST
193 223
added spike/nim/reconnect_probe.nim +182 -0
new file mode 100644
@@ -0,0 +1,182 @@
1+## Checks the reconnect/backoff supervisor.
2+##
3+## nim c -r -d:ssl reconnect_probe.nim
4+##
5+## The backoff curve and cursor arithmetic are pure and tested directly. The
6+## reconnect behaviour itself is tested against a local websocket server that
7+## is deliberately hostile -- it drops every connection after a few frames --
8+## because a consumer that "works" against a healthy server proves nothing
9+## about the case this module exists for.
10+
11+import std/[json, sequtils, strformat, strutils]
12+import pkg/chronicles
13+import pkg/chronos
14+import pkg/websock/websock
15+import atproto/[jetstream, jetstream_run]
16+
17+var failures = 0
18+
19+proc report(name: string, ok: bool, detail = "") =
20+ if not ok: inc failures
21+ let label = if ok: "PASS" else: "FAIL"
22+ if detail.len > 0: echo &" {label} {name} -- {detail}"
23+ else: echo &" {label} {name}"
24+
25+# --- a server that hangs up ------------------------------------------------
26+
27+type FlakyServer = ref object
28+ http: HttpServer
29+ connections*: int
30+ framesPerConn*: int
31+ cursorsSeen*: seq[string] ## the cursor query param of each connection
32+
33+proc start(framesPerConn: int): FlakyServer =
34+ let srv = FlakyServer(framesPerConn: framesPerConn)
35+
36+ proc handle(request: HttpRequest) {.async, gcsafe.} =
37+ srv.connections.inc
38+ srv.cursorsSeen.add request.uri.query
39+ let server = WSServer.new(protos = ["", "jetstream"])
40+ let ws = await server.handleRequest(request)
41+ # Send a few well-formed events, then drop the connection mid-stream.
42+ for i in 0 ..< srv.framesPerConn:
43+ let ev = %*{
44+ "did": "did:plc:probe",
45+ "time_us": 1_700_000_000_000_000'i64 + srv.connections * 1000 + i,
46+ "kind": "commit",
47+ "commit": {"rev": "r", "operation": "create",
48+ "collection": "app.test.item", "rkey": $i,
49+ "record": {"n": i}},
50+ }
51+ await ws.send($ev)
52+ await ws.stream.closeWait()
53+
54+ srv.http = HttpServer.create(initTAddress("127.0.0.1:0"), handle)
55+ srv.http.start()
56+ srv
57+
58+proc port(s: FlakyServer): int = s.http.localAddress.port.int
59+
60+# --- checks ----------------------------------------------------------------
61+
62+proc main() {.async.} =
63+ echo "Jetstream reconnect probe"
64+ echo ""
65+
66+ # --- backoff curve ------------------------------------------------------
67+ block:
68+ let b = BackoffConfig(initial: 1.seconds, maxDelay: 30.seconds,
69+ factor: 2.0, jitter: 0.0)
70+ report("first retry uses the initial delay",
71+ delayFor(b, 0) == 1.seconds, $delayFor(b, 0).milliseconds & "ms")
72+ report("the first retry is jittered too (herd control)",
73+ delayFor(BackoffConfig(initial: 1.seconds, maxDelay: 10.seconds,
74+ factor: 2.0, jitter: 0.5),
75+ 0, rand01 = 1.0) == 500.milliseconds)
76+ report("delay grows geometrically",
77+ delayFor(b, 1) == 2.seconds and delayFor(b, 2) == 4.seconds,
78+ &"{delayFor(b, 1).milliseconds}ms, {delayFor(b, 2).milliseconds}ms")
79+ report("delay is capped",
80+ delayFor(b, 20) == 30.seconds, $delayFor(b, 20).milliseconds & "ms")
81+
82+ block:
83+ # Jitter must reduce, never exceed, so the cap stays a real ceiling.
84+ let b = BackoffConfig(initial: 1.seconds, maxDelay: 10.seconds,
85+ factor: 2.0, jitter: 0.5)
86+ let atMax = delayFor(b, 0, rand01 = 0.0)
87+ let atMin = delayFor(b, 0, rand01 = 1.0)
88+ report("jitter only shortens the delay",
89+ atMax == 1.seconds and atMin == 500.milliseconds,
90+ &"{atMin.milliseconds}..{atMax.milliseconds}ms")
91+ report("a jittered capped delay never exceeds the cap",
92+ delayFor(b, 20, rand01 = 0.0) <= 10.seconds)
93+
94+ # --- cursor rewind ------------------------------------------------------
95+ block:
96+ let store = CursorStore(cursor: 1_700_000_000_000_000'i64)
97+ let c = rewoundCursor(store, 5.seconds)
98+ report("resume rewinds past the unflushed window",
99+ c == 1_700_000_000_000_000'i64 - 5_000_000,
100+ &"{store.cursor} -> {c}")
101+
102+ # Starting live must stay live; rewinding zero would ask for 1970.
103+ report("no cursor stays live", rewoundCursor(CursorStore(), 5.seconds) == 0)
104+
105+ let young = CursorStore(cursor: 1000)
106+ report("rewind clamps at zero", rewoundCursor(young, 5.seconds) == 0)
107+
108+ # --- reconnect against a server that hangs up ---------------------------
109+ let server = start(framesPerConn = 3)
110+ var received: seq[Event]
111+
112+ proc onEvent(ev: Event) {.gcsafe, raises: [].} =
113+ received.add ev
114+
115+ var cfg = newSupervisorConfig(JetstreamConfig(
116+ url: &"ws://127.0.0.1:{server.port}",
117+ wantedCollections: @["app.test.item"],
118+ ))
119+ # Tight timings so the probe finishes; rotation off, so every reconnect
120+ # here is a genuine recovery from the server hanging up.
121+ cfg.backoff = BackoffConfig(initial: 50.milliseconds,
122+ maxDelay: 200.milliseconds,
123+ factor: 2.0, jitter: 0.1)
124+ cfg.rotateEvery = ZeroDuration
125+ cfg.rewind = 0.seconds
126+
127+ let sup = newSupervisor(cfg, onEvent)
128+ let runner = sup.run()
129+ await sleepAsync(2.seconds)
130+ sup.stop()
131+ await runner.cancelAndWait()
132+
133+ report("survived the server hanging up", server.connections > 1,
134+ &"{server.connections} connections")
135+ report("kept receiving across reconnects", received.len > 3,
136+ &"{received.len} events over {server.connections} connections")
137+ report("events arrived from more than one connection",
138+ received.len >= server.connections,
139+ &"{received.len} events")
140+
141+ # The cursor must advance across connections, or a restart would replay
142+ # from the beginning forever.
143+ report("cursor advanced to the newest event seen",
144+ sup.store.load() == received[^1].timeUs,
145+ &"{sup.store.load()}")
146+
147+ # And it must actually reach the server. Tracking a cursor that is never
148+ # sent looks identical from the client side and loses every event in the
149+ # gap on each reconnect.
150+ block:
151+ let withCursor = server.cursorsSeen.filterIt("cursor=" in it)
152+ report("reconnects send a cursor",
153+ withCursor.len > 0,
154+ &"{withCursor.len} of {server.cursorsSeen.len} connections")
155+ report("the first connection has no cursor (starts live)",
156+ server.cursorsSeen.len > 0 and "cursor=" notin server.cursorsSeen[0])
157+ if withCursor.len >= 2:
158+ proc cursorOf(q: string): int64 =
159+ for part in q.split('&'):
160+ if part.startsWith("cursor="):
161+ return try: parseBiggestInt(part[7 .. ^1]).int64
162+ except ValueError: 0
163+ 0
164+ report("the cursor advances between reconnects",
165+ cursorOf(withCursor[^1]) > cursorOf(withCursor[0]),
166+ &"{cursorOf(withCursor[0])} -> {cursorOf(withCursor[^1])}")
167+
168+ # Stop accepting before closing: a connection still being torn down by the
169+ # supervisor would otherwise keep closeWait pending indefinitely.
170+ server.http.stop()
171+ if not await server.http.closeWait().withTimeout(5.seconds):
172+ report("server shut down cleanly", false, "closeWait timed out")
173+
174+ echo ""
175+ if failures == 0:
176+ echo "RESULT: reconnect, backoff and cursor resume all work."
177+ else:
178+ echo &"RESULT: {failures} check(s) failed."
179+ quit 1
180+
181+when isMainModule:
182+ waitFor main()
new file mode 100644
@@ -0,0 +1,182 @@
1+## Checks the reconnect/backoff supervisor.
2+##
3+## nim c -r -d:ssl reconnect_probe.nim
4+##
5+## The backoff curve and cursor arithmetic are pure and tested directly. The
6+## reconnect behaviour itself is tested against a local websocket server that
7+## is deliberately hostile -- it drops every connection after a few frames --
8+## because a consumer that "works" against a healthy server proves nothing
9+## about the case this module exists for.
10+
11+import std/[json, sequtils, strformat, strutils]
12+import pkg/chronicles
13+import pkg/chronos
14+import pkg/websock/websock
15+import atproto/[jetstream, jetstream_run]
16+
17+var failures = 0
18+
19+proc report(name: string, ok: bool, detail = "") =
20+ if not ok: inc failures
21+ let label = if ok: "PASS" else: "FAIL"
22+ if detail.len > 0: echo &" {label} {name} -- {detail}"
23+ else: echo &" {label} {name}"
24+
25+# --- a server that hangs up ------------------------------------------------
26+
27+type FlakyServer = ref object
28+ http: HttpServer
29+ connections*: int
30+ framesPerConn*: int
31+ cursorsSeen*: seq[string] ## the cursor query param of each connection
32+
33+proc start(framesPerConn: int): FlakyServer =
34+ let srv = FlakyServer(framesPerConn: framesPerConn)
35+
36+ proc handle(request: HttpRequest) {.async, gcsafe.} =
37+ srv.connections.inc
38+ srv.cursorsSeen.add request.uri.query
39+ let server = WSServer.new(protos = ["", "jetstream"])
40+ let ws = await server.handleRequest(request)
41+ # Send a few well-formed events, then drop the connection mid-stream.
42+ for i in 0 ..< srv.framesPerConn:
43+ let ev = %*{
44+ "did": "did:plc:probe",
45+ "time_us": 1_700_000_000_000_000'i64 + srv.connections * 1000 + i,
46+ "kind": "commit",
47+ "commit": {"rev": "r", "operation": "create",
48+ "collection": "app.test.item", "rkey": $i,
49+ "record": {"n": i}},
50+ }
51+ await ws.send($ev)
52+ await ws.stream.closeWait()
53+
54+ srv.http = HttpServer.create(initTAddress("127.0.0.1:0"), handle)
55+ srv.http.start()
56+ srv
57+
58+proc port(s: FlakyServer): int = s.http.localAddress.port.int
59+
60+# --- checks ----------------------------------------------------------------
61+
62+proc main() {.async.} =
63+ echo "Jetstream reconnect probe"
64+ echo ""
65+
66+ # --- backoff curve ------------------------------------------------------
67+ block:
68+ let b = BackoffConfig(initial: 1.seconds, maxDelay: 30.seconds,
69+ factor: 2.0, jitter: 0.0)
70+ report("first retry uses the initial delay",
71+ delayFor(b, 0) == 1.seconds, $delayFor(b, 0).milliseconds & "ms")
72+ report("the first retry is jittered too (herd control)",
73+ delayFor(BackoffConfig(initial: 1.seconds, maxDelay: 10.seconds,
74+ factor: 2.0, jitter: 0.5),
75+ 0, rand01 = 1.0) == 500.milliseconds)
76+ report("delay grows geometrically",
77+ delayFor(b, 1) == 2.seconds and delayFor(b, 2) == 4.seconds,
78+ &"{delayFor(b, 1).milliseconds}ms, {delayFor(b, 2).milliseconds}ms")
79+ report("delay is capped",
80+ delayFor(b, 20) == 30.seconds, $delayFor(b, 20).milliseconds & "ms")
81+
82+ block:
83+ # Jitter must reduce, never exceed, so the cap stays a real ceiling.
84+ let b = BackoffConfig(initial: 1.seconds, maxDelay: 10.seconds,
85+ factor: 2.0, jitter: 0.5)
86+ let atMax = delayFor(b, 0, rand01 = 0.0)
87+ let atMin = delayFor(b, 0, rand01 = 1.0)
88+ report("jitter only shortens the delay",
89+ atMax == 1.seconds and atMin == 500.milliseconds,
90+ &"{atMin.milliseconds}..{atMax.milliseconds}ms")
91+ report("a jittered capped delay never exceeds the cap",
92+ delayFor(b, 20, rand01 = 0.0) <= 10.seconds)
93+
94+ # --- cursor rewind ------------------------------------------------------
95+ block:
96+ let store = CursorStore(cursor: 1_700_000_000_000_000'i64)
97+ let c = rewoundCursor(store, 5.seconds)
98+ report("resume rewinds past the unflushed window",
99+ c == 1_700_000_000_000_000'i64 - 5_000_000,
100+ &"{store.cursor} -> {c}")
101+
102+ # Starting live must stay live; rewinding zero would ask for 1970.
103+ report("no cursor stays live", rewoundCursor(CursorStore(), 5.seconds) == 0)
104+
105+ let young = CursorStore(cursor: 1000)
106+ report("rewind clamps at zero", rewoundCursor(young, 5.seconds) == 0)
107+
108+ # --- reconnect against a server that hangs up ---------------------------
109+ let server = start(framesPerConn = 3)
110+ var received: seq[Event]
111+
112+ proc onEvent(ev: Event) {.gcsafe, raises: [].} =
113+ received.add ev
114+
115+ var cfg = newSupervisorConfig(JetstreamConfig(
116+ url: &"ws://127.0.0.1:{server.port}",
117+ wantedCollections: @["app.test.item"],
118+ ))
119+ # Tight timings so the probe finishes; rotation off, so every reconnect
120+ # here is a genuine recovery from the server hanging up.
121+ cfg.backoff = BackoffConfig(initial: 50.milliseconds,
122+ maxDelay: 200.milliseconds,
123+ factor: 2.0, jitter: 0.1)
124+ cfg.rotateEvery = ZeroDuration
125+ cfg.rewind = 0.seconds
126+
127+ let sup = newSupervisor(cfg, onEvent)
128+ let runner = sup.run()
129+ await sleepAsync(2.seconds)
130+ sup.stop()
131+ await runner.cancelAndWait()
132+
133+ report("survived the server hanging up", server.connections > 1,
134+ &"{server.connections} connections")
135+ report("kept receiving across reconnects", received.len > 3,
136+ &"{received.len} events over {server.connections} connections")
137+ report("events arrived from more than one connection",
138+ received.len >= server.connections,
139+ &"{received.len} events")
140+
141+ # The cursor must advance across connections, or a restart would replay
142+ # from the beginning forever.
143+ report("cursor advanced to the newest event seen",
144+ sup.store.load() == received[^1].timeUs,
145+ &"{sup.store.load()}")
146+
147+ # And it must actually reach the server. Tracking a cursor that is never
148+ # sent looks identical from the client side and loses every event in the
149+ # gap on each reconnect.
150+ block:
151+ let withCursor = server.cursorsSeen.filterIt("cursor=" in it)
152+ report("reconnects send a cursor",
153+ withCursor.len > 0,
154+ &"{withCursor.len} of {server.cursorsSeen.len} connections")
155+ report("the first connection has no cursor (starts live)",
156+ server.cursorsSeen.len > 0 and "cursor=" notin server.cursorsSeen[0])
157+ if withCursor.len >= 2:
158+ proc cursorOf(q: string): int64 =
159+ for part in q.split('&'):
160+ if part.startsWith("cursor="):
161+ return try: parseBiggestInt(part[7 .. ^1]).int64
162+ except ValueError: 0
163+ 0
164+ report("the cursor advances between reconnects",
165+ cursorOf(withCursor[^1]) > cursorOf(withCursor[0]),
166+ &"{cursorOf(withCursor[0])} -> {cursorOf(withCursor[^1])}")
167+
168+ # Stop accepting before closing: a connection still being torn down by the
169+ # supervisor would otherwise keep closeWait pending indefinitely.
170+ server.http.stop()
171+ if not await server.http.closeWait().withTimeout(5.seconds):
172+ report("server shut down cleanly", false, "closeWait timed out")
173+
174+ echo ""
175+ if failures == 0:
176+ echo "RESULT: reconnect, backoff and cursor resume all work."
177+ else:
178+ echo &"RESULT: {failures} check(s) failed."
179+ quit 1
180+
181+when isMainModule:
182+ waitFor main()