Skip to main content

max / makenotwork

perf: daily-gate grace hiding, concurrent appeal resumes, batch push-ref issue lookups - enforce_post_grace_hiding moved from every 60s tick into the daily block; it is self-draining (mark_grace_enforced) so per-tick it only re-ran an empty query on a 30-day grace window. - appeal approval resumes paused subscriptions on Stripe with a bounded JoinSet (8-way) instead of one serial network round-trip per sub. - git push-refs batch-fetches referenced issues in one query (number = ANY) and mutates a local map, replacing the per-commit-reference get_issue_by_number N+1. - add tracing spans to the mt/wam outbound clients (were uninstrumented). ultra-fuzz Run 5 M-Perf1 + perf cold spots (deep). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-24 02:41 UTC
Commit: ed25e447654cd77cdb3c5df26404f6232af78f1a
Parent: ea9eb66
6 files changed, +75 insertions, -16 deletions
M server/src/db/issues.rs +22
@@ -85,6 +85,28 @@ pub async fn get_issue_by_number(
85 85 Ok(issue)
86 86 }
87 87
88 + /// Fetch multiple issues by number in a single query, keyed by number. Used by
89 + /// the git push-refs handler so a push touching many commits doesn't run one
90 + /// `get_issue_by_number` per commit-reference (N+1).
91 + #[tracing::instrument(skip_all)]
92 + pub async fn get_issues_by_numbers(
93 + pool: &PgPool,
94 + repo_id: GitRepoId,
95 + numbers: &[i32],
96 + ) -> Result<std::collections::HashMap<i32, DbIssue>> {
97 + if numbers.is_empty() {
98 + return Ok(std::collections::HashMap::new());
99 + }
100 + let issues = sqlx::query_as::<_, DbIssue>(
101 + "SELECT * FROM issues WHERE repo_id = $1 AND number = ANY($2)",
102 + )
103 + .bind(repo_id)
104 + .bind(numbers)
105 + .fetch_all(pool)
106 + .await?;
107 + Ok(issues.into_iter().map(|i| (i.number, i)).collect())
108 + }
109 +
88 110 /// List issues with author username and comment count. Returns (issues, total_count).
89 111 #[tracing::instrument(skip_all)]
90 112 pub async fn list_issues(
M server/src/mt_client.rs +4
@@ -167,6 +167,7 @@ impl MtClient {
167 167 }
168 168
169 169 /// Create or retrieve an existing community on MT.
170 + #[tracing::instrument(skip_all)]
170 171 pub async fn create_community(
171 172 &self,
172 173 req: &CreateCommunityRequest,
@@ -175,6 +176,7 @@ impl MtClient {
175 176 }
176 177
177 178 /// Create a discussion thread on MT linked to MNW content.
179 + #[tracing::instrument(skip_all)]
178 180 pub async fn create_thread(
179 181 &self,
180 182 req: &CreateThreadRequest,
@@ -183,6 +185,7 @@ impl MtClient {
183 185 }
184 186
185 187 /// Add a reply to an existing thread on MT.
188 + #[tracing::instrument(skip_all, fields(thread_id = %thread_id))]
186 189 pub async fn create_post(
187 190 &self,
188 191 thread_id: MtThreadId,
@@ -193,6 +196,7 @@ impl MtClient {
193 196 }
194 197
195 198 /// Get thread stats (post count + last activity).
199 + #[tracing::instrument(skip_all, fields(thread_id = %thread_id))]
196 200 pub async fn get_thread_stats(
197 201 &self,
198 202 thread_id: MtThreadId,
M server/src/routes/admin/moderation.rs +16 -2
@@ -71,11 +71,25 @@ pub(super) async fn admin_decide_appeal(
71 71 && let Some(ref account_id) = db_user.stripe_account_id
72 72 {
73 73 let resumed = db::subscriptions::resume_subscriptions_for_creator(&state.db, user_id).await?;
74 + // Resume each paused subscription on Stripe with bounded concurrency, so
75 + // a creator with many fans doesn't serialize one network round-trip per
76 + // subscription (the appeal handler would otherwise block for N x RTT).
77 + const RESUME_PARALLELISM: usize = 8;
78 + let mut set = tokio::task::JoinSet::new();
74 79 for sub in &resumed {
75 - if let Err(e) = stripe.resume_subscription(&sub.stripe_subscription_id, account_id).await {
76 - tracing::error!(stripe_sub_id = %sub.stripe_subscription_id, error = ?e, "failed to resume subscription on appeal approval");
80 + if set.len() >= RESUME_PARALLELISM {
81 + let _ = set.join_next().await;
77 82 }
83 + let stripe = stripe.clone();
84 + let account_id = account_id.clone();
85 + let sub_id = sub.stripe_subscription_id.clone();
86 + set.spawn(async move {
87 + if let Err(e) = stripe.resume_subscription(&sub_id, &account_id).await {
88 + tracing::error!(stripe_sub_id = %sub_id, error = ?e, "failed to resume subscription on appeal approval");
89 + }
90 + });
78 91 }
92 + while set.join_next().await.is_some() {}
79 93 if !resumed.is_empty() {
80 94 tracing::info!(user_id = %user_id, resumed = resumed.len(), "resumed fan subscriptions on appeal approval");
81 95 }
M server/src/routes/git_issues/push_refs.rs +26 -11
@@ -187,21 +187,34 @@ pub(super) async fn process_push(
187 187 .await
188 188 .context("git push-refs walk task")??;
189 189
190 + // Batch-fetch every referenced issue once (deduped across commits) so a push
191 + // touching many commits doesn't run one get_issue_by_number per reference.
192 + // The local map is mutated as statuses change, so repeated references to the
193 + // same issue see its current state without re-querying.
194 + let mut numbers: Vec<i32> = commit_refs
195 + .iter()
196 + .flat_map(|(_, _, refs)| refs.iter().map(|r| r.number))
197 + .collect();
198 + numbers.sort_unstable();
199 + numbers.dedup();
200 + let mut issues = db::issues::get_issues_by_numbers(&state.db, repo.id, &numbers).await?;
201 +
190 202 // Now process DB operations (async-safe, git2 types are dropped)
191 203 let mut processed = 0u32;
192 204 for (oid_str, short_oid, refs) in &commit_refs {
193 205 for issue_ref in refs {
194 - let issue = match db::issues::get_issue_by_number(&state.db, repo.id, issue_ref.number).await {
195 - Ok(Some(i)) => i,
196 - _ => continue,
206 + let Some(issue) = issues.get_mut(&issue_ref.number) else {
207 + continue;
197 208 };
198 209
199 210 match issue_ref.action {
200 211 IssueRefAction::Close => {
201 - if issue.status == IssueStatus::Open
202 - && let Err(e) = db::issues::update_issue_status(&state.db, issue.id, IssueStatus::Closed).await
203 - {
204 - tracing::warn!(issue_id = %issue.id, error = ?e, "failed to close issue via push");
212 + 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 + }
205 218 }
206 219 let body_md = format!(
207 220 "Closed via commit [`{}`](/git/{}/{}/commit/{}) on `{}`.",
@@ -215,10 +228,12 @@ pub(super) async fn process_push(
215 228 }
216 229 }
217 230 IssueRefAction::Reopen => {
218 - if issue.status == IssueStatus::Closed
219 - && let Err(e) = db::issues::update_issue_status(&state.db, issue.id, IssueStatus::Open).await
220 - {
221 - tracing::warn!(issue_id = %issue.id, error = ?e, "failed to reopen issue via push");
231 + 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 + }
222 237 }
223 238 let body_md = format!(
224 239 "Reopened via commit [`{}`](/git/{}/{}/commit/{}) on `{}`.",
M server/src/scheduler/mod.rs +6 -3
@@ -199,9 +199,6 @@ pub fn spawn_scheduler(
199 199 }
200 200 }
201 201
202 - // Enforce post-grace item hiding (canceled 30+ days ago)
203 - integrity::enforce_post_grace_hiding(&state).await;
204 -
205 202 // Clean up expired idempotency keys (every tick is fine — cheap DELETE)
206 203 if let Err(e) = db::idempotency::cleanup_expired(&state.db).await {
207 204 tracing::error!(error = ?e, "failed to clean up expired idempotency keys");
@@ -263,6 +260,12 @@ pub fn spawn_scheduler(
263 260 integrity::check_stale_subscriptions(&state).await;
264 261 integrity::check_email_bounce_spike(&state).await;
265 262
263 + // Enforce post-grace item hiding (canceled 30+ days ago). Daily
264 + // is ample for a 30-day grace window, and it's self-draining
265 + // (mark_grace_enforced), so running it every 60s tick only
266 + // re-issued an empty query — moved here off the hot tick path.
267 + integrity::enforce_post_grace_hiding(&state).await;
268 +
266 269 // Prune session records inactive for 90+ days
267 270 let session_threshold = chrono::Utc::now() - chrono::Duration::days(90);
268 271 match db::sessions::prune_expired_sessions(&state.db, session_threshold).await {
M server/src/wam_client.rs +1
@@ -41,6 +41,7 @@ impl WamClient {
41 41
42 42 /// Create a ticket in WAM. Errors are logged but never propagated — WAM
43 43 /// is a best-effort notification channel, not a critical path.
44 + #[tracing::instrument(skip_all, fields(source = %source, priority = %priority))]
44 45 pub async fn create_ticket(
45 46 &self,
46 47 title: &str,