Skip to main content

max / makenotwork

Test db money/race cold spots; fix inert cancel_stale_rotation Close the deferred db Testing axis from the Run 16 audit with 34 Postgres-backed layer tests: - creator_tiers: storage-cap CAS concurrency invariant (never over-sells), atomic replace/clamp, subscription lifecycle, 30-day grace sweep - webhook_events: concurrent-duplicate mark idempotency, get_retryable_events SKIP LOCKED claim-and-defer, backoff/dead-letter, retention prune - synckit rotation: concurrent begin_key_rotation yields exactly one rotation (pins the Phase 3 FOR UPDATE fix), full begin/complete/cancel lifecycle The rotation tests surfaced a live bug: cancel_stale_rotation bound stale_hours as i64, so make_interval(hours => $3) resolved to make_interval(hours => bigint) and errored 42883 on every call, leaving the stale-rotation sweep inert. Fixed with $3::int. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-02 14:31 UTC
Commit: 95370d433be7ec08f3d790a54c7283e6dcc94822
Parent: dfdf4b9
5 files changed, +936 insertions, -1 deletion
M server/src/db/synckit/rotation.rs +1 -1
@@ -303,7 +303,7 @@ pub async fn cancel_stale_rotation(
303 303 r#"
304 304 DELETE FROM sync_key_rotations
305 305 WHERE app_id = $1 AND user_id = $2
306 - AND updated_at < NOW() - make_interval(hours => $3)
306 + AND updated_at < NOW() - make_interval(hours => $3::int)
307 307 "#,
308 308 )
309 309 .bind(app_id)
A server/tests/workflows/db_creator_tiers_layer.rs +359
@@ -0,0 +1,359 @@
1 + //! DB-layer contract tests for `db::creator_tiers` — the money/quota module.
2 + //!
3 + //! Audit Run 16 flagged the creator-tier DB layer as the Testing cold spot: the
4 + //! 280-line in-file test block exercises only the pure `label`/`price_cents`
5 + //! table, while the storage-cap CAS and the subscription lifecycle — the parts
6 + //! that actually enforce money and quota — were asserted only indirectly through
7 + //! HTTP upload flows ("money/race modules lack in-file `#[sqlx::test]`"). These
8 + //! call the `db::creator_tiers` functions directly against real Postgres so the
9 + //! invariants live where they belong:
10 + //!
11 + //! 1. `try_increment_storage` is a real compare-and-set — it never lets
12 + //! concurrent uploads push a creator over their cap (the race a plain
13 + //! read-then-write would lose).
14 + //! 2. `try_replace_storage_on` swaps sizes atomically and clamps the decrement.
15 + //! 3. The subscription lifecycle (create / duplicate-webhook / cancel /
16 + //! resubscribe) and the 30-day grace window resolve as documented.
17 +
18 + use crate::harness::db::TestDb;
19 + use makenotwork::db::{self, CreatorTier, UserId};
20 +
21 + const MB: i64 = 1024 * 1024;
22 +
23 + /// Seed a minimal verified user (rows FK `users(id)`).
24 + async fn seed_user(pool: &sqlx::PgPool, username: &str) -> UserId {
25 + let hash = makenotwork::auth::hash_password("password123").expect("hash");
26 + sqlx::query_scalar::<_, UserId>(
27 + "INSERT INTO users (username, email, password_hash, email_verified)
28 + VALUES ($1, $2, $3, true) RETURNING id",
29 + )
30 + .bind(username)
31 + .bind(format!("{username}@test.com"))
32 + .bind(&hash)
33 + .fetch_one(pool)
34 + .await
35 + .expect("seed user")
36 + }
37 +
38 + async fn storage_used(pool: &sqlx::PgPool, user: UserId) -> i64 {
39 + db::creator_tiers::get_storage_used(pool, user).await.unwrap()
40 + }
41 +
42 + // ── try_increment_storage — the cap-checked CAS ──────────────────────────────
43 +
44 + #[tokio::test]
45 + async fn increment_within_cap_advances_the_counter() {
46 + let db = TestDb::new().await;
47 + let user = seed_user(&db.pool, "ct_inc_ok").await;
48 +
49 + db::creator_tiers::try_increment_storage(&db.pool, user, 10 * MB, 100 * MB)
50 + .await
51 + .expect("increment under cap must succeed");
52 + assert_eq!(storage_used(&db.pool, user).await, 10 * MB);
53 + }
54 +
55 + #[tokio::test]
56 + async fn increment_to_exact_cap_succeeds_then_one_more_byte_fails() {
57 + let db = TestDb::new().await;
58 + let user = seed_user(&db.pool, "ct_inc_boundary").await;
59 + let cap = 50 * MB;
60 +
61 + // Filling to exactly the cap is allowed (the check is `<= cap`).
62 + db::creator_tiers::try_increment_storage(&db.pool, user, cap, cap)
63 + .await
64 + .expect("filling to exactly the cap must succeed");
65 + assert_eq!(storage_used(&db.pool, user).await, cap);
66 +
67 + // One more byte crosses the cap and is rejected, leaving the counter intact.
68 + let err = db::creator_tiers::try_increment_storage(&db.pool, user, 1, cap)
69 + .await
70 + .expect_err("crossing the cap must be rejected");
71 + assert!(
72 + matches!(err, makenotwork::error::AppError::BadRequest(_)),
73 + "cap rejection surfaces as a user-facing BadRequest, got {err:?}"
74 + );
75 + assert_eq!(
76 + storage_used(&db.pool, user).await,
77 + cap,
78 + "a rejected increment must not move the counter"
79 + );
80 + }
81 +
82 + /// The invariant a plain read-then-write would break: many uploads confirm at
83 + /// once, each incrementing the shared counter. The cap-checked UPDATE is a CAS,
84 + /// so the sum of the winners can never exceed the cap — no over-sell.
85 + #[tokio::test]
86 + async fn concurrent_increments_never_exceed_the_cap() {
87 + let db = TestDb::new().await;
88 + let user = seed_user(&db.pool, "ct_inc_race").await;
89 +
90 + let chunk = 10 * MB;
91 + let cap = 5 * chunk; // exactly five chunks fit
92 + let racers = 20;
93 +
94 + let mut handles = Vec::new();
95 + for _ in 0..racers {
96 + let pool = db.pool.clone();
97 + handles.push(tokio::spawn(async move {
98 + db::creator_tiers::try_increment_storage(&pool, user, chunk, cap).await.is_ok()
99 + }));
100 + }
101 + let mut winners = 0;
102 + for h in handles {
103 + if h.await.expect("task panicked") {
104 + winners += 1;
105 + }
106 + }
107 +
108 + assert_eq!(winners, 5, "exactly cap/chunk increments may win the CAS");
109 + let final_used = storage_used(&db.pool, user).await;
110 + assert_eq!(final_used, cap, "the counter lands exactly on the cap");
111 + assert!(final_used <= cap, "the CAS must never over-sell storage");
112 + }
113 +
114 + // ── try_replace_storage_on / try_apply_storage_on — atomic swap + clamps ─────
115 +
116 + #[tokio::test]
117 + async fn replace_storage_swaps_old_for_new_atomically() {
118 + let db = TestDb::new().await;
119 + let user = seed_user(&db.pool, "ct_replace").await;
120 + let cap = 100 * MB;
121 +
122 + db::creator_tiers::try_increment_storage(&db.pool, user, 30 * MB, cap).await.unwrap();
123 +
124 + let mut conn = db.pool.acquire().await.unwrap();
125 + // Swap the 30MB file for a 20MB one: net -10MB in a single statement.
126 + db::creator_tiers::try_replace_storage_on(&mut conn, user, 30 * MB, 20 * MB, cap)
127 + .await
128 + .expect("replace under cap must succeed");
129 + drop(conn);
130 +
131 + assert_eq!(storage_used(&db.pool, user).await, 20 * MB);
132 + }
133 +
134 + #[tokio::test]
135 + async fn replace_storage_over_cap_is_rejected_without_drift() {
136 + let db = TestDb::new().await;
137 + let user = seed_user(&db.pool, "ct_replace_cap").await;
138 + let cap = 50 * MB;
139 +
140 + db::creator_tiers::try_increment_storage(&db.pool, user, 40 * MB, cap).await.unwrap();
141 +
142 + let mut conn = db.pool.acquire().await.unwrap();
143 + // Replacing the 40MB file with a 60MB one would land at 60MB > cap: rejected.
144 + let err = db::creator_tiers::try_replace_storage_on(&mut conn, user, 40 * MB, 60 * MB, cap)
145 + .await
146 + .expect_err("replace that overflows the cap must be rejected");
147 + assert!(matches!(err, makenotwork::error::AppError::BadRequest(_)));
148 + drop(conn);
149 +
150 + assert_eq!(
151 + storage_used(&db.pool, user).await,
152 + 40 * MB,
153 + "a rejected replace must leave the counter untouched (no drift)"
154 + );
155 + }
156 +
157 + #[tokio::test]
158 + async fn replace_storage_clamps_the_decrement_at_zero() {
159 + let db = TestDb::new().await;
160 + let user = seed_user(&db.pool, "ct_replace_clamp").await;
161 + let cap = 100 * MB;
162 +
163 + // Counter is 10MB but we claim to remove a 999MB "old" file: the
164 + // GREATEST(0, ...) clamp prevents a negative counter, then adds the new 5MB.
165 + db::creator_tiers::try_increment_storage(&db.pool, user, 10 * MB, cap).await.unwrap();
166 + let mut conn = db.pool.acquire().await.unwrap();
167 + db::creator_tiers::try_replace_storage_on(&mut conn, user, 999 * MB, 5 * MB, cap)
168 + .await
169 + .expect("clamped replace succeeds");
170 + drop(conn);
171 +
172 + assert_eq!(
173 + storage_used(&db.pool, user).await,
174 + 5 * MB,
175 + "the decrement clamps to zero, so the result is just the new size"
176 + );
177 + }
178 +
179 + #[tokio::test]
180 + async fn apply_storage_dispatches_increment_vs_replace() {
181 + let db = TestDb::new().await;
182 + let user = seed_user(&db.pool, "ct_apply").await;
183 + let cap = 100 * MB;
184 +
185 + // None => fresh increment.
186 + let mut conn = db.pool.acquire().await.unwrap();
187 + db::creator_tiers::try_apply_storage_on(&mut conn, user, None, 15 * MB, cap)
188 + .await
189 + .unwrap();
190 + // Some(old) => replace.
191 + db::creator_tiers::try_apply_storage_on(&mut conn, user, Some(15 * MB), 25 * MB, cap)
192 + .await
193 + .unwrap();
194 + drop(conn);
195 +
196 + assert_eq!(storage_used(&db.pool, user).await, 25 * MB);
197 + }
198 +
199 + #[tokio::test]
200 + async fn decrement_storage_clamps_at_zero() {
201 + let db = TestDb::new().await;
202 + let user = seed_user(&db.pool, "ct_dec").await;
203 +
204 + db::creator_tiers::try_increment_storage(&db.pool, user, 5 * MB, 100 * MB).await.unwrap();
205 + // Over-decrement (double-delete, stale size) must floor at zero, not wrap negative.
206 + db::creator_tiers::decrement_storage_used(&db.pool, user, 999 * MB).await.unwrap();
207 +
208 + assert_eq!(storage_used(&db.pool, user).await, 0);
209 + }
210 +
211 + // ── subscription lifecycle ───────────────────────────────────────────────────
212 +
213 + #[tokio::test]
214 + async fn create_then_read_active_tier() {
215 + let db = TestDb::new().await;
216 + let user = seed_user(&db.pool, "ct_sub_create").await;
217 +
218 + assert!(
219 + db::creator_tiers::get_active_creator_tier(&db.pool, user).await.unwrap().is_none(),
220 + "no subscription => no active tier"
221 + );
222 +
223 + let created = db::creator_tiers::create_creator_subscription(
224 + &db.pool, user, "sub_ct_1", "cus_ct_1", CreatorTier::BigFiles,
225 + )
226 + .await
227 + .unwrap();
228 + assert!(created.is_some(), "a fresh insert returns the new row");
229 +
230 + assert_eq!(
231 + db::creator_tiers::get_active_creator_tier(&db.pool, user).await.unwrap(),
232 + Some(CreatorTier::BigFiles),
233 + );
234 + }
235 +
236 + #[tokio::test]
237 + async fn duplicate_webhook_insert_is_a_noop() {
238 + let db = TestDb::new().await;
239 + let user = seed_user(&db.pool, "ct_sub_dup").await;
240 +
241 + db::creator_tiers::create_creator_subscription(
242 + &db.pool, user, "sub_dup", "cus_dup", CreatorTier::SmallFiles,
243 + )
244 + .await
245 + .unwrap();
246 +
247 + // Same subscription id, already active => the guarded ON CONFLICT is a no-op
248 + // and RETURNING yields no row. This is what makes a Stripe redelivery safe.
249 + let again = db::creator_tiers::create_creator_subscription(
250 + &db.pool, user, "sub_dup", "cus_dup", CreatorTier::SmallFiles,
251 + )
252 + .await
253 + .unwrap();
254 + assert!(again.is_none(), "a duplicate active-webhook insert returns None");
255 + }
256 +
257 + #[tokio::test]
258 + async fn cancel_clears_active_tier_and_resubscribe_reactivates() {
259 + let db = TestDb::new().await;
260 + let user = seed_user(&db.pool, "ct_sub_cycle").await;
261 +
262 + db::creator_tiers::create_creator_subscription(
263 + &db.pool, user, "sub_cycle_a", "cus_cycle", CreatorTier::Everything,
264 + )
265 + .await
266 + .unwrap();
267 +
268 + let canceled = db::creator_tiers::cancel_creator_sub(&db.pool, "sub_cycle_a").await.unwrap();
269 + assert!(canceled.is_some(), "cancel returns the updated row");
270 + assert!(
271 + db::creator_tiers::get_active_creator_tier(&db.pool, user).await.unwrap().is_none(),
272 + "a canceled subscription is no longer the active tier"
273 + );
274 +
275 + // Re-subscribing with a new Stripe subscription id reactivates the row.
276 + let re = db::creator_tiers::create_creator_subscription(
277 + &db.pool, user, "sub_cycle_b", "cus_cycle", CreatorTier::Basic,
278 + )
279 + .await
280 + .unwrap();
281 + assert!(re.is_some(), "a new subscription id after cancel reactivates");
282 + assert_eq!(
283 + db::creator_tiers::get_active_creator_tier(&db.pool, user).await.unwrap(),
284 + Some(CreatorTier::Basic),
285 + "reactivation adopts the new tier",
286 + );
287 + }
288 +
289 + #[tokio::test]
290 + async fn count_active_paying_counts_only_active_subscriptions() {
291 + let db = TestDb::new().await;
292 + let a = seed_user(&db.pool, "ct_count_a").await;
293 + let b = seed_user(&db.pool, "ct_count_b").await;
294 +
295 + db::creator_tiers::create_creator_subscription(&db.pool, a, "sub_ca", "cus_ca", CreatorTier::Basic)
296 + .await
297 + .unwrap();
298 + db::creator_tiers::create_creator_subscription(&db.pool, b, "sub_cb", "cus_cb", CreatorTier::Basic)
299 + .await
300 + .unwrap();
301 + assert_eq!(db::creator_tiers::count_active_paying(&db.pool).await.unwrap(), 2);
302 +
303 + // Cancelling one drops the strict paying count.
304 + db::creator_tiers::cancel_creator_sub(&db.pool, "sub_cb").await.unwrap();
305 + assert_eq!(db::creator_tiers::count_active_paying(&db.pool).await.unwrap(), 1);
306 + }
307 +
308 + // ── 30-day cancellation grace window ─────────────────────────────────────────
309 +
310 + #[tokio::test]
311 + async fn grace_window_opens_on_cancel_and_closes_after_thirty_days() {
312 + let db = TestDb::new().await;
313 + let user = seed_user(&db.pool, "ct_grace").await;
314 +
315 + db::creator_tiers::create_creator_subscription(
316 + &db.pool, user, "sub_grace", "cus_grace", CreatorTier::SmallFiles,
317 + )
318 + .await
319 + .unwrap();
320 + db::creator_tiers::cancel_creator_sub(&db.pool, "sub_grace").await.unwrap();
321 +
322 + // Just-canceled => inside the grace window.
323 + assert!(db::creator_tiers::is_in_grace_period(&db.pool, user).await.unwrap());
324 +
325 + // Backdate the cancellation past 30 days => grace closed.
326 + sqlx::query("UPDATE creator_subscriptions SET canceled_at = NOW() - INTERVAL '31 days' WHERE user_id = $1")
327 + .bind(user)
328 + .execute(&db.pool)
329 + .await
330 + .unwrap();
331 + assert!(!db::creator_tiers::is_in_grace_period(&db.pool, user).await.unwrap());
332 + }
333 +
334 + #[tokio::test]
335 + async fn expired_grace_creators_surface_once_then_mark_enforced() {
336 + let db = TestDb::new().await;
337 + let user = seed_user(&db.pool, "ct_grace_sweep").await;
338 +
339 + db::creator_tiers::create_creator_subscription(
340 + &db.pool, user, "sub_sweep", "cus_sweep", CreatorTier::Basic,
341 + )
342 + .await
343 + .unwrap();
344 + db::creator_tiers::cancel_creator_sub(&db.pool, "sub_sweep").await.unwrap();
345 + // Age the cancellation past the 30-day grace so the sweep picks it up.
346 + sqlx::query("UPDATE creator_subscriptions SET canceled_at = NOW() - INTERVAL '31 days' WHERE user_id = $1")
347 + .bind(user)
348 + .execute(&db.pool)
349 + .await
350 + .unwrap();
351 +
352 + let expired = db::creator_tiers::get_expired_grace_creators(&db.pool).await.unwrap();
353 + assert!(expired.contains(&user), "an over-grace creator surfaces for enforcement");
354 +
355 + // Stamping grace_enforced_at removes it from the next sweep (idempotent tick).
356 + db::creator_tiers::mark_grace_enforced_batch(&db.pool, &[user]).await.unwrap();
357 + let after = db::creator_tiers::get_expired_grace_creators(&db.pool).await.unwrap();
358 + assert!(!after.contains(&user), "an enforced creator is not swept twice");
359 + }
A server/tests/workflows/db_synckit_rotation.rs +281
@@ -0,0 +1,281 @@
1 + //! DB-layer contract tests for `db::synckit::rotation` — the SyncKit encryption
2 + //! key-rotation state machine.
3 + //!
4 + //! Audit Run 16 flagged `rotation` as a Concurrency/Testing cold spot (B+), and
5 + //! Phase 3 of the Run 16 remediation made `begin_key_rotation` transactional
6 + //! (`SELECT ... FOR UPDATE` on the `sync_keys` row) precisely to close a
7 + //! check-then-insert race where two devices both observe "no rotation" and both
8 + //! INSERT. These tests pin that fix and the rest of the begin/complete/cancel
9 + //! lifecycle at the DB layer, calling the `db::synckit::rotation` functions
10 + //! directly against real Postgres.
11 +
12 + use crate::harness::db::TestDb;
13 + use makenotwork::db::synckit;
14 + use makenotwork::db::{SyncAppId, SyncDeviceId, UserId};
15 +
16 + /// Seed a verified user.
17 + async fn seed_user(pool: &sqlx::PgPool, username: &str) -> UserId {
18 + let hash = makenotwork::auth::hash_password("password123").expect("hash");
19 + sqlx::query_scalar::<_, UserId>(
20 + "INSERT INTO users (username, email, password_hash, email_verified)
21 + VALUES ($1, $2, $3, true) RETURNING id",
22 + )
23 + .bind(username)
24 + .bind(format!("{username}@test.com"))
25 + .bind(&hash)
26 + .fetch_one(pool)
27 + .await
28 + .expect("seed user")
29 + }
30 +
31 + /// Seed a sync app owned by `user`.
32 + async fn seed_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
33 + sqlx::query_scalar::<_, SyncAppId>(
34 + "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix)
35 + VALUES ($1, $2, $3, $4) RETURNING id",
36 + )
37 + .bind(user)
38 + .bind(name)
39 + .bind(format!("hash_{name}"))
40 + .bind(&name[..name.len().min(8)])
41 + .fetch_one(pool)
42 + .await
43 + .expect("seed sync app")
44 + }
45 +
46 + /// Seed the app's sync_keys row (key_version=1, key_id=1 by default).
47 + async fn seed_key(pool: &sqlx::PgPool, app: SyncAppId, user: UserId) {
48 + sqlx::query("INSERT INTO sync_keys (app_id, user_id, encrypted_key) VALUES ($1, $2, 'enc_v1')")
49 + .bind(app)
50 + .bind(user)
51 + .execute(pool)
52 + .await
53 + .expect("seed sync key");
54 + }
55 +
56 + /// Seed a device row (`sync_key_rotations.device_id` FKs `sync_devices`).
57 + async fn seed_device(pool: &sqlx::PgPool, app: SyncAppId, user: UserId, name: &str) -> SyncDeviceId {
58 + sqlx::query_scalar::<_, SyncDeviceId>(
59 + "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
60 + VALUES ($1, $2, $3, 'macos') RETURNING id",
61 + )
62 + .bind(app)
63 + .bind(user)
64 + .bind(name)
65 + .fetch_one(pool)
66 + .await
67 + .expect("seed device")
68 + }
69 +
70 + async fn rotation_row_count(pool: &sqlx::PgPool, app: SyncAppId) -> i64 {
71 + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM sync_key_rotations WHERE app_id = $1")
72 + .bind(app)
73 + .fetch_one(pool)
74 + .await
75 + .expect("count rotations")
76 + }
77 +
78 + // ── begin_key_rotation preconditions ─────────────────────────────────────────
79 +
80 + #[tokio::test]
81 + async fn begin_rotation_creates_a_row_at_matching_version() {
82 + let db = TestDb::new().await;
83 + let user = seed_user(&db.pool, "rot_begin").await;
84 + let app = seed_app(&db.pool, user, "rotbegin").await;
85 + seed_key(&db.pool, app, user).await;
86 + let device = seed_device(&db.pool, app, user, "dev").await;
87 +
88 + let out = synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
89 + .await
90 + .unwrap();
91 + let created = out.expect("matching version begins a rotation");
92 + // Fresh key was key_id=1, so the rotation targets key_id=2.
93 + assert_eq!(created.new_key_id, 2);
94 + assert_eq!(created.device_id, device);
95 + assert_eq!(rotation_row_count(&db.pool, app).await, 1);
96 + }
97 +
98 + #[tokio::test]
99 + async fn begin_rotation_rejects_version_mismatch() {
100 + let db = TestDb::new().await;
101 + let user = seed_user(&db.pool, "rot_ver").await;
102 + let app = seed_app(&db.pool, user, "rotver").await;
103 + seed_key(&db.pool, app, user).await;
104 +
105 + // Client believes the key is at version 99; server is at 1 -> caller 409s.
106 + let out = synckit::begin_key_rotation(&db.pool, app, user, SyncDeviceId::new(), "enc_v2", 99)
107 + .await
108 + .unwrap();
109 + assert_eq!(out.err(), Some("key version mismatch"));
110 + assert_eq!(rotation_row_count(&db.pool, app).await, 0, "a mismatch inserts nothing");
111 + }
112 +
113 + #[tokio::test]
114 + async fn begin_rotation_errors_when_no_key_exists() {
115 + let db = TestDb::new().await;
116 + let user = seed_user(&db.pool, "rot_nokey").await;
117 + let app = seed_app(&db.pool, user, "rotnokey").await;
118 + // Deliberately no seed_key.
119 +
120 + let out = synckit::begin_key_rotation(&db.pool, app, user, SyncDeviceId::new(), "enc_v2", 1)
121 + .await
122 + .unwrap();
123 + assert_eq!(out.err(), Some("no encryption key exists"));
124 + }
125 +
126 + #[tokio::test]
127 + async fn begin_rotation_is_resumable_by_the_same_device() {
128 + let db = TestDb::new().await;
129 + let user = seed_user(&db.pool, "rot_resume").await;
130 + let app = seed_app(&db.pool, user, "rotresume").await;
131 + seed_key(&db.pool, app, user).await;
132 + let device = seed_device(&db.pool, app, user, "dev").await;
133 +
134 + let first = synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
135 + .await
136 + .unwrap()
137 + .expect("first begin");
138 + // The same device re-issuing begin resumes the identical rotation, not a new one.
139 + let resumed = synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
140 + .await
141 + .unwrap()
142 + .expect("resume");
143 + assert_eq!(resumed.id, first.id, "the same device resumes its own rotation");
144 + assert_eq!(rotation_row_count(&db.pool, app).await, 1, "resume must not create a second row");
145 + }
146 +
147 + #[tokio::test]
148 + async fn begin_rotation_blocks_a_second_device() {
149 + let db = TestDb::new().await;
150 + let user = seed_user(&db.pool, "rot_block").await;
151 + let app = seed_app(&db.pool, user, "rotblock").await;
152 + seed_key(&db.pool, app, user).await;
153 + let dev_a = seed_device(&db.pool, app, user, "dev-a").await;
154 + let dev_b = seed_device(&db.pool, app, user, "dev-b").await;
155 +
156 + synckit::begin_key_rotation(&db.pool, app, user, dev_a, "enc_v2", 1)
157 + .await
158 + .unwrap()
159 + .expect("first device begins");
160 + let second = synckit::begin_key_rotation(&db.pool, app, user, dev_b, "enc_v2", 1)
161 + .await
162 + .unwrap();
163 + assert_eq!(second.err(), Some("rotation already in progress by another device"));
164 + assert_eq!(rotation_row_count(&db.pool, app).await, 1);
165 + }
166 +
167 + /// The Phase 3 remediation's raison d'être: two devices racing `begin` for the
168 + /// same (app, user). The `FOR UPDATE` on the sync_keys row serializes the
169 + /// check-then-insert, so exactly one wins and exactly one rotation row exists —
170 + /// never two rotations, never a unique-constraint 500.
171 + #[tokio::test]
172 + async fn concurrent_begins_yield_exactly_one_rotation() {
173 + let db = TestDb::new().await;
174 + let user = seed_user(&db.pool, "rot_race").await;
175 + let app = seed_app(&db.pool, user, "rotrace").await;
176 + seed_key(&db.pool, app, user).await;
177 +
178 + let mut handles = Vec::new();
179 + for i in 0..8 {
180 + let pool = db.pool.clone();
181 + let device = seed_device(&db.pool, app, user, &format!("dev-{i}")).await;
182 + handles.push(tokio::spawn(async move {
183 + synckit::begin_key_rotation(&pool, app, user, device, "enc_v2", 1).await
184 + }));
185 + }
186 +
187 + let mut winners = 0;
188 + for h in handles {
189 + // No task may error out (a lost race resolves to Ok(Err(..)), not Err).
190 + let out = h.await.expect("task panicked").expect("begin must not hard-error under contention");
191 + if out.is_ok() {
192 + winners += 1;
193 + }
194 + }
195 +
196 + assert_eq!(winners, 1, "exactly one racing device may open the rotation");
197 + assert_eq!(
198 + rotation_row_count(&db.pool, app).await,
199 + 1,
200 + "the FOR UPDATE serialization must leave exactly one rotation row"
201 + );
202 + }
203 +
204 + // ── complete + cancel ────────────────────────────────────────────────────────
205 +
206 + #[tokio::test]
207 + async fn complete_rotation_with_no_entries_swaps_the_key() {
208 + let db = TestDb::new().await;
209 + let user = seed_user(&db.pool, "rot_complete").await;
210 + let app = seed_app(&db.pool, user, "rotcomp").await;
211 + seed_key(&db.pool, app, user).await;
212 + let device = seed_device(&db.pool, app, user, "dev").await;
213 +
214 + let started = synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
215 + .await
216 + .unwrap()
217 + .expect("begin");
218 +
219 + // No sync_log entries exist, so nothing needs re-encryption: complete succeeds.
220 + let done = synckit::complete_key_rotation(&db.pool, app, user, started.id)
221 + .await
222 + .unwrap();
223 + assert_eq!(done, Ok(started.new_key_id), "complete returns the new key id");
224 +
225 + // The rotation is consumed and the live key advanced.
226 + assert_eq!(rotation_row_count(&db.pool, app).await, 0, "the rotation row is deleted");
227 + let (version, key_id, enc): (i32, i32, String) =
228 + sqlx::query_as("SELECT key_version, key_id, encrypted_key FROM sync_keys WHERE app_id = $1 AND user_id = $2")
229 + .bind(app)
230 + .bind(user)
231 + .fetch_one(&db.pool)
232 + .await
233 + .unwrap();
234 + assert_eq!(version, 2, "key_version is bumped on completion");
235 + assert_eq!(key_id, started.new_key_id, "the new key id is now the active one");
236 + assert_eq!(enc, "enc_v2", "the new encrypted key is now live");
237 + }
238 +
239 + #[tokio::test]
240 + async fn complete_rotation_is_noop_for_unknown_rotation() {
241 + let db = TestDb::new().await;
242 + let user = seed_user(&db.pool, "rot_unknown").await;
243 + let app = seed_app(&db.pool, user, "rotunkn").await;
244 + seed_key(&db.pool, app, user).await;
245 +
246 + // A random rotation id that was never started -> Err(0), no key change.
247 + let out = synckit::complete_key_rotation(&db.pool, app, user, uuid::Uuid::new_v4())
248 + .await
249 + .unwrap();
250 + assert_eq!(out, Err(0));
251 + }
252 +
253 + #[tokio::test]
254 + async fn cancel_stale_rotation_respects_the_age_threshold() {
255 + let db = TestDb::new().await;
256 + let user = seed_user(&db.pool, "rot_stale").await;
257 + let app = seed_app(&db.pool, user, "rotstale").await;
258 + seed_key(&db.pool, app, user).await;
259 + let device = seed_device(&db.pool, app, user, "dev").await;
260 +
261 + synckit::begin_key_rotation(&db.pool, app, user, device, "enc_v2", 1)
262 + .await
263 + .unwrap()
264 + .expect("begin");
265 +
266 + // A fresh rotation is not stale, so a 24h cancel is a no-op.
267 + assert!(
268 + !synckit::cancel_stale_rotation(&db.pool, app, user, 24).await.unwrap(),
269 + "a fresh rotation must not be cancelled as stale"
270 + );
271 + assert_eq!(rotation_row_count(&db.pool, app).await, 1);
272 +
273 + // Age it past the threshold -> the stale sweep removes it, unblocking new begins.
274 + sqlx::query("UPDATE sync_key_rotations SET updated_at = NOW() - INTERVAL '48 hours' WHERE app_id = $1")
275 + .bind(app)
276 + .execute(&db.pool)
277 + .await
278 + .unwrap();
279 + assert!(synckit::cancel_stale_rotation(&db.pool, app, user, 24).await.unwrap());
280 + assert_eq!(rotation_row_count(&db.pool, app).await, 0);
281 + }
A server/tests/workflows/db_webhook_events.rs +292
@@ -0,0 +1,292 @@
1 + //! DB-layer contract tests for `db::webhook_events` — the webhook dedup +
2 + //! retry-queue module.
3 + //!
4 + //! Audit Run 16 flagged `webhook_events` as a Testing/Concurrency cold spot
5 + //! (type B+): the dedup marker and the retry-claim query are exercised only
6 + //! indirectly through the Stripe webhook flow, with "no two-concurrent-
7 + //! deliveries test." These call the `db::webhook_events` functions directly so
8 + //! the two race-sensitive invariants live where they belong:
9 + //!
10 + //! 1. `mark_event_processed` is idempotent under concurrent duplicate delivery
11 + //! (Stripe at-least-once) — N racing marks of one event leave exactly one row.
12 + //! 2. `get_retryable_events` claims-and-defers so an overlapping scheduler tick
13 + //! (or a second replica) can't re-claim the same due events within the window.
14 + //!
15 + //! Plus the backoff/dead-letter progression and the retention prune.
16 +
17 + use crate::harness::db::TestDb;
18 + use makenotwork::db::webhook_events;
19 +
20 + /// Count rows in `processed_webhook_events` (dedup markers).
21 + async fn processed_count(pool: &sqlx::PgPool) -> i64 {
22 + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM processed_webhook_events")
23 + .fetch_one(pool)
24 + .await
25 + .expect("count processed")
26 + }
27 +
28 + /// Insert a failed event and return its id, backdating `next_retry_at` into the
29 + /// past so it is immediately due (the column defaults to NOW()+60s).
30 + async fn seed_due_event(pool: &sqlx::PgPool, event_type: &str) -> uuid::Uuid {
31 + webhook_events::insert_failed_event(pool, "stripe", event_type, "{}", None, "boom")
32 + .await
33 + .expect("insert failed event");
34 + sqlx::query_scalar::<_, uuid::Uuid>(
35 + "UPDATE webhook_events SET next_retry_at = NOW() - INTERVAL '1 minute' \
36 + WHERE event_type = $1 RETURNING id",
37 + )
38 + .bind(event_type)
39 + .fetch_one(pool)
40 + .await
41 + .expect("backdate next_retry_at")
42 + }
43 +
44 + // ── dedup marker (`processed_webhook_events`) ────────────────────────────────
45 +
46 + #[tokio::test]
47 + async fn is_event_processed_false_before_mark_true_after() {
48 + let db = TestDb::new().await;
49 + let id = "evt_dedup_basic";
50 +
51 + assert!(
52 + !webhook_events::is_event_processed(&db.pool, id).await.unwrap(),
53 + "an unseen event must not read as processed"
54 + );
55 +
56 + webhook_events::mark_event_processed(&db.pool, id).await.unwrap();
57 +
58 + assert!(
59 + webhook_events::is_event_processed(&db.pool, id).await.unwrap(),
60 + "a marked event must read as processed"
61 + );
62 + }
63 +
64 + #[tokio::test]
65 + async fn mark_event_processed_is_idempotent() {
66 + let db = TestDb::new().await;
67 + let id = "evt_dedup_idem";
68 +
69 + // Serial re-marks (redelivery) must not error and must not duplicate the row.
70 + webhook_events::mark_event_processed(&db.pool, id).await.unwrap();
71 + webhook_events::mark_event_processed(&db.pool, id).await.unwrap();
72 + webhook_events::mark_event_processed(&db.pool, id).await.unwrap();
73 +
74 + assert_eq!(processed_count(&db.pool).await, 1, "re-marking must be a no-op");
75 + }
76 +
77 + /// The core concurrency invariant: Stripe delivers at-least-once, so two
78 + /// deliveries of the same event can race the post-handler mark. `ON CONFLICT DO
79 + /// NOTHING` must collapse them to exactly one dedup row — never a duplicate-key
80 + /// error, never two rows.
81 + #[tokio::test]
82 + async fn mark_event_processed_concurrent_duplicates_leave_one_row() {
83 + let db = TestDb::new().await;
84 + let id = "evt_dedup_race";
85 +
86 + let mut handles = Vec::new();
87 + for _ in 0..24 {
88 + let pool = db.pool.clone();
89 + let id = id.to_string();
90 + handles.push(tokio::spawn(async move {
91 + webhook_events::mark_event_processed(&pool, &id).await
92 + }));
93 + }
94 + for h in handles {
95 + // No task may observe a unique-violation — the ON CONFLICT swallows it.
96 + h.await.expect("task panicked").expect("mark must not error under contention");
97 + }
98 +
99 + assert_eq!(
100 + processed_count(&db.pool).await,
101 + 1,
102 + "concurrent duplicate deliveries must yield exactly one dedup row"
103 + );
104 + }
105 +
106 + #[tokio::test]
107 + async fn prune_processed_events_respects_retention_boundary() {
108 + let db = TestDb::new().await;
109 +
110 + webhook_events::mark_event_processed(&db.pool, "evt_old").await.unwrap();
111 + webhook_events::mark_event_processed(&db.pool, "evt_fresh").await.unwrap();
112 + // Age the old marker past a 30-day retention; leave the fresh one at NOW().
113 + sqlx::query("UPDATE processed_webhook_events SET processed_at = NOW() - INTERVAL '40 days' WHERE event_id = 'evt_old'")
114 + .execute(&db.pool)
115 + .await
116 + .unwrap();
117 +
118 + let deleted = webhook_events::prune_processed_events(&db.pool, 30).await.unwrap();
119 + assert_eq!(deleted, 1, "exactly the aged marker is pruned");
120 +
121 + assert!(!webhook_events::is_event_processed(&db.pool, "evt_old").await.unwrap());
122 + assert!(
123 + webhook_events::is_event_processed(&db.pool, "evt_fresh").await.unwrap(),
124 + "a within-retention marker must survive the prune"
125 + );
126 + }
127 +
128 + // ── retry queue (`webhook_events`) claim-and-defer ───────────────────────────
129 +
130 + /// A due event is claimed once, and the same immediate tick can't re-claim it:
131 + /// `get_retryable_events` pushes `next_retry_at` out by 2 minutes as it selects,
132 + /// so an overlapping scheduler tick (or a second replica) gets nothing. This is
133 + /// the "two-concurrent-deliveries" guard the audit noted was untested.
134 + #[tokio::test]
135 + async fn get_retryable_events_claims_and_defers() {
136 + let db = TestDb::new().await;
137 + seed_due_event(&db.pool, "claim.a").await;
138 + seed_due_event(&db.pool, "claim.b").await;
139 +
140 + let first = webhook_events::get_retryable_events(&db.pool).await.unwrap();
141 + assert_eq!(first.len(), 2, "both due events are claimed on the first tick");
142 +
143 + let second = webhook_events::get_retryable_events(&db.pool).await.unwrap();
144 + assert!(
145 + second.is_empty(),
146 + "a claimed event's next_retry_at is deferred, so an immediate re-tick claims nothing"
147 + );
148 + }
149 +
150 + /// Two overlapping ticks racing the same due events must partition them — never
151 + /// hand the same event to both callers (would double-run the handler).
152 + #[tokio::test]
153 + async fn get_retryable_events_concurrent_ticks_do_not_double_claim() {
154 + let db = TestDb::new().await;
155 + for i in 0..6 {
156 + seed_due_event(&db.pool, &format!("race.claim.{i}")).await;
157 + }
158 +
159 + let p1 = db.pool.clone();
160 + let p2 = db.pool.clone();
161 + let (a, b) = tokio::join!(
162 + tokio::spawn(async move { webhook_events::get_retryable_events(&p1).await }),
163 + tokio::spawn(async move { webhook_events::get_retryable_events(&p2).await }),
164 + );
165 + let a = a.unwrap().unwrap();
166 + let b = b.unwrap().unwrap();
167 +
168 + let mut ids: Vec<_> = a.iter().chain(b.iter()).map(|e| e.id).collect();
169 + let total = ids.len();
170 + ids.sort();
171 + ids.dedup();
172 + assert_eq!(
173 + ids.len(),
174 + total,
175 + "no event id may be claimed by both concurrent ticks"
176 + );
177 + assert_eq!(ids.len(), 6, "every due event is claimed exactly once across the two ticks");
178 + }
179 +
180 + #[tokio::test]
181 + async fn get_retryable_events_excludes_future_and_exhausted() {
182 + let db = TestDb::new().await;
183 +
184 + // Due now — eligible.
185 + seed_due_event(&db.pool, "elig.due").await;
186 + // Scheduled in the future — not yet eligible.
187 + webhook_events::insert_failed_event(&db.pool, "stripe", "elig.future", "{}", None, "e")
188 + .await
189 + .unwrap();
190 + // Due but retry-exhausted (attempts >= 5) — must be skipped so it can't storm.
191 + let exhausted = seed_due_event(&db.pool, "elig.exhausted").await;
192 + sqlx::query("UPDATE webhook_events SET attempts = 5 WHERE id = $1")
193 + .bind(exhausted)
194 + .execute(&db.pool)
195 + .await
196 + .unwrap();
197 +
198 + let due = webhook_events::get_retryable_events(&db.pool).await.unwrap();
199 + let types: Vec<_> = due.iter().map(|e| e.event_type.as_str()).collect();
200 + assert_eq!(types, ["elig.due"], "only the due, non-exhausted event is returned");
201 + }
202 +
203 + // ── backoff + dead-letter progression ────────────────────────────────────────
204 +
205 + #[tokio::test]
206 + async fn schedule_retry_sets_retrying_and_backs_off() {
207 + let db = TestDb::new().await;
208 + let id = seed_due_event(&db.pool, "backoff.evt").await;
209 +
210 + // First failed retry: attempt count 1, status -> retrying, deferred forward.
211 + webhook_events::schedule_retry(&db.pool, id, 1, "still failing").await.unwrap();
212 +
213 + let (status, attempts, deferred): (String, i32, bool) = sqlx::query_as(
214 + "SELECT status, attempts, next_retry_at > NOW() FROM webhook_events WHERE id = $1",
215 + )
216 + .bind(id)
217 + .fetch_one(&db.pool)
218 + .await
219 + .unwrap();
220 + assert_eq!(status, "retrying");
221 + assert_eq!(attempts, 1);
222 + assert!(deferred, "next_retry_at is pushed into the future by the backoff");
223 +
224 + // Immediately after scheduling, it is not due, so a tick skips it.
225 + assert!(
226 + webhook_events::get_retryable_events(&db.pool).await.unwrap().is_empty(),
227 + "a just-rescheduled event is not yet due"
228 + );
229 + }
230 +
231 + #[tokio::test]
232 + async fn schedule_retry_dead_letters_at_max_attempts() {
233 + let db = TestDb::new().await;
234 + let id = seed_due_event(&db.pool, "dead.evt").await;
235 +
236 + // Reaching the max attempts marks the event dead — off the retry queue,
237 + // onto the operator dead-letter list.
238 + webhook_events::schedule_retry(&db.pool, id, 5, "exhausted").await.unwrap();
239 +
240 + let status: String = sqlx::query_scalar("SELECT status FROM webhook_events WHERE id = $1")
241 + .bind(id)
242 + .fetch_one(&db.pool)
243 + .await
244 + .unwrap();
245 + assert_eq!(status, "dead");
246 +
247 + assert!(
248 + webhook_events::get_retryable_events(&db.pool).await.unwrap().is_empty(),
249 + "a dead event must never be re-claimed for retry"
250 + );
251 + let dead = webhook_events::get_dead_events(&db.pool).await.unwrap();
252 + assert!(dead.iter().any(|e| e.id == id), "a dead event surfaces on the operator list");
253 + }
254 +
255 + #[tokio::test]
256 + async fn retry_dead_event_resurrects_only_dead_rows() {
257 + let db = TestDb::new().await;
258 +
259 + // A live (failed) event is not a dead-letter, so resetting it is a no-op.
260 + let live = seed_due_event(&db.pool, "resurrect.live").await;
261 + assert!(
262 + !webhook_events::retry_dead_event(&db.pool, live).await.unwrap(),
263 + "retry_dead_event must only act on status = 'dead'"
264 + );
265 +
266 + // A genuinely dead event flips back to 'failed' and becomes claimable again.
267 + let dead = seed_due_event(&db.pool, "resurrect.dead").await;
268 + webhook_events::schedule_retry(&db.pool, dead, 5, "exhausted").await.unwrap();
269 + assert!(webhook_events::retry_dead_event(&db.pool, dead).await.unwrap());
270 +
271 + let status: String = sqlx::query_scalar("SELECT status FROM webhook_events WHERE id = $1")
272 + .bind(dead)
273 + .fetch_one(&db.pool)
274 + .await
275 + .unwrap();
276 + assert_eq!(status, "failed", "a resurrected event returns to the retry queue");
277 + }
278 +
279 + #[tokio::test]
280 + async fn mark_processed_removes_event_from_queue() {
281 + let db = TestDb::new().await;
282 + let id = seed_due_event(&db.pool, "done.evt").await;
283 +
284 + webhook_events::mark_processed(&db.pool, id).await.unwrap();
285 +
286 + let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events WHERE id = $1")
287 + .bind(id)
288 + .fetch_one(&db.pool)
289 + .await
290 + .unwrap();
291 + assert_eq!(remaining, 0, "a successfully processed event is deleted from the queue");
292 + }
M server/tests/workflows/mod.rs +3
@@ -98,4 +98,7 @@ mod cart;
98 98 mod bundles;
99 99 mod idempotency;
100 100 mod db_payments_layer;
101 + mod db_creator_tiers_layer;
102 + mod db_synckit_rotation;
103 + mod db_webhook_events;
101 104 mod security_idor_moderation;