Skip to main content

max / makenotwork

14.6 KB · 461 lines History Blame Raw
1 //! DB-layer contract tests for the SyncKit change log, blobs and keys
2 //! (`db::synckit::{log, blobs, keys}`), which had none of their own.
3 //!
4 //! Its siblings (`db_synckit_rotation`, `db_synckit_groups`,
5 //! `db_synckit_invitations`) already pin the rotation state machine, the group
6 //! changelog and the invitation lifecycle at this layer;
7 //! `db_synckit_accounts_layer` covers devices, apps, subscriptions and the
8 //! audit log. What was reachable only through the HTTP workflows is the owner
9 //! scoping every one of these queries carries, the `app_id = $1 AND
10 //! user_id = $2` pair: a route test authenticates as one user, so it cannot
11 //! see a query that returns another user's rows.
12 //!
13 //! Deliberately not re-asserted here, because it is covered elsewhere: the
14 //! table and `since` pull filters (`synckit_selective`), the storage quota and
15 //! blob-delete paths through a real subscription (`synckit_paid_sync`,
16 //! `synckit_per_key_storage`), and compaction against a freshly registered
17 //! device (`synckit_paid_sync`).
18
19 use super::db_synckit_accounts_layer::seed_active_subscription;
20 use crate::harness::db::TestDb;
21 use crate::harness::seed_user;
22
23 use makenotwork::db::synckit;
24 use makenotwork::db::synckit::BlobConfirm;
25 use makenotwork::db::{SyncAppId, SyncDeviceId, SyncPlatform, UserId};
26 use uuid::Uuid;
27
28 /// Seed a sync app owned by `user`, with its usage row.
29 async fn seed_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
30 synckit::create_sync_app(pool, user, name, &format!("key_{name}_padding"), None, None)
31 .await
32 .expect("seed sync app")
33 .id
34 }
35
36 /// Mark an app first-party, so the end-user subscription model applies.
37 async fn make_internal(pool: &sqlx::PgPool, app: SyncAppId) {
38 sqlx::query("UPDATE sync_apps SET is_internal = true WHERE id = $1")
39 .bind(app)
40 .execute(pool)
41 .await
42 .expect("mark internal");
43 }
44
45 async fn seed_device(
46 pool: &sqlx::PgPool,
47 app: SyncAppId,
48 user: UserId,
49 name: &str,
50 ) -> SyncDeviceId {
51 synckit::upsert_sync_device(pool, app, user, name, SyncPlatform::Macos, None)
52 .await
53 .expect("seed device")
54 .id
55 }
56
57 /// One INSERT change tuple in the shape `push_sync_changes` expects.
58 fn change(
59 table: &str,
60 row: &str,
61 ) -> (
62 String,
63 String,
64 String,
65 chrono::DateTime<chrono::Utc>,
66 Option<serde_json::Value>,
67 ) {
68 (
69 table.to_string(),
70 "INSERT".to_string(),
71 row.to_string(),
72 chrono::Utc::now(),
73 Some(serde_json::json!({ "row": row })),
74 )
75 }
76
77 // ── log ─────────────────────────────────────────────────────────────────────
78
79 #[tokio::test]
80 async fn an_append_keeps_every_entry_and_the_order_it_arrived_in() {
81 let db = TestDb::new().await;
82 let user = seed_user(&db.pool, "sklog_append").await;
83 let app = seed_app(&db.pool, user, "logappend").await;
84 let device = seed_device(&db.pool, app, user, "laptop").await;
85
86 let changes: Vec<_> = (0..5).map(|i| change("tasks", &format!("r{i}"))).collect();
87 let cursor = synckit::push_sync_changes(&db.pool, app, user, device, Uuid::new_v4(), &changes)
88 .await
89 .unwrap();
90
91 let entries = synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
92 .await
93 .unwrap();
94 assert_eq!(entries.len(), 5, "an append drops nothing");
95 let rows: Vec<&str> = entries.iter().map(|e| e.row_id.as_str()).collect();
96 assert_eq!(rows, ["r0", "r1", "r2", "r3", "r4"], "and reorders nothing");
97 assert!(
98 entries.windows(2).all(|w| w[0].seq < w[1].seq),
99 "seq is strictly increasing: {:?}",
100 entries.iter().map(|e| e.seq).collect::<Vec<_>>()
101 );
102 assert_eq!(
103 cursor,
104 entries.last().unwrap().seq,
105 "the returned cursor is the highest seq assigned"
106 );
107 }
108
109 #[tokio::test]
110 async fn cursor_paging_returns_each_entry_exactly_once() {
111 let db = TestDb::new().await;
112 let user = seed_user(&db.pool, "sklog_page").await;
113 let app = seed_app(&db.pool, user, "logpage").await;
114 let device = seed_device(&db.pool, app, user, "laptop").await;
115
116 for i in 0..5 {
117 synckit::push_sync_changes(
118 &db.pool,
119 app,
120 user,
121 device,
122 Uuid::new_v4(),
123 &[change("tasks", &format!("r{i}"))],
124 )
125 .await
126 .unwrap();
127 }
128
129 // Walk the log two at a time the way a client does, carrying the last seq
130 // forward as the next cursor.
131 let mut seen: Vec<String> = Vec::new();
132 let mut cursor = 0i64;
133 loop {
134 let page = synckit::pull_sync_changes(&db.pool, app, user, cursor, 2)
135 .await
136 .unwrap();
137 if page.is_empty() {
138 break;
139 }
140 cursor = page.last().unwrap().seq;
141 seen.extend(page.into_iter().map(|e| e.row_id));
142 assert!(seen.len() <= 5, "paging must terminate: {seen:?}");
143 }
144 assert_eq!(
145 seen,
146 ["r0", "r1", "r2", "r3", "r4"],
147 "every entry once, in order, with no gap at a page boundary"
148 );
149 }
150
151 #[tokio::test]
152 async fn a_replayed_batch_appends_nothing_and_returns_the_same_cursor() {
153 let db = TestDb::new().await;
154 let user = seed_user(&db.pool, "sklog_replay").await;
155 let app = seed_app(&db.pool, user, "logreplay").await;
156 let device = seed_device(&db.pool, app, user, "laptop").await;
157
158 let batch = Uuid::new_v4();
159 let changes = [change("tasks", "r0"), change("tasks", "r1")];
160 let first = synckit::push_sync_changes(&db.pool, app, user, device, batch, &changes)
161 .await
162 .unwrap();
163 // The client never saw the response and retried the same batch id.
164 let second = synckit::push_sync_changes(&db.pool, app, user, device, batch, &changes)
165 .await
166 .unwrap();
167
168 assert_eq!(first, second, "at most once: the same cursor comes back");
169 let entries = synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
170 .await
171 .unwrap();
172 assert_eq!(entries.len(), 2, "the retry inserted nothing: {entries:?}");
173 }
174
175 #[tokio::test]
176 async fn a_pull_never_reaches_another_user_or_another_app() {
177 let db = TestDb::new().await;
178 let alice = seed_user(&db.pool, "sklog_alice").await;
179 let bob = seed_user(&db.pool, "sklog_bob").await;
180 let app = seed_app(&db.pool, alice, "logscope").await;
181 let other_app = seed_app(&db.pool, alice, "logscope2").await;
182 let alice_dev = seed_device(&db.pool, app, alice, "alice-laptop").await;
183 let bob_dev = seed_device(&db.pool, app, bob, "bob-laptop").await;
184 let alice_other_dev = seed_device(&db.pool, other_app, alice, "alice-phone").await;
185
186 synckit::push_sync_changes(
187 &db.pool,
188 app,
189 alice,
190 alice_dev,
191 Uuid::new_v4(),
192 &[change("tasks", "alice")],
193 )
194 .await
195 .unwrap();
196 synckit::push_sync_changes(
197 &db.pool,
198 app,
199 bob,
200 bob_dev,
201 Uuid::new_v4(),
202 &[change("tasks", "bob")],
203 )
204 .await
205 .unwrap();
206 synckit::push_sync_changes(
207 &db.pool,
208 other_app,
209 alice,
210 alice_other_dev,
211 Uuid::new_v4(),
212 &[change("tasks", "alice-other-app")],
213 )
214 .await
215 .unwrap();
216
217 let alice_entries = synckit::pull_sync_changes(&db.pool, app, alice, 0, 100)
218 .await
219 .unwrap();
220 let rows: Vec<&str> = alice_entries.iter().map(|e| e.row_id.as_str()).collect();
221 assert_eq!(
222 rows,
223 ["alice"],
224 "one user's log is one user's, per app: {rows:?}"
225 );
226 }
227
228 // ── blobs ───────────────────────────────────────────────────────────────────
229
230 #[tokio::test]
231 async fn a_blob_is_scoped_to_the_user_who_stored_it() {
232 let db = TestDb::new().await;
233 let alice = seed_user(&db.pool, "skblob_alice").await;
234 let bob = seed_user(&db.pool, "skblob_bob").await;
235 let app = seed_app(&db.pool, alice, "blobscope").await;
236 make_internal(&db.pool, app).await;
237 seed_active_subscription(&db.pool, alice, app, "sub_blob_alice", 1_000_000).await;
238 seed_active_subscription(&db.pool, bob, app, "sub_blob_bob", 1_000_000).await;
239
240 let stored =
241 synckit::confirm_internal_blob(&db.pool, app, alice, "hash-a", 400, "s3/alice", "default")
242 .await
243 .unwrap();
244 assert_eq!(stored, BlobConfirm::Stored);
245
246 assert!(
247 synckit::get_sync_blob_by_hash(&db.pool, app, alice, "hash-a")
248 .await
249 .unwrap()
250 .is_some()
251 );
252 assert!(
253 synckit::get_sync_blob_by_hash(&db.pool, app, bob, "hash-a")
254 .await
255 .unwrap()
256 .is_none(),
257 "the hash is the same bytes, but it is not bob's row"
258 );
259 assert_eq!(
260 synckit::storage_used_bytes(&db.pool, app, bob)
261 .await
262 .unwrap(),
263 0,
264 "and it is not charged to bob's quota"
265 );
266 assert_eq!(
267 synckit::storage_used_bytes(&db.pool, app, alice)
268 .await
269 .unwrap(),
270 400
271 );
272 }
273
274 #[tokio::test]
275 async fn re_confirming_the_same_hash_stores_one_row_and_charges_once() {
276 let db = TestDb::new().await;
277 let user = seed_user(&db.pool, "skblob_dedup").await;
278 let app = seed_app(&db.pool, user, "blobdedup").await;
279 make_internal(&db.pool, app).await;
280 seed_active_subscription(&db.pool, user, app, "sub_blob_dedup", 1_000).await;
281
282 let first =
283 synckit::confirm_internal_blob(&db.pool, app, user, "hash-x", 400, "s3/x", "default")
284 .await
285 .unwrap();
286 let second =
287 synckit::confirm_internal_blob(&db.pool, app, user, "hash-x", 400, "s3/x", "default")
288 .await
289 .unwrap();
290 assert_eq!(first, BlobConfirm::Stored);
291 assert_eq!(
292 second,
293 BlobConfirm::AlreadyStored,
294 "content-addressed: the same hash is the same object"
295 );
296
297 assert_eq!(
298 synckit::storage_used_bytes(&db.pool, app, user)
299 .await
300 .unwrap(),
301 400,
302 "a re-upload must not count twice, or a retry would eat the cap"
303 );
304 let rows: i64 = sqlx::query_scalar(
305 "SELECT COUNT(*) FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = 'hash-x'",
306 )
307 .bind(app)
308 .bind(user)
309 .fetch_one(&db.pool)
310 .await
311 .unwrap();
312 assert_eq!(rows, 1);
313 }
314
315 // ── keys ────────────────────────────────────────────────────────────────────
316
317 #[tokio::test]
318 async fn a_key_upsert_takes_only_the_version_the_caller_expected() {
319 let db = TestDb::new().await;
320 let user = seed_user(&db.pool, "skkey_occ").await;
321 let app = seed_app(&db.pool, user, "keyocc").await;
322
323 assert!(
324 synckit::upsert_sync_key(&db.pool, app, user, "env_v1", 0)
325 .await
326 .unwrap(),
327 "the first key inserts"
328 );
329 let info = synckit::get_sync_key(&db.pool, app, user)
330 .await
331 .unwrap()
332 .expect("key exists");
333 assert_eq!(info.encrypted_key, "env_v1");
334 assert_eq!(info.key_version, 1);
335 assert!(info.pending_key.is_none());
336
337 // A second device that still believes it is at version 0 loses.
338 assert!(
339 !synckit::upsert_sync_key(&db.pool, app, user, "env_stale", 0)
340 .await
341 .unwrap(),
342 "a stale expected_version is a conflict, not a write"
343 );
344 assert_eq!(
345 synckit::get_sync_key(&db.pool, app, user)
346 .await
347 .unwrap()
348 .unwrap()
349 .encrypted_key,
350 "env_v1",
351 "and the losing envelope must not have landed"
352 );
353
354 assert!(
355 synckit::upsert_sync_key(&db.pool, app, user, "env_v2", 1)
356 .await
357 .unwrap()
358 );
359 let info = synckit::get_sync_key(&db.pool, app, user)
360 .await
361 .unwrap()
362 .unwrap();
363 assert_eq!(info.encrypted_key, "env_v2");
364 assert_eq!(info.key_version, 2);
365 }
366
367 #[tokio::test]
368 async fn a_key_belongs_to_one_user_within_one_app() {
369 let db = TestDb::new().await;
370 let alice = seed_user(&db.pool, "skkey_alice").await;
371 let bob = seed_user(&db.pool, "skkey_bob").await;
372 let app = seed_app(&db.pool, alice, "keyscope").await;
373 synckit::upsert_sync_key(&db.pool, app, alice, "alice_env", 0)
374 .await
375 .unwrap();
376
377 assert!(
378 synckit::get_sync_key(&db.pool, app, bob)
379 .await
380 .unwrap()
381 .is_none(),
382 "bob has no key here, and must not be handed alice's envelope"
383 );
384 }
385
386 #[tokio::test]
387 async fn pruning_refuses_a_non_positive_horizon() {
388 let db = TestDb::new().await;
389 let user = seed_user(&db.pool, "skkey_prune").await;
390 let app = seed_app(&db.pool, user, "keyprune").await;
391 let device = seed_device(&db.pool, app, user, "laptop").await;
392 synckit::push_sync_changes(
393 &db.pool,
394 app,
395 user,
396 device,
397 Uuid::new_v4(),
398 &[change("tasks", "r0")],
399 )
400 .await
401 .unwrap();
402
403 for horizon in [0, -1] {
404 assert_eq!(
405 synckit::prune_sync_log(&db.pool, horizon).await.unwrap(),
406 0,
407 "a zero or negative retention is a mistake, not an instruction to \
408 delete the whole log"
409 );
410 }
411 assert_eq!(
412 synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
413 .await
414 .unwrap()
415 .len(),
416 1
417 );
418 // A real horizon spares entries inside it.
419 assert_eq!(synckit::prune_sync_log(&db.pool, 30).await.unwrap(), 0);
420 }
421
422 #[tokio::test]
423 async fn compaction_holds_back_a_log_no_device_has_pulled() {
424 let db = TestDb::new().await;
425 let user = seed_user(&db.pool, "skkey_compact").await;
426 let app = seed_app(&db.pool, user, "keycompact").await;
427 let device = seed_device(&db.pool, app, user, "laptop").await;
428 synckit::push_sync_changes(
429 &db.pool,
430 app,
431 user,
432 device,
433 Uuid::new_v4(),
434 &[change("tasks", "r0")],
435 )
436 .await
437 .unwrap();
438
439 assert_eq!(
440 synckit::compact_sync_log(&db.pool, app, user, 0)
441 .await
442 .unwrap(),
443 0,
444 "a zero-day safety margin compacts nothing"
445 );
446 assert_eq!(
447 synckit::compact_sync_log(&db.pool, app, user, 7)
448 .await
449 .unwrap(),
450 0,
451 "the device sits at cursor 0, so nothing is known-pulled"
452 );
453 assert_eq!(
454 synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
455 .await
456 .unwrap()
457 .len(),
458 1
459 );
460 }
461