feat: exception issue lookup (#25878)

Co-authored-by: Oliver Browne <[email protected]>

rust/cymbal/src/issue_resolution.rs

@@ -0,0 +1,166 @@
+use sqlx::postgres::any::AnyConnectionBackend;
+use uuid::Uuid;
+
+use crate::error::UnhandledError;
+
+pub struct IssueFingerprintOverride {
+    pub id: Uuid,
+    pub team_id: i32,
+    pub issue_id: Uuid,
+    pub fingerprint: String,
+    pub version: i64,
+}
+
+pub struct Issue {
+    pub id: Uuid,
+    pub team_id: i32,
+    pub status: String,
+}
+
+impl Issue {
+    pub fn new(team_id: i32) -> Self {
+        Self {
+            id: Uuid::new_v4(),
+            team_id,
+            status: "active".to_string(), // TODO - we should at some point use an enum here
+        }
+    }
+
+    pub async fn load<'c, E>(
+        executor: E,
+        team_id: i32,
+        issue_id: Uuid,
+    ) -> Result<Option<Self>, UnhandledError>
+    where
+        E: sqlx::Executor<'c, Database = sqlx::Postgres>,
+    {
+        let res = sqlx::query_as!(
+            Issue,
+            r#"
+            SELECT id, team_id, status FROM posthog_errortrackingissue
+            WHERE team_id = $1 AND id = $2
+            "#,
+            team_id,
+            issue_id
+        )
+        .fetch_optional(executor)
+        .await?;
+
+        Ok(res)
+    }
+
+    pub async fn insert<'c, E>(&self, executor: E) -> Result<bool, UnhandledError>
+    where
+        E: sqlx::Executor<'c, Database = sqlx::Postgres>,
+    {
+        let did_insert = sqlx::query_scalar!(
+            r#"
+            INSERT INTO posthog_errortrackingissue (id, team_id, status, created_at)
+            VALUES ($1, $2, $3, NOW())
+            ON CONFLICT (id) DO NOTHING
+            RETURNING (xmax = 0) AS was_inserted
+            "#,
+            self.id,
+            self.team_id,
+            self.status
+        )
+        .fetch_one(executor)
+        .await?;
+
+        // TODO - I'm fairly sure the Option here is a bug in sqlx, so the unwrap will
+        // never be hit, but nonetheless I'm not 100% sure the "no rows" case actually
+        // means the insert was not done.
+        Ok(did_insert.unwrap_or(false))
+    }
+}
+
+impl IssueFingerprintOverride {
+    pub async fn load<'c, E>(
+        executor: E,
+        team_id: i32,
+        fingerprint: &str,
+    ) -> Result<Option<Self>, UnhandledError>
+    where
+        E: sqlx::Executor<'c, Database = sqlx::Postgres>,
+    {
+        let res = sqlx::query_as!(
+            IssueFingerprintOverride,
+            r#"
+            SELECT id, team_id, issue_id, fingerprint, version FROM posthog_errortrackingissuefingerprintv2
+            WHERE team_id = $1 AND fingerprint = $2
+            "#,
+            team_id,
+            fingerprint
+        ).fetch_optional(executor).await?;
+
+        Ok(res)
+    }
+
+    pub async fn create_or_load<'c, E>(
+        executor: E,
+        team_id: i32,
+        fingerprint: &str,
+        issue: &Issue,
+    ) -> Result<Self, UnhandledError>
+    where
+        E: sqlx::Executor<'c, Database = sqlx::Postgres>,
+    {
+        // We do an "ON CONFLICT DO NOTHING" here because callers can compare the returned issue id
+        // to the passed Issue, to see if the issue was actually inserted or not.
+        let res = sqlx::query_as!(
+            IssueFingerprintOverride,
+            r#"
+            INSERT INTO posthog_errortrackingissuefingerprintv2 (id, team_id, issue_id, fingerprint, version, created_at)
+            VALUES ($1, $2, $3, $4, 0, NOW())
+            ON CONFLICT (team_id, fingerprint) DO NOTHING
+            RETURNING id, team_id, issue_id, fingerprint, version
+            "#,
+            Uuid::new_v4(),
+            team_id,
+            issue.id,
+            fingerprint
+        ).fetch_one(executor).await?;
+
+        Ok(res)
+    }
+}
+
+pub async fn resolve_issue<'c, A>(
+    con: A,
+    fingerprint: &str,
+    team_id: i32,
+) -> Result<IssueFingerprintOverride, UnhandledError>
+where
+    A: sqlx::Acquire<'c, Database = sqlx::Postgres>,
+{
+    let mut conn = con.acquire().await?;
+    // If an override already exists, just fast-path, skipping the transaction
+    if let Some(issue_override) =
+        IssueFingerprintOverride::load(&mut *conn, team_id, fingerprint).await?
+    {
+        return Ok(issue_override);
+    }
+
+    // Start a transaction, so we can roll it back on override insert failure
+    conn.begin().await?;
+    // Insert a new issue
+    let issue = Issue::new(team_id);
+    // We don't actually care if we insert the issue here or not - conflicts aren't possible at
+    // this stage.
+    issue.insert(&mut *conn).await?;
+    // Insert the fingerprint override
+    let issue_override =
+        IssueFingerprintOverride::create_or_load(&mut *conn, team_id, fingerprint, &issue).await?;
+
+    // If we actually inserted a new row for the issue override, commit the transaction,
+    // saving both the issue and the override. Otherwise, rollback the transaction, and
+    // use the retrieved issue override.
+    let was_created = issue_override.issue_id == issue.id;
+    if !was_created {
+        conn.rollback().await?;
+    } else {
+        conn.commit().await?;
+    }
+
+    Ok(issue_override)
+}

