feat: bulk cancel by filter + sub-job tags (#54, #57)

CHANGELOG.md

@@ -5,6 +5,30 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+* `JobsClient.cancel_where(filter, reason)` — bulk cancel all jobs matching a
+  `JobFilter` in a single set-based operation. Pending/scheduled jobs go straight
+  to terminal `cancelled`; running jobs get cooperative cancel (`cancel_phase=1`).
+  Returns `BulkCancelResult` with counts and affected IDs. Empty filters are
+  rejected with `EmptyFilterError` unless `allow_empty_filter=True` is passed.
+* `SubJobEnqueuer.enqueue()` now accepts `tags`, `inherit_tags`,
+  `schedule_to_close`, `start_to_close`, and `heartbeat_timeout` parameters.
+  Sub-jobs inherit the parent job's tags by default (`inherit_tags=True`); pass
+  `inherit_tags=False` to suppress inheritance for a specific sub-job.
+* `BulkCancelResult` and `EmptyFilterError` exported from `taskq` top-level.
+
+### Changed
+
+* **Sub-jobs now inherit parent tags by default.** Every `ctx.jobs.enqueue()`
+  call inside an actor body now propagates the parent job's tags to the sub-job,
+  making sub-jobs findable by `JobFilter(tags=...)` and cancellable by
+  `cancel_where`. Pass `inherit_tags=False` per-call to opt out. This is a
+  behavior change for any code that relied on sub-job tags being empty —
+  inherited tags make sub-jobs visible to tag-based filters and bulk cancels.
+
 ## [0.2.2](https://github.com/AZX-PBC-OSS/TaskQ/compare/v0.2.1...v0.2.2) (2026-07-22)
 
 

README.md

@@ -47,7 +47,9 @@ Async-native, Postgres-backed background job library for Python 3.12+.
   isn't inside an actor should instead poll `BatchHandle.status(db_connection)`.
   See [Jobs & Clients](docs/guides/jobs-clients.md#enqueue_batch).
 - **Cancellation** — cooperative cancellation with grace periods and
-  force-cancel sweeps; `ctx.check_cancelled()` inside actor bodies.
+  force-cancel sweeps; `ctx.check_cancelled()` inside actor bodies; bulk
+  `cancel_where(filter)` for set-based cancellation by tag, queue, actor,
+  or batch ID.
 - **Progress tracking** — `ctx.progress(...)` events buffered and published
   to subscribers and the admin UI.
 - **Workgroups** — multi-worker process supervision with a shared heartbeat

docs/architecture.md

@@ -102,6 +102,7 @@ class Backend(Protocol):
 
     # Cancel signals
     async def write_cancel_request(self, job_id, reason) -> bool: ...
+    async def cancel_where(self, filter: JobFilter, reason: str | None) -> BulkCancelResult: ...
     async def poll_cancel_flags(self, worker_id) -> list[CancelFlag]: ...
 
     # Admin operations

docs/guides/actors.md

@@ -689,6 +689,13 @@ INSERTs are part of the parent's database transaction:
 This is the correct default for fan-out patterns where sub-jobs should only exist if the parent
 completes successfully.
 
+**Tag inheritance.** Sub-jobs enqueued via `ctx.jobs.enqueue()` inherit the parent
+job's tags by default. This makes sub-jobs findable by `JobFilter(tags=...)` and
+cancellable by `cancel_where`. Pass `inherit_tags=False` to suppress inheritance
+for a specific sub-job, or pass explicit `tags=[...]` to merge with inherited
+tags. See [Jobs & Clients — Sub-job tag inheritance](jobs-clients.md#tag-inheritance)
+for the full semantics table.
+
 **Autonomous fallback.** If no LOOP-scope `asyncpg.Connection` is registered in the DI
 container, `ctx.jobs.enqueue()` falls back to the worker pool and commits each INSERT
 independently. In this mode, sub-jobs are persisted even if the parent subsequently raises

docs/guides/cancellation.md

@@ -30,6 +30,22 @@ result = await handle.cancel(reason="deadline exceeded")
 
 `JobHandle.cancel()` delegates directly to `JobsClient.cancel(handle.job_id, reason)`. The handle must have been constructed with a `JobsClient` (i.e. via `client.enqueue()` or `client.get()`); handles obtained from inside an actor body via `ctx.jobs.enqueue()` do not have a client and will raise `RuntimeError`.
 
+### Via `JobsClient.cancel_where()`
+
+```python
+from taskq import JobFilter
+
+result = await client.cancel_where(
+    JobFilter(tags=("tenant-acme",), active=True),
+    reason="tenant offboarded",
+)
+```
+
+`cancel_where()` cancels all jobs matching a `JobFilter` in a single set-based SQL
+operation. Pending/scheduled jobs go straight to terminal `cancelled`; running jobs get
+`cancel_phase=1` (cooperative cancel). Returns a `BulkCancelResult` with counts and
+affected IDs. See [jobs-clients.md](jobs-clients.md#cancel_where) for the full API.
+
 ### Effect by prior status
 
 | Prior status | Effect |

docs/guides/jobs-clients.md

@@ -52,13 +52,14 @@ Terminal statuses (`succeeded`, `failed`, `cancelled`, `crashed`, `abandoned`) h
 6. [`JobHandle[R]`](#jobhandler)
 7. [`get()`](#get)
 8. [`cancel()`](#cancel)
-9. [`list()`](#list)
-10. [`SubJobEnqueuer`](#subjobenqueuer)
-11. [Error handling](#error-handling)
-12. [Full enqueue-and-wait example](#full-enqueue-and-wait-example)
-13. [Idempotency example](#idempotency-example)
-14. [Batch enqueue example](#batch-enqueue-example)
-15. [Tags](#tags)
+9. [`cancel_where()`](#cancel_where)
+10. [`list()`](#list)
+11. [`SubJobEnqueuer`](#subjobenqueuer)
+12. [Error handling](#error-handling)
+13. [Full enqueue-and-wait example](#full-enqueue-and-wait-example)
+14. [Idempotency example](#idempotency-example)
+15. [Batch enqueue example](#batch-enqueue-example)
+16. [Tags](#tags)
 
 ---
 
@@ -636,6 +637,63 @@ else:
 
 ---
 
+## `cancel_where()`
+
+```python
+async def cancel_where(
+    self,
+    filter: JobFilter,
+    reason: str | None = None,
+    *,
+    allow_empty_filter: bool = False,
+) -> BulkCancelResult: ...
+```
+
+Cancel all jobs matching `filter` in a single set-based operation. Pending/scheduled
+jobs go straight to terminal `cancelled`; running jobs get `cancel_phase=1` (cooperative
+cancel) — the worker's heartbeat observes the phase change and sets the in-process
+`cancel_event` at the next tick.
+
+**Guardrail:** a filter with no predicates (no `queue`, `status`, `actor`,
+`identity_key`, `batch_id`, `tags`, or `active`) is rejected with `EmptyFilterError`
+unless `allow_empty_filter=True` is passed. This prevents accidental full-table cancels.
+
+**Filter fields used:** `queue`, `status`, `actor`, `identity_key`, `batch_id`, `tags`,
+`active`. The `limit`, `cursor`, and `order_by` fields are ignored — a bulk cancel is
+not paginated.
+
+**Snapshot boundary:** jobs matching the filter that are enqueued *after* the
+statement's snapshot escape this call. Stop producers first, or issue a follow-up call;
+the returned counts make non-convergence detectable.
+
+```python
+result = await client.cancel_where(
+    JobFilter(tags=("tenant-acme",), active=True),
+    reason="tenant offboarded",
+)
+print(f"cancelled {result.cancelled_directly} pending, "
+      f"requested cancel for {result.cancel_requested} running")
+```
+
+### `BulkCancelResult`
+
+Frozen Pydantic model returned by `cancel_where()`.
+
+| Field | Type | Description |
+|---|---|---|
+| `cancelled_directly` | `int` | Pending/scheduled jobs moved straight to terminal `cancelled`. |
+| `cancel_requested` | `int` | Running jobs with `cancel_phase=1` set (cooperative cancel). |
+| `cancelled_ids` | `list[UUID]` | IDs of jobs cancelled directly. |
+| `cancel_requested_ids` | `list[UUID]` | IDs of running jobs with cancel requested. |
+| `total_affected` | `int` (property) | `cancelled_directly + cancel_requested`. |
+
+For tenant-scale cancels (10^5+ matching rows), partition via filter (e.g.
+`JobFilter(queue=..., tags=...)` to split by queue). A single transaction covering 10^6
+rows would hold locks too long. The counts let the caller verify completeness and issue
+follow-up calls for remaining partitions.
+
+---
+
 ## `list()`
 
 ```python
@@ -781,17 +839,52 @@ async def enqueue(
     unique_for: timedelta | None = None,
     unique_states: tuple[JobStatus, ...] | None = None,
     max_pending: int | None = None,
+    tags: list[str] | None = None,
+    inherit_tags: bool = True,
+    schedule_to_close: datetime | None = None,
+    start_to_close: timedelta | None = None,
+    heartbeat_timeout: timedelta | None = None,
 ) -> JobHandle[R]: ...
 ```
 
 Enqueues a single sub-job. Accepts the same options as `JobsClient.enqueue()` except:
 
 - No `queue` override (sub-jobs use the actor's declared queue).
-- No `schedule_to_close`, `start_to_close`, or `heartbeat_timeout` (set on the actor declaration).
 - No explicit `trace_id` / `span_id` (extracted from the active OTel span).
 - `connection` may be passed to use a specific `asyncpg.Connection` rather than the LOOP-scope
   connection.
 
+#### Tag inheritance
+
+Sub-jobs inherit the parent job's tags by default (`inherit_tags=True`). When no
+explicit `tags` are passed, the sub-job carries the parent's tags. When explicit
+`tags` are passed, they merge with the parent's tags (parent first, deduplicated).
+
+| `inherit_tags` | `tags` | Resulting job tags |
+|---|---|---|
+| `True` (default) | `None` | Parent job's tags (or `()` if parent has none) |
+| `True` (default) | `["new-tag"]` | Parent tags + explicit tags, merged (union, parent-first, deduped) |
+| `False` | `None` | `()` (no inheritance) |
+| `False` | `["new-tag"]` | `("new-tag",)` (explicit only) |
+
+Pass `inherit_tags=False` to suppress inheritance for a specific sub-job.
+
+**Blast radius:** inherited tags make sub-jobs visible to `cancel_where` filters
+matching those tags. A shared/utility sub-job enqueued by a tenant-tagged parent
+will be swept up in that tenant's `cancel_where`.
+
+`enqueue_batch()` does **not** inherit parent tags — batch items carry their own
+`EnqueueItem.tags`. This asymmetry is deliberate: batch fan-out callers typically
+set per-item tags explicitly.
+
+#### `schedule_to_close` / `start_to_close` / `heartbeat_timeout`
+
+`schedule_to_close` and `start_to_close` override the actor's declared defaults for
+this specific sub-job. `heartbeat_timeout` has no actor-level declaration — the
+per-call value is the only source. Note that `schedule_to_close` bounds total
+wall-clock time *including* time snoozed on `wait_for_batch` — finalizer-style
+sub-jobs that snooze for long periods should set it generously or not at all.
+
 The per-call `max_pending=` argument is resolved against the operator-owned stored cap and
 the `@actor(...)` literal, not in place of them: against a non-NULL stored
 `actor_config.max_pending` the tighter of the two wins (`min(stored, per_call)`) — explicit
@@ -1014,7 +1107,7 @@ Tags are user-defined keyword labels stored in `jobs.tags text[]`. They have no
 handle = await client.enqueue(
     send_email,
     EmailPayload(to="user@example.com"),
-    tags=["notification", "priority:high", "tenant:acme"],
+    tags=["notification", "priority-high", "tenant-acme"],
 )
 ```
 
@@ -1046,7 +1139,7 @@ Use `JobFilter.tags` with array-overlap semantics (matches jobs that have **any*
 page = await client.list(JobFilter(
     actor="send_email",
     status="failed",
-    tags=["priority:high", "tenant:acme"],
+    tags=["priority-high", "tenant-acme"],
     limit=50,
 ))
 ```

pyproject.toml

@@ -202,6 +202,11 @@ ignore = [
 "src/taskq/backend/_terminal.py" = ["S608", "SIM117"]
 "src/taskq/backend/_enqueue.py" = ["S608", "SIM117"]
 "src/taskq/backend/_reads.py" = ["S608"]
+# S608: schema name validated against _IDENT_RE before interpolation; same
+# rationale as _reads.py. All user-supplied values use $N parameter binding.
+# SIM117: pool.acquire() and conn.transaction() cannot be flattened.
+# S311: random module used for jitter in deadlock retry backoff, not crypto.
+"src/taskq/backend/_cancel_bulk.py" = ["S608", "SIM117", "S311"]
 "src/taskq/backend/_dispatch.py" = ["S608", "SIM117"]
 "src/taskq/backend/_sql_templates.py" = ["S608"]
 # SIM117: pool.acquire() and conn.transaction() cannot be flattened into a
@@ -257,6 +262,7 @@ ignore = [
 # Same rationale — schema name validated via PostgresBackend constructor;
 # all user-supplied values use $N parameter binding.
 "tests/test_backend_equivalence.py" = ["S608"]
+"tests/test_cancel_where_pg.py" = ["S608"]
 # Same rationale — schema name validated against _IDENT_RE; all
 # user-supplied values use $N parameter binding.
 "tests/test_heartbeat_integration.py" = ["S608"]

src/taskq/__init__.py

@@ -37,7 +37,7 @@
     ScheduleRecord,
 )
 from taskq.batch import BatchCompletionStatus, BatchHandle, EnqueueItem, wait_for_batch
-from taskq.client import CancelResult, JobEvent, JobHandle, JobsClient, TaskQ
+from taskq.client import BulkCancelResult, CancelResult, JobEvent, JobHandle, JobsClient, TaskQ
 from taskq.client._enqueuer import SubJobEnqueuer
 from taskq.connections import ConnFactory, PoolFactory, RedisFactory, WorkerConnections
 from taskq.context import JobContext
@@ -48,6 +48,7 @@
     BackpressureError,
     DependencyCycle,
     DIError,
+    EmptyFilterError,
     IllegalStateTransition,
     JobFailed,
     MaxPendingExceededError,
@@ -93,13 +94,15 @@
     "BackpressureError",
     "BatchCompletionStatus",
     "BatchHandle",
+    "BulkCancelResult",
     "CancelPhase",
     "CancelResult",
     "ConnFactory",
     "CronScheduleSpec",
     "DIError",
     "DependencyCycle",
     "DstStrategy",
+    "EmptyFilterError",
     "EnqueueItem",
     "ErrorReporter",
     "EventRow",

src/taskq/backend/__init__.py

@@ -20,6 +20,7 @@
     AttemptRow,
     Backend,
     BackendDeps,
+    BulkCancelResult,
     CancelFlag,
     DstStrategy,
     EnqueueArgs,
@@ -57,6 +58,7 @@ def __getattr__(name: str) -> object:
     "AttemptRow",
     "Backend",
     "BackendDeps",
+    "BulkCancelResult",
     "CancelFlag",
     "DstStrategy",
     "EnqueueArgs",