Skip to main content

max / makenotwork

12.1 KB · 320 lines History Blame Raw
1 //! DB-layer contract tests for `db::pending_uploads`, the tracking table that
2 //! decides whether an abandoned upload's object is reclaimed or leaked.
3 //!
4 //! Every function in the module is `pub(crate)`, so its reclaim/expiry half is
5 //! reached the way production reaches it, through the orphan-upload reaper
6 //! (`TestHarness::run_orphan_upload_reaper`), which calls
7 //! `get_stale_pending_uploads` and `delete_pending_uploads` and nothing else.
8 //! Those two carry the retry machinery this table exists for, and they are what
9 //! the assertions below are about: the age cutoff on both sides, oldest-first
10 //! draining under the `STALE_UPLOAD_BATCH` cap, the `(s3_key, bucket)` pair
11 //! scoping on the delete, and the reclaim case where a live row took the key
12 //! back and the object must survive while the tracking row goes.
13 //!
14 //! Not asserted here, and deliberately not claimed: `record_pending_upload`,
15 //! `record_pending_multipart_upload`, `declared_size`, `touch_pending_upload`,
16 //! `remove_pending_upload` and `is_owned`, which an integration test can only
17 //! reach through the presign handlers.
18 //!
19 //! Delete this file and a reaper that either deletes a reclaimed object or
20 //! leaves an orphan behind forever passes silently.
21
22 use crate::harness::{TestHarness, seed_project};
23 use makenotwork::db::items::update_item_file_cas;
24 use makenotwork::db::{ItemId, ProjectId, UserId};
25 use makenotwork::storage::{FileType, StorageBackend};
26 // ── helpers ──────────────────────────────────────────────────────────────────
27
28 async fn seed_item(pool: &sqlx::PgPool, project: ProjectId, slug: &str) -> ItemId {
29 sqlx::query_scalar::<_, ItemId>(
30 "INSERT INTO items (project_id, title, item_type, price_cents, slug)
31 VALUES ($1, $2, 'digital', 1000, $3) RETURNING id",
32 )
33 .bind(project)
34 .bind(format!("Item {slug}"))
35 .bind(slug)
36 .fetch_one(pool)
37 .await
38 .expect("seed item")
39 }
40
41 /// The audio pair as stored.
42 async fn audio_cols(pool: &sqlx::PgPool, item: ItemId) -> (Option<String>, Option<i64>) {
43 sqlx::query_as("SELECT audio_s3_key, audio_file_size_bytes FROM items WHERE id = $1")
44 .bind(item)
45 .fetch_one(pool)
46 .await
47 .expect("read audio columns")
48 }
49
50 // ── db::pending_uploads: the expiry and reclaim half ─────────────────────────
51 //
52 // Reached through the reaper, which is the only production caller of
53 // `get_stale_pending_uploads` and `delete_pending_uploads`.
54
55 /// Insert a pending upload aged `hours` old, with its object staged in storage.
56 async fn pending_upload(h: &TestHarness, user: UserId, key: &str, bucket: &str, hours: i32) {
57 h.storage
58 .as_ref()
59 .expect("with_storage provides a backend")
60 .put(key, b"orphan".to_vec());
61 sqlx::query(
62 "INSERT INTO pending_uploads (user_id, s3_key, bucket, created_at)
63 VALUES ($1, $2, $3, NOW() - make_interval(hours => $4))",
64 )
65 .bind(user)
66 .bind(key)
67 .bind(bucket)
68 .bind(hours)
69 .execute(&h.db)
70 .await
71 .expect("insert pending upload");
72 }
73
74 async fn pending_rows(h: &TestHarness, key: &str, bucket: &str) -> i64 {
75 sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads WHERE s3_key = $1 AND bucket = $2")
76 .bind(key)
77 .bind(bucket)
78 .fetch_one(&h.db)
79 .await
80 .expect("count pending rows")
81 }
82
83 /// The cutoff is 24 hours and it is a strict age test, so both sides of it are
84 /// asserted in one run: a 25-hour-old upload is reaped and a 23-hour-old one is
85 /// left completely alone. A test that only checked the old row could not tell
86 /// "older than 24h" from "every row".
87 #[tokio::test]
88 async fn the_reaper_takes_uploads_past_the_cutoff_and_leaves_younger_ones_untouched() {
89 let mut h = TestHarness::with_storage().await;
90 let user = h
91 .signup("reaper_cutoff", "reaper_cutoff@test.com", "pass1234")
92 .await;
93 let storage = h.storage.clone().expect("with_storage provides a backend");
94
95 pending_upload(&h, user, "staging/too-old.bin", "main", 25).await;
96 pending_upload(&h, user, "staging/still-young.bin", "main", 23).await;
97
98 h.run_orphan_upload_reaper().await;
99
100 assert_eq!(
101 pending_rows(&h, "staging/too-old.bin", "main").await,
102 0,
103 "an upload older than the 24h cutoff must lose its tracking row"
104 );
105 assert!(
106 !storage.object_exists("staging/too-old.bin").await.unwrap(),
107 "and its object must be deleted"
108 );
109
110 assert_eq!(
111 pending_rows(&h, "staging/still-young.bin", "main").await,
112 1,
113 "an upload younger than the cutoff is still in flight and must be kept"
114 );
115 assert!(
116 storage
117 .object_exists("staging/still-young.bin")
118 .await
119 .unwrap(),
120 "deleting a young upload's object would destroy a transfer in progress"
121 );
122 }
123
124 /// The reaper is capped at `STALE_UPLOAD_BATCH` (200) rows per tick and drains
125 /// oldest first, so a backlog cannot wedge the tick. 201 rows put the cap on both
126 /// sides in one run: exactly one row survives, and it is the youngest, which is
127 /// what distinguishes oldest-first from an unordered or newest-first scan.
128 #[tokio::test]
129 async fn the_reaper_drains_the_oldest_two_hundred_and_leaves_the_rest_for_the_next_tick() {
130 let mut h = TestHarness::with_storage().await;
131 let user = h
132 .signup("reaper_batch", "reaper_batch@test.com", "pass1234")
133 .await;
134 let storage = h.storage.clone().expect("with_storage provides a backend");
135
136 // 201 stale rows, ages 48h down to about 44h40m, so the ordering is total and
137 // every one of them is past the 24h cutoff.
138 for i in 0..201i32 {
139 let key = format!("staging/batch-{i:03}.bin");
140 storage.put(&key, b"orphan".to_vec());
141 }
142 sqlx::query(
143 "INSERT INTO pending_uploads (user_id, s3_key, bucket, created_at)
144 SELECT $1, 'staging/batch-' || to_char(i, 'FM000') || '.bin', 'main',
145 NOW() - INTERVAL '48 hours' + make_interval(mins => i)
146 FROM generate_series(0, 200) AS i",
147 )
148 .bind(user)
149 .execute(&h.db)
150 .await
151 .expect("insert 201 stale uploads");
152
153 h.run_orphan_upload_reaper().await;
154
155 let remaining: Vec<String> =
156 sqlx::query_scalar("SELECT s3_key FROM pending_uploads ORDER BY s3_key")
157 .fetch_all(&h.db)
158 .await
159 .expect("read remaining rows");
160 assert_eq!(
161 remaining,
162 vec!["staging/batch-200.bin".to_string()],
163 "exactly the youngest row is left for the next tick, got {remaining:?}"
164 );
165 assert!(
166 storage
167 .object_exists("staging/batch-200.bin")
168 .await
169 .unwrap(),
170 "the row that was not reaped keeps its object"
171 );
172 assert!(
173 !storage
174 .object_exists("staging/batch-000.bin")
175 .await
176 .unwrap(),
177 "the oldest row is the first one drained"
178 );
179
180 // The next tick finishes the backlog, which is what makes the cap a drain
181 // rather than a permanent leak.
182 h.run_orphan_upload_reaper().await;
183 let left: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads")
184 .fetch_one(&h.db)
185 .await
186 .expect("count remaining");
187 assert_eq!(left, 0, "the second tick drains what the cap held back");
188 }
189
190 /// The uniqueness key is `(s3_key, bucket)`, so the same key can be pending in
191 /// two buckets at once and the delete must match the pair. Only the main-bucket
192 /// row is stale here: a delete keyed on `s3_key` alone would take the synckit
193 /// row with it and the reaper would forget an upload that is still in flight.
194 #[tokio::test]
195 async fn reaping_one_bucket_leaves_another_buckets_row_for_the_same_key() {
196 let mut h = TestHarness::with_storage().await;
197 let user = h
198 .signup("reaper_bucket", "reaper_bucket@test.com", "pass1234")
199 .await;
200
201 pending_upload(&h, user, "staging/shared-key.bin", "main", 30).await;
202 // Young, and in a different bucket: not eligible for this reap at all.
203 sqlx::query(
204 "INSERT INTO pending_uploads (user_id, s3_key, bucket, created_at)
205 VALUES ($1, 'staging/shared-key.bin', 'synckit', NOW())",
206 )
207 .bind(user)
208 .execute(&h.db)
209 .await
210 .expect("insert synckit pending row");
211
212 h.run_orphan_upload_reaper().await;
213
214 assert_eq!(
215 pending_rows(&h, "staging/shared-key.bin", "main").await,
216 0,
217 "the stale main-bucket row is reaped"
218 );
219 assert_eq!(
220 pending_rows(&h, "staging/shared-key.bin", "synckit").await,
221 1,
222 "the synckit row shares the key but not the bucket and must survive"
223 );
224 }
225
226 /// The reclaim path, which is why the reaper routes through the live-key guard.
227 /// A confirm can take a staging key over onto a live item row before the reaper
228 /// reaches the stale pending record. The object now belongs to that item, so it
229 /// must survive; only the tracking row goes.
230 #[tokio::test]
231 async fn a_key_reclaimed_by_a_live_item_keeps_its_object_and_only_loses_the_pending_row() {
232 let mut h = TestHarness::with_storage().await;
233 let user = h
234 .signup("reaper_reclaim", "reaper_reclaim@test.com", "pass1234")
235 .await;
236 let storage = h.storage.clone().expect("with_storage provides a backend");
237
238 let project = seed_project(&h.db, user, "reclaim-proj").await;
239 let item = seed_item(&h.db, project, "reclaim-item").await;
240 pending_upload(&h, user, "staging/reclaimed.bin", "main", 40).await;
241
242 // The confirm that beat the reaper: the key is now the item's live audio.
243 update_item_file_cas(
244 &h.db,
245 item,
246 user,
247 FileType::Audio,
248 None,
249 "staging/reclaimed.bin",
250 3_300_000,
251 )
252 .await
253 .expect("confirm reclaims the key");
254
255 h.run_orphan_upload_reaper().await;
256
257 assert!(
258 storage
259 .object_exists("staging/reclaimed.bin")
260 .await
261 .unwrap(),
262 "the object is a live item file now; deleting it would destroy a paid-for upload"
263 );
264 assert_eq!(
265 pending_rows(&h, "staging/reclaimed.bin", "main").await,
266 0,
267 "the stale tracking row is still cleared, or the reaper retries it forever"
268 );
269 let queued: i64 =
270 sqlx::query_scalar("SELECT COUNT(*) FROM pending_s3_deletions WHERE s3_key = $1")
271 .bind("staging/reclaimed.bin")
272 .fetch_one(&h.db)
273 .await
274 .expect("count queued deletions");
275 assert_eq!(
276 queued, 0,
277 "a live key must not be handed to the deletion queue either"
278 );
279 let audio = audio_cols(&h.db, item).await;
280 assert_eq!(
281 audio,
282 (Some("staging/reclaimed.bin".to_string()), Some(3_300_000)),
283 "and the item still points at it, got {audio:?}"
284 );
285 }
286
287 /// Running the reaper twice over the same backlog must delete each object once.
288 /// The tracking rows are gone after the first pass, so the second finds nothing:
289 /// this is the replay guard on the reap itself.
290 #[tokio::test]
291 async fn a_second_reaper_pass_over_a_drained_backlog_deletes_nothing_again() {
292 let mut h = TestHarness::with_storage().await;
293 let user = h
294 .signup("reaper_twice", "reaper_twice@test.com", "pass1234")
295 .await;
296 let storage = h.storage.clone().expect("with_storage provides a backend");
297
298 pending_upload(&h, user, "staging/once-a.bin", "main", 26).await;
299 pending_upload(&h, user, "staging/once-b.bin", "main", 27).await;
300
301 h.run_orphan_upload_reaper().await;
302 let after_first = storage.faults().calls("delete_object");
303 assert_eq!(
304 after_first, 2,
305 "the first pass deletes each of the two orphans exactly once"
306 );
307
308 h.run_orphan_upload_reaper().await;
309 assert_eq!(
310 storage.faults().calls("delete_object"),
311 after_first,
312 "the second pass has no rows to act on and must issue no further deletes"
313 );
314 let left: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads")
315 .fetch_one(&h.db)
316 .await
317 .expect("count remaining");
318 assert_eq!(left, 0, "nothing is left pending");
319 }
320