perf: change relay payload to reduce json size (#69916) (#69990)

Since the incident is resolved, we can now revert the revert.

Let's wait for the Snuba deployment to be complete to be `extra` safe.
I'm opening the PR now, to give time for @getsentry/data and
@getsentry/owners-snuba @getsentry/snuba-maintainers teams to look into
this before landing.

Aiming to land this on Wednesday (May 1st)

src/sentry/replays/lib/event_linking.py

@@ -5,8 +5,6 @@ import uuid
 from hashlib import md5
 from typing import TYPE_CHECKING, TypedDict, Union
 
-from sentry.utils import json
-
 if TYPE_CHECKING:
     from sentry.eventstore.models import BaseEvent
 
@@ -17,7 +15,9 @@ class EventLinkKafkaMessage(TypedDict):
     replay_id: str
     project_id: int
     segment_id: None
-    payload: list[int]
+    payload: list[
+        int
+    ] | EventLinkPayloadDebugId | EventLinkPayloadInfoId | EventLinkPayloadWarningId | EventLinkPayloadErrorId | EventLinkPayloadFatalId
     retention_days: int
 
 
@@ -133,7 +133,7 @@ def transform_event_for_linking_payload(replay_id: str, event: BaseEvent) -> Eve
         "project_id": event.project.id,
         "segment_id": None,
         "retention_days": 90,
-        "payload": list(bytes(json.dumps(_make_json_binary_payload()).encode())),
+        "payload": _make_json_binary_payload(),
     }
 
 

src/sentry/tasks/post_process.py

@@ -1005,11 +1005,11 @@ def process_replay_link(job: PostProcessJob) -> None:
         kafka_payload = transform_event_for_linking_payload(replay_id, group_event)
     except ValueError:
         metrics.incr("post_process.process_replay_link.id_invalid")
-
-    publisher.publish(
-        "ingest-replay-events",
-        json.dumps(kafka_payload),
-    )
+    else:
+        publisher.publish(
+            "ingest-replay-events",
+            json.dumps(kafka_payload),
+        )
 
 
 def process_rules(job: PostProcessJob) -> None:

tests/sentry/tasks/test_post_process.py

@@ -1935,11 +1935,7 @@ class ReplayLinkageTestMixin(BasePostProgressGroupMixin):
         assert ret_value["project_id"] == self.project.id
         assert ret_value["segment_id"] is None
         assert ret_value["retention_days"] == 90
-
-        # convert ret_value_payload which is a list of bytes to a string
-        ret_value_payload = json.loads(bytes(ret_value["payload"]).decode("utf-8"))
-
-        assert ret_value_payload == {
+        assert ret_value["payload"] == {
             "type": "event_link",
             "replay_id": replay_id,
             "error_id": event.event_id,
@@ -1974,11 +1970,7 @@ class ReplayLinkageTestMixin(BasePostProgressGroupMixin):
         assert ret_value["project_id"] == self.project.id
         assert ret_value["segment_id"] is None
         assert ret_value["retention_days"] == 90
-
-        # convert ret_value_payload which is a list of bytes to a string
-        ret_value_payload = json.loads(bytes(ret_value["payload"]).decode("utf-8"))
-
-        assert ret_value_payload == {
+        assert ret_value["payload"] == {
             "type": "event_link",
             "replay_id": replay_id,
             "error_id": event.event_id,