Skip to main content

max / makenotwork

DB concurrency: project-split lock, issue retry/search, webhook claim /audit Run 13 DB-concurrency cold spots: - project_members: serialize revenue-split mutations on the parent projects row (FOR UPDATE) instead of the member row-set, which does not block a concurrent INSERT of the first members and let the total race past 100%. Also drop the invalid `SUM(...) FOR UPDATE` in update_member_split (Postgres rejects FOR UPDATE with an aggregate — it 500'd on every call); shared lock_project_for_splits helper. - issues: retry create_issue only on a unique violation (was any Err, so FK/timeout errors were pointlessly retried); make list_issues search match body_markdown, not just title (list + count). - webhook_events: atomically CLAIM due retry events (FOR UPDATE SKIP LOCKED + push next_retry_at out) so a second scheduler replica can't double-run the same event; handlers are idempotent so at-least-once is safe. Also fix the Postmark inbound integration tests to carry an aligned SPF/DKIM verdict (normal production inbound) now that the server requires sender authentication, and add spoofed/missing-auth rejection tests. Integration: git_issues + patches + revenue_splits + stripe_webhooks all green (90 tests). clippy --tests clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-01 21:26 UTC
Commit: c57008daec06a80107a8597f25c242c35f01c7f6
Parent: a6fd76d
5 files changed, +147 insertions, -32 deletions
M server/src/db/issues.rs +8 -4
@@ -24,10 +24,14 @@ pub async fn create_issue(
24 24 let result = try_create_issue(pool, repo_id, author_id, title, body_md, body_html).await;
25 25 match result {
26 26 Ok(issue) => Ok(issue),
27 - Err(_) => {
28 - // Retry once on conflict (concurrent number assignment)
27 + // Retry once ONLY on a unique violation: concurrent `MAX(number)+1`
28 + // assignment can collide on the `(repo_id, number)` index. Any other
29 + // error (FK violation, timeout, pool exhaustion) is not helped by an
30 + // immediate retry and should surface (audit Run 13 Conc).
31 + Err(e) if crate::helpers::is_unique_violation(&e) => {
29 32 try_create_issue(pool, repo_id, author_id, title, body_md, body_html).await
30 33 }
34 + Err(e) => Err(e),
31 35 }
32 36 }
33 37
@@ -134,7 +138,7 @@ pub async fn list_issues(
134 138 JOIN users u ON u.id = i.author_user_id
135 139 WHERE i.repo_id = $1
136 140 AND ($2::TEXT IS NULL OR i.status = $2)
137 - AND ($3::TEXT IS NULL OR i.title ILIKE $3)
141 + AND ($3::TEXT IS NULL OR i.title ILIKE $3 OR i.body_markdown ILIKE $3)
138 142 ORDER BY i.created_at DESC
139 143 LIMIT $4 OFFSET $5
140 144 "#,
@@ -153,7 +157,7 @@ pub async fn list_issues(
153 157 FROM issues
154 158 WHERE repo_id = $1
155 159 AND ($2::TEXT IS NULL OR status = $2)
156 - AND ($3::TEXT IS NULL OR title ILIKE $3)
160 + AND ($3::TEXT IS NULL OR title ILIKE $3 OR body_markdown ILIKE $3)
157 161 "#,
158 162 )
159 163 .bind(repo_id)
M server/src/db/project_members.rs +31 -19
@@ -9,6 +9,21 @@ use crate::error::{AppError, Result};
9 9
10 10 // ── Project Members ──
11 11
12 + /// Take a row lock on the parent `projects` row so every revenue-split mutation
13 + /// for the project serializes — including the first two members added
14 + /// concurrently, which member-row locks alone would not block (audit Run 13
15 + /// Conc TOCTOU). Callers must already be inside a transaction.
16 + async fn lock_project_for_splits(
17 + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
18 + project_id: ProjectId,
19 + ) -> Result<()> {
20 + sqlx::query("SELECT 1 FROM projects WHERE id = $1 FOR UPDATE")
21 + .bind(project_id)
22 + .execute(&mut **tx)
23 + .await?;
24 + Ok(())
25 + }
26 +
12 27 /// Add a member to a project with a revenue split percentage (0-100).
13 28 ///
14 29 /// The sum of all member splits (excluding the owner's implicit remainder)
@@ -22,11 +37,6 @@ pub async fn add_project_member(
22 37 split_percent: i16,
23 38 added_by: UserId,
24 39 ) -> Result<DbProjectMember> {
25 - let mut tx = pool.begin().await?;
26 -
27 - // Lock all existing members to prevent concurrent split modifications.
28 - // FOR UPDATE cannot be combined with aggregate functions, so we lock
29 - // the rows first, then compute the sum.
30 40 // Reject negative and out-of-range splits before any DB write. The Run #6
31 41 // audit caught the missing lower bound — negative percentages flow through
32 42 // `compute_splits` and record negative obligations.
@@ -36,12 +46,14 @@ pub async fn add_project_member(
36 46 )));
37 47 }
38 48
39 - let _locked: Vec<(i16,)> = sqlx::query_as(
40 - "SELECT split_percent FROM project_members WHERE project_id = $1 FOR UPDATE",
41 - )
42 - .bind(project_id)
43 - .fetch_all(&mut *tx)
44 - .await?;
49 + let mut tx = pool.begin().await?;
50 +
51 + // Serialize every split mutation for this project on the parent row. Locking
52 + // only the member rows does NOT block a concurrent INSERT of a new member
53 + // when none exist yet, so two "first members" could each pass the cap check
54 + // and race the total past 100% (audit Run 13 Conc TOCTOU). The projects row
55 + // always exists, so `FOR UPDATE` on it serializes all adds/updates.
56 + lock_project_for_splits(&mut tx, project_id).await?;
45 57
46 58 // Subtract the existing row's split (if this is an upsert) before the cap
47 59 // check; otherwise a legitimate update is rejected as "> 100%" whenever
@@ -164,9 +176,14 @@ pub async fn update_member_split(
164 176 }
165 177 let mut tx = pool.begin().await?;
166 178
167 - // Lock all members for this project to prevent concurrent split modifications
179 + // Serialize on the parent project row (see `add_project_member`). The prior
180 + // `SUM(...) FOR UPDATE` below was invalid — Postgres rejects FOR UPDATE with
181 + // an aggregate, so this function 500'd on every call (audit Run 13); the lock
182 + // now lives on the projects row and the SUM runs plain.
183 + lock_project_for_splits(&mut tx, project_id).await?;
184 +
168 185 let current: Option<DbProjectMember> = sqlx::query_as(
169 - "SELECT * FROM project_members WHERE project_id = $1 AND user_id = $2 FOR UPDATE",
186 + "SELECT * FROM project_members WHERE project_id = $1 AND user_id = $2",
170 187 )
171 188 .bind(project_id)
172 189 .bind(user_id)
@@ -176,12 +193,7 @@ pub async fn update_member_split(
176 193 let current_member_split = current.map(|m| m.split_percent as i64).unwrap_or(0);
177 194
178 195 let total_row: (Option<i64>,) = sqlx::query_as(
179 - r#"
180 - SELECT SUM(split_percent)::BIGINT
181 - FROM project_members
182 - WHERE project_id = $1
183 - FOR UPDATE
184 - "#,
196 + "SELECT SUM(split_percent)::BIGINT FROM project_members WHERE project_id = $1",
185 197 )
186 198 .bind(project_id)
187 199 .fetch_one(&mut *tx)
M server/src/db/webhook_events.rs +21 -6
@@ -106,13 +106,28 @@ pub async fn insert_failed_event(
106 106 /// Returns up to 10 at a time.
107 107 #[tracing::instrument(skip_all)]
108 108 pub async fn get_retryable_events(pool: &PgPool) -> Result<Vec<DbWebhookEvent>> {
109 + // Atomically CLAIM the due events rather than plain-SELECT them: select
110 + // `FOR UPDATE SKIP LOCKED` and push `next_retry_at` out so a second scheduler
111 + // replica (or an overlapping tick) can't grab the same rows and double-run the
112 + // handler (audit Run 13 Conc). A claimed event whose process crashes before
113 + // resolution becomes eligible again after the claim window — handlers are
114 + // idempotent, so at-least-once is safe. Post-processing (`mark_processed` /
115 + // `schedule_retry`) rewrites `status`/`next_retry_at` for the terminal state.
109 116 let events = sqlx::query_as::<_, DbWebhookEvent>(
110 - r#"SELECT * FROM webhook_events
111 - WHERE status IN ('failed', 'retrying')
112 - AND attempts < 5
113 - AND next_retry_at <= NOW()
114 - ORDER BY next_retry_at
115 - LIMIT 10"#,
117 + r#"
118 + UPDATE webhook_events
119 + SET next_retry_at = NOW() + INTERVAL '2 minutes'
120 + WHERE id IN (
121 + SELECT id FROM webhook_events
122 + WHERE status IN ('failed', 'retrying')
123 + AND attempts < 5
124 + AND next_retry_at <= NOW()
125 + ORDER BY next_retry_at
126 + LIMIT 10
127 + FOR UPDATE SKIP LOCKED
128 + )
129 + RETURNING *
130 + "#,
116 131 )
117 132 .fetch_all(pool)
118 133 .await?;
M server/tests/workflows/git_issues.rs +72 -2
@@ -72,15 +72,35 @@ async fn setup_with_inbound(tmp: &tempfile::TempDir) -> TestHarness {
72 72 h
73 73 }
74 74
75 - /// Build a Postmark inbound payload JSON string.
75 + /// Build a Postmark inbound payload JSON string. Includes an SPF/DKIM verdict
76 + /// aligned with the From domain — normal production inbound — which the server
77 + /// now requires before trusting the sender's identity (Run 13 sender-spoofing).
76 78 fn inbound_payload(to: &str, from_email: &str, subject: &str, body: &str) -> String {
79 + let domain = from_email.rsplit('@').next().unwrap_or("example.com");
80 + let auth = format!("mx.postmark.com; spf=pass smtp.mailfrom={from_email}; dkim=pass header.d={domain}");
81 + inbound_payload_with_auth(to, from_email, subject, body, Some(&auth))
82 + }
83 +
84 + /// Like `inbound_payload` but with an explicit `Authentication-Results` value
85 + /// (`None` = no auth header at all) — for exercising the sender-auth gate.
86 + fn inbound_payload_with_auth(
87 + to: &str,
88 + from_email: &str,
89 + subject: &str,
90 + body: &str,
91 + auth_results: Option<&str>,
92 + ) -> String {
93 + let headers = match auth_results {
94 + Some(v) => serde_json::json!([{ "Name": "Authentication-Results", "Value": v }]),
95 + None => serde_json::json!([]),
96 + };
77 97 serde_json::json!({
78 98 "FromFull": { "Email": from_email, "Name": "" },
79 99 "To": to,
80 100 "Subject": subject,
81 101 "TextBody": body,
82 102 "MessageID": format!("test-msg-{}", uuid::Uuid::new_v4()),
83 - "Headers": []
103 + "Headers": headers
84 104 })
85 105 .to_string()
86 106 }
@@ -305,6 +325,56 @@ async fn inbound_new_issue_creates_issue() {
305 325 }
306 326
307 327 #[tokio::test]
328 + async fn inbound_spoofed_sender_is_rejected() {
329 + // A forged `From:` matching a verified user, but the SPF/DKIM verdict is
330 + // aligned to the attacker's domain, not the From domain — must NOT create an
331 + // issue as that user (Run 13 sender-spoofing). Returns 200 (no oracle) but
332 + // writes nothing.
333 + let tmp = tempfile::TempDir::new().unwrap();
334 + make_test_repo(tmp.path());
335 + let mut h = setup_with_inbound(&tmp).await;
336 +
337 + h.client.set_bearer_token("test-inbound-secret");
338 + let spoofed = inbound_payload_with_auth(
339 + "testowner+testrepo@issues.makenot.work",
340 + "testowner@example.com",
341 + "Spoofed issue",
342 + "I am pretending to be the owner.",
343 + Some("mx.postmark.com; spf=pass smtp.mailfrom=mallory@evil.test; dkim=pass header.d=evil.test"),
344 + );
345 + let resp = h.client.post_json("/postmark/inbound-issues", &spoofed).await;
346 + assert_eq!(resp.status, 200, "spoofed inbound should ack without an oracle: {}", resp.text);
347 +
348 + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM issues")
349 + .fetch_one(&h.db).await.unwrap();
350 + assert_eq!(count, 0, "a spoofed sender must not create an issue");
351 + }
352 +
353 + #[tokio::test]
354 + async fn inbound_missing_auth_is_rejected() {
355 + // No SPF/DKIM verdict at all (enforce mode) — the sender is unauthenticated,
356 + // so no issue is attributed to the claimed From user.
357 + let tmp = tempfile::TempDir::new().unwrap();
358 + make_test_repo(tmp.path());
359 + let mut h = setup_with_inbound(&tmp).await;
360 +
361 + h.client.set_bearer_token("test-inbound-secret");
362 + let no_auth = inbound_payload_with_auth(
363 + "testowner+testrepo@issues.makenot.work",
364 + "testowner@example.com",
365 + "Unauthenticated issue",
366 + "No auth headers present.",
367 + None,
368 + );
369 + let resp = h.client.post_json("/postmark/inbound-issues", &no_auth).await;
370 + assert_eq!(resp.status, 200);
371 +
372 + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM issues")
373 + .fetch_one(&h.db).await.unwrap();
374 + assert_eq!(count, 0, "an unauthenticated sender must not create an issue");
375 + }
376 +
377 + #[tokio::test]
308 378 async fn inbound_new_issue_requires_verified_sender() {
309 379 let tmp = tempfile::TempDir::new().unwrap();
310 380 make_test_repo(tmp.path());
M server/tests/workflows/patches.rs +15 -1
@@ -37,13 +37,27 @@ fn inbound_payload(
37 37 message_id: &str,
38 38 extra_headers: &[(&str, &str)],
39 39 ) -> String {
40 - let headers: Vec<serde_json::Value> = extra_headers
40 + let mut headers: Vec<serde_json::Value> = extra_headers
41 41 .iter()
42 42 .map(|(name, value)| {
43 43 serde_json::json!({"Name": name, "Value": value})
44 44 })
45 45 .collect();
46 46
47 + // Default to a passing SPF/DKIM verdict aligned with the From domain (normal
48 + // production inbound) unless the caller supplied their own auth header — the
49 + // server requires alignment before trusting the sender (Run 13 spoofing).
50 + let has_auth = extra_headers.iter().any(|(n, _)| {
51 + n.eq_ignore_ascii_case("Authentication-Results") || n.eq_ignore_ascii_case("Received-SPF")
52 + });
53 + if !has_auth {
54 + let domain = from_email.rsplit('@').next().unwrap_or("example.com");
55 + headers.push(serde_json::json!({
56 + "Name": "Authentication-Results",
57 + "Value": format!("mx.postmark.com; spf=pass smtp.mailfrom={from_email}; dkim=pass header.d={domain}")
58 + }));
59 + }
60 +
47 61 serde_json::json!({
48 62 "FromFull": {"Email": from_email, "Name": "Test Sender"},
49 63 "From": from_email,