Skip to main content

max / makenotwork

Kill route N+1s, parallelize builds by host, unify scan entry points Performance axis remediation (Run 9): - enforce_post_grace_hiding: replace the per-creator hide+mark N+1 on the lock-held scheduler tick with two set-based statements (hide_all_items_for_users + mark_grace_enforced_batch); delete the now-unused single-row variants. - grant_bundle_items: claim all bundle children in one UNNEST INSERT (claim_free_items_batch) instead of N per-child round-trips, each re-acquiring a pool connection, on the Stripe webhook hot path. - admin_mt_provision: fan the per-project MT HTTP call out over a bounded JoinSet instead of awaiting them strictly serially over the whole table. - build_runner: group targets by build host and run host groups concurrently (targets within a host stay serial), so a linux+darwin release overlaps across machines instead of serializing at up to 30 min/target. CHRONIC (scan/scan_stream twin divergence, Run 6-9): collapse both entry points into thin pub(crate) adapters over one private run_scan(ScanInput). All scan policy (CPU off-runtime, ClamAV, URLhaus host extraction, external lookups) now lives in exactly one body, so it can no longer drift between buffered and streaming. Delete the buffered-only check_urlhaus helper the divergence rode on. Reconciled: Perf-S5 (internal export_sales) was already paged + 1M-capped at the DB layer in a prior run; no change needed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-25 04:01 UTC
Commit: bacacb7a1caa2526b26fd36263cdbd29e237eb8f
Parent: 5dd4ce3
10 files changed, +361 insertions, -221 deletions
M server/src/build_runner.rs +64 -7
@@ -152,6 +152,13 @@ fn build_failure_message(succeeded: usize, failed: usize, first_error: Option<&s
152 152 }
153 153 }
154 154
155 + /// A successfully built target: `(target_os, arch, s3_key, signature)`.
156 + type TargetArtifact = (String, String, String, String);
157 + /// A failed target: `(target_str, error)`.
158 + type TargetError = (String, String);
159 + /// One host group's results: its artifacts and its per-target failures.
160 + type GroupOutput = (Vec<TargetArtifact>, Vec<TargetError>);
161 +
155 162 /// Execute a full build: iterate targets, SSH to hosts, build, upload artifacts.
156 163 #[tracing::instrument(skip_all, name = "build_runner::run_build", fields(build_id = %build.id, version = %build.version))]
157 164 async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
@@ -159,6 +166,12 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
159 166 let mut failed_count: usize = 0;
160 167 let mut first_error: Option<String> = None;
161 168
169 + // Resolve each target to its build host up front. Synchronous failures (bad
170 + // target format, no host configured) are tallied here; resolvable targets are
171 + // grouped by host so independent hosts (e.g. linux vs darwin) build
172 + // concurrently while same-host targets stay serial — a multi-target release no
173 + // longer serializes end to end at up to 30 min/target (Perf-S2, Run 9).
174 + let mut groups: Vec<(String, Vec<(String, String)>)> = Vec::new(); // host -> [(os, arch)]
162 175 for target_str in &config.targets {
163 176 let Some((target_os, arch)): Option<(&str, &str)> = target_str.split_once('/') else {
164 177 let msg = format!("invalid target format: {target_str}\n");
@@ -184,22 +197,66 @@ async fn run_build(state: &AppState, build: &DbBuild, config: &DbBuildConfig) {
184 197 }
185 198 };
186 199
187 - match execute_target(state, build, config, host, target_os, arch).await {
188 - Ok((s3_key, signature)) => {
189 - artifact_keys.push((target_os.to_string(), arch.to_string(), s3_key, signature));
200 + let entry = (target_os.to_string(), arch.to_string());
201 + match groups.iter_mut().find(|(h, _)| h == host) {
202 + Some((_, targets)) => targets.push(entry),
203 + None => groups.push((host.to_string(), vec![entry])),
204 + }
205 + }
206 +
207 + // One task per host; targets within a host run serially. Results are gathered
208 + // by group index so the merged artifact order stays deterministic regardless
209 + // of which host finishes first.
210 + let group_count = groups.len();
211 + let mut set: tokio::task::JoinSet<(usize, GroupOutput)> = tokio::task::JoinSet::new();
212 + for (idx, (host, targets)) in groups.into_iter().enumerate() {
213 + let state = state.clone();
214 + let build = build.clone();
215 + let config = config.clone();
216 + set.spawn(async move {
217 + let mut oks: Vec<TargetArtifact> = Vec::new();
218 + let mut errs: Vec<TargetError> = Vec::new();
219 + for (target_os, arch) in &targets {
220 + match execute_target(&state, &build, &config, &host, target_os, arch).await {
221 + Ok((s3_key, signature)) => {
222 + oks.push((target_os.clone(), arch.clone(), s3_key, signature));
223 + }
224 + Err(e) => errs.push((format!("{target_os}/{arch}"), e)),
225 + }
190 226 }
227 + (idx, (oks, errs))
228 + });
229 + }
230 +
231 + let mut gathered: Vec<Option<GroupOutput>> = (0..group_count).map(|_| None).collect();
232 + while let Some(res) = set.join_next().await {
233 + match res {
234 + Ok((idx, out)) => gathered[idx] = Some(out),
191 235 Err(e) => {
192 - let msg = format!("target {target_str} failed: {e}\n");
193 - tracing::error!("{}", msg.trim());
194 - let _ = append_log_bounded(state, build.id, &msg).await;
236 + // A host group's task panicked; the build can't be considered
237 + // complete, so fail it rather than silently dropping its targets.
238 + tracing::error!(error = ?e, "build host group task panicked");
195 239 failed_count += 1;
196 240 if first_error.is_none() {
197 - first_error = Some(e);
241 + first_error = Some("a build host group task panicked".to_string());
198 242 }
199 243 }
200 244 }
201 245 }
202 246
247 + for (oks, errs) in gathered.into_iter().flatten() {
248 + artifact_keys.extend(oks);
249 + for (target_str, e) in errs {
250 + let msg = format!("target {target_str} failed: {e}\n");
251 + tracing::error!("{}", msg.trim());
252 + let _ = append_log_bounded(state, build.id, &msg).await;
253 + failed_count += 1;
254 + if first_error.is_none() {
255 + first_error = Some(e);
256 + }
257 + }
258 + }
259 +
203 260 if artifact_keys.is_empty() || failed_count > 0 {
204 261 let err_msg = build_failure_message(artifact_keys.len(), failed_count, first_error.as_deref());
205 262 if let Err(e) = db::builds::update_build_status(
M server/src/db/creator_tiers.rs +9 -4
@@ -474,13 +474,18 @@ pub async fn get_expired_grace_creators(pool: &PgPool) -> Result<Vec<UserId>> {
474 474 Ok(ids)
475 475 }
476 476
477 - /// Mark a creator's post-grace enforcement as applied.
477 + /// Stamp `grace_enforced_at` for all given creators in one statement, paired with
478 + /// `items::hide_all_items_for_users` on the post-grace sweep (Perf-S4, Run 9).
479 + /// No-op on an empty slice.
478 480 #[tracing::instrument(skip_all)]
479 - pub async fn mark_grace_enforced(pool: &PgPool, user_id: UserId) -> Result<()> {
481 + pub async fn mark_grace_enforced_batch(pool: &PgPool, user_ids: &[UserId]) -> Result<()> {
482 + if user_ids.is_empty() {
483 + return Ok(());
484 + }
480 485 sqlx::query(
481 - "UPDATE creator_subscriptions SET grace_enforced_at = NOW() WHERE user_id = $1",
486 + "UPDATE creator_subscriptions SET grace_enforced_at = NOW() WHERE user_id = ANY($1)",
482 487 )
483 - .bind(user_id)
488 + .bind(user_ids)
484 489 .execute(pool)
485 490 .await?;
486 491
M server/src/db/items/mod.rs +9 -5
@@ -1026,18 +1026,22 @@ pub async fn get_item_by_project_and_slug(
1026 1026 Ok(item)
1027 1027 }
1028 1028
1029 - /// Hide all items for a user (set is_public = false). Used for post-grace enforcement.
1030 - /// Returns the number of items hidden.
1029 + /// Hide every public item for ALL given creators in one statement, for the
1030 + /// post-grace scheduler sweep — replaces N per-user UPDATEs on the lock-held tick
1031 + /// (Perf-S4, Run 9). Returns the total rows affected. No-op on an empty slice.
1031 1032 #[tracing::instrument(skip_all)]
1032 - pub async fn hide_all_items_for_user(pool: &PgPool, user_id: UserId) -> Result<u64> {
1033 + pub async fn hide_all_items_for_users(pool: &PgPool, user_ids: &[UserId]) -> Result<u64> {
1034 + if user_ids.is_empty() {
1035 + return Ok(0);
1036 + }
1033 1037 let result = sqlx::query(
1034 1038 r#"
1035 1039 UPDATE items SET is_public = false
1036 - WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
1040 + WHERE project_id IN (SELECT id FROM projects WHERE user_id = ANY($1))
1037 1041 AND is_public = true
1038 1042 "#,
1039 1043 )
1040 - .bind(user_id)
1044 + .bind(user_ids)
1041 1045 .execute(pool)
1042 1046 .await?;
1043 1047
M server/src/db/transactions.rs +49
@@ -510,6 +510,55 @@ pub async fn claim_free_item<'e>(
510 510 Ok(result.rows_affected() > 0)
511 511 }
512 512
513 + /// Batch variant of [`claim_free_item`] for bundle grants: claims every child
514 + /// item for one buyer in a single INSERT instead of N round-trips (each with its
515 + /// own pool acquire) on the Stripe webhook / checkout hot path (Perf-S4, Run 9).
516 + /// Each row is idempotent via the same partial-unique ON CONFLICT as the single
517 + /// claim, and child items deliberately do not increment `sales_count`. Returns the
518 + /// number of rows actually inserted. No-op on an empty slice.
519 + #[tracing::instrument(skip_all)]
520 + pub async fn claim_free_items_batch<'e>(
521 + executor: impl sqlx::PgExecutor<'e>,
522 + buyer_id: UserId,
523 + seller_id: UserId,
524 + seller_username: &str,
525 + parent_transaction_id: Option<TransactionId>,
526 + items: &[(ItemId, &str)],
527 + ) -> Result<u64> {
528 + if items.is_empty() {
529 + return Ok(0);
530 + }
531 + let item_ids: Vec<ItemId> = items.iter().map(|(id, _)| *id).collect();
532 + let item_titles: Vec<&str> = items.iter().map(|(_, title)| *title).collect();
533 +
534 + // The per-row claim id mirrors the single claim's `free-claim-{buyer}-{item}`
535 + // so a later single claim of the same item still collides idempotently.
536 + let result = sqlx::query(
537 + r#"
538 + INSERT INTO transactions
539 + (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents,
540 + stripe_checkout_session_id, status, completed_at, item_title,
541 + seller_username, share_contact, parent_transaction_id)
542 + SELECT
543 + $1, $2, t.item_id, 0, 0,
544 + 'free-claim-' || $1::text || '-' || t.item_id::text,
545 + 'completed', NOW(), t.item_title, $3, false, $4
546 + FROM UNNEST($5::uuid[], $6::text[]) AS t(item_id, item_title)
547 + ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
548 + "#,
549 + )
550 + .bind(buyer_id)
551 + .bind(seller_id)
552 + .bind(seller_username)
553 + .bind(parent_transaction_id)
554 + .bind(&item_ids)
555 + .bind(&item_titles)
556 + .execute(executor)
557 + .await?;
558 +
559 + Ok(result.rows_affected())
560 + }
561 +
513 562 /// Optional parameters for generating a license key inside a claim transaction.
514 563 pub struct LicenseKeyParams<'a> {
515 564 pub key_code: &'a KeyCode,
M server/src/routes/admin/mod.rs +41 -25
@@ -105,39 +105,55 @@ async fn admin_mt_provision(
105 105 let owner_map: std::collections::HashMap<db::UserId, &db::DbUser> =
106 106 owners.iter().map(|u| (u.id, u)).collect();
107 107
108 + // Fan the per-project MT HTTP call (5s timeout each) out over a bounded
109 + // JoinSet instead of awaiting them strictly serially — provisioning latency
110 + // was MT-latency x project-count over the whole table (Perf-S3, Run 9). Cap
111 + // concurrency so a large backfill can't open hundreds of MT connections at
112 + // once; mirror admin_shutdown_notice's bound.
113 + let parallelism = crate::constants::BROADCAST_PARALLELISM;
114 + let mut set: tokio::task::JoinSet<bool> = tokio::task::JoinSet::new();
115 +
108 116 for project in &projects {
109 117 let Some(user) = owner_map.get(&project.user_id) else {
110 118 failed += 1;
111 119 continue;
112 120 };
113 121
114 - match mt
115 - .create_community(&crate::mt_client::CreateCommunityRequest {
116 - name: project.title.clone(),
117 - slug: project.slug.to_string(),
118 - description: project.description.clone(),
119 - owner_mnw_id: *user.id,
120 - owner_username: user.username.to_string(),
121 - owner_display_name: user.display_name.clone(),
122 - })
123 - .await
124 - {
125 - Ok(resp) => {
126 - if let Err(e) =
127 - db::projects::set_mt_community_id(&state.db, project.id, resp.community_id)
128 - .await
129 - {
130 - tracing::error!(error = ?e, project_id = %project.id, "failed to store MT community ID");
131 - failed += 1;
132 - } else {
133 - provisioned += 1;
122 + if set.len() >= parallelism && let Some(res) = set.join_next().await {
123 + if res.unwrap_or(false) { provisioned += 1 } else { failed += 1 }
124 + }
125 +
126 + let mt = mt.clone();
127 + let pool = state.db.clone();
128 + let req = crate::mt_client::CreateCommunityRequest {
129 + name: project.title.clone(),
130 + slug: project.slug.to_string(),
131 + description: project.description.clone(),
132 + owner_mnw_id: *user.id,
133 + owner_username: user.username.to_string(),
134 + owner_display_name: user.display_name.clone(),
135 + };
136 + let project_id = project.id;
137 + let slug = project.slug.to_string();
138 + set.spawn(async move {
139 + match mt.create_community(&req).await {
140 + Ok(resp) => match db::projects::set_mt_community_id(&pool, project_id, resp.community_id).await {
141 + Ok(()) => true,
142 + Err(e) => {
143 + tracing::error!(error = ?e, project_id = %project_id, "failed to store MT community ID");
144 + false
145 + }
146 + },
147 + Err(e) => {
148 + tracing::error!(error = ?e, slug = %slug, "MT community provisioning failed");
149 + false
134 150 }
135 151 }
136 - Err(e) => {
137 - tracing::error!(error = ?e, slug = %project.slug, "MT community provisioning failed");
138 - failed += 1;
139 - }
140 - }
152 + });
153 + }
154 +
155 + while let Some(res) = set.join_next().await {
156 + if res.unwrap_or(false) { provisioned += 1 } else { failed += 1 }
141 157 }
142 158
143 159 tracing::info!(total, provisioned, failed, "MT community backfill complete");
M server/src/routes/stripe/checkout/item.rs +24 -24
@@ -396,29 +396,29 @@ pub(crate) async fn grant_bundle_items(
396 396 _ => return,
397 397 };
398 398
399 - for child in &child_items {
400 - let claim = db::transactions::ClaimParams {
401 - buyer_id,
402 - item_id: child.id,
403 - seller_id,
404 - item_title: &child.title,
405 - seller_username: &seller.username,
406 - share_contact: false,
407 - parent_transaction_id,
408 - };
409 - // Idempotent: ON CONFLICT DO NOTHING if already claimed
410 - if let Err(e) = db::transactions::claim_free_item(&state.db, &claim).await {
411 - tracing::warn!(
412 - child_item_id = %child.id, bundle_id = %bundle_id,
413 - error = ?e, "failed to grant bundle child item"
414 - );
415 - }
416 - // Deliberately NOT incrementing sales_count for child items
399 + // Claim all children in one INSERT instead of N per-child round-trips (each
400 + // re-acquiring a pool connection) on the webhook hot path (Perf-S4, Run 9).
401 + // Idempotent via ON CONFLICT; child items deliberately do not bump sales_count.
402 + let items: Vec<(db::ItemId, &str)> =
403 + child_items.iter().map(|c| (c.id, c.title.as_str())).collect();
404 + match db::transactions::claim_free_items_batch(
405 + &state.db,
406 + buyer_id,
407 + seller_id,
408 + &seller.username,
409 + parent_transaction_id,
410 + &items,
411 + )
412 + .await
413 + {
414 + Ok(granted) => tracing::info!(
415 + bundle_id = %bundle_id, buyer_id = %buyer_id,
416 + child_count = child_items.len(), granted,
417 + "granted bundle child items"
418 + ),
419 + Err(e) => tracing::warn!(
420 + bundle_id = %bundle_id, error = ?e,
421 + "failed to grant bundle child items"
422 + ),
417 423 }
418 -
419 - tracing::info!(
420 - bundle_id = %bundle_id, buyer_id = %buyer_id,
421 - child_count = child_items.len(),
422 - "granted bundle child items"
423 - );
424 424 }
M server/src/scanning/mod.rs +122 -110
@@ -238,6 +238,65 @@ pub struct ScanPipeline {
238 238 metadefender_api_key: Option<String>,
239 239 }
240 240
241 + /// How the bytes to scan are sourced. This is the ONLY thing that differs between
242 + /// the buffered and streaming scan paths — all scan POLICY (CPU layers off the
243 + /// runtime, ClamAV, URLhaus host extraction, external lookups, byte caps) runs in
244 + /// the single private [`ScanPipeline::run_scan`], so a policy cannot drift between
245 + /// the two entry points (ultra-fuzz CHRONIC: scan()/scan_stream() twin divergence,
246 + /// closed Run 9).
247 + enum ScanInput {
248 + /// Whole object held in memory — small uploads.
249 + Buffered(bytes::Bytes),
250 + /// Object spooled to a tempfile and scanned via a memory map. The retained
251 + /// `SpoolHandle` keeps the file on disk (and unlinks it on drop) so ClamAV can
252 + /// stream it by path while the CPU layers read the map.
253 + Spooled {
254 + map: std::sync::Arc<memmap2::Mmap>,
255 + spool: spool::SpoolHandle,
256 + },
257 + }
258 +
259 + /// A cheaply-cloneable, `Send + 'static` view of the scan bytes for the
260 + /// `spawn_blocking` CPU work (a `Bytes` refcount bump or an `Arc<Mmap>` clone).
261 + #[derive(Clone)]
262 + enum ScanBytes {
263 + Buffered(bytes::Bytes),
264 + Mapped(std::sync::Arc<memmap2::Mmap>),
265 + }
266 +
267 + impl ScanBytes {
268 + fn as_slice(&self) -> &[u8] {
269 + match self {
270 + ScanBytes::Buffered(b) => b,
271 + ScanBytes::Mapped(m) => &m[..],
272 + }
273 + }
274 + }
275 +
276 + /// Where ClamAV reads its bytes: the in-memory buffer, or the spool path it
277 + /// streams via INSTREAM frames.
278 + enum ClamavSource {
279 + Buffered(bytes::Bytes),
280 + Path(std::path::PathBuf),
281 + }
282 +
283 + impl ScanInput {
284 + /// A `Send + 'static` byte handle for `spawn_blocking` CPU work.
285 + fn byte_handle(&self) -> ScanBytes {
286 + match self {
287 + ScanInput::Buffered(b) => ScanBytes::Buffered(b.clone()),
288 + ScanInput::Spooled { map, .. } => ScanBytes::Mapped(std::sync::Arc::clone(map)),
289 + }
290 + }
291 +
292 + fn len(&self) -> usize {
293 + match self {
294 + ScanInput::Buffered(b) => b.len(),
295 + ScanInput::Spooled { map, .. } => map.len(),
296 + }
297 + }
298 + }
299 +
241 300 impl ScanPipeline {
242 301 /// Create a new pipeline, compiling YARA rules from the configured directory.
243 302 pub fn new(config: &ScanConfig) -> Result<Self, String> {
@@ -312,111 +371,32 @@ impl ScanPipeline {
312 371 Ok(())
313 372 }
314 373
315 - /// Run all applicable scanning layers against file data.
316 - ///
317 - /// CPU-bound layers (sha256, content-type, structural, archive, yara) run
318 - /// on a blocking-pool thread via `spawn_blocking` so they don't stall the
319 - /// tokio runtime — the 4 in-process layers can each take seconds on a
320 - /// 100 MB file. Network-bound layers (ClamAV, MalwareBazaar) run
321 - /// concurrently with the sync block via `tokio::join!`.
322 - pub async fn scan(
374 + /// Buffered scan entry point — small uploads held in memory. A thin adapter
375 + /// over [`run_scan`](Self::run_scan); all scan policy lives there.
376 + pub(crate) async fn scan(
323 377 self: std::sync::Arc<Self>,
324 378 data: impl Into<bytes::Bytes>,
325 379 file_type: FileType,
326 380 ) -> ScanResult {
327 381 // `bytes::Bytes` is already a cheaply-cloneable refcounted buffer, so the
328 382 // download hands its single aggregated allocation straight here with no
329 - // extra copy — the buffered scan path previously aggregated then `to_vec`'d
330 - // the body, transiently doubling to ~200 MB for a 100 MB file (Run #2
331 - // Performance SERIOUS). Each consumer below clones the `Bytes` handle (a
332 - // refcount bump) and reads it as `&[u8]` by deref coercion.
333 - let data: bytes::Bytes = data.into();
334 - let file_size = data.len() as u64;
335 -
336 - // Sync layers + hash, off the runtime
337 - let sync_data = data.clone();
338 - let sync_self = std::sync::Arc::clone(&self);
339 - let sync_fut =
340 - tokio::task::spawn_blocking(move || sync_self.run_sync_layers(&sync_data, file_type));
341 -
342 - // Async layers — ClamAV needs the bytes, MalwareBazaar needs only the hash
343 - // but the hash is computed in the sync block, so we run MalwareBazaar
344 - // after the sync block returns (its endpoint is fast). ClamAV can run
345 - // concurrently with the sync block.
346 - let clamav_data = data.clone();
347 - let clamav_socket = self.clamav_socket.clone();
348 - let clamav_fut = async move {
349 - match clamav_socket {
350 - Some(socket) => clamav::scan_with_clamav(&socket, &clamav_data).await,
351 - None => LayerResult {
352 - layer: "clamav",
353 - verdict: LayerVerdict::Skip,
354 - detail: Some("ClamAV not configured".to_string()),
355 - },
356 - }
357 - };
358 -
359 - // Layer 7: URLhaus — extract URLs from the bytes, query hosts.
360 - // Runs concurrently with the sync block (independent of the hash).
361 - let urlhaus_data = data.clone();
362 - let urlhaus_enabled = self.urlhaus_enabled;
363 - let urlhaus_key = self.abuse_ch_auth_key.clone();
364 - let urlhaus_fut = async move {
365 - if urlhaus_enabled {
366 - urlhaus::check_urlhaus(&urlhaus_data, urlhaus_key.as_deref()).await
367 - } else {
368 - LayerResult {
369 - layer: "urlhaus",
370 - verdict: LayerVerdict::Skip,
371 - detail: Some("URLhaus lookups disabled".to_string()),
372 - }
373 - }
374 - };
375 -
376 - let (sync_result, clamav_result, urlhaus_result) =
377 - tokio::join!(sync_fut, clamav_fut, urlhaus_fut);
378 - let (mut layers, sha256) = match sync_result {
379 - Ok(v) => v,
380 - Err(join_err) => {
381 - tracing::error!(
382 - error = %join_err,
383 - is_panic = join_err.is_panic(),
384 - "scan sync layers panicked; holding file for review (worker survives)"
385 - );
386 - return panicked_sync_result(file_size);
387 - }
388 - };
389 - layers.push(clamav_result);
390 - layers.push(urlhaus_result);
391 - self.push_external_lookups(&mut layers, &sha256).await;
392 -
393 - let status = final_status(&layers);
394 -
395 - ScanResult {
396 - status,
397 - layers,
398 - sha256,
399 - file_size,
400 - }
383 + // extra copy (the buffered path previously aggregated then `to_vec`'d the
384 + // body, transiently doubling to ~200 MB for a 100 MB file — Run #2).
385 + self.run_scan(ScanInput::Buffered(data.into()), file_type).await
401 386 }
402 387
403 - /// Streaming counterpart to `scan`. Runs against a spooled tempfile so
404 - /// the >100 MB case doesn't hold the whole object in RAM. The CPU
405 - /// layers operate on a memory mapping (`spool::mmap_read`) — pages are
406 - /// demand-paged by the kernel as goblin / yara-x / archive walk them —
407 - /// and ClamAV streams the file via INSTREAM frames.
408 - pub async fn scan_stream(
388 + /// Streaming scan entry point — large uploads spooled to a tempfile and
389 + /// scanned via a memory map, so the >100 MB case doesn't hold the whole object
390 + /// in RAM. A thin adapter over [`run_scan`](Self::run_scan).
391 + pub(crate) async fn scan_stream(
409 392 self: std::sync::Arc<Self>,
410 393 spool: spool::SpoolHandle,
411 394 file_type: FileType,
412 395 ) -> ScanResult {
413 - let file_size = std::fs::metadata(spool.path())
414 - .map(|m| m.len())
415 - .unwrap_or(0);
416 -
417 396 let map = match spool::mmap_read(spool.path()) {
418 397 Ok(m) => std::sync::Arc::new(m),
419 398 Err(e) => {
399 + let file_size = std::fs::metadata(spool.path()).map(|m| m.len()).unwrap_or(0);
420 400 let layer = LayerResult {
421 401 layer: "spool",
422 402 verdict: LayerVerdict::Error,
@@ -430,17 +410,52 @@ impl ScanPipeline {
430 410 };
431 411 }
432 412 };
413 + self.run_scan(ScanInput::Spooled { map, spool }, file_type).await
414 + }
433 415
434 - let sync_map = std::sync::Arc::clone(&map);
416 + /// The single scan body shared by [`scan`](Self::scan) and
417 + /// [`scan_stream`](Self::scan_stream). Every scan policy lives here exactly
418 + /// once; the entry points differ only in how `input` sources its bytes, so a
419 + /// policy (CPU work off the runtime, ClamAV, URLhaus host extraction, external
420 + /// lookups, byte caps) can no longer drift between buffered and streaming
421 + /// (ultra-fuzz CHRONIC, closed Run 9).
422 + ///
423 + /// CPU-bound layers (sha256, content-type, structural, archive, yara) run on a
424 + /// blocking-pool thread via `spawn_blocking`; ClamAV and URLhaus run
425 + /// concurrently with them via `tokio::join!`.
426 + async fn run_scan(
427 + self: std::sync::Arc<Self>,
428 + input: ScanInput,
429 + file_type: FileType,
430 + ) -> ScanResult {
431 + let file_size = input.len() as u64;
432 +
433 + // CPU layers + hash, off the runtime.
434 + let sync_bytes = input.byte_handle();
435 435 let sync_self = std::sync::Arc::clone(&self);
436 - let sync_fut =
437 - tokio::task::spawn_blocking(move || sync_self.run_sync_layers(&sync_map, file_type));
436 + let sync_fut = tokio::task::spawn_blocking(move || {
437 + sync_self.run_sync_layers(sync_bytes.as_slice(), file_type)
438 + });
438 439
440 + // ClamAV: buffered scans the in-memory bytes; spooled streams the file by
441 + // path (INSTREAM frames) so a >100 MB object isn't re-buffered. The
442 + // retained SpoolHandle in `input` keeps the file alive across this join.
439 443 let clamav_socket = self.clamav_socket.clone();
440 - let clamav_path = spool.path().to_path_buf();
444 + let clamav_source = match &input {
445 + ScanInput::Buffered(b) => ClamavSource::Buffered(b.clone()),
446 + ScanInput::Spooled { spool, .. } => ClamavSource::Path(spool.path().to_path_buf()),
447 + };
441 448 let clamav_fut = async move {
442 - match clamav_socket {
443 - Some(socket) => match tokio::fs::File::open(&clamav_path).await {
449 + let Some(socket) = clamav_socket else {
450 + return LayerResult {
451 + layer: "clamav",
452 + verdict: LayerVerdict::Skip,
453 + detail: Some("ClamAV not configured".to_string()),
454 + };
455 + };
456 + match clamav_source {
457 + ClamavSource::Buffered(data) => clamav::scan_with_clamav(&socket, &data).await,
458 + ClamavSource::Path(path) => match tokio::fs::File::open(&path).await {
444 459 Ok(file) => clamav::scan_with_clamav_stream(&socket, file).await,
445 460 Err(e) => LayerResult {
446 461 layer: "clamav",
@@ -448,25 +463,20 @@ impl ScanPipeline {
448 463 detail: Some(format!("open spool for clamav: {e}")),
449 464 },
450 465 },
451 - None => LayerResult {
452 - layer: "clamav",
453 - verdict: LayerVerdict::Skip,
454 - detail: Some("ClamAV not configured".to_string()),
455 - },
456 466 }
457 467 };
458 468
459 - let urlhaus_map = std::sync::Arc::clone(&map);
469 + // URLhaus: extract candidate hosts on the blocking pool (the byte walk can
470 + // page-fault on the mmap), then do only the network lookups async. This is
471 + // ONE policy for both entry points — the exact step that used to drift
472 + // between scan() and scan_stream() (ultra-fuzz CHRONIC).
473 + let urlhaus_bytes = input.byte_handle();
460 474 let urlhaus_enabled = self.urlhaus_enabled;
461 475 let urlhaus_key = self.abuse_ch_auth_key.clone();
462 476 let urlhaus_fut = async move {
463 477 if urlhaus_enabled {
464 - // Host extraction walks the memory-mapped buffer and can
465 - // page-fault; run it on the blocking pool so it doesn't stall a
466 - // tokio worker, then do only the network lookups async
467 - // (ultra-fuzz Run 4 Perf S2).
468 478 let hosts = tokio::task::spawn_blocking(move || {
469 - urlhaus::extract_unique_hosts(&urlhaus_map[..], urlhaus::MAX_HOSTS_PER_FILE)
479 + urlhaus::extract_unique_hosts(urlhaus_bytes.as_slice(), urlhaus::MAX_HOSTS_PER_FILE)
470 480 })
471 481 .await
472 482 .unwrap_or_default();
@@ -488,7 +498,7 @@ impl ScanPipeline {
488 498 tracing::error!(
489 499 error = %join_err,
490 500 is_panic = join_err.is_panic(),
491 - "scan_stream sync layers panicked; holding file for review (worker survives)"
501 + "scan sync layers panicked; holding file for review (worker survives)"
492 502 );
493 503 return panicked_sync_result(file_size);
494 504 }
@@ -498,8 +508,10 @@ impl ScanPipeline {
498 508 self.push_external_lookups(&mut layers, &sha256).await;
499 509
500 510 let status = final_status(&layers);
501 - drop(map);
502 - drop(spool);
511 +
512 + // Dropping `input` here releases the mmap and unlinks the spool tempfile
513 + // (if any), after ClamAV has finished streaming it.
514 + drop(input);
503 515
504 516 ScanResult {
505 517 status,
M server/src/scanning/urlhaus.rs +14 -22
@@ -29,26 +29,10 @@ const MAX_SCAN_BYTES: usize = 4 * 1024 * 1024;
29 29 /// Minimum printable-ASCII run length to consider as a potential string.
30 30 const MIN_RUN_LEN: usize = 6;
31 31
32 - /// Check the file for URLs that appear in URLhaus's known-bad index.
33 - pub async fn check_urlhaus(data: &[u8], auth_key: Option<&str>) -> LayerResult {
34 - if auth_key.is_none() {
35 - return LayerResult {
36 - layer: "urlhaus",
37 - verdict: LayerVerdict::Skip,
38 - detail: Some("No abuse.ch Auth-Key configured".to_string()),
39 - };
40 - }
41 - // In-memory path (small uploads, tests): extraction is cheap on a heap
42 - // buffer. The large-file mmap path extracts inside `spawn_blocking` and calls
43 - // `check_urlhaus_hosts` directly (ultra-fuzz Run 4 Perf S2).
44 - let hosts = extract_unique_hosts(data, MAX_HOSTS_PER_FILE);
45 - check_urlhaus_hosts(hosts, auth_key).await
46 - }
47 -
48 - /// URLhaus lookup over already-extracted hosts. Split from [`check_urlhaus`] so
49 - /// the mmap scan path can run the page-fault-prone host extraction off the async
50 - /// runtime (in `spawn_blocking`) and pass the result here, leaving only the
51 - /// network lookups on the runtime.
32 + /// URLhaus lookup over already-extracted hosts. The scan pipeline always extracts
33 + /// candidate hosts off the async runtime (in `spawn_blocking`, since the byte walk
34 + /// can page-fault on an mmap) and passes the result here, leaving only the network
35 + /// lookups on the runtime — one path for buffered and streamed scans alike.
52 36 pub(crate) async fn check_urlhaus_hosts(hosts: Vec<String>, auth_key: Option<&str>) -> LayerResult {
53 37 let Some(key) = auth_key else {
54 38 return LayerResult {
@@ -295,16 +279,24 @@ mod tests {
295 279 use super::*;
296 280 use serde_json::json;
297 281
282 + /// Exercise the production scan path: extract hosts, then look them up. The
283 + /// scan pipeline always extracts on a blocking thread then calls
284 + /// `check_urlhaus_hosts`; there is no separate buffered helper.
285 + async fn check(data: &[u8], auth_key: Option<&str>) -> LayerResult {
286 + let hosts = extract_unique_hosts(data, MAX_HOSTS_PER_FILE);
287 + check_urlhaus_hosts(hosts, auth_key).await
288 + }
289 +
298 290 #[tokio::test]
299 291 async fn no_auth_key_returns_skip() {
300 - let result = check_urlhaus(b"http://example.com/", None).await;
292 + let result = check(b"http://example.com/", None).await;
301 293 assert_eq!(result.verdict, LayerVerdict::Skip);
302 294 }
303 295
304 296 #[tokio::test]
305 297 async fn no_urls_in_data_passes() {
306 298 // With an auth key, an empty buffer should still pass (no URLs to check).
307 - let result = check_urlhaus(b"plain binary blob no urls here", Some("test")).await;
299 + let result = check(b"plain binary blob no urls here", Some("test")).await;
308 300 assert_eq!(result.verdict, LayerVerdict::Pass);
309 301 assert!(result.detail.unwrap().contains("No URLs"));
310 302 }
M server/src/scheduler/integrity.rs +28 -23
@@ -20,32 +20,37 @@ pub(super) async fn recalculate_all_storage_used(state: &AppState) {
20 20
21 21 /// Enforce post-grace item hiding for creators whose cancellation grace period has expired.
22 22 pub(super) async fn enforce_post_grace_hiding(state: &AppState) {
23 - match db::creator_tiers::get_expired_grace_creators(&state.db).await {
24 - Ok(user_ids) => {
25 - for user_id in user_ids {
26 - match db::items::hide_all_items_for_user(&state.db, user_id).await {
27 - Ok(count) => {
28 - if count > 0 {
29 - tracing::info!(
30 - user_id = %user_id,
31 - items_hidden = count,
32 - "post-grace enforcement: items hidden"
33 - );
34 - }
35 - }
36 - Err(e) => {
37 - tracing::error!(error = ?e, user_id = %user_id, "failed to hide items for post-grace enforcement");
38 - continue;
39 - }
40 - }
41 - if let Err(e) = db::creator_tiers::mark_grace_enforced(&state.db, user_id).await {
42 - tracing::error!(error = ?e, user_id = %user_id, "failed to mark grace enforced");
43 - }
44 - }
45 - }
23 + let user_ids = match db::creator_tiers::get_expired_grace_creators(&state.db).await {
24 + Ok(ids) => ids,
46 25 Err(e) => {
47 26 tracing::error!(error = ?e, "failed to query expired grace creators");
27 + return;
48 28 }
29 + };
30 + if user_ids.is_empty() {
31 + return;
32 + }
33 +
34 + // Set-based hide + mark in two statements instead of 2N per-creator round-trips
35 + // inline on the lock-held tick (Perf-S4, Run 9). Both are idempotent, so if the
36 + // mark fails after the hide the next sweep simply re-hides (a no-op) and re-marks.
37 + let hidden = match db::items::hide_all_items_for_users(&state.db, &user_ids).await {
38 + Ok(count) => count,
39 + Err(e) => {
40 + tracing::error!(error = ?e, "failed to hide items for post-grace enforcement");
41 + return;
42 + }
43 + };
44 + if let Err(e) = db::creator_tiers::mark_grace_enforced_batch(&state.db, &user_ids).await {
45 + tracing::error!(error = ?e, "failed to mark grace enforced");
46 + return;
47 + }
48 + if hidden > 0 {
49 + tracing::info!(
50 + creators = user_ids.len(),
51 + items_hidden = hidden,
52 + "post-grace enforcement: items hidden"
53 + );
49 54 }
50 55 }
51 56
M server/src/scheduler/mod.rs +1 -1
@@ -262,7 +262,7 @@ pub fn spawn_scheduler(
262 262
263 263 // Enforce post-grace item hiding (canceled 30+ days ago). Daily
264 264 // is ample for a 30-day grace window, and it's self-draining
265 - // (mark_grace_enforced), so running it every 60s tick only
265 + // (mark_grace_enforced_batch), so running it every 60s tick only
266 266 // re-issued an empty query — moved here off the hot tick path.
267 267 integrity::enforce_post_grace_hiding(&state).await;
268 268