Skip to main content

max / makenotwork

perf: one maintenance scheduler, shared scan external-lookup helper Drive Performance A -> A+ (ultra-fuzz Run 7 --deep). Consolidate the two periodic-maintenance engines into one (Perf-S3 mandatory surprise). The monitor ran a daily prune/compaction block (health history, sync-log prune + cursor compaction, OAuth code/refresh-token cleanup) behind its OWN advisory lock -- a second, parallel scheduler. Those five ops move into the real scheduler's daily block under the single SCHEDULER_ADVISORY_LOCK the rest of periodic maintenance already uses; MONITOR_PRUNE_LOCK_ID and the monitor's run_daily_maintenance(_locked) are deleted. The monitor is now health-snapshot/alerting/session-cache only. One lock, one place to add a daily job. Dedup the external-lookup block duplicated between scan() and scan_stream() into one push_external_lookups helper (MalwareBazaar by hash, then the suspicion-gated MetaDefender second opinion, each under the shared lookup_with_timeout deadline) so the buffered and streaming paths can't drift. Gate: cargo clippy --all-targets clean, cargo test --lib (1770), cargo test --test integration (987). No version bump, no deploy. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-25 01:09 UTC
Commit: 5a70e7603a2904329c0bd632f08f0019bbedc54f
Parent: 0cb8326
3 files changed, +54 insertions, -158 deletions
M server/src/monitor.rs +4 -111
@@ -141,7 +141,6 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
141 141 // true so the first overrun after boot can alert.
142 142 let mut pool_pressure_armed: bool = true;
143 143 let mut last_pg_activity_alert_at: Option<Instant> = None;
144 - let mut prune_counter: u64 = 0;
145 144
146 145 loop {
147 146 tokio::select! {
@@ -438,116 +437,10 @@ async fn run_monitor_loop(state: AppState, mut shutdown_rx: watch::Receiver<()>)
438 437 .session_cache
439 438 .retain(|_, validated_at| validated_at.elapsed() < cache_ttl);
440 439
441 - // Prune old records once per day (~1440 checks at 60s interval)
442 - prune_counter += 1;
443 - if prune_counter.is_multiple_of(1440) {
444 - run_daily_maintenance_locked(&state).await;
445 - }
446 - }
447 - }
448 -
449 - /// PostgreSQL advisory-lock id serializing the monitor's daily maintenance across
450 - /// replicas (Perf-S3). Distinct from the scheduler's lock so the two never contend.
451 - const MONITOR_PRUNE_LOCK_ID: i64 = 0x4D4E575F4D4F4E; // "MNW_MON" truncated
452 -
453 - /// Run the once-daily prune/compaction under a cross-replica advisory lock. The
454 - /// monitor is effectively a second maintenance scheduler; it previously ran these
455 - /// ops WITHOUT the advisory lock the real scheduler uses, so on two replicas they
456 - /// double-executed (Perf-S3 mandatory surprise). `pg_try_advisory_lock` is
457 - /// session-scoped — held on `lock_conn` for the duration — so only one instance
458 - /// runs the block per day; the other skips.
459 - async fn run_daily_maintenance_locked(state: &AppState) {
460 - let mut lock_conn = match state.db.acquire().await {
461 - Ok(c) => c,
462 - Err(e) => {
463 - tracing::warn!(error = ?e, "monitor: failed to acquire connection for daily-prune lock, skipping");
464 - return;
465 - }
466 - };
467 - let locked: bool = match sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
468 - .bind(MONITOR_PRUNE_LOCK_ID)
469 - .fetch_one(&mut *lock_conn)
470 - .await
471 - {
472 - Ok(v) => v,
473 - Err(e) => {
474 - tracing::warn!(error = ?e, "monitor: failed to acquire daily-prune advisory lock, skipping");
475 - return;
476 - }
477 - };
478 - if !locked {
479 - tracing::debug!("monitor: daily-prune lock held by another instance, skipping");
480 - return;
481 - }
482 -
483 - run_daily_maintenance(state).await;
484 -
485 - // Explicit unlock (also released when lock_conn drops; explicit survives refactors).
486 - let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
487 - .bind(MONITOR_PRUNE_LOCK_ID)
488 - .execute(&mut *lock_conn)
489 - .await;
490 - }
491 -
492 - /// The actual daily prune/compaction ops, extracted from the monitor loop so the
493 - /// advisory-locked wrapper above can guard them. Each op logs and continues on
494 - /// error — one failing prune must not skip the rest.
495 - async fn run_daily_maintenance(state: &AppState) {
496 - match db::monitor::prune_health_history(&state.db, constants::HEALTH_HISTORY_RETAIN_DAYS).await
497 - {
498 - Ok(deleted) if deleted > 0 => {
499 - tracing::info!(deleted = deleted, "pruned old health history records");
500 - }
501 - Err(e) => {
502 - tracing::warn!(error = ?e, "failed to prune health history");
503 - }
504 - _ => {}
505 - }
506 -
507 - // Prune old sync log entries (age-based: 90 days)
508 - match db::synckit::prune_sync_log(&state.db, constants::SYNC_LOG_RETAIN_DAYS).await {
509 - Ok(deleted) if deleted > 0 => {
510 - tracing::info!(deleted = deleted, "pruned old sync log records");
511 - }
512 - Err(e) => {
513 - tracing::warn!(error = ?e, "failed to prune sync log");
514 - }
515 - _ => {}
516 - }
517 -
518 - // Cursor-based compaction: remove entries all devices have pulled (7-day safety margin)
519 - match db::synckit::compact_all_sync_logs(&state.db, constants::SYNC_LOG_COMPACT_MIN_AGE_DAYS)
520 - .await
521 - {
522 - Ok(deleted) if deleted > 0 => {
523 - tracing::info!(deleted = deleted, "compacted sync log (cursor-based)");
524 - }
525 - Err(e) => {
526 - tracing::warn!(error = ?e, "failed to compact sync log");
527 - }
528 - _ => {}
529 - }
530 -
531 - // Clean up expired/used OAuth authorization codes
532 - match db::oauth::cleanup_expired_oauth_codes(&state.db).await {
533 - Ok(deleted) if deleted > 0 => {
534 - tracing::info!(deleted = deleted, "cleaned up expired OAuth codes");
535 - }
536 - Err(e) => {
537 - tracing::warn!(error = ?e, "failed to clean up OAuth codes");
538 - }
539 - _ => {}
540 - }
541 -
542 - // Clean up expired/used/revoked OAuth refresh tokens
543 - match db::oauth::cleanup_expired_refresh_tokens(&state.db).await {
544 - Ok(deleted) if deleted > 0 => {
545 - tracing::info!(deleted = deleted, "cleaned up expired OAuth refresh tokens");
546 - }
547 - Err(e) => {
548 - tracing::warn!(error = ?e, "failed to clean up OAuth refresh tokens");
549 - }
550 - _ => {}
440 + // Daily prune/compaction (health history, sync log, OAuth cleanup) used
441 + // to live here behind a SECOND advisory lock — a parallel maintenance
442 + // scheduler. It now runs in the real scheduler's daily block under the
443 + // one scheduler lock (Perf-S3). The monitor is health/alerting only.
551 444 }
552 445 }
553 446
M server/src/scanning/mod.rs +24 -47
@@ -388,39 +388,7 @@ impl ScanPipeline {
388 388 };
389 389 layers.push(clamav_result);
390 390 layers.push(urlhaus_result);
391 -
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.
394 - layers.push(if self.malwarebazaar_enabled {
395 - lookup_with_timeout(
396 - "malwarebazaar",
397 - hash_lookup::check_malwarebazaar(&sha256, self.abuse_ch_auth_key.as_deref()),
398 - )
399 - .await
400 - } else {
401 - LayerResult {
402 - layer: "malwarebazaar",
403 - verdict: LayerVerdict::Skip,
404 - detail: Some("MalwareBazaar lookups disabled".to_string()),
405 - }
406 - });
407 -
408 - // Layer 9: MetaDefender (second-opinion). Only invoked when a prior
409 - // layer flagged the file as suspicious — keeps us within the free-tier
410 - // quota and avoids spending budget on uncontroversial uploads.
411 - layers.push(if suspicion_present(&layers) {
412 - lookup_with_timeout(
413 - "metadefender",
414 - metadefender::check_metadefender(&sha256, self.metadefender_api_key.as_deref()),
415 - )
416 - .await
417 - } else {
418 - LayerResult {
419 - layer: "metadefender",
420 - verdict: LayerVerdict::Skip,
421 - detail: Some("No prior suspicion; second-opinion not invoked".to_string()),
422 - }
423 - });
391 + self.push_external_lookups(&mut layers, &sha256).await;
424 392
425 393 let status = final_status(&layers);
426 394
@@ -527,11 +495,31 @@ impl ScanPipeline {
527 495 };
528 496 layers.push(clamav_result);
529 497 layers.push(urlhaus_result);
498 + self.push_external_lookups(&mut layers, &sha256).await;
499 +
500 + let status = final_status(&layers);
501 + drop(map);
502 + drop(spool);
503 +
504 + ScanResult {
505 + status,
506 + layers,
507 + sha256,
508 + file_size,
509 + }
510 + }
530 511
512 + /// Append the post-sync, hash-keyed external-lookup layers to `layers`:
513 + /// Layer 6 MalwareBazaar (by hash) then Layer 9 MetaDefender (a
514 + /// suspicion-gated second opinion that only fires when a prior layer flagged
515 + /// the file, to stay within the free-tier quota). Each runs under the shared
516 + /// aggregate external-lookup deadline (`lookup_with_timeout`, Perf-S1). Shared
517 + /// by `scan` and `scan_stream` so the buffered and streaming paths can't drift.
518 + async fn push_external_lookups(&self, layers: &mut Vec<LayerResult>, sha256: &str) {
531 519 layers.push(if self.malwarebazaar_enabled {
532 520 lookup_with_timeout(
533 521 "malwarebazaar",
534 - hash_lookup::check_malwarebazaar(&sha256, self.abuse_ch_auth_key.as_deref()),
522 + hash_lookup::check_malwarebazaar(sha256, self.abuse_ch_auth_key.as_deref()),
535 523 )
536 524 .await
537 525 } else {
@@ -542,10 +530,10 @@ impl ScanPipeline {
542 530 }
543 531 });
544 532
545 - layers.push(if suspicion_present(&layers) {
533 + layers.push(if suspicion_present(layers) {
546 534 lookup_with_timeout(
547 535 "metadefender",
548 - metadefender::check_metadefender(&sha256, self.metadefender_api_key.as_deref()),
536 + metadefender::check_metadefender(sha256, self.metadefender_api_key.as_deref()),
549 537 )
550 538 .await
551 539 } else {
@@ -555,17 +543,6 @@ impl ScanPipeline {
555 543 detail: Some("No prior suspicion; second-opinion not invoked".to_string()),
556 544 }
557 545 });
558 -
559 - let status = final_status(&layers);
560 - drop(map);
561 - drop(spool);
562 -
563 - ScanResult {
564 - status,
565 - layers,
566 - sha256,
567 - file_size,
568 - }
569 546 }
570 547
571 548 /// CPU-bound layers + SHA-256. Pure sync; safe to call from `spawn_blocking`.
M server/src/scheduler/mod.rs +26
@@ -325,6 +325,32 @@ pub fn spawn_scheduler(
325 325 }
326 326 Err(e) => tracing::error!(error = ?e, "failed to prune processed-webhook markers"),
327 327 }
328 +
329 + // Health/sync/OAuth prune+compaction. These ran in the monitor
330 + // loop under a SECOND advisory lock (a parallel maintenance
331 + // scheduler — Perf-S3 mandatory surprise); consolidated here so
332 + // all periodic maintenance lives under the one scheduler lock and
333 + // there is a single place to add a daily job.
334 + match db::monitor::prune_health_history(&state.db, constants::HEALTH_HISTORY_RETAIN_DAYS).await {
335 + Ok(n) => { if n > 0 { tracing::info!(deleted = n, "pruned old health history records"); } }
336 + Err(e) => tracing::warn!(error = ?e, "failed to prune health history"),
337 + }
338 + match db::synckit::prune_sync_log(&state.db, constants::SYNC_LOG_RETAIN_DAYS).await {
339 + Ok(n) => { if n > 0 { tracing::info!(deleted = n, "pruned old sync log records"); } }
340 + Err(e) => tracing::warn!(error = ?e, "failed to prune sync log"),
341 + }
342 + match db::synckit::compact_all_sync_logs(&state.db, constants::SYNC_LOG_COMPACT_MIN_AGE_DAYS).await {
343 + Ok(n) => { if n > 0 { tracing::info!(deleted = n, "compacted sync log (cursor-based)"); } }
344 + Err(e) => tracing::warn!(error = ?e, "failed to compact sync log"),
345 + }
346 + match db::oauth::cleanup_expired_oauth_codes(&state.db).await {
347 + Ok(n) => { if n > 0 { tracing::info!(deleted = n, "cleaned up expired OAuth codes"); } }
348 + Err(e) => tracing::warn!(error = ?e, "failed to clean up OAuth codes"),
349 + }
350 + match db::oauth::cleanup_expired_refresh_tokens(&state.db).await {
351 + Ok(n) => { if n > 0 { tracing::info!(deleted = n, "cleaned up expired OAuth refresh tokens"); } }
352 + Err(e) => tracing::warn!(error = ?e, "failed to clean up OAuth refresh tokens"),
353 + }
328 354 }
329 355
330 356 })