fix(eap-outcomes): Add trace ID and timestamp to dedup key

rust_snuba/src/strategies/accepted_outcomes/aggregator.rs

@@ -91,10 +91,26 @@ impl<TNext> OutcomesAggregator<TNext> {
         let item_type =
             TraceItemType::try_from(trace_item.item_type).unwrap_or(TraceItemType::Unspecified);
 
+        if item_type != TraceItemType::Span {
+            return false;
+        }
+
+        let timestamp = trace_item
+            .timestamp
+            .as_ref()
+            .map(|t| t.seconds as u32)
+            .unwrap_or(0);
+
+        let trace_id = uuid::Uuid::try_parse(&trace_item.trace_id)
+            .map(|u| *u.as_bytes())
+            .unwrap_or([0u8; 16]);
+
         if let Ok(item_id) = <[u8; 16]>::try_from(trace_item.item_id.as_slice()) {
             let dedup_key = ItemDedupKey {
                 org_id,
                 project_id,
+                timestamp,
+                trace_id,
                 item_id,
             };
             return self.batch.record_if_duplicate(item_type, dedup_key);
@@ -782,6 +798,7 @@ mod tests {
             organization_id: org_id,
             project_id,
             item_id: item_id.to_vec(),
+            item_type: TraceItemType::Span.into(),
             received: Some(Timestamp {
                 seconds: ts_secs,
                 nanos: 0,

rust_snuba/src/types.rs

@@ -619,11 +619,15 @@ pub struct TrackOutcome {
 }
 
 /// Key used to deduplicate items within an outcomes batch.
-/// Uses the relevant fields from the eap_items table sorting key.
+/// Uses the relevant fields from the eap_items table sorting key:
+/// (organization_id, project_id, item_type, timestamp, trace_id, item_id)
+/// Note: item_type is handled separately via the HashMap in seen_items.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct ItemDedupKey {
     pub org_id: u64,
     pub project_id: u64,
+    pub timestamp: u32,
+    pub trace_id: [u8; 16],
     pub item_id: [u8; 16],
 }