diff --git a/CHANGELOG.md b/CHANGELOG.md index 60d4b1f0..97dfba30 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -203,6 +203,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `stranded_jobs_query_failed`, and the `failed_details` payload of `sub_enqueue_flush_failed`). Log pipelines querying `fields.message` on these two events must switch to `error_message`. +- **Breaking: `taskq.worker.actor_config` moved to `taskq.actor_config`.** + The `ActorConfig` dataclass (released in v0.2.0–v0.2.2 at + `taskq.worker.actor_config`) has moved to the top-level + `taskq.actor_config` module. It is shared by the client, CLI, and admin + UI, not worker-internal. The old import path raises `ImportError`. See + [docs/guides/upgrading.md](docs/guides/upgrading.md) for the full + migration mapping. The companion `actor_config_ops` module (listing, + inspecting, tuning, and deregistering actors) has likewise moved from + `taskq.worker.actor_config_ops` to `taskq.actor_config_ops`; it was + never released under the `worker.*` path. ### Security diff --git a/docs/api-reference/client.md b/docs/api-reference/client.md index 33d66b38..c9214de5 100644 --- a/docs/api-reference/client.md +++ b/docs/api-reference/client.md @@ -1,9 +1,11 @@ # Client -`TaskQ`, `JobsClient`, `JobHandle`, and `CancelResult`. +`TaskQ`, `JobsClient`, `JobHandle`, `CancelResult`, and `ActorsClient`. ::: taskq.client._taskq ::: taskq.client._jobs ::: taskq.client._handle + +::: taskq.client._actors.ActorsClient diff --git a/docs/guides/actors.md b/docs/guides/actors.md index 2d1705e5..7edf7f08 100644 --- a/docs/guides/actors.md +++ b/docs/guides/actors.md @@ -25,6 +25,7 @@ the actor's payload and result types end-to-end. 15. [Progress reporting](#progress-reporting) 16. [Testing actors without a database](#testing-actors-without-a-database) 17. [Full worked example](#full-worked-example) +18. [Actor deregistration](#actor-deregistration) --- @@ -944,3 +945,82 @@ async def submit_order(client, order_id: str, customer_id: str, amount_cents: in print(f"confirmed: {result.confirmation_number}") return handle.job_id ``` + +--- + +## Actor deregistration + +Actors registered by worker startup create `actor_config` rows that persist +until explicitly removed. For long-lived deployments this is intentional — +the row is the source of truth for capacity and routing. For ephemeral, +per-run deployments (e.g. `my-actor.`), each run leaves a row behind. + +### `client.actors.deregister()` + +```python +async with TaskQ(dsn=...) as tq: + result = await tq.actors.deregister("my-actor.run-123") + # force=False: refuses if non-terminal jobs or enabled schedules exist +``` + +**Safety checks (force=False):** +- Refuses if any non-terminal jobs (pending/scheduled/running) reference the + actor. +- Refuses if any enabled cron schedules reference the actor. + +**force=True:** +- Still refuses if **running** jobs exist (they are actively executing). +- Cancels pending/scheduled jobs (marks as `cancelled` with + `error_class='ActorDeregistered'`). +- Disables enabled cron schedules (sets `enabled=false`). + +**Terminal job history** is never deleted. The `jobs.actor` column is plain +text, not a foreign key — terminal rows remain queryable by actor name after +deregistration. + +**Queue cleanup** (`purge_queue=True`): deletes the `queues` row if no other +`actor_config` references the same queue. A shared queue is never purged. + +### Enqueue after deregistration + +After deregistration, any client can still `enqueue()` the dead actor name — +the `INSERT` succeeds (there is no foreign key from `jobs.actor` to +`actor_config.actor`), and the job sits in `pending` status forever. Because +the dispatch query inner-joins `actor_config`, the job will **never be +dispatched** and no background sweep will reap it. + +**Operational discipline:** stop enqueuing to an actor *before* deregistering +it. Deregistration is best-effort against concurrent enqueue/dispatch; +callers must quiesce the actor first. + +**Stop workers first:** A concurrent worker startup (`sync_actor_config`) +can re-create the `actor_config` row after deregistration, with capacity +fields reset to `@actor(...)` defaults. Stop all workers for the actor +before calling `deregister`. + +### Idempotent deregistration + +A second `deregister` call on an already-deregistered actor raises +`ActorNotFoundError`. For cleanup-automation loops: + +```python +from taskq.exceptions import ActorNotFoundError + +try: + await tq.actors.deregister(actor_name, force=True, purge_queue=True) +except ActorNotFoundError: + pass # already deregistered — idempotent +``` + +### CLI + +```bash +taskq actor-config deregister my-actor.run-123 +taskq actor-config deregister my-actor.run-123 --force --purge-queue +``` + +### Admin UI + +The `/admin/actors` page lists all `actor_config` rows with active job counts +and schedule counts. Each row has a deregister form with `force` and +`purge_queue` checkboxes (requires `TASKQ_ADMIN_ACTIONS_ENABLED=true`). diff --git a/docs/guides/admin-ui.md b/docs/guides/admin-ui.md index cf93f82c..e3f34e21 100644 --- a/docs/guides/admin-ui.md +++ b/docs/guides/admin-ui.md @@ -342,6 +342,32 @@ Rate-limit state page. Reads all rows from `rate_limit_buckets` (bucket name, ki Reservation slot summary. For each `bucket_name` in `reservation_slots`, shows the count of held slots (where `job_id IS NOT NULL`), free slots, and total slots. +### `GET /admin/actors` + +The `/admin/actors` page lists all stored `actor_config` rows with: + +- Actor name, queue, max concurrent, max pending +- Active job count (pending + scheduled + running) +- Enabled schedule count +- Last updated timestamp + +Each row has a **Deregister** button with `force` and `purge queue` checkboxes. +Deregistration requires `TASKQ_ADMIN_ACTIONS_ENABLED=true`. The form is +CSRF-protected via the synchronizer-token pattern. + +### POST /admin/actors/{actor}/deregister + +Deregisters an actor. Form fields: +- `csrf_token` — CSRF synchronizer token (set by GET) +- `force` — checkbox; cancels pending/scheduled jobs and disables schedules +- `purge_queue` — checkbox; deletes the orphaned queues row + +Response codes: +- `303` — success, redirects to `/actors?notice=deregistered+{actor}` +- `403` — admin actions disabled or CSRF validation failed +- `404` — actor not found (no `actor_config` row) +- `409` — actor has active jobs or enabled schedules (force=False) + ### `GET /admin/sse/{topic}` SSE (Server-Sent Events) endpoint. Accepts any `topic` string. On connect it emits an initial `event: status` frame with `{"status": "awaiting_progress_backend"}`, then sends `: keepalive` comments every 30 seconds to prevent connection timeout. See [Real-time vs polling mode](#real-time-vs-polling-mode) below. diff --git a/docs/guides/cli.md b/docs/guides/cli.md index 441d389b..ab8ee85c 100644 --- a/docs/guides/cli.md +++ b/docs/guides/cli.md @@ -303,6 +303,22 @@ Per actor and capacity field this prints the `@actor(...)` literal, the stored v `taskq actor-config set` requires the actor to have a stored row already (created by a worker startup that registered it). `queue` and `metadata` are structural and are only ever changed by redeploying with a new `@actor(...)` registration (plus `--force-update-actor-config` if a stored row already exists). +### `taskq actor-config deregister` + +Deregister an actor: delete its `actor_config` row with safety checks. + +```bash +taskq actor-config deregister [--force] [--purge-queue] +``` + +- `` — actor name (positional argument) +- `--force` — cancel pending/scheduled jobs, disable enabled cron schedules, + and proceed despite non-terminal jobs. Running jobs still block. +- `--purge-queue` — also delete the orphaned `queues` row if no other actor + references it. + +Exit code 0 on success, 1 on refusal (with error message) or not found. + ### Exit codes | Code | Meaning | diff --git a/docs/guides/jobs-clients.md b/docs/guides/jobs-clients.md index 30ddf59e..94417df0 100644 --- a/docs/guides/jobs-clients.md +++ b/docs/guides/jobs-clients.md @@ -4,6 +4,9 @@ `Backend` and adds typed payload serialisation, `JobHandle[R]` construction, and `CancelResult` building. +**Actor management:** `tq.actors` provides `list()`, `get()`, `set_capacity()`, +and `deregister()` — see [Actor deregistration](actors.md#actor-deregistration). + --- ## Job lifecycle diff --git a/docs/guides/upgrading.md b/docs/guides/upgrading.md index ec4d8b94..5672b99b 100644 --- a/docs/guides/upgrading.md +++ b/docs/guides/upgrading.md @@ -61,3 +61,61 @@ This is a deliberate tradeoff, not a missing feature: - Pin `taskq-py` back to the previous version until the issue is resolved, since the previous version's code may not be compatible with the new schema. + +--- + +## Breaking import path changes + +### `taskq.worker.actor_config` → `taskq.actor_config` + +> **Released in v0.2.0–v0.2.2, moved in unreleased.** This is a breaking +> change for anyone importing `ActorConfig` from the old path. + +The `ActorConfig` dataclass has moved from `taskq.worker.actor_config` to +the top-level `taskq.actor_config` module. It is a shared carrier used by +the client, CLI, and admin UI — not worker-internal. + +**Old (v0.2.0–v0.2.2):** + +```python +from taskq.worker.actor_config import ActorConfig +``` + +**New:** + +```python +from taskq.actor_config import ActorConfig +``` + +The old import path raises `ImportError` — update your imports. + +### `taskq.worker.actor_config_ops` → `taskq.actor_config_ops` + +The `actor_config_ops` module — listing, inspecting, tuning, and +deregistering actors on a live deployment — has moved from +`taskq.worker.actor_config_ops` to the top-level +`taskq.actor_config_ops`. This module was introduced on the unreleased +branch; if you were importing it from the `worker.*` path during +development, update to the top-level path. + +**Old (unreleased branch only):** + +```python +from taskq.worker.actor_config_ops import ( + list_actor_configs, + get_actor_config, + set_actor_config_capacity, + deregister_actor, +) +``` + +**New:** + +```python +from taskq.actor_config_ops import ( + list_actor_configs, + get_actor_config, + set_actor_config_capacity, + deregister_actor, +) +``` diff --git a/pyproject.toml b/pyproject.toml index db397dc6..26e33d6a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -274,6 +274,9 @@ ignore = [ # Same rationale — schema name validated against _IDENT_RE; all # user-supplied values use $N parameter binding. "tests/test_web_admin_integration.py" = ["S608"] +# Same rationale — schema name validated against _IDENT_RE; all +# user-supplied values use $N parameter binding. +"tests/test_web_admin_actors.py" = ["S608"] # FastAPI Depends() in test route-handler defaults is the same idiomatic # declarative-injection pattern as web/admin — not a real mutable default. "tests/test_sso_session.py" = ["B008"] diff --git a/src/taskq/__init__.py b/src/taskq/__init__.py index e18c8f1f..b73613fb 100644 --- a/src/taskq/__init__.py +++ b/src/taskq/__init__.py @@ -11,6 +11,7 @@ import importlib.metadata from taskq.actor import ActorFn, ActorFnWithCtx, ActorHandler, ActorRef, actor +from taskq.actor_config_ops import ActorConfigRow, DeregisterResult from taskq.auth import ( PgCredential, PgCredentialProvider, @@ -38,6 +39,7 @@ ) from taskq.batch import BatchCompletionStatus, BatchHandle, EnqueueItem, wait_for_batch from taskq.client import CancelResult, JobEvent, JobHandle, JobsClient, TaskQ +from taskq.client._actors import ActorsClient from taskq.client._enqueuer import SubJobEnqueuer from taskq.connections import ConnFactory, PoolFactory, RedisFactory, WorkerConnections from taskq.context import JobContext @@ -45,6 +47,10 @@ from taskq.exceptions import ( ActorConfigDriftError, ActorConfigDriftList, + ActorDeregistrationError, + ActorHasActiveJobsError, + ActorHasEnabledSchedulesError, + ActorNotFoundError, BackpressureError, DependencyCycle, DIError, @@ -86,10 +92,16 @@ __all__ = [ "ActorConfigDriftError", "ActorConfigDriftList", + "ActorConfigRow", + "ActorDeregistrationError", "ActorFn", "ActorFnWithCtx", "ActorHandler", + "ActorHasActiveJobsError", + "ActorHasEnabledSchedulesError", + "ActorNotFoundError", "ActorRef", + "ActorsClient", "BackpressureError", "BatchCompletionStatus", "BatchHandle", @@ -99,6 +111,7 @@ "CronScheduleSpec", "DIError", "DependencyCycle", + "DeregisterResult", "DstStrategy", "EnqueueItem", "ErrorReporter", diff --git a/src/taskq/worker/actor_config.py b/src/taskq/actor_config.py similarity index 100% rename from src/taskq/worker/actor_config.py rename to src/taskq/actor_config.py diff --git a/src/taskq/actor_config_ops.py b/src/taskq/actor_config_ops.py new file mode 100644 index 00000000..f77abb5c --- /dev/null +++ b/src/taskq/actor_config_ops.py @@ -0,0 +1,506 @@ +"""Operator surface for reading and tuning stored `{schema}.actor_config` rows. + +Complements :func:`taskq.worker.startup.sync_actor_config`: that function +only ever *seeds* a row's capacity fields (``max_concurrent``, +``max_pending``, ``result_ttl``) on first registration and otherwise +leaves them untouched. This module is how an operator changes them +afterwards — on a live deployment, without a code change or a worker +restart. All three are re-read by the engine without a restart: + +* ``max_concurrent`` — the dispatch query joins ``actor_config`` fresh + on every dispatch cycle (``taskq/backend/_dispatch_sql.py``); a change + is effective immediately. +* ``result_ttl`` — the terminal-write UPDATE recomputes + ``result_expires_at`` from the stored value for every completing job + (``taskq/backend/_sql_templates.py::mark_succeeded``); a change is + effective for jobs completing after the write. +* ``max_pending`` — enqueue-side processes hold a TTL-bounded cache of + this table (``taskq/client/_capacity.py``, default 5s staleness); a + change is effective fleet-wide within seconds, with no redeploy. + +Clearing semantics differ by field on purpose. ``--clear-max-concurrent`` +writes NULL, which the dispatch SQL reads as *unlimited* — the SQL +cannot see the code literal once the row exists. ``--clear-max-pending`` +and ``--clear-result-ttl`` write NULL, which their enforcement paths +read as *fall back to the ``@actor(...)`` literal* — clearing reverts an +override to the code default. + +This module also provides :func:`deregister_actor` — the transactional +removal of an ``actor_config`` row with safety checks for active jobs +and enabled schedules, optional forced cancellation of pending/scheduled +jobs, optional disabling of cron schedules, and optional purging of +orphaned queues. See :class:`DeregisterResult` for the return contract. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass +from typing import TYPE_CHECKING, Final + +from taskq._json import loads +from taskq.backend._protocol import ConnLike +from taskq.backend._records import jsonb_param +from taskq.backend._sql import INSERT_EVENT_SQL +from taskq.backend.statemachine import ACTIVE_STATUSES, TERMINAL_STATUSES +from taskq.constants import ( + _IDENT_RE, # pyright: ignore[reportPrivateUsage] # Why: reusing the canonical identifier regex rather than redefining it +) +from taskq.exceptions import ( + ActorHasActiveJobsError, + ActorHasEnabledSchedulesError, + ActorNotFoundError, +) + +if TYPE_CHECKING: + import asyncpg + +__all__ = [ + "UNSET", + "ActorConfigRow", + "DeregisterResult", + "Unset", + "deregister_actor", + "get_actor_config", + "list_actor_configs", + "list_actor_summaries", + "set_actor_config_capacity", +] + + +class Unset: + """Sentinel distinguishing 'leave unchanged' from an explicit ``None`` (clear).""" + + def __repr__(self) -> str: + return "UNSET" + + +UNSET: Final = Unset() + + +@dataclass(frozen=True, slots=True) +class ActorConfigRow: + """Snapshot of one `{schema}.actor_config` row.""" + + actor: str + max_concurrent: int | None + max_pending: int | None + queue: str + result_ttl: float | None + metadata: dict[str, object] + updated_at: str + + +@dataclass(frozen=True, slots=True) +class DeregisterResult: + """Outcome of a ``deregister_actor`` call. + + ``actor_config_deleted`` is always ``True`` — if the row is not found, + ``deregister_actor`` raises :class:`ActorNotFoundError` instead of + returning a result with ``False``. The field is retained for API + contract clarity and consumer assertions. + """ + + actor: str + queue: str + actor_config_deleted: bool + schedules_disabled: int + jobs_cancelled: int + terminal_jobs_remaining: int + queue_purged: bool + + +_LIST_ACTOR_CONFIG_SQL = """ +SELECT actor, max_concurrent, max_pending, queue, result_ttl, + metadata::text AS metadata, updated_at::text AS updated_at + FROM "{schema}".actor_config + ORDER BY actor +""".strip() + +_GET_ACTOR_CONFIG_SQL = """ +SELECT actor, max_concurrent, max_pending, queue, result_ttl, + metadata::text AS metadata, updated_at::text AS updated_at + FROM "{schema}".actor_config + WHERE actor = $1 +""".strip() + +# Each capacity column is only overwritten when its paired boolean +# "touch" flag is true; otherwise the CASE expression preserves the +# current value. This lets one statement express "set to N", "clear to +# NULL", and "leave alone" for all three fields without dynamic SQL. +_SET_ACTOR_CONFIG_CAPACITY_SQL = """ +UPDATE "{schema}".actor_config + SET max_concurrent = CASE WHEN $2 THEN $3 ELSE max_concurrent END, + max_pending = CASE WHEN $4 THEN $5 ELSE max_pending END, + result_ttl = CASE WHEN $6 THEN $7 ELSE result_ttl END, + updated_at = now() + WHERE actor = $1 +RETURNING actor, max_concurrent, max_pending, queue, result_ttl, + metadata::text AS metadata, updated_at::text AS updated_at +""".strip() + + +def _row_to_dataclass(row: asyncpg.Record) -> ActorConfigRow: + return ActorConfigRow( + actor=row["actor"], + max_concurrent=row["max_concurrent"], + max_pending=row["max_pending"], + queue=row["queue"], + result_ttl=row["result_ttl"], + metadata=loads(row["metadata"]), + updated_at=row["updated_at"], + ) + + +async def list_actor_configs(conn: ConnLike, *, schema: str = "taskq") -> list[ActorConfigRow]: + """Return every stored `{schema}.actor_config` row, ordered by actor name.""" + if not _IDENT_RE.match(schema): + raise ValueError(f"invalid schema identifier: {schema!r}") + rows = await conn.fetch(_LIST_ACTOR_CONFIG_SQL.format(schema=schema)) + return [_row_to_dataclass(row) for row in rows] + + +_ACTOR_SUMMARIES_SQL = """ +SELECT ac.actor, ac.max_concurrent, ac.max_pending, ac.queue, + ac.updated_at::text AS updated_at, + (SELECT count(*) FROM "{schema}".jobs j + WHERE j.actor = ac.actor + AND j.status = ANY($1::"{schema}".job_status[])) AS active_job_count, + (SELECT count(*) FROM "{schema}".cron_schedules cs + WHERE cs.actor = ac.actor AND cs.enabled = true) AS enabled_schedule_count + FROM "{schema}".actor_config ac + ORDER BY ac.actor +""".strip() + + +async def list_actor_summaries(conn: ConnLike, *, schema: str = "taskq") -> list[dict[str, object]]: + """Return actor_config rows with active job and schedule counts for display.""" + if not _IDENT_RE.match(schema): + raise ValueError(f"invalid schema identifier: {schema!r}") + rows = await conn.fetch( + _ACTOR_SUMMARIES_SQL.format(schema=schema), + list(ACTIVE_STATUSES), + ) + return [dict(r) for r in rows] + + +async def get_actor_config( + conn: ConnLike, actor: str, *, schema: str = "taskq" +) -> ActorConfigRow | None: + """Return the stored row for *actor*, or ``None`` if it has never been synced.""" + if not _IDENT_RE.match(schema): + raise ValueError(f"invalid schema identifier: {schema!r}") + row = await conn.fetchrow(_GET_ACTOR_CONFIG_SQL.format(schema=schema), actor) + return _row_to_dataclass(row) if row is not None else None + + +def _validate_int_field(name: str, value: int | None | Unset) -> None: + """Reject the two shapes that slip past ``isinstance(x, int) and x < 0``: + ``bool`` (an ``int`` subclass — ``False`` would be written as 0, flooring + the dispatch residual ``GREATEST(cap - in_flight, 0)`` and silently + pausing the actor) and negative values.""" + if isinstance(value, bool): + raise ValueError(f"{name} must be a non-negative integer; got {value!r} (bool)") + if isinstance(value, int) and value < 0: + raise ValueError(f"{name} must be a non-negative integer; got {value!r}") + + +def _validate_result_ttl(value: float | None | Unset) -> None: + """Reject bool, negative, and non-finite ``result_ttl``. + + NaN sails through ``value < 0`` (NaN compares False) and then breaks + every completion for the actor — ``now() + NaN * interval '1 second'`` + raises ``interval out of range`` in the terminal-write UPDATE. ±inf is + rejected on the same grounds (``interval out of range`` / meaningless + expiry).""" + if isinstance(value, bool): + raise ValueError( + f"result_ttl must be a non-negative number of seconds; got {value!r} (bool)" + ) + if isinstance(value, (int, float)): + if not math.isfinite(value): + raise ValueError(f"result_ttl must be finite; got {value!r}") + if value < 0: + raise ValueError(f"result_ttl must be a non-negative number of seconds; got {value!r}") + + +async def set_actor_config_capacity( + conn: ConnLike, + actor: str, + *, + max_concurrent: int | None | Unset = UNSET, + max_pending: int | None | Unset = UNSET, + result_ttl: float | None | Unset = UNSET, + schema: str = "taskq", +) -> ActorConfigRow | None: + """Update capacity fields on an existing `{schema}.actor_config` row. + + Only fields passed as something other than :data:`UNSET` are + changed. Pass ``None`` explicitly to clear a field — precisely what + that means depends on the field's enforcement path: clearing + ``max_concurrent`` makes the actor *unlimited* (the dispatch SQL + reads a stored NULL as no cap), while clearing ``max_pending`` or + ``result_ttl`` reverts enforcement to the ``@actor(...)`` literal + (those paths can still see the code default). Returns ``None`` if + *actor* has no stored row — a row is only created by + :func:`taskq.worker.startup.sync_actor_config` at worker startup, so + an actor must have been registered by at least one worker before its + capacity can be tuned here. + + Raises :class:`ValueError` if ``max_concurrent`` or ``max_pending`` + is a negative integer or a ``bool`` — the same guard ``@actor(...)`` + applies at decoration time (``taskq/actor.py``), plus the bool case + (``False`` is an ``int`` and would be written as 0). Without these, + an operator typo here (e.g. ``--max-concurrent -5``) would write + silently into the dispatch CTE's ``GREATEST(ac.max_concurrent - + in_flight, 0)`` residual calculation, floor to zero, and pause the + actor indefinitely with no error anywhere in the path. ``result_ttl`` + is likewise rejected when negative, non-finite, or ``bool`` — a + negative TTL would set ``result_expires_at`` to a past timestamp in + the terminal-write UPDATE (silently expiring every result the moment + it is written), and NaN/±inf raise ``interval out of range`` in that + same UPDATE, failing every completion for the actor. + """ + if not _IDENT_RE.match(schema): + raise ValueError(f"invalid schema identifier: {schema!r}") + _validate_int_field("max_concurrent", max_concurrent) + _validate_int_field("max_pending", max_pending) + _validate_result_ttl(result_ttl) + + row = await conn.fetchrow( + _SET_ACTOR_CONFIG_CAPACITY_SQL.format(schema=schema), + actor, + not isinstance(max_concurrent, Unset), + None if isinstance(max_concurrent, Unset) else max_concurrent, + not isinstance(max_pending, Unset), + None if isinstance(max_pending, Unset) else max_pending, + not isinstance(result_ttl, Unset), + None if isinstance(result_ttl, Unset) else result_ttl, + ) + return _row_to_dataclass(row) if row is not None else None + + +# ── deregister_actor ──────────────────────────────────────────────────── + +_RUNNING_STATUS: str = "running" + +_DEREGISTER_CHECK_ACTOR_EXISTS_SQL = """ +SELECT 1 FROM "{schema}".actor_config WHERE actor = $1 +""".strip() + +_DEREGISTER_CHECK_ACTIVE_JOBS_SQL = """ +SELECT status, count(*) AS cnt + FROM "{schema}".jobs + WHERE actor = $1 AND status = ANY($2::"{schema}".job_status[]) + GROUP BY status +""".strip() + +_DEREGISTER_CHECK_SCHEDULES_SQL = """ +SELECT id::text FROM "{schema}".cron_schedules + WHERE actor = $1 AND enabled = true +""".strip() + +_DEREGISTER_CANCEL_PENDING_SQL = """ +UPDATE "{schema}".jobs + SET status = 'cancelled', + finished_at = now(), + error_class = 'ActorDeregistered', + error_message = 'Job cancelled by actor deregistration (force=True)' + WHERE actor = $1 + AND status IN ('pending', 'scheduled') +RETURNING id +""".strip() + +_DEREGISTER_DISABLE_SCHEDULES_SQL = """ +UPDATE "{schema}".cron_schedules + SET enabled = false + WHERE actor = $1 AND enabled = true +""".strip() + +_DEREGISTER_DELETE_ACTOR_CONFIG_SQL = """ +DELETE FROM "{schema}".actor_config WHERE actor = $1 +RETURNING queue +""".strip() + +_DEREGISTER_PURGE_QUEUE_SQL = """ +DELETE FROM "{schema}".queues + WHERE name = $1 + AND NOT EXISTS ( + SELECT 1 FROM "{schema}".actor_config WHERE queue = $1 + ) +RETURNING name +""".strip() + +_DEREGISTER_COUNT_TERMINAL_SQL = """ +SELECT count(*) FROM "{schema}".jobs + WHERE actor = $1 AND status = ANY($2::"{schema}".job_status[]) +""".strip() + + +async def deregister_actor( + conn: ConnLike, + actor: str, + *, + force: bool = False, + purge_queue: bool = False, + schema: str = "taskq", +) -> DeregisterResult: + """Deregister an actor: delete its ``actor_config`` row with safety checks. + + **Default (force=False):** + 1. Refuse if any non-terminal jobs (pending/scheduled/running) reference + the actor — raises :class:`ActorHasActiveJobsError`. + 2. Refuse if any enabled cron schedules reference the actor — raises + :class:`ActorHasEnabledSchedulesError`. + 3. Delete the ``actor_config`` row. + 4. Optionally purge the orphaned queue (if ``purge_queue=True`` and no + other ``actor_config`` row references the same queue). + + **force=True:** + 1. Refuse if any running jobs reference the actor — raises + :class:`ActorHasActiveJobsError`. + 2. Cancel pending/scheduled jobs for this actor. + 3. Disable enabled cron schedules for this actor. + 4. Delete the ``actor_config`` row. + 5. Optionally purge the orphaned queue. + + Terminal job history is never deleted or modified. The entire operation + runs inside a single ``conn.transaction()`` block. If the actor has no + stored ``actor_config`` row, raises :class:`ActorNotFoundError`. + + .. warning:: + **Concurrent enqueue / dispatch race (TOCTOU).** The transaction + uses READ COMMITTED isolation. Callers must quiesce the actor first + — stop enqueuing, disable cron schedules, and wait for running jobs + to reach a terminal state — before calling deregister. + + **Concurrent worker startup (sync_actor_config)** can re-create the + ``actor_config`` row after this function returns, with capacity fields + reset to ``@actor(...)`` defaults. Stop all workers for this actor + before calling deregister. + + A job dispatched between the running check and the cancel UPDATE + (force=True) will be left running with no ``actor_config`` row. + When its lock expires, the leader sweep transitions it to a + terminal state (crashed/cancelled) — it is NOT re-dispatched. + + A cron schedule that fires between the disable UPDATE and the + ``actor_config`` DELETE will enqueue a job that can never be + dispatched (same stranding as enqueue-during-deregister). + + Concurrent deregistration of the last two actors sharing a queue + may leave the queue row orphaned (both transactions see the + other's ``actor_config`` row as still present under READ + COMMITTED). The queue can be manually deleted if needed. + """ + if not _IDENT_RE.match(schema): + raise ValueError(f"invalid schema identifier: {schema!r}") + + async with conn.transaction(): + # Check actor_config row exists first — ActorNotFoundError takes + # precedence over all other checks so callers don't get misleading + # errors for actors that are already deregistered but have stranded + # jobs. + exists = await conn.fetchval( + _DEREGISTER_CHECK_ACTOR_EXISTS_SQL.format(schema=schema), + actor, + ) + if not exists: + raise ActorNotFoundError(actor) + + if not force: + active_rows = await conn.fetch( + _DEREGISTER_CHECK_ACTIVE_JOBS_SQL.format(schema=schema), + actor, + list(ACTIVE_STATUSES), + ) + if active_rows: + status_counts: dict[str, int] = { + str(row["status"]): int(row["cnt"]) for row in active_rows + } + active_count = sum(status_counts.values()) + raise ActorHasActiveJobsError(actor, active_count, status_counts) + + schedule_rows = await conn.fetch( + _DEREGISTER_CHECK_SCHEDULES_SQL.format(schema=schema), + actor, + ) + if schedule_rows: + schedule_ids = [row["id"] for row in schedule_rows] + raise ActorHasEnabledSchedulesError(actor, schedule_ids) + + jobs_cancelled = 0 + schedules_disabled = 0 + else: + running_rows = await conn.fetch( + _DEREGISTER_CHECK_ACTIVE_JOBS_SQL.format(schema=schema), + actor, + [_RUNNING_STATUS], + ) + if running_rows: + status_counts: dict[str, int] = { + str(row["status"]): int(row["cnt"]) for row in running_rows + } + active_count = sum(status_counts.values()) + raise ActorHasActiveJobsError(actor, active_count, status_counts, force=True) + + cancelled_rows = await conn.fetch( + _DEREGISTER_CANCEL_PENDING_SQL.format(schema=schema), + actor, + ) + jobs_cancelled = len(cancelled_rows) + if cancelled_rows: + detail = jsonb_param( + { + "from_state": "pending_or_scheduled", + "to_state": "cancelled", + "reason": "actor_deregistered", + } + ) + event_sql = INSERT_EVENT_SQL.format(schema=schema) + await conn.executemany( + event_sql, + [(row["id"], "state_change", detail) for row in cancelled_rows], + ) + + disable_result = await conn.execute( + _DEREGISTER_DISABLE_SCHEDULES_SQL.format(schema=schema), + actor, + ) + schedules_disabled = int(disable_result.split()[-1]) if disable_result else 0 + + deleted_rows = await conn.fetch( + _DEREGISTER_DELETE_ACTOR_CONFIG_SQL.format(schema=schema), + actor, + ) + if not deleted_rows: + # Handles the concurrent-delete race: under READ COMMITTED, a + # concurrent transaction could delete the row between our + # preflight check and this DELETE. + raise ActorNotFoundError(actor) + + queue_name = deleted_rows[0]["queue"] + + terminal_count = await conn.fetchval( + _DEREGISTER_COUNT_TERMINAL_SQL.format(schema=schema), + actor, + list(TERMINAL_STATUSES), + ) + + queue_purged = False + if purge_queue: + purged_name = await conn.fetchval( + _DEREGISTER_PURGE_QUEUE_SQL.format(schema=schema), + queue_name, + ) + queue_purged = purged_name is not None + + return DeregisterResult( + actor=actor, + queue=queue_name, + actor_config_deleted=True, + schedules_disabled=schedules_disabled, + jobs_cancelled=jobs_cancelled, + terminal_jobs_remaining=terminal_count or 0, + queue_purged=queue_purged, + ) diff --git a/src/taskq/cli.py b/src/taskq/cli.py index 865db7eb..12a5aa5e 100644 --- a/src/taskq/cli.py +++ b/src/taskq/cli.py @@ -30,16 +30,17 @@ close_redis_bounded, ) from taskq.actor import ActorRef -from taskq.exceptions import ActorConfigDriftList -from taskq.settings import TaskQSettings, WorkerSettings -from taskq.worker.actor_config_ops import ( +from taskq.actor_config_ops import ( UNSET, ActorConfigRow, Unset, + deregister_actor, get_actor_config, list_actor_configs, set_actor_config_capacity, ) +from taskq.exceptions import ActorConfigDriftList, ActorDeregistrationError, ActorNotFoundError +from taskq.settings import TaskQSettings, WorkerSettings from taskq.worker.dev import dev_watch_loop from taskq.worker.run import worker_main as _worker_main @@ -517,6 +518,81 @@ async def _actor_config_set( _print_actor_config_row(row) +@actor_config_app.command("deregister") +def actor_config_deregister( + actor: Annotated[str, typer.Argument(help="Actor name to deregister.")], + force: Annotated[ + bool, + typer.Option( + "--force", + help="Cancel pending/scheduled jobs and disable enabled cron schedules" + " instead of refusing. Running jobs still block deregistration.", + ), + ] = False, + purge_queue: Annotated[ + bool, + typer.Option( + "--purge-queue", + help="Also delete the orphaned queues row if no other actor_config" + " references the same queue.", + ), + ] = False, +) -> None: + """Deregister an actor: delete its actor_config row with safety checks. + + By default refuses if non-terminal jobs or enabled cron schedules + reference the actor. Use --force to cancel pending/scheduled jobs and + disable schedules. Running jobs always block (force or not). Use + --purge-queue to also delete the queues row if no other actor uses it. + + Exit codes: 0 success, 2 refusal (active jobs/schedules or invalid + schema), 3 not found. + """ + settings = TaskQSettings.load() + asyncio.run(_actor_config_deregister(settings, actor, force, purge_queue)) + + +async def _actor_config_deregister( + settings: TaskQSettings, + actor: str, + force: bool, + purge_queue: bool, +) -> None: + conn = await asyncpg.connect(str(settings.pg_dsn)) + try: + result = await deregister_actor( + conn, + actor, + force=force, + purge_queue=purge_queue, + schema=settings.schema_name, + ) + except ActorNotFoundError as exc: + typer.echo(str(exc), err=True) + raise typer.Exit(code=3) from None + except (ActorDeregistrationError, ValueError) as exc: + typer.echo(str(exc), err=True) + raise typer.Exit(code=2) from None + finally: + await conn.close() + + typer.echo( + f"Deregistered actor {result.actor!r}:" + f" actor_config_deleted={result.actor_config_deleted}" + f" queue={result.queue!r}" + f" schedules_disabled={result.schedules_disabled}" + f" jobs_cancelled={result.jobs_cancelled}" + f" terminal_jobs_remaining={result.terminal_jobs_remaining}" + f" queue_purged={result.queue_purged}" + ) + typer.echo( + f"WARNING: Actor {result.actor!r} is now unregistered. Any future enqueue()" + f" to this actor name will create a stranded pending job that will never" + f" be dispatched. Stop enqueuing before deregistering.", + err=True, + ) + + _CAPACITY_DIFF_FIELDS = ("max_concurrent", "max_pending", "result_ttl") diff --git a/src/taskq/client/__init__.py b/src/taskq/client/__init__.py index 8418f1f5..99d8421c 100644 --- a/src/taskq/client/__init__.py +++ b/src/taskq/client/__init__.py @@ -1,14 +1,16 @@ """TaskQ client — the public surface for enqueuing, querying, and -cancelling jobs. +cancelling jobs, plus actor management (listing, capacity tuning, +deregistration). -Re-exports :class:`JobsClient`, :class:`JobHandle`, :class:`TaskQ`, and -:class:`JobEvent`. +Re-exports :class:`JobsClient`, :class:`ActorsClient`, :class:`JobHandle`, +:class:`TaskQ`, and :class:`JobEvent`. Import from ``taskq.client`` (or from ``taskq`` which re-exports these names). anchors: (public API ownership). """ +from taskq.client._actors import ActorsClient from taskq.client._enqueuer import SubJobEnqueuer from taskq.client._handle import JobHandle from taskq.client._jobs import JobsClient @@ -16,6 +18,7 @@ from taskq.types import CancelResult __all__ = [ + "ActorsClient", "CancelResult", "JobEvent", "JobHandle", diff --git a/src/taskq/client/_actors.py b/src/taskq/client/_actors.py new file mode 100644 index 00000000..2714157b --- /dev/null +++ b/src/taskq/client/_actors.py @@ -0,0 +1,104 @@ +"""ActorsClient — pool-wrapping facade for actor configuration operations. + +Provides a typed surface for listing, inspecting, tuning, and deregistering +stored ``actor_config`` rows. Each method acquires a connection from the +injected pool, delegates to ``taskq.actor_config_ops``, and returns +the result. +""" + +from typing import TYPE_CHECKING + +from taskq.actor_config_ops import ( + UNSET, + ActorConfigRow, + DeregisterResult, + Unset, + deregister_actor, + get_actor_config, + list_actor_configs, + set_actor_config_capacity, +) + +if TYPE_CHECKING: + import asyncpg + +__all__ = ["ActorsClient"] + + +class ActorsClient: + """Pool-wrapping facade for actor configuration operations. + + Acquires a connection from the injected pool for each call, delegates + to ``taskq.actor_config_ops``, and returns the result. The + caller must have opened the pool; this class does not manage its + lifecycle. + + .. note:: + This client is Postgres-only — it delegates to + :mod:`taskq.actor_config_ops`, which executes raw SQL against + the ``actor_config`` table. The :class:`~taskq.backend._protocol.Backend` + protocol does not include actor config operations, so + :class:`~taskq.testing.InMemoryBackend` does not support + ``ActorsClient``. + + Parameters + ---------- + pool: + An open ``asyncpg.Pool``. The caller retains ownership. + schema: + TaskQ schema name. Defaults to ``"taskq"``. + """ + + def __init__(self, pool: "asyncpg.Pool", *, schema: str = "taskq") -> None: + self._pool = pool + self._schema = schema + + async def list(self) -> list[ActorConfigRow]: + """List all stored actor_config rows, ordered by actor name.""" + async with self._pool.acquire() as conn: + return await list_actor_configs(conn, schema=self._schema) + + async def get(self, actor: str) -> ActorConfigRow | None: + """Get one actor_config row, or ``None`` if not found.""" + async with self._pool.acquire() as conn: + return await get_actor_config(conn, actor, schema=self._schema) + + async def set_capacity( + self, + actor: str, + *, + max_concurrent: int | None | Unset = UNSET, + max_pending: int | None | Unset = UNSET, + result_ttl: float | None | Unset = UNSET, + ) -> ActorConfigRow | None: + """Update capacity fields on an existing actor_config row.""" + async with self._pool.acquire() as conn: + return await set_actor_config_capacity( + conn, + actor, + max_concurrent=max_concurrent, + max_pending=max_pending, + result_ttl=result_ttl, + schema=self._schema, + ) + + async def deregister( + self, + actor: str, + *, + force: bool = False, + purge_queue: bool = False, + ) -> DeregisterResult: + """Deregister an actor with safety checks. + + See :func:`taskq.actor_config_ops.deregister_actor` for + the full semantics. + """ + async with self._pool.acquire() as conn: + return await deregister_actor( + conn, + actor, + force=force, + purge_queue=purge_queue, + schema=self._schema, + ) diff --git a/src/taskq/client/_taskq.py b/src/taskq/client/_taskq.py index fc21d32a..ec8d8e2d 100644 --- a/src/taskq/client/_taskq.py +++ b/src/taskq/client/_taskq.py @@ -67,6 +67,7 @@ async def create_task(payload: MyPayload): ) from taskq.backend.statemachine import TERMINAL_STATUSES from taskq.batch import BatchHandle, EnqueueItem +from taskq.client._actors import ActorsClient from taskq.client._handle import JobHandle from taskq.client._jobs import JobsClient from taskq.constants import RECLAIM_EVENT_VISIBILITY_DELAY, progress_channel, wake_channel @@ -74,7 +75,7 @@ async def create_task(payload: MyPayload): from taskq.progress._events import ProgressEvent from taskq.types import CancelResult -__all__ = ["EventRow", "JobEvent", "TaskQ"] +__all__ = ["ActorsClient", "EventRow", "JobEvent", "TaskQ"] logger = structlog.get_logger("taskq.client._taskq") @@ -219,6 +220,7 @@ def __init__( self._reclaim_event_visibility_delay = reclaim_event_visibility_delay self._owns_pool = pool is None self._client: JobsClient | None = None + self._actors_client: ActorsClient | None = None # ── Lifecycle ────────────────────────────────────────────────────────── @@ -273,6 +275,7 @@ async def open(self) -> None: if self._redis_url is not None: settings.redis_url = self._redis_url # type: ignore[assignment] # Why: dotenvmodel PostgresDsn/RedisDsn fields accept str values at runtime but pyright cannot verify the coercion through the model's __setattr__. self._client = JobsClient(backend, settings=settings) + self._actors_client = ActorsClient(pool, schema=self._schema) if self._redis_client is not None: self._client._redis_client = self._redis_client # pyright: ignore[reportPrivateUsage] # Why: TaskQ owns the JobsClient lifecycle; assigning the caller-owned redis_client directly bypasses _open_redis so the client is NOT entered on the exit stack — TaskQ.close() must not close a caller-owned client. elif self._redis_url is not None: @@ -287,6 +290,7 @@ async def close(self) -> None: if self._client is not None: await self._client.close() self._client = None + self._actors_client = None if self._owns_pool and self._pool is not None: # Why bounded: an enqueue in flight at close time can stall # Pool.close() indefinitely against a dead PG. @@ -309,6 +313,21 @@ def _require_open(self) -> JobsClient: ) return self._client + # ── Actor configuration ──────────────────────────────────────────────── + + @property + def actors(self) -> ActorsClient: + """Actor configuration client — list, get, set capacity, deregister. + + Raises RuntimeError if called before ``open()`` or outside an + ``async with`` block. + """ + if self._actors_client is None: + raise RuntimeError( + "TaskQ is not open. Call 'await tq.open()' or use 'async with TaskQ(...) as tq:'" + ) + return self._actors_client + # ── Job operations ───────────────────────────────────────────────────── async def enqueue[P: BaseModel, R: BaseModel | None]( diff --git a/src/taskq/exceptions.py b/src/taskq/exceptions.py index 6b0b107e..8587c5ea 100644 --- a/src/taskq/exceptions.py +++ b/src/taskq/exceptions.py @@ -343,6 +343,82 @@ def __init__(self, drifts: tuple[ActorConfigDriftError, ...]) -> None: super().__init__("\n".join(lines)) +class ActorDeregistrationError(TaskQError): + """Base for actor deregistration refusals.""" + + def __init__(self, actor: str, detail: str) -> None: + self.actor = actor + super().__init__(f"Cannot deregister actor {actor!r}: {detail}") + + +class ActorHasActiveJobsError(ActorDeregistrationError): + """Non-terminal jobs reference the actor. + + Carries the count and per-status breakdown of the blocking jobs so the + caller can decide whether to cancel them first or use ``force=True``. + + When *force* is ``True``, the message reflects that running jobs + cannot be cancelled by ``force=True`` — the caller must wait for + them to finish or cancel them individually first. + """ + + def __init__( + self, + actor: str, + active_count: int, + status_counts: dict[str, int], + *, + force: bool = False, + ) -> None: + self.active_count = active_count + self.status_counts = status_counts + if force: + detail = ( + f"{active_count} running job(s) still reference this actor" + f" (breakdown: {status_counts}). Running jobs cannot be" + f" cancelled by force=True \u2014 wait for them to finish or" + f" cancel them individually first." + ) + else: + detail = ( + f"{active_count} non-terminal job(s) still reference this actor" + f" (breakdown: {status_counts}). Cancel them first or pass" + f" force=True to cancel pending/scheduled jobs automatically." + ) + super().__init__(actor, detail) + + +class ActorHasEnabledSchedulesError(ActorDeregistrationError): + """Enabled cron schedules reference the actor. + + Carries the schedule IDs so the caller can disable or delete them first. + """ + + def __init__( + self, + actor: str, + schedule_ids: list[str], + ) -> None: + self.schedule_ids = schedule_ids + detail = ( + f"{len(schedule_ids)} enabled cron schedule(s) reference this actor" + f" (ids: {schedule_ids}). Disable or delete them first or pass" + f" force=True to disable them automatically." + ) + super().__init__(actor, detail) + + +class ActorNotFoundError(ActorDeregistrationError): + """The actor_config row does not exist — nothing to deregister. + + Currently raised only by :func:`deregister_actor`. Other ops + (``get``, ``set_capacity``) return ``None`` for missing rows. + """ + + def __init__(self, actor: str) -> None: + super().__init__(actor, "no stored actor_config row for this actor") + + class PartialBatchError(TaskQError): """Raised when an autonomous enqueue_batch partially fails. diff --git a/src/taskq/testing/_runner.py b/src/taskq/testing/_runner.py index e8b46dc4..1fdbb8e5 100644 --- a/src/taskq/testing/_runner.py +++ b/src/taskq/testing/_runner.py @@ -20,6 +20,7 @@ import structlog from pydantic import BaseModel +from taskq.actor_config import ActorConfig from taskq.backend._protocol import ( EventRow, JobId, @@ -31,7 +32,6 @@ from taskq.context import JobContext from taskq.exceptions import Snooze from taskq.retry import OnRetryExhausted, OnSuccess, RetryClassifierHook, RetryPolicy -from taskq.worker.actor_config import ActorConfig if TYPE_CHECKING: from taskq.testing.in_memory import InMemoryBackend diff --git a/src/taskq/testing/in_memory.py b/src/taskq/testing/in_memory.py index 56d663fc..cf8f90c2 100644 --- a/src/taskq/testing/in_memory.py +++ b/src/taskq/testing/in_memory.py @@ -30,6 +30,7 @@ from pydantic import BaseModel from taskq._ids import new_uuid +from taskq.actor_config import ActorConfig from taskq.backend._cursor import decode_cursor, encode_cursor from taskq.backend._notify import _SubscriberContext from taskq.backend._protocol import ( @@ -121,7 +122,6 @@ _write_attempt, _write_cancel_escalation, ) -from taskq.worker.actor_config import ActorConfig if TYPE_CHECKING: from taskq.worker.leader import ArchiveExpiryResult, PruneResult diff --git a/src/taskq/web/admin/_factory.py b/src/taskq/web/admin/_factory.py index 7547a10b..2906b608 100644 --- a/src/taskq/web/admin/_factory.py +++ b/src/taskq/web/admin/_factory.py @@ -147,8 +147,16 @@ def get_backend(request: Request) -> Backend | None: def get_schema(request: Request) -> str: - """Dependency: yields the schema name from ``app.state``.""" + """Dependency: yields the schema name from ``app.state``. + + Re-validates against :data:`_IDENT_RE` as defence-in-depth — the schema + was validated at ``create_router`` construction time, but this ensures a + runtime mutation of ``app.state.schema`` (e.g. by a misconfigured test + fixture) cannot reach SQL interpolation. + """ s: str = request.app.state.schema + if not _IDENT_RE.match(s): + raise HTTPException(status_code=500, detail="invalid schema configuration") return s diff --git a/src/taskq/web/admin/actors.py b/src/taskq/web/admin/actors.py new file mode 100644 index 00000000..9a93036e --- /dev/null +++ b/src/taskq/web/admin/actors.py @@ -0,0 +1,87 @@ +"""Actors overview and deregister admin pages.""" + +import asyncpg +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import HTMLResponse, RedirectResponse +from jinja2 import Environment + +from taskq.actor_config_ops import deregister_actor, list_actor_summaries +from taskq.exceptions import ActorDeregistrationError, ActorNotFoundError +from taskq.settings import TaskQSettings +from taskq.web.admin._factory import ( + get_base_path, + get_csrf_token, + get_pg_pool, + get_realtime_ctx, + get_schema, + get_settings, + get_templates, + validate_csrf, +) + +_NOTICE_MESSAGES: dict[str, str] = { + "deregistered": "Actor deregistered successfully.", +} + + +def register(router: APIRouter) -> None: + """Attach actors overview and deregister routes to *router*.""" + + @router.get("/actors", response_class=HTMLResponse) + async def actors_overview( # pyright: ignore[reportUnusedFunction] # Why: registered via FastAPI decorator; pyright cannot see the route registration. + pool: asyncpg.Pool = Depends(get_pg_pool), + schema: str = Depends(get_schema), + tmpl: Environment = Depends(get_templates), + realtime_ctx: tuple[str, str] = Depends(get_realtime_ctx), + csrf_token: str = Depends(get_csrf_token), + notice: str | None = None, + ) -> HTMLResponse: + actors: list[dict[str, object]] = [] + async with pool.acquire() as conn: + actors = await list_actor_summaries(conn, schema=schema) + realtime_mode, mode_label = realtime_ctx + notice_text: str | None = _NOTICE_MESSAGES.get(notice) if notice else None + html = tmpl.get_template("actors.html").render( + actors=actors, + realtime_mode=realtime_mode, + mode_label=mode_label, + csrf_token=csrf_token, + active_page="actors", + notice=notice_text, + ) + return HTMLResponse(content=html) + + # Note: {actor} matches a single path segment — actor names containing "/" + # cannot be deregistered via the admin UI (use the CLI or client API instead). + # This is an accepted limitation; %2F in URLs is decoded before routing. + @router.post("/actors/{actor}/deregister") + async def actor_deregister( # pyright: ignore[reportUnusedFunction] # Why: registered via FastAPI decorator; pyright cannot see the route registration. + actor: str, + request: Request, + _csrf: None = Depends(validate_csrf), + pool: asyncpg.Pool = Depends(get_pg_pool), + schema: str = Depends(get_schema), + base_path: str = Depends(get_base_path), + settings: TaskQSettings = Depends(get_settings), + ) -> RedirectResponse: + if not settings.admin_actions_enabled: + raise HTTPException(status_code=403, detail="Admin actions are disabled") + + form = await request.form() + force = form.get("force") == "true" + purge_queue = form.get("purge_queue") == "true" + + async with pool.acquire() as conn: + try: + await deregister_actor( + conn, actor, force=force, purge_queue=purge_queue, schema=schema + ) + except ActorNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from None + except ActorDeregistrationError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from None + + return RedirectResponse( + url=f"{base_path}/actors?notice=deregistered", + status_code=303, + ) diff --git a/src/taskq/web/templates/_base.html b/src/taskq/web/templates/_base.html index 53eaaa54..b13615e2 100644 --- a/src/taskq/web/templates/_base.html +++ b/src/taskq/web/templates/_base.html @@ -51,6 +51,9 @@

