Bulk cancel jobs by filter (tags/batch_id/queue)

[rcbevans]

I tag jobs per tenant when enqueueing (EnqueueItem.tags), and when a tenant needs to be offboarded — or a particular run needs aborting — I want to cancel everything matching that tag. The only cancel surface today is per-job: JobsClient.cancel(job_id), TaskQ.cancel(job_id), JobHandle.cancel(). So the operator path is paginate-and-cancel:

cursor = None
while True:
    page = await client.list(
        JobFilter(tags=("tenant:acme",), active=True, limit=100, cursor=cursor)
    )
    for row in page.jobs:
        await client.cancel(row.id, reason="tenant offboarded")
    if page.next_cursor is None:
        break
    cursor = page.next_cursor

Two problems. First, it's slow for large tenants — one round trip per job on top of the page reads. Second, it's racy: workers keep dispatching jobs from pages I haven't reached yet, and nothing stops new matching jobs from landing behind my cursor, so "cancel everything for this tenant" needs an outer retry loop to actually converge. The fleet is actively working against me while I iterate.

What I'd rather write:

result = await client.cancel_where(
    JobFilter(tags=("tenant:acme",), active=True),
    reason="tenant offboarded",
)

i.e. one set-based write that issues cancel requests (phase 1, cooperative) for every row matching the filter, returning counts and maybe the affected ids for observability. The per-row semantics already exist in write_cancel_request and the cancellation state machine; this is the same write over a filter instead of a single id.

Open questions from my side: should matching rows in pending/scheduled go straight to terminal cancelled instead of through the phase-1 request (they're not running, so there's nothing to cooperate with)? And is a guardrail wanted — e.g. refusing a filter with no predicates so you can't cancel the whole table by accident?

Is a bulk-cancel primitive something you'd consider, or is the paginated loop the intended answer? Happy to be told I'm missing an existing escape hatch.

[rcbevans]

Two clarifications after building on this.

A tag filter under-selects, so bulk-cancel-by-tag would be incomplete. tags can only be set from JobsClient.enqueue and EnqueueItem; SubJobEnqueuer.enqueue has no tags parameter and sub-jobs don't inherit the parent's (#57). So in any pipeline where actors chain by enqueueing the next stage from inside the handler, those jobs carry no tags and a cancel_where(JobFilter(tags=...)) would silently skip exactly the jobs that advance the pipeline — cancelling the fanned-out work while leaving the continuation to fire. Whichever of the two lands first, the pair is what makes "cancel this run" a single operation; on its own this one is still useful for batch_id and queue filters.

Sending pending/scheduled straight to cancelled matters more than it looks. You floated it as an open question; from the outside it's the difference between converging and not. The race in the paginated loop is almost entirely against the dispatcher promoting rows out from under the cursor — a row that's already running needs the cooperative phase, but a row that hasn't been picked up yet is exactly the one that will be picked up while I'm still paginating. Terminating those in the same set-based write is what makes one call sufficient instead of needing an outer retry loop.

On the guardrail question: refusing a filter with no predicates seems right, and active=True alone probably shouldn't count as a predicate — "cancel everything not yet finished, fleet-wide" is the accident worth protecting against, and it's the filter someone reaches for while debugging.

A returned count of (requested, already_terminal, transitioned) rather than a single number would help; "I asked to cancel 400 and 12 had already finished" is the normal case and worth being able to distinguish from a partial failure.