max / synckit

3.5 KB · 83 lines History Blame Raw

1	//! The SSE subscribe stream and its reconnect state machine.
2
3	use crate::common::*;
4
5	// ── SSE subscribe / reconnect state machine ──
6	//
7	// These drive the real `subscribe()` -> `SyncNotifyStream::next_change()` path
8	// against wiremock so the reconnect state machine (subscribe.rs) is exercised
9	// end to end: a live notification, a transparent reconnect across a stream
10	// drop, the fatal auth-rejection exit, and the give-up-after-N-failures cap.
11
12	const SUBSCRIBE_PATH: &str = "/api/v1/sync/subscribe";
13
14	/// One complete SSE "changed" block, body-terminated so the client sees a full
15	/// event and then end-of-stream.
16	fn sse_changed_block() -> &'static str {
17	"event: changed\n\n"
18	}
19
20	#[tokio::test]
21	async fn subscribe_yields_changed_event() {
22	let kit = MockKit::start().await;
23	kit.get(SUBSCRIBE_PATH).text(sse_changed_block()).await;
24
25	let client = kit.authed();
26	let mut stream = client.subscribe().await.expect("subscribe should succeed");
27
28	// The first block is delivered inside the initial response body, so this
29	// returns without any reconnect.
30	assert_eq!(stream.next_change().await, Some(()));
31	}
32
33	#[tokio::test]
34	async fn subscribe_reconnects_transparently_after_stream_drop() {
35	let kit = MockKit::start().await;
36	// Every connection serves one block then ends (Content-Length terminates the
37	// body). Consuming the first event, then reading past it, drops the stream
38	// and forces a reconnect that must transparently yield the next event.
39	kit.get(SUBSCRIBE_PATH).text(sse_changed_block()).await;
40
41	let client = kit.authed();
42	let mut stream = client.subscribe().await.expect("subscribe should succeed");
43
44	// #1 comes from the initial connection; #2 can only arrive after the stream
45	// drops (body EOF) and reconnect() re-establishes it.
46	assert_eq!(stream.next_change().await, Some(()));
47	assert_eq!(stream.next_change().await, Some(()));
48
49	// subscribe() opened one connection; the second event required at least one
50	// reconnect, so the server saw two or more subscribe requests.
51	let hits = kit.hits(SUBSCRIBE_PATH).await;
52	assert!(
53	hits >= 2,
54	"expected a reconnect (>=2 subscribe requests), got {hits}"
55	);
56	}
57
58	#[tokio::test]
59	async fn subscribe_stream_closes_on_auth_rejection() {
60	let kit = MockKit::start().await;
61	// First connection opens cleanly but carries no event and ends immediately,
62	// forcing a reconnect. The reconnect is rejected for auth -> fatal, so the
63	// stream ends with `None` rather than retrying.
64	kit.get(SUBSCRIBE_PATH).once().text("").await;
65	kit.get(SUBSCRIBE_PATH).code(401).text("unauthorized").await;
66
67	let client = kit.authed();
68	let mut stream = client.subscribe().await.expect("initial subscribe is 200");
69
70	// Empty body -> EOF -> reconnect -> 401 -> fatal -> None.
71	assert_eq!(stream.next_change().await, None);
72	}
73
74	// Note: the "give up after MAX_RECONNECT_ATTEMPTS consecutive failures" path is
75	// deliberately not covered end-to-end here. Exercising it against a live mock
76	// server would incur the real exponential backoff (growing to a 60s cap, minutes
77	// of wall-clock), and `tokio::time::pause()` cannot collapse it: the wiremock
78	// server runs on its own runtime, so a paused clock auto-advances past the
79	// cross-thread request to the nearest timer and the reconnect never completes.
80	// The two pieces of that path are covered separately: the backoff schedule by
81	// `reconnect_delay_grows_then_caps` (subscribe.rs), and the fatal-exit mechanics
82	// (returning `None` and clearing state) by the auth-rejection test above.
83