Skip to main content

max / synckit

11.6 KB · 317 lines History Blame Raw
1 //! SSE push notification stream for real-time sync notifications.
2 //!
3 //! The server sends zero-data `event: changed` events over SSE whenever
4 //! another device pushes changes. The client should pull on each event.
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::RngExt;
13
14 use crate::error::{Result, SyncKitError};
15
16 use super::{SecretToken, SyncKitClient};
17
18 /// A stream of SSE "changed" notifications from the SyncKit server.
19 ///
20 /// Created by [`SyncKitClient::subscribe`]. Wraps a raw HTTP response byte
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).
28 pub struct SyncNotifyStream {
29 buffer: String,
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<SecretToken>,
35 reconnect_attempt: u32,
36 }
37
38 /// Maximum SSE buffer size (1 MB). If the server sends data without a
39 /// block terminator (`\n\n`) exceeding this limit, we treat it as an error
40 /// and close the stream to prevent unbounded memory growth.
41 const MAX_SSE_BUFFER: usize = 1024 * 1024;
42
43 /// Cap on the reconnect backoff delay.
44 const MAX_RECONNECT_DELAY: Duration = Duration::from_mins(1);
45
46 /// Maximum *consecutive* failed reconnect attempts before the stream gives up.
47 /// A single success resets the counter, so this only fires when the server has
48 /// been unreachable or erroring for a sustained stretch (with the 60s backoff
49 /// cap, several minutes). Without it, a permanently-degraded non-auth server
50 /// (persistent 500s, connection refused) is retried silently forever, the
51 /// caller is better told the stream is dead so it can re-subscribe later.
52 const MAX_RECONNECT_ATTEMPTS: u32 = 10;
53
54 /// Reconnect backoff: 1s, 2s, 4s, ... capped at [`MAX_RECONNECT_DELAY`], with
55 /// +/-20% jitter so a fleet of clients doesn't reconnect in lockstep.
56 fn reconnect_delay(attempt: u32) -> Duration {
57 let base = Duration::from_secs(1)
58 .saturating_mul(2u32.saturating_pow(attempt.min(6)))
59 .min(MAX_RECONNECT_DELAY);
60 let millis = base.as_millis() as u64;
61 let span = millis / 5;
62 let delta = rand::rng().random_range(0..=2 * span.max(1));
63 Duration::from_millis(millis.saturating_sub(span).saturating_add(delta))
64 }
65
66 impl SyncNotifyStream {
67 pub(crate) fn new(
68 http: reqwest::Client,
69 url: String,
70 token: Arc<SecretToken>,
71 response: reqwest::Response,
72 ) -> Self {
73 Self {
74 buffer: String::new(),
75 response: Some(response),
76 http,
77 url,
78 token,
79 reconnect_attempt: 0,
80 }
81 }
82
83 /// Re-establish the SSE connection after a drop. Loops with backoff on
84 /// transient failures; returns `false` (fatal) if the server rejects the
85 /// request for auth reasons (caller re-authenticates) or after
86 /// [`MAX_RECONNECT_ATTEMPTS`] consecutive failures (caller re-subscribes
87 /// later) rather than retrying a dead server forever.
88 async fn reconnect(&mut self) -> bool {
89 loop {
90 self.reconnect_attempt += 1;
91 if self.reconnect_attempt > MAX_RECONNECT_ATTEMPTS {
92 tracing::warn!(
93 attempts = self.reconnect_attempt - 1,
94 "SSE reconnect gave up after too many consecutive failures; stream closing"
95 );
96 return false;
97 }
98 tokio::time::sleep(reconnect_delay(self.reconnect_attempt)).await;
99
100 match self
101 .http
102 .get(&self.url)
103 .bearer_auth(&*self.token)
104 .send()
105 .await
106 {
107 Ok(resp) => {
108 let status = resp.status().as_u16();
109 if status == 401 || status == 403 {
110 tracing::warn!(status, "SSE reconnect rejected (auth); stream closing");
111 return false;
112 }
113 if status >= 400 {
114 tracing::debug!(
115 status,
116 attempt = self.reconnect_attempt,
117 "SSE reconnect got error status; backing off"
118 );
119 continue;
120 }
121 tracing::debug!(attempt = self.reconnect_attempt, "SSE reconnected");
122 self.reconnect_attempt = 0;
123 self.buffer.clear();
124 self.response = Some(resp);
125 return true;
126 }
127 Err(e) => {
128 tracing::debug!(
129 error = %e,
130 attempt = self.reconnect_attempt,
131 "SSE reconnect failed; backing off"
132 );
133 }
134 }
135 }
136 }
137
138 /// Wait for the next "changed" notification.
139 ///
140 /// Transparently reconnects across transient stream drops. Returns `Some(())`
141 /// when the server signals new changes. Returns `None` only on a fatal
142 /// condition: the reconnect is rejected for auth, or the stream violates the
143 /// SSE protocol (invalid UTF-8 / oversized block). Keepalive comments (lines
144 /// starting with `:`) are silently ignored.
145 pub async fn next_change(&mut self) -> Option<()> {
146 loop {
147 if self.response.is_none() && !self.reconnect().await {
148 return None;
149 }
150
151 // Try to extract a complete SSE block from the buffer
152 if let Some(pos) = self.buffer.find("\n\n") {
153 let block = self.buffer[..pos].to_string();
154 self.buffer = self.buffer[pos + 2..].to_string();
155
156 if parse_sse_block_is_changed(&block) {
157 return Some(());
158 }
159 // Not a "changed" event, skip and try next block
160 continue;
161 }
162
163 // Need more data from the stream
164 let chunk = self
165 .response
166 .as_mut()
167 .expect("connected above")
168 .chunk()
169 .await;
170 match chunk {
171 Ok(Some(chunk)) => {
172 match std::str::from_utf8(&chunk) {
173 Ok(text) => self.buffer.push_str(text),
174 Err(_) => {
175 tracing::warn!("SSE stream contained invalid UTF-8, closing");
176 return None;
177 }
178 }
179 if self.buffer.len() > MAX_SSE_BUFFER {
180 tracing::warn!(
181 "SSE buffer exceeded {MAX_SSE_BUFFER} bytes, closing stream"
182 );
183 return None;
184 }
185 }
186 // Stream ended or errored: drop it and reconnect on the next loop
187 // instead of ending, this is the auto-reconnect.
188 Ok(None) => {
189 tracing::debug!("SSE stream ended; will reconnect");
190 self.response = None;
191 }
192 Err(e) => {
193 tracing::debug!(error = %e, "SSE stream error; will reconnect");
194 self.response = None;
195 }
196 }
197 }
198 }
199 }
200
201 /// Parse an SSE block and return `true` if it contains `event: changed`.
202 ///
203 /// An SSE block is one or more lines separated by `\n`, terminated by `\n\n`.
204 /// Lines starting with `:` are comments (keepalive). The `event:` field
205 /// specifies the event type, `data:` the payload (ignored here).
206 fn parse_sse_block_is_changed(block: &str) -> bool {
207 for line in block.lines() {
208 let trimmed = line.trim();
209 // Skip comments
210 if trimmed.starts_with(':') {
211 continue;
212 }
213 if let Some(value) = trimmed.strip_prefix("event:")
214 && value.trim() == "changed"
215 {
216 return true;
217 }
218 }
219 false
220 }
221
222 impl SyncKitClient {
223 /// Open an SSE connection for real-time push notifications.
224 ///
225 /// Returns a [`SyncNotifyStream`] that yields `Some(())` each time the
226 /// server signals new changes are available. The caller should pull after
227 /// each notification to get the actual changes.
228 ///
229 /// The returned stream auto-reconnects with backoff across transient drops;
230 /// `next_change()` returns `None` only on a fatal condition (auth rejected on
231 /// reconnect, or a protocol violation), at which point the caller should
232 /// re-authenticate and subscribe again.
233 #[tracing::instrument(skip(self))]
234 pub async fn subscribe(&self) -> Result<SyncNotifyStream> {
235 let token = self.require_token()?;
236 let (app_id, _user_id) = self.require_session_ids()?;
237 let url = format!("{}?app_id={}", self.endpoints.subscribe, app_id);
238
239 let resp = self
240 .http_stream
241 .get(&url)
242 .bearer_auth(&*token)
243 .send()
244 .await
245 .map_err(SyncKitError::Http)?;
246
247 let status = resp.status().as_u16();
248 if status >= 400 {
249 let message = crate::client::helpers::read_text_capped(
250 resp,
251 crate::client::helpers::MAX_CONTROL_BODY_BYTES,
252 )
253 .await;
254 return Err(SyncKitError::Server {
255 status,
256 message,
257 retry_after_secs: None,
258 });
259 }
260
261 // Hand the stream everything it needs to reconnect itself.
262 Ok(SyncNotifyStream::new(
263 self.http_stream.clone(),
264 url,
265 token,
266 resp,
267 ))
268 }
269 }
270
271 #[cfg(test)]
272 mod tests {
273 use super::*;
274
275 #[test]
276 fn parse_changed_event() {
277 let block = "event: changed\ndata: {}";
278 assert!(parse_sse_block_is_changed(block));
279 }
280
281 #[test]
282 fn parse_ignores_keepalive_comments() {
283 let block = ": keepalive";
284 assert!(!parse_sse_block_is_changed(block));
285 }
286
287 #[test]
288 fn parse_handles_malformed_lines() {
289 let block = "garbled nonsense";
290 assert!(!parse_sse_block_is_changed(block));
291
292 let block2 = "";
293 assert!(!parse_sse_block_is_changed(block2));
294
295 let block3 = "event:other\ndata: test";
296 assert!(!parse_sse_block_is_changed(block3));
297 }
298
299 #[test]
300 fn parse_changed_with_extra_whitespace() {
301 let block = "event: changed \ndata: {}";
302 assert!(parse_sse_block_is_changed(block));
303 }
304
305 #[test]
306 fn reconnect_delay_grows_then_caps() {
307 // Within +/-20% jitter, the delay grows with the attempt and never
308 // exceeds the cap plus jitter headroom.
309 let d1 = reconnect_delay(1);
310 let d6 = reconnect_delay(6);
311 assert!(d1 >= Duration::from_millis(1600) && d1 <= Duration::from_millis(2400));
312 assert!(d6 <= MAX_RECONNECT_DELAY + Duration::from_secs(12));
313 // Deep attempts stay capped (exponent is clamped).
314 assert!(reconnect_delay(100) <= MAX_RECONNECT_DELAY + Duration::from_secs(12));
315 }
316 }
317