feat(replays): Add ReplayVideo envelope item type (#3105)

Add support for replay-video event types.

Changes:

- Refactors event validation in the processor. This is to DRY the
processing logic between the three replay event types.
- Adds new replay_video field to replay-recording Kafka message.
- Add outcomes.
- Configure rate-limits.

Relates to:
 * getsentry/sentry#63255

---------

Co-authored-by: Joshua Ferge <[email protected]>
Co-authored-by: Josh Ferge <[email protected]>
Co-authored-by: Joris Bayer <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Extend GPU context with data for Unreal Engine crash reports. ([#3144](https://github.com/getsentry/relay/pull/3144))
 - Parametrize transaction in dynamic sampling context. ([#3141](https://github.com/getsentry/relay/pull/3141))
+- Adds ReplayVideo envelope-item type. ([#3105](https://github.com/getsentry/relay/pull/3105))
 - Parse & scrub span description for supabase. ([#3153](https://github.com/getsentry/relay/pull/3153), [#3156](https://github.com/getsentry/relay/pull/3156))
 
 **Bug Fixes**:

Cargo.lock

@@ -10,6 +10,7 @@ license-file = "../LICENSE.md"
 publish = false
 
 [dependencies]
+bytes = { version = "1.4.0" }
 flate2 = "1.0.19"
 once_cell = { workspace = true }
 relay-common = { path = "../relay-common" }

relay-replays/Cargo.toml

@@ -23,7 +23,6 @@ processing = [
     "dep:symbolic-common",
     "dep:symbolic-unreal",
     "dep:zstd",
-    "bytes/serde",
     "relay-cardinality/redis",
     "relay-config/processing",
     "relay-kafka/producer",
@@ -47,7 +46,7 @@ axum-server = "0.4.7"
 backoff = "0.4.0"
 brotli = "3.3.4"
 bytecount = "0.6.0"
-bytes = { version = "1.4.0" }
+bytes = { version = "1.4.0", features = ["serde"] }
 chrono = { workspace = true, features = ["clock"] }
 data-encoding = "2.3.3"
 flate2 = "1.0.19"
@@ -97,6 +96,7 @@ reqwest = { version = "0.11.1", features = [
 rmp-serde = "1.1.1"
 rust-embed = { version = "8.0.0", optional = true }
 serde = { workspace = true }
+serde_bytes = "0.11"
 serde_json = { workspace = true }
 smallvec = { workspace = true, features = ["drain_filter"] }
 sqlx = { version = "0.7.3", features = [

relay-server/Cargo.toml

@@ -114,6 +114,8 @@ pub enum ItemType {
     ReplayEvent,
     /// Replay Recording data.
     ReplayRecording,
+    /// Replay Video data.
+    ReplayVideo,
     /// Monitor check-in encoded as JSON.
     CheckIn,
     /// A standalone span.
@@ -168,6 +170,7 @@ impl ItemType {
             Self::Profile => "profile",
             Self::ReplayEvent => "replay_event",
             Self::ReplayRecording => "replay_recording",
+            Self::ReplayVideo => "replay_video",
             Self::CheckIn => "check_in",
             Self::Span => "span",
             Self::OtelSpan => "otel_span",
@@ -222,6 +225,7 @@ impl std::str::FromStr for ItemType {
             "profile" => Self::Profile,
             "replay_event" => Self::ReplayEvent,
             "replay_recording" => Self::ReplayRecording,
+            "replay_video" => Self::ReplayVideo,
             "check_in" => Self::CheckIn,
             "span" => Self::Span,
             "otel_span" => Self::OtelSpan,
@@ -661,7 +665,9 @@ impl Item {
             } else {
                 DataCategory::Profile
             }),
-            ItemType::ReplayEvent | ItemType::ReplayRecording => Some(DataCategory::Replay),
+            ItemType::ReplayEvent | ItemType::ReplayRecording | ItemType::ReplayVideo => {
+                Some(DataCategory::Replay)
+            }
             ItemType::ClientReport => None,
             ItemType::CheckIn => Some(DataCategory::Monitor),
             ItemType::Span | ItemType::OtelSpan => Some(if indexed {
@@ -873,6 +879,7 @@ impl Item {
             | ItemType::ClientReport
             | ItemType::ReplayEvent
             | ItemType::ReplayRecording
+            | ItemType::ReplayVideo
             | ItemType::Profile
             | ItemType::CheckIn
             | ItemType::Span
@@ -907,6 +914,7 @@ impl Item {
             ItemType::MetricMeta => false,
             ItemType::ClientReport => false,
             ItemType::ReplayRecording => false,
+            ItemType::ReplayVideo => false,
             ItemType::Profile => true,
             ItemType::CheckIn => false,
             ItemType::Span => false,
@@ -1765,6 +1773,22 @@ mod tests {
         assert_eq!(items[0].ty(), &ItemType::ReplayRecording);
     }
 
+    #[test]
+    fn test_deserialize_envelope_replay_video() {
+        let bytes = Bytes::from(
+            "\
+             {\"event_id\":\"9ec79c33ec9942ab8353589fcb2e04dc\",\"dsn\":\"https://e12d836b15bb49d7bbf99e64295d995b:@sentry.io/42\"}\n\
+             {\"type\":\"replay_video\"}\n\
+             helloworld\n\
+             ",
+        );
+
+        let envelope = Envelope::parse_bytes(bytes).unwrap();
+        assert_eq!(envelope.len(), 1);
+        let items: Vec<_> = envelope.items().collect();
+        assert_eq!(items[0].ty(), &ItemType::ReplayVideo);
+    }
+
     #[test]
     fn test_deserialize_envelope_view_hierarchy() {
         let bytes = Bytes::from(

relay-server/src/envelope.rs

@@ -366,6 +366,7 @@ pub enum DiscardReason {
     InvalidReplayEventNoPayload,
     InvalidReplayEventPii,
     InvalidReplayRecordingEvent,
+    InvalidReplayVideoEvent,
 
     /// (Relay) Profiling related discard reasons
     Profiling(&'static str),
@@ -413,6 +414,7 @@ impl DiscardReason {
             DiscardReason::InvalidReplayEventNoPayload => "invalid_replay_no_payload",
             DiscardReason::InvalidReplayEventPii => "invalid_replay_pii_scrubber_failed",
             DiscardReason::InvalidReplayRecordingEvent => "invalid_replay_recording",
+            DiscardReason::InvalidReplayVideoEvent => "invalid_replay_video",
             DiscardReason::Profiling(reason) => reason,
             DiscardReason::InvalidSpan => "invalid_span",
         }

relay-server/src/services/outcome.rs

@@ -226,7 +226,7 @@ impl ProcessingGroup {
         let replay_items = envelope.take_items_by(|item| {
             matches!(
                 item.ty(),
-                &ItemType::ReplayEvent | &ItemType::ReplayRecording
+                &ItemType::ReplayEvent | &ItemType::ReplayRecording | &ItemType::ReplayVideo
             )
         });
         if !replay_items.is_empty() {

relay-server/src/services/processor.rs

@@ -464,6 +464,7 @@ fn is_duplicate(item: &Item, processing_enabled: bool) -> bool {
         ItemType::Profile => false,
         ItemType::ReplayEvent => false,
         ItemType::ReplayRecording => false,
+        ItemType::ReplayVideo => false,
         ItemType::CheckIn => false,
         ItemType::Span => false,
         ItemType::OtelSpan => false,