Skip to main content

max / makenotwork

synckit: auto-reconnect the SSE notification stream (audit Phase 5) Resilience remediation from the 2026-07-02 /audit arbiter: SyncNotifyStream now transparently reconnects across transient connection drops (exponential backoff, capped at 60s, +/-20% jitter) instead of ending on the first drop. SSE `changed` events are contentless pokes, so a reconnect needs no cursor. next_change() returns None only on a fatal condition now: the server rejecting the reconnect with 401/403 (caller must re-authenticate) or a protocol violation (invalid UTF-8, oversized block). The stream carries its own http_stream client, URL, and token to reconnect itself; consumer reconnect loops (e.g. GoingsOn's scheduler) stay compatible since None still means "fatal". Gate: 407 crate tests pass, cargo clippy --all-targets clean, GoingsOn cargo check clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-03 01:49 UTC
Commit: 04d605b421743a4e4e0ddcb5141b89ed02294c99
Parent: 2ed0884
1 file changed, +130 insertions, -17 deletions
M shared/synckit-client/src/client/subscribe.rs +130 -17
@@ -2,7 +2,14 @@
2 2 //!
3 3 //! The server sends zero-data `event: changed` events over SSE whenever
4 4 //! another device pushes changes. The client should pull on each event.
5 - //! No auto-reconnect — consumer apps handle reconnect in their scheduler.
5 + //! The stream auto-reconnects with backoff across transient drops; it only
6 + //! ends (`next_change` returns `None`) on a fatal condition — the server
7 + //! rejecting the reconnect for auth reasons, or a protocol violation.
8 +
9 + use std::sync::Arc;
10 + use std::time::Duration;
11 +
12 + use rand::Rng;
6 13
7 14 use crate::error::{Result, SyncKitError};
8 15
@@ -11,11 +18,21 @@ use super::SyncKitClient;
11 18 /// A stream of SSE "changed" notifications from the SyncKit server.
12 19 ///
13 20 /// Created by [`SyncKitClient::subscribe`]. Wraps a raw HTTP response byte
14 - /// stream and parses SSE events line by line. Yields `Some(())` for each
15 - /// `event: changed`, `None` when the stream ends or encounters an error.
21 + /// stream and parses SSE events line by line, yielding `Some(())` for each
22 + /// `event: changed`. When the underlying connection drops, it transparently
23 + /// reconnects (exponential backoff, capped, jittered) rather than ending —
24 + /// SSE `changed` events are contentless pokes, so a reconnect needs no cursor.
25 + /// `next_change` returns `None` only when reconnection is futile: the server
26 + /// rejects the request with `401`/`403` (the caller must re-authenticate), or
27 + /// the stream violates the protocol (invalid UTF-8, oversized block).
16 28 pub struct SyncNotifyStream {
17 29 buffer: String,
18 - response: reqwest::Response,
30 + /// `None` between a drop and the next successful reconnect.
31 + response: Option<reqwest::Response>,
32 + http: reqwest::Client,
33 + url: String,
34 + token: Arc<String>,
35 + reconnect_attempt: u32,
19 36 }
20 37
21 38 /// Maximum SSE buffer size (1 MB). If the server sends data without a
@@ -23,21 +40,92 @@ pub struct SyncNotifyStream {
23 40 /// and close the stream to prevent unbounded memory growth.
24 41 const MAX_SSE_BUFFER: usize = 1024 * 1024;
25 42
43 + /// Cap on the reconnect backoff delay.
44 + const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(60);
45 +
46 + /// Reconnect backoff: 1s, 2s, 4s, ... capped at [`MAX_RECONNECT_DELAY`], with
47 + /// +/-20% jitter so a fleet of clients doesn't reconnect in lockstep.
48 + fn reconnect_delay(attempt: u32) -> Duration {
49 + let base = Duration::from_secs(1)
50 + .saturating_mul(2u32.saturating_pow(attempt.min(6)))
51 + .min(MAX_RECONNECT_DELAY);
52 + let millis = base.as_millis() as u64;
53 + let span = millis / 5;
54 + let delta = rand::rng().random_range(0..=2 * span.max(1));
55 + Duration::from_millis(millis.saturating_sub(span).saturating_add(delta))
56 + }
57 +
26 58 impl SyncNotifyStream {
27 - pub(crate) fn new(response: reqwest::Response) -> Self {
59 + pub(crate) fn new(
60 + http: reqwest::Client,
61 + url: String,
62 + token: Arc<String>,
63 + response: reqwest::Response,
64 + ) -> Self {
28 65 Self {
29 66 buffer: String::new(),
30 - response,
67 + response: Some(response),
68 + http,
69 + url,
70 + token,
71 + reconnect_attempt: 0,
72 + }
73 + }
74 +
75 + /// Re-establish the SSE connection after a drop. Loops with backoff on
76 + /// transient failures; returns `false` (fatal) if the server rejects the
77 + /// request for auth reasons, so the caller stops and re-authenticates.
78 + async fn reconnect(&mut self) -> bool {
79 + loop {
80 + self.reconnect_attempt += 1;
81 + tokio::time::sleep(reconnect_delay(self.reconnect_attempt)).await;
82 +
83 + match self.http.get(&self.url).bearer_auth(&*self.token).send().await {
84 + Ok(resp) => {
85 + let status = resp.status().as_u16();
86 + if status == 401 || status == 403 {
87 + tracing::warn!(status, "SSE reconnect rejected (auth); stream closing");
88 + return false;
89 + }
90 + if status >= 400 {
91 + tracing::debug!(
92 + status,
93 + attempt = self.reconnect_attempt,
94 + "SSE reconnect got error status; backing off"
95 + );
96 + continue;
97 + }
98 + tracing::debug!(attempt = self.reconnect_attempt, "SSE reconnected");
99 + self.reconnect_attempt = 0;
100 + self.buffer.clear();
101 + self.response = Some(resp);
102 + return true;
103 + }
104 + Err(e) => {
105 + tracing::debug!(
106 + error = %e,
107 + attempt = self.reconnect_attempt,
108 + "SSE reconnect failed; backing off"
109 + );
110 + continue;
111 + }
112 + }
31 113 }
32 114 }
33 115
34 116 /// Wait for the next "changed" notification.
35 117 ///
36 - /// Returns `Some(())` when the server signals new changes are available.
37 - /// Returns `None` when the stream ends (server closed, network error).
38 - /// Keepalive comments (lines starting with `:`) are silently ignored.
118 + /// Transparently reconnects across transient stream drops. Returns `Some(())`
119 + /// when the server signals new changes. Returns `None` only on a fatal
120 + /// condition: the reconnect is rejected for auth, or the stream violates the
121 + /// SSE protocol (invalid UTF-8 / oversized block). Keepalive comments (lines
122 + /// starting with `:`) are silently ignored.
39 123 pub async fn next_change(&mut self) -> Option<()> {
40 124 loop {
125 + if self.response.is_none() && !self.reconnect().await {
126 + return None;
127 + }
128 +
41 129 // Try to extract a complete SSE block from the buffer
42 130 if let Some(pos) = self.buffer.find("\n\n") {
43 131 let block = self.buffer[..pos].to_string();
@@ -51,7 +139,8 @@ impl SyncNotifyStream {
51 139 }
52 140
53 141 // Need more data from the stream
54 - match self.response.chunk().await {
142 + let chunk = self.response.as_mut().expect("connected above").chunk().await;
143 + match chunk {
55 144 Ok(Some(chunk)) => {
56 145 match std::str::from_utf8(&chunk) {
57 146 Ok(text) => self.buffer.push_str(text),
@@ -65,10 +154,15 @@ impl SyncNotifyStream {
65 154 return None;
66 155 }
67 156 }
68 - Ok(None) => return None,
157 + // Stream ended or errored: drop it and reconnect on the next loop
158 + // instead of ending — this is the auto-reconnect.
159 + Ok(None) => {
160 + tracing::debug!("SSE stream ended; will reconnect");
161 + self.response = None;
162 + }
69 163 Err(e) => {
70 - tracing::debug!(error = %e, "SSE stream error, closing");
71 - return None;
164 + tracing::debug!(error = %e, "SSE stream error; will reconnect");
165 + self.response = None;
72 166 }
73 167 }
74 168 }
@@ -103,9 +197,10 @@ impl SyncKitClient {
103 197 /// server signals new changes are available. The caller should pull after
104 198 /// each notification to get the actual changes.
105 199 ///
106 - /// The stream does not auto-reconnect. If the connection drops,
107 - /// `next_change()` returns `None` and the caller should reconnect
108 - /// (typically after a short delay).
200 + /// The returned stream auto-reconnects with backoff across transient drops;
201 + /// `next_change()` returns `None` only on a fatal condition (auth rejected on
202 + /// reconnect, or a protocol violation), at which point the caller should
203 + /// re-authenticate and subscribe again.
109 204 #[tracing::instrument(skip(self))]
110 205 pub async fn subscribe(&self) -> Result<SyncNotifyStream> {
111 206 let token = self.require_token()?;
@@ -126,7 +221,13 @@ impl SyncKitClient {
126 221 return Err(SyncKitError::Server { status, message, retry_after_secs: None });
127 222 }
128 223
129 - Ok(SyncNotifyStream::new(resp))
224 + // Hand the stream everything it needs to reconnect itself.
225 + Ok(SyncNotifyStream::new(
226 + self.http_stream.clone(),
227 + url,
228 + token,
229 + resp,
230 + ))
130 231 }
131 232 }
132 233
@@ -163,4 +264,16 @@ mod tests {
163 264 let block = "event: changed \ndata: {}";
164 265 assert!(parse_sse_block_is_changed(block));
165 266 }
267 +
268 + #[test]
269 + fn reconnect_delay_grows_then_caps() {
270 + // Within +/-20% jitter, the delay grows with the attempt and never
271 + // exceeds the cap plus jitter headroom.
272 + let d1 = reconnect_delay(1);
273 + let d6 = reconnect_delay(6);
274 + assert!(d1 >= Duration::from_millis(1600) && d1 <= Duration::from_millis(2400));
275 + assert!(d6 <= MAX_RECONNECT_DELAY + Duration::from_secs(12));
276 + // Deep attempts stay capped (exponent is clamped).
277 + assert!(reconnect_delay(100) <= MAX_RECONNECT_DELAY + Duration::from_secs(12));
278 + }
166 279 }