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

[rcbevans]

Summary

Adds set-based cancel_where(JobFilter) for bulk cancellation in a single SQL round-trip, and widens SubJobEnqueuer.enqueue() with tags, inherit_tags, and missing timeout fields so sub-jobs are visible to tag-based filters via parent-tag inheritance.

Issues addressed

• #54 — bulk cancel by filter
• #57 — sub-job tags

Implementation

Bulk cancel by filter (#54)

BulkCancelResult (backend/_protocol.py, re-exported through types → client → __init__): frozen Pydantic model with cancelled_directly, cancel_requested, cancelled_ids, cancel_requested_ids, and total_affected property. Defined in _protocol.py (not types.py) to avoid the circular import types → _protocol → types.

EmptyFilterError (exceptions.py): TaskQError subclass. Raised by JobsClient.cancel_where when the filter has no predicates (would cancel the entire table). Bypassed with allow_empty_filter=True.

build_filter_conditions() (backend/_filter_sql.py, new): extracted from _reads._list_jobs so reads and cancel share identical filter semantics. Returns FilterSQL(conditions, params). Translates only predicate fields: queue, status, actor, identity_key, batch_id (via metadata @> $N::jsonb), tags (via tags && $N::text[]), active (expands to status = ANY($N) with active/terminal status sets). cursor, limit, and order_by are excluded — callers handle them separately.

JobsClient.cancel_where(filter, reason, *, allow_empty_filter=False) (client/_jobs.py): client-layer method with the empty-filter guardrail, taskq.cancellation.requested counter increment, and schema-error translation. Returns BulkCancelResult.

TaskQ.cancel_where() (client/_taskq.py): delegate to JobsClient.

Backend protocol (backend/_protocol.py): cancel_where(filter, reason) -> BulkCancelResult added. Purely additive — no protocol version bump; a backend lacking the method raises AttributeError loudly.

PostgresBackend.cancel_where() (backend/postgres.py): delegates to _cancel_bulk._cancel_where(), then sends batched NOTIFY via SELECT pg_notify(channel, payload) FROM unnest($1::text[], $2::text[]) for running jobs (fleet channel + per-worker channel). Increments taskq.cancel.notify_sent once per notified job.

_cancel_bulk._cancel_where() (backend/_cancel_bulk.py, new): single CTE statement with two UPDATEs:

• cancelled CTE: pending/scheduled → terminal cancelled (captures prev_status for accurate state_change events)
• cancel_requested CTE: running + cancel_phase=0 → cancel_phase=1 (cooperative)

EPQ-safe predicates are duplicated in each UPDATE's own WHERE clause (not just in the matching CTE) so READ COMMITTED re-evaluation works correctly. ORDER BY id in the matching CTE ensures deterministic lock ordering. Events inserted via executemany within the same transaction. Deadlock retry: 3 attempts with jittered exponential backoff. Returns (BulkCancelResult, list[NotifyTarget]).

InMemoryBackend.cancel_where() (testing/_cancel_bulk.py, new): sanitizes the filter (limit=2^31, cursor=None, order_by=None) before calling _list_jobs to avoid pagination capping. Same event parity as Postgres: state_change + cancel_request for pending/scheduled, cancel_request only for running. Wakes cancel subscribers for running jobs.

Sub-job tags (#57)

SubJobEnqueuer.enqueue() (client/_enqueuer.py): new keyword-only params tags, inherit_tags=True, schedule_to_close, start_to_close, heartbeat_timeout. Tag resolution via _resolve_tags():

inherit_tags	tags	Result
True (default)	None	Parent job's tags
True (default)	["new"]	Parent tags + explicit, merged (union, parent-first, deduped)
False	None	() (current behavior)
False	["new"]	("new",) only

Parent tag propagation via contextvars.ContextVar (_parent_tags_var in _enqueuer.py): the consumer sets parent tags before actor invocation and resets after. asyncio Tasks copy context, so concurrent consumers each see their own value — no per-instance field race.

Consumer integration (worker/_consumer.py, worker/run.py): set_parent_tags(tuple(job.tags)) called before actor invocation, _parent_tags_var.reset(token) in finally. The stub consumer in run.py does this unconditionally; the real consumer always sets parent tags (per-call inherit_tags=False is the opt-out).

Breaking changes

• SubJobEnqueuer.enqueue() signature widened with new keyword-only params (tags, inherit_tags, schedule_to_close, start_to_close, heartbeat_timeout). All default to backward-compatible values — pre-existing callers are unaffected.
• Backend protocol gains cancel_where: additive, no version bump. Backends without the method raise AttributeError when called (loud failure, not silent).

Test coverage

File	Scope	What it covers
tests/test_filter_sql.py	Unit	build_filter_conditions for all predicate fields, empty filter, combined filters, cursor/order_by exclusion. SQL injection safety: metacharacter payloads parameterized, conditions contain only column names + $N placeholders
tests/test_bulk_cancel_types.py	Unit	BulkCancelResult construction, frozen enforcement, zero counts. EmptyFilterError is TaskQError subclass, message mentions guardrail
tests/test_cancel_where.py	Unit (in-memory)	Pending→cancelled, running→cooperative, mixed statuses, no matches, already-cancelled, batch_id filter, queue+actor filter, active filter, ignores filter.limit (pagination guard)
tests/test_cancel_where_pg.py	Integration (Postgres)	Bulk cancel pending/scheduled/running/mixed, filter by tags/queue/actor/batch_id, event insertion, deadlock retry (succeed-on-retry, exhausted-raises, no-retry-on-success)
tests/test_cancel_where_client.py	Unit (client)	Empty filter guardrail, allow_empty_filter bypass, counter increment, schema error translation, TaskQ delegate
tests/test_sub_job_tags.py	Unit	Tag inheritance (default inherit, explicit tags, merge, inherit_tags=False, backward compat)
tests/test_backend_protocol.py	Unit	cancel_where in protocol member set, member count updated
tests/e2e/test_sub_job_tags.py	E2E	Real pipeline: tagged parent → sub-job tag inheritance and merge in a worker container
tests/e2e/test_cancellation.py	E2E	Bulk cancel in a real worker container (merged from standalone file)

Security

The filter→SQL builder (_filter_sql.py) uses positional parameters ($N) for all user-supplied values. Column names are hardcoded literals, never derived from input. No f-string interpolation of user values into SQL condition strings. Dedicated SQL injection tests (TestSQLInjectionSafety in test_filter_sql.py) verify that metacharacter payloads ('; DROP TABLE jobs; --, etc.) end up in params, not in conditions.

---

Closes #54, closes #57