fix(symbolicator): Refresh storage token on task resubmission

The storage token was minted once before entering the retry loop in
_process. On TaskIdNotFound or ServiceUnavailable the task is
resubmitted with the same token, which may have expired by then.

Add a kwargs_cb parameter to _process that is called on every new task
submission and merged over the static kwargs. The objectstore callers
use this to mint a fresh token each time instead of reusing the
original one.

Author: jan-auer

src/sentry/lang/native/symbolicator.py

@@ -114,11 +114,21 @@ def __init__(
         self.project = project
         self.event_id = event_id
 
-    def _process(self, task_name: str, path: str, **kwargs):
+    def _process(
+        self,
+        task_name: str,
+        path: str,
+        kwargs_cb: Callable[[], dict[str, Any]] | None = None,
+        **kwargs: Any,
+    ) -> Any:
         """
         This function will submit a symbolication task to a Symbolicator and handle
         polling it using the `SymbolicatorSession`.
         It will also correctly handle `TaskIdNotFound` and `ServiceUnavailable` errors.
+
+        `kwargs_cb`, if provided, is called on every new task submission and its result
+        is merged over `kwargs`. Use this for values that must be fresh on each
+        (re)submission, such as expiring tokens.
         """
         session = SymbolicatorSession(
             url=self.base_url,
@@ -137,7 +147,8 @@ def _process(self, task_name: str, path: str, **kwargs):
                 try:
                     if not task_id:
                         # We are submitting a new task to Symbolicator
-                        json_response = session.create_task(path, **kwargs)
+                        create_kwargs = {**kwargs, **(kwargs_cb() if kwargs_cb else {})}
+                        json_response = session.create_task(path, **create_kwargs)
                     else:
                         # The task has already been submitted to Symbolicator and we are polling
                         json_response = session.query_task(task_id)
@@ -190,7 +201,6 @@ def process_minidump(
         if minidump.stored_id:
             session = get_attachments_session(self.project.organization_id, self.project.id)
             storage_url = get_symbolicator_url(session, minidump.stored_id)
-            storage_token = session.mint_token()
             json: dict[str, Any] = {
                 "platform": platform,
                 "sources": sources,
@@ -199,11 +209,15 @@ def process_minidump(
                 "symbolicate": {
                     "type": "minidump",
                     "storage_url": storage_url,
-                    "storage_token": storage_token,
                     "rewrite_first_module": rewrite_first_module,
                 },
             }
-            res = self._process("process_minidump", "symbolicate-any", json=json)
+
+            def cb() -> dict[str, Any]:
+                json["symbolicate"]["storage_token"] = session.mint_token()
+                return {"json": json}
+
+            res = self._process("process_minidump", "symbolicate-any", kwargs_cb=cb)
             return process_response(res)
 
         data = {
@@ -237,7 +251,12 @@ def process_applecrashreport(self, platform: str, report: CachedAttachment):
                     "storage_token": storage_token,
                 },
             }
-            res = self._process("process_applecrashreport", "symbolicate-any", json=json)
+
+            def cb() -> dict[str, Any]:
+                json["symbolicate"]["storage_token"] = session.mint_token()
+                return {"json": json}
+
+            res = self._process("process_applecrashreport", "symbolicate-any", kwargs_cb=cb)
             return process_response(res)
 
         data = {