feat(taskworker) Add concurrent worker

src/sentry/runner/commands/run.py

@@ -242,6 +242,7 @@ def worker(ignore_unknown_queues: bool, **options: Any) -> None:
 @click.option(
     "--max-task-count", help="Number of tasks this worker should run before exiting", default=10000
 )
+@click.option("--concurrency", help="Number of child worker processes to create.", default=1)
 @click.option(
     "--namespace", help="The dedicated task namespace that taskworker operates on", default=None
 )
@@ -258,7 +259,7 @@ def taskworker(**options: Any) -> None:
 
 
 def run_taskworker(
-    rpc_host: str, max_task_count: int, namespace: str | None, **options: Any
+    rpc_host: str, max_task_count: int, namespace: str | None, concurrency: int, **options: Any
 ) -> None:
     """
     taskworker factory that can be reloaded
@@ -267,7 +268,11 @@ def run_taskworker(
 
     with managed_bgtasks(role="taskworker"):
         worker = TaskWorker(
-            rpc_host=rpc_host, max_task_count=max_task_count, namespace=namespace, **options
+            rpc_host=rpc_host,
+            max_task_count=max_task_count,
+            namespace=namespace,
+            concurrency=concurrency,
+            **options,
         )
         exitcode = worker.start()
         raise SystemExit(exitcode)

src/sentry/taskworker/client.py

@@ -44,7 +44,10 @@ def get_task(self, namespace: str | None = None) -> TaskActivation | None:
         return None
 
     def update_task(
-        self, task_id: str, status: TaskActivationStatus.ValueType, fetch_next_task: FetchNextTask
+        self,
+        task_id: str,
+        status: TaskActivationStatus.ValueType,
+        fetch_next_task: FetchNextTask | None = None,
     ) -> TaskActivation | None:
         """
         Update the status for a given task activation.