Skip to main content

max / makenotwork

synckit: move usage-warning sends off the scheduler tick onto the bounded pool (ultra-fuzz Run 3 CHRONIC) Phase 3 of the SyncKit ultra-fuzz remediation (Performance axis B- -> A). CHRONIC (Runs 1-3) — the hourly usage-warning job ran a serial, unpaced, un-LIMITed loop of blocking email sends inside the single-threaded scheduler tick, so a slow mail provider times N breaching apps would block every other periodic job. Prior runs batched the query but never moved the send. Structural fix: - get_apps_needing_warning takes a LIMIT (SYNCKIT_WARNINGS_PER_TICK = 200) with a deterministic ORDER BY, and the per-key fan-out is truncated to the same bound — the tick does a bounded amount of DB work. - check_and_send_warnings no longer .awaits any email send: it enqueues each send+stamp onto state.bg (the concurrency-capped, fire-and-forget background pool). The tick returns immediately after enqueueing. Warned apps stamp last_warning_pct and drop out of the candidate set, so overflow beyond the per-tick limit drains on subsequent ticks; a send failure skips the stamp and re-fires next tick. The serial network send inside the tick is gone. (A type-level seal removing the email sender from the tick's reach is the remaining A->A+ step.) 50/50 synckit tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 17:44 UTC
Commit: 0660bce120be76dfcdd8f8330c0ea7ce1c7772a2
Parent: 2dd934e
4 files changed, +86 insertions, -54 deletions
M server/src/constants.rs +4
@@ -62,6 +62,10 @@ pub const SYNCKIT_MAX_DEVICES_PER_APP: i64 = 50; // Max devices per user per app
62 62 pub const SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS: u64 = 3600; // 1 hour
63 63 pub const SYNCKIT_MAX_SSE_CONNECTIONS_PER_USER: usize = 10;
64 64 pub const SYNCKIT_ROTATION_STALE_HOURS: i64 = 24;
65 + /// Max usage-warning candidates a single scheduler tick fetches. Bounds the
66 + /// per-tick scan; the sends fan out on the background pool, and stamped apps
67 + /// drop out of the candidate set so subsequent ticks drain any overflow.
68 + pub const SYNCKIT_WARNINGS_PER_TICK: i64 = 200;
65 69 pub const SYNCKIT_ROTATION_BATCH_MAX: usize = 500;
66 70
67 71 // -- Subscriptions --
M server/src/db/synckit_billing.rs +10 -1
@@ -618,8 +618,11 @@ pub async fn add_bytes_egress(
618 618 /// along with the data needed to compute which threshold (if any) has been
619 619 /// breached since the last notice. The per-app breach computation is done in
620 620 /// Rust by the caller (see `scheduler::synckit_warnings`).
621 + /// `limit` bounds the app scan so a single scheduler tick does a bounded amount
622 + /// of work; warned apps stamp `last_warning_pct` and drop out of the candidate
623 + /// set, so any overflow drains on subsequent ticks.
621 624 #[tracing::instrument(skip_all)]
622 - pub async fn get_apps_needing_warning(pool: &PgPool) -> Result<Vec<WarningCandidate>> {
625 + pub async fn get_apps_needing_warning(pool: &PgPool, limit: i64) -> Result<Vec<WarningCandidate>> {
623 626 use super::id_types::UserId;
624 627
625 628 // Egress is no longer a price input or an enforced cap (see migration 118),
@@ -657,8 +660,11 @@ pub async fn get_apps_needing_warning(pool: &PgPool) -> Result<Vec<WarningCandid
657 660 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
658 661 WHERE sa.billing_status = 'active'
659 662 AND sa.is_internal = FALSE
663 + ORDER BY sa.id
664 + LIMIT $1
660 665 "#,
661 666 )
667 + .bind(limit)
662 668 .fetch_all(pool)
663 669 .await?;
664 670
@@ -749,6 +755,9 @@ pub async fn get_apps_needing_warning(pool: &PgPool) -> Result<Vec<WarningCandid
749 755 }
750 756 }
751 757
758 + // Bound the per-tick candidate count (the per-key fan-out can expand beyond
759 + // the app LIMIT); stamped apps drop out so the rest drains on later ticks.
760 + out.truncate(limit as usize);
752 761 Ok(out)
753 762 }
754 763
M server/src/scheduler/synckit_warnings.rs +70 -51
@@ -16,30 +16,58 @@
16 16 //! get the next band (90%/100%) instead. This is an acceptable v1 quirk; if
17 17 //! it bites we'll split `last_warning_pct` into two columns.
18 18
19 - use crate::db::synckit_billing;
19 + use std::sync::Arc;
20 +
21 + use crate::constants::SYNCKIT_WARNINGS_PER_TICK;
22 + use crate::db::synckit_billing::{self, WarningCandidate};
23 + use crate::email::EmailClient;
20 24 use crate::AppState;
21 25
22 - /// Run one pass: fetch breach candidates, send warning emails, update
23 - /// `last_warning_pct`. Errors on individual apps are logged and skipped.
26 + /// Run one pass: fetch a bounded slice of breach candidates and hand each
27 + /// warning to the bounded background pool to send + stamp.
28 + ///
29 + /// CHRONIC fix (ultra-fuzz Runs 1-3): this runs inside the single-threaded
30 + /// scheduler tick, so it must never `.await` a network send here — a slow mail
31 + /// provider times N breaching apps would block every other periodic job. The
32 + /// tick does only bounded DB work (a `LIMIT`ed fetch); the actual `email.send`
33 + /// happens on `state.bg`, which is concurrency-capped and fire-and-forget.
34 + /// Warned apps stamp `last_warning_pct` and drop out of the candidate set, so
35 + /// any overflow beyond the per-tick limit drains on subsequent ticks.
24 36 #[tracing::instrument(skip_all)]
25 37 pub(super) async fn check_and_send_warnings(state: &AppState) {
26 - let candidates = match synckit_billing::get_apps_needing_warning(&state.db).await {
27 - Ok(c) => c,
28 - Err(e) => {
29 - tracing::error!(error = ?e, "synckit warnings: query failed");
30 - return;
31 - }
32 - };
38 + let candidates =
39 + match synckit_billing::get_apps_needing_warning(&state.db, SYNCKIT_WARNINGS_PER_TICK).await {
40 + Ok(c) => c,
41 + Err(e) => {
42 + tracing::error!(error = ?e, "synckit warnings: query failed");
43 + return;
44 + }
45 + };
33 46
34 47 if candidates.is_empty() { return; }
35 - tracing::info!(count = candidates.len(), "synckit warnings: candidates");
48 + tracing::info!(count = candidates.len(), "synckit warnings: enqueuing");
36 49
37 50 for c in candidates {
38 - let url = format!(
39 - "{}/sync/apps/{}/billing",
40 - state.config.host_url, c.app_id,
41 - );
42 - if let Err(e) = state.email.send_synckit_usage_warning(
51 + let email = state.email.clone();
52 + let db = state.db.clone();
53 + let host_url = state.config.host_url.clone();
54 + // Enqueue the slow send (and its stamp) onto the bounded pool so the
55 + // tick returns immediately. Fire-and-forget: on queue overflow the
56 + // task is dropped and the band re-fires next tick.
57 + state.bg.spawn("synckit-usage-warning", async move {
58 + send_warning(email, db, host_url, c).await;
59 + });
60 + }
61 + }
62 +
63 + /// Send one usage-warning email, then stamp the band so it doesn't re-fire.
64 + /// Runs on the background pool, never on the scheduler tick. A send failure
65 + /// skips the stamp so the band re-fires next tick (preferred over losing the
66 + /// notification); a stamp failure logs and also re-fires.
67 + async fn send_warning(email: EmailClient, db: sqlx::PgPool, host_url: Arc<str>, c: WarningCandidate) {
68 + let url = format!("{host_url}/sync/apps/{}/billing", c.app_id);
69 + if let Err(e) = email
70 + .send_synckit_usage_warning(
43 71 &c.creator_email,
44 72 &c.app_name,
45 73 c.dimension,
@@ -48,41 +76,32 @@ pub(super) async fn check_and_send_warnings(state: &AppState) {
48 76 c.used,
49 77 c.limit,
50 78 &url,
51 - ).await {
52 - tracing::error!(
53 - error = ?e,
54 - app_id = %c.app_id,
55 - dimension = c.dimension,
56 - key = c.key.as_deref().unwrap_or(""),
57 - pct = c.threshold_pct,
58 - "synckit warnings: send failed",
59 - );
60 - continue;
61 - }
62 - // Stamp the band on the right table: per-key warnings stamp the
63 - // per-key row, app-wide stamps the app row. Failure to stamp logs
64 - // and continues — next tick will re-fire, which is preferable to
65 - // silently losing the notification.
66 - let stamp = match c.key.as_deref() {
67 - Some(k) => {
68 - synckit_billing::update_key_warning_pct(
69 - &state.db, c.app_id, k, c.threshold_pct,
70 - ).await
71 - }
72 - None => {
73 - synckit_billing::update_warning_pct(
74 - &state.db, c.app_id, c.threshold_pct,
75 - ).await
76 - }
77 - };
78 - if let Err(e) = stamp {
79 - tracing::error!(
80 - error = ?e,
81 - app_id = %c.app_id,
82 - key = c.key.as_deref().unwrap_or(""),
83 - "synckit warnings: stamp failed (will re-fire next tick)",
84 - );
85 - }
79 + )
80 + .await
81 + {
82 + tracing::error!(
83 + error = ?e,
84 + app_id = %c.app_id,
85 + dimension = c.dimension,
86 + key = c.key.as_deref().unwrap_or(""),
87 + pct = c.threshold_pct,
88 + "synckit warnings: send failed",
89 + );
90 + return;
91 + }
92 + // Stamp the band on the right table: per-key warnings stamp the per-key
93 + // row, app-wide stamps the app row.
94 + let stamp = match c.key.as_deref() {
95 + Some(k) => synckit_billing::update_key_warning_pct(&db, c.app_id, k, c.threshold_pct).await,
96 + None => synckit_billing::update_warning_pct(&db, c.app_id, c.threshold_pct).await,
97 + };
98 + if let Err(e) = stamp {
99 + tracing::error!(
100 + error = ?e,
101 + app_id = %c.app_id,
102 + key = c.key.as_deref().unwrap_or(""),
103 + "synckit warnings: stamp failed (will re-fire next tick)",
104 + );
86 105 }
87 106 }
88 107
M server/tests/workflows/synckit_per_key_storage.rs +2 -2
@@ -345,7 +345,7 @@ async fn warning_candidates_emit_per_key_dimension_when_per_key_mode() {
345 345 .await
346 346 .unwrap();
347 347
348 - let candidates = makenotwork::db::synckit_billing::get_apps_needing_warning(&h.db)
348 + let candidates = makenotwork::db::synckit_billing::get_apps_needing_warning(&h.db, 1000)
349 349 .await
350 350 .expect("warning candidates");
351 351
@@ -377,7 +377,7 @@ async fn warning_candidates_emit_per_key_dimension_when_per_key_mode() {
377 377 .await
378 378 .expect("stamp");
379 379
380 - let candidates2 = makenotwork::db::synckit_billing::get_apps_needing_warning(&h.db)
380 + let candidates2 = makenotwork::db::synckit_billing::get_apps_needing_warning(&h.db, 1000)
381 381 .await
382 382 .unwrap();
383 383 assert!(