Skip to main content

max / makenotwork

perf: scan timeouts on both paths, supervise+lock monitor, background appeal fan-out Drive Performance to A (ultra-fuzz Run 7). Perf-S1: the buffered scan() path lacked the aggregate external-lookup timeout that scan_stream applies, so a small upload had only the per-request client timeout. Extract a shared lookup_with_timeout helper used by both paths (also dedups the inline timeout in scan_stream). Perf-S2: the urlhaus host loop had only a per-host timeout, so up to MAX_HOSTS_PER_FILE sequential lookups could stack (~25s) holding a scan worker. Wrap the loop in an aggregate deadline. The urlhaus/metadefender/hash_lookup reqwest clients had no timeouts at all — build all three with .timeout() and .connect_timeout() (new SCAN_HTTP_CONNECT_TIMEOUT_SECS). Perf-S3: the monitor's outer loop was unsupervised — a panic killed it and silently stopped all health checks, alerting, and daily maintenance. Wrap run_monitor_loop in a re-spawning supervisor (panic -> restart, via tokio's task-boundary panic catching, no new dep). The daily prune/compaction block now runs under a MONITOR_PRUNE_LOCK_ID advisory lock so it can't double-execute across replicas, and a per-tick record_job_run("health_monitor") heartbeat makes a stalled monitor externally detectable. Perf-MIN: admin/moderation appeal-resume now backgrounds via spawn_fan_sub_fanout (the 6th fan-out site) instead of blocking the request; the issue-notification and 403-page bundle N+1s are batched (get_users_by_ids / get_public_items_by_ids). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-24 22:46 UTC
Commit: 69ab513027aa33d668c739bd2df4a59d883e7e50
Parent: 5540d16
9 files changed, +634 insertions, -363 deletions
M server/src/constants.rs +8 -1
@@ -257,6 +257,10 @@ pub const SCAN_ZIP_MAX_UNCOMPRESSED: u64 = 2 * 1024 * 1024 * 1024; // 2 GB uncom
257 257 // far past any legitimate sample pack / content bundle; beyond it, fail closed.
258 258 pub const SCAN_ZIP_MAX_ENTRIES: usize = 100_000;
259 259 pub const SCAN_MALWAREBAZAAR_TIMEOUT_SECS: u64 = 5;
260 + /// TCP connect timeout for the external-lookup HTTP clients (MalwareBazaar,
261 + /// MetaDefender, URLhaus). Bounds the time spent establishing a connection to a
262 + /// hung/blackholed host so a stalled connect can't pin a scan worker (Perf-S2).
263 + pub const SCAN_HTTP_CONNECT_TIMEOUT_SECS: u64 = 5;
260 264 pub const SCAN_CLAMAV_TIMEOUT_SECS: u64 = 30;
261 265 // How often the background probe pings clamd to maintain the runtime liveness
262 266 // flag. The boot gate proves clamd was up at startup; this catches a death
@@ -526,7 +530,10 @@ mod tests {
526 530 fn build_allowed_targets_are_os_slash_arch() {
527 531 for target in BUILD_ALLOWED_TARGETS {
528 532 assert!(!target.is_empty());
529 - assert!(target.contains('/'), "target should be os/arch format: {target}");
533 + assert!(
534 + target.contains('/'),
535 + "target should be os/arch format: {target}"
536 + );
530 537 }
531 538 }
532 539 }
M server/src/monitor.rs +253 -201
@@ -11,9 +11,9 @@ use std::time::Instant;
11 11 use tokio::sync::watch;
12 12 use tokio::task::JoinHandle;
13 13
14 + use crate::AppState;
14 15 use crate::constants;
15 16 use crate::db;
16 - use crate::AppState;
17 17
18 18 /// Overall service status.
19 19 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -65,12 +65,11 @@ pub async fn run_health_check(state: &AppState) -> HealthSnapshot {
65 65 };
66 66
67 67 // 3. Sessions: probe the session table
68 - let sessions_ok = sqlx::query_scalar::<_, bool>(
69 - "SELECT EXISTS(SELECT 1 FROM tower_sessions.session)",
70 - )
71 - .fetch_one(&state.db)
72 - .await
73 - .is_ok();
68 + let sessions_ok =
69 + sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM tower_sessions.session)")
70 + .fetch_one(&state.db)
71 + .await
72 + .is_ok();
74 73
75 74 let elapsed = start.elapsed();
76 75 let check_duration_ms = elapsed.as_millis().min(i32::MAX as u128) as i32;
@@ -93,365 +92,467 @@ pub async fn run_health_check(state: &AppState) -> HealthSnapshot {
93 92 }
94 93
95 94 /// Spawn the background monitor loop. Drop `shutdown_tx` to stop it.
96 - pub fn spawn_monitor(
97 - state: AppState,
98 - mut shutdown_rx: watch::Receiver<()>,
99 - ) -> JoinHandle<()> {
95 + pub fn spawn_monitor(state: AppState, shutdown_rx: watch::Receiver<()>) -> JoinHandle<()> {
96 + // Supervise the loop (Perf-S3): a panic in the loop body would otherwise kill
97 + // the task and silently stop all health checks, alerting, and daily
98 + // maintenance — with no alert, since the alerting lives in the same loop.
99 + // Re-spawn on panic so monitoring survives; a clean shutdown ends supervision.
100 100 tokio::spawn(async move {
101 - let alert_email = std::env::var("ALERT_EMAIL").ok();
102 - match &alert_email {
103 - Some(email) => tracing::info!(alert_email = %email, "health monitor started"),
104 - None => tracing::info!("Health monitor started (ALERT_EMAIL not set — alerts disabled)"),
101 + loop {
102 + match tokio::spawn(run_monitor_loop(state.clone(), shutdown_rx.clone())).await {
103 + Ok(()) => return, // clean shutdown
104 + Err(e) if e.is_panic() => {
105 + tracing::error!(error = ?e, "health monitor loop panicked; restarting in 5s");
106 + tokio::time::sleep(std::time::Duration::from_secs(5)).await;
107 + }
108 + Err(_) => return, // task cancelled
109 + }
105 110 }
111 + })
112 + }
106 113
107 - let interval_secs = std::env::var("HEALTH_CHECK_INTERVAL_SECS")
108 - .ok()
109 - .and_then(|v| v.parse::<u64>().ok())
110 - .unwrap_or(constants::HEALTH_CHECK_INTERVAL_SECS);
111 -
112 - let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
113 - // Skip (not burst) missed ticks so a slow health check doesn't cause a
114 - // catch-up burst on the next wake.
115 - interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
116 - interval.tick().await; // consume immediate first tick
117 -
118 - let mut previous_status: Option<MonitorStatus> = None;
119 - let mut last_alert_at: Option<Instant> = None;
120 - let mut last_pool_alert_at: Option<Instant> = None;
121 - // Hysteresis arming flag for DB pool pressure. True = next crossing
122 - // above the high threshold is allowed to fire a ticket. Resets to
123 - // true whenever pressure drops below the low threshold. Starts
124 - // true so the first overrun after boot can alert.
125 - let mut pool_pressure_armed: bool = true;
126 - let mut last_pg_activity_alert_at: Option<Instant> = None;
127 - let mut prune_counter: u64 = 0;
114 + /// The health-monitor loop body. Extracted from `spawn_monitor` so a panic here is
115 + /// caught at the task boundary by the supervisor (Perf-S3) instead of unwinding the
116 + /// whole monitor task and silently halting background maintenance.
117 + async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>) {
118 + let alert_email = std::env::var("ALERT_EMAIL").ok();
119 + match &alert_email {
120 + Some(email) => tracing::info!(alert_email = %email, "health monitor started"),
121 + None => tracing::info!("Health monitor started (ALERT_EMAIL not set — alerts disabled)"),
122 + }
128 123
129 - loop {
130 - tokio::select! {
131 - _ = interval.tick() => {}
132 - _ = shutdown_rx.changed() => {
133 - tracing::info!("Health monitor shutting down");
134 - return;
135 - }
124 + let interval_secs = std::env::var("HEALTH_CHECK_INTERVAL_SECS")
125 + .ok()
126 + .and_then(|v| v.parse::<u64>().ok())
127 + .unwrap_or(constants::HEALTH_CHECK_INTERVAL_SECS);
128 +
129 + let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
130 + // Skip (not burst) missed ticks so a slow health check doesn't cause a
131 + // catch-up burst on the next wake.
132 + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
133 + interval.tick().await; // consume immediate first tick
134 +
135 + let mut previous_status: Option<MonitorStatus> = None;
136 + let mut last_alert_at: Option<Instant> = None;
137 + let mut last_pool_alert_at: Option<Instant> = None;
138 + // Hysteresis arming flag for DB pool pressure. True = next crossing
139 + // above the high threshold is allowed to fire a ticket. Resets to
140 + // true whenever pressure drops below the low threshold. Starts
141 + // true so the first overrun after boot can alert.
142 + let mut pool_pressure_armed: bool = true;
143 + let mut last_pg_activity_alert_at: Option<Instant> = None;
144 + let mut prune_counter: u64 = 0;
145 +
146 + loop {
147 + tokio::select! {
148 + _ = interval.tick() => {}
149 + _ = shutdown_rx.changed() => {
150 + tracing::info!("Health monitor shutting down");
151 + return;
136 152 }
153 + }
137 154
138 - let snap = run_health_check(&state).await;
139 -
140 - // Update Prometheus gauges on every tick. DB pool + domain cache
141 - // are local and cheap; the pg_stat_activity probe is one fast
142 - // query against the shared Postgres. Storage fill aggregates
143 - // across all paying creators via a JOIN of users + creator_subs +
144 - // tier caps — sub-second today but grows linearly with paying
145 - // creator count, so we cache the result for 5 minutes rather
146 - // than burning a multi-second query 2880×/day at 10k+ creators.
147 - crate::metrics::record_db_pool_stats(&state.db);
148 - crate::metrics::record_domain_cache_size(state.domain_cache.len());
149 - static STORAGE_FILL_LAST: std::sync::OnceLock<
150 - std::sync::Mutex<std::time::Instant>,
151 - > = std::sync::OnceLock::new();
152 - const STORAGE_FILL_TTL: std::time::Duration =
153 - std::time::Duration::from_secs(300);
154 - let last_lock = STORAGE_FILL_LAST.get_or_init(|| {
155 - // Initialize to "long ago" so the first tick refreshes.
156 - std::sync::Mutex::new(
157 - std::time::Instant::now() - STORAGE_FILL_TTL,
158 - )
159 - });
160 - let should_refresh = {
161 - // Recover from a poisoned lock rather than panicking the monitor
162 - // loop — the critical section is a trivial timestamp swap.
163 - let mut last = last_lock.lock().unwrap_or_else(|e| e.into_inner());
164 - if last.elapsed() >= STORAGE_FILL_TTL {
165 - *last = std::time::Instant::now();
166 - true
167 - } else {
168 - false
169 - }
170 - };
171 - if should_refresh {
172 - crate::metrics::record_storage_fill_stats(&state.db).await;
173 - crate::metrics::record_custom_pages_stats(&state.db).await;
155 + let snap = run_health_check(&state).await;
156 +
157 + // Heartbeat (Perf-S3): stamp this tick so an external watcher (the PoM
158 + // health contract / dashboard) can detect a stalled monitor — a stale
159 + // `health_monitor` last_ran_at means the loop died and the supervisor
160 + // couldn't bring it back.
161 + if let Err(e) = db::scheduler_jobs::record_job_run(&state.db, "health_monitor", 0).await {
162 + tracing::warn!(error = ?e, "failed to record health-monitor heartbeat");
163 + }
164 +
165 + // Update Prometheus gauges on every tick. DB pool + domain cache
166 + // are local and cheap; the pg_stat_activity probe is one fast
167 + // query against the shared Postgres. Storage fill aggregates
168 + // across all paying creators via a JOIN of users + creator_subs +
169 + // tier caps — sub-second today but grows linearly with paying
170 + // creator count, so we cache the result for 5 minutes rather
171 + // than burning a multi-second query 2880×/day at 10k+ creators.
172 + crate::metrics::record_db_pool_stats(&state.db);
173 + crate::metrics::record_domain_cache_size(state.domain_cache.len());
174 + static STORAGE_FILL_LAST: std::sync::OnceLock<std::sync::Mutex<std::time::Instant>> =
175 + std::sync::OnceLock::new();
176 + const STORAGE_FILL_TTL: std::time::Duration = std::time::Duration::from_secs(300);
177 + let last_lock = STORAGE_FILL_LAST.get_or_init(|| {
178 + // Initialize to "long ago" so the first tick refreshes.
179 + std::sync::Mutex::new(std::time::Instant::now() - STORAGE_FILL_TTL)
180 + });
181 + let should_refresh = {
182 + // Recover from a poisoned lock rather than panicking the monitor
183 + // loop — the critical section is a trivial timestamp swap.
184 + let mut last = last_lock.lock().unwrap_or_else(|e| e.into_inner());
185 + if last.elapsed() >= STORAGE_FILL_TTL {
186 + *last = std::time::Instant::now();
187 + true
188 + } else {
189 + false
174 190 }
191 + };
192 + if should_refresh {
193 + crate::metrics::record_storage_fill_stats(&state.db).await;
194 + crate::metrics::record_custom_pages_stats(&state.db).await;
195 + }
175 196
176 - // Log status changes. Skip the bootstrap None->Operational transition
177 - // so clean restarts don't fire a spurious "recovered" alert.
178 - let status_changed = previous_status != Some(snap.status);
179 - let is_bootstrap_ok =
180 - previous_status.is_none() && snap.status == MonitorStatus::Operational;
181 - if status_changed && !is_bootstrap_ok {
182 - match snap.status {
183 - MonitorStatus::Operational => {
184 - if previous_status.is_some() {
185 - tracing::info!(duration_ms = snap.check_duration_ms, "health recovered, operational");
186 - }
187 - }
188 - MonitorStatus::Degraded => {
189 - tracing::warn!(
190 - db = snap.db_ok, s3 = snap.s3_ok, sessions = snap.sessions_ok,
191 - duration_ms = snap.check_duration_ms, "health degraded"
192 - );
193 - }
194 - MonitorStatus::Error => {
195 - tracing::error!(
196 - db = snap.db_ok, s3 = snap.s3_ok, sessions = snap.sessions_ok,
197 - duration_ms = snap.check_duration_ms, "health error"
197 + // Log status changes. Skip the bootstrap None->Operational transition
198 + // so clean restarts don't fire a spurious "recovered" alert.
199 + let status_changed = previous_status != Some(snap.status);
200 + let is_bootstrap_ok =
201 + previous_status.is_none() && snap.status == MonitorStatus::Operational;
202 + if status_changed && !is_bootstrap_ok {
203 + match snap.status {
204 + MonitorStatus::Operational => {
205 + if previous_status.is_some() {
206 + tracing::info!(
207 + duration_ms = snap.check_duration_ms,
208 + "health recovered, operational"
198 209 );
199 210 }
200 211 }
212 + MonitorStatus::Degraded => {
213 + tracing::warn!(
214 + db = snap.db_ok,
215 + s3 = snap.s3_ok,
216 + sessions = snap.sessions_ok,
217 + duration_ms = snap.check_duration_ms,
218 + "health degraded"
219 + );
220 + }
221 + MonitorStatus::Error => {
222 + tracing::error!(
223 + db = snap.db_ok,
224 + s3 = snap.s3_ok,
225 + sessions = snap.sessions_ok,
226 + duration_ms = snap.check_duration_ms,
227 + "health error"
228 + );
229 + }
230 + }
201 231
202 - // Status-change notifications (admin alert + user notifications) share
203 - // a single cooldown so a flapping monitor cannot spam either audience.
204 - let cooldown_elapsed = last_alert_at
205 - .is_none_or(|t| t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS);
206 -
207 - if cooldown_elapsed {
208 - if let Some(ref to) = alert_email {
209 - let (subject, body) = build_alert(previous_status, &snap);
210 - match state.email.send_alert(to, &subject, &body).await {
211 - Ok(()) => tracing::info!(recipient = %to, "alert email sent"),
212 - Err(e) => tracing::error!(error = ?e, "failed to send alert email"),
213 - }
232 + // Status-change notifications (admin alert + user notifications) share
233 + // a single cooldown so a flapping monitor cannot spam either audience.
234 + let cooldown_elapsed = last_alert_at
235 + .is_none_or(|t| t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS);
236 +
237 + if cooldown_elapsed {
238 + if let Some(ref to) = alert_email {
239 + let (subject, body) = build_alert(previous_status, &snap);
240 + match state.email.send_alert(to, &subject, &body).await {
241 + Ok(()) => tracing::info!(recipient = %to, "alert email sent"),
242 + Err(e) => tracing::error!(error = ?e, "failed to send alert email"),
214 243 }
244 + }
215 245
216 - // Notify opted-in users of status changes (fire-and-forget).
217 - // Batched pacing: pause 1s every 50 sends (matching the
218 - // scheduler's announcement fan-out) instead of a hard 100ms
219 - // between every send, which turned a 1000-subscriber notify
220 - // into a ~100s single task holding the email-client clone.
221 - // The subscriber list is bounded by the query's LIMIT.
222 - {
223 - let pool = state.db.clone();
224 - let email_client = state.email.clone();
225 - let host_url = state.config.host_url.clone();
226 - let signing_secret = state.config.signing_secret.clone();
227 - let current_status = snap.status.as_str().to_string();
228 - let prev_status = previous_status.map_or("unknown", |s| s.as_str()).to_string();
229 - tokio::spawn(async move {
230 - match db::users::get_status_alert_subscribers(&pool).await {
231 - Ok(subscribers) if !subscribers.is_empty() => {
232 - tracing::info!(count = subscribers.len(), "sending status notifications to opted-in users");
233 - for (i, sub) in subscribers.iter().enumerate() {
234 - if i > 0 && i % 50 == 0 {
235 - tokio::time::sleep(std::time::Duration::from_secs(1)).await;
236 - }
237 - let unsub_url = crate::email::generate_unsubscribe_url(
238 - &host_url, sub.id, crate::email::UnsubscribeAction::Status, &sub.id.to_string(), &signing_secret,
239 - );
240 - let _ = email_client.send_status_notification(
246 + // Notify opted-in users of status changes (fire-and-forget).
247 + // Batched pacing: pause 1s every 50 sends (matching the
248 + // scheduler's announcement fan-out) instead of a hard 100ms
249 + // between every send, which turned a 1000-subscriber notify
250 + // into a ~100s single task holding the email-client clone.
251 + // The subscriber list is bounded by the query's LIMIT.
252 + {
253 + let pool = state.db.clone();
254 + let email_client = state.email.clone();
255 + let host_url = state.config.host_url.clone();
256 + let signing_secret = state.config.signing_secret.clone();
257 + let current_status = snap.status.as_str().to_string();
258 + let prev_status = previous_status
259 + .map_or("unknown", |s| s.as_str())
260 + .to_string();
261 + tokio::spawn(async move {
262 + match db::users::get_status_alert_subscribers(&pool).await {
263 + Ok(subscribers) if !subscribers.is_empty() => {
264 + tracing::info!(
265 + count = subscribers.len(),
266 + "sending status notifications to opted-in users"
267 + );
268 + for (i, sub) in subscribers.iter().enumerate() {
269 + if i > 0 && i % 50 == 0 {
270 + tokio::time::sleep(std::time::Duration::from_secs(1)).await;
271 + }
272 + let unsub_url = crate::email::generate_unsubscribe_url(
273 + &host_url,
274 + sub.id,
275 + crate::email::UnsubscribeAction::Status,
276 + &sub.id.to_string(),
277 + &signing_secret,
278 + );
279 + let _ = email_client
280 + .send_status_notification(
241 281 &sub.email,
242 282 sub.display_name.as_deref(),
243 283 &current_status,
244 284 &prev_status,
245 285 &unsub_url,
246 - ).await;
247 - }
248 - }
249 - Err(e) => {
250 - tracing::error!(error = ?e, "failed to query status alert subscribers");
286 + )
287 + .await;
251 288 }
252 - _ => {}
253 289 }
254 - });
255 - }
256 -
257 - // Only record the cooldown timestamp once we've issued at least
258 - // one notification path (admin or subscribers); admin block above
259 - // is gated on `alert_email` being set, but subscriber fan-out is
260 - // always spawned. Always-set is fine since the goal is "don't
261 - // re-notify within ALERT_COOLDOWN_SECS regardless of audience."
262 - last_alert_at = Some(Instant::now());
290 + Err(e) => {
291 + tracing::error!(error = ?e, "failed to query status alert subscribers");
292 + }
293 + _ => {}
294 + }
295 + });
263 296 }
264 297
265 - // Create WAM ticket on degradation/error transitions
266 - if snap.status != MonitorStatus::Operational
267 - && let Some(ref wam) = state.wam
268 - {
269 - let priority = match snap.status {
270 - MonitorStatus::Error => "critical",
271 - MonitorStatus::Degraded => "high",
272 - MonitorStatus::Operational => unreachable!(),
273 - };
274 - let title = format!("Health status: {}", snap.status.as_str());
275 - let body = format!(
276 - "db: {}\ns3: {}\nsessions: {}\ncheck_ms: {}",
277 - snap.db_ok, snap.s3_ok, snap.sessions_ok, snap.check_duration_ms,
278 - );
279 - wam.create_ticket(&title, Some(&body), priority, "health-status-change", None).await;
280 - }
298 + // Only record the cooldown timestamp once we've issued at least
299 + // one notification path (admin or subscribers); admin block above
300 + // is gated on `alert_email` being set, but subscriber fan-out is
301 + // always spawned. Always-set is fine since the goal is "don't
302 + // re-notify within ALERT_COOLDOWN_SECS regardless of audience."
303 + last_alert_at = Some(Instant::now());
281 304 }
282 305
283 - previous_status = Some(snap.status);
284 -
285 - // DB pool pressure check with hysteresis: open at 80%, only
286 - // re-alert after pressure drops below 60% and climbs back over
287 - // 80%. Without hysteresis, a load that oscillates around the
288 - // threshold (e.g. each scheduler tick briefly maxes the pool)
289 - // spams a ticket every ALERT_COOLDOWN_SECS window.
306 + // Create WAM ticket on degradation/error transitions
307 + if snap.status != MonitorStatus::Operational
308 + && let Some(ref wam) = state.wam
290 309 {
291 - let pool_size = state.db.size();
292 - let pool_idle = state.db.num_idle() as u32;
293 - let active = pool_size.saturating_sub(pool_idle);
294 - let pct = (active * 100).checked_div(pool_size).unwrap_or(0);
295 - let high = 80u32;
296 - let low = 60u32;
297 -
298 - if pct > high {
299 - tracing::warn!(pool_size, active, idle = pool_idle, pct, "DB pool pressure >80%");
300 - // Only fire a ticket on the rising edge (was below the low
301 - // threshold last we recovered). The cooldown is still here
302 - // as a backstop in case the hysteresis state machine ever
303 - // gets confused.
304 - let cooldown_ok = last_pool_alert_at
305 - .is_none_or(|t| t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS);
306 - if cooldown_ok && pool_pressure_armed && let Some(ref wam) = state.wam {
307 - let title = format!("DB pool pressure: {active}/{pool_size} active");
308 - wam.create_ticket(&title, None, "high", "db-pool-pressure", None).await;
309 - last_pool_alert_at = Some(Instant::now());
310 - pool_pressure_armed = false; // wait for drop below `low` before re-arming
311 - }
312 - } else if pct < low {
313 - pool_pressure_armed = true;
310 + let priority = match snap.status {
311 + MonitorStatus::Error => "critical",
312 + MonitorStatus::Degraded => "high",
313 + MonitorStatus::Operational => unreachable!(),
314 + };
315 + let title = format!("Health status: {}", snap.status.as_str());
316 + let body = format!(
317 + "db: {}\ns3: {}\nsessions: {}\ncheck_ms: {}",
318 + snap.db_ok, snap.s3_ok, snap.sessions_ok, snap.check_duration_ms,
319 + );
320 + wam.create_ticket(&title, Some(&body), priority, "health-status-change", None)
321 + .await;
322 + }
323 + }
324 +
325 + previous_status = Some(snap.status);
326 +
327 + // DB pool pressure check with hysteresis: open at 80%, only
328 + // re-alert after pressure drops below 60% and climbs back over
329 + // 80%. Without hysteresis, a load that oscillates around the
330 + // threshold (e.g. each scheduler tick briefly maxes the pool)
331 + // spams a ticket every ALERT_COOLDOWN_SECS window.
332 + {
333 + let pool_size = state.db.size();
334 + let pool_idle = state.db.num_idle() as u32;
335 + let active = pool_size.saturating_sub(pool_idle);
336 + let pct = (active * 100).checked_div(pool_size).unwrap_or(0);
337 + let high = 80u32;
338 + let low = 60u32;
339 +
340 + if pct > high {
341 + tracing::warn!(
342 + pool_size,
343 + active,
344 + idle = pool_idle,
345 + pct,
346 + "DB pool pressure >80%"
347 + );
348 + // Only fire a ticket on the rising edge (was below the low
349 + // threshold last we recovered). The cooldown is still here
350 + // as a backstop in case the hysteresis state machine ever
351 + // gets confused.
352 + let cooldown_ok = last_pool_alert_at
353 + .is_none_or(|t| t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS);
354 + if cooldown_ok
355 + && pool_pressure_armed
356 + && let Some(ref wam) = state.wam
357 + {
358 + let title = format!("DB pool pressure: {active}/{pool_size} active");
359 + wam.create_ticket(&title, None, "high", "db-pool-pressure", None)
360 + .await;
361 + last_pool_alert_at = Some(Instant::now());
362 + pool_pressure_armed = false; // wait for drop below `low` before re-arming
314 363 }
315 - // Between `low` and `high` we hold the current armed state —
316 - // that's the dead-band where neither edge triggers.
364 + } else if pct < low {
365 + pool_pressure_armed = true;
317 366 }
367 + // Between `low` and `high` we hold the current armed state —
368 + // that's the dead-band where neither edge triggers.
369 + }
318 370
Lines truncated
M server/src/routes/admin/moderation.rs +14 -21
@@ -71,27 +71,20 @@ pub(super) async fn admin_decide_appeal(
71 71 && let Some(ref account_id) = db_user.stripe_account_id
72 72 {
73 73 let resumed = db::subscriptions::resume_subscriptions_for_creator(&state.db, user_id).await?;
74 - // Resume each paused subscription on Stripe with bounded concurrency, so
75 - // a creator with many fans doesn't serialize one network round-trip per
76 - // subscription (the appeal handler would otherwise block for N x RTT).
77 - const RESUME_PARALLELISM: usize = 8;
78 - let mut set = tokio::task::JoinSet::new();
79 - for sub in &resumed {
80 - if set.len() >= RESUME_PARALLELISM {
81 - let _ = set.join_next().await;
82 - }
83 - let stripe = stripe.clone();
84 - let account_id = account_id.clone();
85 - let sub_id = sub.stripe_subscription_id.clone();
86 - set.spawn(async move {
87 - if let Err(e) = stripe.resume_subscription(&sub_id, &account_id).await {
88 - tracing::error!(stripe_sub_id = %sub_id, error = ?e, "failed to resume subscription on appeal approval");
89 - }
90 - });
91 - }
92 - while set.join_next().await.is_some() {}
93 - if !resumed.is_empty() {
94 - tracing::info!(user_id = %user_id, resumed = resumed.len(), "resumed fan subscriptions on appeal approval");
74 + // Fan-out the Stripe resume on the bounded background queue so the appeal
75 + // handler returns immediately instead of blocking on N Stripe round-trips
76 + // (Perf-MIN: the 6th fan-out site, now consistent with the other five
77 + // via spawn_fan_sub_fanout rather than an inline blocking JoinSet).
78 + let ids: Vec<String> = resumed.iter().map(|s| s.stripe_subscription_id.clone()).collect();
79 + if !ids.is_empty() {
80 + tracing::info!(user_id = %user_id, resumed = ids.len(), "resuming fan subscriptions on appeal approval (backgrounded)");
81 + crate::payments::fan_ops::spawn_fan_sub_fanout(
82 + &state.bg,
83 + std::sync::Arc::clone(stripe),
84 + account_id.clone(),
85 + ids,
86 + crate::payments::fan_ops::FanSubOp::Resume,
87 + );
95 88 }
96 89 }
97 90
M server/src/routes/pages/public/content/library.rs +10 -9
@@ -90,16 +90,17 @@ pub(in crate::routes::pages::public) async fn library_page(
90 90 let containing_bundles: Vec<Item> = if !db_item.listed {
91 91 let bundle_ids =
92 92 db::bundles::get_bundles_containing_item(&state.db, db_item.id).await?;
93 - let mut bundles = Vec::new();
94 - for bid in &bundle_ids {
95 - if let Some(b) = db::items::get_item_by_id(&state.db, *bid).await?
96 - && b.is_public
97 - {
93 + // Batch-fetch the public bundles in one query (Perf-MIN N+1);
94 + // get_public_items_by_ids already filters to is_public, matching the
95 + // per-item check the old loop did.
96 + db::items::get_public_items_by_ids(&state.db, &bundle_ids)
97 + .await?
98 + .iter()
99 + .map(|b| {
98 100 let tags = Vec::new();
99 - bundles.push(Item::from_db_list(&b, &tags, b.price_cents == 0, false));
100 - }
101 - }
102 - bundles
101 + Item::from_db_list(b, &tags, b.price_cents == 0, false)
102 + })
103 + .collect()
103 104 } else {
104 105 Vec::new()
105 106 };
M server/src/routes/postmark/issues.rs +9 -8
@@ -360,17 +360,18 @@ async fn handle_issue_reply(
360 360 let issue_url = format!("{}/git/{}/{}/issues/{}", host_url, owner_name, repo_name, issue_number);
361 361 let original_msg_id = format!("<issue-{}-{}@makenot.work>", issue_id, chrono::Utc::now().timestamp());
362 362
363 - for participant_id in participants {
364 - if participant_id == commenter_id {
365 - continue;
366 - }
367 - let user = match db::users::get_user_by_id(&db, participant_id).await {
368 - Ok(Some(u)) => u,
369 - _ => continue,
370 - };
363 + // Batch-fetch every recipient in one query instead of one per participant
364 + // (Perf-MIN N+1). The email sends below are inherently per-recipient.
365 + let recipient_ids: Vec<_> =
366 + participants.into_iter().filter(|p| *p != commenter_id).collect();
367 + let users = db::users::get_users_by_ids(&db, &recipient_ids)
368 + .await
369 + .unwrap_or_default();
370 + for user in users {
371 371 if !user.notify_issues {
372 372 continue;
373 373 }
374 + let participant_id = user.id;
374 375 let unsub_url = crate::email::generate_unsubscribe_url(
375 376 &host_url, participant_id, crate::email::UnsubscribeAction::Issue, &participant_id.to_string(), &signing_secret,
376 377 );
M server/src/scanning/hash_lookup.rs +12 -2
@@ -52,8 +52,18 @@ pub async fn check_malwarebazaar(sha256: &str, auth_key: Option<&str>) -> LayerR
52 52 }
53 53
54 54 async fn query_hash(sha256: &str, auth_key: &str) -> Result<LayerResult, String> {
55 - static CLIENT: std::sync::LazyLock<reqwest::Client> =
56 - std::sync::LazyLock::new(reqwest::Client::new);
55 + // Explicit request + connect timeouts (Perf-S2); a default client has neither.
56 + static CLIENT: std::sync::LazyLock<reqwest::Client> = std::sync::LazyLock::new(|| {
57 + reqwest::Client::builder()
58 + .timeout(std::time::Duration::from_secs(
59 + crate::constants::SCAN_MALWAREBAZAAR_TIMEOUT_SECS,
60 + ))
61 + .connect_timeout(std::time::Duration::from_secs(
62 + crate::constants::SCAN_HTTP_CONNECT_TIMEOUT_SECS,
63 + ))
64 + .build()
65 + .unwrap_or_default()
66 + });
57 67
58 68 let client = &*CLIENT;
59 69 let params = [("query", "get_info"), ("hash", sha256)];
M server/src/scanning/metadefender.rs +22 -5
@@ -61,8 +61,18 @@ pub async fn check_metadefender(sha256: &str, api_key: Option<&str>) -> LayerRes
61 61 }
62 62
63 63 async fn query_hash(sha256: &str, api_key: &str) -> Result<LayerResult, String> {
64 - static CLIENT: std::sync::LazyLock<reqwest::Client> =
65 - std::sync::LazyLock::new(reqwest::Client::new);
64 + // Explicit request + connect timeouts (Perf-S2); a default client has neither.
65 + static CLIENT: std::sync::LazyLock<reqwest::Client> = std::sync::LazyLock::new(|| {
66 + reqwest::Client::builder()
67 + .timeout(std::time::Duration::from_secs(
68 + crate::constants::SCAN_MALWAREBAZAAR_TIMEOUT_SECS,
69 + ))
70 + .connect_timeout(std::time::Duration::from_secs(
71 + crate::constants::SCAN_HTTP_CONNECT_TIMEOUT_SECS,
72 + ))
73 + .build()
74 + .unwrap_or_default()
75 + });
66 76 let client = &*CLIENT;
67 77
68 78 let url = format!("{API_URL_PREFIX}{sha256}");
@@ -141,12 +151,14 @@ fn parse_metadefender_response(body: &serde_json::Value) -> LayerResult {
141 151 let mut threats: Vec<String> = Vec::new();
142 152 if let Some(details) = scan_results.get("scan_details").and_then(|d| d.as_object()) {
143 153 for (engine, info) in details {
144 - let detected = info.get("scan_result_i")
154 + let detected = info
155 + .get("scan_result_i")
145 156 .and_then(|v| v.as_i64())
146 157 .unwrap_or(0);
147 158 // 1 = infected, 2 = suspicious in MetaDefender's enum. 0 = clean.
148 159 if detected == 1 || detected == 2 {
149 - let threat = info.get("threat_found")
160 + let threat = info
161 + .get("threat_found")
150 162 .and_then(|v| v.as_str())
151 163 .unwrap_or("malicious");
152 164 threats.push(format!("{engine}: {threat}"));
@@ -158,7 +170,12 @@ fn parse_metadefender_response(body: &serde_json::Value) -> LayerResult {
158 170 let threats_str = if threats.is_empty() {
159 171 format!("{}/{} engines flagged", total_detected, total_avs)
160 172 } else {
161 - format!("{}/{} engines flagged — {}", total_detected, total_avs, threats.join(", "))
173 + format!(
174 + "{}/{} engines flagged — {}",
175 + total_detected,
176 + total_avs,
177 + threats.join(", ")
178 + )
162 179 };
163 180 LayerResult {
164 181 layer: "metadefender",
M server/src/scanning/mod.rs +220 -77
@@ -98,7 +98,10 @@ fn error_policy_for(layer: &str) -> ErrorPolicy {
98 98 "signing_linux" => signing_linux::ERROR_POLICY,
99 99 "metadefender" => metadefender::ERROR_POLICY,
100 100 other => {
101 - tracing::error!(layer = other, "unknown scan layer; defaulting to FailClosed");
101 + tracing::error!(
102 + layer = other,
103 + "unknown scan layer; defaulting to FailClosed"
104 + );
102 105 ErrorPolicy::FailClosed
103 106 }
104 107 }
@@ -110,15 +113,37 @@ fn error_policy_for(layer: &str) -> ErrorPolicy {
110 113 /// to escalate". Pure `Error`s from fail-open external layers do not — those
111 114 /// are operational noise, not malware signals.
112 115 fn suspicion_present(layers: &[LayerResult]) -> bool {
113 - layers.iter().any(|l| {
114 - match l.verdict {
115 - LayerVerdict::Fail => true,
116 - LayerVerdict::Error => error_policy_for(l.layer) == ErrorPolicy::FailClosed,
117 - _ => false,
118 - }
116 + layers.iter().any(|l| match l.verdict {
117 + LayerVerdict::Fail => true,
118 + LayerVerdict::Error => error_policy_for(l.layer) == ErrorPolicy::FailClosed,
119 + _ => false,
119 120 })
120 121 }
121 122
123 + /// Run an external hash-lookup layer under the aggregate external-lookup deadline,
124 + /// degrading to a Skip (FailOpen) on timeout. Shared by the buffered `scan()` and
125 + /// the streaming `scan_stream()` so BOTH paths get the same defense-in-depth
126 + /// deadline (Perf-S1) — previously only `scan_stream` wrapped these calls, so a
127 + /// small upload had only the per-request client timeout with no aggregate ceiling.
128 + async fn lookup_with_timeout(
129 + layer: &'static str,
130 + fut: impl std::future::Future<Output = LayerResult>,
131 + ) -> LayerResult {
132 + let lookup_timeout =
133 + std::time::Duration::from_secs(crate::constants::SCAN_EXTERNAL_LOOKUP_TIMEOUT_SECS);
134 + match tokio::time::timeout(lookup_timeout, fut).await {
135 + Ok(layer_result) => layer_result,
136 + Err(_) => {
137 + tracing::warn!("{layer} lookup timed out; treating as Skip (FailOpen)");
138 + LayerResult {
139 + layer,
140 + verdict: LayerVerdict::Skip,
141 + detail: Some(format!("{layer} lookup timed out")),
142 + }
143 + }
144 + }
145 + }
146 +
122 147 /// Aggregate per-layer results into a final scan status.
123 148 ///
124 149 /// - Any layer `Fail` → `Quarantined` (terminal).
@@ -311,7 +336,8 @@ impl ScanPipeline {
311 336 // Sync layers + hash, off the runtime
312 337 let sync_data = data.clone();
313 338 let sync_self = std::sync::Arc::clone(&self);
314 - let sync_fut = tokio::task::spawn_blocking(move || sync_self.run_sync_layers(&sync_data, file_type));
339 + let sync_fut =
340 + tokio::task::spawn_blocking(move || sync_self.run_sync_layers(&sync_data, file_type));
315 341
316 342 // Async layers — ClamAV needs the bytes, MalwareBazaar needs only the hash
317 343 // but the hash is computed in the sync block, so we run MalwareBazaar
@@ -363,9 +389,14 @@ impl ScanPipeline {
363 389 layers.push(clamav_result);
364 390 layers.push(urlhaus_result);
365 391
366 - // Layer 6: MalwareBazaar — needs the hash from the sync block
392 + // Layer 6: MalwareBazaar — needs the hash from the sync block. Wrapped in
393 + // the aggregate external-lookup deadline (Perf-S1), same as scan_stream.
367 394 layers.push(if self.malwarebazaar_enabled {
368 - hash_lookup::check_malwarebazaar(&sha256, self.abuse_ch_auth_key.as_deref()).await
395 + lookup_with_timeout(
396 + "malwarebazaar",
397 + hash_lookup::check_malwarebazaar(&sha256, self.abuse_ch_auth_key.as_deref()),
398 + )
399 + .await
369 400 } else {
370 401 LayerResult {
371 402 layer: "malwarebazaar",
@@ -378,7 +409,11 @@ impl ScanPipeline {
378 409 // layer flagged the file as suspicious — keeps us within the free-tier
379 410 // quota and avoids spending budget on uncontroversial uploads.
380 411 layers.push(if suspicion_present(&layers) {
381 - metadefender::check_metadefender(&sha256, self.metadefender_api_key.as_deref()).await
412 + lookup_with_timeout(
413 + "metadefender",
414 + metadefender::check_metadefender(&sha256, self.metadefender_api_key.as_deref()),
415 + )
416 + .await
382 417 } else {
383 418 LayerResult {
384 419 layer: "metadefender",
@@ -430,7 +465,8 @@ impl ScanPipeline {
430 465
431 466 let sync_map = std::sync::Arc::clone(&map);
432 467 let sync_self = std::sync::Arc::clone(&self);
433 - let sync_fut = tokio::task::spawn_blocking(move || sync_self.run_sync_layers(&sync_map, file_type));
468 + let sync_fut =
469 + tokio::task::spawn_blocking(move || sync_self.run_sync_layers(&sync_map, file_type));
434 470
435 471 let clamav_socket = self.clamav_socket.clone();
436 472 let clamav_path = spool.path().to_path_buf();
@@ -492,22 +528,12 @@ impl ScanPipeline {
492 528 layers.push(clamav_result);
493 529 layers.push(urlhaus_result);
494 530
495 - let lookup_timeout =
496 - std::time::Duration::from_secs(crate::constants::SCAN_EXTERNAL_LOOKUP_TIMEOUT_SECS);
497 -
498 531 layers.push(if self.malwarebazaar_enabled {
499 - let fut = hash_lookup::check_malwarebazaar(&sha256, self.abuse_ch_auth_key.as_deref());
500 - match tokio::time::timeout(lookup_timeout, fut).await {
501 - Ok(layer) => layer,
502 - Err(_) => {
503 - tracing::warn!("MalwareBazaar lookup timed out; treating as Skip (FailOpen)");
504 - LayerResult {
505 - layer: "malwarebazaar",
506 - verdict: LayerVerdict::Skip,
507 - detail: Some("MalwareBazaar lookup timed out".to_string()),
508 - }
509 - }
510 - }
532 + lookup_with_timeout(
533 + "malwarebazaar",
534 + hash_lookup::check_malwarebazaar(&sha256, self.abuse_ch_auth_key.as_deref()),
535 + )
536 + .await
511 537 } else {
512 538 LayerResult {
513 539 layer: "malwarebazaar",
@@ -517,19 +543,11 @@ impl ScanPipeline {
517 543 });
518 544
519 545 layers.push(if suspicion_present(&layers) {
520 - let fut =
521 - metadefender::check_metadefender(&sha256, self.metadefender_api_key.as_deref());
522 - match tokio::time::timeout(lookup_timeout, fut).await {
523 - Ok(layer) => layer,
524 - Err(_) => {
525 - tracing::warn!("MetaDefender lookup timed out; treating as Skip (FailOpen)");
526 - LayerResult {
527 - layer: "metadefender",
528 - verdict: LayerVerdict::Skip,
529 - detail: Some("MetaDefender lookup timed out".to_string()),
530 - }
531 - }
532 - }
546 + lookup_with_timeout(
547 + "metadefender",
548 + metadefender::check_metadefender(&sha256, self.metadefender_api_key.as_deref()),
549 + )
550 + .await
533 551 } else {
534 552 LayerResult {
535 553 layer: "metadefender",
@@ -709,7 +727,9 @@ mod tests {
709 727 #[tokio::test]
710 728 async fn assert_live_refuses_degraded_yara_corpus() {
711 729 let mut compiler = yara_x::Compiler::new();
712 - compiler.add_source(r#"rule r { strings: $a = "x" condition: $a }"#).unwrap();
730 + compiler
731 + .add_source(r#"rule r { strings: $a = "x" condition: $a }"#)
732 + .unwrap();
713 733 let pipeline = std::sync::Arc::new(ScanPipeline {
714 734 yara_rules: Some(compiler.build()),
715 735 yara_rule_count: 1,
@@ -720,15 +740,23 @@ mod tests {
720 740 abuse_ch_auth_key: None,
721 741 metadefender_api_key: None,
722 742 });
723 - let err = pipeline.assert_live().await.expect_err("degraded corpus must refuse boot");
724 - assert!(err.contains("YARA corpus degraded"), "unexpected error: {err}");
743 + let err = pipeline
744 + .assert_live()
745 + .await
746 + .expect_err("degraded corpus must refuse boot");
747 + assert!(
748 + err.contains("YARA corpus degraded"),
749 + "unexpected error: {err}"
750 + );
725 751 }
726 752
727 753 /// The complement: a corpus meeting the floor boots, with YARA counted live.
728 754 #[tokio::test]
729 755 async fn assert_live_accepts_corpus_meeting_floor() {
730 756 let mut compiler = yara_x::Compiler::new();
731 - compiler.add_source(r#"rule r { strings: $a = "x" condition: $a }"#).unwrap();
757 + compiler
758 + .add_source(r#"rule r { strings: $a = "x" condition: $a }"#)
759 + .unwrap();
732 760 let pipeline = std::sync::Arc::new(ScanPipeline {
733 761 yara_rules: Some(compiler.build()),
734 762 yara_rule_count: 3,
@@ -739,13 +767,19 @@ mod tests {
739 767 abuse_ch_auth_key: None,
740 768 metadefender_api_key: None,
741 769 });
742 - pipeline.assert_live().await.expect("a corpus meeting the floor must boot");
770 + pipeline
771 + .assert_live()
772 + .await
773 + .expect("a corpus meeting the floor must boot");
743 774 }
744 775
745 776 #[tokio::test]
746 777 async fn pipeline_clean_download_passes() {
747 778 let pipeline = make_pipeline();
748 - let result = pipeline.clone().scan(b"just some file content".to_vec(), FileType::Download).await;
779 + let result = pipeline
780 + .clone()
781 + .scan(b"just some file content".to_vec(), FileType::Download)
782 + .await;
749 783 assert_eq!(result.status, FileScanStatus::Clean);
750 784 assert_eq!(result.file_size, 22);
751 785 assert!(!result.sha256.is_empty());
@@ -756,7 +790,10 @@ mod tests {
756 790 async fn pipeline_unrecognized_audio_quarantined() {
757 791 let pipeline = make_pipeline();
758 792 // Unrecognized data claimed as audio should be rejected by content_type layer
759 - let result = pipeline.clone().scan(b"audio data here".to_vec(), FileType::Audio).await;
793 + let result = pipeline
794 + .clone()
795 + .scan(b"audio data here".to_vec(), FileType::Audio)
796 + .await;
760 797 assert_eq!(result.status, FileScanStatus::Quarantined);
761 798 }
762 799
@@ -774,10 +811,17 @@ mod tests {
774 811 let pipeline = make_pipeline();
775 812 // PE magic bytes — content-type layer should detect application/* and fail
776 813 let pe_header = b"MZ\x90\x00\x03\x00\x00\x00";
777 - let result = pipeline.clone().scan(pe_header.to_vec(), FileType::Audio).await;
814 + let result = pipeline
815 + .clone()
816 + .scan(pe_header.to_vec(), FileType::Audio)
817 + .await;
778 818 assert_eq!(result.status, FileScanStatus::Quarantined);
779 819 // Verify content_type layer produced the fail
780 - let content_type_layer = result.layers.iter().find(|l| l.layer == "content_type").unwrap();
820 + let content_type_layer = result
821 + .layers
822 + .iter()
823 + .find(|l| l.layer == "content_type")
824 + .unwrap();
781 825 assert_eq!(content_type_layer.verdict, LayerVerdict::Fail);
782 826 }
783 827
@@ -785,7 +829,10 @@ mod tests {
785 829 async fn pipeline_pe_as_cover_quarantined() {
786 830 let pipeline = make_pipeline();
787 831 let pe_header = b"MZ\x90\x00\x03\x00\x00\x00";
788 - let result = pipeline.clone().scan(pe_header.to_vec(), FileType::Cover).await;
832 + let result = pipeline
833 + .clone()
834 + .scan(pe_header.to_vec(), FileType::Cover)
835 + .await;
789 836 assert_eq!(result.status, FileScanStatus::Quarantined);
790 837 }
791 838
@@ -793,15 +840,24 @@ mod tests {
793 840 async fn pipeline_sha256_is_deterministic() {
794 841 let pipeline = make_pipeline();
795 842 let data = b"deterministic hash test";
796 - let r1 = pipeline.clone().scan(data.to_vec(), FileType::Download).await;
797 - let r2 = pipeline.clone().scan(data.to_vec(), FileType::Download).await;
843 + let r1 = pipeline
844 + .clone()
845 + .scan(data.to_vec(), FileType::Download)
846 + .await;
847 + let r2 = pipeline
848 + .clone()
849 + .scan(data.to_vec(), FileType::Download)
850 + .await;
798 851 assert_eq!(r1.sha256, r2.sha256);
799 852 }
800 853
801 854 #[tokio::test]
802 855 async fn pipeline_skips_optional_layers_when_unconfigured() {
803 856 let pipeline = make_pipeline();
804 - let result = pipeline.clone().scan(b"test".to_vec(), FileType::Download).await;
857 + let result = pipeline
858 + .clone()
859 + .scan(b"test".to_vec(), FileType::Download)
860 + .await;
805 861
806 862 let yara = result.layers.iter().find(|l| l.layer == "yara").unwrap();
807 863 assert_eq!(yara.verdict, LayerVerdict::Skip);
@@ -809,7 +865,11 @@ mod tests {
809 865 let clamav = result.layers.iter().find(|l| l.layer == "clamav").unwrap();
810 866 assert_eq!(clamav.verdict, LayerVerdict::Skip);
811 867
812 - let mb = result.layers.iter().find(|l| l.layer == "malwarebazaar").unwrap();
868 + let mb = result
869 + .layers
870 + .iter()
871 + .find(|l| l.layer == "malwarebazaar")
872 + .unwrap();
813 873 assert_eq!(mb.verdict, LayerVerdict::Skip);
814 874
815 875 let uh = result.layers.iter().find(|l| l.layer == "urlhaus").unwrap();
@@ -822,7 +882,12 @@ mod tests {
822 882 for file_type in [FileType::Audio, FileType::Cover, FileType::Download] {
823 883 let result = pipeline.clone().scan(b"data".to_vec(), file_type).await;
824 884 // 11 base layers + the recursive archive-interior layer (archive_nested).
825 - assert_eq!(result.layers.len(), 12, "Expected 12 layers for {:?}", file_type);
885 + assert_eq!(
886 + result.layers.len(),
887 + 12,
888 + "Expected 12 layers for {:?}",
889 + file_type
890 + );
826 891 }
827 892 }
828 893
@@ -869,21 +934,44 @@ mod tests {
869 934 // -- Per-layer fail policy tests --
870 935
871 936 fn err(layer: &'static str) -> LayerResult {
872 - LayerResult { layer, verdict: LayerVerdict::Error, detail: None }
937 + LayerResult {
938 + layer,
939 + verdict: LayerVerdict::Error,
940 + detail: None,
941 + }
873 942 }
874 943 fn pass(layer: &'static str) -> LayerResult {
875 - LayerResult { layer, verdict: LayerVerdict::Pass, detail: None }
944 + LayerResult {
945 + layer,
946 + verdict: LayerVerdict::Pass,
947 + detail: None,
948 + }
876 949 }
877 950 fn skip(layer: &'static str) -> LayerResult {
878 - LayerResult { layer, verdict: LayerVerdict::Skip, detail: None }
951 + LayerResult {
952 + layer,
953 + verdict: LayerVerdict::Skip,
954 + detail: None,
955 + }
879 956 }
880 957 fn fail(layer: &'static str) -> LayerResult {
881 - LayerResult { layer, verdict: LayerVerdict::Fail, detail: None }
958 + LayerResult {
959 + layer,
960 + verdict: LayerVerdict::Fail,
961 + detail: None,
962 + }
882 963 }
883 964
884 965 #[test]
885 966 fn final_status_clean_when_all_pass() {
886 - let layers = vec![pass("content_type"), pass("structural"), pass("archive"), skip("yara"), skip("clamav"), skip("malwarebazaar")];
967 + let layers = vec![
968 + pass("content_type"),
969 + pass("structural"),
970 + pass("archive"),
971 + skip("yara"),
972 + skip("clamav"),
973 + skip("malwarebazaar"),
974 + ];
887 975 assert_eq!(final_status(&layers), FileScanStatus::Clean);
888 976 }
889 977
@@ -911,7 +999,14 @@ mod tests {
911 999 fn final_status_clean_on_fail_open_error_only() {
912 1000 // malwarebazaar is FailOpen — its Error must NOT hold the file.
913 1001 // This is the regression of 2026-05-10 that motivated the audit.
914 - let layers = vec![pass("content_type"), pass("structural"), pass("archive"), skip("yara"), skip("clamav"), err("malwarebazaar")];
1002 + let layers = vec![
1003 + pass("content_type"),
1004 + pass("structural"),
1005 + pass("archive"),
1006 + skip("yara"),
1007 + skip("clamav"),
1008 + err("malwarebazaar"),
1009 + ];
915 1010 assert_eq!(final_status(&layers), FileScanStatus::Clean);
916 1011 }
917 1012
@@ -920,7 +1015,14 @@ mod tests {
920 1015 // Worst-case external-services outage: every network/daemon layer
921 1016 // erroring at once. As long as the in-process layers pass, the file
922 1017 // is Clean. Health is surfaced separately via per-layer monitoring.
923 - let layers = vec![pass("content_type"), pass("structural"), pass("archive"), skip("yara"), err("clamav"), err("malwarebazaar")];
1018 + let layers = vec![
1019 + pass("content_type"),
1020 + pass("structural"),
1021 + pass("archive"),
1022 + skip("yara"),
1023 + err("clamav"),
1024 + err("malwarebazaar"),
1025 + ];
924 1026 assert_eq!(final_status(&layers), FileScanStatus::Clean);
925 1027 }
926 1028
@@ -929,9 +1031,18 @@ mod tests {
929 1031 // CHRONIC S1: a reachable-but-incomplete clamav scan is emitted under the
930 1032 // `clamav_incomplete` layer, which must be FailClosed → HeldForReview,
931 1033 // distinct from the FailOpen `clamav` layer used for an unreachable daemon.
932 - assert_eq!(error_policy_for("clamav_incomplete"), ErrorPolicy::FailClosed);
1034 + assert_eq!(
1035 + error_policy_for("clamav_incomplete"),
1036 + ErrorPolicy::FailClosed
1037 + );
933 1038 assert_eq!(error_policy_for("clamav"), ErrorPolicy::FailOpen);
934 - let layers = vec![pass("content_type"), pass("structural"), pass("archive"), skip("yara"), err("clamav_incomplete")];
1039 + let layers = vec![
1040 + pass("content_type"),
1041 + pass("structural"),
1042 + pass("archive"),
1043 + skip("yara"),
1044 + err("clamav_incomplete"),
1045 + ];
935 1046 assert_eq!(final_status(&layers), FileScanStatus::HeldForReview);
936 1047 }
937 1048
@@ -940,7 +1051,10 @@ mod tests {
940 1051 // Defensive default: an unknown layer name that errors falls through
941 1052 // to FailClosed. This is what catches a new layer added without
942 1053 // wiring its policy into `error_policy_for`.
943 - let layers = vec![pass("content_type"), err("brand_new_layer_someone_forgot_to_register")];
1054 + let layers = vec![
1055 + pass("content_type"),
1056 + err("brand_new_layer_someone_forgot_to_register"),
1057 + ];
944 1058 assert_eq!(final_status(&layers), FileScanStatus::HeldForReview);
945 1059 }
946 1060
@@ -949,7 +1063,14 @@ mod tests {
949 1063 // Every layer name produced by the pipeline must have an explicit
950 1064 // declaration in `error_policy_for`. The default branch is reserved
951 1065 // for genuine programmer error (new layer, forgot to register).
952 - for name in ["content_type", "structural", "archive", "yara", "clamav", "malwarebazaar"] {
1066 + for name in [
1067 + "content_type",
1068 + "structural",
1069 + "archive",
1070 + "yara",
1071 + "clamav",
1072 + "malwarebazaar",
1073 + ] {
953 1074 let policy = error_policy_for(name);
954 1075 // Both values are valid; we just want this to not hit the default.
955 1076 // If a layer is renamed without updating `error_policy_for`, this
@@ -960,25 +1081,47 @@ mod tests {
960 1081 }
961 1082
962 1083 #[test]
963 - fn content_type_is_fail_closed() { assert_eq!(error_policy_for("content_type"), ErrorPolicy::FailClosed); }
1084 + fn content_type_is_fail_closed() {
1085 + assert_eq!(error_policy_for("content_type"), ErrorPolicy::FailClosed);
1086 + }
964 1087 #[test]
965 - fn structural_is_fail_closed() { assert_eq!(error_policy_for("structural"), ErrorPolicy::FailClosed); }
1088 + fn structural_is_fail_closed() {
1089 + assert_eq!(error_policy_for("structural"), ErrorPolicy::FailClosed);
1090 + }
966 1091 #[test]
967 - fn archive_is_fail_closed() { assert_eq!(error_policy_for("archive"), ErrorPolicy::FailClosed); }
1092 + fn archive_is_fail_closed() {
1093 + assert_eq!(error_policy_for("archive"), ErrorPolicy::FailClosed);
1094 + }
968 1095 #[test]
969 - fn yara_is_fail_closed() { assert_eq!(error_policy_for("yara"), ErrorPolicy::FailClosed); }
1096 + fn yara_is_fail_closed() {
1097 + assert_eq!(error_policy_for("yara"), ErrorPolicy::FailClosed);
1098 + }
970 1099 #[test]
971 - fn clamav_is_fail_open() { assert_eq!(error_policy_for("clamav"), ErrorPolicy::FailOpen); }
1100 + fn clamav_is_fail_open() {
1101 + assert_eq!(error_policy_for("clamav"), ErrorPolicy::FailOpen);
1102 + }
972 1103 #[test]
973 - fn malwarebazaar_is_fail_open() { assert_eq!(error_policy_for("malwarebazaar"), ErrorPolicy::FailOpen); }
1104 + fn malwarebazaar_is_fail_open() {
1105 + assert_eq!(error_policy_for("malwarebazaar"), ErrorPolicy::FailOpen);
1106 + }
974 1107 #[test]
975 - fn urlhaus_is_fail_open() { assert_eq!(error_policy_for("urlhaus"), ErrorPolicy::FailOpen); }
1108 + fn urlhaus_is_fail_open() {
1109 + assert_eq!(error_policy_for("urlhaus"), ErrorPolicy::FailOpen);
1110 + }
976 1111 #[test]
977 - fn signing_macos_is_fail_open() { assert_eq!(error_policy_for("signing_macos"), ErrorPolicy::FailOpen); }
1112 + fn signing_macos_is_fail_open() {
1113 + assert_eq!(error_policy_for("signing_macos"), ErrorPolicy::FailOpen);
1114 + }
978 1115 #[test]
979 - fn metadefender_is_fail_open() { assert_eq!(error_policy_for("metadefender"), ErrorPolicy::FailOpen); }
1116 + fn metadefender_is_fail_open() {
1117 + assert_eq!(error_policy_for("metadefender"), ErrorPolicy::FailOpen);
1118 + }
980 1119 #[test]
981 - fn signing_windows_is_fail_open() { assert_eq!(error_policy_for("signing_windows"), ErrorPolicy::FailOpen); }
1120 + fn signing_windows_is_fail_open() {
1121 + assert_eq!(error_policy_for("signing_windows"), ErrorPolicy::FailOpen);
1122 + }
982 1123 #[test]
983 - fn signing_linux_is_fail_open() { assert_eq!(error_policy_for("signing_linux"), ErrorPolicy::FailOpen); }
1124 + fn signing_linux_is_fail_open() {
1125 + assert_eq!(error_policy_for("signing_linux"), ErrorPolicy::FailOpen);
1126 + }
984 1127 }
M server/src/scanning/urlhaus.rs +86 -39
@@ -66,47 +66,66 @@ pub(crate) async fn check_urlhaus_hosts(hosts: Vec<String>, auth_key: Option<&st
66 66 };
67 67 }
68 68
69 - let timeout = Duration::from_secs(constants::SCAN_MALWAREBAZAAR_TIMEOUT_SECS);
70 - for host in &hosts {
71 - match tokio::time::timeout(timeout, query_host(host, key)).await {
72 - Ok(Ok(verdict)) => match verdict {
73 - HostVerdict::Clean => continue,
74 - HostVerdict::Bad(reason) => {
69 + let per_host = Duration::from_secs(constants::SCAN_MALWAREBAZAAR_TIMEOUT_SECS);
70 + // Aggregate deadline across ALL hosts (Perf-S2): the per-host timeout alone
71 + // let up to MAX_HOSTS_PER_FILE sequential lookups stack (5 × per_host ≈ 25s),
72 + // holding one of the two scan workers the whole time. Cap the total at the
73 + // shared external-lookup budget so two adversarial uploads can't saturate the
74 + // pool. On the deadline we return Error (degraded per urlhaus's fail policy),
75 + // same verdict shape as a per-host timeout.
76 + let aggregate = Duration::from_secs(constants::SCAN_EXTERNAL_LOOKUP_TIMEOUT_SECS);
77 + let host_count = hosts.len();
78 + let scan = async {
79 + for host in &hosts {
80 + match tokio::time::timeout(per_host, query_host(host, key)).await {
81 + Ok(Ok(verdict)) => match verdict {
82 + HostVerdict::Clean => continue,
83 + HostVerdict::Bad(reason) => {
84 + return LayerResult {
85 + layer: "urlhaus",
86 + verdict: LayerVerdict::Fail,
87 + detail: Some(format!("Known-malicious host {host}: {reason}")),
88 + };
89 + }
90 + HostVerdict::AuthError(detail) => {
91 + return LayerResult {
92 + layer: "urlhaus",
93 + verdict: LayerVerdict::Error,
94 + detail: Some(detail),
95 + };
96 + }
97 + },
98 + Ok(Err(e)) => {
75 99 return LayerResult {
76 100 layer: "urlhaus",
77 - verdict: LayerVerdict::Fail,
78 - detail: Some(format!("Known-malicious host {host}: {reason}")),
101 + verdict: LayerVerdict::Error,
102 + detail: Some(format!("URLhaus error: {e}")),
79 103 };
80 104 }
81 - HostVerdict::AuthError(detail) => {
105 + Err(_) => {
82 106 return LayerResult {
83 107 layer: "urlhaus",
84 108 verdict: LayerVerdict::Error,
85 - detail: Some(detail),
109 + detail: Some("URLhaus lookup timed out".to_string()),
86 110 };
87 111 }
88 - },
89 - Ok(Err(e)) => {
90 - return LayerResult {
91 - layer: "urlhaus",
92 - verdict: LayerVerdict::Error,
93 - detail: Some(format!("URLhaus error: {e}")),
94 - };
95 - }
96 - Err(_) => {
97 - return LayerResult {
98 - layer: "urlhaus",
99 - verdict: LayerVerdict::Error,
100 - detail: Some("URLhaus lookup timed out".to_string()),
101 - };
102 112 }
103 113 }
104 - }
105 114
106 - LayerResult {
107 - layer: "urlhaus",
108 - verdict: LayerVerdict::Pass,
109 - detail: Some(format!("Checked {} host(s); none known-bad", hosts.len())),
115 + LayerResult {
116 + layer: "urlhaus",
117 + verdict: LayerVerdict::Pass,
118 + detail: Some(format!("Checked {host_count} host(s); none known-bad")),
119 + }
120 + };
121 +
122 + match tokio::time::timeout(aggregate, scan).await {
123 + Ok(result) => result,
124 + Err(_) => LayerResult {
125 + layer: "urlhaus",
126 + verdict: LayerVerdict::Error,
127 + detail: Some("URLhaus aggregate lookup deadline exceeded".to_string()),
128 + },
110 129 }
111 130 }
112 131
@@ -117,8 +136,20 @@ enum HostVerdict {
117 136 }
118 137
119 138 async fn query_host(host: &str, auth_key: &str) -> Result<HostVerdict, String> {
120 - static CLIENT: std::sync::LazyLock<reqwest::Client> =
121 - std::sync::LazyLock::new(reqwest::Client::new);
139 + // Explicit request + connect timeouts (Perf-S2): a default client has neither,
140 + // so a hung host could block past the layer's own per-host timeout. Falls back
141 + // to a default client only if the builder somehow fails.
142 + static CLIENT: std::sync::LazyLock<reqwest::Client> = std::sync::LazyLock::new(|| {
143 + reqwest::Client::builder()
144 + .timeout(Duration::from_secs(
145 + constants::SCAN_MALWAREBAZAAR_TIMEOUT_SECS,
146 + ))
147 + .connect_timeout(Duration::from_secs(
148 + constants::SCAN_HTTP_CONNECT_TIMEOUT_SECS,
149 + ))
150 + .build()
151 + .unwrap_or_default()
152 + });
122 153 let client = &*CLIENT;
123 154
124 155 let params = [("host", host)];
@@ -179,7 +210,11 @@ fn classify_urlhaus_response(body: &serde_json::Value) -> HostVerdict {
179 210 /// Pull printable-ASCII URL hosts out of the byte buffer. Cap at `max` unique
180 211 /// hosts to bound per-upload work and free-tier quota use.
181 212 pub(crate) fn extract_unique_hosts(data: &[u8], max: usize) -> Vec<String> {
182 - let scan = if data.len() > MAX_SCAN_BYTES { &data[..MAX_SCAN_BYTES] } else { data };
213 + let scan = if data.len() > MAX_SCAN_BYTES {
214 + &data[..MAX_SCAN_BYTES]
215 + } else {
216 + data
217 + };
183 218
184 219 let mut hosts: HashSet<String> = HashSet::new();
185 220 let mut out: Vec<String> = Vec::new();
@@ -234,7 +269,12 @@ fn find_scheme(bytes: &[u8]) -> Option<usize> {
234 269 // Find ":/" position then back up to the scheme start.
235 270 bytes.windows(3).position(|w| w == b"://").map(|colon_at| {
236 271 let mut start = colon_at;
237 - while start > 0 && (bytes[start - 1].is_ascii_alphanumeric() || bytes[start - 1] == b'+' || bytes[start - 1] == b'-' || bytes[start - 1] == b'.') {
272 + while start > 0
273 + && (bytes[start - 1].is_ascii_alphanumeric()
274 + || bytes[start - 1] == b'+'
275 + || bytes[start - 1] == b'-'
276 + || bytes[start - 1] == b'.')
277 + {
238 278 start -= 1;
239 279 }
240 280 start
@@ -243,9 +283,7 @@ fn find_scheme(bytes: &[u8]) -> Option<usize> {
243 283
244 284 fn extract_host(url_run: &str) -> Option<&str> {
245 285 let after_scheme = url_run.split_once("://")?.1;
246 - let host = after_scheme
247 - .split(['/', '?', '#', ':'])
248 - .next()?;
286 + let host = after_scheme.split(['/', '?', '#', ':']).next()?;
249 287 if host.is_empty() || !host.contains('.') {
250 288 return None;
251 289 }
@@ -325,7 +363,10 @@ mod tests {
325 363 #[test]
326 364 fn no_results_is_clean() {
327 365 let body = json!({"query_status": "no_results"});
328 - assert!(matches!(classify_urlhaus_response(&body), HostVerdict::Clean));
366 + assert!(matches!(
367 + classify_urlhaus_response(&body),
368 + HostVerdict::Clean
369 + ));
329 370 }
330 371
331 372 #[test]
@@ -352,7 +393,10 @@ mod tests {
352 393 #[test]
353 394 fn invalid_host_treated_as_clean() {
354 395 let body = json!({"query_status": "invalid_host"});
355 - assert!(matches!(classify_urlhaus_response(&body), HostVerdict::Clean));
396 + assert!(matches!(
397 + classify_urlhaus_response(&body),
398 + HostVerdict::Clean
399 + ));
356 400 }
357 401
358 402 #[test]
@@ -361,6 +405,9 @@ mod tests {
361 405 // shouldn't fail the upload — the layer aggregator's error counts
362 406 // will surface this via the dashboard health panel separately.
363 407 let body = json!({"query_status": "totally_new_thing"});
364 - assert!(matches!(classify_urlhaus_response(&body), HostVerdict::Clean));
408 + assert!(matches!(
409 + classify_urlhaus_response(&body),
410 + HostVerdict::Clean
411 + ));
365 412 }
366 413 }