ref(spanv2): Disallow multiple span containers (#5247)

Already not allowed in the other span processing path and consistent
with other items, like logs.

Author: Dav1dde

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

@@ -137,7 +137,7 @@ impl processing::Processor for LogsProcessor {
         mut logs: Managed<Self::UnitOfWork>,
         ctx: Context<'_>,
     ) -> Result<Output<Self::Output>, Rejected<Error>> {
-        validate::container(&logs)?;
+        validate::container(&logs).reject(&logs)?;
 
         if ctx.is_proxy() {
             // If running in proxy mode, just apply cached rate limits and forward without

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

@@ -1,5 +1,5 @@
-use crate::processing::logs::{Error, SerializedLogs};
-use crate::processing::{Managed, Rejected};
+use crate::processing::Managed;
+use crate::processing::logs::{Error, Result, SerializedLogs};
 
 /// Validates that there is only a single log container processed at a time.
 ///
@@ -10,10 +10,10 @@ use crate::processing::{Managed, Rejected};
 ///
 /// This limit mostly exists to incentivise SDKs to batch multiple logs into a single container,
 /// technically it can be removed without issues.
-pub fn container(logs: &Managed<SerializedLogs>) -> Result<(), Rejected<Error>> {
+pub fn container(logs: &Managed<SerializedLogs>) -> Result<()> {
     // It's fine if there was no log container, as we still accept OTel logs.
     if logs.logs.len() > 1 {
-        return Err(logs.reject_err(Error::DuplicateContainer));
+        return Err(Error::DuplicateContainer);
     }
 
     Ok(())

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

@@ -22,11 +22,15 @@ mod integrations;
 mod process;
 #[cfg(feature = "processing")]
 mod store;
+mod validate;
 
 type Result<T, E = Error> = std::result::Result<T, E>;
 
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
+    /// Multiple item containers for spans in a single envelope are not allowed.
+    #[error("duplicate span container")]
+    DuplicateContainer,
     /// Standalone spans filtered because of a missing feature flag.
     #[error("spans feature flag missing")]
     FilterFeatureFlag,
@@ -49,6 +53,7 @@ impl OutcomeError for Error {
 
     fn consume(self) -> (Option<Outcome>, Self::Error) {
         let outcome = match &self {
+            Self::DuplicateContainer => Some(Outcome::Invalid(DiscardReason::DuplicateItem)),
             Self::FilterFeatureFlag => None,
             Self::Filtered(f) => Some(Outcome::Filtered(f.clone())),
             Self::RateLimited(limits) => {
@@ -97,7 +102,7 @@ impl processing::Processor for SpansProcessor {
 
         let spans = envelope
             .envelope_mut()
-            .take_items_by(|item| matches!(*item.ty(), ItemType::Span))
+            .take_items_by(ItemContainer::<SpanV2>::is_container)
             .into_vec();
 
         let integrations = envelope
@@ -119,6 +124,7 @@ impl processing::Processor for SpansProcessor {
         ctx: Context<'_>,
     ) -> Result<Output<Self::Output>, Rejected<Self::Error>> {
         filter::feature_flag(ctx).reject(&spans)?;
+        validate::container(&spans).reject(&spans)?;
 
         if ctx.is_proxy() {
             // If running in proxy mode, just apply cached rate limits and forward without

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

@@ -0,0 +1,19 @@
+use crate::managed::Managed;
+use crate::processing::spans::{Error, Result, SerializedSpans};
+
+/// Validates that there is only a single spans container processed at a time.
+///
+/// Currently it is only allowed to send a single span container in an envelope.
+/// Since all spans of the same envelope must belong to the same trace (due to the dynamic sampling
+/// context on the envelope), they also should be collapsed by SDKs into a single container.
+///
+/// Once we lift the requirement of a single trace per envelope, we may want to also consider
+/// lifting this restriction.
+pub fn container(spans: &Managed<SerializedSpans>) -> Result<()> {
+    // It's fine if there was no container, as we still accept OTel spans.
+    if spans.spans.len() > 1 {
+        return Err(Error::DuplicateContainer);
+    }
+
+    Ok(())
+}

tests/integration/test_spansv2.py

@@ -532,3 +532,73 @@ def test_spanv2_inbound_filters(
     ]
 
     assert mini_sentry.captured_events.empty()
+
+
+def test_spans_v2_multiple_containers_not_allowed(
+    mini_sentry,
+    relay,
+):
+    project_id = 42
+    project_config = mini_sentry.add_full_project_config(project_id)
+    project_config["config"]["features"] = [
+        "organizations:standalone-span-ingestion",
+        "projects:span-v2-experimental-processing",
+    ]
+
+    relay = relay(mini_sentry, options=TEST_CONFIG)
+    start = datetime.now(timezone.utc)
+    envelope = Envelope()
+
+    payload = {
+        "start_timestamp": start.timestamp(),
+        "end_timestamp": start.timestamp() + 0.500,
+        "trace_id": "5b8efff798038103d269b633813fc60c",
+        "span_id": "eee19b7ec3c1b175",
+        "name": "some op",
+        "is_remote": False,
+        "status": "ok",
+    }
+    envelope.add_item(
+        Item(
+            type="span",
+            payload=PayloadRef(json={"items": [payload]}),
+            content_type="application/vnd.sentry.items.span.v2+json",
+            headers={"item_count": 1},
+        )
+    )
+    envelope.add_item(
+        Item(
+            type="span",
+            payload=PayloadRef(json={"items": [payload, payload]}),
+            content_type="application/vnd.sentry.items.span.v2+json",
+            headers={"item_count": 2},
+        )
+    )
+
+    relay.send_envelope(project_id, envelope)
+
+    assert mini_sentry.get_outcomes(2) == [
+        {
+            "category": DataCategory.SPAN.value,
+            "timestamp": time_within_delta(),
+            "key_id": 123,
+            "org_id": 1,
+            "outcome": 3,  # Invalid
+            "project_id": 42,
+            "quantity": 3,
+            "reason": "duplicate_item",
+        },
+        {
+            "category": DataCategory.SPAN_INDEXED.value,
+            "timestamp": time_within_delta(),
+            "key_id": 123,
+            "org_id": 1,
+            "outcome": 3,  # Invalid
+            "project_id": 42,
+            "quantity": 3,
+            "reason": "duplicate_item",
+        },
+    ]
+
+    assert mini_sentry.captured_events.empty()
+    assert mini_sentry.captured_outcomes.empty()