ref(replays): Always combine payloads and remove feature flag (#4812)

We want to always combine payloads.  This flag is always true.

Author: cmanallen

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+**Internal**:
+
+- Always combine replay payloads and remove feature flag guarding it. ([#4812](https://github.com/getsentry/relay/pull/4812))
+
 ## 25.6.0
 
 **Features**:

relay-dynamic-config/src/feature.rs

@@ -23,12 +23,6 @@ pub enum Feature {
     /// Serialized as `organizations:session-replay-recording-scrubbing`.
     #[serde(rename = "organizations:session-replay-recording-scrubbing")]
     SessionReplayRecordingScrubbing,
-    /// Enables combining session replay envelope items (Replay Recordings and Replay Events).
-    /// into one Kafka message.
-    ///
-    /// Serialized as `organizations:session-replay-combined-envelope-items`.
-    #[serde(rename = "organizations:session-replay-combined-envelope-items")]
-    SessionReplayCombinedEnvelopeItems,
     /// Disables select organizations from processing mobile replay events.
     ///
     /// Serialized as `organizations:session-replay-video-disabled`.

relay-server/src/envelope/item.rs

@@ -33,7 +33,6 @@ impl Item {
                 filename: None,
                 routing_hint: None,
                 rate_limited: false,
-                replay_combined_payload: false,
                 source_quantities: None,
                 other: BTreeMap::new(),
                 metrics_extracted: false,
@@ -282,17 +281,6 @@ impl Item {
         self.headers.source_quantities = Some(source_quantities);
     }
 
-    /// Returns if the payload's replay items should be combined into one kafka message.
-    #[cfg(feature = "processing")]
-    pub fn replay_combined_payload(&self) -> bool {
-        self.headers.replay_combined_payload
-    }
-
-    /// Sets the replay_combined_payload for this item.
-    pub fn set_replay_combined_payload(&mut self, combined_payload: bool) {
-        self.headers.replay_combined_payload = combined_payload;
-    }
-
     /// Returns the metrics extracted flag.
     pub fn metrics_extracted(&self) -> bool {
         self.headers.metrics_extracted
@@ -771,11 +759,6 @@ pub struct ItemHeaders {
     #[serde(default, skip)]
     rate_limited: bool,
 
-    /// Indicates that this item should be combined into one payload with other replay item.
-    /// NOTE: This is internal-only and not exposed into the Envelope.
-    #[serde(default, skip)]
-    replay_combined_payload: bool,
-
     /// Contains the amount of events this item was generated and aggregated from.
     ///
     /// A [metrics buckets](`ItemType::MetricBuckets`) item contains metrics extracted and

relay-server/src/services/processor/replay.rs

@@ -82,10 +82,6 @@ pub fn process(
     };
 
     for item in managed_envelope.envelope_mut().items_mut() {
-        if project_info.has_feature(Feature::SessionReplayCombinedEnvelopeItems) {
-            item.set_replay_combined_payload(true);
-        }
-
         match item.ty() {
             ItemType::ReplayEvent => {
                 let replay_event = handle_replay_event_item(item.payload(), &rpc)?;

relay-server/src/services/store.rs

@@ -279,10 +279,7 @@ impl StoreService {
                     replay_recording = Some(item);
                 }
                 ItemType::ReplayEvent => {
-                    if item.replay_combined_payload() {
-                        replay_event = Some(item);
-                    }
-
+                    replay_event = Some(item);
                     self.produce_replay_event(
                         event_id.ok_or(StoreError::NoEventId)?,
                         scoping.project_id,

tests/integration/test_replay_combined_payload.py

@@ -16,14 +16,7 @@ def test_replay_combined_with_processing(
     relay = relay_with_processing()
     mini_sentry.add_basic_project_config(
         project_id,
-        extra={
-            "config": {
-                "features": [
-                    "organizations:session-replay",
-                    "organizations:session-replay-combined-envelope-items",
-                ]
-            }
-        },
+        extra={"config": {"features": ["organizations:session-replay"]}},
     )
     replay_recordings_consumer = replay_recordings_consumer()
     replay_events_consumer = replay_events_consumer(timeout=10)
@@ -62,50 +55,3 @@ def test_replay_combined_with_processing(
     assert replay_event["type"] == "replay_event"
     assert replay_event["replay_id"] == replay_id
     assert replay_event_message["retention_days"] == 90
-
-
-def test_replay_combined_with_processing_no_flag_set(
-    mini_sentry, relay_with_processing, replay_recordings_consumer
-):
-    project_id = 42
-    replay_id = "515539018c9b4260a6f999572f1661ee"
-    relay = relay_with_processing()
-    mini_sentry.add_basic_project_config(
-        project_id,
-        extra={
-            "config": {
-                "features": [
-                    "organizations:session-replay",
-                ]
-            }
-        },
-    )
-    replay_recordings_consumer = replay_recordings_consumer()
-
-    envelope = Envelope(
-        headers=[
-            [
-                "event_id",
-                replay_id,
-            ],
-            ["attachment_type", "replay_recording"],
-        ]
-    )
-    payload = recording_payload(b"[]")
-    envelope.add_item(Item(payload=PayloadRef(bytes=payload), type="replay_recording"))
-
-    replay_event = generate_replay_sdk_event(replay_id=replay_id)
-    envelope.add_item(Item(payload=PayloadRef(json=replay_event), type="replay_event"))
-
-    relay.send_envelope(project_id, envelope)
-
-    replay_recording_message = replay_recordings_consumer.get_not_chunked_replay(
-        timeout=10
-    )
-
-    assert replay_recording_message["type"] == "replay_recording_not_chunked"
-    assert replay_recording_message["replay_id"] == replay_id
-
-    assert replay_recording_message["payload"] == payload
-
-    assert replay_recording_message["replay_event"] is None