max / makenotwork

Give scheduler lock a dedicated pool; drain worker quarantine spawns The advisory lock borrowed a connection from the 25-conn request pool and held it for the whole tick (24/25 serving capacity during ticks) — an accepted residual now closed: spawn_scheduler creates a dedicated 1-connection pool once and acquires the lock from it, falling back to the request pool if the side-pool can't be created so the deploy-critical scheduler still runs. The two fire-and-forget quarantine side-effects (WAM ticket, CF purge) were raw tokio::spawn — unbounded and dropped on shutdown. Route them through the existing bounded, shutdown-drained background pool via a new WorkerContext.bg handle. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: Max Johnson <me@maxj.phd> · 2026-06-25 05:12 UTC

Commit: e9ecf1b7490e1e39d43c03545c2f3d5271dd25bd

Parent: 0994faa

4 files changed, +24 insertions, -12 deletions

M server/src/main.rs +1

			@@ -397,6 +397,7 @@ async fn main() {
397	397		pipeline: scanner,
398	398		scan_semaphore: state.scan_semaphore.clone(),
399	399		wam: state.wam.clone(),
	400	+	bg: state.bg.clone(),
400	401		cloudflare: makenotwork::cloudflare::CloudflarePurger::from_env(),
401	402		cdn_base_url: state.config.cdn_base_url.as_deref().map(std::sync::Arc::from),
402	403		});

M server/src/scanning/worker.rs +6 -2

			@@ -39,6 +39,10 @@ pub struct WorkerContext {
39	39		pub pipeline: Arc<ScanPipeline>,
40	40		pub scan_semaphore: Arc<Semaphore>,
41	41		pub wam: Option<WamClient>,
	42	+	/// Bounded, shutdown-drained background pool. Quarantine side-effects (WAM
	43	+	/// ticket, CF purge) go through this rather than a raw `tokio::spawn` so they
	44	+	/// can't accumulate unbounded or be dropped on shutdown (Run 9).
	45	+	pub bg: crate::background::BackgroundTx,
42	46		/// Cloudflare edge-cache purger, `None` when `CF_API_TOKEN`/`CF_ZONE_ID`
43	47		/// aren't configured. On quarantine we delete the origin object and, if a
44	48		/// purger is present, evict its `cdn_base_url`-prefixed URL from the edge so
			@@ -290,7 +294,7 @@ async fn run_pipeline_and_decide(
290	294		job.user_id, failed_layers.join(", "), job.file_size_bytes,
291	295		);
292	296		let s3_key = job.s3_key.clone();
293		-	tokio::spawn(async move {
	297	+	ctx.bg.spawn("malware quarantine ticket", async move {
294	298		wam.create_ticket(&title, Some(&body), "high", "malware-quarantine", Some(&s3_key)).await;
295	299		});
296	300		}
			@@ -387,7 +391,7 @@ async fn run_pipeline_and_decide(
387	391		// Purging a never-cached URL is harmless, so we don't gate on kind.
388	392		if let (Some(cf), Some(base)) = (ctx.cloudflare.clone(), ctx.cdn_base_url.clone()) {
389	393		let url = format!("{}/{}", base.trim_end_matches('/'), job.s3_key);
390		-	tokio::spawn(async move {
	394	+	ctx.bg.spawn("malware quarantine cdn purge", async move {
391	395		cf.purge_urls(vec![url]).await;
392	396		});
393	397		}

M server/src/scheduler/mod.rs +16 -10

			@@ -81,6 +81,17 @@ pub fn spawn_scheduler(
81	81		let mut last_overrun_alert: Option<std::time::Instant> = None;
82	82		let mut last_panic_alert: Option<std::time::Instant> = None;
83	83
	84	+	// Dedicated 1-connection pool for the advisory lock, created once. Holding
	85	+	// the lock across a tick on this pool no longer borrows from the 25-conn
	86	+	// request pool (was an accepted residual; Run 9). Falls back to the request
	87	+	// pool if the side-pool can't be created so the scheduler still runs.
	88	+	let lock_pool = sqlx::postgres::PgPoolOptions::new()
	89	+	.max_connections(1)
	90	+	.connect(&state.config.database_url)
	91	+	.await
	92	+	.map_err(\|e\| tracing::warn!(error = ?e, "scheduler: dedicated lock pool unavailable, using request pool"))
	93	+	.ok();
	94	+
84	95		loop {
85	96		tokio::select! {
86	97		_ = interval.tick() => {}
			@@ -95,16 +106,11 @@ pub fn spawn_scheduler(
95	106
96	107		// Pin advisory lock to a dedicated connection held for the entire tick.
97	108		// pg_try_advisory_lock is session-scoped — holding the connection prevents
98		-	// another instance from acquiring the lock until this tick completes.
99		-	//
100		-	// Accepted residual (ultra-fuzz Run 4 Perf, decision 2026-06-23): this
101		-	// borrows 1 of the request pool's connections for the tick's duration
102		-	// (effective serving capacity 24/25 during a tick). The architecturally
103		-	// cleaner shape is a dedicated 1-connection side-pool for the lock, but
104		-	// that change touches the deploy-critical scheduler's connection
105		-	// lifecycle for a small steady-state gain; revisit if pool-pressure
106		-	// metrics show it matters.
107		-	let mut lock_conn = match state.db.acquire().await {
	109	+	// another instance from acquiring the lock until this tick completes. The
	110	+	// connection comes from the dedicated `lock_pool` (1 conn), or the request
	111	+	// pool if that pool couldn't be created.
	112	+	let lock_source = lock_pool.as_ref().unwrap_or(&state.db);
	113	+	let mut lock_conn = match lock_source.acquire().await {
108	114		Ok(conn) => conn,
109	115		Err(e) => {
110	116		tracing::warn!(error = ?e, "scheduler: failed to acquire connection for advisory lock, skipping tick");

M server/tests/harness/mod.rs +1

			@@ -545,6 +545,7 @@ impl TestHarness {
545	545		pipeline: deps.pipeline.clone(),
546	546		scan_semaphore: deps.semaphore.clone(),
547	547		wam: None,
	548	+	bg: makenotwork::background::spawn_pool(),
548	549		// No Cloudflare purge in tests; quarantine still deletes from origin.
549	550		cloudflare: None,
550	551		cdn_base_url: None,