Skip to main content

max / makenotwork

7.9 KB · 221 lines History Blame Raw
1 //! Hourly SyncKit usage warning emails.
2 //!
3 //! Walks every `active`, non-internal sync app and computes the highest
4 //! warning band (75/90/100%) currently breached above the previously-stamped
5 //! `last_warning_pct`. For each breach, emails the app owner and stamps the
6 //! threshold so we don't re-fire on subsequent ticks.
7 //!
8 //! Thresholds reset at period rollover via `reset_period_usage` (called from
9 //! the `invoice.paid` webhook handler), so each billing cycle can fire warnings
10 //! again from scratch.
11 //!
12 //! Storage breaches do NOT reset at period rollover (the cap is absolute, not
13 //! per-period), but `last_warning_pct` is shared across both dimensions for
14 //! simplicity. A creator who silenced storage warnings at 75% and then crosses
15 //! egress 75% in the same cycle won't get re-notified for egress 75%, they'll
16 //! get the next band (90%/100%) instead. This is an acceptable v1 quirk; if
17 //! it bites we'll split `last_warning_pct` into two columns.
18
19 use std::sync::Arc;
20
21 use crate::AppState;
22 use crate::constants::SYNCKIT_WARNINGS_PER_TICK;
23 use crate::db::synckit_billing::{self, WarningCandidate};
24 use crate::email::EmailClient;
25
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 /// 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.
36 #[tracing::instrument(skip_all)]
37 pub(super) async fn check_and_send_warnings(state: &AppState) {
38 let candidates =
39 match synckit_billing::get_apps_needing_warning(&state.db, SYNCKIT_WARNINGS_PER_TICK).await
40 {
41 Ok(c) => c,
42 Err(e) => {
43 tracing::error!(error = ?e, "synckit warnings: query failed");
44 return;
45 }
46 };
47
48 if candidates.is_empty() {
49 return;
50 }
51 tracing::info!(count = candidates.len(), "synckit warnings: enqueuing");
52
53 for c in candidates {
54 let email = state.email.clone();
55 let db = state.db.clone();
56 let host_url = state.config.host_url.clone();
57 // Enqueue the slow send (and its stamp) onto the bounded pool so the
58 // tick returns immediately. Fire-and-forget: on queue overflow the
59 // task is dropped and the band re-fires next tick.
60 state.bg.spawn("synckit-usage-warning", async move {
61 send_warning(email, db, host_url, c).await;
62 });
63 }
64 }
65
66 /// Send one usage-warning email, then stamp the band so it doesn't re-fire.
67 /// Runs on the background pool, never on the scheduler tick. A send failure
68 /// skips the stamp so the band re-fires next tick (preferred over losing the
69 /// notification); a stamp failure logs and also re-fires.
70 async fn send_warning(
71 email: EmailClient,
72 db: sqlx::PgPool,
73 host_url: Arc<str>,
74 c: WarningCandidate,
75 ) {
76 let url = format!("{host_url}/sync/apps/{}/billing", c.app_id);
77 if let Err(e) = email
78 .send_synckit_usage_warning(
79 &c.creator_email,
80 &c.app_name,
81 c.dimension,
82 c.key.as_deref(),
83 c.threshold_pct,
84 c.used,
85 c.limit,
86 &url,
87 )
88 .await
89 {
90 tracing::error!(
91 error = ?e,
92 app_id = %c.app_id,
93 dimension = c.dimension,
94 key = c.key.as_deref().unwrap_or(""),
95 pct = c.threshold_pct,
96 "synckit warnings: send failed",
97 );
98 return;
99 }
100 // Stamp the band on the right table: per-key warnings stamp the per-key
101 // row, app-wide stamps the app row.
102 let stamp = match c.key.as_deref() {
103 Some(k) => synckit_billing::update_key_warning_pct(&db, c.app_id, k, c.threshold_pct).await,
104 None => synckit_billing::update_warning_pct(&db, c.app_id, c.threshold_pct).await,
105 };
106 if let Err(e) = stamp {
107 tracing::error!(
108 error = ?e,
109 app_id = %c.app_id,
110 key = c.key.as_deref().unwrap_or(""),
111 "synckit warnings: stamp failed (will re-fire next tick)",
112 );
113 }
114 }
115
116 #[cfg(test)]
117 mod tests {
118 use crate::db::synckit_billing::highest_breached_threshold;
119
120 /// Seal for the warning loop. The send must only ever happen on the bounded
121 /// background pool, inside `send_warning`, never inline in the scheduler
122 /// tick. This fails the build if `check_and_send_warnings` ever names the
123 /// email send directly, so a serial send inside the tick cannot appear
124 /// unnoticed.
125 #[test]
126 fn tick_never_sends_inline() {
127 let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
128 .join("src/scheduler/synckit_warnings.rs");
129 let src = std::fs::read_to_string(&path).expect("read warnings source");
130
131 let start = src
132 .find("async fn check_and_send_warnings")
133 .expect("tick fn present");
134 let rest = &src[start..];
135 // The tick body ends where the next top-level fn (send_warning) begins.
136 let end = rest[1..]
137 .find("\nasync fn ")
138 .or_else(|| rest[1..].find("\nfn "))
139 .map_or(rest.len(), |i| i + 1);
140 let tick_body = &rest[..end];
141
142 assert!(
143 !tick_body.contains("send_synckit_usage_warning"),
144 "CHRONIC seal violated: check_and_send_warnings must not perform an \
145 email send inline, enqueue onto state.bg and send in send_warning."
146 );
147 assert!(
148 tick_body.contains("state.bg.spawn"),
149 "warning tick must dispatch sends onto the bounded background pool",
150 );
151 }
152
153 #[test]
154 fn no_breach_below_first_threshold() {
155 // 50% used, no warning yet
156 assert!(highest_breached_threshold(50, 100, 0).is_none());
157 }
158
159 #[test]
160 fn first_breach_at_75() {
161 assert_eq!(highest_breached_threshold(75, 100, 0), Some(75));
162 assert_eq!(highest_breached_threshold(80, 100, 0), Some(75));
163 }
164
165 #[test]
166 fn no_re_fire_at_same_band() {
167 // 80% used, already warned at 75%, don't fire again at 75%.
168 assert!(highest_breached_threshold(80, 100, 75).is_none());
169 }
170
171 #[test]
172 fn next_band_fires_after_previous() {
173 // 90% used, last fired at 75%, fire 90%.
174 assert_eq!(highest_breached_threshold(90, 100, 75), Some(90));
175 }
176
177 #[test]
178 fn jumps_to_highest_band_only() {
179 // 100% used after only firing 75%, fire 100%, skip 90%.
180 assert_eq!(highest_breached_threshold(100, 100, 75), Some(100));
181 }
182
183 #[test]
184 fn overshoot_caps_at_100() {
185 // 150% used, none fired, still goes straight to 100% (highest band).
186 assert_eq!(highest_breached_threshold(150, 100, 0), Some(100));
187 }
188
189 #[test]
190 fn nothing_above_100_fires() {
191 // Already at 100%; further growth doesn't trigger anything new.
192 assert!(highest_breached_threshold(200, 100, 100).is_none());
193 }
194
195 #[test]
196 fn zero_limit_is_safe() {
197 // Defensive: don't divide by zero.
198 assert!(highest_breached_threshold(50, 0, 0).is_none());
199 }
200
201 #[test]
202 fn exactly_75_percent() {
203 // Floor of 75.0 is 75, should breach the 75 band.
204 assert_eq!(highest_breached_threshold(75, 100, 0), Some(75));
205 }
206
207 #[test]
208 fn just_under_75_does_not_fire() {
209 // 74.99%, floor is 74, below the 75 threshold.
210 assert!(highest_breached_threshold(74, 100, 0).is_none());
211 }
212
213 #[test]
214 fn large_bytes_no_overflow() {
215 // 800 GiB of 1 TiB, no warning yet.
216 let used: i64 = 800 * 1024 * 1024 * 1024;
217 let limit: i64 = 1024 * 1024 * 1024 * 1024;
218 assert_eq!(highest_breached_threshold(used, limit, 0), Some(75));
219 }
220 }
221