TaskQ Admin

Workers + Actors Schedules diff --git a/src/taskq/web/templates/actors.html b/src/taskq/web/templates/actors.html new file mode 100644 index 00000000..280c6493 --- /dev/null +++ b/src/taskq/web/templates/actors.html @@ -0,0 +1,79 @@ +{% extends "_base.html" %} +{% block title %}Actors — TaskQ Admin{% endblock %} +{% block head %} + +{% endblock %} +{% block content %} +
+

Actors

+ {% if notice %} +
+ {{ notice }} +
+ {% endif %} + {% if actors %} +
+ + + + + + + + + + + + + + + {% for a in actors %} + + + + + + + + + + + {% endfor %} + +
ActorQueueMax ConcurrentMax PendingActive JobsSchedulesUpdatedActions
{{ a.actor }}{{ a.queue }}{{ a.max_concurrent if a.max_concurrent is not none else '∞' }}{{ a.max_pending or '—' }} + + {{ a.active_job_count }} + + {{ a.enabled_schedule_count }}{{ a.updated_at | time_ago }} +
+ + + + +
+
+
+ {% else %} +

No actor_config rows.

+ {% endif %} +
+{% endblock %} diff --git a/src/taskq/worker/__init__.py b/src/taskq/worker/__init__.py index 8f46fb39..81f6b4bf 100644 --- a/src/taskq/worker/__init__.py +++ b/src/taskq/worker/__init__.py @@ -1,7 +1,7 @@ """Worker subsystem: pool management, deps, heartbeat, and connection budgeting. All imports are lazy via ``__getattr__`` so that importing any submodule -(e.g. ``taskq.worker.actor_config``) does not pull in ``asyncpg`` or +(e.g. ``taskq.worker.budget``) does not pull in ``asyncpg`` or concrete backends. This keeps ``import taskq.testing`` lean. The names below are also imported under ``TYPE_CHECKING`` so static tools diff --git a/src/taskq/worker/_bootstrap.py b/src/taskq/worker/_bootstrap.py index 040020a1..3fad43c4 100644 --- a/src/taskq/worker/_bootstrap.py +++ b/src/taskq/worker/_bootstrap.py @@ -23,6 +23,7 @@ from taskq._di.scopes import LoopScope, ProcessScope, ThreadScope, make_resolver from taskq._dsn import dsn_host as _dsn_host from taskq.actor import ActorRef +from taskq.actor_config import ActorConfig from taskq.backend._protocol import Backend, JobRow, ScheduleCreateArgs from taskq.backend.clock import Clock, SystemClock from taskq.backend.postgres import PostgresBackend @@ -42,7 +43,6 @@ from taskq.ratelimit.registry import registry as rl_registry from taskq.settings import WorkerSettings from taskq.worker._watchdog import LoopLagWatchdog, ShutdownWatchdog, loop_watchdog_loop -from taskq.worker.actor_config import ActorConfig from taskq.worker.cancel import make_cancel_controller from taskq.worker.deps import WorkerDeps, open_worker_deps from taskq.worker.health import HealthServer diff --git a/src/taskq/worker/actor_config_ops.py b/src/taskq/worker/actor_config_ops.py deleted file mode 100644 index ceeb51eb..00000000 --- a/src/taskq/worker/actor_config_ops.py +++ /dev/null @@ -1,217 +0,0 @@ -"""Operator surface for reading and tuning stored `{schema}.actor_config` rows. - -Complements :func:`taskq.worker.startup.sync_actor_config`: that function -only ever *seeds* a row's capacity fields (``max_concurrent``, -``max_pending``, ``result_ttl``) on first registration and otherwise -leaves them untouched. This module is how an operator changes them -afterwards — on a live deployment, without a code change or a worker -restart. All three are re-read by the engine without a restart: - -* ``max_concurrent`` — the dispatch query joins ``actor_config`` fresh - on every dispatch cycle (``taskq/backend/_dispatch_sql.py``); a change - is effective immediately. -* ``result_ttl`` — the terminal-write UPDATE recomputes - ``result_expires_at`` from the stored value for every completing job - (``taskq/backend/_sql_templates.py::mark_succeeded``); a change is - effective for jobs completing after the write. -* ``max_pending`` — enqueue-side processes hold a TTL-bounded cache of - this table (``taskq/client/_capacity.py``, default 5s staleness); a - change is effective fleet-wide within seconds, with no redeploy. - -Clearing semantics differ by field on purpose. ``--clear-max-concurrent`` -writes NULL, which the dispatch SQL reads as *unlimited* — the SQL -cannot see the code literal once the row exists. ``--clear-max-pending`` -and ``--clear-result-ttl`` write NULL, which their enforcement paths -read as *fall back to the ``@actor(...)`` literal* — clearing reverts an -override to the code default. -""" - -import math -from dataclasses import dataclass -from typing import Final - -import asyncpg - -from taskq._json import loads -from taskq.backend._protocol import ConnLike -from taskq.constants import ( - _IDENT_RE, # pyright: ignore[reportPrivateUsage] # Why: reusing the canonical identifier regex rather than redefining it -) - -__all__ = [ - "UNSET", - "ActorConfigRow", - "Unset", - "get_actor_config", - "list_actor_configs", - "set_actor_config_capacity", -] - - -class Unset: - """Sentinel distinguishing 'leave unchanged' from an explicit ``None`` (clear).""" - - def __repr__(self) -> str: - return "UNSET" - - -UNSET: Final = Unset() - - -@dataclass(frozen=True, slots=True) -class ActorConfigRow: - """Snapshot of one `{schema}.actor_config` row.""" - - actor: str - max_concurrent: int | None - max_pending: int | None - queue: str - result_ttl: float | None - metadata: dict[str, object] - updated_at: str - - -_LIST_ACTOR_CONFIG_SQL = """ -SELECT actor, max_concurrent, max_pending, queue, result_ttl, - metadata::text AS metadata, updated_at::text AS updated_at - FROM "{schema}".actor_config - ORDER BY actor -""".strip() - -_GET_ACTOR_CONFIG_SQL = """ -SELECT actor, max_concurrent, max_pending, queue, result_ttl, - metadata::text AS metadata, updated_at::text AS updated_at - FROM "{schema}".actor_config - WHERE actor = $1 -""".strip() - -# Each capacity column is only overwritten when its paired boolean -# "touch" flag is true; otherwise the CASE expression preserves the -# current value. This lets one statement express "set to N", "clear to -# NULL", and "leave alone" for all three fields without dynamic SQL. -_SET_ACTOR_CONFIG_CAPACITY_SQL = """ -UPDATE "{schema}".actor_config - SET max_concurrent = CASE WHEN $2 THEN $3 ELSE max_concurrent END, - max_pending = CASE WHEN $4 THEN $5 ELSE max_pending END, - result_ttl = CASE WHEN $6 THEN $7 ELSE result_ttl END, - updated_at = now() - WHERE actor = $1 -RETURNING actor, max_concurrent, max_pending, queue, result_ttl, - metadata::text AS metadata, updated_at::text AS updated_at -""".strip() - - -def _row_to_dataclass(row: asyncpg.Record) -> ActorConfigRow: - return ActorConfigRow( - actor=row["actor"], - max_concurrent=row["max_concurrent"], - max_pending=row["max_pending"], - queue=row["queue"], - result_ttl=row["result_ttl"], - metadata=loads(row["metadata"]), - updated_at=row["updated_at"], - ) - - -async def list_actor_configs(conn: ConnLike, *, schema: str = "taskq") -> list[ActorConfigRow]: - """Return every stored `{schema}.actor_config` row, ordered by actor name.""" - if not _IDENT_RE.match(schema): - raise ValueError(f"invalid schema identifier: {schema!r}") - rows = await conn.fetch(_LIST_ACTOR_CONFIG_SQL.format(schema=schema)) - return [_row_to_dataclass(row) for row in rows] - - -async def get_actor_config( - conn: ConnLike, actor: str, *, schema: str = "taskq" -) -> ActorConfigRow | None: - """Return the stored row for *actor*, or ``None`` if it has never been synced.""" - if not _IDENT_RE.match(schema): - raise ValueError(f"invalid schema identifier: {schema!r}") - row = await conn.fetchrow(_GET_ACTOR_CONFIG_SQL.format(schema=schema), actor) - return _row_to_dataclass(row) if row is not None else None - - -def _validate_int_field(name: str, value: int | None | Unset) -> None: - """Reject the two shapes that slip past ``isinstance(x, int) and x < 0``: - ``bool`` (an ``int`` subclass — ``False`` would be written as 0, flooring - the dispatch residual ``GREATEST(cap - in_flight, 0)`` and silently - pausing the actor) and negative values.""" - if isinstance(value, bool): - raise ValueError(f"{name} must be a non-negative integer; got {value!r} (bool)") - if isinstance(value, int) and value < 0: - raise ValueError(f"{name} must be a non-negative integer; got {value!r}") - - -def _validate_result_ttl(value: float | None | Unset) -> None: - """Reject bool, negative, and non-finite ``result_ttl``. - - NaN sails through ``value < 0`` (NaN compares False) and then breaks - every completion for the actor — ``now() + NaN * interval '1 second'`` - raises ``interval out of range`` in the terminal-write UPDATE. ±inf is - rejected on the same grounds (``interval out of range`` / meaningless - expiry).""" - if isinstance(value, bool): - raise ValueError( - f"result_ttl must be a non-negative number of seconds; got {value!r} (bool)" - ) - if isinstance(value, (int, float)): - if not math.isfinite(value): - raise ValueError(f"result_ttl must be finite; got {value!r}") - if value < 0: - raise ValueError(f"result_ttl must be a non-negative number of seconds; got {value!r}") - - -async def set_actor_config_capacity( - conn: ConnLike, - actor: str, - *, - max_concurrent: int | None | Unset = UNSET, - max_pending: int | None | Unset = UNSET, - result_ttl: float | None | Unset = UNSET, - schema: str = "taskq", -) -> ActorConfigRow | None: - """Update capacity fields on an existing `{schema}.actor_config` row. - - Only fields passed as something other than :data:`UNSET` are - changed. Pass ``None`` explicitly to clear a field — precisely what - that means depends on the field's enforcement path: clearing - ``max_concurrent`` makes the actor *unlimited* (the dispatch SQL - reads a stored NULL as no cap), while clearing ``max_pending`` or - ``result_ttl`` reverts enforcement to the ``@actor(...)`` literal - (those paths can still see the code default). Returns ``None`` if - *actor* has no stored row — a row is only created by - :func:`taskq.worker.startup.sync_actor_config` at worker startup, so - an actor must have been registered by at least one worker before its - capacity can be tuned here. - - Raises :class:`ValueError` if ``max_concurrent`` or ``max_pending`` - is a negative integer or a ``bool`` — the same guard ``@actor(...)`` - applies at decoration time (``taskq/actor.py``), plus the bool case - (``False`` is an ``int`` and would be written as 0). Without these, - an operator typo here (e.g. ``--max-concurrent -5``) would write - silently into the dispatch CTE's ``GREATEST(ac.max_concurrent - - in_flight, 0)`` residual calculation, floor to zero, and pause the - actor indefinitely with no error anywhere in the path. ``result_ttl`` - is likewise rejected when negative, non-finite, or ``bool`` — a - negative TTL would set ``result_expires_at`` to a past timestamp in - the terminal-write UPDATE (silently expiring every result the moment - it is written), and NaN/±inf raise ``interval out of range`` in that - same UPDATE, failing every completion for the actor. - """ - if not _IDENT_RE.match(schema): - raise ValueError(f"invalid schema identifier: {schema!r}") - _validate_int_field("max_concurrent", max_concurrent) - _validate_int_field("max_pending", max_pending) - _validate_result_ttl(result_ttl) - - row = await conn.fetchrow( - _SET_ACTOR_CONFIG_CAPACITY_SQL.format(schema=schema), - actor, - not isinstance(max_concurrent, Unset), - None if isinstance(max_concurrent, Unset) else max_concurrent, - not isinstance(max_pending, Unset), - None if isinstance(max_pending, Unset) else max_pending, - not isinstance(result_ttl, Unset), - None if isinstance(result_ttl, Unset) else result_ttl, - ) - return _row_to_dataclass(row) if row is not None else None diff --git a/src/taskq/worker/startup.py b/src/taskq/worker/startup.py index f81e9f49..f8103db9 100644 --- a/src/taskq/worker/startup.py +++ b/src/taskq/worker/startup.py @@ -6,13 +6,13 @@ import structlog from taskq._json import dumps_str, loads +from taskq.actor_config import ActorConfig from taskq.backend._protocol import ConnLike from taskq.constants import ( _IDENT_RE, # pyright: ignore[reportPrivateUsage] # Why: reusing the canonical identifier regex rather than redefining it ) from taskq.exceptions import ActorConfigDriftError, ActorConfigDriftList from taskq.obs import get_logger -from taskq.worker.actor_config import ActorConfig logger: structlog.stdlib.BoundLogger = get_logger(__name__) diff --git a/tests/e2e/test_actor_deregistration.py b/tests/e2e/test_actor_deregistration.py new file mode 100644 index 00000000..29423abf --- /dev/null +++ b/tests/e2e/test_actor_deregistration.py @@ -0,0 +1,167 @@ +"""E2E: actor deregistration lifecycle with a real worker. + +Each test uses a **different actor** to avoid cross-test interference: +``e2e_worker`` is module-scoped and ``sync_actor_config`` runs only at +bootstrap, so once a test deregisters an actor's ``actor_config`` row, +later tests cannot enqueue to that same actor (the dispatch query +inner-joins ``actor_config`` — jobs would never be dispatched). + +Actors used (all defined in ``tests/e2e/actors.py``): +- ``quick_result`` — 0.05 s sleep, simple payload/result. +- ``long_running_job`` — 30 s sleep. Used for the refusal-with-active-jobs test. +- ``short_lived_job`` — 0.5 s sleep. Used for the force+purge_queue test. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from ._assertions import poll_until, wait_for_handle_status +from .actors import ( + LongRunningPayload, + QuickResultPayload, + ShortJobPayload, + long_running_job, + quick_result, + short_lived_job, +) + +if TYPE_CHECKING: + import asyncpg + + from taskq import TaskQ + + from .conftest import E2ESchema, E2EWorker + +pytestmark = [pytest.mark.e2e, pytest.mark.timeout(900)] + + +async def test_deregister_after_jobs_complete( + e2e_client: TaskQ, + e2e_worker: E2EWorker, + e2e_pg_pool: asyncpg.Pool, + e2e_schema: E2ESchema, + run_id: str, +) -> None: + """Deregister an actor after all its jobs are terminal.""" + schema = e2e_schema.schema_name + actor_name = quick_result.name + + handle = await e2e_client.enqueue(quick_result, QuickResultPayload(run_id=run_id, value="test")) + await handle.wait(timeout=60) + + ac_count = await e2e_pg_pool.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', + actor_name, + ) + assert ac_count == 1 + + result = await e2e_client.actors.deregister(actor_name) + + assert result.actor_config_deleted is True + assert result.terminal_jobs_remaining == 1 + + ac_count = await e2e_pg_pool.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', + actor_name, + ) + assert ac_count == 0 + + job_count = await e2e_pg_pool.fetchval( + f"SELECT count(*) FROM \"{schema}\".jobs WHERE actor = $1 AND status = 'succeeded'", + actor_name, + ) + assert job_count == 1 + + +async def test_deregister_refuses_with_active_jobs( + e2e_client: TaskQ, + e2e_worker: E2EWorker, + e2e_pg_pool: asyncpg.Pool, + e2e_schema: E2ESchema, + run_id: str, +) -> None: + """Deregistration refuses when a job is running.""" + from taskq.exceptions import ActorHasActiveJobsError + + schema = e2e_schema.schema_name + actor_name = long_running_job.name + + handle = await e2e_client.enqueue(long_running_job, LongRunningPayload(run_id=run_id)) + + async def _is_running() -> bool: + status = await e2e_pg_pool.fetchval( + f'SELECT status FROM "{schema}".jobs WHERE id = $1', + handle.job_id, + ) + return status == "running" + + await poll_until(_is_running, timeout=30.0, interval=0.5) + + with pytest.raises(ActorHasActiveJobsError) as exc_info: + await e2e_client.actors.deregister(actor_name) + + assert exc_info.value.actor == actor_name + assert exc_info.value.active_count == 1 + assert "running" in exc_info.value.status_counts + + ac_count = await e2e_pg_pool.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', + actor_name, + ) + assert ac_count == 1 + + # Cleanup: cancel the job, wait for terminal, then force-deregister. + # Do NOT use handle.wait() — it raises JobFailed for cancelled status. + # long_running_job never calls ctx.check_cancelled(), so the cancel + # lands only after the 30s sleep finishes — budget the full duration. + await handle.cancel() + await wait_for_handle_status(handle, "cancelled", timeout=60) + + result = await e2e_client.actors.deregister(actor_name, force=True) + assert result.actor_config_deleted is True + + ac_count = await e2e_pg_pool.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', + actor_name, + ) + assert ac_count == 0 + + +async def test_deregister_force_with_purge_queue_after_completion( + e2e_client: TaskQ, + e2e_worker: E2EWorker, + e2e_pg_pool: asyncpg.Pool, + e2e_schema: E2ESchema, + run_id: str, +) -> None: + """force=True + purge_queue=True deregister succeeds when all jobs are terminal. + + Uses short_lived_job (0.5s sleep). Enqueues a job, waits for completion, + then force-deregisters with purge_queue. The force path cancels 0 jobs + (none are pending) and the queue is purged if no other actor uses it. + """ + schema = e2e_schema.schema_name + actor_name = short_lived_job.name + + handle = await e2e_client.enqueue( + short_lived_job, ShortJobPayload(run_id=run_id, label="force-test") + ) + await handle.wait(timeout=60) + + result = await e2e_client.actors.deregister(actor_name, force=True, purge_queue=True) + + assert result.actor_config_deleted is True + assert result.jobs_cancelled == 0 + assert result.terminal_jobs_remaining == 1 + # All e2e actors share queue="e2e" — purge_queue=True is a safe no-op + # because the orphan guard correctly refuses to delete a shared queue. + assert result.queue_purged is False + + ac_count = await e2e_pg_pool.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', + actor_name, + ) + assert ac_count == 0 diff --git a/tests/test_actor_capacity_pg.py b/tests/test_actor_capacity_pg.py index 4fc03e1f..0f1c0d23 100644 --- a/tests/test_actor_capacity_pg.py +++ b/tests/test_actor_capacity_pg.py @@ -17,11 +17,11 @@ from pydantic import BaseModel from taskq.actor import actor +from taskq.actor_config import ActorConfig +from taskq.actor_config_ops import set_actor_config_capacity from taskq.client import JobsClient from taskq.exceptions import MaxPendingExceededError from taskq.testing.fixtures import JobsApp -from taskq.worker.actor_config import ActorConfig -from taskq.worker.actor_config_ops import set_actor_config_capacity from taskq.worker.startup import sync_actor_config pytestmark = [pytest.mark.asyncio, pytest.mark.integration] diff --git a/tests/test_actor_config.py b/tests/test_actor_config.py index 2a7ea55b..53d439e5 100644 --- a/tests/test_actor_config.py +++ b/tests/test_actor_config.py @@ -4,7 +4,7 @@ import pytest -from taskq.worker.actor_config import ActorConfig +from taskq.actor_config import ActorConfig # ── Construction ─────────────────────────────────────────────────────────── diff --git a/tests/test_actor_config_ops.py b/tests/test_actor_config_ops.py index 0a2e5706..7b5ba7a2 100644 --- a/tests/test_actor_config_ops.py +++ b/tests/test_actor_config_ops.py @@ -1,4 +1,4 @@ -"""Tests for `taskq.worker.actor_config_ops`: the operator read/tune surface +"""Tests for `taskq.actor_config_ops`: the operator read/tune surface that backs `taskq actor-config get/set/list`. These are integration-tier (real Postgres) because the whole point of the @@ -11,8 +11,8 @@ import pytest from taskq._ids import new_base62 -from taskq.worker.actor_config import ActorConfig -from taskq.worker.actor_config_ops import ( +from taskq.actor_config import ActorConfig +from taskq.actor_config_ops import ( get_actor_config, list_actor_configs, set_actor_config_capacity, diff --git a/tests/test_actor_config_ops_validation.py b/tests/test_actor_config_ops_validation.py index a5e47bbe..15af83cc 100644 --- a/tests/test_actor_config_ops_validation.py +++ b/tests/test_actor_config_ops_validation.py @@ -19,7 +19,7 @@ import pytest -from taskq.worker.actor_config_ops import UNSET, set_actor_config_capacity +from taskq.actor_config_ops import UNSET, set_actor_config_capacity class _RecordingConn: diff --git a/tests/test_actor_config_sync.py b/tests/test_actor_config_sync.py index 19efa276..ee95c697 100644 --- a/tests/test_actor_config_sync.py +++ b/tests/test_actor_config_sync.py @@ -12,8 +12,8 @@ from taskq._ids import new_base62 from taskq._json import dumps_str +from taskq.actor_config import ActorConfig from taskq.exceptions import ActorConfigDriftList -from taskq.worker.actor_config import ActorConfig from taskq.worker.startup import sync_actor_config diff --git a/tests/test_actor_deregistration.py b/tests/test_actor_deregistration.py new file mode 100644 index 00000000..1a91c748 --- /dev/null +++ b/tests/test_actor_deregistration.py @@ -0,0 +1,681 @@ +"""Integration tests for ``deregister_actor`` — the transactional +``actor_config`` deletion with safety checks (force=False / force=True, +purge_queue). + +These require real Postgres (marked ``integration``) because the function +executes hand-written SQL against the fully migrated schema — a fake +connection would only prove the query string looks right, not that +Postgres executes it correctly with enum casts, transactional rollback, +and the ``NOT EXISTS`` subquery for queue purging. +""" + +import asyncio +from uuid import uuid4 + +import asyncpg +import pytest + +from taskq.actor_config import ActorConfig +from taskq.actor_config_ops import DeregisterResult, deregister_actor, get_actor_config +from taskq.exceptions import ( + ActorHasActiveJobsError, + ActorHasEnabledSchedulesError, + ActorNotFoundError, +) +from taskq.testing.fixtures import ModulePgSchema +from taskq.worker.startup import sync_actor_config + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +# ── Helpers ────────────────────────────────────────────────────────────── + + +async def _insert_job( + conn: asyncpg.Connection, + schema: str, + *, + actor: str, + status: str, + queue: str = "default", +) -> str: + """Insert a minimal job row with the given status; return its UUID string.""" + job_id = uuid4() + await conn.execute( + f"""INSERT INTO "{schema}".jobs ( + id, actor, queue, payload, max_attempts, retry_kind, status + ) VALUES ( + $1, $2, $3, $4::jsonb, $5, $6, $7::"{schema}".job_status + )""", # noqa: S608 # Why: schema validated by _IDENT_RE in apply_pending; actor/status/queue are test constants, not user input. + job_id, + actor, + queue, + '{"key": "value"}', + 3, + "transient", + status, + ) + return str(job_id) + + +async def _insert_schedule( + conn: asyncpg.Connection, + schema: str, + *, + actor: str, + enabled: bool = True, +) -> str: + """Insert a cron schedule row; return its UUID string.""" + schedule_id = uuid4() + await conn.execute( + f"""INSERT INTO "{schema}".cron_schedules ( + id, actor, cron_expr, enabled, next_fire_at + ) VALUES ( + $1, $2, $3, $4, now() + )""", # noqa: S608 # Why: schema validated by _IDENT_RE in apply_pending; actor is a test constant. + schedule_id, + actor, + "*/5 * * * *", + enabled, + ) + return str(schedule_id) + + +async def _insert_queue( + conn: asyncpg.Connection, + schema: str, + name: str, +) -> None: + """Insert a queue row (the ``queues`` table is not populated by sync_actor_config).""" + await conn.execute( + f'INSERT INTO "{schema}".queues (name) VALUES ($1)', # noqa: S608 # Why: schema validated by _IDENT_RE in apply_pending; name is a test constant. + name, + ) + + +async def _queue_exists(conn: asyncpg.Connection, schema: str, name: str) -> bool: + """Check whether a queue row exists.""" + return bool( + await conn.fetchval( + f'SELECT 1 FROM "{schema}".queues WHERE name = $1', # noqa: S608 + name, + ) + ) + + +async def _job_status(conn: asyncpg.Connection, schema: str, job_id: str) -> str: + """Return the current status of a job row.""" + return str( + await conn.fetchval( + f'SELECT status::text FROM "{schema}".jobs WHERE id = $1', # noqa: S608 + job_id, + ) + ) + + +async def _schedule_state(conn: asyncpg.Connection, schema: str, schedule_id: str) -> str | None: + """Return 'enabled', 'disabled', or None (deleted).""" + row = await conn.fetchrow( + f'SELECT enabled FROM "{schema}".cron_schedules WHERE id = $1', # noqa: S608 + schedule_id, + ) + if row is None: + return None + return "enabled" if row["enabled"] else "disabled" + + +# ── force=False path ──────────────────────────────────────────────────── + + +async def test_deregister_raises_not_found_for_unknown_actor( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + + with pytest.raises(ActorNotFoundError, match="no stored actor_config row"): + await deregister_actor(clean_pg_conn, "ghost", schema=schema) + + +async def test_deregister_succeeds_when_no_jobs_or_schedules( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="clean_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + + result = await deregister_actor(clean_pg_conn, "clean_actor", schema=schema) + + assert isinstance(result, DeregisterResult) + assert result.actor == "clean_actor" + assert result.queue == "default" + assert result.actor_config_deleted is True + assert result.schedules_disabled == 0 + assert result.jobs_cancelled == 0 + assert result.terminal_jobs_remaining == 0 + assert result.queue_purged is False + + assert await get_actor_config(clean_pg_conn, "clean_actor", schema=schema) is None + + +async def test_deregister_refuses_with_pending_jobs( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="busy_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + await _insert_job(clean_pg_conn, schema, actor="busy_actor", status="pending") + + with pytest.raises(ActorHasActiveJobsError) as exc_info: + await deregister_actor(clean_pg_conn, "busy_actor", schema=schema) + + assert exc_info.value.active_count == 1 + assert exc_info.value.status_counts == {"pending": 1} + + # Row must still exist — the transaction rolled back. + assert await get_actor_config(clean_pg_conn, "busy_actor", schema=schema) is not None + + +async def test_deregister_refuses_with_running_jobs( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="run_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + await _insert_job(clean_pg_conn, schema, actor="run_actor", status="running") + + with pytest.raises(ActorHasActiveJobsError) as exc_info: + await deregister_actor(clean_pg_conn, "run_actor", schema=schema) + + assert exc_info.value.active_count == 1 + assert exc_info.value.status_counts == {"running": 1} + assert await get_actor_config(clean_pg_conn, "run_actor", schema=schema) is not None + + +async def test_deregister_refuses_with_enabled_schedules( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="sched_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + schedule_id = await _insert_schedule(clean_pg_conn, schema, actor="sched_actor", enabled=True) + + with pytest.raises(ActorHasEnabledSchedulesError) as exc_info: + await deregister_actor(clean_pg_conn, "sched_actor", schema=schema) + + assert exc_info.value.schedule_ids == [schedule_id] + assert await get_actor_config(clean_pg_conn, "sched_actor", schema=schema) is not None + + +async def test_deregister_succeeds_with_disabled_schedules( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="dis_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + await _insert_schedule(clean_pg_conn, schema, actor="dis_actor", enabled=False) + + result = await deregister_actor(clean_pg_conn, "dis_actor", schema=schema) + + assert result.actor_config_deleted is True + assert result.schedules_disabled == 0 + assert await get_actor_config(clean_pg_conn, "dis_actor", schema=schema) is None + + +async def test_deregister_succeeds_with_terminal_jobs( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="term_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + await _insert_job(clean_pg_conn, schema, actor="term_actor", status="succeeded") + await _insert_job(clean_pg_conn, schema, actor="term_actor", status="failed") + + result = await deregister_actor(clean_pg_conn, "term_actor", schema=schema) + + assert result.actor_config_deleted is True + assert result.terminal_jobs_remaining == 2 + assert await get_actor_config(clean_pg_conn, "term_actor", schema=schema) is None + + +# ── force=True path ───────────────────────────────────────────────────── + + +async def test_deregister_force_cancels_pending_and_disables_schedules( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="force_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + pending_id = await _insert_job(clean_pg_conn, schema, actor="force_actor", status="pending") + scheduled_id = await _insert_job(clean_pg_conn, schema, actor="force_actor", status="scheduled") + schedule_id = await _insert_schedule(clean_pg_conn, schema, actor="force_actor", enabled=True) + + result = await deregister_actor(clean_pg_conn, "force_actor", force=True, schema=schema) + + assert result.actor_config_deleted is True + assert result.jobs_cancelled == 2 + assert result.schedules_disabled == 1 + # The 2 cancelled jobs are now terminal — terminal_jobs_remaining + # counts all non-pending/scheduled/running rows, including the + # newly-cancelled ones. + assert result.terminal_jobs_remaining == 2 + + # Verify DB state directly. + assert await _job_status(clean_pg_conn, schema, pending_id) == "cancelled" + assert await _job_status(clean_pg_conn, schema, scheduled_id) == "cancelled" + assert await _schedule_state(clean_pg_conn, schema, schedule_id) == "disabled" + assert await get_actor_config(clean_pg_conn, "force_actor", schema=schema) is None + + # Verify audit trail fields on cancelled jobs (M12). + job_row = await clean_pg_conn.fetchrow( + f"SELECT error_class, error_message, finished_at " # noqa: S608 # Why: schema validated by _IDENT_RE; pending_id is a test-generated UUID. + f'FROM "{schema}".jobs WHERE id = $1', + pending_id, + ) + assert job_row is not None + assert job_row["error_class"] == "ActorDeregistered" + assert job_row["error_message"] is not None + assert "actor deregistration" in job_row["error_message"] + assert job_row["finished_at"] is not None + + +async def test_deregister_force_writes_job_events_on_cancel( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """H3: force=True cancel must insert job_events state_change rows — + every other cancel path does, and audit consumers rely on them.""" + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="evt_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + pending_id = await _insert_job(clean_pg_conn, schema, actor="evt_actor", status="pending") + scheduled_id = await _insert_job(clean_pg_conn, schema, actor="evt_actor", status="scheduled") + + await deregister_actor(clean_pg_conn, "evt_actor", force=True, schema=schema) + + for jid in (pending_id, scheduled_id): + events = await clean_pg_conn.fetch( + f"SELECT kind, detail::text AS detail " # noqa: S608 # Why: schema validated by _IDENT_RE; jid is a test UUID. + f' FROM "{schema}".job_events ' + f" WHERE job_id = $1 " + f" ORDER BY occurred_at", + jid, + ) + assert len(events) >= 1, f"no job_events for cancelled job {jid}" + state_changes = [e for e in events if e["kind"] == "state_change"] + assert len(state_changes) == 1 + import json + + detail = json.loads(state_changes[0]["detail"]) + assert detail["to_state"] == "cancelled" + assert detail["reason"] == "actor_deregistered" + + +async def test_deregister_force_refuses_with_running_jobs( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="frun_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + await _insert_job(clean_pg_conn, schema, actor="frun_actor", status="running") + + with pytest.raises(ActorHasActiveJobsError) as exc_info: + await deregister_actor(clean_pg_conn, "frun_actor", force=True, schema=schema) + + assert exc_info.value.active_count == 1 + assert exc_info.value.status_counts == {"running": 1} + assert await get_actor_config(clean_pg_conn, "frun_actor", schema=schema) is not None + + +async def test_deregister_force_with_running_and_pending_only_reports_running( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """force=True checks only running jobs — pending jobs are not in the error + because they would be cancelled, not blocking.""" + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="mix_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + await _insert_job(clean_pg_conn, schema, actor="mix_actor", status="running") + await _insert_job(clean_pg_conn, schema, actor="mix_actor", status="pending") + + with pytest.raises(ActorHasActiveJobsError) as exc_info: + await deregister_actor(clean_pg_conn, "mix_actor", force=True, schema=schema) + + assert exc_info.value.active_count == 1 + assert exc_info.value.status_counts == {"running": 1} + assert "pending" not in exc_info.value.status_counts + # Row still exists — transaction rolled back. + assert await get_actor_config(clean_pg_conn, "mix_actor", schema=schema) is not None + # The pending job must still be pending — the transaction rolled back + # on the raise, so the cancel UPDATE never committed. + pending_count = await clean_pg_conn.fetchval( + f"SELECT count(*) FROM \"{schema}\".jobs WHERE actor = $1 AND status = 'pending'", # noqa: S608 # Why: schema validated by _IDENT_RE in apply_pending; actor/status are test constants. + "mix_actor", + ) + assert pending_count == 1 + + +async def test_deregister_force_keeps_terminal_history( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Terminal job rows are never modified — only pending/scheduled are cancelled.""" + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="hist_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + pending_id = await _insert_job(clean_pg_conn, schema, actor="hist_actor", status="pending") + succeeded_id = await _insert_job(clean_pg_conn, schema, actor="hist_actor", status="succeeded") + failed_id = await _insert_job(clean_pg_conn, schema, actor="hist_actor", status="failed") + + result = await deregister_actor(clean_pg_conn, "hist_actor", force=True, schema=schema) + + assert result.jobs_cancelled == 1 + # terminal_jobs_remaining counts all terminal rows including the + # newly-cancelled pending job: 1 cancelled + 1 succeeded + 1 failed. + assert result.terminal_jobs_remaining == 3 + + assert await _job_status(clean_pg_conn, schema, pending_id) == "cancelled" + assert await _job_status(clean_pg_conn, schema, succeeded_id) == "succeeded" + assert await _job_status(clean_pg_conn, schema, failed_id) == "failed" + + +# ── purge_queue path ───────────────────────────────────────────────────── + + +async def test_deregister_purge_queue_deletes_orphaned_queue( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await _insert_queue(clean_pg_conn, schema, "solo_queue") + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="solo_actor", max_concurrent=5, queue="solo_queue")], + schema=schema, + ) + + result = await deregister_actor(clean_pg_conn, "solo_actor", purge_queue=True, schema=schema) + + assert result.queue == "solo_queue" + assert result.queue_purged is True + assert await _queue_exists(clean_pg_conn, schema, "solo_queue") is False + + +async def test_deregister_purge_queue_keeps_shared_queue( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await _insert_queue(clean_pg_conn, schema, "shared_queue") + await sync_actor_config( + clean_pg_conn, + [ + ActorConfig(actor="shared_a", max_concurrent=5, queue="shared_queue"), + ActorConfig(actor="shared_b", max_concurrent=5, queue="shared_queue"), + ], + schema=schema, + ) + + result = await deregister_actor(clean_pg_conn, "shared_a", purge_queue=True, schema=schema) + + assert result.queue == "shared_queue" + assert result.queue_purged is False + # The queue survives because shared_b still references it. + assert await _queue_exists(clean_pg_conn, schema, "shared_queue") is True + # shared_b's row must still exist. + assert await get_actor_config(clean_pg_conn, "shared_b", schema=schema) is not None + + +async def test_deregister_without_purge_queue_keeps_queue( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + schema = module_pg_schema.schema_name + await _insert_queue(clean_pg_conn, schema, "kept_queue") + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="keep_actor", max_concurrent=5, queue="kept_queue")], + schema=schema, + ) + + result = await deregister_actor(clean_pg_conn, "keep_actor", schema=schema) + + assert result.queue_purged is False + assert await _queue_exists(clean_pg_conn, schema, "kept_queue") is True + + +# ── idempotency ───────────────────────────────────────────────────────── + + +async def test_double_deregister_raises_not_found( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """A second deregister call on an already-deregistered actor raises ActorNotFoundError. + + This is the primary consumer pattern (cleanup loops using try/except ActorNotFoundError). + The idempotency guarantee must be tested — an implementation bug that silently returns + actor_config_deleted=False instead of raising would not be caught otherwise. + """ + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="idem_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + + result = await deregister_actor(clean_pg_conn, "idem_actor", schema=schema) + assert result.actor_config_deleted is True + + with pytest.raises(ActorNotFoundError, match="no stored actor_config row"): + await deregister_actor(clean_pg_conn, "idem_actor", schema=schema) + + +# ── combined force + purge_queue ──────────────────────────────────────── + + +async def test_deregister_force_with_purge_queue( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """force=True + purge_queue=True simultaneously — the exact pattern downstream + consumers (aacrtool) use for ephemeral actor cleanup.""" + schema = module_pg_schema.schema_name + await _insert_queue(clean_pg_conn, schema, "ephemeral_queue") + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="ephemeral_actor", max_concurrent=5, queue="ephemeral_queue")], + schema=schema, + ) + await _insert_job(clean_pg_conn, schema, actor="ephemeral_actor", status="pending") + await _insert_job(clean_pg_conn, schema, actor="ephemeral_actor", status="scheduled") + + result = await deregister_actor( + clean_pg_conn, "ephemeral_actor", force=True, purge_queue=True, schema=schema + ) + + assert result.actor_config_deleted is True + assert result.jobs_cancelled == 2 + assert result.queue_purged is True + assert await get_actor_config(clean_pg_conn, "ephemeral_actor", schema=schema) is None + assert await _queue_exists(clean_pg_conn, schema, "ephemeral_queue") is False + + +async def test_deregister_purge_queue_noop_when_queue_row_absent( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """purge_queue=True is a safe no-op when the queues row was never created. + + The queues table is metadata-only and not always populated — operator-managed + deployments may never create a row. The DELETE returns 0 rows and + queue_purged is False, which is correct. + """ + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="noqueue_actor", max_concurrent=5, queue="never_created")], + schema=schema, + ) + # Deliberately do NOT create a queues row for "never_created". + + result = await deregister_actor(clean_pg_conn, "noqueue_actor", purge_queue=True, schema=schema) + + assert result.actor_config_deleted is True + assert result.queue_purged is False + + +# ── concurrent deregistration ──────────────────────────────────────────── + + +async def test_concurrent_force_deregister_one_succeeds_one_raises( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Two concurrent force=True deregister calls: one wins, one raises. + + Both connections target the same actor that has a pending job and an + enabled schedule. Under READ COMMITTED, row-level locking serializes + the UPDATEs: the first transaction locks the job rows, the second + blocks, then sees 0 rows after the first commits. The DELETE returns + a row for only one transaction; the other gets 0 rows and raises + ActorNotFoundError. + """ + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="concurrent_force_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + job_id = await _insert_job( + clean_pg_conn, schema, actor="concurrent_force_actor", status="pending" + ) + sched_id = await _insert_schedule( + clean_pg_conn, schema, actor="concurrent_force_actor", enabled=True + ) + + conn2 = await asyncpg.connect(module_pg_schema.pg_dsn) + try: + results: list[BaseException | DeregisterResult] = [] + + async def _deregister(conn: asyncpg.Connection) -> None: + try: + result = await deregister_actor( + conn, "concurrent_force_actor", force=True, schema=schema + ) + results.append(result) + except ActorNotFoundError as exc: + results.append(exc) + + await asyncio.gather( + _deregister(clean_pg_conn), + _deregister(conn2), + ) + + successes = [r for r in results if isinstance(r, DeregisterResult)] + not_found = [r for r in results if isinstance(r, ActorNotFoundError)] + assert len(successes) == 1 + assert len(not_found) == 1 + + # The winner should have cancelled exactly 1 job and disabled 1 schedule + # — not double-cancelled by both transactions. + assert successes[0].jobs_cancelled == 1 + assert successes[0].schedules_disabled == 1 + + # Verify final DB state — job cancelled, schedule disabled, actor_config gone. + assert ( + await get_actor_config(clean_pg_conn, "concurrent_force_actor", schema=schema) is None + ) + assert await _job_status(clean_pg_conn, schema, job_id) == "cancelled" + assert await _schedule_state(clean_pg_conn, schema, sched_id) == "disabled" + finally: + await conn2.close() + + +# ── refusal gap coverage (L16) ─────────────────────────────────────────── + + +async def test_deregister_refuses_with_scheduled_jobs( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """force=False refuses with scheduled (not just pending) jobs.""" + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="sched_job_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + await _insert_job(clean_pg_conn, schema, actor="sched_job_actor", status="scheduled") + with pytest.raises(ActorHasActiveJobsError) as exc_info: + await deregister_actor(clean_pg_conn, "sched_job_actor", schema=schema) + assert exc_info.value.status_counts == {"scheduled": 1} + + +async def test_deregister_active_jobs_takes_precedence_over_schedules( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """When both active jobs AND enabled schedules exist, the jobs error is raised first.""" + schema = module_pg_schema.schema_name + await sync_actor_config( + clean_pg_conn, + [ActorConfig(actor="precedence_actor", max_concurrent=5, queue="default")], + schema=schema, + ) + await _insert_job(clean_pg_conn, schema, actor="precedence_actor", status="pending") + await _insert_schedule(clean_pg_conn, schema, actor="precedence_actor", enabled=True) + with pytest.raises(ActorHasActiveJobsError): + await deregister_actor(clean_pg_conn, "precedence_actor", schema=schema) + + +async def test_deregister_invalid_schema_raises_value_error( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Invalid schema identifier raises ValueError before any DB access.""" + with pytest.raises(ValueError, match="invalid schema identifier"): + await deregister_actor(clean_pg_conn, "any_actor", schema="bad; DROP TABLE") diff --git a/tests/test_actor_deregistration_client.py b/tests/test_actor_deregistration_client.py new file mode 100644 index 00000000..319854f9 --- /dev/null +++ b/tests/test_actor_deregistration_client.py @@ -0,0 +1,211 @@ +"""Full-stack integration tests for actor deregistration via TaskQ.actors. + +Exercises the complete client path: TaskQ → ActorsClient → pool → +deregister_actor → real Postgres. No Docker worker container needed — +the tests seed actor_config rows directly and call through the client. +""" + +from uuid import uuid4 + +import asyncpg +import pytest + +from taskq.actor_config import ActorConfig +from taskq.exceptions import ( + ActorHasActiveJobsError, + ActorNotFoundError, +) +from taskq.testing.fixtures import ModulePgSchema +from taskq.worker.startup import sync_actor_config + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +async def _seed_actor( + conn: asyncpg.Connection, schema: str, actor: str, queue: str = "default" +) -> None: + await sync_actor_config( + conn, + [ActorConfig(actor=actor, max_concurrent=5, queue=queue)], + schema=schema, + ) + + +async def _insert_job(conn: asyncpg.Connection, schema: str, actor: str, status: str) -> None: + await conn.execute( + f'INSERT INTO "{schema}".jobs (id, actor, queue, payload, status, max_attempts, retry_kind) ' # noqa: S608 # Why: schema validated by _IDENT_RE in apply_pending; actor/status are test constants. + f"VALUES ($1, $2, 'default', '{{}}'::jsonb, $3::\"{schema}\".job_status, 3, 'transient')", + uuid4(), + actor, + status, + ) + + +async def _insert_queue(conn: asyncpg.Connection, schema: str, name: str) -> None: + await conn.execute(f'INSERT INTO "{schema}".queues (name) VALUES ($1)', name) # noqa: S608 # Why: schema validated by _IDENT_RE in apply_pending; name is a test constant. + + +async def test_client_deregister_clean_actor( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Full client path: deregister an actor with no jobs or schedules.""" + from taskq import TaskQ + + schema = module_pg_schema.schema_name + await _seed_actor(clean_pg_conn, schema, "client_clean_actor") + + async with TaskQ(dsn=module_pg_schema.pg_dsn, schema=schema) as tq: + result = await tq.actors.deregister("client_clean_actor") + + assert result.actor_config_deleted is True + assert result.actor == "client_clean_actor" + + +async def test_client_deregister_not_found( + module_pg_schema: ModulePgSchema, +) -> None: + """Full client path: ActorNotFoundError for unknown actor.""" + from taskq import TaskQ + + async with TaskQ(dsn=module_pg_schema.pg_dsn, schema=module_pg_schema.schema_name) as tq: + with pytest.raises(ActorNotFoundError, match="no stored actor_config row"): + await tq.actors.deregister("nonexistent_actor") + + +async def test_client_deregister_refuses_with_active_jobs( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Full client path: ActorHasActiveJobsError when pending jobs exist.""" + from taskq import TaskQ + + schema = module_pg_schema.schema_name + await _seed_actor(clean_pg_conn, schema, "client_busy_actor") + await _insert_job(clean_pg_conn, schema, "client_busy_actor", "pending") + + async with TaskQ(dsn=module_pg_schema.pg_dsn, schema=schema) as tq: + with pytest.raises(ActorHasActiveJobsError) as exc_info: + await tq.actors.deregister("client_busy_actor") + + assert exc_info.value.active_count == 1 + assert exc_info.value.status_counts == {"pending": 1} + + +async def test_client_deregister_force_cancels_jobs( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Full client path: force=True cancels pending jobs through the client.""" + from taskq import TaskQ + + schema = module_pg_schema.schema_name + await _seed_actor(clean_pg_conn, schema, "client_force_actor") + await _insert_job(clean_pg_conn, schema, "client_force_actor", "pending") + await _insert_job(clean_pg_conn, schema, "client_force_actor", "scheduled") + + async with TaskQ(dsn=module_pg_schema.pg_dsn, schema=schema) as tq: + result = await tq.actors.deregister("client_force_actor", force=True) + + assert result.actor_config_deleted is True + assert result.jobs_cancelled == 2 + + # Verify jobs are cancelled in DB + cancelled_count = await clean_pg_conn.fetchval( + f"SELECT count(*) FROM \"{schema}\".jobs WHERE actor = 'client_force_actor' AND status = 'cancelled'" # noqa: S608 # Why: schema validated by _IDENT_RE in apply_pending; actor/status are test constants. + ) + assert cancelled_count == 2 + + +async def test_client_deregister_with_purge_queue( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Full client path: purge_queue=True deletes orphaned queue.""" + from taskq import TaskQ + + schema = module_pg_schema.schema_name + await _insert_queue(clean_pg_conn, schema, "client_solo_queue") + await _seed_actor(clean_pg_conn, schema, "client_purge_actor", queue="client_solo_queue") + + async with TaskQ(dsn=module_pg_schema.pg_dsn, schema=schema) as tq: + result = await tq.actors.deregister("client_purge_actor", purge_queue=True) + + assert result.queue_purged is True + assert result.queue == "client_solo_queue" + + queue_count = await clean_pg_conn.fetchval( + f"SELECT count(*) FROM \"{schema}\".queues WHERE name = 'client_solo_queue'" # noqa: S608 # Why: schema validated by _IDENT_RE in apply_pending; name is a test constant. + ) + assert queue_count == 0 + + +async def test_client_double_deregister_raises_not_found( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Full client path: second deregister raises ActorNotFoundError (idempotency).""" + from taskq import TaskQ + + schema = module_pg_schema.schema_name + await _seed_actor(clean_pg_conn, schema, "client_idem_actor") + + async with TaskQ(dsn=module_pg_schema.pg_dsn, schema=schema) as tq: + result = await tq.actors.deregister("client_idem_actor") + assert result.actor_config_deleted is True + + with pytest.raises(ActorNotFoundError): + await tq.actors.deregister("client_idem_actor") + + +async def test_client_actors_list_returns_rows( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Full client path: tq.actors.list() returns seeded actor_config rows.""" + from taskq import TaskQ + + schema = module_pg_schema.schema_name + await _seed_actor(clean_pg_conn, schema, "client_list_actor") + + async with TaskQ(dsn=module_pg_schema.pg_dsn, schema=schema) as tq: + rows = await tq.actors.list() + + actors = [r.actor for r in rows] + assert "client_list_actor" in actors + + +async def test_client_actors_get_returns_row( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Full client path: tq.actors.get() returns the specific actor row.""" + from taskq import TaskQ + + schema = module_pg_schema.schema_name + await _seed_actor(clean_pg_conn, schema, "client_get_actor") + + async with TaskQ(dsn=module_pg_schema.pg_dsn, schema=schema) as tq: + row = await tq.actors.get("client_get_actor") + assert row is not None + assert row.actor == "client_get_actor" + + missing = await tq.actors.get("nonexistent") + assert missing is None + + +async def test_client_actors_set_capacity( + clean_pg_conn: asyncpg.Connection, + module_pg_schema: ModulePgSchema, +) -> None: + """Full client path: tq.actors.set_capacity() updates max_concurrent.""" + from taskq import TaskQ + + schema = module_pg_schema.schema_name + await _seed_actor(clean_pg_conn, schema, "client_setcap_actor") + + async with TaskQ(dsn=module_pg_schema.pg_dsn, schema=schema) as tq: + row = await tq.actors.set_capacity("client_setcap_actor", max_concurrent=10) + + assert row is not None + assert row.max_concurrent == 10 diff --git a/tests/test_actors_client.py b/tests/test_actors_client.py new file mode 100644 index 00000000..c5e35d87 --- /dev/null +++ b/tests/test_actors_client.py @@ -0,0 +1,211 @@ +"""Tests for ActorsClient — the pool-wrapping facade over actor_config_ops. + +These tests use a fake pool to verify the delegation wiring without +requiring real Postgres (the ops functions themselves are integration-tested +in test_actor_deregistration.py and test_actor_config_ops.py). +""" + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from taskq.actor_config_ops import ( + ActorConfigRow, + DeregisterResult, +) +from taskq.exceptions import ActorNotFoundError + +pytestmark = [pytest.mark.asyncio] + + +class _FakeConn: + """Fake connection — just needs to be passable to the ops functions.""" + + +class _FakePool: + """Minimal pool that yields a fake connection via async context manager.""" + + def __init__(self, conn: Any) -> None: + self._conn = conn + + def acquire(self) -> Any: + cm = MagicMock() + cm.__aenter__ = AsyncMock(return_value=self._conn) + cm.__aexit__ = AsyncMock(return_value=None) + return cm + + +async def test_actors_client_list_delegates(monkeypatch: pytest.MonkeyPatch) -> None: + from taskq.client._actors import ActorsClient + + conn = _FakeConn() + pool = _FakePool(conn) + client = ActorsClient(pool, schema="test_schema") + + mock_result = [ + ActorConfigRow( + actor="a", + max_concurrent=1, + max_pending=None, + queue="q", + result_ttl=None, + metadata={}, + updated_at="2026-01-01", + ) + ] + import taskq.client._actors as actors_mod + + mock_list = AsyncMock(return_value=mock_result) + monkeypatch.setattr(actors_mod, "list_actor_configs", mock_list) + result = await client.list() + assert result == mock_result + mock_list.assert_called_once_with(conn, schema="test_schema") + + +async def test_actors_client_deregister_delegates(monkeypatch: pytest.MonkeyPatch) -> None: + from taskq.client._actors import ActorsClient + + conn = _FakeConn() + pool = _FakePool(conn) + client = ActorsClient(pool, schema="test_schema") + + expected = DeregisterResult( + actor="test-actor", + queue="q", + actor_config_deleted=True, + schedules_disabled=0, + jobs_cancelled=0, + terminal_jobs_remaining=0, + queue_purged=False, + ) + + import taskq.client._actors as actors_mod + + mock_deregister = AsyncMock(return_value=expected) + monkeypatch.setattr(actors_mod, "deregister_actor", mock_deregister) + result = await client.deregister("test-actor", force=True, purge_queue=True) + assert result == expected + mock_deregister.assert_called_once_with( + conn, "test-actor", force=True, purge_queue=True, schema="test_schema" + ) + + +async def test_actors_client_get_delegates(monkeypatch: pytest.MonkeyPatch) -> None: + from taskq.client._actors import ActorsClient + + conn = _FakeConn() + pool = _FakePool(conn) + client = ActorsClient(pool, schema="test_schema") + + expected = ActorConfigRow( + actor="a", + max_concurrent=1, + max_pending=None, + queue="q", + result_ttl=None, + metadata={}, + updated_at="2026-01-01", + ) + + import taskq.client._actors as actors_mod + + mock_get = AsyncMock(return_value=expected) + monkeypatch.setattr(actors_mod, "get_actor_config", mock_get) + result = await client.get("a") + assert result == expected + mock_get.assert_called_once_with(conn, "a", schema="test_schema") + + +async def test_actors_client_set_capacity_delegates(monkeypatch: pytest.MonkeyPatch) -> None: + from taskq.client._actors import ActorsClient + + conn = _FakeConn() + pool = _FakePool(conn) + client = ActorsClient(pool, schema="test_schema") + + expected = ActorConfigRow( + actor="a", + max_concurrent=10, + max_pending=None, + queue="q", + result_ttl=None, + metadata={}, + updated_at="2026-01-01", + ) + + import taskq.client._actors as actors_mod + + mock_set_capacity = AsyncMock(return_value=expected) + monkeypatch.setattr(actors_mod, "set_actor_config_capacity", mock_set_capacity) + result = await client.set_capacity("a", max_concurrent=10) + assert result == expected + mock_set_capacity.assert_called_once_with( + conn, + "a", + max_concurrent=10, + max_pending=actors_mod.UNSET, + result_ttl=actors_mod.UNSET, + schema="test_schema", + ) + + +async def test_actors_client_deregister_propagates_errors(monkeypatch: pytest.MonkeyPatch) -> None: + """Exceptions from the ops function must propagate through the pool-wrapper — + not be silently swallowed. This is a lifecycle concern, not a delegation one.""" + from taskq.client._actors import ActorsClient + + conn = _FakeConn() + pool = _FakePool(conn) + client = ActorsClient(pool, schema="test_schema") + + import taskq.client._actors as actors_mod + + monkeypatch.setattr( + actors_mod, + "deregister_actor", + AsyncMock(side_effect=ActorNotFoundError("bad-actor")), + ) + with pytest.raises(ActorNotFoundError): + await client.deregister("bad-actor") + + +async def test_actors_client_set_capacity_forwards_none_vs_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """None means 'clear'; UNSET (default) means 'leave unchanged'. + + The client must forward both faithfully to set_actor_config_capacity + so operators can distinguish clearing a field from not touching it. + """ + from taskq.client._actors import UNSET, ActorsClient + + conn = _FakeConn() + pool = _FakePool(conn) + client = ActorsClient(pool, schema="test_schema") + + expected = ActorConfigRow( + actor="a", + max_concurrent=None, + max_pending=5, + queue="q", + result_ttl=None, + metadata={}, + updated_at="2026-01-01", + ) + + import taskq.client._actors as actors_mod + + mock_fn = AsyncMock(return_value=expected) + monkeypatch.setattr(actors_mod, "set_actor_config_capacity", mock_fn) + + await client.set_capacity("a", max_concurrent=None, max_pending=5) + + mock_fn.assert_called_once_with( + conn, + "a", + max_concurrent=None, + max_pending=5, + result_ttl=UNSET, + schema="test_schema", + ) diff --git a/tests/test_cli_actor_config.py b/tests/test_cli_actor_config.py index 5d2b60c4..9f476fa5 100644 --- a/tests/test_cli_actor_config.py +++ b/tests/test_cli_actor_config.py @@ -15,8 +15,8 @@ from typer.testing import CliRunner from taskq.actor import ActorRef, actor +from taskq.actor_config_ops import ActorConfigRow from taskq.cli import app -from taskq.worker.actor_config_ops import ActorConfigRow runner = CliRunner() diff --git a/tests/test_cli_actor_deregister.py b/tests/test_cli_actor_deregister.py new file mode 100644 index 00000000..a64c1c62 --- /dev/null +++ b/tests/test_cli_actor_deregister.py @@ -0,0 +1,156 @@ +"""Tests for `taskq actor-config deregister` CLI command. + +Monkeypatches the ops function and asyncpg.connect to pin the CLI's +argument parsing, error handling, and output shape without requiring +real Postgres (integration coverage is in test_actor_deregistration.py). +""" + +from typing import Any + +import pytest +from typer.testing import CliRunner + +from taskq.actor_config_ops import DeregisterResult +from taskq.cli import app + +runner = CliRunner() + + +def _patch_deregister( + monkeypatch: pytest.MonkeyPatch, + *, + result: DeregisterResult | None = None, + raises: Exception | None = None, +) -> dict[str, Any]: + """Fake asyncpg.connect + deregister_actor; return captured call kwargs.""" + captured: dict[str, Any] = {} + + class _FakeConn: + async def close(self) -> None: ... + + async def fake_connect(dsn: str) -> Any: + return _FakeConn() + + async def fake_deregister(conn: Any, actor: str, **kwargs: Any) -> Any: + captured["actor"] = actor + captured["kwargs"] = kwargs + if raises is not None: + raise raises + return result or DeregisterResult( + actor=actor, + queue="default", + actor_config_deleted=True, + schedules_disabled=0, + jobs_cancelled=0, + terminal_jobs_remaining=0, + queue_purged=False, + ) + + monkeypatch.setattr("taskq.cli.asyncpg.connect", fake_connect) + monkeypatch.setattr("taskq.cli.deregister_actor", fake_deregister) + return captured + + +def test_deregister_default_no_force_no_purge(monkeypatch: pytest.MonkeyPatch) -> None: + captured = _patch_deregister(monkeypatch) + result = runner.invoke(app, ["actor-config", "deregister", "my-actor.run-123"]) + assert result.exit_code == 0, f"stderr: {result.stderr}" + assert captured["actor"] == "my-actor.run-123" + assert captured["kwargs"]["force"] is False + assert captured["kwargs"]["purge_queue"] is False + + +def test_deregister_force_flag(monkeypatch: pytest.MonkeyPatch) -> None: + captured = _patch_deregister(monkeypatch) + result = runner.invoke(app, ["actor-config", "deregister", "my-actor", "--force"]) + assert result.exit_code == 0, f"stderr: {result.stderr}" + assert captured["kwargs"]["force"] is True + + +def test_deregister_purge_queue_flag(monkeypatch: pytest.MonkeyPatch) -> None: + captured = _patch_deregister(monkeypatch) + result = runner.invoke(app, ["actor-config", "deregister", "my-actor", "--purge-queue"]) + assert result.exit_code == 0, f"stderr: {result.stderr}" + assert captured["kwargs"]["purge_queue"] is True + + +def test_deregister_not_found_exit_three(monkeypatch: pytest.MonkeyPatch) -> None: + from taskq.exceptions import ActorNotFoundError + + _patch_deregister(monkeypatch, raises=ActorNotFoundError("ghost")) + result = runner.invoke(app, ["actor-config", "deregister", "ghost"]) + assert result.exit_code == 3 + assert "no stored actor_config row" in result.stderr + assert "Cannot deregister actor" in result.stderr + + +def test_deregister_active_jobs_error_exit_two(monkeypatch: pytest.MonkeyPatch) -> None: + from taskq.exceptions import ActorHasActiveJobsError + + _patch_deregister( + monkeypatch, + raises=ActorHasActiveJobsError( + "busy", active_count=3, status_counts={"pending": 2, "running": 1} + ), + ) + result = runner.invoke(app, ["actor-config", "deregister", "busy"]) + assert result.exit_code == 2 + assert "non-terminal" in result.stderr + assert "force=True" in result.stderr + assert "Cannot deregister actor" in result.stderr + + +def test_deregister_schedules_error_exit_two(monkeypatch: pytest.MonkeyPatch) -> None: + from taskq.exceptions import ActorHasEnabledSchedulesError + + _patch_deregister( + monkeypatch, + raises=ActorHasEnabledSchedulesError("sched-actor", ["s1", "s2"]), + ) + result = runner.invoke(app, ["actor-config", "deregister", "sched-actor"]) + assert result.exit_code == 2 + assert "enabled cron schedule" in result.stderr + + +def test_deregister_output_shows_result(monkeypatch: pytest.MonkeyPatch) -> None: + result = DeregisterResult( + actor="my-actor", + queue="my-queue", + actor_config_deleted=True, + schedules_disabled=2, + jobs_cancelled=5, + terminal_jobs_remaining=10, + queue_purged=True, + ) + _patch_deregister(monkeypatch, result=result) + output = runner.invoke( + app, ["actor-config", "deregister", "my-actor", "--force", "--purge-queue"] + ) + assert output.exit_code == 0 + assert "deregistered" in output.output.lower() + assert "actor_config_deleted=true" in output.output.lower() + assert "queue='my-queue'" in output.output.lower() + assert "schedules_disabled=2" in output.output + assert "jobs_cancelled=5" in output.output + assert "terminal_jobs_remaining=10" in output.output + assert "queue_purged=true" in output.output.lower() + assert "WARNING" in output.output + assert "stranded pending job" in output.output + + +def test_deregister_double_deregister_exit_three(monkeypatch: pytest.MonkeyPatch) -> None: + """Second deregister on an already-deregistered actor exits 3 with ActorNotFoundError.""" + from taskq.exceptions import ActorNotFoundError + + _patch_deregister(monkeypatch, raises=ActorNotFoundError("already-gone")) + result = runner.invoke(app, ["actor-config", "deregister", "already-gone"]) + assert result.exit_code == 3 + assert "no stored actor_config row" in result.stderr + + +def test_deregister_forwards_configured_schema(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TASKQ_SCHEMA_NAME", "my_test_schema") + captured = _patch_deregister(monkeypatch) + result = runner.invoke(app, ["actor-config", "deregister", "my-actor"]) + assert result.exit_code == 0, f"stderr: {result.stderr}" + assert captured["kwargs"]["schema"] == "my_test_schema" diff --git a/tests/test_dispatch_pg.py b/tests/test_dispatch_pg.py index 0e8c9dc0..065f34d6 100644 --- a/tests/test_dispatch_pg.py +++ b/tests/test_dispatch_pg.py @@ -17,11 +17,11 @@ import pytest from taskq._ids import new_base62, new_uuid +from taskq.actor_config import ActorConfig from taskq.backend.postgres import PostgresBackend from taskq.testing.fixtures import JobsApp, _open_pg_backend from taskq.testing.jobs import make_enqueue_args from taskq.testing.pg import create_worker -from taskq.worker.actor_config import ActorConfig from taskq.worker.run import register_worker from taskq.worker.startup import sync_actor_config diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py index ba1af63f..9c8bcb62 100644 --- a/tests/test_exceptions.py +++ b/tests/test_exceptions.py @@ -476,3 +476,91 @@ def test_missing_provider_isinstance_taskqerror() -> None: """MissingProvider is a TaskQError subclass.""" exc = MissingProvider(type_name="X", required_by="Y") assert isinstance(exc, TaskQError) + + +class TestActorDeregistrationErrors: + """Tests for the deregistration refusal exception hierarchy.""" + + def test_actor_has_active_jobs_error_carries_counts(self) -> None: + from taskq.exceptions import ActorHasActiveJobsError + + err = ActorHasActiveJobsError( + actor="my-actor.run-123", + active_count=3, + status_counts={"pending": 2, "running": 1}, + ) + assert err.actor == "my-actor.run-123" + assert err.active_count == 3 + assert err.status_counts == {"pending": 2, "running": 1} + assert "3 non-terminal" in str(err) + assert "force=True" in str(err) + + def test_actor_has_active_jobs_error_force_true_message(self) -> None: + from taskq.exceptions import ActorHasActiveJobsError + + err = ActorHasActiveJobsError( + actor="my-actor.run-123", + active_count=2, + status_counts={"running": 2}, + force=True, + ) + assert err.actor == "my-actor.run-123" + assert err.active_count == 2 + assert err.status_counts == {"running": 2} + msg = str(err) + assert "2 running job(s)" in msg + assert "cannot be cancelled by force=True" in msg + assert "wait for them to finish" in msg + # Must NOT contain the force=False remediation advice + assert "pass" not in msg.lower().split("force=true")[0] + + def test_actor_has_enabled_schedules_error_carries_ids(self) -> None: + from taskq.exceptions import ActorHasEnabledSchedulesError + + err = ActorHasEnabledSchedulesError( + actor="my-actor.run-123", + schedule_ids=["sched-1", "sched-2"], + ) + assert err.actor == "my-actor.run-123" + assert err.schedule_ids == ["sched-1", "sched-2"] + assert "2 enabled cron schedule" in str(err) + assert "force=True" in str(err) + + def test_actor_not_found_error(self) -> None: + from taskq.exceptions import ActorNotFoundError + + err = ActorNotFoundError("ghost-actor") + assert err.actor == "ghost-actor" + assert "no stored actor_config row" in str(err) + + def test_deregistration_errors_inherit_taskq_error(self) -> None: + from taskq.exceptions import ( + ActorDeregistrationError, + ActorHasActiveJobsError, + ActorHasEnabledSchedulesError, + ActorNotFoundError, + TaskQError, + ) + + for cls in ( + ActorDeregistrationError, + ActorHasActiveJobsError, + ActorHasEnabledSchedulesError, + ActorNotFoundError, + ): + assert issubclass(cls, TaskQError) + + def test_specific_errors_inherit_deregistration_error(self) -> None: + from taskq.exceptions import ( + ActorDeregistrationError, + ActorHasActiveJobsError, + ActorHasEnabledSchedulesError, + ActorNotFoundError, + ) + + for cls in ( + ActorHasActiveJobsError, + ActorHasEnabledSchedulesError, + ActorNotFoundError, + ): + assert issubclass(cls, ActorDeregistrationError) diff --git a/tests/test_in_memory_terminal_writes.py b/tests/test_in_memory_terminal_writes.py index 7fe2ae95..ca3abb20 100644 --- a/tests/test_in_memory_terminal_writes.py +++ b/tests/test_in_memory_terminal_writes.py @@ -14,11 +14,11 @@ import pytest from taskq._ids import new_job_id, new_uuid +from taskq.actor_config import ActorConfig from taskq.backend._protocol import EnqueueArgs, ErrorInfo, JobId, RetryKind from taskq.exceptions import WorkerOwnershipMismatch from taskq.testing.clock import FakeClock from taskq.testing.in_memory import InMemoryBackend -from taskq.worker.actor_config import ActorConfig # ── Helpers ──────────────────────────────────────────────────────────── diff --git a/tests/test_postgres_terminal_writes.py b/tests/test_postgres_terminal_writes.py index d3d69c81..0b54f59d 100644 --- a/tests/test_postgres_terminal_writes.py +++ b/tests/test_postgres_terminal_writes.py @@ -990,7 +990,7 @@ async def _seed_cleared_ttl_and_aged_job( result_expires_at is pinned in the past — the state the enqueue path leaves behind (enqueue_now + literal) after the job sat in the queue longer than its TTL.""" - from taskq.worker.actor_config_ops import set_actor_config_capacity + from taskq.actor_config_ops import set_actor_config_capacity deps = jobs_app.deps schema = deps.settings.schema_name @@ -1067,7 +1067,7 @@ async def test_stored_result_ttl_wins_over_fallback_at_completion( ) -> None: """Operator override (stored 300s) beats the worker's fallback literal (5s) for jobs completing after the override.""" - from taskq.worker.actor_config_ops import set_actor_config_capacity + from taskq.actor_config_ops import set_actor_config_capacity deps = clean_jobs_app.deps backend = clean_jobs_app.backend diff --git a/tests/test_taskq_actors_property.py b/tests/test_taskq_actors_property.py new file mode 100644 index 00000000..ca0420e5 --- /dev/null +++ b/tests/test_taskq_actors_property.py @@ -0,0 +1,69 @@ +"""Tests for TaskQ.actors property and public API exports.""" + +from __future__ import annotations + +import pytest + +from taskq.testing.fixtures import ModulePgSchema + + +def test_actors_client_importable_from_taskq() -> None: + from taskq import ActorsClient + from taskq.client._actors import ActorsClient as ClientActorsClient + + assert ActorsClient is ClientActorsClient + + +def test_deregister_result_importable_from_taskq() -> None: + from taskq import DeregisterResult + from taskq.actor_config_ops import DeregisterResult as OpsDeregisterResult + + assert DeregisterResult is OpsDeregisterResult + + +def test_actor_config_row_importable_from_taskq() -> None: + from taskq import ActorConfigRow + from taskq.actor_config_ops import ActorConfigRow as OpsActorConfigRow + + assert ActorConfigRow is OpsActorConfigRow + + +def test_deregistration_exceptions_importable_from_taskq() -> None: + from taskq import ( + ActorDeregistrationError, + ActorHasActiveJobsError, + ActorHasEnabledSchedulesError, + ActorNotFoundError, + ) + + assert ActorDeregistrationError is not None + assert ActorHasActiveJobsError is not None + assert ActorHasEnabledSchedulesError is not None + assert ActorNotFoundError is not None + + +def test_taskq_actors_raises_before_open() -> None: + """Accessing tq.actors before open() raises RuntimeError.""" + from taskq import TaskQ + + tq = TaskQ(dsn="postgresql://fake:fake@localhost/fake") + with pytest.raises(RuntimeError, match="not open"): + _ = tq.actors + + +@pytest.mark.asyncio +@pytest.mark.integration +async def test_taskq_actors_property_returns_actors_client( + module_pg_schema: ModulePgSchema, +) -> None: + """TaskQ.actors returns an ActorsClient bound to the same pool and schema.""" + from taskq import TaskQ + from taskq.client._actors import ActorsClient + + async with TaskQ( + dsn=module_pg_schema.pg_dsn, + schema=module_pg_schema.schema_name, + ) as tq: + client = tq.actors + assert isinstance(client, ActorsClient) + assert client._schema == module_pg_schema.schema_name diff --git a/tests/test_web_admin_actors.py b/tests/test_web_admin_actors.py new file mode 100644 index 00000000..6214d2c0 --- /dev/null +++ b/tests/test_web_admin_actors.py @@ -0,0 +1,347 @@ +"""Tests for the admin UI actors page and deregister route. + +Follows the pattern of tests/test_web_admin_integration.py: a per-test +asyncpg pool on the module's migrated schema, a FastAPI app built via +create_router + setup_admin_state + include_router, and httpx.AsyncClient +with ASGITransport. + +CSRF uses the synchronizer-token pattern: GET sets the taskq_csrf_token +cookie; POST must include it as the csrf_token form field. +""" + +import asyncpg +import httpx +import pytest +from fastapi import FastAPI + +from taskq.actor_config import ActorConfig +from taskq.testing.fixtures import ModulePgSchema +from taskq.web.admin import create_router, setup_admin_state +from taskq.worker.startup import sync_actor_config + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +def _make_admin_app( + pool: asyncpg.Pool, + schema: str, + monkeypatch: pytest.MonkeyPatch, + *, + admin_actions_enabled: bool, +) -> FastAPI: + # setenv must precede create_router — it calls TaskQSettings.load() internally + monkeypatch.setenv("TASKQ_ENVIRONMENT", "dev") + monkeypatch.setenv("TASKQ_ADMIN_ACTIONS_ENABLED", "true" if admin_actions_enabled else "false") + bundle = create_router(pool, schema=schema, base_path="/admin") + app = FastAPI() + setup_admin_state(app, bundle) + app.include_router(bundle.router, prefix="/admin") + return app + + +async def _seed_actor_config( + conn: asyncpg.Connection, + schema: str, + actor: str, + queue: str = "default", +) -> None: + await sync_actor_config( + conn, + [ActorConfig(actor=actor, max_concurrent=1, queue=queue)], + schema=schema, + ) + + +async def _get_csrf_then_post( + app: FastAPI, + get_url: str, + post_url: str, + data: dict[str, str] | None = None, +) -> httpx.Response: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test", + follow_redirects=False, + ) as client: + get_resp = await client.get(get_url) + assert get_resp.status_code == 200 + csrf_token = get_resp.cookies.get("taskq_csrf_token", "") + assert csrf_token, "GET must set the taskq_csrf_token cookie" + return await client.post(post_url, data={"csrf_token": csrf_token, **(data or {})}) + + +async def test_actors_page_lists_actor_config_rows( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + schema = module_pg_schema.schema_name + await _seed_actor_config(clean_pg_conn, schema, "test-actor-1", queue="default") + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + resp = await client.get("/admin/actors") + + assert resp.status_code == 200 + assert "test-actor-1" in resp.text + assert "default" in resp.text + + +async def test_actors_page_shows_deregister_form( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + schema = module_pg_schema.schema_name + await _seed_actor_config(clean_pg_conn, schema, "button-actor", queue="default") + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + resp = await client.get("/admin/actors") + + assert resp.status_code == 200 + # Targeted assertions — check for the specific form action and form fields, + # not just the word "Deregister" which could appear anywhere. + assert 'action="/admin/actors/button-actor/deregister"' in resp.text + assert 'name="force"' in resp.text + assert 'name="purge_queue"' in resp.text + + +async def test_deregister_route_returns_403_when_admin_actions_disabled( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + schema = module_pg_schema.schema_name + await _seed_actor_config(clean_pg_conn, schema, "disabled-actor", queue="default") + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=False) + + resp = await _get_csrf_then_post( + app, "/admin/actors", "/admin/actors/disabled-actor/deregister" + ) + + assert resp.status_code == 403 + count = await clean_pg_conn.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', + "disabled-actor", + ) + assert count == 1 + + +async def test_deregister_route_succeeds_for_clean_actor( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + schema = module_pg_schema.schema_name + await _seed_actor_config(clean_pg_conn, schema, "clean-deregister-actor", queue="default") + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + resp = await _get_csrf_then_post( + app, "/admin/actors", "/admin/actors/clean-deregister-actor/deregister" + ) + + assert resp.status_code == 303 + assert "/actors" in resp.headers["location"] + + count = await clean_pg_conn.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', + "clean-deregister-actor", + ) + assert count == 0 + + +async def test_deregister_route_returns_409_when_actor_has_active_jobs( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """POST deregister with an active job returns 409 Conflict.""" + from uuid import uuid4 + + schema = module_pg_schema.schema_name + await _seed_actor_config(clean_pg_conn, schema, "blocked-actor", queue="default") + # Insert a pending job so force=False deregistration refuses. + await clean_pg_conn.execute( + f'INSERT INTO "{schema}".jobs (id, actor, queue, payload, status, max_attempts, retry_kind) ' + f"VALUES ($1, 'blocked-actor', 'default', '{{}}'::jsonb, 'pending'::\"{schema}\".job_status, 3, 'transient')", + uuid4(), + ) + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + resp = await _get_csrf_then_post(app, "/admin/actors", "/admin/actors/blocked-actor/deregister") + + assert resp.status_code == 409 + assert "non-terminal" in resp.text + # Row must still exist — deregistration was refused. + count = await clean_pg_conn.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', + "blocked-actor", + ) + assert count == 1 + + +async def test_deregister_route_with_force_cancels_pending_job( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """POST deregister with force=true cancels pending jobs through the admin route.""" + from uuid import uuid4 + + schema = module_pg_schema.schema_name + await _seed_actor_config(clean_pg_conn, schema, "force-admin-actor", queue="default") + job_id = uuid4() + await clean_pg_conn.execute( + f'INSERT INTO "{schema}".jobs (id, actor, queue, payload, status, max_attempts, retry_kind) ' + f"VALUES ($1, 'force-admin-actor', 'default', '{{}}'::jsonb, 'pending'::\"{schema}\".job_status, 3, 'transient')", + job_id, + ) + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + resp = await _get_csrf_then_post( + app, + "/admin/actors", + "/admin/actors/force-admin-actor/deregister", + data={"force": "true"}, + ) + + assert resp.status_code == 303 + # Job should be cancelled + status = await clean_pg_conn.fetchval( + f'SELECT status::text FROM "{schema}".jobs WHERE id = $1', job_id + ) + assert status == "cancelled" + # Actor config should be gone + count = await clean_pg_conn.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', "force-admin-actor" + ) + assert count == 0 + + +async def test_deregister_route_returns_404_for_unknown_actor( + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """POST deregister for a non-existent actor returns 404, not 409.""" + schema = module_pg_schema.schema_name + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + resp = await _get_csrf_then_post( + app, "/admin/actors", "/admin/actors/nonexistent-actor/deregister" + ) + + assert resp.status_code == 404 + assert "no stored actor_config row" in resp.text + + +async def test_actors_page_shows_notice_after_deregister( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """GET /actors?notice=... renders the notice banner.""" + schema = module_pg_schema.schema_name + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + resp = await client.get("/admin/actors?notice=deregistered") + + assert resp.status_code == 200 + assert "Actor deregistered successfully" in resp.text + # The notice should be in a styled banner div, not just in the page somewhere + assert "bg-green" in resp.text + + +async def test_deregister_route_returns_409_for_enabled_schedules( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """POST deregister with an enabled schedule returns 409 Conflict.""" + from uuid import uuid4 + + schema = module_pg_schema.schema_name + await _seed_actor_config(clean_pg_conn, schema, "sched-409-actor", queue="default") + await clean_pg_conn.execute( + f'INSERT INTO "{schema}".cron_schedules (id, actor, cron_expr, enabled, next_fire_at) ' + f"VALUES ($1, 'sched-409-actor', '*/5 * * * *', true, now())", + uuid4(), + ) + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + resp = await _get_csrf_then_post( + app, "/admin/actors", "/admin/actors/sched-409-actor/deregister" + ) + + assert resp.status_code == 409 + assert "enabled cron schedule" in resp.text + count = await clean_pg_conn.fetchval( + f'SELECT count(*) FROM "{schema}".actor_config WHERE actor = $1', + "sched-409-actor", + ) + assert count == 1 + + +async def test_deregister_route_with_purge_queue_deletes_queue( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """POST deregister with purge_queue=true deletes the orphaned queue.""" + schema = module_pg_schema.schema_name + await clean_pg_conn.execute( + f'INSERT INTO "{schema}".queues (name) VALUES ($1)', + "admin-purge-queue", + ) + await _seed_actor_config(clean_pg_conn, schema, "admin-purge-actor", queue="admin-purge-queue") + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + resp = await _get_csrf_then_post( + app, + "/admin/actors", + "/admin/actors/admin-purge-actor/deregister", + data={"purge_queue": "true"}, + ) + + assert resp.status_code == 303 + queue_count = await clean_pg_conn.fetchval( + f'SELECT count(*) FROM "{schema}".queues WHERE name = $1', + "admin-purge-queue", + ) + assert queue_count == 0 + + +async def test_deregister_route_returns_403_without_csrf_token( + clean_pg_conn: asyncpg.Connection, + module_pg_pool: asyncpg.Pool, + module_pg_schema: ModulePgSchema, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """POST without CSRF token returns 403.""" + schema = module_pg_schema.schema_name + await _seed_actor_config(clean_pg_conn, schema, "csrf-test-actor", queue="default") + app = _make_admin_app(module_pg_pool, schema, monkeypatch, admin_actions_enabled=True) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + resp = await client.post("/admin/actors/csrf-test-actor/deregister") + + assert resp.status_code == 403