Skip to main content

max / makenotwork

Background Stripe fan-outs and close perf cliffs (ultra-fuzz Run 6 Performance) S1/S2/R6-Perf-M1: extract payments::fan_ops::spawn_fan_sub_fanout, a single helper that runs a creator's per-fan-subscription Stripe fan-out on the bounded background queue. Wire it into creator pause, webhook re-subscribe unpause, and admin suspend/unsuspend/terminate so none of those hot paths block on N serial Stripe round-trips (the webhook one no longer risks a timeout/retry storm). S3: cap YARA input at SCAN_YARA_MAX_BYTES (512 MiB) so a large object no longer faults its whole mmap resident per scan; SHA-256 and ClamAV still see the full file. S4: stop the decompression-bomb walk at min(absolute, compressed*MAX_RATIO) instead of always expanding to the 2 GB absolute cap (single-stream and per-entry paths); the ratio verdict fires identically on the early count. S6: batch the git push-refs issue writes into db::issues::create_comments + update_issue_statuses (two queries) instead of one UPDATE + INSERT per referenced commit. R6-Perf-M2: stream export_contacts through spawn_paginated_csv like its sibling exports. R6-Perf-M3: bound the scan spool write at claimed size + slack, not the global 8 GiB ceiling. R6-Perf-M4: creator_stats counts projects/items with COUNT(*) rather than fetching rows to .len(). R6-Perf-M5: suggest_slugs resolves availability with one slug = ANY($1) query instead of up to 9 point queries. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-24 15:39 UTC
Commit: a57614cdc8cb85c4e5ae5bb5844fccfc73c1524c
Parent: f2dec50
17 files changed, +458 insertions, -145 deletions
M server/src/constants.rs +13
@@ -221,6 +221,19 @@ pub const SCAN_SPOOL_MAX_BYTES: u64 = 8 * 1024 * 1024 * 1024;
221 221 /// pending object. The scanner refuses if `statvfs(free) - expected_size`
222 222 /// would drop below this threshold.
223 223 pub const SCAN_SPOOL_FREE_RESERVE_BYTES: u64 = 2 * 1024 * 1024 * 1024;
224 + /// Slack added to a scan job's claimed object size when bounding how many bytes
225 + /// the spool writer will accept. The S3 object size is authoritative, but a
226 + /// small margin absorbs benign content-length/multipart rounding without
227 + /// letting an under-reported object stream the full `SCAN_SPOOL_MAX_BYTES` to
228 + /// scratch before the writer aborts.
229 + pub const SCAN_SPOOL_SLACK_BYTES: u64 = 16 * 1024 * 1024; // 16 MiB
230 + /// Maximum number of bytes fed to YARA in a single scan. yara-x's `Scanner`
231 + /// walks the whole slice, which demand-pages the entire mmap resident — so an
232 + /// 8 GiB object would otherwise pin 8 GiB of page cache per scan (×
233 + /// `SCAN_MAX_CONCURRENT`). Malware signatures cluster near a file's start, and
234 + /// ClamAV (streamed, uncapped) is the full-file backstop, so scanning a generous
235 + /// prefix is the right trade. Above this, YARA sees the prefix and logs the cap.
236 + pub const SCAN_YARA_MAX_BYTES: usize = 512 * 1024 * 1024; // 512 MiB
224 237
225 238 // -- Caddy on-demand TLS --
226 239 // Caps concurrent cache-miss DB lookups in `/api/domains/caddy-ask`. Cache hits
M server/src/db/issues.rs +73
@@ -257,6 +257,79 @@ pub async fn create_comment(
257 257 Ok(comment)
258 258 }
259 259
260 + /// Bulk-insert issue comments in a single statement, then touch each affected
261 + /// issue's `updated_at` once. The git-push processor would otherwise issue one
262 + /// INSERT (and one updated_at UPDATE) per referenced commit, ~100-150 sequential
263 + /// writes per push (ultra-fuzz Run 6 S6). All comments share one author.
264 + #[tracing::instrument(skip_all)]
265 + pub async fn create_comments(
266 + pool: &PgPool,
267 + author_id: UserId,
268 + comments: &[(IssueId, String, String)],
269 + ) -> Result<()> {
270 + if comments.is_empty() {
271 + return Ok(());
272 + }
273 + let issue_ids: Vec<sqlx::types::Uuid> =
274 + comments.iter().map(|(id, _, _)| *id.as_uuid()).collect();
275 + let bodies_md: Vec<&str> = comments.iter().map(|(_, md, _)| md.as_str()).collect();
276 + let bodies_html: Vec<&str> = comments.iter().map(|(_, _, html)| html.as_str()).collect();
277 +
278 + sqlx::query(
279 + r#"
280 + INSERT INTO issue_comments (issue_id, author_user_id, body_markdown, body_html)
281 + SELECT u.issue_id, $1, u.body_md, u.body_html
282 + FROM UNNEST($2::uuid[], $3::text[], $4::text[]) AS u(issue_id, body_md, body_html)
283 + "#,
284 + )
285 + .bind(author_id)
286 + .bind(&issue_ids)
287 + .bind(&bodies_md)
288 + .bind(&bodies_html)
289 + .execute(pool)
290 + .await?;
291 +
292 + // Touch updated_at once per distinct affected issue.
293 + let mut distinct = issue_ids;
294 + distinct.sort_unstable();
295 + distinct.dedup();
296 + sqlx::query("UPDATE issues SET updated_at = NOW() WHERE id = ANY($1)")
297 + .bind(&distinct)
298 + .execute(pool)
299 + .await?;
300 +
301 + Ok(())
302 + }
303 +
304 + /// Bulk status update for the git-push processor. Each entry is `(issue, final
305 + /// status)`; the caller resolves repeated references to one final state so each
306 + /// issue appears at most once.
307 + #[tracing::instrument(skip_all)]
308 + pub async fn update_issue_statuses(
309 + pool: &PgPool,
310 + updates: &[(IssueId, IssueStatus)],
311 + ) -> Result<()> {
312 + if updates.is_empty() {
313 + return Ok(());
314 + }
315 + let ids: Vec<sqlx::types::Uuid> = updates.iter().map(|(id, _)| *id.as_uuid()).collect();
316 + let statuses: Vec<String> = updates.iter().map(|(_, s)| s.to_string()).collect();
317 +
318 + sqlx::query(
319 + r#"
320 + UPDATE issues AS i SET status = u.status, updated_at = NOW()
321 + FROM UNNEST($1::uuid[], $2::text[]) AS u(id, status)
322 + WHERE i.id = u.id
323 + "#,
324 + )
325 + .bind(&ids)
326 + .bind(&statuses)
327 + .execute(pool)
328 + .await?;
329 +
330 + Ok(())
331 + }
332 +
260 333 /// List all comments on an issue with author usernames.
261 334 #[tracing::instrument(skip_all)]
262 335 pub async fn list_comments(
M server/src/db/items/mod.rs +18
@@ -218,6 +218,24 @@ pub async fn get_items_by_user(pool: &PgPool, user_id: UserId) -> Result<Vec<DbI
218 218 Ok(items)
219 219 }
220 220
221 + /// Count a user's (non-deleted) items without materializing the rows. For
222 + /// callers that only need the total, this avoids fetching up to 500 full rows
223 + /// just to `.len()` them (ultra-fuzz Run 6 R6-Perf-M4).
224 + #[tracing::instrument(skip_all)]
225 + pub async fn count_items_by_user(pool: &PgPool, user_id: UserId) -> Result<i64> {
226 + let count: i64 = sqlx::query_scalar(
227 + r#"
228 + SELECT COUNT(*) FROM items i
229 + JOIN projects p ON i.project_id = p.id
230 + WHERE p.user_id = $1 AND i.deleted_at IS NULL
231 + "#,
232 + )
233 + .bind(user_id)
234 + .fetch_one(pool)
235 + .await?;
236 + Ok(count)
237 + }
238 +
221 239 /// Count items per project for all projects owned by a user.
222 240 ///
223 241 /// Returns `(project_id, count)` tuples. Used by the CLI to avoid N+1 queries.
M server/src/db/projects.rs +30
@@ -123,6 +123,36 @@ pub async fn get_projects_by_user(pool: &PgPool, user_id: UserId) -> Result<Vec<
123 123 Ok(projects)
124 124 }
125 125
126 + /// Of the given candidate slugs, return those already taken by a project owned
127 + /// by `user_id`. One indexed `slug = ANY($2)` query replaces a per-candidate
128 + /// point-query loop (ultra-fuzz Run 6 R6-Perf-M5).
129 + #[tracing::instrument(skip_all)]
130 + pub async fn filter_taken_slugs(
131 + pool: &PgPool,
132 + user_id: UserId,
133 + slugs: &[String],
134 + ) -> Result<Vec<String>> {
135 + let taken: Vec<String> =
136 + sqlx::query_scalar("SELECT slug FROM projects WHERE user_id = $1 AND slug = ANY($2)")
137 + .bind(user_id)
138 + .bind(slugs)
139 + .fetch_all(pool)
140 + .await?;
141 + Ok(taken)
142 + }
143 +
144 + /// Count a user's projects without materializing the rows. For callers that only
145 + /// need the total (e.g. stats), this avoids fetching up to 500 full rows just to
146 + /// `.len()` them (ultra-fuzz Run 6 R6-Perf-M4).
147 + #[tracing::instrument(skip_all)]
148 + pub async fn count_projects_by_user(pool: &PgPool, user_id: UserId) -> Result<i64> {
149 + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM projects WHERE user_id = $1")
150 + .bind(user_id)
151 + .fetch_one(pool)
152 + .await?;
153 + Ok(count)
154 + }
155 +
126 156 /// Partially update a project's fields (COALESCE keeps existing values when `None`).
127 157 ///
128 158 /// When `features` is `Some`, the project_type is auto-derived from the new features.
M server/src/db/transactions.rs +43
@@ -1132,6 +1132,49 @@ pub async fn get_seller_contacts(
1132 1132 Ok(rows)
1133 1133 }
1134 1134
1135 + /// One page of a seller's sharing-opted-in contacts for CSV export, newest
1136 + /// first. Paginated so the contacts export streams in bounded batches like its
1137 + /// sibling exports instead of buffering one capped query into a single `String`
1138 + /// (ultra-fuzz Run 6 R6-Perf-M2). `(MAX(completed_at), buyer_id)` ordering is
1139 + /// stable so OFFSET batches don't reorder.
1140 + #[tracing::instrument(skip_all)]
1141 + pub async fn get_seller_contacts_page(
1142 + pool: &PgPool,
1143 + seller_id: UserId,
1144 + limit: i64,
1145 + offset: i64,
1146 + ) -> Result<Vec<DbContactRow>> {
1147 + let rows = sqlx::query_as::<_, DbContactRow>(
1148 + r#"
1149 + SELECT
1150 + u.username,
1151 + u.email,
1152 + COUNT(*) AS total_purchases,
1153 + COALESCE(SUM(t.amount_cents), 0)::BIGINT AS total_spent_cents,
1154 + MAX(t.completed_at) AS last_purchase_at
1155 + FROM transactions t
1156 + JOIN users u ON u.id = t.buyer_id
1157 + WHERE t.seller_id = $1
1158 + AND t.status = 'completed'
1159 + AND t.share_contact = true
1160 + AND NOT EXISTS (
1161 + SELECT 1 FROM contact_revocations cr
1162 + WHERE cr.buyer_id = t.buyer_id AND cr.seller_id = t.seller_id
1163 + )
1164 + GROUP BY t.buyer_id, u.username, u.email
1165 + ORDER BY MAX(t.completed_at) DESC, t.buyer_id
1166 + LIMIT $2 OFFSET $3
1167 + "#,
1168 + )
1169 + .bind(seller_id)
1170 + .bind(limit)
1171 + .bind(offset)
1172 + .fetch_all(pool)
1173 + .await?;
1174 +
1175 + Ok(rows)
1176 + }
1177 +
1135 1178 /// Get seller transactions for CSV export, with conditional buyer email.
1136 1179 ///
1137 1180 /// Respects contact revocations: if a buyer revoked sharing, their email
A server/src/payments/fan_ops.rs +87
@@ -0,0 +1,87 @@
1 + //! Background fan-out of a Stripe subscription operation across all of a
2 + //! creator's fan subscriptions.
3 + //!
4 + //! Pausing, suspending, or terminating a creator requires one Stripe call per
5 + //! fan subscription. Run inline on a request or webhook handler, that serial
6 + //! fan-out ties the hot path up for the duration of N Stripe round-trips —
7 + //! minutes for a large creator, and on the webhook path long enough to trip
8 + //! Stripe's delivery timeout and trigger a retry storm that re-runs the loop
9 + //! (ultra-fuzz Run 6 S1/S2/R6-Perf-M1). This module moves the loop onto the
10 + //! bounded background queue so the handler returns immediately. The operations
11 + //! are fire-and-forget — errors were only logged inline, and still are.
12 +
13 + use std::sync::Arc;
14 +
15 + use crate::background::BackgroundTx;
16 + use crate::payments::PaymentProvider;
17 +
18 + /// Which Stripe subscription operation to apply to each fan subscription.
19 + #[derive(Clone, Copy, Debug)]
20 + pub enum FanSubOp {
21 + /// Set (`true`) or clear (`false`) cancel_at_period_end — creator pause/unpause.
22 + CancelAtPeriodEnd(bool),
23 + /// Pause collection — admin suspend.
24 + Pause,
25 + /// Resume collection — admin unsuspend.
26 + Resume,
27 + /// Cancel outright — admin terminate.
28 + Cancel,
29 + }
30 +
31 + impl FanSubOp {
32 + fn label(self) -> &'static str {
33 + match self {
34 + FanSubOp::CancelAtPeriodEnd(true) => "set_cancel_at_period_end",
35 + FanSubOp::CancelAtPeriodEnd(false) => "clear_cancel_at_period_end",
36 + FanSubOp::Pause => "pause",
37 + FanSubOp::Resume => "resume",
38 + FanSubOp::Cancel => "cancel",
39 + }
40 + }
41 +
42 + async fn apply(
43 + self,
44 + stripe: &Arc<dyn PaymentProvider>,
45 + sub_id: &str,
46 + account_id: &str,
47 + ) -> crate::error::Result<()> {
48 + match self {
49 + FanSubOp::CancelAtPeriodEnd(c) => {
50 + stripe.set_cancel_at_period_end(sub_id, account_id, c).await
51 + }
52 + FanSubOp::Pause => stripe.pause_subscription(sub_id, account_id).await,
53 + FanSubOp::Resume => stripe.resume_subscription(sub_id, account_id).await,
54 + FanSubOp::Cancel => stripe.cancel_subscription(sub_id, account_id).await,
55 + }
56 + }
57 + }
58 +
59 + /// Apply `op` to every subscription in `sub_ids` on the background queue. Returns
60 + /// immediately; the loop runs off the request/webhook hot path. Per-subscription
61 + /// failures are logged and do not abort the rest. No-op for an empty list.
62 + pub fn spawn_fan_sub_fanout(
63 + bg: &BackgroundTx,
64 + stripe: Arc<dyn PaymentProvider>,
65 + account_id: String,
66 + sub_ids: Vec<String>,
67 + op: FanSubOp,
68 + ) {
69 + if sub_ids.is_empty() {
70 + return;
71 + }
72 + bg.spawn("fan-sub stripe fan-out", async move {
73 + let total = sub_ids.len();
74 + let mut failed = 0u32;
75 + for sub_id in &sub_ids {
76 + if let Err(e) = op.apply(&stripe, sub_id, &account_id).await {
77 + failed += 1;
78 + tracing::warn!(stripe_sub_id = %sub_id, op = op.label(), error = ?e, "fan subscription op failed");
79 + }
80 + }
81 + if failed > 0 {
82 + tracing::warn!(total, failed, op = op.label(), "fan subscription fan-out completed with failures");
83 + } else {
84 + tracing::debug!(total, op = op.label(), "fan subscription fan-out completed");
85 + }
86 + });
87 + }
M server/src/payments/mod.rs +1
@@ -17,6 +17,7 @@
17 17 mod checkout;
18 18 mod checkout_metadata;
19 19 mod connect;
20 + pub mod fan_ops;
20 21 pub mod synckit_app_pricing;
21 22 pub mod synckit_billing;
22 23 mod webhooks;
M server/src/routes/admin/users.rs +36 -19
@@ -172,19 +172,25 @@ pub(super) async fn admin_suspend_user(
172 172 // Record moderation action for audit trail
173 173 db::moderation::create_action(&state.db, id, admin.id, ModerationActionType::Suspension, reason, None).await?;
174 174
175 - // Pause fan subscriptions to this creator's projects
175 + // Pause fan subscriptions to this creator's projects. The DB pause runs
176 + // inline (fast, single statement); the per-sub Stripe calls are fanned out
177 + // on the background queue so a creator with many fans doesn't stall the
178 + // admin request.
176 179 if let Some(ref stripe) = state.stripe
177 180 && let Some(ref account_id) = db_user.stripe_account_id
178 181 {
179 182 let subs = db::subscriptions::get_active_subscriptions_by_creator(&state.db, id).await?;
180 - let count = subs.len();
181 - for sub in &subs {
182 - if let Err(e) = stripe.pause_subscription(&sub.stripe_subscription_id, account_id).await {
183 - tracing::error!(stripe_sub_id = %sub.stripe_subscription_id, error = ?e, "failed to pause subscription on Stripe");
184 - }
185 - }
183 + let ids: Vec<String> = subs.into_iter().map(|s| s.stripe_subscription_id).collect();
184 + let queued = ids.len();
185 + crate::payments::fan_ops::spawn_fan_sub_fanout(
186 + &state.bg,
187 + std::sync::Arc::clone(stripe),
188 + account_id.clone(),
189 + ids,
190 + crate::payments::fan_ops::FanSubOp::Pause,
191 + );
186 192 let paused = db::subscriptions::pause_subscriptions_for_creator(&state.db, id).await?;
187 - tracing::info!(user_id = %id, stripe_paused = count, db_paused = paused, "paused fan subscriptions for suspended creator");
193 + tracing::info!(user_id = %id, stripe_queued = queued, db_paused = paused, "paused fan subscriptions for suspended creator");
188 194 }
189 195
190 196 // Send notification email (fire-and-forget)
@@ -222,12 +228,16 @@ pub(super) async fn admin_unsuspend_user(
222 228 && let Some(ref account_id) = db_user.stripe_account_id
223 229 {
224 230 let resumed = db::subscriptions::resume_subscriptions_for_creator(&state.db, id).await?;
225 - for sub in &resumed {
226 - if let Err(e) = stripe.resume_subscription(&sub.stripe_subscription_id, account_id).await {
227 - tracing::error!(stripe_sub_id = %sub.stripe_subscription_id, error = ?e, "failed to resume subscription on Stripe");
228 - }
229 - }
230 - tracing::info!(user_id = %id, resumed = resumed.len(), "resumed fan subscriptions for unsuspended creator");
231 + let ids: Vec<String> = resumed.into_iter().map(|s| s.stripe_subscription_id).collect();
232 + let queued = ids.len();
233 + crate::payments::fan_ops::spawn_fan_sub_fanout(
234 + &state.bg,
235 + std::sync::Arc::clone(stripe),
236 + account_id.clone(),
237 + ids,
238 + crate::payments::fan_ops::FanSubOp::Resume,
239 + );
240 + tracing::info!(user_id = %id, resumed = queued, "resumed fan subscriptions for unsuspended creator");
231 241 }
232 242
233 243 tracing::info!(user_id = %id, "admin unsuspended user");
@@ -277,11 +287,18 @@ pub(super) async fn admin_terminate_user(
277 287 {
278 288 let active_subs = db::subscriptions::get_active_subscriptions_by_creator(&state.db, id).await?;
279 289 let paused_subs = db::subscriptions::get_paused_subscriptions_by_creator(&state.db, id).await?;
280 - for sub in active_subs.iter().chain(paused_subs.iter()) {
281 - if let Err(e) = stripe.cancel_subscription(&sub.stripe_subscription_id, account_id).await {
282 - tracing::error!(stripe_sub_id = %sub.stripe_subscription_id, error = ?e, "failed to cancel subscription on termination");
283 - }
284 - }
290 + let ids: Vec<String> = active_subs
291 + .into_iter()
292 + .chain(paused_subs)
293 + .map(|s| s.stripe_subscription_id)
294 + .collect();
295 + crate::payments::fan_ops::spawn_fan_sub_fanout(
296 + &state.bg,
297 + std::sync::Arc::clone(stripe),
298 + account_id.clone(),
299 + ids,
300 + crate::payments::fan_ops::FanSubOp::Cancel,
301 + );
285 302 }
286 303
287 304 // Send termination email
M server/src/routes/api/exports/mod.rs +25 -26
@@ -683,32 +683,31 @@ pub(super) async fn export_contacts(
683 683 AuthUser(user): AuthUser,
684 684 ) -> Result<Response> {
685 685 let is_htmx = is_htmx_request(&headers);
686 - let contacts = db::transactions::get_seller_contacts(&state.db, user.id).await?;
687 -
688 - let mut csv_content = String::from("Username,Email,Purchases,Total Spent,Last Purchase\n");
689 - for c in &contacts {
690 - csv_content.push_str(&format!(
691 - "{},{},{},{},{}\n",
692 - sanitize_csv_cell(&c.username),
693 - sanitize_csv_cell(&c.email),
694 - c.total_purchases,
695 - crate::formatting::format_dollars_plain(c.total_spent_cents),
696 - c.last_purchase_at.format("%Y-%m-%d"),
697 - ));
698 - }
699 -
700 - if is_htmx {
701 - let data_uri = format!(
702 - "data:text/csv;charset=utf-8,{}",
703 - urlencoding::encode(&csv_content)
704 - );
705 - return Ok(ExportDownloadTemplate {
706 - data_uri,
707 - filename: "makenot-work-contacts.csv".to_string(),
708 - }.into_response());
709 - }
710 -
711 - download_response(csv_content.into_bytes(), "makenot-work-contacts.csv", "text/csv")
686 + let pool = state.db.clone();
687 + let uid = user.id;
688 + let rx = spawn_paginated_csv(
689 + "Username,Email,Purchases,Total Spent,Last Purchase\n",
690 + move |limit, offset| {
691 + let pool = pool.clone();
692 + async move {
693 + let contacts =
694 + db::transactions::get_seller_contacts_page(&pool, uid, limit, offset).await?;
695 + let mut buf = String::new();
696 + for c in &contacts {
697 + buf.push_str(&format!(
698 + "{},{},{},{},{}\n",
699 + sanitize_csv_cell(&c.username),
700 + sanitize_csv_cell(&c.email),
701 + c.total_purchases,
702 + crate::formatting::format_dollars_plain(c.total_spent_cents),
703 + c.last_purchase_at.format("%Y-%m-%d"),
704 + ));
705 + }
706 + Ok((buf, contacts.len()))
707 + }
708 + },
709 + );
710 + finish_csv(is_htmx, "makenot-work-contacts.csv", rx).await
712 711 }
713 712
714 713 #[cfg(test)]
M server/src/routes/api/internal/creators.rs +4 -4
@@ -161,8 +161,8 @@ pub(super) async fn creator_stats(
161 161 let comparison =
162 162 db::analytics::get_period_comparison(&state.db, query.user_id, None, None, &range).await?;
163 163
164 - let projects = db::projects::get_projects_by_user(&state.db, query.user_id).await?;
165 - let items = db::items::get_items_by_user(&state.db, query.user_id).await?;
164 + let total_projects = db::projects::count_projects_by_user(&state.db, query.user_id).await?;
165 + let total_items = db::items::count_items_by_user(&state.db, query.user_id).await?;
166 166
167 167 Ok(Json(CreatorStats {
168 168 current_revenue_cents: comparison.current_revenue_cents.as_i64(),
@@ -171,8 +171,8 @@ pub(super) async fn creator_stats(
171 171 previous_sales: comparison.previous_sales,
172 172 current_followers: comparison.current_followers,
173 173 previous_followers: comparison.previous_followers,
174 - total_projects: projects.len() as i64,
175 - total_items: items.len() as i64,
174 + total_projects,
175 + total_items,
176 176 }))
177 177 }
178 178
M server/src/routes/api/users/profile.rs +11 -14
@@ -314,22 +314,19 @@ pub(in crate::routes::api) async fn pause_creator(
314 314 tracing::warn!(error = ?e, "failed to cancel creator tier subscription on Stripe during pause");
315 315 }
316 316
317 - // Set cancel_at_period_end on all active fan subscriptions (connected account)
317 + // Set cancel_at_period_end on all active fan subscriptions (connected
318 + // account). Fanned out on the background queue so a creator with many
319 + // fans doesn't tie the request up for N serial Stripe round-trips.
318 320 if let Some(ref stripe_account_id) = db_user.stripe_account_id {
319 321 let fan_subs = db::subscriptions::get_active_subscriptions_by_creator(&state.db, user.id).await?;
320 - for sub in &fan_subs {
321 - if let Err(e) = stripe.set_cancel_at_period_end(
322 - &sub.stripe_subscription_id,
323 - stripe_account_id,
324 - true,
325 - ).await {
326 - tracing::warn!(
327 - stripe_sub_id = %sub.stripe_subscription_id,
328 - error = ?e,
329 - "failed to set cancel_at_period_end on fan subscription during pause"
330 - );
331 - }
332 - }
322 + let ids = fan_subs.into_iter().map(|s| s.stripe_subscription_id).collect();
323 + crate::payments::fan_ops::spawn_fan_sub_fanout(
324 + &state.bg,
325 + std::sync::Arc::clone(stripe),
326 + stripe_account_id.clone(),
327 + ids,
328 + crate::payments::fan_ops::FanSubOp::CancelAtPeriodEnd(true),
329 + );
333 330 }
334 331 }
335 332
M server/src/routes/api/validate.rs +20 -16
@@ -33,23 +33,27 @@ async fn suggest_slugs(
33 33 })
34 34 .unwrap_or(base);
35 35
36 - let mut suggestions = Vec::new();
37 - for n in 2..=10 {
38 - let candidate = format!("{}-{}", root, n);
39 - if let Ok(slug) = Slug::new(&candidate) {
40 - let taken = db::projects::get_project_by_user_and_slug(db, user_id, &slug)
41 - .await
42 - .map(|p| p.is_some())
43 - .unwrap_or(true);
44 - if !taken {
45 - suggestions.push(candidate);
46 - if suggestions.len() >= 3 {
47 - break;
48 - }
49 - }
50 - }
36 + // Build all valid candidate slugs in order, then resolve availability with a
37 + // single `slug = ANY(...)` query instead of one point-query per candidate.
38 + let candidates: Vec<String> = (2..=10)
39 + .map(|n| format!("{}-{}", root, n))
40 + .filter(|c| Slug::new(c).is_ok())
41 + .collect();
42 + if candidates.is_empty() {
43 + return Vec::new();
51 44 }
52 - suggestions
45 +
46 + // On a DB error, suggest nothing rather than risk offering a taken slug.
47 + let Ok(taken) = db::projects::filter_taken_slugs(db, user_id, &candidates).await else {
48 + return Vec::new();
49 + };
50 + let taken: std::collections::HashSet<String> = taken.into_iter().collect();
51 +
52 + candidates
53 + .into_iter()
54 + .filter(|c| !taken.contains(c))
55 + .take(3)
56 + .collect()
53 57 }
54 58
55 59 #[derive(Debug, Deserialize)]
M server/src/routes/git_issues/push_refs.rs +33 -45
@@ -199,70 +199,58 @@ pub(super) async fn process_push(
199 199 numbers.dedup();
200 200 let mut issues = db::issues::get_issues_by_numbers(&state.db, repo.id, &numbers).await?;
201 201
202 - // Now process DB operations (async-safe, git2 types are dropped)
202 + // Collect the DB writes and flush them in two bulk statements at the end
203 + // rather than one status UPDATE + one comment INSERT per referenced commit
204 + // (ultra-fuzz Run 6 S6). The local `issue.status` is still mutated as we go,
205 + // so repeated references to the same issue within one push observe the
206 + // running state; `status_changes` holds the final status to persist.
203 207 let mut processed = 0u32;
208 + let mut status_changes: HashMap<db::IssueId, IssueStatus> = HashMap::new();
209 + let mut comments: Vec<(db::IssueId, String, String)> = Vec::new();
210 +
204 211 for (oid_str, short_oid, refs) in &commit_refs {
205 212 for issue_ref in refs {
206 213 let Some(issue) = issues.get_mut(&issue_ref.number) else {
207 214 continue;
208 215 };
209 216
210 - match issue_ref.action {
217 + let lead = match issue_ref.action {
211 218 IssueRefAction::Close => {
212 219 if issue.status == IssueStatus::Open {
213 - if let Err(e) = db::issues::update_issue_status(&state.db, issue.id, IssueStatus::Closed).await {
214 - tracing::warn!(issue_id = %issue.id, error = ?e, "failed to close issue via push");
215 - } else {
216 - issue.status = IssueStatus::Closed;
217 - }
218 - }
219 - let body_md = format!(
220 - "Closed via commit [`{}`](/git/{}/{}/commit/{}) on `{}`.",
221 - short_oid, req.repo_owner, req.repo_name, oid_str, req.ref_name,
222 - );
223 - let body_html = docengine::render_permissive(&body_md);
224 - if let Err(e) = db::issues::create_comment(
225 - &state.db, issue.id, owner_user.id, &body_md, &body_html,
226 - ).await {
227 - tracing::warn!(issue_id = %issue.id, error = ?e, "failed to create close comment via push");
220 + issue.status = IssueStatus::Closed;
221 + status_changes.insert(issue.id, IssueStatus::Closed);
228 222 }
223 + "Closed via"
229 224 }
230 225 IssueRefAction::Reopen => {
231 226 if issue.status == IssueStatus::Closed {
232 - if let Err(e) = db::issues::update_issue_status(&state.db, issue.id, IssueStatus::Open).await {
233 - tracing::warn!(issue_id = %issue.id, error = ?e, "failed to reopen issue via push");
234 - } else {
235 - issue.status = IssueStatus::Open;
236 - }
237 - }
238 - let body_md = format!(
239 - "Reopened via commit [`{}`](/git/{}/{}/commit/{}) on `{}`.",
240 - short_oid, req.repo_owner, req.repo_name, oid_str, req.ref_name,
241 - );
242 - let body_html = docengine::render_permissive(&body_md);
243 - if let Err(e) = db::issues::create_comment(
244 - &state.db, issue.id, owner_user.id, &body_md, &body_html,
245 - ).await {
246 - tracing::warn!(issue_id = %issue.id, error = ?e, "failed to create reopen comment via push");
227 + issue.status = IssueStatus::Open;
228 + status_changes.insert(issue.id, IssueStatus::Open);
247 229 }
230 + "Reopened via"
248 231 }
249 - IssueRefAction::Reference => {
250 - let body_md = format!(
251 - "Referenced in commit [`{}`](/git/{}/{}/commit/{}) on `{}`.",
252 - short_oid, req.repo_owner, req.repo_name, oid_str, req.ref_name,
253 - );
254 - let body_html = docengine::render_permissive(&body_md);
255 - if let Err(e) = db::issues::create_comment(
256 - &state.db, issue.id, owner_user.id, &body_md, &body_html,
257 - ).await {
258 - tracing::warn!(issue_id = %issue.id, error = ?e, "failed to create reference comment via push");
259 - }
260 - }
261 - }
232 + IssueRefAction::Reference => "Referenced in",
233 + };
234 +
235 + let body_md = format!(
236 + "{lead} commit [`{}`](/git/{}/{}/commit/{}) on `{}`.",
237 + short_oid, req.repo_owner, req.repo_name, oid_str, req.ref_name,
238 + );
239 + let body_html = docengine::render_permissive(&body_md);
240 + comments.push((issue.id, body_md, body_html));
262 241 processed += 1;
263 242 }
264 243 }
265 244
245 + // Flush: final status per changed issue, then every comment, in two queries.
246 + let status_updates: Vec<(db::IssueId, IssueStatus)> = status_changes.into_iter().collect();
247 + if let Err(e) = db::issues::update_issue_statuses(&state.db, &status_updates).await {
248 + tracing::warn!(error = ?e, "failed to bulk-update issue statuses via push");
249 + }
250 + if let Err(e) = db::issues::create_comments(&state.db, owner_user.id, &comments).await {
251 + tracing::warn!(error = ?e, "failed to bulk-create issue comments via push");
252 + }
253 +
266 254 Ok((StatusCode::OK, Json(serde_json::json!({ "processed": processed }))))
267 255 }
268 256
M server/src/routes/stripe/webhook/checkout.rs +12 -14
@@ -533,24 +533,22 @@ pub(super) async fn handle_creator_tier_checkout_completed(
533 533 .await
534 534 .with_context(|| format!("unpause creator {user_id}"))?;
535 535
536 - // Un-cancel active fan subscriptions (clear cancel_at_period_end)
536 + // Un-cancel active fan subscriptions (clear cancel_at_period_end). Fanned
537 + // out on the background queue: doing it inline here let a creator with
538 + // many fans stall the webhook past Stripe's delivery timeout, which
539 + // triggers a retry that re-runs the whole loop.
537 540 if let (Some(stripe), Some(stripe_account_id)) = (&state.stripe, &db_user.stripe_account_id) {
538 541 let fan_subs = db::subscriptions::get_active_subscriptions_by_creator(&state.db, user_id)
539 542 .await
540 543 .with_context(|| format!("fetch active fan subs for unpause {user_id}"))?;
541 - for fan_sub in &fan_subs {
542 - if let Err(e) = stripe.set_cancel_at_period_end(
543 - &fan_sub.stripe_subscription_id,
544 - stripe_account_id,
545 - false,
546 - ).await {
547 - tracing::warn!(
548 - stripe_sub_id = %fan_sub.stripe_subscription_id,
549 - error = ?e,
550 - "failed to clear cancel_at_period_end on fan sub during unpause"
551 - );
552 - }
553 - }
544 + let ids = fan_subs.into_iter().map(|s| s.stripe_subscription_id).collect();
545 + crate::payments::fan_ops::spawn_fan_sub_fanout(
546 + &state.bg,
547 + std::sync::Arc::clone(stripe),
548 + stripe_account_id.clone(),
549 + ids,
550 + crate::payments::fan_ops::FanSubOp::CancelAtPeriodEnd(false),
551 + );
554 552 }
555 553
556 554 tracing::info!(user_id = %user_id, "creator auto-unpaused after re-subscribing to tier");
M server/src/scanning/archive.rs +24 -3
@@ -106,6 +106,19 @@ enum ContentBuf {
106 106 /// `Err(detail)` is a mid-stream decode failure: the caller decides what that
107 107 /// means for each dimension (ZIP bomb accounting uses a conservative estimate
108 108 /// and continues; a single-stream compressor fails closed).
109 + /// Byte limit at which `tee_decompress` should stop expanding a stream for bomb
110 + /// accounting: the smaller of the 2 GB absolute cap and `compressed * MAX_RATIO`.
111 + /// A stream that blows past its ratio budget is flagged as a bomb regardless, so
112 + /// there is no reason to keep decompressing it up to the full 2 GB first. Falls
113 + /// back to the absolute cap when the compressed size is unknown (0).
114 + fn entry_bomb_stop_limit(compressed_size: u64) -> u64 {
115 + if compressed_size == 0 {
116 + return constants::SCAN_ZIP_MAX_UNCOMPRESSED;
117 + }
118 + let ratio_limit = compressed_size.saturating_mul(constants::SCAN_ZIP_MAX_RATIO as u64);
119 + constants::SCAN_ZIP_MAX_UNCOMPRESSED.min(ratio_limit)
120 + }
121 +
109 122 fn tee_decompress(
110 123 reader: &mut dyn Read,
111 124 bomb_abs_limit: u64,
@@ -308,12 +321,16 @@ fn walk_zip<R: std::io::Read + std::io::Seek>(
308 321 }
309 322
310 323 // Decompress the entry exactly once, teeing the full size (bomb) and the
311 - // buffered prefix (content).
324 + // buffered prefix (content). Stop early once the entry blows past
325 + // `compressed * MAX_RATIO` rather than always decompressing to the 2 GB
326 + // absolute cap — a high-ratio entry is flagged below regardless, so there
327 + // is no reason to expand it fully first.
328 + let entry_stop_limit = entry_bomb_stop_limit(entry_compressed);
312 329 let want_content = content && nested_res.is_none();
313 330 let (counted, content_buf, decode_err) = match archive.by_index(i) {
314 331 Ok(mut entry) => match tee_decompress(
315 332 &mut entry,
316 - constants::SCAN_ZIP_MAX_UNCOMPRESSED,
333 + entry_stop_limit,
317 334 want_content,
318 335 &mut interior_budget,
319 336 ) {
@@ -496,8 +513,12 @@ fn walk_compressed<R: Read>(
496 513 let ratio_limit = compressed_size.saturating_mul(constants::SCAN_ZIP_MAX_RATIO as u64);
497 514 let mut interior_budget: u64 = constants::SCAN_ZIP_MAX_UNCOMPRESSED;
498 515
516 + // Stop expanding once the stream blows past `compressed * MAX_RATIO` rather
517 + // than always running to the 2 GB absolute cap; the ratio verdict below fires
518 + // identically on the early-stopped count.
519 + let stop_limit = entry_bomb_stop_limit(compressed_size);
499 520 let (counted, content_buf, decode_err) =
500 - match tee_decompress(decoder.as_mut(), abs_limit, content, &mut interior_budget) {
521 + match tee_decompress(decoder.as_mut(), stop_limit, content, &mut interior_budget) {
501 522 Ok((c, cb)) => (c, cb, None),
502 523 Err((c, why)) => (c, None, Some(why)),
503 524 };
M server/src/scanning/mod.rs +17 -1
@@ -547,7 +547,23 @@ impl ScanPipeline {
547 547 layers.push(archive_layer);
548 548 layers.push(archive_nested_layer);
549 549 layers.push(match self.yara_rules {
550 - Some(ref rules) => yara::scan_with_yara(rules, data),
550 + Some(ref rules) => {
551 + // Cap YARA's input: it walks the whole slice, faulting the entire
552 + // mmap resident. ClamAV (streamed, uncapped) is the full-file
553 + // backstop, so scanning a generous prefix bounds peak RAM without
554 + // surrendering the floor.
555 + let yara_input = if data.len() > crate::constants::SCAN_YARA_MAX_BYTES {
556 + tracing::warn!(
557 + full_bytes = data.len(),
558 + capped_bytes = crate::constants::SCAN_YARA_MAX_BYTES,
559 + "YARA input capped; scanning prefix only (ClamAV scans the full file)"
560 + );
561 + &data[..crate::constants::SCAN_YARA_MAX_BYTES]
562 + } else {
563 + data
564 + };
565 + yara::scan_with_yara(rules, yara_input)
566 + }
551 567 None => LayerResult {
552 568 layer: "yara",
553 569 verdict: LayerVerdict::Skip,
M server/src/scanning/spool.rs +11 -3
@@ -11,7 +11,7 @@
11 11
12 12 use crate::constants::{
13 13 SCAN_SPOOL_FREE_RESERVE_BYTES, SCAN_SPOOL_MAX_BYTES, SCAN_SPOOL_ORPHAN_AGE_SECS,
14 - SCAN_WORKER_COUNT,
14 + SCAN_SPOOL_SLACK_BYTES, SCAN_WORKER_COUNT,
15 15 };
16 16 use s3_storage::ByteStream;
17 17 use std::path::{Path, PathBuf};
@@ -148,6 +148,14 @@ pub async fn download_into_tempfile(
148 148 // — funnels through a single cleanup that unlinks the partial file. The
149 149 // earlier code only removed it on the over-cap branch, leaking a partial
150 150 // tempfile on a write/flush/stream error until the orphan reaper swept it.
151 + // Bound the streamed bytes by the *claimed* object size plus a small slack,
152 + // not just the global ceiling: an object that under-reports its size must not
153 + // be allowed to stream the full `SCAN_SPOOL_MAX_BYTES` to scratch before we
154 + // abort (Run 6 R6-Perf-M3). `check_free_space`/worker already cap
155 + // `expected_size` at the global max, so this is always the tighter bound.
156 + let write_cap = expected_size
157 + .saturating_add(SCAN_SPOOL_SLACK_BYTES)
158 + .min(SCAN_SPOOL_MAX_BYTES);
151 159 let write_result: Result<(), String> = async {
152 160 let mut written: u64 = 0;
153 161 while let Some(chunk) = stream
@@ -156,9 +164,9 @@ pub async fn download_into_tempfile(
156 164 .map_err(|e| format!("read S3 stream: {e}"))?
157 165 {
158 166 written += chunk.len() as u64;
159 - if written > SCAN_SPOOL_MAX_BYTES {
167 + if written > write_cap {
160 168 return Err(format!(
161 - "S3 stream exceeded SCAN_SPOOL_MAX_BYTES ({SCAN_SPOOL_MAX_BYTES} bytes); aborting scan spool"
169 + "S3 stream exceeded expected size bound ({write_cap} bytes, claimed {expected_size}); aborting scan spool"
162 170 ));
163 171 }
164 172 file.write_all(&chunk)