Skip to main content

max / makenotwork

synckit: ship blob-delete API + fix silently-broken sync_log retention (ultra-fuzz --deep Storage A+) Storage A->A+ cold spots from the Run 3 SyncKit pass: - Blob-delete API (F3): DELETE /api/v1/sync/blobs/{hash}. db::synckit:: delete_sync_blob removes the row, refunds the developer-billing counters, and dead-letters the S3 object (bucket "synckit") in one transaction, so a delete can never overstate bytes_stored or orphan the object on a crash. This is the live storage shrink path that previously only the weekly drift job provided. Idempotent; allowed regardless of billing status. - new-device replay-after-compaction (latent): compact_all_sync_logs filtered WHERE last_pulled_seq > 0, excluding a freshly-registered device sitting at seq 0 from the cursor MIN -- so compaction could delete entries that device had never pulled, and its first pull-from-0 would silently miss them. Now every device counts; a not-yet-pulled device pins MIN to 0 and blocks the pair until it catches up. prune_sync_log's retention is the abandoned-device backstop. - pre-existing bug surfaced by the new test: prune_sync_log / compact_all_sync_logs / compact_sync_log bound bigint into make_interval(days => $n), which errors (days is int) -- so BOTH daily sync_log maintenance jobs have been failing whenever they had rows to delete (the monitor swallowed the warn). The log was effectively unbounded at runtime. Added ::int casts to match monitor.rs / webhook_events.rs. Tests: +3 (blob delete frees space + dead-letters, idempotent no-op delete, compaction waits for a fresh device then compacts). Full synckit workflow suite green (52); clippy clean. db::synckit made pub for the test crate (mirrors db::idempotency). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 19:53 UTC
Commit: f28489d81c0cf97c857445ad41bcee5510e3e4f7
Parent: 167097e
6 files changed, +246 insertions, -13 deletions
M server/src/db/mod.rs +1 -1
@@ -24,7 +24,7 @@ pub(crate) mod auth;
24 24 pub mod waitlist;
25 25 pub(crate) mod blog_posts;
26 26 pub(crate) mod license_keys;
27 - pub(crate) mod synckit;
27 + pub mod synckit; // pub so the integration test crate can exercise compaction directly
28 28 pub mod synckit_billing;
29 29 pub(crate) mod oauth;
30 30 pub(crate) mod promo_codes;
M server/src/db/synckit/blobs.rs +86
@@ -274,6 +274,92 @@ pub async fn confirm_developer_blob(
274 274 Ok(BlobConfirm::Stored)
275 275 }
276 276
277 + /// Outcome of a blob delete. Terminal for one call.
278 + #[derive(Debug, Clone, PartialEq, Eq)]
279 + pub enum BlobDelete {
280 + /// Row removed; S3 object dead-lettered; usage counters refunded.
281 + Deleted { size_bytes: i64 },
282 + /// No blob with that hash for this `(app, user)` — idempotent no-op.
283 + NotFound,
284 + }
285 +
286 + /// Delete a blob by hash for a user within an app. Atomic in one transaction:
287 + /// removes the `sync_blobs` row, refunds the developer-billing counters, and
288 + /// dead-letters the S3 object via `pending_s3_deletions`. This is the sole
289 + /// shrink path for storage usage that doesn't wait on the weekly drift job —
290 + /// and because the row delete, the counter refund, and the S3-delete enqueue
291 + /// commit together, a delete can never leave `bytes_stored` overstating reality
292 + /// or orphan the object on a mid-operation crash.
293 + ///
294 + /// Internal (first-party) apps keep no counter — `confirm_internal_blob` sums
295 + /// usage straight from `sync_blobs` — so for them the counter UPDATEs match no
296 + /// row and are harmless no-ops; removing the row is the whole refund. Idempotent:
297 + /// deleting an absent blob returns `NotFound` with nothing written.
298 + #[tracing::instrument(skip_all)]
299 + pub async fn delete_sync_blob(
300 + pool: &PgPool,
301 + app_id: SyncAppId,
302 + user_id: UserId,
303 + hash: &str,
304 + ) -> Result<BlobDelete> {
305 + let mut tx = pool.begin().await?;
306 +
307 + // Remove the row, capturing what we need to refund counters and dead-letter
308 + // the object. SyncKit blob keys are per-`(app, user, hash)` (no cross-user
309 + // dedup), so deleting this row's object frees only this user's copy.
310 + let row: Option<(i64, String, String)> = sqlx::query_as(
311 + "DELETE FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3
312 + RETURNING size_bytes, s3_key, key",
313 + )
314 + .bind(app_id)
315 + .bind(user_id)
316 + .bind(hash)
317 + .fetch_optional(&mut *tx)
318 + .await?;
319 +
320 + let Some((size_bytes, s3_key, key)) = row else {
321 + tx.commit().await?;
322 + return Ok(BlobDelete::NotFound);
323 + };
324 +
325 + // Refund the developer-billing counters. `GREATEST(_, 0)` floors at zero so
326 + // a drifted counter can never go negative; internal apps have no usage row
327 + // and these UPDATEs touch nothing.
328 + sqlx::query(
329 + "UPDATE sync_app_usage_current
330 + SET bytes_stored = GREATEST(bytes_stored - $2, 0), updated_at = NOW()
331 + WHERE app_id = $1",
332 + )
333 + .bind(app_id)
334 + .bind(size_bytes)
335 + .execute(&mut *tx)
336 + .await?;
337 + sqlx::query(
338 + "UPDATE sync_key_usage_current
339 + SET bytes_stored = GREATEST(bytes_stored - $3, 0), updated_at = NOW()
340 + WHERE app_id = $1 AND key = $2",
341 + )
342 + .bind(app_id)
343 + .bind(&key)
344 + .bind(size_bytes)
345 + .execute(&mut *tx)
346 + .await?;
347 +
348 + // Dead-letter the object in the SAME tx as the row delete + refund (model:
349 + // `versions::delete_version`). After commit the row is gone, so the key is
350 + // non-live and the deletion worker (`retry_pending_s3_deletions`, which
351 + // routes `bucket = "synckit"` to `synckit_s3`) can act on it.
352 + crate::db::pending_s3_deletions::enqueue_deletions(
353 + &mut *tx,
354 + &[(s3_key, "synckit".to_string())],
355 + "synckit_blob_delete",
356 + )
357 + .await?;
358 +
359 + tx.commit().await?;
360 + Ok(BlobDelete::Deleted { size_bytes })
361 + }
362 +
277 363 /// Count devices registered for a user/app pair.
278 364 #[tracing::instrument(skip_all)]
279 365 pub async fn count_sync_devices(
M server/src/db/synckit/keys.rs +12 -6
@@ -123,7 +123,7 @@ pub async fn prune_sync_log(pool: &PgPool, retain_days: i64) -> Result<u64> {
123 123 return Ok(0);
124 124 }
125 125 let result = sqlx::query(
126 - "DELETE FROM sync_log WHERE created_at < NOW() - make_interval(days => $1)",
126 + "DELETE FROM sync_log WHERE created_at < NOW() - make_interval(days => $1::int)",
127 127 )
128 128 .bind(retain_days)
129 129 .execute(pool)
@@ -187,7 +187,7 @@ pub async fn compact_sync_log(
187 187 DELETE FROM sync_log
188 188 WHERE app_id = $1 AND user_id = $2
189 189 AND seq <= $3
190 - AND created_at < NOW() - make_interval(days => $4)
190 + AND created_at < NOW() - make_interval(days => $4::int)
191 191 "#,
192 192 )
193 193 .bind(app_id)
@@ -211,13 +211,19 @@ pub async fn compact_all_sync_logs(pool: &PgPool, min_age_days: i64) -> Result<u
211 211 return Ok(0);
212 212 }
213 213
214 - // Find (app_id, user_id) pairs where compaction is possible:
215 - // all devices have pulled past seq 0, and there are old entries to delete.
214 + // Find (app_id, user_id) pairs where compaction is safe: EVERY registered
215 + // device has pulled past the compaction floor. We deliberately do NOT filter
216 + // `last_pulled_seq > 0` in the WHERE — a freshly-registered device sits at
217 + // seq 0 until its first pull, and excluding it would let us delete entries
218 + // it has never seen, so its first pull-from-0 would silently miss changes.
219 + // Including every device means a single not-yet-pulled device pins MIN to 0
220 + // and the `HAVING MIN > 0` skips the pair entirely until it catches up. A
221 + // truly-abandoned never-pulled device is bounded by `prune_sync_log`'s
222 + // retention backstop, not this cursor path.
216 223 let pairs: Vec<(SyncAppId, UserId, i64)> = sqlx::query_as(
217 224 r#"
218 225 SELECT app_id, user_id, MIN(last_pulled_seq) AS min_seq
219 226 FROM sync_devices
220 - WHERE last_pulled_seq > 0
221 227 GROUP BY app_id, user_id
222 228 HAVING MIN(last_pulled_seq) > 0
223 229 "#,
@@ -232,7 +238,7 @@ pub async fn compact_all_sync_logs(pool: &PgPool, min_age_days: i64) -> Result<u
232 238 DELETE FROM sync_log
233 239 WHERE app_id = $1 AND user_id = $2
234 240 AND seq <= $3
235 - AND created_at < NOW() - make_interval(days => $4)
241 + AND created_at < NOW() - make_interval(days => $4::int)
236 242 "#,
237 243 )
238 244 .bind(app_id)
M server/src/routes/synckit/blobs.rs +23 -6
@@ -267,9 +267,26 @@ pub(super) async fn blob_download_url(
267 267 Ok(Json(BlobDownloadUrlResponse { download_url }).into_response())
268 268 }
269 269
270 - // NOTE: No active blob-delete API path was found in v1. Storage counters can
271 - // only grow (until period rollover or manual reset via reset_period_usage,
272 - // which only resets egress, not storage). The weekly drift correction job in
273 - // `db::synckit_billing::recalculate_synckit_app_storage` reconciles
274 - // `bytes_stored` against `sync_blobs` and is the only way storage can shrink
275 - // without DB intervention. Known limitation; revisit when blob deletion ships.
270 + /// Delete a blob by hash for the authenticated user.
271 + ///
272 + /// Frees storage immediately: the row, the S3 object, and the usage counters
273 + /// are released in one atomic step (see `db::synckit::delete_sync_blob`). This
274 + /// is the live shrink path that the weekly drift job used to be the only source
275 + /// of. Allowed regardless of billing status — a user must always be able to
276 + /// reclaim space, even on a lapsed subscription. Idempotent: deleting a hash
277 + /// that isn't stored returns 204.
278 + #[utoipa::path(delete, path = "/api/v1/sync/blobs/{hash}", tag = "SyncKit",
279 + params(("hash" = String, Path, description = "Content hash of the blob to delete")),
280 + responses((status = 204, description = "Blob deleted (or already absent)")),
281 + security(("bearer" = [])),
282 + )]
283 + #[tracing::instrument(skip_all, name = "synckit::blob_delete")]
284 + pub(super) async fn blob_delete(
285 + State(state): State<AppState>,
286 + sync_user: SyncUser,
287 + axum::extract::Path(hash): axum::extract::Path<String>,
288 + ) -> Result<Response> {
289 + validation::validate_sync_blob_hash(&hash)?;
290 + db::synckit::delete_sync_blob(&state.db, sync_user.app_id, sync_user.user_id, &hash).await?;
291 + Ok(StatusCode::NO_CONTENT.into_response())
292 + }
M server/src/routes/synckit/mod.rs +2
@@ -609,6 +609,8 @@ pub fn synckit_routes(synckit_jwt_secret: Option<std::sync::Arc<String>>) -> Csr
609 609 .route("/api/v1/sync/blobs/confirm", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_confirm_upload))
610 610 .route("/api/sync/blobs/download", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_download_url))
611 611 .route("/api/v1/sync/blobs/download", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_download_url))
612 + .route("/api/sync/blobs/{hash}", delete_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_delete))
613 + .route("/api/v1/sync/blobs/{hash}", delete_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_delete))
612 614 // Per-app rate limit (inner layer runs first): prevents one developer's
613 615 // app from starving other apps. Extracts app ID from JWT payload.
614 616 .route_layer(GovernorLayer {
M server/tests/workflows/synckit_paid_sync.rs +122
@@ -225,3 +225,125 @@ async fn confirm_uses_authoritative_object_size_not_client_claim() {
225 225 let body: serde_json::Value = resp.json();
226 226 assert_eq!(body["reason"], "storage_limit_reached");
227 227 }
228 +
229 + #[tokio::test]
230 + async fn delete_blob_removes_row_dead_letters_object_and_frees_space() {
231 + let (mut h, blobs) = harness_with_blobs().await;
232 + let user_id = h.signup("paid_del", "paid_del@example.com", "Password1!").await;
233 + let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
234 + seed_subscription(&h.db, user_id, app_id, "active", 1500).await; // 1500-byte cap
235 + auth_as(&mut h, user_id, app_id, "user-key");
236 +
237 + // Fill most of the cap with one 1000-byte blob.
238 + let hash = fake_hash(0x06);
239 + let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash, 1000, 1000).await;
240 + assert_eq!(resp.status, 204, "first confirm should succeed: {}", resp.text);
241 +
242 + // A second 1000-byte blob would breach the 1500 cap (1000 + 1000 > 1500).
243 + let hash2 = fake_hash(0x07);
244 + let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash2, 1000, 1000).await;
245 + assert_eq!(resp.status, 402, "second confirm should breach the cap: {}", resp.text);
246 +
247 + // Delete the first blob.
248 + let s3_key = format!("{app_id}/{user_id}/{hash}");
249 + let resp = h.client.delete(&format!("/api/sync/blobs/{hash}")).await;
250 + assert_eq!(resp.status, 204, "delete should succeed: {}", resp.text);
251 +
252 + // Row is gone.
253 + let remaining: i64 = sqlx::query_scalar(
254 + "SELECT COUNT(*) FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
255 + )
256 + .bind(app_id)
257 + .bind(user_id)
258 + .bind(&hash)
259 + .fetch_one(&h.db)
260 + .await
261 + .unwrap();
262 + assert_eq!(remaining, 0, "deleted blob row must be gone");
263 +
264 + // The S3 object was dead-lettered for the synckit bucket, not dropped.
265 + let dead_lettered: i64 = sqlx::query_scalar(
266 + "SELECT COUNT(*) FROM pending_s3_deletions
267 + WHERE s3_key = $1 AND bucket = 'synckit' AND source = 'synckit_blob_delete'",
268 + )
269 + .bind(&s3_key)
270 + .fetch_one(&h.db)
271 + .await
272 + .unwrap();
273 + assert_eq!(dead_lettered, 1, "deleted object must be enqueued to the dead-letter ladder");
274 +
275 + // Space is freed: the previously-rejected second blob now fits (internal
276 + // usage is summed from sync_blobs, so removing the row shrinks usage).
277 + let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash2, 1000, 1000).await;
278 + assert_eq!(resp.status, 204, "after delete, the second blob should fit: {}", resp.text);
279 + }
280 +
281 + #[tokio::test]
282 + async fn compaction_waits_for_a_freshly_registered_device() {
283 + // Regression: a new device sits at last_pulled_seq = 0 until its first pull.
284 + // Compaction must NOT delete log entries below the other devices' cursors
285 + // while a never-pulled device exists, or that device's first pull-from-0
286 + // would silently miss the compacted-away changes.
287 + let (mut h, _blobs) = harness_with_blobs().await;
288 + let user_id = h.signup("compact_dev", "compact_dev@example.com", "Password1!").await;
289 + let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
290 +
291 + // Device A has pulled; device B is freshly registered (seq 0).
292 + let dev_a: uuid::Uuid = sqlx::query_scalar(
293 + "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
294 + VALUES ($1, $2, 'A', 'test') RETURNING id",
295 + )
296 + .bind(app_id).bind(user_id).fetch_one(&h.db).await.unwrap();
297 + let _dev_b: uuid::Uuid = sqlx::query_scalar(
298 + "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
299 + VALUES ($1, $2, 'B', 'test') RETURNING id",
300 + )
301 + .bind(app_id).bind(user_id).fetch_one(&h.db).await.unwrap();
302 +
303 + // One old log entry (backdated well past the 7-day compaction margin).
304 + let seq: i64 = sqlx::query_scalar(
305 + "INSERT INTO sync_log
306 + (app_id, user_id, device_id, table_name, operation, row_id, client_timestamp, created_at)
307 + VALUES ($1, $2, $3, 'tasks', 'INSERT', 'r1', NOW(), NOW() - INTERVAL '30 days')
308 + RETURNING seq",
309 + )
310 + .bind(app_id).bind(user_id).bind(dev_a).fetch_one(&h.db).await.unwrap();
311 +
312 + // Device A has pulled past the entry; device B has not pulled at all.
313 + sqlx::query("UPDATE sync_devices SET last_pulled_seq = $1 WHERE id = $2")
314 + .bind(seq).bind(dev_a).execute(&h.db).await.unwrap();
315 +
316 + // With device B pinned at 0, compaction must delete nothing.
317 + let deleted = makenotwork::db::synckit::compact_all_sync_logs(&h.db, 7).await.unwrap();
318 + assert_eq!(deleted, 0, "compaction must wait for the never-pulled device");
319 + let present: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sync_log WHERE seq = $1")
320 + .bind(seq).fetch_one(&h.db).await.unwrap();
321 + assert_eq!(present, 1, "log entry must survive while a new device sits at seq 0");
322 +
323 + // Once device B catches up, the same entry is safely compactable.
324 + sqlx::query("UPDATE sync_devices SET last_pulled_seq = $1 WHERE device_name = 'B'")
325 + .bind(seq).execute(&h.db).await.unwrap();
326 + let deleted = makenotwork::db::synckit::compact_all_sync_logs(&h.db, 7).await.unwrap();
327 + assert_eq!(deleted, 1, "after every device caught up, the old entry compacts");
328 + }
329 +
330 + #[tokio::test]
331 + async fn delete_absent_blob_is_idempotent_no_op() {
332 + let (mut h, _blobs) = harness_with_blobs().await;
333 + let user_id = h.signup("paid_del2", "paid_del2@example.com", "Password1!").await;
334 + let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
335 + seed_subscription(&h.db, user_id, app_id, "active", GIB).await;
336 + auth_as(&mut h, user_id, app_id, "user-key");
337 +
338 + // Deleting a hash that was never stored is a 204 no-op, and enqueues nothing.
339 + let resp = h.client.delete(&format!("/api/sync/blobs/{}", fake_hash(0x08))).await;
340 + assert_eq!(resp.status, 204, "deleting an absent blob should be a 204 no-op: {}", resp.text);
341 +
342 + let enqueued: i64 = sqlx::query_scalar(
343 + "SELECT COUNT(*) FROM pending_s3_deletions WHERE source = 'synckit_blob_delete'",
344 + )
345 + .fetch_one(&h.db)
346 + .await
347 + .unwrap();
348 + assert_eq!(enqueued, 0, "no-op delete must not dead-letter anything");
349 + }