Repository navigation
Conversation
A pool can only execute tasks for the applications its workers serve. Nothing checked that, so a misrouted activation was immortal: in pull mode no worker ever claimed it and it sat Pending until the lag alert paged someone, and in push mode it was claimed, found no worker, had its claim expire, and was fetched again forever. Add an `applications` allowlist to broker config and discard anything outside it in ActivationBatcher::reduce, alongside the existing killswitch and expiry filters, publishing the payload to the pool's deadletter topic so a misconfiguration is recoverable rather than silent data loss. Push pools derive the set from `worker_map`, which already lists every application they can deliver to. Pull pools are never given a `worker_map`, so an unset `applications` admits everything and they keep running until ops opts them in; a pool whose `worker_map` names an application outside the allowlist now fails to start, since its tasks would be discarded before reaching that worker. Two supporting changes for activations that are already stored, which the consumer filter cannot reach: the push thread now sets Failure instead of returning silently, handing the task to upkeep's discard/deadletter path, and get_task returns failed_precondition for an unserved application so a worker pointed at the wrong pool fails instead of polling an empty queue. The producer swap and batch send in flush are extracted into helpers and shared with the forwarding path rather than copied. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The existing test stops at `reduce`, so the publish in `flush` was never exercised. Produce a discarded activation through a real broker and consume it back off the pool's deadletter topic, following the pattern upkeep uses for its own deadletter test. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
![sentry[bot]](https://avatars.githubusercontent.com/in/12637?s=60&v=4){kind=small}
…orrect-applications main replaced the batcher's raw FutureProducer with ProducerBackend, which also switches on the use_arroyo_producer runtime flag. Both of its swap blocks move into `producer_for`, which now takes the flag, and `send_all` produces through the backend. The runtime config read is hoisted so the deadletter batch can see the flag too. Also restores the ActivationStatus import in push/thread.rs, which main dropped when it reworked claim undo, and parameterizes the deadletter test over the Arroyo flag to match main's forwarding test.
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.
Reviewed by Cursor Bugbot for commit 7bf2633. Configure here.
- `worker_map` is only delivered through in push mode, and `from_args` gives pull pools a default entry they never read, so the subset check rejected a pull pool opting in to any other application. Scope the check to push. - `set_task_status` claims through the same path as `get_task`, so apply the check there too. It returns no task rather than an error, because an error makes the worker re-send the status update. - Discards and forwards bypass `batch`, so `is_full` never tripped on them and a misrouted topic only flushed on the timer. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
evanh
approved these changes
Oct 5, 2026
evanh
left a comment
evanh
left a comment
Member
There was a problem hiding this comment.
One small change. Also please ensure that any new metrics/changes to existing metrics are updated in our DD dashboards.
|
|
||
| // Same check as get_task. Returning an error here would make the worker | ||
| // re-send the status update, so an unserved application just gets no task. | ||
| if let Some(application) = application |
Member
There was a problem hiding this comment.
This check should not fail silently. It should log a metric and/or log an error.
The fetch-next path returned no task without any signal, so a worker pointed at the wrong pool looked identical to an idle queue. Both claim paths now emit `grpc_server.unserved_application` (tagged by method) and log the application. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
ShashSriv
deleted the
ShashSriv/stream-1807-discard-tasks-from-incorrect-applications
branch
October 5, 2026 21:47
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
A pool can only run tasks for the applications its workers serve. Nothing checked
that, so a misrouted task never died:
This pr allows the consumer to drop it before it's ever stored: metric, error log, and a copy
published to the pool's deadletter topic so it's recoverable. Tasks already in the
DB get the same treatment via the push thread setting
Failure, which hands them toupkeep's existing discard/deadletter path.
Push pools derive the allowed set from
worker_mapand are covered on deploy. Pullpools have no
worker_map, so they stay unrestricted until ops setsapplicationson them — follow-up PR in
ops. Nothing changes behaviour until then.