Skip to main content

max / synckit

5.9 KB · 217 lines History Blame Raw
1 //! Concurrent use of one client: interleaved push/pull, parallel reads of the
2 //! session and key state, and the stress cases. These assert the absence of
3 //! panics and data corruption, not a particular interleaving.
4
5 use crate::common::*;
6
7 const PUSH_PATH: &str = "/api/v1/sync/push";
8 const PULL_PATH: &str = "/api/v1/sync/pull";
9
10 // ── Concurrent access ──
11
12 #[tokio::test]
13 async fn concurrent_push_pull_no_panics() {
14 let kit = MockKit::start().await;
15 kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;
16 kit.post(PULL_PATH)
17 .json(json!({
18 "changes": [],
19 "cursor": 0,
20 "has_more": false,
21 }))
22 .await;
23
24 let device_id = DeviceId::new(Uuid::new_v4());
25 let (client, _key) = kit.keyed();
26 let client = Arc::new(client);
27
28 let mut handles = Vec::new();
29 for _ in 0..4 {
30 let c = Arc::clone(&client);
31 let did = device_id;
32 handles.push(tokio::spawn(async move {
33 let _ = c.push(did, vec![]).await;
34 let _ = c.pull(did, 0).await;
35 }));
36 }
37
38 for h in handles {
39 h.await.unwrap(); // No panics
40 }
41 }
42
43 // ── Concurrent operations ──
44
45 #[tokio::test]
46 async fn concurrent_push_operations_no_data_corruption() {
47 let kit = MockKit::start().await;
48 kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;
49
50 let (client, _key) = kit.keyed();
51 let client = Arc::new(client);
52
53 let mut handles = Vec::new();
54 for i in 0..8 {
55 let c = Arc::clone(&client);
56 handles.push(tokio::spawn(async move {
57 let device_id = DeviceId::new(Uuid::new_v4());
58 let entry = ChangeEntry {
59 table: format!("table_{i}"),
60 op: ChangeOp::Insert,
61 row_id: format!("row_{i}"),
62 timestamp: Utc::now(),
63 hlc: Hlc::zero(DeviceId::nil()),
64 data: Some(json!({"index": i})),
65 extra: serde_json::Map::default(),
66 };
67 c.push(device_id, vec![entry]).await
68 }));
69 }
70
71 for h in handles {
72 let result = h.await.unwrap();
73 assert!(result.is_ok(), "Concurrent push should succeed: {result:?}");
74 }
75 }
76
77 #[tokio::test]
78 async fn concurrent_push_and_pull_interleaved() {
79 let kit = MockKit::start().await;
80 let device_id = DeviceId::new(Uuid::new_v4());
81
82 kit.post(PUSH_PATH).json(json!({"cursor": 10})).await;
83 kit.post(PULL_PATH)
84 .json(json!({
85 "changes": [],
86 "cursor": 10,
87 "has_more": false,
88 }))
89 .await;
90
91 let (client, _key) = kit.keyed();
92 let client = Arc::new(client);
93
94 let mut handles = Vec::new();
95 for i in 0..4 {
96 let c = Arc::clone(&client);
97 let did = device_id;
98 handles.push(tokio::spawn(async move {
99 // Alternate push and pull
100 if i % 2 == 0 {
101 c.push(did, vec![]).await.map(|_| ())
102 } else {
103 c.pull(did, 0).await.map(|_| ())
104 }
105 }));
106 }
107
108 for h in handles {
109 let result = h.await.unwrap();
110 assert!(
111 result.is_ok(),
112 "Interleaved push/pull should succeed: {result:?}"
113 );
114 }
115 }
116
117 // ── Concurrency stress tests ──
118
119 #[tokio::test]
120 async fn concurrent_session_info_reads() {
121 let kit = MockKit::start().await;
122 let client = Arc::new(kit.authed());
123
124 let mut handles = Vec::new();
125 for _ in 0..50 {
126 let c = Arc::clone(&client);
127 handles.push(tokio::spawn(async move { c.session_info() }));
128 }
129
130 for h in handles {
131 let info = h.await.unwrap();
132 assert!(
133 info.is_some(),
134 "All concurrent reads should see the session"
135 );
136 }
137 }
138
139 #[tokio::test]
140 async fn concurrent_has_master_key_reads() {
141 let kit = MockKit::start().await;
142 let (client, _key) = kit.keyed();
143 let client = Arc::new(client);
144
145 let mut handles = Vec::new();
146 for _ in 0..50 {
147 let c = Arc::clone(&client);
148 handles.push(tokio::spawn(async move { c.has_master_key() }));
149 }
150
151 for h in handles {
152 let has_key = h.await.unwrap();
153 assert!(has_key, "All concurrent reads should see the master key");
154 }
155 }
156
157 #[tokio::test]
158 async fn concurrent_status_checks() {
159 let kit = MockKit::start().await;
160 kit.get("/api/v1/sync/status")
161 .json(json!({"total_changes": 5, "latest_cursor": 3}))
162 .await;
163
164 let client = Arc::new(kit.authed());
165
166 let mut handles = Vec::new();
167 for _ in 0..20 {
168 let c = Arc::clone(&client);
169 handles.push(tokio::spawn(async move { c.status().await }));
170 }
171
172 for h in handles {
173 let result = h.await.unwrap();
174 assert!(
175 result.is_ok(),
176 "All concurrent status checks should succeed: {result:?}"
177 );
178 assert_eq!(result.unwrap().total_changes, 5);
179 }
180 }
181
182 #[tokio::test]
183 async fn concurrent_push_100_entries_each() {
184 let kit = MockKit::start().await;
185 kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;
186
187 let (client, _key) = kit.keyed();
188 let client = Arc::new(client);
189
190 let mut handles = Vec::new();
191 for batch in 0..4 {
192 let c = Arc::clone(&client);
193 handles.push(tokio::spawn(async move {
194 let changes: Vec<ChangeEntry> = (0..100)
195 .map(|i| ChangeEntry {
196 table: format!("batch_{batch}"),
197 op: ChangeOp::Insert,
198 row_id: format!("row_{i}"),
199 timestamp: Utc::now(),
200 hlc: Hlc::zero(DeviceId::nil()),
201 data: Some(json!({"index": i})),
202 extra: serde_json::Map::default(),
203 })
204 .collect();
205 c.push(DeviceId::new(Uuid::new_v4()), changes).await
206 }));
207 }
208
209 for h in handles {
210 let result = h.await.unwrap();
211 assert!(
212 result.is_ok(),
213 "Concurrent 100-entry push should succeed: {result:?}"
214 );
215 }
216 }
217