Skip to main content

max / makenotwork

Bound unbounded work and unblock the runtime (ultra-fuzz Run 10 Performance) - S2: stream SSH build output through a per-stream capped reader instead of Command::output(), so a chatty build can't pin 100+ MB in RAM before the log cap applies. Drains past the cap to avoid blocking the child. - S3: chunk the weekly storage recalc into 500-creator batches instead of one UPDATE with nine LATERAL SUMs over every creator, so it can't become a multi-minute statement that trips the scheduler overrun alert. - S4: run the git-issues default-ref lookup (blocking libgit2) on spawn_blocking so a nav-bar ref lookup can't stall a worker thread. - S5: page the internal sales export (bounding peak memory, with a row ceiling) and cap the admin held-uploads queries at 1000 rows (oldest first) with a warn when the cap is hit. - S6: config-gate the dashboard Forums-tab visibility (matching the other call sites) instead of a synchronous 3s-timeout MT round-trip on every render. M3 (announcements fan-out) left as-is: spawn_bounded_fanout is a deliberate, documented single rate-limited task (Run #14 chronic remediation), not N concurrent spawns. M6 (try_join dashboard awaits) is marginal latency polish now that S6 removed the real per-render cost; the remaining awaits have data dependencies or are cheap session reads. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-30 14:28 UTC
Commit: c906608abcc2898247269be81956618ff192f23a
Parent: 5716d14
9 files changed, +175 insertions, -65 deletions
M server/src/build_runner.rs +64 -8
@@ -546,24 +546,65 @@ async fn run_ssh_command(host: &str, command: &str) -> std::result::Result<Strin
546 546 }
547 547 args.push(host);
548 548 args.push(command);
549 - let output = tokio::process::Command::new("ssh")
549 + let mut child = tokio::process::Command::new("ssh")
550 550 .args(&args)
551 - .output()
552 - .await
551 + .stdin(std::process::Stdio::null())
552 + .stdout(std::process::Stdio::piped())
553 + .stderr(std::process::Stdio::piped())
554 + .spawn()
553 555 .map_err(|e| format!("failed to spawn ssh: {e}"))?;
554 556
555 - if output.status.success() {
556 - Ok(String::from_utf8_lossy(&output.stdout).to_string())
557 + let stdout_pipe = child.stdout.take().expect("stdout piped");
558 + let stderr_pipe = child.stderr.take().expect("stderr piped");
559 +
560 + // Stream both pipes with a per-stream cap instead of Command::output(), which
561 + // would buffer all of a chatty 30-min build's output in RAM before the log
562 + // cap is ever applied. read_capped keeps draining past the cap (so the child
563 + // never blocks on a full pipe) but only retains the first BUILD_MAX_LOG_BYTES
564 + // (ultra-fuzz Run 10 Perf S2). Reading both concurrently with wait() avoids
565 + // the single-pipe-fills-and-deadlocks hazard.
566 + let (stdout_buf, stderr_buf, status) = tokio::join!(
567 + read_capped(stdout_pipe, BUILD_MAX_LOG_BYTES),
568 + read_capped(stderr_pipe, BUILD_MAX_LOG_BYTES),
569 + child.wait(),
570 + );
571 + let status = status.map_err(|e| format!("ssh wait failed: {e}"))?;
572 +
573 + if status.success() {
574 + Ok(stdout_buf)
557 575 } else {
558 - let stderr = String::from_utf8_lossy(&output.stderr);
559 576 Err(format!(
560 577 "exit code {}: {}",
561 - output.status.code().unwrap_or(-1),
562 - stderr.trim()
578 + status.code().unwrap_or(-1),
579 + stderr_buf.trim()
563 580 ))
564 581 }
565 582 }
566 583
584 + /// Drain an async reader to completion but retain only the first `cap` bytes.
585 + /// Draining past the cap keeps the child process from blocking on a full pipe
586 + /// buffer; retaining only `cap` bounds memory regardless of output volume.
587 + async fn read_capped<R>(mut reader: R, cap: usize) -> String
588 + where
589 + R: tokio::io::AsyncRead + Unpin,
590 + {
591 + use tokio::io::AsyncReadExt;
592 + let mut kept = Vec::new();
593 + let mut chunk = [0u8; 8192];
594 + loop {
595 + match reader.read(&mut chunk).await {
596 + Ok(0) | Err(_) => break,
597 + Ok(n) => {
598 + if kept.len() < cap {
599 + let take = n.min(cap - kept.len());
600 + kept.extend_from_slice(&chunk[..take]);
601 + }
602 + }
603 + }
604 + }
605 + String::from_utf8_lossy(&kept).into_owned()
606 + }
607 +
567 608 /// Download a file from a remote host via SCP.
568 609 async fn run_scp_download(
569 610 host: &str,
@@ -736,6 +777,21 @@ fn shell_escape(s: &str) -> String {
736 777 mod tests {
737 778 use super::*;
738 779
780 + #[tokio::test]
781 + async fn read_capped_truncates_to_cap() {
782 + // 10k bytes through a 4k cap retains exactly 4k (the rest is drained and
783 + // discarded so the child never blocks on a full pipe).
784 + let data = vec![b'x'; 10_000];
785 + let out = read_capped(&data[..], 4096).await;
786 + assert_eq!(out.len(), 4096);
787 + }
788 +
789 + #[tokio::test]
790 + async fn read_capped_returns_all_when_under_cap() {
791 + let out = read_capped(&b"hello world"[..], 4096).await;
792 + assert_eq!(out, "hello world");
793 + }
794 +
739 795 #[test]
740 796 fn build_failure_message_partial() {
741 797 assert_eq!(
M server/src/db/creator_tiers.rs +32 -5
@@ -558,12 +558,38 @@ pub async fn is_in_grace_period(pool: &PgPool, user_id: UserId) -> Result<bool>
558 558 Ok(in_grace)
559 559 }
560 560
561 - /// Batch-recalculate storage_used_bytes for ALL creators in a single query.
562 - ///
563 - /// Uses the same CTE logic as `recalculate_storage_used` but operates on all
564 - /// creator users at once, avoiding the N+1 loop. Returns the number of rows updated.
561 + /// Creators per batch for the weekly storage recalc. Bounds each UPDATE's
562 + /// LATERAL SUM fan-out so a large creator base can't produce one multi-minute
563 + /// statement that trips the scheduler overrun alert; the connection is released
564 + /// between batches (ultra-fuzz Run 10 Perf S3).
565 + const STORAGE_RECALC_BATCH: usize = 500;
566 +
567 + /// IDs of every creator the storage recalc covers.
568 + async fn get_storage_recalc_creator_ids(pool: &PgPool) -> Result<Vec<UserId>> {
569 + let ids: Vec<UserId> =
570 + sqlx::query_scalar("SELECT id FROM users WHERE can_create_projects = true ORDER BY id")
571 + .fetch_all(pool)
572 + .await?;
573 + Ok(ids)
574 + }
575 +
576 + /// Weekly storage drift correction. Chunks creators into bounded batches so each
577 + /// statement stays short, returning the total number of corrected rows.
565 578 #[tracing::instrument(skip_all)]
566 579 pub async fn recalculate_all_storage_batch(pool: &PgPool) -> Result<u64> {
580 + let creator_ids = get_storage_recalc_creator_ids(pool).await?;
581 + let mut corrected = 0u64;
582 + for batch in creator_ids.chunks(STORAGE_RECALC_BATCH) {
583 + corrected += recalculate_storage_for_user_batch(pool, batch).await?;
584 + }
585 + Ok(corrected)
586 + }
587 +
588 + /// Recalculate `storage_used_bytes` for a specific batch of creators.
589 + async fn recalculate_storage_for_user_batch(pool: &PgPool, user_ids: &[UserId]) -> Result<u64> {
590 + if user_ids.is_empty() {
591 + return Ok(0);
592 + }
567 593 let result = sqlx::query(
568 594 r#"
569 595 UPDATE users SET storage_used_bytes = totals.total
@@ -629,11 +655,12 @@ pub async fn recalculate_all_storage_batch(pool: &PgPool) -> Result<u64> {
629 655 FROM project_images pi JOIN projects p ON pi.project_id = p.id
630 656 WHERE p.user_id = u.id
631 657 ) project_gallery ON true
632 - WHERE u.can_create_projects = true
658 + WHERE u.can_create_projects = true AND u.id = ANY($1)
633 659 ) totals
634 660 WHERE users.id = totals.user_id AND users.storage_used_bytes IS DISTINCT FROM totals.total
635 661 "#,
636 662 )
663 + .bind(user_ids)
637 664 .execute(pool)
638 665 .await?;
639 666
M server/src/db/scanning.rs +2
@@ -187,6 +187,7 @@ pub async fn get_held_items(db: &PgPool) -> Result<Vec<HeldItemRow>, sqlx::Error
187 187 JOIN users u ON u.id = p.user_id
188 188 WHERE i.scan_status = 'held_for_review'
189 189 ORDER BY i.updated_at ASC
190 + LIMIT 1000
190 191 "#,
191 192 )
192 193 .fetch_all(db)
@@ -217,6 +218,7 @@ pub async fn get_held_versions(db: &PgPool) -> Result<Vec<HeldVersionRow>, sqlx:
217 218 JOIN users u ON u.id = p.user_id
218 219 WHERE v.scan_status = 'held_for_review'
219 220 ORDER BY v.created_at ASC
221 + LIMIT 1000
220 222 "#,
221 223 )
222 224 .fetch_all(db)
M server/src/routes/admin/uploads.rs +9
@@ -114,6 +114,15 @@ pub(super) async fn admin_uploads(
114 114
115 115 let held_items = db::scanning::get_held_items(&state.db).await?;
116 116 let held_versions = db::scanning::get_held_versions(&state.db).await?;
117 + // The held queries are capped at 1000 rows each (oldest first). Hitting the
118 + // cap is an incident-scale backlog; surface it rather than silently truncate.
119 + if held_items.len() >= 1000 || held_versions.len() >= 1000 {
120 + tracing::warn!(
121 + items = held_items.len(),
122 + versions = held_versions.len(),
123 + "admin held-uploads list hit the 1000-row display cap (oldest shown first)"
124 + );
125 + }
117 126
118 127 let mut held_uploads: Vec<AdminHeldUploadRow> = Vec::new();
119 128 held_uploads.extend(held_items.iter().map(AdminHeldUploadRow::from_held_item));
M server/src/routes/api/internal/creators.rs +49 -23
@@ -304,31 +304,57 @@ pub(super) async fn export_sales(
304 304 _auth: ServiceAuth,
305 305 Query(query): Query<UserIdQuery>,
306 306 ) -> Result<impl IntoResponse> {
307 - let rows =
308 - db::transactions::get_seller_transactions_for_export(&state.db, query.user_id).await?;
307 + // Page the read so a whale seller's full history isn't loaded by one
308 + // unbounded query; peak memory is one batch plus the accumulating CSV.
309 + // Mirrors the streaming web export's batch size and row ceiling (ultra-fuzz
310 + // Run 10 Perf S5).
311 + const BATCH: i64 = 2_000;
312 + const MAX_ROWS: usize = 1_000_000;
309 313
310 314 let mut csv = String::from("Date,Item ID,Item Title,Amount,Status,Buyer Email\n");
311 - for tx in &rows {
312 - let date = tx.created_at.format("%Y-%m-%d %H:%M:%S").to_string();
313 - let item_id = tx
314 - .item_id
315 - .map(|id| id.to_string())
316 - .unwrap_or_default();
317 - let item_title = tx.item_title.as_deref().unwrap_or("");
318 - let amount = format!("{}.{:02}", tx.amount_cents / 100, tx.amount_cents.abs() % 100);
319 - let status = tx.status.to_string();
320 - let email = tx.buyer_email.as_deref().unwrap_or("");
321 -
322 - csv.push_str(&format!(
323 - "{},{},{},{},{},{}\n",
324 - helpers::sanitize_csv_cell(&date),
325 - helpers::sanitize_csv_cell(&item_id),
326 - helpers::sanitize_csv_cell(item_title),
327 - amount,
328 - helpers::sanitize_csv_cell(&status),
329 - helpers::sanitize_csv_cell(email),
330 - ));
315 + let mut offset = 0i64;
316 + let mut total = 0usize;
317 + loop {
318 + let rows = db::transactions::get_seller_transactions_for_export_page(
319 + &state.db,
320 + query.user_id,
321 + BATCH,
322 + offset,
323 + )
324 + .await?;
325 + let n = rows.len();
326 + for tx in &rows {
327 + let date = tx.created_at.format("%Y-%m-%d %H:%M:%S").to_string();
328 + let item_id = tx.item_id.map(|id| id.to_string()).unwrap_or_default();
329 + let item_title = tx.item_title.as_deref().unwrap_or("");
330 + let amount = format!("{}.{:02}", tx.amount_cents / 100, tx.amount_cents.abs() % 100);
331 + let status = tx.status.to_string();
332 + let email = tx.buyer_email.as_deref().unwrap_or("");
333 +
334 + csv.push_str(&format!(
335 + "{},{},{},{},{},{}\n",
336 + helpers::sanitize_csv_cell(&date),
337 + helpers::sanitize_csv_cell(&item_id),
338 + helpers::sanitize_csv_cell(item_title),
339 + amount,
340 + helpers::sanitize_csv_cell(&status),
341 + helpers::sanitize_csv_cell(email),
342 + ));
343 + }
344 + offset += n as i64;
345 + total += n;
346 + if (n as i64) < BATCH {
347 + break;
348 + }
349 + if total >= MAX_ROWS {
350 + tracing::warn!(
351 + user_id = %query.user_id,
352 + rows = total,
353 + "internal sales export hit the row ceiling; truncating"
354 + );
355 + break;
356 + }
331 357 }
332 358
333 - Ok(Json(serde_json::json!({ "csv": csv, "row_count": rows.len() })))
359 + Ok(Json(serde_json::json!({ "csv": csv, "row_count": total })))
334 360 }
M server/src/routes/git_issues/issues.rs +2 -2
@@ -56,7 +56,7 @@ pub(super) async fn issue_list(
56 56
57 57 let total_pages = (total + per_page - 1) / per_page;
58 58 let (open_count, closed_count) = db::issues::get_issue_counts(&state.db, resolved.db_repo.id).await?;
59 - let current_ref = default_ref(&state, &owner, &repo_name);
59 + let current_ref = default_ref(&state, &owner, &repo_name).await;
60 60 let csrf_token = get_csrf_token(&session).await;
61 61
62 62 let is_owner = maybe_user.as_ref().map(|u| u.id) == Some(resolved.db_user.id);
@@ -105,7 +105,7 @@ pub(super) async fn issue_detail(
105 105 let is_owner = maybe_user.as_ref().map(|u| u.id) == Some(resolved.db_user.id);
106 106
107 107 let (open_issue_count, _) = db::issues::get_issue_counts(&state.db, resolved.db_repo.id).await?;
108 - let current_ref = default_ref(&state, &owner, &repo_name);
108 + let current_ref = default_ref(&state, &owner, &repo_name).await;
109 109 let csrf_token = get_csrf_token(&session).await;
110 110
111 111 let email_address = format!("{}+{}@issues.makenot.work", owner, repo_name);
M server/src/routes/git_issues/mod.rs +11 -7
@@ -28,16 +28,20 @@ pub fn git_issue_routes() -> CsrfRouter<AppState> {
28 28 }
29 29
30 30 /// Get the default branch name for a repo (for nav bar links).
31 - fn default_ref(state: &AppState, owner: &str, repo_name: &str) -> String {
31 + async fn default_ref(state: &AppState, owner: &str, repo_name: &str) -> String {
32 32 let root = match repos_root(state) {
33 33 Ok(r) => r,
34 34 Err(_) => return "main".to_string(),
35 35 };
36 - match crate::git::open_repo(&root, owner, repo_name) {
37 - Ok(repo) => {
38 - let info = crate::git::repo_info(&repo, repo_name);
39 - info.default_branch
40 - }
36 + let owner = owner.to_string();
37 + let repo_name = repo_name.to_string();
38 + // libgit2 open + repo_info are blocking filesystem work; run them on the
39 + // blocking pool so a nav-bar ref lookup can't stall a worker thread
40 + // (ultra-fuzz Run 10 Perf S4).
41 + tokio::task::spawn_blocking(move || match crate::git::open_repo(&root, &owner, &repo_name) {
42 + Ok(repo) => crate::git::repo_info(&repo, &repo_name).default_branch,
41 43 Err(_) => "main".to_string(),
42 - }
44 + })
45 + .await
46 + .unwrap_or_else(|_| "main".to_string())
43 47 }
M server/src/routes/git_issues/settings.rs +1 -1
@@ -38,7 +38,7 @@ pub(super) async fn repo_settings_form(
38 38 let projects = db::projects::get_projects_by_user(&state.db, user.id).await?;
39 39 let linked_project_id = resolved.db_repo.project_id.map(|pid| pid.to_string()).unwrap_or_default();
40 40 let (open_issue_count, _) = db::issues::get_issue_counts(&state.db, resolved.db_repo.id).await.unwrap_or((0, 0));
41 - let current_ref = default_ref(&state, &owner, &repo_name);
41 + let current_ref = default_ref(&state, &owner, &repo_name).await;
42 42 let csrf_token = get_csrf_token(&session).await;
43 43
44 44 Ok(GitRepoSettingsTemplate {
M server/src/routes/pages/dashboard/main.rs +5 -19
@@ -117,25 +117,11 @@ pub(super) async fn dashboard(
117 117 };
118 118
119 119 // Check if user has MT forum memberships (for Forums tab visibility)
120 - let has_mt_memberships = if let Some(ref mt_url) = state.config.mt_base_url {
121 - let url = format!("{}/api/user/{}/summary", mt_url, session_user.id);
122 - match crate::helpers::HTTP_CLIENT
123 - .get(&url)
124 - .timeout(std::time::Duration::from_secs(3))
125 - .send()
126 - .await
127 - {
128 - Ok(resp) if resp.status().is_success() => resp
129 - .json::<serde_json::Value>()
130 - .await
131 - .ok()
132 - .and_then(|j| j["memberships"].as_array().map(|a| !a.is_empty()))
133 - .unwrap_or(false),
134 - _ => false,
135 - }
136 - } else {
137 - false
138 - };
120 + // Forums-tab visibility is config-gated (matching the landing and user-settings
121 + // call sites): show the tab whenever MT is configured; the tab lazy-loads the
122 + // user's actual memberships on open. Previously this blocked every dashboard
123 + // render on a synchronous 3s-timeout MT round-trip (ultra-fuzz Run 10 Perf S6).
124 + let has_mt_memberships = state.config.mt_base_url.is_some();
139 125
140 126 let suspended = db_user.is_suspended();
141 127 let suspension_reason = db_user.suspension_reason.clone();