feat(processor): Playstation endpoint

Cargo.toml

@@ -124,6 +124,7 @@ json-forensics = "0.1.1"
 libc = "0.2.167"
 liblzma = "0.3.5"
 lru = "0.12.5"
+lz4_flex = "0.11.3"
 maxminddb = "0.24.0"
 memchr = "2.7.4"
 md5 = "0.7.0"
@@ -179,7 +180,7 @@ socket2 = "0.5.8"
 sqlparser = "0.44.0"
 sqlx = { version = "0.8.2", default-features = false }
 # See statsdproxy PR: https://github.com/getsentry/statsdproxy/pull/55
-statsdproxy = { git = "https://github.com/getsentry/statsdproxy", rev = "50155c5c5dc103fb1557e5ccde6b2cf6dfe82b9f", default-features = false}
+statsdproxy = { git = "https://github.com/getsentry/statsdproxy", rev = "50155c5c5dc103fb1557e5ccde6b2cf6dfe82b9f", default-features = false }
 symbolic-common = { version = "12.12.3", default-features = false }
 symbolic-unreal = { version = "12.12.3", default-features = false }
 syn = { version = "2.0.90" }

relay-dynamic-config/src/feature.rs

@@ -61,6 +61,11 @@ pub enum Feature {
     /// Serialized as `projects:relay-otel-endpoint`.
     #[serde(rename = "projects:relay-otel-endpoint")]
     OtelEndpoint,
+    /// Enable playstation crash dump ingestion via the `/playstation/` endpoint.
+    ///
+    /// Serialized as `project:relay-playstation-endpoint`.
+    #[serde(rename = "projects:relay-playstation-endpoint")]
+    PlaystationEndpoint,
     /// Discard transactions in a spans-only world.
     ///
     /// Serialized as `projects:discard-transaction`.

relay-server/Cargo.toml

@@ -52,6 +52,7 @@ itertools = { workspace = true }
 json-forensics = { workspace = true }
 libc = { workspace = true }
 liblzma = { workspace = true }
+lz4_flex = { workspace = true }
 mime = { workspace = true }
 minidump = { workspace = true, optional = true }
 multer = { workspace = true }

relay-server/src/endpoints/common.rs

@@ -63,6 +63,12 @@ pub enum BadStoreRequest {
     #[error("missing minidump")]
     MissingMinidump,
 
+    #[error("invalid prosperodump")]
+    InvalidProsperodump,
+
+    #[error("missing prosperodump")]
+    MissingProsperodump,
+
     #[error("invalid compression container")]
     InvalidCompressionContainer(#[source] std::io::Error),
 

relay-server/src/endpoints/mod.rs

@@ -15,6 +15,7 @@ mod health_check;
 mod minidump;
 mod monitor;
 mod nel;
+mod playstation;
 mod project_configs;
 mod public_keys;
 mod security_report;
@@ -74,6 +75,7 @@ pub fn routes(config: &Config) -> Router<ServiceState>{
         // No mandatory trailing slash here because people already use it like this.
         .route("/api/{project_id}/minidump", minidump::route(config))
         .route("/api/{project_id}/minidump/", minidump::route(config))
+        .route("/api/{project_id}/playstation/", playstation::route(config))
         .route("/api/{project_id}/events/{event_id}/attachments/", post(attachments::handle))
         .route("/api/{project_id}/unreal/{sentry_key}/", unreal::route(config))
         .route("/api/{project_id}/otlp/v1/traces/", traces::route(config))

relay-server/src/endpoints/playstation.rs

@@ -0,0 +1,153 @@
+use axum::extract::{DefaultBodyLimit, Request};
+use axum::response::IntoResponse;
+use axum::routing::{post, MethodRouter};
+use axum::RequestExt;
+use bytes::Bytes;
+use lz4_flex::frame::FrameDecoder as lz4Decoder;
+use multer::Multipart;
+use relay_config::Config;
+use relay_dynamic_config::Feature;
+use relay_event_schema::protocol::EventId;
+use std::io::Cursor;
+use std::io::Read;
+
+use crate::endpoints::common::{self, BadStoreRequest, TextResponse};
+use crate::envelope::ContentType::OctetStream;
+use crate::envelope::{AttachmentType, Envelope};
+use crate::extractors::{RawContentType, Remote, RequestMeta};
+use crate::service::ServiceState;
+use crate::utils;
+
+/// The extension of a prosperodump in the multipart form-data upload.
+const PROSPERODUMP_EXTENSION: &str = "prosperodmp";
+
+/// The extension of a screenshot in the multipart form-data upload.
+const SCREENSHOT_EXTENSION: &str = "jpg";
+
+/// The extension of a video in the multipart form-data upload.
+const VIDEO_EXTENSION: &str = "webm";
+
+/// The extension of a memorydump in the multipart form-data upload.
+const MEMORYDUMP_EXTENSION: &str = "prosperomemdmp";
+
+/// Prosperodump attachments should have these magic bytes
+const PROSPERODUMP_MAGIC_HEADER: &[u8] = b"\x7FELF";
+
+/// Magic bytes for lz4 compressed prosperodump containers.
+const LZ4_MAGIC_HEADER: &[u8] = b"\x04\x22\x4d\x18";
+
+fn validate_prosperodump(data: &[u8]) -> Result<(), BadStoreRequest> {
+    if !data.starts_with(PROSPERODUMP_MAGIC_HEADER) {
+        relay_log::trace!("invalid prosperodump file");
+        return Err(BadStoreRequest::InvalidProsperodump);
+    }
+
+    Ok(())
+}
+
+// TODO: Decide if we want to move this into a utils since it is duplicate.
+/// Convenience wrapper to let a decoder decode its full input into a buffer
+fn run_decoder(decoder: &mut Box<dyn Read>) -> std::io::Result<Vec<u8>> {
+    let mut buffer = Vec::new();
+    decoder.read_to_end(&mut buffer)?;
+    Ok(buffer)
+}
+
+// TODO: Decide if this extra function is worth it if we only have one encoder
+/// Creates a decoder based on the magic bytes the prosperodump payload
+fn decoder_from(prosperodump_data: Bytes) -> Option<Box<dyn Read>> {
+    if prosperodump_data.starts_with(LZ4_MAGIC_HEADER) {
+        return Some(Box::new(lz4Decoder::new(Cursor::new(prosperodump_data))));
+    }
+    None
+}
+
+/// Tries to decode a prosperodump using any of the supported compression formats
+/// or returns the provided minidump payload untouched if no format where detected
+fn decode_prosperodump(prosperodump_data: Bytes) -> Result<Bytes, BadStoreRequest> {
+    match decoder_from(prosperodump_data.clone()) {
+        Some(mut decoder) => {
+            match run_decoder(&mut decoder) {
+                Ok(decoded) => Ok(Bytes::from(decoded)),
+                Err(err) => {
+                    // we detected a compression container but failed to decode it
+                    relay_log::trace!("invalid compression container");
+                    Err(BadStoreRequest::InvalidCompressionContainer(err))
+                }
+            }
+        }
+        None => {
+            // this means we haven't detected any compression container
+            // TODO: Decide if we want to fail here
+            Ok(prosperodump_data)
+        }
+    }
+}
+
+fn infer_attachment_type(field_name: Option<&str>) -> AttachmentType {
+    match field_name.unwrap_or("") {
+        PROSPERODUMP_EXTENSION => AttachmentType::Prosperodump,
+        // TODO: Think about if we want these to be a special attachment type.
+        SCREENSHOT_EXTENSION | VIDEO_EXTENSION | MEMORYDUMP_EXTENSION | _ => {
+            AttachmentType::Attachment
+        }
+    }
+}
+
+async fn extract_multipart(
+    multipart: Multipart<'static>,
+    meta: RequestMeta,
+) -> Result<Box<Envelope>, BadStoreRequest> {
+    let mut items = utils::multipart_items_by_extension(multipart, infer_attachment_type).await?;
+
+    let prosperodump_item = items
+        .iter_mut()
+        .find(|item| item.attachment_type() == Some(&AttachmentType::Prosperodump))
+        .ok_or(BadStoreRequest::MissingProsperodump)?;
+
+    // TODO: Think about if we want a ContentType::Prosperodump ?
+    prosperodump_item.set_payload(
+        OctetStream,
+        decode_prosperodump(prosperodump_item.payload())?,
+    );
+
+    validate_prosperodump(&prosperodump_item.payload())?;
+
+    let event_id = common::event_id_from_items(&items)?.unwrap_or_else(EventId::new);
+    let mut envelope = Envelope::from_request(Some(event_id), meta);
+
+    for item in items {
+        envelope.add_item(item);
+    }
+
+    Ok(envelope)
+}
+
+async fn handle(
+    state: ServiceState,
+    meta: RequestMeta,
+    _content_type: RawContentType,
+    request: Request,
+) -> axum::response::Result<impl IntoResponse> {
+    // The crash dumps are transmitted as `...` in a multipart form-data/ request.
+    let Remote(multipart) = request.extract_with_state(&state).await?;
+    let mut envelope = extract_multipart(multipart, meta).await?;
+    envelope.require_feature(Feature::PlaystationEndpoint);
+
+    let id = envelope.event_id();
+
+    // Never respond with a 429 since clients often retry these
+    match common::handle_envelope(&state, envelope).await {
+        Ok(_) | Err(BadStoreRequest::RateLimited(_)) => (),
+        Err(error) => return Err(error.into()),
+    };
+
+    // Return here needs to be a 200 with arbitrary text to make the sender happy.
+    // TODO: Think about if there is something else to return here
+    Ok(TextResponse(id))
+}
+
+pub fn route(config: &Config) -> MethodRouter<ServiceState> {
+    // TODO: Check if this even has an effect since we will always have a multipart message.
+    post(handle).route_layer(DefaultBodyLimit::max(config.max_attachment_size()))
+}

relay-server/src/envelope.rs

@@ -415,6 +415,9 @@ pub enum AttachmentType {
     /// This attachment is processed by Relay immediately and never forwarded or persisted.
     Breadcrumbs,
 
+    // A prosperodump crash report (binary data)
+    Prosperodump,
+
     /// This is a binary attachment present in Unreal 4 events containing event context information.
     ///
     /// This can be deserialized using the `symbolic` crate see
@@ -452,6 +455,7 @@ impl fmt::Display for AttachmentType {
             AttachmentType::Minidump => write!(f, "event.minidump"),
             AttachmentType::AppleCrashReport => write!(f, "event.applecrashreport"),
             AttachmentType::EventPayload => write!(f, "event.payload"),
+            AttachmentType::Prosperodump => write!(f, "playstation.prosperodump"),
             AttachmentType::Breadcrumbs => write!(f, "event.breadcrumbs"),
             AttachmentType::UnrealContext => write!(f, "unreal.context"),
             AttachmentType::UnrealLogs => write!(f, "unreal.logs"),
@@ -935,6 +939,7 @@ impl Item {
                     AttachmentType::AppleCrashReport
                     | AttachmentType::Minidump
                     | AttachmentType::EventPayload
+                    | AttachmentType::Prosperodump
                     | AttachmentType::Breadcrumbs => true,
                     AttachmentType::Attachment
                     | AttachmentType::UnrealContext

relay-server/src/services/processor.rs

@@ -98,6 +98,7 @@ mod span;
 mod transaction;
 pub use span::extract_transaction_span;
 
+mod playstation;
 mod standalone;
 #[cfg(feature = "processing")]
 mod unreal;
@@ -1554,6 +1555,21 @@ impl EnvelopeProcessorService {
             unreal::expand(managed_envelope, &self.inner.config)?;
         });
 
+        // TODO: Check if this is the correct way of doing this. Or if we should go via
+        // should_filter.
+        if managed_envelope
+            .envelope()
+            .required_features()
+            .contains(&Feature::PlaystationEndpoint)
+            && !project_info.has_feature(Feature::PlaystationEndpoint)
+        {
+            managed_envelope.drop_items_silently();
+            return Ok(None);
+        }
+
+        // TODO: Would probably either want to "expand here" (guard with some flag)
+        playstation::expand(managed_envelope, &self.inner.config)?;
+
         let extraction_result = event::extract(
             managed_envelope,
             &mut metrics,
@@ -1568,6 +1584,12 @@ impl EnvelopeProcessorService {
             {
                 event_fully_normalized = inner_event_fully_normalized;
             }
+            // TODO: Process
+            if let Some(inner_event_fully_normalized) =
+                playstation::process(managed_envelope, &mut event)?
+            {
+                event_fully_normalized = inner_event_fully_normalized;
+            }
             if let Some(inner_event_fully_normalized) =
                 attachment::create_placeholders(managed_envelope, &mut event, &mut metrics)
             {

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

@@ -0,0 +1,33 @@
+//! Playstation related code.
+//!
+//! These functions are included only in the processing mode.
+
+use crate::envelope::{AttachmentType, ItemType};
+use crate::services::processor::{ErrorGroup, EventFullyNormalized, ProcessingError};
+use crate::utils::TypedEnvelope;
+use relay_config::Config;
+use relay_event_schema::protocol::Event;
+use relay_protocol::Annotated;
+
+pub fn expand(
+    managed_envelope: &mut TypedEnvelope<ErrorGroup>,
+    _config: &Config,
+) -> Result<(), ProcessingError> {
+    let envelope = &mut managed_envelope.envelope_mut();
+
+    if let Some(item) = envelope.take_item_by(|item| item.ty() == &ItemType::Attachment) {
+        if let Some(&AttachmentType::Prosperodump) = item.attachment_type() {
+            // TODO: Do some work here
+        }
+    }
+
+    Ok(())
+}
+
+// FIXME: Decide on weather we also want to keep the double function here to do the extraction work on the custom tags.
+pub fn process(
+    _managed_envelope: &mut TypedEnvelope<ErrorGroup>,
+    _event: &mut Annotated<Event>,
+) -> Result<Option<EventFullyNormalized>, ProcessingError> {
+    Ok(None)
+}