rust/cymbal/src/lib.rs

@@ -4,6 +4,7 @@ use app_context::AppContext;
 use common_types::ClickHouseEvent;
 use error::{EventError, UnhandledError};
 use fingerprinting::generate_fingerprint;
+use issue_resolution::resolve_issue;
 use tracing::warn;
 use types::{ErrProps, Exception, Stacktrace};
 use uuid::Uuid;
@@ -13,6 +14,7 @@ pub mod config;
 pub mod error;
 pub mod fingerprinting;
 pub mod frames;
+pub mod issue_resolution;
 pub mod langs;
 pub mod metric_consts;
 pub mod symbol_store;
@@ -50,7 +52,12 @@ pub async fn handle_event(
         results.push(process_exception(context, event.team_id, exception).await?);
     }
 
-    props.fingerprint = Some(generate_fingerprint(&results));
+    let fingerprint = generate_fingerprint(&results);
+
+    let issue_override = resolve_issue(&context.pool, &fingerprint, event.team_id).await?;
+
+    props.fingerprint = Some(fingerprint);
+    props.resolved_issue_id = Some(issue_override.issue_id);
     props.exception_list = Some(results);
 
     event.properties = Some(serde_json::to_string(&props).unwrap());

rust/cymbal/src/types/mod.rs

@@ -3,6 +3,7 @@ use std::collections::HashMap;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use sha2::{digest::Update, Sha512};
+use uuid::Uuid;
 
 use crate::frames::{Frame, RawFrame};
 
@@ -53,6 +54,11 @@ pub struct ErrProps {
         skip_serializing_if = "Option::is_none"
     )]
     pub fingerprint: Option<String>, // We expect this not to exist when the event is received, and we populate it as part of processing
+    #[serde(
+        rename = "$exception_issue_id",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub resolved_issue_id: Option<Uuid>, // We populate the exception issue id as part of processing
     #[serde(flatten)]
     // A catch-all for all the properties we don't "care" about, so when we send back to kafka we don't lose any info
     pub other: HashMap<String, Value>,

rust/cymbal/tests/test_migrations/20241101134611_test_migration_for_symbol_set_saving_tests.sql

@@ -11,7 +11,6 @@ CREATE TABLE posthog_errortrackingsymbolset (
 -- Create index for team_id and ref combination
 CREATE INDEX idx_error_tracking_symbol_sets_team_ref ON posthog_errortrackingsymbolset(team_id, ref);
 
--- Add migration script here
 CREATE TABLE IF NOT EXISTS posthog_errortrackingstackframe (
     id UUID PRIMARY KEY,
     raw_id TEXT NOT NULL,
@@ -23,3 +22,20 @@ CREATE TABLE IF NOT EXISTS posthog_errortrackingstackframe (
     context TEXT,
     UNIQUE(raw_id, team_id)
 );
+
+CREATE TABLE IF NOT EXISTS posthog_errortrackingissue (
+    id UUID PRIMARY KEY,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    status TEXT,
+    team_id INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS posthog_errortrackingissuefingerprintv2 (
+    id UUID PRIMARY KEY,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    fingerprint TEXT NOT NULL,
+    version BIGINT NOT NULL,
+    team_id INTEGER NOT NULL,
+    issue_id UUID NOT NULL,
+    UNIQUE(team_id, fingerprint)
+);