feat(taskworker): Add Push Mode to Taskworker#576
Conversation
![sentry[bot]](https://avatars.githubusercontent.com/in/12637?s=60&v=4){kind=small}
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
evanh
left a comment
evanh
left a comment
There was a problem hiding this comment.
Still a couple bugs from the AI, but otherwise looks good!
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
Autofix Details
Bugbot Autofix prepared a fix for the issue found in the latest run.
- ✅ Fixed: Unused
_push_modeand_grpc_portinTaskbrokerClient- Removed the unused
self._push_modeandself._grpc_portassignments fromTaskbrokerClient.__init__, eliminating the dead stores without changing behavior.
- Removed the unused
Or push these changes by commenting:
@cursor push fb5751843e
Preview (fb5751843e)
diff --git a/clients/python/src/taskbroker_client/worker/client.py b/clients/python/src/taskbroker_client/worker/client.py
--- a/clients/python/src/taskbroker_client/worker/client.py
+++ b/clients/python/src/taskbroker_client/worker/client.py
@@ -147,8 +147,6 @@
self._hosts = hosts
self._rpc_secret = rpc_secret
self._metrics = metrics
- self._push_mode = push_mode
- self._grpc_port = grpc_port
self._grpc_options: list[tuple[str, Any]] = [
("grpc.max_receive_message_length", MAX_ACTIVATION_SIZE)This Bugbot Autofix run was free. To enable autofix for future PRs, go to the Cursor dashboard.
| mock_metrics = mock.MagicMock() | ||
| taskworker._metrics = mock_metrics | ||
| mock_queue = mock.MagicMock() | ||
| mock_queue.full.return_value = False | ||
| taskworker._child_tasks = mock_queue |
There was a problem hiding this comment.
Should we have optional constructor parameters on Taskworker to make it easier to pass mocks in?
| """ | ||
| Records a gauge metric (a point-in-time value). | ||
| """ | ||
| raise NotImplementedError |
There was a problem hiding this comment.
We'll have to be careful about implementing this when sentry's client library version is updated. I don't think there are any tests in sentry that will fail because of this new abstract method.
There was a problem hiding this comment.
Sentry (and launchpad) now have implementations of this MetricsBackend but the won't have this method defined. When the worker runtime calls metrics.gauge() it will raise as the applications won't have implemented this abstract method.
When we update applications to the new client library release, we have to remember to implement this method, or the push worker will be broken.
evanh
left a comment
evanh
left a comment
There was a problem hiding this comment.
Except for those bugs around auth I think this PR looks good!
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
| if getattr(_RPC_SIGNATURE_AUTH_TLS, "failed", False): | ||
| context.abort(grpc.StatusCode.UNAUTHENTICATED, "Authentication failed") | ||
|
|
||
| return original(request, context) |
There was a problem hiding this comment.
Auth bypass: context.abort() doesn't stop handler execution
High Severity
In gRPC Python's synchronous server, context.abort() does not raise an exception — it sends the error to the client but execution continues (this is a known, unresolved gRPC Python bug: grpc/grpc#30306, #37518). In RequestSignatureServerInterceptor, when HMAC verification fails, context.abort(UNAUTHENTICATED) is called but original(request, context) still executes on the next line, passing the (empty-deserialized) request to the real PushTask handler. This effectively bypasses authentication, allowing unauthenticated requests to enqueue tasks. The same pattern in WorkerServicer.PushTask is less severe but similarly relies on abort() halting execution. A return statement is needed after each context.abort() call.
Additional Locations (1)
There was a problem hiding this comment.
Pretty sure this is not a problem. But something to remember for the future.
george-sentry
deleted the
george/push-taskbroker/add-push-mode-to-taskworker
branch
3 participants
Linear
Completes STREAM-822
Description
Currently, taskworkers pull tasks from taskbrokers via RPC. This approach works, but has some drawbacks. Therefore, we want taskbrokers to push tasks to taskworkers instead. Read this page on Notion for more information.
This PR allows users to run the taskworker in push mode via the
--push-modeand--grpc-portCLI options.--push-modeboolFalse--grpc-portint50052Details
Dependencies
sentry-protosfrom 0.4.11 to 0.8.5 (to use the new worker service schema)Additions
WorkerServicerclass inworker.pypush_modeandgrpc_portfields toTaskWorkerandTaskbrokerClientclassespush_taskmethod toTaskWorkerclassgaugemethod to metrics backend abstract classModifications
_get_stubmethod toTaskbrokerClientto connect to brokers on the fly by using the activation callback URL