From 36aa88a8141c6bff9ddaff0138475f257cf7a128 Mon Sep 17 00:00:00 2001 From: ajianaz Date: Tue, 13 Oct 2026 08:36:26 +0700 Subject: [PATCH] fix(db): atomic record_event, safe orphan purge, RFC3339 retention cutoff (#332) - record_event upserts issue + inserts event in one transaction (both backends); digest uses it - purge_orphan_issues(days) only removes unresolved orphans with last_seen < cutoff - bind Rust-computed RFC3339 cutoff for events/auth_attempts purge on both backends Co-Authored-By: Claude Sonnet 5.5 --- crates/trapfall-core/src/store.rs | 13 ++ crates/trapfall-db/src/common.rs | 6 + crates/trapfall-db/src/lib.rs | 17 +- crates/trapfall-db/src/postgres.rs | 126 ++++++++++----- crates/trapfall-db/src/sqlite.rs | 240 +++++++++++++++++++++++------ crates/trapfall-db/tests/common.rs | 2 +- crates/trapfalld/src/digest.rs | 9 +- crates/trapfalld/src/retention.rs | 2 +- 8 files changed, 316 insertions(+), 99 deletions(-) diff --git a/crates/trapfall-core/src/store.rs b/crates/trapfall-core/src/store.rs index dfd68b8..9f20064 100644 --- a/crates/trapfall-core/src/store.rs +++ b/crates/trapfall-core/src/store.rs @@ -95,6 +95,19 @@ impl Store { self.db.upsert_issue(project_id, fingerprint, title, culprit, level).await } + /// Atomically upsert an issue and insert its event. + pub async fn record_event( + &self, + project_id: &str, + fingerprint: &str, + title: &str, + culprit: Option<&str>, + level: Level, + event_data: &str, + ) -> Result { + self.db.record_event(project_id, fingerprint, title, culprit, level, event_data).await + } + pub async fn get_issue(&self, issue_id: &str) -> Result> { self.db.get_issue(issue_id).await } diff --git a/crates/trapfall-db/src/common.rs b/crates/trapfall-db/src/common.rs index ee18fb0..374b54b 100644 --- a/crates/trapfall-db/src/common.rs +++ b/crates/trapfall-db/src/common.rs @@ -246,6 +246,12 @@ pub fn extract_dsn_host(dsn: &str) -> String { .to_string() } +/// RFC3339 cutoff `days` days before now. Same format as stored timestamps +/// (`now_rfc3339`), so text comparison is consistent on both backends. +pub fn cutoff_rfc3339_days(days: i64) -> String { + (chrono::Utc::now() - chrono::Duration::days(days)).to_rfc3339() +} + /// Current time as RFC3339 string. pub fn now_rfc3339() -> String { chrono::Utc::now().to_rfc3339() diff --git a/crates/trapfall-db/src/lib.rs b/crates/trapfall-db/src/lib.rs index 4d4ad51..0a54d27 100644 --- a/crates/trapfall-db/src/lib.rs +++ b/crates/trapfall-db/src/lib.rs @@ -244,6 +244,17 @@ pub trait Database: Send + Sync { // ── Events ────────────────────────────────────────────────────────── async fn insert_event(&self, issue_id: &str, project_id: &str, event_data: &str) -> Result; + /// Upsert the issue and insert its event in ONE transaction, so an event + /// failure never leaves a bumped issue count without an event row. + async fn record_event( + &self, + project_id: &str, + fingerprint: &str, + title: &str, + culprit: Option<&str>, + level: Level, + event_data: &str, + ) -> Result; async fn list_events(&self, issue_id: &str, limit: i64, offset: i64) -> Result>; async fn count_events(&self, issue_id: &str) -> Result; @@ -337,8 +348,10 @@ pub trait Database: Send + Sync { /// Delete events older than `days` days. Returns count of deleted rows. async fn purge_old_events(&self, days: i64) -> Result; - /// Delete orphaned issues (no remaining events). Best-effort. - async fn purge_orphan_issues(&self) -> Result<()>; + /// Delete orphaned issues: no remaining events, not resolved/ignored, and + /// `last_seen` older than `days` days (so a freshly upserted issue whose + /// event is not yet inserted is never removed). Best-effort. + async fn purge_orphan_issues(&self, days: i64) -> Result<()>; /// Delete auth attempts older than 30 days. Best-effort. async fn purge_stale_auth_attempts(&self) -> Result<()>; diff --git a/crates/trapfall-db/src/postgres.rs b/crates/trapfall-db/src/postgres.rs index c83476f..b3d44f5 100644 --- a/crates/trapfall-db/src/postgres.rs +++ b/crates/trapfall-db/src/postgres.rs @@ -201,29 +201,25 @@ impl Database for PostgresBackend { culprit: Option<&str>, level: Level, ) -> Result { - let level_str = level_to_str(level); - let id = new_id(); - let now = now_rfc3339(); - - // INSERT ... ON CONFLICT DO UPDATE — returns the row in both cases. - let row = sqlx::query_as::<_, IssueRow>( - "INSERT INTO issues (id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen) \ - VALUES ($1, $2, $3, $4, $5, 'unresolved', $6, 1, 0, $7, $7) \ - ON CONFLICT (project_id, fingerprint) DO UPDATE SET \ - count = issues.count + 1, last_seen = $7, level = $6 \ - RETURNING id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen", - ) - .bind(&id) - .bind(project_id) - .bind(fingerprint) - .bind(title) - .bind(culprit) - .bind(&level_str) - .bind(&now) - .fetch_one(&self.pool) - .await?; + let mut conn = self.pool.acquire().await?; + upsert_issue_conn(&mut conn, project_id, fingerprint, title, culprit, level).await + } - Ok(row.into()) + async fn record_event( + &self, + project_id: &str, + fingerprint: &str, + title: &str, + culprit: Option<&str>, + level: Level, + event_data: &str, + ) -> Result { + // One transaction: if the event insert fails the issue bump rolls back. + let mut tx = self.pool.begin().await?; + let issue = upsert_issue_conn(&mut tx, project_id, fingerprint, title, culprit, level).await?; + insert_event_conn(&mut tx, &issue.id, project_id, event_data).await?; + tx.commit().await?; + Ok(issue) } async fn get_issue(&self, issue_id: &str) -> Result> { @@ -321,17 +317,8 @@ impl Database for PostgresBackend { // ── Events ───────────────────────────────────────────────────────── async fn insert_event(&self, issue_id: &str, project_id: &str, event_data: &str) -> Result { - let id = new_id(); - let now = now_rfc3339(); - sqlx::query("INSERT INTO events (id, issue_id, project_id, data, received_at) VALUES ($1, $2, $3, $4, $5)") - .bind(&id) - .bind(issue_id) - .bind(project_id) - .bind(event_data) - .bind(&now) - .execute(&self.pool) - .await?; - Ok(id) + let mut conn = self.pool.acquire().await?; + insert_event_conn(&mut conn, issue_id, project_id, event_data).await } async fn list_events(&self, issue_id: &str, limit: i64, offset: i64) -> Result> { @@ -843,20 +830,27 @@ impl Database for PostgresBackend { // ── Retention ────────────────────────────────────────────────────── async fn purge_old_events(&self, days: i64) -> Result { - let cutoff = (chrono::Utc::now() - chrono::Duration::days(days)).to_rfc3339(); + let cutoff = cutoff_rfc3339_days(days); let result = sqlx::query("DELETE FROM events WHERE received_at < $1").bind(&cutoff).execute(&self.pool).await?; Ok(result.rows_affected()) } - async fn purge_orphan_issues(&self) -> Result<()> { - sqlx::query("DELETE FROM issues WHERE id NOT IN (SELECT DISTINCT issue_id FROM events)") - .execute(&self.pool) - .await?; + async fn purge_orphan_issues(&self, days: i64) -> Result<()> { + // Single atomic statement. `last_seen < cutoff` protects a freshly + // upserted issue whose event is not yet inserted; resolved and ignored + // issues are kept as history. + sqlx::query( + "DELETE FROM issues WHERE last_seen < $1 AND status NOT IN ('resolved', 'ignored') \ + AND NOT EXISTS (SELECT 1 FROM events WHERE events.issue_id = issues.id)", + ) + .bind(cutoff_rfc3339_days(days)) + .execute(&self.pool) + .await?; Ok(()) } async fn purge_stale_auth_attempts(&self) -> Result<()> { - let cutoff = (chrono::Utc::now() - chrono::Duration::days(30)).to_rfc3339(); + let cutoff = cutoff_rfc3339_days(30); sqlx::query("DELETE FROM auth_attempts WHERE created_at < $1").bind(&cutoff).execute(&self.pool).await?; Ok(()) } @@ -1072,3 +1066,57 @@ impl Database for PostgresBackend { Ok(row.map(Into::into)) } } + +/// Atomic issue upsert on an existing connection/transaction. +async fn upsert_issue_conn( + conn: &mut sqlx::PgConnection, + project_id: &str, + fingerprint: &str, + title: &str, + culprit: Option<&str>, + level: Level, +) -> Result { + let level_str = level_to_str(level); + let id = new_id(); + let now = now_rfc3339(); + + // INSERT ... ON CONFLICT DO UPDATE returns the row in both cases. + let row = sqlx::query_as::<_, IssueRow>( + "INSERT INTO issues (id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen) \ + VALUES ($1, $2, $3, $4, $5, 'unresolved', $6, 1, 0, $7, $7) \ + ON CONFLICT (project_id, fingerprint) DO UPDATE SET \ + count = issues.count + 1, last_seen = $7, level = $6 \ + RETURNING id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen", + ) + .bind(&id) + .bind(project_id) + .bind(fingerprint) + .bind(title) + .bind(culprit) + .bind(&level_str) + .bind(&now) + .fetch_one(&mut *conn) + .await?; + + Ok(row.into()) +} + +/// Insert an event row on an existing connection/transaction. +async fn insert_event_conn( + conn: &mut sqlx::PgConnection, + issue_id: &str, + project_id: &str, + event_data: &str, +) -> Result { + let id = new_id(); + let now = now_rfc3339(); + sqlx::query("INSERT INTO events (id, issue_id, project_id, data, received_at) VALUES ($1, $2, $3, $4, $5)") + .bind(&id) + .bind(issue_id) + .bind(project_id) + .bind(event_data) + .bind(&now) + .execute(&mut *conn) + .await?; + Ok(id) +} diff --git a/crates/trapfall-db/src/sqlite.rs b/crates/trapfall-db/src/sqlite.rs index 73a7d06..e4fad6d 100644 --- a/crates/trapfall-db/src/sqlite.rs +++ b/crates/trapfall-db/src/sqlite.rs @@ -202,40 +202,27 @@ impl Database for SqliteBackend { culprit: Option<&str>, level: Level, ) -> Result { - let level_str = level_to_str(level); - let now = chrono::Utc::now().to_rfc3339(); - - // Atomic upsert via ON CONFLICT — avoids the SELECT-then-UPDATE race - // where two concurrent ingests could both read count=N and both write - // N+1 (lost update). The unique constraint on (project_id, fingerprint) - // makes this safe. - let row = sqlx::query_as::<_, IssueRow>( - "INSERT INTO issues (id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen) \n VALUES (?, ?, ?, ?, ?, 'unresolved', ?, 1, 0, ?, ?) \n ON CONFLICT(project_id, fingerprint) DO UPDATE SET \n count = count + 1,\n last_seen = excluded.last_seen,\n level = excluded.level\n RETURNING id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen", - ) - .bind(new_id()) - .bind(project_id) - .bind(fingerprint) - .bind(title) - .bind(culprit) - .bind(&level_str) - .bind(&now) - .bind(&now) - .fetch_one(&self.pool) - .await?; + let mut conn = self.pool.acquire().await?; + upsert_issue_conn(&mut conn, project_id, fingerprint, title, culprit, level).await + } - Ok(Issue { - id: row.id, - project_id: row.project_id, - fingerprint: row.fingerprint, - title: row.title, - culprit: row.culprit, - status: str_to_status(&row.status), - level, - count: row.count, - user_count: row.user_count, - first_seen: row.first_seen, - last_seen: row.last_seen, - }) + async fn record_event( + &self, + project_id: &str, + fingerprint: &str, + title: &str, + culprit: Option<&str>, + level: Level, + event_data: &str, + ) -> Result { + // One transaction: if the event insert fails the issue bump rolls back. + // The first statement is a write, so the deferred BEGIN takes the + // write lock immediately (no read->write upgrade BUSY under WAL). + let mut tx = self.pool.begin().await?; + let issue = upsert_issue_conn(&mut tx, project_id, fingerprint, title, culprit, level).await?; + insert_event_conn(&mut tx, &issue.id, project_id, event_data).await?; + tx.commit().await?; + Ok(issue) } async fn get_issue(&self, issue_id: &str) -> Result> { @@ -327,17 +314,8 @@ impl Database for SqliteBackend { // ── Events ───────────────────────────────────────────────────────── async fn insert_event(&self, issue_id: &str, project_id: &str, event_data: &str) -> Result { - let id = new_id(); - let now = chrono::Utc::now().to_rfc3339(); - sqlx::query("INSERT INTO events (id, issue_id, project_id, data, received_at) VALUES (?, ?, ?, ?, ?)") - .bind(&id) - .bind(issue_id) - .bind(project_id) - .bind(event_data) - .bind(&now) - .execute(&self.pool) - .await?; - Ok(id) + let mut conn = self.pool.acquire().await?; + insert_event_conn(&mut conn, issue_id, project_id, event_data).await } async fn list_events(&self, issue_id: &str, limit: i64, offset: i64) -> Result> { @@ -844,22 +822,32 @@ impl Database for SqliteBackend { // ── Retention ────────────────────────────────────────────────────── async fn purge_old_events(&self, days: i64) -> Result { - let result = sqlx::query("DELETE FROM events WHERE received_at < datetime('now', ?)") - .bind(format!("-{days} days")) + // Bind an RFC3339 cutoff (same format as `received_at` writers): + // `datetime('now', ..)` uses a space separator and skews text compares. + let result = sqlx::query("DELETE FROM events WHERE received_at < ?") + .bind(cutoff_rfc3339_days(days)) .execute(&self.pool) .await?; Ok(result.rows_affected()) } - async fn purge_orphan_issues(&self) -> Result<()> { - sqlx::query("DELETE FROM issues WHERE id NOT IN (SELECT DISTINCT issue_id FROM events)") - .execute(&self.pool) - .await?; + async fn purge_orphan_issues(&self, days: i64) -> Result<()> { + // Single atomic statement. `last_seen < cutoff` protects an issue that + // was just upserted but whose event is not yet inserted; resolved and + // ignored issues are kept as history. + sqlx::query( + "DELETE FROM issues WHERE last_seen < ? AND status NOT IN ('resolved', 'ignored') \ + AND NOT EXISTS (SELECT 1 FROM events WHERE events.issue_id = issues.id)", + ) + .bind(cutoff_rfc3339_days(days)) + .execute(&self.pool) + .await?; Ok(()) } async fn purge_stale_auth_attempts(&self) -> Result<()> { - sqlx::query("DELETE FROM auth_attempts WHERE created_at < datetime('now', '-30 days')") + sqlx::query("DELETE FROM auth_attempts WHERE created_at < ?") + .bind(cutoff_rfc3339_days(30)) .execute(&self.pool) .await?; Ok(()) @@ -1078,6 +1066,70 @@ impl Database for SqliteBackend { // ── Tests ────────────────────────────────────────────────────────────── +/// Atomic issue upsert on an existing connection/transaction. +async fn upsert_issue_conn( + conn: &mut sqlx::SqliteConnection, + project_id: &str, + fingerprint: &str, + title: &str, + culprit: Option<&str>, + level: Level, +) -> Result { + let level_str = level_to_str(level); + let now = chrono::Utc::now().to_rfc3339(); + + // Atomic upsert via ON CONFLICT: avoids the SELECT-then-UPDATE race where + // two concurrent ingests both read count=N and both write N+1 (lost + // update). The unique (project_id, fingerprint) constraint makes it safe. + let row = sqlx::query_as::<_, IssueRow>( + "INSERT INTO issues (id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen) \n VALUES (?, ?, ?, ?, ?, 'unresolved', ?, 1, 0, ?, ?) \n ON CONFLICT(project_id, fingerprint) DO UPDATE SET \n count = count + 1,\n last_seen = excluded.last_seen,\n level = excluded.level\n RETURNING id, project_id, fingerprint, title, culprit, status, level, count, user_count, first_seen, last_seen", + ) + .bind(new_id()) + .bind(project_id) + .bind(fingerprint) + .bind(title) + .bind(culprit) + .bind(&level_str) + .bind(&now) + .bind(&now) + .fetch_one(&mut *conn) + .await?; + + Ok(Issue { + id: row.id, + project_id: row.project_id, + fingerprint: row.fingerprint, + title: row.title, + culprit: row.culprit, + status: str_to_status(&row.status), + level, + count: row.count, + user_count: row.user_count, + first_seen: row.first_seen, + last_seen: row.last_seen, + }) +} + +/// Insert an event row on an existing connection/transaction. +async fn insert_event_conn( + conn: &mut sqlx::SqliteConnection, + issue_id: &str, + project_id: &str, + event_data: &str, +) -> Result { + let id = new_id(); + let now = chrono::Utc::now().to_rfc3339(); + sqlx::query("INSERT INTO events (id, issue_id, project_id, data, received_at) VALUES (?, ?, ?, ?, ?)") + .bind(&id) + .bind(issue_id) + .bind(project_id) + .bind(event_data) + .bind(&now) + .execute(&mut *conn) + .await?; + Ok(id) +} + #[cfg(test)] mod tests { use super::*; @@ -1115,6 +1167,92 @@ mod tests { SqliteBackend::new(pool) } + #[tokio::test] + async fn test_record_event_is_atomic() { + let db = open_backend().await; + let p = db.create_project("atomic", "Atomic").await.unwrap(); + sqlx::query("CREATE TRIGGER fail_ev BEFORE INSERT ON events BEGIN SELECT RAISE(ABORT, 'boom'); END") + .execute(&db.pool) + .await + .unwrap(); + let res = db.record_event(&p.id, "fp1", "T", None, Level::Error, "{}").await; + assert!(res.is_err()); + // Issue upsert must have been rolled back with the failed event insert. + assert_eq!(db.count_issues(&p.id, None, None).await.unwrap(), 0); + + sqlx::query("DROP TRIGGER fail_ev").execute(&db.pool).await.unwrap(); + let issue = db.record_event(&p.id, "fp1", "T", None, Level::Error, "{}").await.unwrap(); + assert_eq!(issue.count, 1); + assert_eq!(db.count_events(&issue.id).await.unwrap(), 1); + } + + #[tokio::test] + async fn test_purge_old_events_cutoff_format() { + let db = open_backend().await; + let p = db.create_project("cutoff", "Cutoff").await.unwrap(); + let issue = db.upsert_issue(&p.id, "fp", "T", None, Level::Error).await.unwrap(); + let old = (chrono::Utc::now() - chrono::Duration::hours(25)).to_rfc3339(); + let fresh = (chrono::Utc::now() - chrono::Duration::hours(23)).to_rfc3339(); + for (id, ts) in [("old", &old), ("fresh", &fresh)] { + sqlx::query("INSERT INTO events (id, issue_id, project_id, data, received_at) VALUES (?, ?, ?, '{}', ?)") + .bind(id) + .bind(&issue.id) + .bind(&p.id) + .bind(ts) + .execute(&db.pool) + .await + .unwrap(); + } + // 25h-old event is past a 1-day retention; 23h-old is not. + assert_eq!(db.purge_old_events(1).await.unwrap(), 1); + assert_eq!(db.count_events(&issue.id).await.unwrap(), 1); + } + + #[tokio::test] + async fn test_purge_stale_auth_attempts_cutoff_format() { + let db = open_backend().await; + let old = (chrono::Utc::now() - chrono::Duration::days(30) - chrono::Duration::hours(2)).to_rfc3339(); + let fresh = (chrono::Utc::now() - chrono::Duration::days(30) + chrono::Duration::hours(2)).to_rfc3339(); + for (id, ts) in [("a", &old), ("b", &fresh)] { + sqlx::query("INSERT INTO auth_attempts (id, email, ip, success, created_at) VALUES (?, 'e', 'i', 0, ?)") + .bind(id) + .bind(ts) + .execute(&db.pool) + .await + .unwrap(); + } + db.purge_stale_auth_attempts().await.unwrap(); + let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM auth_attempts").fetch_one(&db.pool).await.unwrap(); + assert_eq!(n, 1); + } + + #[tokio::test] + async fn test_purge_orphan_issues_safety() { + let db = open_backend().await; + let p = db.create_project("orph", "Orph").await.unwrap(); + let fresh = db.upsert_issue(&p.id, "fresh", "Fresh", None, Level::Error).await.unwrap(); + let old = db.upsert_issue(&p.id, "old", "Old", None, Level::Error).await.unwrap(); + let resolved = db.upsert_issue(&p.id, "res", "Resolved", None, Level::Error).await.unwrap(); + db.set_issue_status(&resolved.id, IssueStatus::Resolved).await.unwrap(); + let ancient = (chrono::Utc::now() - chrono::Duration::days(200)).to_rfc3339(); + sqlx::query("UPDATE issues SET last_seen = ? WHERE id IN (?, ?)") + .bind(&ancient) + .bind(&old.id) + .bind(&resolved.id) + .execute(&db.pool) + .await + .unwrap(); + + db.purge_orphan_issues(90).await.unwrap(); + + // Just-upserted issue (event not yet inserted) survives. + assert!(db.get_issue(&fresh.id).await.unwrap().is_some()); + // Old unresolved orphan is purged. + assert!(db.get_issue(&old.id).await.unwrap().is_none()); + // Resolved history is kept even without events. + assert!(db.get_issue(&resolved.id).await.unwrap().is_some()); + } + #[tokio::test] async fn test_seq_backfill_for_legacy_projects() { let pool = sqlx::sqlite::SqlitePoolOptions::new().max_connections(1).connect("sqlite::memory:").await.unwrap(); diff --git a/crates/trapfall-db/tests/common.rs b/crates/trapfall-db/tests/common.rs index f25982c..2aeb6dc 100644 --- a/crates/trapfall-db/tests/common.rs +++ b/crates/trapfall-db/tests/common.rs @@ -339,7 +339,7 @@ pub async fn retention(db: Arc) { assert_eq!(deleted, 0); // Orphan purge (safe — issue has events) - db.purge_orphan_issues().await.unwrap(); + db.purge_orphan_issues(36500).await.unwrap(); let still_there = db.get_issue(&issue.id).await.unwrap(); assert!(still_there.is_some()); } diff --git a/crates/trapfalld/src/digest.rs b/crates/trapfalld/src/digest.rs index cb25a50..957fd3b 100644 --- a/crates/trapfalld/src/digest.rs +++ b/crates/trapfalld/src/digest.rs @@ -126,12 +126,11 @@ impl DigestTask { let level = ev.event.level; - // Upsert issue (creates or increments count) - let issue = store.upsert_issue(&ev.project_id, &ev.fingerprint.hash, &title, culprit.as_deref(), level).await?; - - // Store raw event JSON + // Upsert issue and store raw event JSON atomically (one transaction) let event_json = serde_json::to_string(&ev.event).unwrap_or_default(); - let _event_id = store.insert_event(&issue.id, &ev.project_id, &event_json).await?; + let issue = store + .record_event(&ev.project_id, &ev.fingerprint.hash, &title, culprit.as_deref(), level, &event_json) + .await?; Ok(issue) } diff --git a/crates/trapfalld/src/retention.rs b/crates/trapfalld/src/retention.rs index 59b9b60..49218ba 100644 --- a/crates/trapfalld/src/retention.rs +++ b/crates/trapfalld/src/retention.rs @@ -26,7 +26,7 @@ pub async fn run_retention(store: &Store, retention_days: Option) { async fn purge_old_events(store: &Store, days: i64) -> anyhow::Result<()> { let db = store.backend(); let deleted = db.purge_old_events(days).await?; - db.purge_orphan_issues().await?; + db.purge_orphan_issues(days).await?; db.purge_stale_auth_attempts().await?; if deleted > 0 {