fix: Don't synthesize event IDs for requires_event items

CHANGELOG.md

@@ -26,6 +26,7 @@
 - Reject Nintendo Switch dying message attachments with an invalid magic number instead of panicking on a short payload. ([#6253](https://github.com/getsentry/relay/pull/6253))
 - Downgrade Kafka to prevent producers from getting stuck. ([#6336](https://github.com/getsentry/relay/pull/6336))
 - Prevent memory bomb in PII processor's `split_chunks`. ([#6343](https://github.com/getsentry/relay/pull/6343))
+- Fill in missing event IDs only if items would create events. ([#6350](https://github.com/getsentry/relay/pull/6350))
 
 **Internal**:
 

relay-server/src/envelope/mod.rs

@@ -330,7 +330,7 @@ impl Envelope {
 
         // Event-related envelopes *must* contain an event id.
         let items = Self::parse_items(&bytes, offset)?;
-        if items.iter().any(Item::requires_event) {
+        if items.iter().any(Item::creates_event) {
             headers.event_id.get_or_insert_with(EventId::new);
         }
 

relay-server/src/processing/attachments/mod.rs

@@ -7,7 +7,6 @@ use crate::envelope::{EnvelopeHeaders, Item, ItemType, Items};
 use crate::managed::{Counted, Managed, ManagedEnvelope, OutcomeError, Quantities, Rejected};
 use crate::processing::utils::attachments;
 use crate::processing::{self, CountRateLimited, Output, QuotaRateLimiter};
-#[cfg(feature = "processing")]
 use crate::services::outcome::DiscardReason;
 use crate::services::outcome::Outcome;
 use crate::statsd::RelayCounters;
@@ -23,7 +22,6 @@ pub enum Error {
     RateLimited(RateLimits),
 
     /// The envelope did not contain an event ID.
-    #[cfg(feature = "processing")]
     #[error("missing event ID")]
     NoEventId,
 }
@@ -37,8 +35,7 @@ impl OutcomeError for Error {
                 let reason_code = limits.longest().and_then(|limit| limit.reason_code.clone());
                 Some(Outcome::RateLimited(reason_code))
             }
-            #[cfg(feature = "processing")]
-            Self::NoEventId => Some(Outcome::Invalid(DiscardReason::Internal)),
+            Self::NoEventId => Some(Outcome::Invalid(DiscardReason::InvalidEventId)),
         };
         (outcome, self)
     }
@@ -98,12 +95,19 @@ impl processing::Processor for AttachmentProcessor {
         mut attachments: Managed<Self::Input>,
         ctx: processing::Context<'_>,
     ) -> Result<processing::Output<Self::Output>, Rejected<Self::Error>> {
+        let has_event_id = attachments.headers.event_id().is_some();
+
         // Temporary counter to figure out which SDKs are still sending standalone attachments.
         relay_statsd::metric!(
             counter(RelayCounters::StandaloneAttachment) += 1,
-            sdk = client_name_tag(attachments.headers.meta().client_name())
+            sdk = client_name_tag(attachments.headers.meta().client_name()),
+            has_event_id = has_event_id.to_string()
         );
 
+        if !has_event_id {
+            return Err(attachments.reject_err(Error::NoEventId));
+        }
+
         attachments::validate_attachments(&mut attachments, |a| &mut a.attachments, ctx);
 
         let mut attachments = self.limiter.enforce_quotas(attachments, ctx).await?;

relay-server/src/processing/replays/forward.rs

@@ -50,11 +50,9 @@ impl Forward for ReplaysOutput {
     ) -> Result<(), Rejected<()>> {
         let Self(replay) = self;
 
-        let event_id = replay.headers.event_id().ok_or_else(|| {
-            replay
-                .reject_err(crate::processing::replays::Error::NoEventId)
-                .map(drop)
-        })?;
+        let Some(event_id) = replay.headers.event_id() else {
+            return Err(replay.reject_err(super::Error::NoEventId).map(drop));
+        };
 
         let ctx = store::Context {
             event_id,

relay-server/src/processing/replays/mod.rs

@@ -76,7 +76,6 @@ pub enum Error {
     #[error("replay recording too large")]
     TooLarge,
     /// The envelope did not contain an event ID.
-    #[cfg(feature = "processing")]
     #[error("missing replay ID")]
     NoEventId,
 }
@@ -117,8 +116,7 @@ impl OutcomeError for Error {
             Self::TooLarge => Some(Outcome::Invalid(DiscardReason::ItemTooLarge(
                 crate::services::outcome::DiscardItemType::ReplayRecording,
             ))),
-            #[cfg(feature = "processing")]
-            Self::NoEventId => Some(Outcome::Invalid(DiscardReason::Internal)),
+            Self::NoEventId => Some(Outcome::Invalid(DiscardReason::InvalidEventId)),
         };
         (outcome, self)
     }

relay-server/src/processing/replays/validate.rs

@@ -1,9 +1,24 @@
 use relay_event_normalization::replay;
 
 use crate::processing::replays::{Error, ExpandedReplay, Result};
+use crate::statsd::RelayCounters;
+use crate::utils::client_name_tag;
 
 /// Checks the structural validity of a replay, rejects it if invalid.
 pub fn validate(replay: &ExpandedReplay) -> Result<()> {
+    let has_event_id = replay.headers.event_id().is_some();
+
+    // Temporary counter to figure out which SDKs are sending replays without event IDs.
+    relay_statsd::metric!(
+        counter(RelayCounters::Replay) += 1,
+        sdk = client_name_tag(replay.headers.meta().client_name()),
+        has_event_id = has_event_id.to_string(),
+    );
+
+    if !has_event_id {
+        return Err(Error::NoEventId);
+    }
+
     let Some(event) = replay.payload.event() else {
         return Ok(());
     };

relay-server/src/processing/user_reports/mod.rs

@@ -4,12 +4,15 @@ use relay_cogs::{AppFeature, FeatureWeights};
 use relay_quotas::RateLimits;
 
 use crate::envelope::{EnvelopeHeaders, Item, ItemType};
-use crate::managed::{Counted, Managed, ManagedEnvelope, OutcomeError, Quantities, Rejected};
+use crate::managed::{
+    Counted, Managed, ManagedEnvelope, ManagedResult, OutcomeError, Quantities, Rejected,
+};
 use crate::processing::{Context, CountRateLimited, Output, Processor, QuotaRateLimiter};
 use crate::services::outcome::Outcome;
 
 mod forward;
 mod process;
+mod validate;
 
 pub use process::process_user_reports;
 
@@ -20,7 +23,6 @@ pub enum Error {
     RateLimited(RateLimits),
 
     /// The envelope did not contain an event ID.
-    #[cfg(feature = "processing")]
     #[error("missing event ID")]
     NoEventId,
 }
@@ -34,9 +36,8 @@ impl OutcomeError for Error {
                 let reason_code = limits.longest().and_then(|limit| limit.reason_code.clone());
                 Some(Outcome::RateLimited(reason_code))
             }
-            #[cfg(feature = "processing")]
             Self::NoEventId => Some(Outcome::Invalid(
-                crate::services::outcome::DiscardReason::Internal,
+                crate::services::outcome::DiscardReason::InvalidEventId,
             )),
         };
         (outcome, self)
@@ -91,6 +92,7 @@ impl Processor for UserReportsProcessor {
         mut reports: Managed<Self::Input>,
         ctx: Context<'_>,
     ) -> Result<Output<Self::Output>, Rejected<Self::Error>> {
+        validate::validate(&reports).reject(&reports)?;
         process::process(&mut reports);
 
         let reports = self.limiter.enforce_quotas(reports, ctx).await?;

relay-server/src/processing/user_reports/validate.rs

@@ -0,0 +1,21 @@
+use crate::processing::user_reports::{Error, SerializedUserReports};
+use crate::statsd::RelayCounters;
+use crate::utils::client_name_tag;
+
+/// Checks that the user reports contain an event ID.
+pub fn validate(reports: &SerializedUserReports) -> Result<(), Error> {
+    let has_event_id = reports.headers.event_id().is_some();
+
+    // Temporary counter to figure out which SDKs are sending user reports without event IDs.
+    relay_statsd::metric!(
+        counter(RelayCounters::UserReport) += 1,
+        sdk = client_name_tag(reports.headers.meta().client_name()),
+        has_event_id = has_event_id.to_string(),
+    );
+
+    if !has_event_id {
+        return Err(Error::NoEventId);
+    }
+
+    Ok(())
+}

relay-server/src/statsd.rs

@@ -1026,7 +1026,20 @@ pub enum RelayCounters {
     ///
     /// This metric is tagged with:
     /// - `sdk`: low-cardinality client name
+    /// - `has_event_id`: whether the envelope contained an event ID
     StandaloneAttachment,
+    /// Amount of user reports processed.
+    ///
+    /// This metric is tagged with:
+    /// - `sdk`: low-cardinality client name
+    /// - `has_event_id`: whether the envelope contained an event ID
+    UserReport,
+    /// Amount of replays processed.
+    ///
+    /// This metric is tagged with:
+    /// - `sdk`: low-cardinality client name
+    /// - `has_event_id`: whether the envelope contained an event ID
+    Replay,
 }
 
 impl CounterMetric for RelayCounters {
@@ -1094,6 +1107,8 @@ impl CounterMetric for RelayCounters {
             RelayCounters::CompressedMinidump => "minidump.compressed.count",
             RelayCounters::TraceMetricNilTraceId => "trace_metric.nil_trace_id",
             RelayCounters::StandaloneAttachment => "processing.standalone_attachment",
+            RelayCounters::UserReport => "processing.user_report",
+            RelayCounters::Replay => "processing.replay",
         }
     }
 }

tests/integration/fixtures/__init__.py

@@ -335,15 +335,12 @@ def send_transaction(
 
         self.send_envelope(project_id, envelope)
 
-    def send_replay_event(self, project_id, payload, item_headers=None):
-        envelope = Envelope()
+    def send_replay_event(self, project_id, payload, envelope_headers=None):
+        envelope = Envelope(headers=envelope_headers)
         envelope.add_item(Item(payload=PayloadRef(json=payload), type="replay_event"))
         envelope.add_item(
             Item(payload=PayloadRef(bytes=b"{}\n[]"), type="replay_recording")
         )
-        if envelope.headers is None:
-            envelope.headers = {}
-
         self.send_envelope(project_id, envelope)
 
     def send_session_aggregates(self, project_id, payload, headers=None):

tests/integration/test_attachments.py

@@ -6,6 +6,7 @@
 
 from requests.exceptions import HTTPError
 from sentry_sdk.envelope import Envelope, Item, PayloadRef
+from sentry_relay.consts import DataCategory
 
 from .asserts import matches_any
 from .test_store import make_transaction
@@ -446,7 +447,7 @@ def test_attachments_pii_logfile(mini_sentry, relay):
 path=c:\Users\yan\mylogfile.txt
 password=mysupersecretpassword123"""
 
-    envelope = Envelope()
+    envelope = Envelope(headers={"event_id": "515539018c9b4260a6f999572f1661ee"})
     item = Item(
         payload=attachment, type="attachment", headers={"filename": "logfile.txt"}
     )
@@ -736,6 +737,53 @@ def test_event_with_attachment(
     assert event["event_id"] == event_id
 
 
+def test_attachment_without_event_id(
+    mini_sentry,
+    relay_with_processing,
+    outcomes_consumer,
+):
+    project_id = 42
+
+    mini_sentry.add_full_project_config(project_id)
+    outcomes_consumer = outcomes_consumer()
+
+    relay = relay_with_processing()
+
+    envelope = Envelope(headers=[])
+    envelope.add_item(
+        Item(
+            type="attachment",
+            payload=PayloadRef(bytes=b"event attachment"),
+            filename="event.txt",
+            content_type="text/plain",
+        )
+    )
+
+    relay.send_envelope(project_id, envelope)
+
+    outcomes = outcomes_consumer.get_aggregated_outcomes(n=2)
+    assert outcomes == [
+        {
+            "category": DataCategory.ATTACHMENT.value,
+            "key_id": 123,
+            "org_id": 1,
+            "outcome": 3,  # Invalid
+            "project_id": 42,
+            "quantity": 16,
+            "reason": "invalid_event_id",
+        },
+        {
+            "category": DataCategory.ATTACHMENT_ITEM.value,
+            "key_id": 123,
+            "org_id": 1,
+            "outcome": 3,  # Invalid
+            "project_id": 42,
+            "quantity": 1,
+            "reason": "invalid_event_id",
+        },
+    ]
+
+
 def test_form_data_is_rejected(
     mini_sentry, relay_with_processing, attachments_consumer, outcomes_consumer
 ):

tests/integration/test_envelope.py

@@ -92,7 +92,7 @@ def test_unknown_item(mini_sentry, relay):
     PROJECT_ID = 42
     mini_sentry.add_basic_project_config(PROJECT_ID)
 
-    envelope = Envelope()
+    envelope = Envelope(headers={"event_id": "d2132d31b39445f1938d7e21b6bf0ec4"})
     envelope.add_item(
         Item(payload=PayloadRef(bytes=b"something"), type="invalid_unknown")
     )
@@ -120,7 +120,7 @@ def test_drop_unknown_item(mini_sentry, relay):
     PROJECT_ID = 42
     mini_sentry.add_basic_project_config(PROJECT_ID)
 
-    envelope = Envelope()
+    envelope = Envelope(headers={"event_id": "d2132d31b39445f1938d7e21b6bf0ec4"})
     envelope.add_item(Item(payload=PayloadRef(bytes=b"something"), type="attachment"))
     envelope.add_item(
         Item(payload=PayloadRef(bytes=b"something"), type="invalid_unknown")

tests/integration/test_replay_events.py

@@ -122,7 +122,9 @@ def test_replay_events_without_processing(mini_sentry, relay_chain):
 
     replay_item = generate_replay_sdk_event()
 
-    relay.send_replay_event(42, replay_item)
+    relay.send_replay_event(
+        42, replay_item, envelope_headers={"event_id": replay_item["event_id"]}
+    )
 
     envelope = mini_sentry.get_captured_envelope(timeout=20)
     assert len(envelope.items) == 2
@@ -147,7 +149,9 @@ def test_replay_events_are_filtered(
     replay = generate_replay_sdk_event()
     replay["request"]["url"] = "http://localhost:1200"
 
-    relay.send_replay_event(42, replay)
+    relay.send_replay_event(
+        42, replay, envelope_headers={"event_id": replay["event_id"]}
+    )
 
     outcome = outcomes_consumer.get_outcome(timeout=10)
     assert outcome["org_id"] == 1
@@ -188,7 +192,9 @@ def test_time_corrections(mini_sentry, relay, delta, error):
     replay["timestamp"] = sdk_ts
     replay["replay_start_timestamp"] = sdk_start_ts
 
-    relay.send_replay_event(project_id, replay)
+    relay.send_replay_event(
+        42, replay, envelope_headers={"event_id": replay["event_id"]}
+    )
     if error == "past_timestamp":
         assert mini_sentry.get_aggregated_outcomes() == [
             {

tests/integration/test_user_report.py

@@ -13,17 +13,19 @@ def test_standalone_user_report(
     relay = relay_with_processing()
     mini_sentry.add_full_project_config(project_id)
 
+    event_id = "4cec9f3e1f214073b816e0f4de5f59b1"
+
     report_payload = {
         "name": "Josh",
         "email": "",
         "comments": "I'm having fun",
-        "event_id": "4cec9f3e1f214073b816e0f4de5f59b1",
+        "event_id": event_id,
     }
 
-    relay.send_user_report(
-        project_id,
-        report_payload,
-    )
+    envelope = Envelope(headers={"event_id": event_id})
+    envelope.add_item(Item(PayloadRef(json=report_payload), type="user_report"))
+
+    relay.send_envelope(project_id, envelope)
 
     report = attachments_consumer.get_user_report(timeout=5)
     assert json.loads(report["payload"]) == report_payload