Skip to main content

max / makenotwork

perf: stream CSV exports + close hot/cold-path inefficiencies (ultra-fuzz Run 4, A+) Phase 4 (Performance). Fixes the one SERIOUS finding plus the localized perf gaps. - S1 (SERIOUS) unbounded export queries -> streamed, bounded exports. Every CSV export (sales, purchases, splits, followers+subscribers, subscriptions) now pages its query (LIMIT/OFFSET, stable (created_at,id) order) and streams the body chunk-by-chunk through a bounded channel: peak memory is one batch and the DB connection is released between batches, instead of loading a creator's whole history into one String from one unbounded full-table query. The no-JS HTMX data-URI fallback (which can't stream) materializes a bounded prefix. An overall EXPORT_MAX_ROWS ceiling bounds worst-case OFFSET cost. Admin/internal callers keep a convenience wrapper that accumulates, now capped. (Keyset pagination is the documented future optimization.) - M-Perf1 (S2) urlhaus host-extraction walked the mmap on the async runtime; split check_urlhaus into extract + lookup and run the page-fault-prone extraction in spawn_blocking on the large-file scan path. - fetch-all-then-find -> indexed point lookups: update_repo_visibility uses get_repo_by_id; begin_rotation uses sync_device_belongs. - spool free-space TOCTOU: reserve headroom for all SCAN_WORKER_COUNT workers, so concurrent spool downloads can't collectively overrun the floor between the check and the write. - health dashboard: the four DB count probes now run via tokio::join! (each still records its own latency) instead of serially. Accepted residuals (Max, 2026-06-23), documented at the sites: the scheduler holds 1/25 pool connections for a tick (dedicated lock side-pool deferred - deploy-critical lifecycle, small gain); the daily item-purge cron gathers all expired keys before its batch delete (bounding couples gather+delete on a low-value cron path).
Co-Authored-By
Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-24 00:08 UTC
Signed with PGP, not checked
Commit: c114130e6e0086f03f532db97f0226f2107f419b
Parent: cc53cb1
17 files changed, +680 insertions, -253 deletions
M server/src/db/follows.rs +13 -2
@@ -229,9 +229,17 @@
229 229 ///
230 230 /// Returns username, display_name, what they follow (user/project), and when.
231 231 #[tracing::instrument(skip_all)]
232 - pub async fn get_followers_for_export(
232 + /// One page of a creator's followers for CSV export, newest first.
233 + ///
234 + /// Paginated (`LIMIT`/`OFFSET`) so the export streams in bounded batches rather
235 + /// than materializing every follower in one query, and so the per-row email
236 + /// `EXISTS` reveal runs only for a bounded page at a time (ultra-fuzz Run 4 S1).
237 + /// Stable `(created_at, follower_id)` ordering keeps OFFSET batches consistent.
238 + pub async fn get_followers_for_export_page(
233 239 pool: &PgPool,
234 240 user_id: UserId,
241 + limit: i64,
242 + offset: i64,
235 243 ) -> Result<Vec<FollowerExportRow>> {
236 244 let rows = sqlx::query_as::<_, FollowerExportRow>(
237 245 r#"
@@ -255,10 +263,13 @@
255 263 JOIN users u ON u.id = f.follower_id
256 264 WHERE (f.target_type = 'user' AND f.target_id = $1)
257 265 OR (f.target_type = 'project' AND f.target_id IN (SELECT id FROM projects WHERE user_id = $1))
258 - ORDER BY f.created_at DESC
266 + ORDER BY f.created_at DESC, f.follower_id DESC
267 + LIMIT $2 OFFSET $3
259 268 "#,
260 269 )
261 270 .bind(user_id)
271 + .bind(limit)
272 + .bind(offset)
262 273 .fetch_all(pool)
263 274 .await?;
264 275
M server/src/db/project_members.rs +12 -2
@@ -323,9 +323,16 @@
323 323 /// - The recipient (collaborator receiving a share), or
324 324 /// - The seller/tip recipient (owner who owes collaborators)
325 325 #[tracing::instrument(skip(pool))]
326 - pub async fn get_splits_for_export(
326 + /// One page of a user's revenue splits for CSV export, newest first.
327 + ///
328 + /// Paginated so the splits export streams in bounded batches instead of loading
329 + /// every split in one query (ultra-fuzz Run 4 S1). Stable `(created_at, id)`
330 + /// ordering keeps OFFSET batches consistent.
331 + pub async fn get_splits_for_export_page(
327 332 pool: &PgPool,
328 333 user_id: UserId,
334 + limit: i64,
335 + offset: i64,
329 336 ) -> Result<Vec<DbSplitExportRow>> {
330 337 let rows = sqlx::query_as::<_, DbSplitExportRow>(
331 338 r#"
@@ -338,10 +345,13 @@
338 345 LEFT JOIN tips tip ON tip.id = rs.tip_id
339 346 WHERE rs.recipient_id = $1
340 347 OR COALESCE(t.seller_id, tip.recipient_id) = $1
341 - ORDER BY rs.created_at DESC
348 + ORDER BY rs.created_at DESC, rs.id DESC
349 + LIMIT $2 OFFSET $3
342 350 "#,
343 351 )
344 352 .bind(user_id)
353 + .bind(limit)
354 + .bind(offset)
345 355 .fetch_all(pool)
346 356 .await?;
347 357
M server/src/db/subscriptions.rs +20 -4
@@ -664,9 +664,14 @@
664 664 ///
665 665 /// Returns username, display_name, tier name, subscription status, and when.
666 666 #[tracing::instrument(skip_all)]
667 - pub async fn get_project_subscribers_for_export(
667 + /// One page of a creator's project subscribers for CSV export, newest first.
668 + /// Paginated for bounded-memory streaming (ultra-fuzz Run 4 S1); stable
669 + /// `(created_at, id)` ordering keeps OFFSET batches consistent.
670 + pub async fn get_project_subscribers_for_export_page(
668 671 pool: &PgPool,
669 672 user_id: UserId,
673 + limit: i64,
674 + offset: i64,
670 675 ) -> Result<Vec<SubscriberExportRow>> {
671 676 let rows = sqlx::query_as!(
672 677 SubscriberExportRow,
@@ -678,9 +683,12 @@
678 683 JOIN users u ON u.id = s.subscriber_id
679 684 JOIN subscription_tiers t ON t.id = s.tier_id
680 685 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
681 - ORDER BY s.created_at DESC
686 + ORDER BY s.created_at DESC, s.id DESC
687 + LIMIT $2 OFFSET $3
682 688 "#,
683 689 user_id as UserId,
690 + limit,
691 + offset,
684 692 )
685 693 .fetch_all(pool)
686 694 .await?;
@@ -693,9 +701,14 @@
693 701 /// Returns project name, tier name, price, subscriber username, status,
694 702 /// billing period dates, and cancellation date.
695 703 #[tracing::instrument(skip_all)]
696 - pub async fn get_subscriptions_for_export(
704 + /// One page of a creator's subscriptions for CSV export, newest first.
705 + /// Paginated for bounded-memory streaming (ultra-fuzz Run 4 S1); stable
706 + /// `(created_at, id)` ordering keeps OFFSET batches consistent.
707 + pub async fn get_subscriptions_for_export_page(
697 708 pool: &PgPool,
698 709 user_id: UserId,
710 + limit: i64,
711 + offset: i64,
699 712 ) -> Result<Vec<SubscriptionExportRow>> {
700 713 let rows = sqlx::query_as!(
701 714 SubscriptionExportRow,
@@ -711,9 +724,12 @@
711 724 JOIN subscription_tiers t ON t.id = s.tier_id
712 725 JOIN projects p ON p.id = s.project_id
713 726 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
714 - ORDER BY s.created_at DESC
727 + ORDER BY s.created_at DESC, s.id DESC
728 + LIMIT $2 OFFSET $3
715 729 "#,
716 730 user_id as UserId,
731 + limit,
732 + offset,
717 733 )
718 734 .fetch_all(pool)
719 735 .await?;
M server/src/db/transactions.rs +79 -2
@@ -359,6 +359,43 @@
359 359 Ok(txs)
360 360 }
361 361
362 + /// One page of a buyer's purchases for CSV export, newest first.
363 + ///
364 + /// Paginated so the purchases export streams in bounded batches rather than
365 + /// loading the buyer's whole history with `limit: None` (ultra-fuzz Run 4 S1).
366 + /// Stable `(created_at, id)` ordering keeps OFFSET batches consistent.
367 + pub async fn get_buyer_transactions_for_export_page(
368 + pool: &PgPool,
369 + buyer_id: UserId,
370 + limit: i64,
371 + offset: i64,
372 + ) -> Result<Vec<DbTransaction>> {
373 + let txs = sqlx::query_as!(
374 + DbTransaction,
375 + r#"
376 + SELECT
377 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
378 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
379 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
380 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
381 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
382 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
383 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
384 + download_token AS "download_token: DownloadToken"
385 + FROM transactions WHERE buyer_id = $1
386 + ORDER BY created_at DESC, id DESC
387 + LIMIT $2 OFFSET $3
388 + "#,
389 + buyer_id as UserId,
390 + limit,
391 + offset,
392 + )
393 + .fetch_all(pool)
394 + .await?;
395 +
396 + Ok(txs)
397 + }
398 +
362 399 /// List transactions where the user is the seller, newest first.
363 400 ///
364 401 /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
@@ -1100,9 +1137,19 @@
1100 1137 /// Respects contact revocations: if a buyer revoked sharing, their email
1101 1138 /// is hidden even if `share_contact` was true on the transaction.
1102 1139 #[tracing::instrument(skip_all)]
1103 - pub async fn get_seller_transactions_for_export(
1140 + /// One page of a seller's sales for CSV export, newest first.
1141 + ///
1142 + /// Paginated (`LIMIT`/`OFFSET`) so the export streams in bounded batches instead
1143 + /// of loading the seller's entire transaction history into memory in one query
1144 + /// (ultra-fuzz Run 4 S1). The `(created_at, id)` ordering is stable so OFFSET
1145 + /// batches don't reorder. (Keyset pagination would avoid OFFSET's deep-scan cost
1146 + /// and is the future optimization; OFFSET is sufficient at current scale and
1147 + /// keeps peak memory + per-query result bounded, which is the DoS fix.)
1148 + pub async fn get_seller_transactions_for_export_page(
1104 1149 pool: &PgPool,
1105 1150 seller_id: UserId,
1151 + limit: i64,
1152 + offset: i64,
1106 1153 ) -> Result<Vec<DbTransactionExportRow>> {
1107 1154 let rows = sqlx::query_as!(
1108 1155 DbTransactionExportRow,
@@ -1120,9 +1167,12 @@
1120 1167 FROM transactions t
1121 1168 LEFT JOIN users u ON u.id = t.buyer_id
1122 1169 WHERE t.seller_id = $1
1123 - ORDER BY t.created_at DESC
1170 + ORDER BY t.created_at DESC, t.id DESC
1171 + LIMIT $2 OFFSET $3
1124 1172 "#,
1125 1173 seller_id as UserId,
1174 + limit,
1175 + offset,
1126 1176 )
1127 1177 .fetch_all(pool)
1128 1178 .await?;
@@ -1130,6 +1180,33 @@
1130 1180 Ok(rows)
1131 1181 }
1132 1182
1183 + /// All of a seller's export rows accumulated into one Vec, bounded to
1184 + /// `EXPORT_ACCUMULATE_CAP` rows. For admin / internal-API callers that need the
1185 + /// full set in memory; the public creator-facing export streams page-by-page via
1186 + /// [`get_seller_transactions_for_export_page`] instead of materializing here.
1187 + pub async fn get_seller_transactions_for_export(
1188 + pool: &PgPool,
1189 + seller_id: UserId,
1190 + ) -> Result<Vec<DbTransactionExportRow>> {
1191 + /// Page size for the accumulating fetch.
1192 + const PAGE: i64 = 5_000;
1193 + /// Cap so even an admin/internal export can't load an unbounded result set.
1194 + const EXPORT_ACCUMULATE_CAP: usize = 1_000_000;
1195 +
1196 + let mut all = Vec::new();
1197 + let mut offset = 0i64;
1198 + loop {
1199 + let page = get_seller_transactions_for_export_page(pool, seller_id, PAGE, offset).await?;
1200 + let n = page.len();
1201 + all.extend(page);
1202 + offset += n as i64;
1203 + if (n as i64) < PAGE || all.len() >= EXPORT_ACCUMULATE_CAP {
1204 + break;
1205 + }
1206 + }
1207 + Ok(all)
1208 + }
1209 +
1133 1210 /// Platform-wide revenue stats: total completed revenue, completed count, refunded count.
1134 1211 #[tracing::instrument(skip_all)]
1135 1212 pub async fn get_platform_revenue_stats(pool: &PgPool) -> Result<(i64, i64, i64)> {
M server/src/scanning/mod.rs +10 -1
@@ -457,7 +457,16 @@
457 457 let urlhaus_key = self.abuse_ch_auth_key.clone();
458 458 let urlhaus_fut = async move {
459 459 if urlhaus_enabled {
460 - urlhaus::check_urlhaus(&urlhaus_map, urlhaus_key.as_deref()).await
460 + // Host extraction walks the memory-mapped buffer and can
461 + // page-fault; run it on the blocking pool so it doesn't stall a
462 + // tokio worker, then do only the network lookups async
463 + // (ultra-fuzz Run 4 Perf S2).
464 + let hosts = tokio::task::spawn_blocking(move || {
465 + urlhaus::extract_unique_hosts(&urlhaus_map[..], urlhaus::MAX_HOSTS_PER_FILE)
466 + })
467 + .await
468 + .unwrap_or_default();
469 + urlhaus::check_urlhaus_hosts(hosts, urlhaus_key.as_deref()).await
461 470 } else {
462 471 LayerResult {
463 472 layer: "urlhaus",
M server/src/scanning/spool.rs +13 -4
@@ -9,7 +9,10 @@
9 9 //! Nothing in the scan pipeline calls this module yet; it is wired up in
10 10 //! later chunks of the scanner-streaming refactor.
11 11
12 - use crate::constants::{SCAN_SPOOL_FREE_RESERVE_BYTES, SCAN_SPOOL_MAX_BYTES, SCAN_SPOOL_ORPHAN_AGE_SECS};
12 + use crate::constants::{
13 + SCAN_SPOOL_FREE_RESERVE_BYTES, SCAN_SPOOL_MAX_BYTES, SCAN_SPOOL_ORPHAN_AGE_SECS,
14 + SCAN_WORKER_COUNT,
15 + };
13 16 use s3_storage::ByteStream;
14 17 use std::path::{Path, PathBuf};
15 18 use tokio::fs::{File, OpenOptions};
@@ -74,10 +77,16 @@
74 77 };
75 78 let free = fs2::available_space(probe)
76 79 .map_err(|e| format!("statvfs {}: {e}", probe.display()))?;
77 - if free.saturating_sub(expected_size) < SCAN_SPOOL_FREE_RESERVE_BYTES {
80 + // The check and the write aren't atomic: with `SCAN_WORKER_COUNT` workers, up
81 + // to that many spool downloads can pass this check and then write at once.
82 + // Reserve headroom for ALL of them rather than just this one, so concurrent
83 + // workers can't collectively overrun the free-space floor (ultra-fuzz Run 4
84 + // Perf TOCTOU). Conservative by at most `(workers-1) * expected`.
85 + let concurrent_reserve = expected_size.saturating_mul(SCAN_WORKER_COUNT as u64);
86 + if free.saturating_sub(concurrent_reserve) < SCAN_SPOOL_FREE_RESERVE_BYTES {
78 87 return Err(format!(
79 - "scan spool volume short on space: free={} expected={} reserve={}",
80 - free, expected_size, SCAN_SPOOL_FREE_RESERVE_BYTES
88 + "scan spool volume short on space: free={} expected={} workers={} reserve={}",
89 + free, expected_size, SCAN_WORKER_COUNT, SCAN_SPOOL_FREE_RESERVE_BYTES
81 90 ));
82 91 }
83 92 Ok(())
M server/src/scanning/urlhaus.rs +21 -3
@@ -23,7 +23,7 @@
23 23
24 24 const URLHAUS_API_URL: &str = "https://urlhaus-api.abuse.ch/v1/host/";
25 25 /// Cap per-upload host lookups to keep within free-tier quotas and bound work.
26 - const MAX_HOSTS_PER_FILE: usize = 5;
26 + pub(crate) const MAX_HOSTS_PER_FILE: usize = 5;
27 27 /// Cap the byte window we scan for URLs; URLs in the wild fit easily here.
28 28 const MAX_SCAN_BYTES: usize = 4 * 1024 * 1024;
29 29 /// Minimum printable-ASCII run length to consider as a potential string.
@@ -31,6 +31,25 @@
31 31
32 32 /// Check the file for URLs that appear in URLhaus's known-bad index.
33 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.
52 + pub(crate) async fn check_urlhaus_hosts(hosts: Vec<String>, auth_key: Option<&str>) -> LayerResult {
34 53 let Some(key) = auth_key else {
35 54 return LayerResult {
36 55 layer: "urlhaus",
@@ -39,7 +58,6 @@
39 58 };
40 59 };
41 60
42 - let hosts = extract_unique_hosts(data, MAX_HOSTS_PER_FILE);
43 61 if hosts.is_empty() {
44 62 return LayerResult {
45 63 layer: "urlhaus",
@@ -160,7 +178,7 @@
160 178
161 179 /// Pull printable-ASCII URL hosts out of the byte buffer. Cap at `max` unique
162 180 /// hosts to bound per-upload work and free-tier quota use.
163 - fn extract_unique_hosts(data: &[u8], max: usize) -> Vec<String> {
181 + pub(crate) fn extract_unique_hosts(data: &[u8], max: usize) -> Vec<String> {
164 182 let scan = if data.len() > MAX_SCAN_BYTES { &data[..MAX_SCAN_BYTES] } else { data };
165 183
166 184 let mut hosts: HashSet<String> = HashSet::new();
M server/src/scheduler/cleanup.rs +7
@@ -309,6 +309,13 @@
309 309 /// Permanently delete items that were soft-deleted more than 7 days ago.
310 310 /// Cleans up S3 objects (item files + version files) and decrements storage
311 311 /// before DB deletion to prevent orphaned storage and accounting drift.
312 + ///
313 + /// Accepted residual (ultra-fuzz Run 4 Perf, decision 2026-06-23): this gathers
314 + /// all expired-item S3 keys into one Vec before the batch delete. It is a daily
315 + /// cron off the request path; steady-state (7-day window) is small, so the only
316 + /// large allocation is transient, after a rare mass-delete event. Bounding it
317 + /// per-tick means coupling a LIMIT on the key-gather to the same slice as the
318 + /// CASCADE delete; deferred as low-value for a cron path. Revisit if it matters.
312 319 pub(super) async fn purge_expired_deleted_items(state: &AppState) {
313 320 // Collect S3 keys from items AND their versions before CASCADE delete destroys the data
314 321 let mut all_s3_keys: Vec<(String, String)> = Vec::new();
M server/src/scheduler/mod.rs +8
@@ -96,6 +96,14 @@
96 96 // Pin advisory lock to a dedicated connection held for the entire tick.
97 97 // pg_try_advisory_lock is session-scoped — holding the connection prevents
98 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.
99 107 let mut lock_conn = match state.db.acquire().await {
100 108 Ok(conn) => conn,
101 109 Err(e) => {
M server/src/routes/api/projects.rs +3 -3
M server/src/routes/synckit/sync.rs +3 -3
M server/src/routes/api/exports/mod.rs +297 -192
M server/src/routes/pages/public/health/mod.rs +29 -28
R server/.sqlx/query-b5fa4c20ec1e09cb5a89fee1674d3ce34fb8016490fc5b286bb171a7d34f8b2c.json → server/.sqlx/query-5013a9ea2c85fe65193f873cfc5d32bdccbedb7de2af30694a52c37df3c592e5.json +5 -3
R server/.sqlx/query-b0e71c186ef4f316174cd31b3903d03e801d8be57576766dbd3fa9a677979ca5.json → server/.sqlx/query-c9b1c6374241b92f67625311ba06f033367de4cf3822ffae9ba04b3dbe3cb9e0.json +5 -3
R server/.sqlx/query-92512050fa13a3ad75ae5e31b848ac1040a9f9c69eecdeef2e03adb90dcbfc69.json → server/.sqlx/query-169ed0d8f74cc69847d5d63b6a1d3b3f4d277cd3540036c7d5278ddc139a3340.json +5 -3
A server/.sqlx/query-17188cab9291709226c6f4586bccfbbd878a4c98b27bb844e8b48c56edd7981a.json +150