diff --git a/.github/workflows/workflow.yaml b/.github/workflows/workflow.yaml index b66a090..04cb52f 100644 --- a/.github/workflows/workflow.yaml +++ b/.github/workflows/workflow.yaml @@ -60,7 +60,7 @@ jobs: python-version: ${{ matrix.python-version }} - name: Install the redis and postgres extras, fastapi and django - run: python -m pip install "redis>=5.0.1" "psycopg[binary]>=3.2" "psycopg-pool>=3.2" fastapi httpx "django>=6.0; python_version >= '3.12'" + run: python -m pip install "redis>=5.0.1,<9" "psycopg[binary]>=3.2,<3.4" "psycopg-pool>=3.2,<3.4" fastapi httpx "django>=6.0; python_version >= '3.12'" - name: Run the test suite run: python -m unittest discover -s tests -t . --verbose @@ -102,15 +102,66 @@ jobs: run: sudo apt-get update && sudo apt-get install -y libpq5 - name: Install redis without hiredis, psycopg without its binary wheel, fastapi and django - run: python -m pip install "redis>=5.0.1" "psycopg>=3.2" "psycopg-pool>=3.2" fastapi httpx "django>=6.0" + run: python -m pip install "redis>=5.0.1,<9" "psycopg>=3.2,<3.4" "psycopg-pool>=3.2,<3.4" fastapi httpx "django>=6.0" - name: Run the test suite with the GIL off run: | python -c "import sys; assert not sys._is_gil_enabled()" python -m unittest discover -s tests -t . --verbose + upgrade: + runs-on: ubuntu-latest + strategy: + fail-fast: true + matrix: + postgres: ["14", "18"] + services: + redis: + image: redis:7.4-alpine + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 2s + --health-timeout 2s + --health-retries 15 + postgres: + image: postgres:${{ matrix.postgres }}-alpine + env: + POSTGRES_PASSWORD: tinyjobs + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U postgres" + --health-interval 2s + --health-timeout 2s + --health-retries 15 + env: + TINYJOBS_REDIS_URL: redis://localhost:6379/0 + TINYJOBS_POSTGRES_URL: postgresql://postgres:tinyjobs@localhost:5432/postgres + TINYJOBS_PREVIOUS_PYTHON: /tmp/previous/bin/python + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - uses: actions/setup-python@6d4fd22b20465ccc0f7354476588e8df7c9ca0e0 # v7.1.0 + with: + python-version: "3.12" + + - name: Install the redis and postgres extras + run: python -m pip install "redis>=5.0.1,<9" "psycopg[binary]>=3.2,<3.4" "psycopg-pool>=3.2,<3.4" + + - name: Install the release before this one in an environment of its own + run: | + version=$(python -c "import re; print(re.search(r'^__version__ = \"(.+)\"$', open('tinyjobs/__init__.py').read(), re.M)[1])") + python -m venv /tmp/previous + /tmp/previous/bin/pip install --quiet "tinyjobs[redis,postgres]<$version" + /tmp/previous/bin/python -I -c "import tinyjobs; print('previous release:', tinyjobs.__version__)" + + - name: Upgrade a store of the previous release and share it with this one + run: python -m unittest tests.test_upgrade --verbose + build: - needs: [test, test-free-threaded] + needs: [test, test-free-threaded, upgrade] runs-on: ubuntu-latest steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 diff --git a/.gitignore b/.gitignore index fca2ab2..d1a9a58 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ dist/ uv.lock .venv/ +.venv-previous/ venv/ *.db diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..d7ad38c --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,412 @@ +# Changelog + +What changed in each release of tinyjobs, newest first, in the layout of [Keep a +Changelog](https://keepachangelog.com/en/1.1.0/). What a release is allowed to change is on the +[compatibility page](https://tinyjobs.readthedocs.io/en/latest/compatibility/). Store schema +versions are the `SCHEMA_VERSION` of each backend. + +## 0.18.0 - unreleased + +### Added +- `tinyjobs.NewerStoreError`, raised when a store was upgraded by a release this one cannot work + beside. +- Importable from `tinyjobs`: the exceptions a caller can meet and could only import from + `tinyjobs.exceptions` (`AppClosedError`, `BackendNotReadyError`, `UnsupportedCapabilityError`, + `SynchronousCallInsideEventLoopError`, `InterruptedCallError`, `EventLoopStoppedError`, + `ExecutorClosedError`, `TaskAlreadyRegisteredError`, `UnknownCodecError`, `ForcedInterrupt`), and + the types the app returns or takes: `Stats`, `QueueStats`, `Page`, `Finding` and `Clock`. +- A compatibility page: what is public, what a release may change, and what has to be true before + 1.0. This changelog. +- For backend authors: a store without `claim_job` passes the contract, as the documentation said; + the tests that claim a job by its id are skipped for it or run without that step. The in-memory + template in `tests/test_async_backend.py` now passes the contract. + +### Changed +- A store too new for the running release is refused with `NewerStoreError` instead of a bare + `RuntimeError`. It is still a `RuntimeError`, so code catching that keeps working. A command + that meets one prints `error: ...` and exits with status 2, where it printed a traceback and + exited with status 1. +- `from tinyjobs import *` brings in the names listed under Added, among them `Page`, `Clock`, + `Stats` and `Finding`, which can shadow a name of your own imported before it. + +### Fixed +- `tinyjobs verify` no longer reports jobs as unfinished when a killed worker's jobs went back to + the queue between two of its reads: it confirms the store is settled with one listing of the + queue. + +### Upgrading +- No schema change and no stop needed: 0.17 and 0.18 processes work on the same store, which every + release now checks on SQLite, Postgres and Redis. From 0.16 or earlier, follow the 0.17 note + first. + +## 0.17.0 - 2026-11-12 + +### Added +- `after=job` (the `Job` or its id): a job waits in the new `waiting` status until another job + succeeds, and is cancelled, with every job waiting behind it, when that job fails for good or is + cancelled. Works with `enqueue`, `app.send` and `tinyjobs send --after`. +- `tinyjobs.JobNotFoundError`, raised by `enqueue` when the job named in `after` is not in the store. +- On Django, `task.after(result)`; every task result reports the task it ran after and why it was + cancelled. +- Every store records `compatible_since`, the oldest schema version whose code can run beside it, + so later upgrades between neighbouring minors need no stop. + +### Changed +- `JobStatus` has a new member, `waiting`. `tinyjobs stats` and the dashboard's queue table gain a + column between held and running, so anything reading those tables by position is one column off; + `/api/overview` counts it in every queue. +- Queued and held jobs are called "queued" everywhere: the dashboard's "Queued by task" (it was + "Waiting by task"), the `check` finding "N queued job(s) for X" (it said "waiting"), and the + help text of `tinyjobs_store_queued_jobs`. Alerts matching the old words need the new ones. +- `jobs retry` and `jobs cancel` word their refusals and help differently, and so do the + dashboard's notices after a retry. +- A job's page and every job in `/api/jobs` show the job it runs after; `/jobs?status=waiting` is a + page instead of a 400; `tinyjobs_store_jobs` can carry `status="waiting"`. +- On Django every task has a `follows` field, `None` unless `.after()` set it. +- On Postgres a completion reads one more row (about 6% on a loop that only claims and completes). + +### Fixed +- `ready()` right after a close no longer raises the `AppClosedError` of a connection attempt that + the close had beaten. +- A job process killed by a signal is reported with the signal's name even when its exit status + arrives late. + +### Upgrading +- Schema: SQLite 5 to 6, Postgres 2 to 3, Redis 3 to 4 (a new column, a field on Redis). +- Stop every tinyjobs process on a store before starting the first 0.17 one. It upgrades the store + on its first call, and older processes then refuse to start against it. Going back to 0.16 means + restoring a backup taken before the upgrade. + +## 0.16.0 - 2026-11-11 + +### Added +- The tinyjobs task options on `django.tasks`: `timeout`, `max_attempts` and the `retry_*` + options, `concurrency`, `rate`, `isolated` and `result_ttl`, on Django's own `@task` (or + `tinyjobs.contrib.django.task` on Django 6.0), or by import path under the `"tasks"` option in + the settings, where they win over the decorator. +- Isolated tasks on Django run inside the job process, with `django.setup()` done there first. + +### Changed +- On Django, `task_finished` fires from a worker thread after the queue has recorded the outcome, + with the result read back from the queue. A receiver that raises no longer fails the task; a + timeout now fires it; a job cancelled while it runs no longer does. +- `@app.task` raises `ConfigurationError` for options it used to accept silently: `max_attempts` + below 1; a `timeout`, `retry_base` or `retry_max` that is not a finite number of seconds + (`timeout` above zero); a `retry_backoff` other than `"exponential"`, `"fixed"` or a function; a + `retry_jitter` that is not a bool; a `retry_on` holding anything but exception classes. + +### Upgrading +- No schema change and no stop needed. + +## 0.15.0 - 2026-11-09 + +### Added +- Per-task limits: `@app.task(concurrency=N, rate="10/m")`, stored in the store and enforced at + the claim across every worker. `tinyjobs.Limit` is exported. +- A `held` job status for jobs a limit stops; they keep their dedup key, can be cancelled, and are + admitted in queue order when the task has room. + +### Changed +- `JobStatus` has a new member, `held`. `tinyjobs stats` and the dashboard's queue table gain a + column for it between queued and running. +- The `check` finding about jobs for an unregistered task says "waiting job(s)" instead of + "queued job(s)"; the dashboard's "Queued by task" became "Waiting by task". +- `jobs cancel` and the dashboard accept held jobs; refusal messages mention "held". +- Eager mode ignores limits; `app.drain()` honours them. +- On Django a held job reads as `READY`. + +### Upgrading +- Schema: SQLite 4 to 5, Postgres 1 to 2, Redis 2 to 3 (the held status and a limits table, keys + on Redis). +- Stop every tinyjobs process on a store before starting the first 0.15 one; older processes + refuse to start against the upgraded store. + +## 0.14.0 - 2026-11-08 + +### Added +- Transactional enqueue on SQLite: `tx=` takes a `sqlite3` connection or cursor on the app's file, + and the job commits or rolls back with the caller's transaction. A process whose first call is + such an enqueue installs the schema through the caller's connection. +- `prefix=` on SQLite, so two apps can share one file or the queue can live in the application's + own database. +- Django's `"database"` option also works on a SQLite alias whose `NAME` is the file in `url`. + +### Changed +- The SQLite tables are renamed `tinyjobs_jobs`, `tinyjobs_executions`, `tinyjobs_schedules` and + `tinyjobs_meta` (they were `jobs`, `executions`, `schedules` and `meta`). + +### Upgrading +- Schema: SQLite 3 to 4. +- Stop every tinyjobs process on the file, workers and enqueuing application processes alike, + then start one 0.14 process: it renames the tables in place and keeps every row. A worker left + running fails every call until stopped, and a job it held runs again when its lease expires. A + pre-0.14 file is migrated under the default prefix only. + +## 0.13.0 - 2026-11-06 + +### Added +- `tinyjobs check` audits the store's configuration against the guarantees: SQLite journal mode, + `synchronous`, filesystem and free space; Redis persistence, eviction policy, memory and role; + Postgres `fsync`, `synchronous_commit` and standby. +- `tinyjobs check --store-warnings` (`app.check(store_warnings=True)`) reports those findings as + `WARN` without failing. +- `tinyjobs verify`, which kills workers under steady load on a store of its own and checks that no + job is lost and every side effect is applied once. +- A guarantees page in the documentation. + +### Changed +- `tinyjobs check` exits 1 on store configuration findings (for example a Redis without + `appendonly`, or a SQLite file not in WAL mode) that passed before; use `--store-warnings` on + development setups. +- Every Postgres connection tinyjobs opens sets `synchronous_commit` to `on` where the server + default is `off`. +- A SQLite app refuses options the SQLite backend does not take (anything but `busy_timeout`) with + `ConfigurationError` when it is built. +- The documentation moved from the README into one page per topic. + +### Upgrading +- No schema change. Check readiness probes that run `tinyjobs check` against development stores. + +## 0.12.0 - 2026-11-06 + +### Added +- `tinyjobs dashboard` (on `127.0.0.1:8070` by default, `--host`/`--port`): queues and health, + jobs with their attempts and tracebacks, schedules, with retry, cancel, pause, resume and run. + No dependencies; no login, meant to sit behind a reverse proxy. +- JSON endpoints `/api/overview`, `/api/jobs`, `/api/jobs/` and `/api/schedules`. +- `tinyjobs.dashboard.wsgi(app)` and `asgi(app)` to mount the pages in an existing app at any + prefix. + +### Upgrading +- No schema change. + +## 0.11.0 - 2026-11-05 + +### Added +- `tinyjobs schedules pause` / `resume` / `run`, and `app.pause_schedule()`, + `app.resume_schedule()`, `app.run_schedule()`. A paused schedule lists as `paused` and shows as + an ok line in `check`. +- `TinyJobs(max_result_bytes=)`: a larger result fails the job with `ResultTooLarge`, without + retry. New exceptions `ResultError`, `ResultTooLarge`, `UnencodableResult`. +- `@app.task(result_ttl=)` to drop a stored result after a while while keeping the job. +- Job fields (`job_id`, `task`, `queue`, `attempt`, `worker_id`, `duration_ms`, and on failure + `error`, `error_message`, `will_retry`, `traceback`) on every log record, and `--log-format json` + on every command; `tinyjobs.logs.JSONFormatter` and `TextFormatter`. +- A Prometheus exporter: `tinyjobs worker --metrics-port` (`--metrics-host`), and + `app.metrics.render()` for embedded workers. +- An operating guide in the documentation. + +### Changed +- `tinyjobs schedules` takes subcommands; with none it lists, as before. +- Durations given on the command line must be finite. + +### Upgrading +- No schema change. Upgrade every worker, scheduler and `tinyjobs schedules` to 0.11 before pausing + a schedule: older ones read a pause as a clock that moved back and un-pause it, and an older + `tinyjobs schedules` fails to list it. + +## 0.10.0 - 2026-11-05 + +### Added +- `@app.task(isolated=True)` runs a `def` task in a reusable job process: hard timeouts (the + process is killed), a crash fails the job with `ProcessDied`, and the process group is killed on + stop. +- Worker options `processes`, `jobs_per_process` and `process_setup`, with `--processes`, + `--jobs-per-process` and `--process-setup`; `TinyJobs(process_setup=)`. +- `ProcessDied` and `IsolatedTaskError` exceptions. Exceptions raised in a job process come back + with their own type and traceback, so `retry_on` matches them. + +### Upgrading +- No schema change. + +## 0.9.0 - 2026-11-04 + +### Added +- On free-threaded builds such as 3.14t the worker keeps the GIL off, and `def` tasks run in + parallel. +- `loops=` on the worker and `tinyjobs worker --loops N`, to run several event loops in one + process (above 1 it is refused on any store but Redis and Postgres, and in an embedded worker). + +### Changed +- `worker.run()` and `tinyjobs worker` open a connection pool of their own for claims, outcomes and + cron, plus one per extra loop, besides the app's pool, which opens only when your code uses it. + Size Postgres `max_connections` or the PgBouncer pool per pool, not per process. + +### Fixed +- On Redis, jobs whose claim reply was lost are handed back when the worker closes. + +### Upgrading +- No schema change. + +## 0.8.0 - 2026-11-03 + +### Added +- A Postgres backend (`pip install 'tinyjobs[postgres]'`, Postgres 14 to 18), with + `LISTEN`/`NOTIFY` wake-ups (`notify=False` to turn off), `prefix=` and `schema=`, and the + server's clock. +- Transactional enqueue on Postgres: `tx=` takes a psycopg connection or cursor, sync or async. +- Django's `"database"` option: every enqueue joins that Postgres database's + `transaction.atomic()`. + +### Changed +- Eager mode runs only the job it sent, ahead of any older job, and returns it finished. Earlier + releases ran every due job on each eager call; call `app.drain()` to run the rest of the queue. + A backend plugin without `claim_job` keeps the old behaviour. +- Shutdown fits inside `shutdown_timeout` (30 s), with time kept for handing back running jobs and + closing connections. A second Ctrl-C under `tinyjobs worker` now cancels running jobs and hands + them back at once; a job that ignores interrupts makes the process exit with status 130 a second + later. +- A threaded `def` task past its timeout counts against `concurrency` until its thread returns, and + a warning names the task. The timeout counts from when the job's thread starts. +- A retention sweep that reaches its cap of 50,000 jobs runs another straight away. +- On Redis and Postgres a worker keeps retrying a job's outcome during a server outage, up to a + second before the lease runs out, so a job finished during a short outage is not run again. + +### Fixed +- Redis scripts the server lost (after a restart or `SCRIPT FLUSH`) are reloaded. +- Every backend call and every stop is bounded by a deadline, so an outage, a stop signal or a stuck + thread no longer hangs a worker. + +### Upgrading +- No schema change (Postgres starts at schema 1). Check tests that relied on eager mode running + older queued jobs. + +## 0.7.3 - 2026-10-27 + +### Added +- Documentation published on Read the Docs, linked from PyPI. + +## 0.7.2 - 2026-10-27 + +### Changed +- README soak figures measured on a Linux VPS, on SQLite and on Redis. No code changes. + +## 0.7.1 - 2026-10-27 + +### Changed +- On Redis, leases, due times, reaping, retention and job ids follow the Redis server's clock + instead of each machine's, so machines with skewed clocks no longer reclaim each other's running + jobs. A `run_at` or `Retry(at=...)` you pass is moved onto the server's clock. A `clock=` you + pass is used as given. +- On Django, results report `run_after` only while the job waits for its first claim. + +### Upgrading +- On Redis, 0.7.0 used each machine's own clock. Where machine clocks drift from the Redis server's + by more than a lease, stop or upgrade every 0.7.0 process, workers and senders alike, before + starting 0.7.1 workers. A Redis user restricted by ACLs needs `TIME` granted. + +## 0.7.0 - 2026-10-26 + +### Added +- A `django.tasks` backend, `tinyjobs.contrib.django.TinyJobsBackend`, with queues, priorities, + `run_after`, async tasks, `takes_context`, `get_result`, the task signals, and + `manage.py tinyjobs` for every CLI command. + +### Changed +- Requires Python 3.12 or newer. +- A tinyjobs call made from a signal handler that lands while the thread holds a tinyjobs lock or + write raises `InterruptedCallError` instead of hanging. +- A call on a closed thread executor raises `ExecutorClosedError` (a subclass of `AppClosedError`). + +### Fixed +- "database is locked" when two processes open a fresh SQLite file at once. +- Interrupted or abandoned SQLite writes are rolled back; concurrent and cancelled closes no longer + stall the loop or leak threads; a forked child does not inherit a pending close. + +### Upgrading +- Schema: SQLite 2 to 3, Redis 1 to 2. +- Stop every 0.6 worker before starting 0.7. The first 0.7 process migrates the store, and from + then on a 0.6 process refuses to start against it. + +## 0.6.0 - 2026-10-22 + +### Added +- `app.lifespan()` and `app.serve()` for FastAPI and Starlette, with an optional embedded worker. +- `max_connections=` and `pool_timeout=` for the Redis pool. + +### Changed +- The Redis pool waits for a free connection instead of failing when all 100 are taken. +- A job whose task the worker does not know is handed back with a delay that doubles each time, up + to `unknown_task_max_delay` (an hour), instead of every minute. +- A closed app no longer reconnects: `enqueue()`, `drain()` and a scheduler tick raise + `AppClosedError`; `start()`, `serve()` or starting a worker open it again. + +### Fixed +- Sync calls from a `def` endpoint on Redis failed every time; the sync facade now runs on the loop + that connected the backend, and raises `EventLoopStoppedError` if that loop stops. +- `RetryPolicy` no longer overflows past attempt 1024. + +## 0.5.0 - 2026-10-12 + +### Added +- A Redis backend, `pip install 'tinyjobs[redis]'` (schema 1). +- `tinyjobs.testing.BackendContract`, the test suite any backend plugin can run. + +### Fixed +- A job whose outcome could not be recorded because the store was unreachable is no longer + recorded as a task failure; it runs again once its lease expires. + +## 0.4.0 - 2026-10-10 + +### Added +- Cron schedules (`@app.task(cron=..., timezone=..., misfire=..., overlap=...)`), fired + leader-free from every worker, with `tinyjobs scheduler`, `tinyjobs schedules`, + `--no-scheduler` and `--misfire-grace`. +- Lifecycle hooks `on_job_start`, `on_job_success`, `on_job_failure`. +- `memory://`, `eager=True`, `app.drain()` and `FrozenClock`, now exported, for testing. +- `tinyjobs stats` and `tinyjobs check` (with `--max-wait`), and `app.stats()` / `app.check()`. + +## 0.3.1 - 2026-09-03 + +### Changed +- `tinyjobs purge` prints durations as `1w`, `30d` instead of seconds. + +### Fixed +- `tinyjobs purge` failed with `BackendNotReadyError` when it was the first command run. + +## 0.3.0 - 2026-09-03 + +### Added +- Retention: finished jobs are pruned automatically, successes and cancellations after a week, + failures after thirty days (`RetentionPolicy`, `KEEP_EVERYTHING`), with worker flags + `--retain-success`, `--retain-cancelled`, `--retain-failed`, `--purge-interval`, `--no-purge`. +- `tinyjobs purge`, with `--before` and `--dry-run`. + +### Changed +- Finished jobs are now deleted after the retention period by default. + +### Fixed +- `tinyjobs jobs list --status` sorted every matching row on large databases; it now has an index + in the order it lists. + +### Upgrading +- Schema: SQLite 1 to 2, migrated automatically on first open. + +## 0.2.0 - 2026-09-02 + +### Added +- `Job.wait()` and `Job.awaited()`, with an optional timeout that raises `WaitTimeout` (also a + `TimeoutError`). + +### Fixed +- The CLI reported the wrong version (0.1.1 said 0.1.0); the version now comes from the package + metadata. + +## 0.1.1 - 2026-09-02 + +### Added +- `Job.refresh()` and `Job.result()`, which returns the decoded result of a `store_result=True` + task. + +### Changed +- `Job.result` (the stored bytes) is renamed `Job.result_payload`. + +## 0.1.0 - 2026-09-02 + +### Added +- First release: background jobs on SQLite, with no dependencies, on Python 3.11 or newer. +- Task registry, `task.enqueue()` with signature validation, `app.send()` by name. +- Async backend protocol with a synchronous facade; atomic claim, leases and a reaper. +- Worker with async and threaded executors, timeouts and graceful shutdown on `SIGTERM`. +- Retries with backoff and jitter, priorities, named queues, delayed jobs, deduplication keys. +- `tinyjobs worker`, `jobs` and `send`, also as `python -m tinyjobs`; the backend plugin seam. diff --git a/README.md b/README.md index af49211..ac4a9a5 100644 --- a/README.md +++ b/README.md @@ -201,6 +201,8 @@ Everything else is at https://tinyjobs.readthedocs.io/: [How it works](https://tinyjobs.readthedocs.io/en/latest/how-it-works/), [Measured](https://tinyjobs.readthedocs.io/en/latest/measured/), [Status](https://tinyjobs.readthedocs.io/en/latest/status/), + [Compatibility](https://tinyjobs.readthedocs.io/en/latest/compatibility/), + [Changelog](https://github.com/Piergiuseppe/tinyjobs/blob/main/CHANGELOG.md), [Design notes](https://tinyjobs.readthedocs.io/en/latest/design/) Young, working and measured: [Status](https://tinyjobs.readthedocs.io/en/latest/status/) lists diff --git a/docs/compatibility.md b/docs/compatibility.md new file mode 100644 index 0000000..4626fbf --- /dev/null +++ b/docs/compatibility.md @@ -0,0 +1,82 @@ +# Compatibility + +tinyjobs is at 0.x: it works and it is measured, and parts of it are still being settled. This page +says what a release may change, how you find out, and what has to be true before a 1.0. + +## What is public + +Public, and treated with the care described below: + +- every name in `tinyjobs.__all__`, whether or not a page here shows it, and the names these + pages tell you to import from `tinyjobs.testing`, `tinyjobs.dashboard`, + `tinyjobs.contrib.django`, `tinyjobs.logs` and `tinyjobs.metrics`; +- the options of `@app.task`, `with_options` and `TinyJobs(...)` that these pages show; +- the `tinyjobs` commands and their flags, and their exit statuses where a page states them; +- the Prometheus metric names and labels, and the fields of the JSON log lines; +- the URL schemes, the store options tinyjobs defines itself, and the Django settings keys. + +Not public, even where nothing hides it: + +- any module, class, method or attribute these pages do not show. Much of the package has no + leading underscore yet, and `TinyJobs` carries methods the worker and the integrations use + between themselves; an import that works is not a promise; +- the tables, columns and keys of a store. Read them for a one-off question, as + [How it works](how-it-works.md) shows, and expect a release to change them through its own + migration; +- the JSON behind the dashboard, and the exact wording of messages, log lines and command output. + +Provisional: + +- the [backend protocol](other-backends.md), which is `BaseBackend`, `Backend` and `Capabilities` + in `tinyjobs.backends`, and its contract tests in `tinyjobs.testing`. Every release that added a + feature a store must carry has changed them, and some of what the core asks of a store is found + by name and not yet declared. A third-party store should pin the minor release it was tested on. + `Job`, `Page`, `Stats`, `QueueStats`, `Execution` and `Finding` are public for the code that + reads them; the fields a store has to fill in are part of this protocol and may grow with it. + +## What a release may change + +- From 0.18, a patch release fixes defects and changes nothing else that is public or provisional, + nor a store's layout, and needs no stop. +- A minor release, 0.x, adds things and may change something public. Every such change is listed + in the [changelog](https://github.com/Piergiuseppe/tinyjobs/blob/main/CHANGELOG.md) under + *Changed*, with what to do about it. +- From 0.18, a public name that is renamed or removed keeps working for at least one minor release + and warns with a `DeprecationWarning` that names its replacement. Where that cannot be done, the + changelog says so in the release that breaks it. +- `JobStatus` and `ExecutionStatus` may gain members in a minor release, as `held` and `waiting` + were added; code that handles every status should have a default branch. + +## Stores and upgrades + +- A release that changes a store's layout upgrades it on the first call. Upgrading to 0.17 itself + needed every process stopped; every release after it keeps the previous minor release working + beside it unless its upgrade note says otherwise, so a rolling deploy from one minor to the next + needs no stop. The upgrade notes are in the changelog, under *Upgrading*. +- A process of a release that a store no longer allows is refused when it starts: every call + raises and nothing is written. From 0.18 the error is `tinyjobs.NewerStoreError`, and a command + prints it and exits with status 2; older releases raise a `RuntimeError`. A process already + running when the store is upgraded does not notice, which is why an upgrade that drops the + previous minor asks for every process to be stopped first. +- A rolling deploy covers one minor at a time. An upgrade that skips a minor is not one: stop every + process of the older release first, and follow the upgrade note of each release it passes. +- Going back works only to a release the store still allows, the previous minor unless the upgrade + note says otherwise. Going further back means restoring a backup taken before the upgrade. + +## Python and dependencies + +tinyjobs runs on the Python versions its package metadata lists, 3.12 and newer today. From 0.18, +support for a Python version ends in a minor release, after that version has reached its own end +of life. + +## Before 1.0 + +1.0 is a promise that the public surface above, and the backend protocol with it, stops changing +in incompatible ways outside a new major version. It will not be made until: + +- the backend protocol declares everything the core asks of a store, and a feature release has + gone by without changing it; +- what is public is separated in the code from what is not; +- a release that changes a store's layout has passed the upgrade check every release runs, with + the previous minor working beside it on SQLite, Postgres and Redis. The check exists from 0.18 + and has not yet met a change of layout. diff --git a/docs/design.md b/docs/design.md index 150cf1c..947e2e9 100644 --- a/docs/design.md +++ b/docs/design.md @@ -527,11 +527,11 @@ exhausted, this job is dead. That distinction is why a separate `RETRYING` job s query today; retrofitting it means migrating a live table later. Semantics: cancelling a `QUEUED` job prevents it from ever being claimed. Cancelling a `RUNNING` job sets the flag and stops future retries. The in-flight attempt is *not* killed, because Python cannot safely interrupt arbitrary -code. Async tasks may cooperate by checking `job.is_cancelled()`. +code. An attempt that finishes after the cancellation cannot report: the store refuses its outcome. ### Execution status -`RUNNING → SUCCESS | FAILED | TIMEOUT | LOST` +`RUNNING → SUCCESS | FAILED | TIMEOUT | LOST | CANCELLED` `LOST` is what the reaper writes when a lease expires: we know the attempt started, we will never know how it ended. Keeping it distinct from `FAILED` matters: a spike in `LOST` means workers are @@ -613,16 +613,16 @@ Plain JSON is too thin: no `datetime`, no `Decimal`, no `UUID`. So the default c tagged-value convention, roughly 60 lines of standard library: ```python -{"$t": "datetime", "v": "2026-08-31T17:34:00+02:00"} -{"$t": "decimal", "v": "19.99"} -{"$t": "uuid", "v": "..."} -{"$t": "set", "v": [1, 2, 3]} -{"$t": "bytes", "v": ""} +{"$t": "datetime", "$v": "2026-08-31T17:34:00+02:00"} +{"$t": "decimal", "$v": "19.99"} +{"$t": "uuid", "$v": "..."} +{"$t": "set", "$v": [1, 2, 3]} +{"$t": "bytes", "$v": ""} ``` Covered out of the box: `str`, `int`, `float`, `bool`, `None`, `list`, `dict`, `tuple`, `set`, -`frozenset`, `bytes`, `datetime`, `date`, `time`, `timedelta`, `Decimal`, `UUID`, `Path`, `Enum`, and -`Path`. Anything whose reconstruction needs a class (a `@dataclass`, an `Enum`, a domain object) +`frozenset`, `bytes`, `datetime`, `date`, `time`, `timedelta`, `Decimal`, `UUID` and `Path`. +Anything whose reconstruction needs a class (a `@dataclass`, an `Enum`, a domain object) must be **registered**, and this is a deliberate refusal rather than a gap: ```python @@ -690,15 +690,14 @@ class Backend(Protocol): async def claim(self, *, worker_id: str, queues: Sequence[str], limit: int, lease_seconds: float, limits: Mapping[str, Limit] | None = None) -> list[Job]: ... - async def claim_job(self, job_id: str, *, worker_id: str, - lease_seconds: float) -> Job | None: ... async def heartbeat( self, job_ids: Sequence[str], lease_seconds: float, worker_id: str | None = None - ) -> None: ... + ) -> Collection[str] | None: ... # the ids whose lease it renewed async def complete(self, job: Job, result: bytes | None) -> bool: ... async def fail(self, job: Job, error: ExecutionError, retry_at: float | None, execution_status: ExecutionStatus = ExecutionStatus.FAILED) -> bool: ... async def release(self, job: Job, delay: float = 0.0) -> bool: ... + async def release_many(self, jobs: Sequence[Job]) -> int: ... async def hand_back(self, job: Job, delay: float = 0.0) -> bool: ... async def release_held(self, tasks: Collection[str], queues: Sequence[str] | None = None) -> int: ... @@ -722,15 +721,41 @@ class Backend(Protocol): # --- housekeeping ----------------------------------------------------- async def reap_expired(self) -> int: ... async def purge(self, policy: RetentionPolicy, now: float) -> int | None: ... + async def expire_results(self, task: str, ttl: float, now: float, + batch: int = 500, since: float = 0.0) -> int: ... ``` +`BaseBackend` gives defaults for `setup`, `close`, `enqueue_many`, `hand_back`, `release_many`, +`release_held`, `purge`, `expire_results` and the three schedule methods, so a store writes the +other fourteen. It also defines `follow_clock`, which a store that reads its own server's time +overrides. Beside the protocol the core looks for a few names a store may leave out, and does +without them: + +- `claim_job(job_id, *, worker_id, lease_seconds)`, which lets an eager call run the one job it + sent; +- `wait_for_jobs`, which lets a worker sleep until a job arrives instead of polling; +- `transient_errors`, a tuple of the store's own exceptions that mean an outage, through which a + worker keeps retrying a job's outcome instead of losing it; +- `reserve`, which tells a pooled store how many jobs a worker runs and returns, as a whole number + of at least 1, how many rounds of hand-backs the worker should allow for when it stops, and + `spare`, which has the store size its pool to what was reserved; +- `audit()`, which answers `tinyjobs check`; +- `refuse_held_transaction`, with which SQLite refuses a write made while the caller holds a + transaction on the same file; +- for a store that declares `transactional_enqueue`, `enqueue_in(tx, jobs)` and + `aenqueue_in(tx, jobs)`. + +These are found by name, not declared, and they are the part of this seam most likely to change +before 1.0 (see [Compatibility](compatibility.md)). + `enqueue_many` earns its place as a distinct method rather than a loop in the core, because **the transaction boundary is the optimization**. A thousand separate `enqueue` calls on SQLite means a -thousand fsyncs; the same thousand inside one transaction means one, which is two orders of magnitude. -Redis backends will want pipelining, Postgres a multi-row `INSERT`. None of that is reachable if the -core does the looping. And since the protocol freezes at v1.0, a seam that is not there at v1.0 cannot -be added later, so it goes in now, with a default implementation in `BaseBackend` (a loop over -`enqueue()`) so that simple backends get the correct behaviour for free. +thousand fsyncs; the same thousand inside one transaction means one, which is two orders of +magnitude. Redis backends will want pipelining, Postgres a multi-row `INSERT`. None of that is +reachable if the core does the looping. After v1.0 a new method could only arrive with a default, so +it goes in now, with a default implementation in `BaseBackend` (a loop over `enqueue()`), so that +simple backends get the correct behaviour for free and the bundled stores override it from the +start. `claim_job` claims one job by its id, only while it is queued and due, which is how an eager call runs the job it sent and no other (see [Testing](#testing)). It is optional: without it an eager @@ -798,20 +823,24 @@ Not every store can do everything. A Redis list has no cheap priority ordering, Backends declare what they support so the core can degrade predictably instead of failing at runtime: ```python -class Capabilities(NamedTuple): - priority: bool - delayed: bool - dedup: bool - transactional_enqueue: bool - retention: bool - schedules: bool - max_payload_bytes: int | None - loop_bound: bool - result_ttl: bool - limits: bool - after: bool +@dataclass(frozen=True, slots=True) +class Capabilities: + priority: bool = True + delayed: bool = True + dedup: bool = False + transactional_enqueue: bool = False + retention: bool = False + schedules: bool = False + max_payload_bytes: int | None = None + loop_bound: bool = False + result_ttl: bool = False + limits: bool = False + after: bool = False ``` +A store names only what differs from the defaults, so a capability added later leaves it as it was. +`max_payload_bytes` is declared and not yet enforced by the core. + Requesting an unsupported feature raises at *enqueue* time with a clear message, not silently later. `limits` says the store enforces a task's `concurrency=` and `rate=` at the claim, which then takes `limits=`, a mapping of task name to `Limit`; a worker or a drain whose app registers a limited task @@ -830,9 +859,13 @@ dependency-free: ```toml # in tinyjobs-redis's pyproject.toml [project.entry-points."tinyjobs.backends"] -redis = "tinyjobs_redis:RedisBackend" +redis = "tinyjobs_redis:RedisBackend.from_url" ``` +The entry point names a factory, which the app calls as `factory(url, clock=clock, **options)`: +the URL it was given, the app's clock, which the store reads instead of `time.time()`, and the +keyword options passed to `TinyJobs`. + The user's side of the swap, in full: ```python @@ -840,10 +873,13 @@ app = TinyJobs("sqlite:///jobs.db") # before app = TinyJobs("redis://localhost/0") # after ``` -The URL scheme selects the backend. The core package never imports, references or knows the names of -optional backends; it looks up the scheme in the entry point group and raises a helpful error -if nothing is registered for it. Bundled: `sqlite://`, `memory://`, `redis://` behind the `redis` -extra and `postgres://` behind the `postgres` extra, each client imported only when its URL is used. +The URL scheme selects the backend. A scheme the package does not bundle is looked up in the entry +point group, with a helpful error if nothing is registered for it. A few answers still come from the +bundled schemes only, such as which stores Django can join a transaction on, which are shared +between processes, and which get a connection pool of their own in `worker.run()` and may run more +than one loop; a plugin cannot give them yet. Bundled: `sqlite://`, `memory://`, `redis://` behind +the `redis` extra and `postgres://` behind the `postgres` extra, each client imported only when its +URL is used. --- @@ -1063,29 +1099,29 @@ are the reason the backend seam exists. ## The Redis backend -Bundled rather than a separate `tinyjobs-redis` package: until the protocol freezes at v1.0, a change -to it would otherwise need two releases in lockstep. It depends on nothing but the public protocol, so -it can move out later without touching a line. The client library is an extra, `tinyjobs[redis]`, and -is imported only when a `redis://` URL is used, so the core stays dependency-free. A url redis-py -cannot read raises `ConfigurationError`, without its reason once the url holds a password, since -that reason may quote part of it. So does a connection option the client does not take, named -unless the url holds a password, before the host or as a `password=` option: a raw `?` in the -password made the rest of the url such an option, which redis-py raised at the first command as a -`TypeError` quoting the password, and a raw `&` in a `password=` value made the part after it one. -A `unix://` url whose socket path holds an `@` behind a password is refused too, since redis-py -took the rest of the password for the path and quoted it when the connection failed. So are -`decode_responses`, which broke every call, and a `timeout=` that is not a number, which redis-py -5.0 kept as text and failed every call with; one that is a number is read as one. So are the -client's own retries (`retry` with any retries, `retry_on_timeout`, `retry_on_error`), on the url -or as keywords. A client handed in, redis-py 8's default included, gives tinyjobs its settings, -not its connections: tinyjobs opens its own pool from them with no retries and leaves the client as -it was, since redis-py keeps the retry on each connection and pinning a shared pool would take the -retries from other code using it. A call sent again after its answer was lost runs its script twice, -so one enqueue counted two `queued` jobs and the `complete` +Bundled rather than a separate `tinyjobs-redis` package: until the protocol stops changing +incompatibly at v1.0, a change to it would otherwise need two releases in lockstep. It depends on +nothing but the public protocol, so it can move out later without touching a line. The client +library is an extra, `tinyjobs[redis]`, and is imported only when a `redis://` URL is used, so the +core stays dependency-free. A url redis-py cannot read raises `ConfigurationError`, without its +reason once the url holds a password, since that reason may quote part of it. So does a connection +option the client does not take, named unless the url holds a password, before the host or as a +`password=` option: a raw `?` in the password made the rest of the url such an option, which +redis-py raised at the first command as a `TypeError` quoting the password, and a raw `&` in a +`password=` value made the part after it one. A `unix://` url whose socket path holds an `@` behind +a password is refused too, since redis-py took the rest of the password for the path and quoted it +when the connection failed. So are `decode_responses`, which broke every call, and a `timeout=` that +is not a number, which redis-py 5.0 kept as text and failed every call with; one that is a number is +read as one. So are the client's own retries (`retry` with any retries, `retry_on_timeout`, +`retry_on_error`), on the url or as keywords. A client handed in, redis-py 8's default included, +gives tinyjobs its settings, not its connections: tinyjobs opens its own pool from them with no +retries and leaves the client as it was, since redis-py keeps the retry on each connection and +pinning a shared pool would take the retries from other code using it. A call sent again after its +answer was lost runs its script twice, so one enqueue counted two `queued` jobs and the `complete` sent again found the job no longer running, returned false and left the counts at one `queued` and one `success`. Refusing them is simpler than making every script safe to run twice, and tinyjobs -already recovers a lost answer itself: outcomes are written again only after checking the claim, -and a lost claim is handed back by its token. +already recovers a lost answer itself: outcomes are written again only after checking the claim, and +a lost claim is handed back by its token. On Python 3.12 a command written to a connection the server had reset raised asyncio's `TypeError: 'NoneType' object is not callable`, a bug fixed in 3.13, so a job whose outcome met it @@ -3445,28 +3481,30 @@ should be benchmarked honestly before anyone opens a `.c` file. **v0.1, the core loop** Task registry, `task.enqueue()` with signature validation, `app.send()` by name, SQLite backend with -the atomic partial-index claim, single worker, sync and async executors, retries with backoff, JSON -codec, `tinyjobs worker`, `tinyjobs jobs`. +the atomic partial-index claim, leases and the reaper, a worker with sync and async executors and a +graceful shutdown, retries with backoff, queues, priorities, delayed jobs and deduplication keys, +the JSON codec, `tinyjobs worker`, `tinyjobs jobs`, and the backend plugin seam. -**v0.2, time** -Cron schedules, `run_at` / delayed jobs, the embedded leader-free scheduler, misfire and overlap -policies, timezone and DST handling. +**v0.1.1 and v0.2, results** +`Job.refresh()`, `Job.result()` and `Job.wait()`. -**v0.3, production shape** -Leases and the reaper, graceful shutdown, multi-queue routing, priorities, deduplication keys, -lifecycle hooks, retention and automatic purge. +**v0.3, retention** +Finished jobs pruned automatically, and `tinyjobs purge`. -**v0.4, comfort** -Result storage, `memory://` backend and the conformance suite, eager and drain test modes, +**v0.4, time and comfort** +Cron schedules with the embedded leader-free scheduler, misfire and overlap policies, timezone and +DST handling; lifecycle hooks; the `memory://` backend, eager and drain test modes; `tinyjobs stats` / `check`. -**v0.5 to v0.6, integrations** -FastAPI and Starlette lifespans with an embedded worker, the `django.tasks` backend and -`manage.py tinyjobs`, the backend plugin seam and its contract suite. +**v0.5, Redis** +A Redis backend on Lua scripts, and the contract suite every backend runs. -**v0.7, Redis** -A Redis backend on Lua scripts, with per-claim tokens, batched outcomes, scripts reloaded when the -server loses them, and the server's clock followed instead of the worker's. +**v0.6, FastAPI** +FastAPI and Starlette lifespans with an embedded worker. + +**v0.7, Django** +The `django.tasks` backend and `manage.py tinyjobs`. On Redis, the server's clock followed instead +of each machine's. **v0.8, Postgres** A Postgres backend with `LISTEN`/`NOTIFY` wake-ups, enqueue inside the caller's transaction, also @@ -3513,11 +3551,20 @@ the backend in the job process. **v0.17, one job after another** `after=job`: a job waits, in a status of its own, until another has succeeded, and is cancelled with everything behind it when that one fails. The store settles it in the step that ends the -parent. With it, `compatible_since`, so that later upgrades need no stop. +parent. With it, `compatible_since`, so that later upgrades between neighbouring minors need no +stop. + +**v0.18, settling** +No feature. A changelog and a [compatibility page](compatibility.md) that says what is public and +what a release may change; the exceptions and return types a caller meets exported from the +package; `NewerStoreError` for a store too new; the in-memory template backend made to pass the +contract; and a release check that upgrades a store of the release before and shares it between +the two versions. **v1.0, commitment** -Frozen backend protocol, documented plugin API, migration tooling, full documentation. After this the -protocol is a compatibility promise. +A backend protocol that changes only compatibly, a documented plugin API, full documentation. After +this the protocol is a compatibility promise. [Compatibility](compatibility.md) lists what has to be +true of the protocol and of the public surface first. **Beyond** Joining a SQLAlchemy session's transaction, designed against a real application rather than against a diff --git a/docs/fastapi.md b/docs/fastapi.md index 3a5bd93..64b22a1 100644 --- a/docs/fastapi.md +++ b/docs/fastapi.md @@ -28,19 +28,20 @@ async def lifespan(api: FastAPI): yield ``` -The process exits within `shutdown_timeout` (30 s), or 1.25 s when that is shorter, so that releasing -still has its second: a last moment, half a second at most, is kept for closing connections, and of -the rest running jobs and a claim still in flight get the first three quarters and releasing what is -still running the last, one second at least; a second Ctrl-C under a server does not cut it short. -Under `tinyjobs worker` it does: running jobs are cancelled and go back to the queue at once, even -when one blocks the event loop; a job that swallows every interrupt makes the process exit a second -later with status 130, leaving its job to the reaper. A `def` job whose +The process exits within `shutdown_timeout` (30 s), or 1.25 s when that is shorter, so that +releasing still has its second: a last moment, half a second at most, is kept for closing +connections, and of the rest running jobs and a claim still in flight get the first three quarters +and releasing what is still running the last, one second at least; a second Ctrl-C under a server +does not cut it short. Under `tinyjobs worker` it does: running jobs are cancelled and go back to +the queue at once, even when one blocks the event loop; a job that swallows every interrupt makes +the process exit a second later with status 130, leaving its job to the reaper. A `def` job whose thread is still running is not released; the worker waits for it and records its outcome within a quarter of the timeout, one second at least, after the thread returns. Keep it under your -orchestrator's grace period: `docker stop` allows 10 s, and past it the process is killed and its jobs -wait out their leases instead of going straight back to the queue. A worker that cannot reach its -backend when it starts, or whose setup fails, exits with status 1 as RQ does, so let the orchestrator -restart it. +orchestrator's grace period: `docker stop` allows 10 s, and past it the process is killed and its +jobs wait out their leases instead of going straight back to the queue. Under `tinyjobs worker`, a +worker that cannot reach its backend when it starts, or whose setup fails, exits with a status other +than 0, so let the orchestrator restart it: 2 for an error tinyjobs raises itself, such as a +configuration or a store it refuses, and 1 for any other. Starlette takes the same `lifespan`. Leave out `worker=True` and the server only enqueues, with `tinyjobs worker --app tasks:app` running the jobs elsewhere. diff --git a/docs/guarantees.md b/docs/guarantees.md index 5661aa7..7bcb39d 100644 --- a/docs/guarantees.md +++ b/docs/guarantees.md @@ -61,7 +61,7 @@ number, so a loss cannot hide behind a send that failed after the store had take enqueued twice is caught. `--duration`, `--rate`, `--workers`, `--slots`, `--kill-every`, `--lease` and `--settle` set the load, with `--settle` above `--lease` plus the 1 s reap interval since a killed job comes back when its lease ends, and sized to twice the lease when only -`--lease` is raised, `--prefix` the namespace on Redis and Postgres, and `--keep` leaves the +`--lease` is raised, `--prefix` the namespace in the store, and `--keep` leaves the finished jobs in the store for `tinyjobs jobs` and the dashboard instead of purging them. It refuses an in-memory store, since the workers could not share it, and a store that holds other jobs, and purges its own at the end, leaving only the store's schema rows behind, unless other diff --git a/docs/limits.md b/docs/limits.md index 7bd768a..f99b7a2 100644 --- a/docs/limits.md +++ b/docs/limits.md @@ -42,7 +42,7 @@ task is reported like any other. ## What the limits mean exactly - The running count is the jobs of the task in the `running` status, so a job whose worker died - counts until its lease expires and the reaper returns it to the queue, within `--lease-seconds`. + counts until its lease expires and the reaper returns it to the queue, within `--lease`. The other way round, a synchronous job that outlives its `timeout=` is recorded as failed while its thread runs on, and the room it took is free again; a task that must never run past its cap wants `isolated=True` or an async body, which a timeout does stop. diff --git a/docs/other-backends.md b/docs/other-backends.md index b351f6f..c04cef8 100644 --- a/docs/other-backends.md +++ b/docs/other-backends.md @@ -8,22 +8,51 @@ The core package doesn't know about third-party backends. It looks an unknown UR mysql = "tinyjobs_mysql:MySQLBackend.from_url" ``` +The entry point names a factory, called as `factory(url, clock=clock, **options)` with the URL given +to `TinyJobs`, the app's clock and the keyword options the app was built with. A store reads the +time from `clock.now()`, never from `time.time()`, which is what lets tests and +[eager mode](testing-your-tasks.md) move it. + +This seam is provisional until 1.0: see [Compatibility](compatibility.md) for what may still change +and how a change is announced. + Backends declare what they support (`priority`, `delayed`, `dedup`, ...) and asking for something unsupported raises when you enqueue rather than failing quietly later. The protocol is `async`, and `tinyjobs.testing.BackendContract` is the test suite a backend has to pass, the same one SQLite, -Postgres and Redis run. `claim_job` is optional, and its contract test is skipped without it. Without -it, an eager call drains every due job, as earlier releases did, and waits for a job it inserted. -A backend that enforces [limits](limits.md) declares `limits` in its capabilities and takes -`limits=`, a mapping of task name to `tinyjobs.Limit` (`concurrency`, and `interval` in seconds), -on `claim`, and implements `release_held(tasks=, queues=)`, which returns the held jobs of the tasks -named, in the queues named or in every queue, to the queue; the contract tests for -them are skipped without the capability, and a worker or a drain whose app registers a limited task -refuses such a backend at start. A backend that keeps [a job waiting on another](jobs.md#one-job-after-another) +Postgres and Redis run: + +```python +class MySQLContractTests(BackendContract, unittest.IsolatedAsyncioTestCase): + async def make_backend(self, clock): + return MySQLBackend.from_url(URL, clock=clock) +``` + +Subclassing `tinyjobs.backends.BaseBackend` leaves fourteen methods to write: `enqueue`, `claim`, +`heartbeat`, `complete`, `fail`, `release`, `cancel`, `requeue`, `get`, `list`, `executions`, +`reap_expired`, `counts_by_status` and `stats`. `heartbeat` takes `worker_id=` and returns the ids +whose lease it renewed. `claim_job` is optional, and the contract tests that claim by id are skipped +without it, or run without that step. Without it, an eager call drains every due job, as earlier +releases did, and waits for a job it inserted. A store that does not declare `loop_bound` is called +from more than one thread: claims, and the hand-backs of a claim given up, run on the app's own +loop, and every other call on the loop that made it, so state it keeps in the process needs a lock. +The template's lock is a `threading.Lock` held for a whole call, which is safe only because none of +its methods awaits anything; a store that does real I/O under it would stall its loop. The race +tests open several instances of the store; a test class whose store keeps its jobs in the process, +where two instances share nothing, sets `independent_clients = False` and races one instance against +itself. A backend that enforces [limits](limits.md) declares `limits` in its capabilities and takes +`limits=`, a mapping of task name to `tinyjobs.Limit` (`concurrency`, and `interval` in seconds), on +`claim`, and implements `release_held(tasks=, queues=)`, which returns the held jobs of the tasks +named, in the queues named or in every queue, to the queue; the contract tests for them are skipped +without the capability, and a worker or a drain whose app registers a limited task refuses such a +backend at start. A backend that keeps [a job waiting on another](jobs.md#one-job-after-another) declares `after`: it places a job whose `after` names a parent by that parent's status when it is inserted, raising `JobNotFoundError` for a parent it does not hold; it queues the waiting children -of a job as it completes it and cancels every waiting descendant of a job that fails -for good or is cancelled; `requeue` asks the parent again; and a waiting job holds its dedup key, which a job placed -cancelled does not, and stays out of the queued figures of `stats`. Without the capability `after=` -raises at enqueue and those contract tests are skipped; one of them cancels a job with 4,000 jobs -waiting on it within `wide_cancel_seconds`, one second unless the test class sets another. `tests/test_async_backend.py` implements a minimal in-memory backend in about 130 -lines if you want a template. +of a job as it completes it and cancels every waiting descendant of a job that fails for good or is +cancelled; `requeue` asks the parent again; and a waiting job holds its dedup key, which a job +placed cancelled does not, and stays out of the queued figures of `stats`. Without the capability +`after=` raises at enqueue and those contract tests are skipped; one of them cancels a job with +4,000 jobs waiting on it within `wide_cancel_seconds`, one second unless the test class sets +another. + +`tests/test_async_backend.py` in the repository holds `RamBackend`, an in-memory backend of about +320 lines that passes the contract with the defaults and `retention`, if you want a template. diff --git a/docs/status.md b/docs/status.md index 6ad5282..bd9a084 100644 --- a/docs/status.md +++ b/docs/status.md @@ -84,5 +84,8 @@ Upgrading from 0.6: stop every 0.6 worker before starting 0.7. The first 0.7 pro store, and from then on a 0.6 process refuses to start against it. 0.7 also needs Python 3.12 or newer. +[Compatibility](compatibility.md) says what a release may change, and the +[changelog](https://github.com/Piergiuseppe/tinyjobs/blob/main/CHANGELOG.md) what each one did. + The [design notes](design.md) explain the design and why it looks like this, including the parts not built yet. diff --git a/docs/tests.md b/docs/tests.md index c8bc83c..2566a98 100644 --- a/docs/tests.md +++ b/docs/tests.md @@ -24,3 +24,19 @@ export TINYJOBS_REDIS_URL=redis://localhost:6379/0 export TINYJOBS_POSTGRES_URL=postgresql://postgres:tinyjobs@localhost:5432/postgres python -m unittest discover -s tests -t . ``` + +One test needs another release. `tests/test_upgrade.py` lets the previous release create a store and +fill it, opens it with this one, and then has both send, run and read each other's jobs, four +processes draining at once at the end, on SQLite and on whichever of Postgres and Redis is set. +Where this release says the previous one cannot work beside it, the test expects the older process +to be refused and the jobs it left to run. It skips unless it is told where the previous release is, +the newest one below the version in `tinyjobs/__init__.py`, `<0.18` here: + +```bash +python -m venv .venv-previous +.venv-previous/bin/pip install 'tinyjobs[redis,postgres]<0.18' +TINYJOBS_PREVIOUS_PYTHON=$PWD/.venv-previous/bin/python python -m unittest tests.test_upgrade +``` + +The workflow runs it on every pull request, push to main and release tag, against the release before +the one in the tree. diff --git a/docs/watching-it.md b/docs/watching-it.md index 0056ab0..d506355 100644 --- a/docs/watching-it.md +++ b/docs/watching-it.md @@ -74,8 +74,10 @@ tinyjobs_jobs_total{task="tasks.generate_pdf",outcome="success"} 1842 tinyjobs_job_duration_seconds_bucket{task="tasks.generate_pdf",le="0.5"} 1790 tinyjobs_jobs_running{task="tasks.generate_pdf"} 2 tinyjobs_store_jobs{queue="default",status="queued"} 14 +tinyjobs_store_queued_jobs{task="tasks.generate_pdf"} 14 tinyjobs_store_oldest_due_seconds{queue="default"} 0.8 tinyjobs_store_expired_leases 0 +tinyjobs_info{version="0.18.0"} 1 ``` The counters, the duration histogram and the running gauge are this process's, fed by the job diff --git a/mkdocs.yml b/mkdocs.yml index 8e1a2c8..9c4c9e2 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -41,6 +41,7 @@ nav: - How it works: how-it-works.md - Measured: measured.md - Status: status.md + - Compatibility: compatibility.md - Project: - Working on it: working-on-it.md - Tests: tests.md diff --git a/pyproject.toml b/pyproject.toml index 6e5bda5..a22cb49 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,7 @@ Homepage = "https://github.com/Piergiuseppe/tinyjobs" Documentation = "https://tinyjobs.readthedocs.io/" Repository = "https://github.com/Piergiuseppe/tinyjobs" Issues = "https://github.com/Piergiuseppe/tinyjobs/issues" +Changelog = "https://github.com/Piergiuseppe/tinyjobs/blob/main/CHANGELOG.md" [project.scripts] tinyjobs = "tinyjobs.cli:run" diff --git a/tests/support.py b/tests/support.py index 6681786..83f742d 100644 --- a/tests/support.py +++ b/tests/support.py @@ -35,6 +35,14 @@ class Injected(BaseException): pass +def freed(reference, seconds: float = 5.0) -> bool: + deadline = time.monotonic() + seconds + while reference() is not None and time.monotonic() < deadline: + gc.collect() + time.sleep(0.05) + return reference() is None + + def flush_control(seconds: float = 5) -> None: from tinyjobs import runtime diff --git a/tests/test_after.py b/tests/test_after.py index 44f709a..c2ac4ac 100644 --- a/tests/test_after.py +++ b/tests/test_after.py @@ -15,7 +15,12 @@ from tinyjobs.backends.sqlite.schema import COMPATIBLE_SINCE, SCHEMA_VERSION from tinyjobs.cli import main from tinyjobs.dashboard.core import Dashboard, Request -from tinyjobs.exceptions import JobNotFoundError, UnsupportedCapabilityError, WaitTimeout +from tinyjobs.exceptions import ( + JobNotFoundError, + NewerStoreError, + UnsupportedCapabilityError, + WaitTimeout, +) from tinyjobs.metrics import _store_lines from tinyjobs.following import AFTER_MESSAGE, PARENT_ENDED_MESSAGE, parent_id from tinyjobs.testing.contract import make_job @@ -455,10 +460,25 @@ def test_a_newer_file_this_release_cannot_work_on_is_refused(self): prepare() app = TinyJobs(f"sqlite:///{self.database}") self.addCleanup(app.close) - with self.assertRaisesRegex(RuntimeError, "newer than this tinyjobs"): + with self.assertRaisesRegex(NewerStoreError, "newer than this tinyjobs"): app.start() self.assertEqual(self.meta(), {"schema_version": newer}) + def test_a_command_on_a_newer_file_says_so_and_leaves_with_a_usage_status(self): + self.opened().close() + self.sql( + "UPDATE tinyjobs_meta SET value = ? " + "WHERE key IN ('schema_version', 'compatible_since')", + str(SCHEMA_VERSION + 1), + ) + app = TinyJobs(f"sqlite:///{self.database}") + self.addCleanup(app.close) + errors = io.StringIO() + with contextlib.redirect_stderr(errors): + self.assertEqual(main(["stats"], app=app), 2) + self.assertRegex(errors.getvalue(), r"^error: database schema version \d+ is newer") + self.assertNotIn("Traceback", errors.getvalue()) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_app.py b/tests/test_app.py index f098527..7818305 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -39,6 +39,7 @@ disarm_alarm, flush_control, fork_without_warning, + freed, interrupt_repeatedly, race_open_and_close, sweep_interrupts, @@ -456,6 +457,33 @@ def from_a_thread() -> None: self.assertEqual(self.fresh.counts()["queued"], 8) +class PublicNamesTests(unittest.TestCase): + def test_every_exception_a_caller_can_meet_is_importable_from_the_package(self): + import tinyjobs + from tinyjobs import exceptions + + defined = { + name + for name, value in vars(exceptions).items() + if isinstance(value, type) + and issubclass(value, BaseException) + and value.__module__ == exceptions.__name__ + } + self.assertEqual(defined - set(tinyjobs.__all__), set()) + + def test_every_exported_name_exists_once(self): + import tinyjobs + + self.assertEqual(len(tinyjobs.__all__), len(set(tinyjobs.__all__))) + self.assertEqual([name for name in tinyjobs.__all__ if not hasattr(tinyjobs, name)], []) + + def test_the_error_for_a_newer_store_is_still_a_runtime_error(self): + import tinyjobs + + self.assertTrue(issubclass(tinyjobs.NewerStoreError, tinyjobs.TinyJobsError)) + self.assertTrue(issubclass(tinyjobs.NewerStoreError, RuntimeError)) + + class FailedConnectionTests(BackendTestCase): def test_an_option_sqlite_does_not_take_is_refused_before_connecting(self): for url in ("sqlite://", f"sqlite:///{self.database}", "memory://"): @@ -929,9 +957,8 @@ def test_an_app_dropped_after_an_enqueue_stops_its_loop(self): loop_thread = app._runner._thread collected = weakref.ref(app) del app, handler - gc.collect() + self.assertTrue(freed(collected)) loop_thread.join(5) - self.assertIsNone(collected()) self.assertFalse(loop_thread.is_alive()) def test_a_child_connects_although_a_parent_thread_held_the_app_lock(self): diff --git a/tests/test_async_backend.py b/tests/test_async_backend.py index 00a892d..ee70f03 100644 --- a/tests/test_async_backend.py +++ b/tests/test_async_backend.py @@ -5,75 +5,119 @@ import threading import time import unittest +from collections import Counter +from contextlib import ExitStack from dataclasses import replace +from functools import wraps +from itertools import count +from unittest import mock from tinyjobs import TinyJobs from tinyjobs.backends import Backend, BaseBackend, Capabilities, register_builtin +from tinyjobs.backends.base import LEASE_EXPIRED_MESSAGE from tinyjobs.backends.managed import ManagedBackend +from tinyjobs.clock import Clock from tinyjobs.exceptions import AppClosedError, SynchronousCallInsideEventLoopError -from tinyjobs.models import ExecutionError, Job, Page, QueueStats, Stats +from tinyjobs.models import Execution, ExecutionError, Job, Page, QueueStats, Stats from tinyjobs.runtime import LoopRunner from tinyjobs.status import ExecutionStatus, JobStatus +from tinyjobs.testing import BackendContract + + +def locked(method): + @wraps(method) + async def called(self, *args, **kwargs): + with self._lock: + return await method(self, *args, **kwargs) + + return called class RamBackend(BaseBackend): - capabilities = Capabilities(dedup=True, retention=True) + capabilities = Capabilities(retention=True) - def __init__(self, url: str, **options) -> None: + def __init__(self, url: str, *, clock: Clock | None = None, **options) -> None: + self._clock = clock or Clock() self._jobs: dict[str, Job] = {} - self._attempts: dict[str, list[str]] = {} - self._lock = asyncio.Lock() - self.thread_hops = 0 + self._order: dict[str, int] = {} + self._sent = count() + self._runs: dict[str, list[Execution]] = {} + self._purged_at: float | None = None + self._lock = threading.Lock() @classmethod def from_url(cls, url: str, **options) -> "RamBackend": return cls(url, **options) + @locked async def enqueue(self, job: Job) -> Job: - async with self._lock: - self._jobs[job.id] = job - return job - - async def claim(self, *, worker_id, queues, limit, lease_seconds) -> list[Job]: - now = time.time() - async with self._lock: - ready = [ - job - for job in self._jobs.values() - if job.status is JobStatus.QUEUED - and job.queue in queues - and job.run_at <= now - ] - ready.sort(key=lambda job: (-job.priority, job.run_at)) - claimed = [] - for job in ready[:limit]: - updated = replace( - job, - status=JobStatus.RUNNING, - worker_id=worker_id, - attempts=job.attempts + 1, - claims=job.claims + 1, - lease_expires_at=now + lease_seconds, + self._jobs[job.id] = replace(job) + self._order[job.id] = next(self._sent) + self._runs[job.id] = [] + return job + + @locked + async def claim( + self, *, worker_id, queues, limit, lease_seconds, limits=None + ) -> list[Job]: + now = self._clock.now() + due = [ + job + for job in self._jobs.values() + if job.status is JobStatus.QUEUED and job.queue in queues and job.run_at <= now + ] + due.sort(key=lambda job: (-job.priority, job.run_at, self._order[job.id])) + claimed = [] + for job in due[:limit]: + taken = self._put( + job, + status=JobStatus.RUNNING, + worker_id=worker_id, + attempts=job.attempts + 1, + claims=job.claims + 1, + lease_expires_at=now + lease_seconds, + ) + self._runs[job.id].append( + Execution( + job.id, + taken.attempts, + ExecutionStatus.RUNNING, + worker_id, + started_at=now, + claim=taken.claims, ) - self._jobs[job.id] = updated - self._attempts.setdefault(job.id, []).append("running") - claimed.append(updated) - return claimed - - async def heartbeat(self, job_ids, lease_seconds) -> None: - return None - - async def complete(self, job: Job, result: bytes | None) -> bool: - async with self._lock: - current = self._held(job) - if current is None: - return False - self._jobs[job.id] = replace( - current, status=JobStatus.SUCCESS, result_payload=result, hand_backs=0 ) - self._attempts[job.id][-1] = "success" - return True + claimed.append(replace(taken)) + return claimed + + @locked + async def heartbeat(self, job_ids, lease_seconds, worker_id=None) -> list[str]: + renewed = [] + for job_id in job_ids: + job = self._jobs.get(job_id) + if job is None or job.status is not JobStatus.RUNNING: + continue + if worker_id is not None and job.worker_id != worker_id: + continue + self._put(job, lease_expires_at=self._clock.now() + lease_seconds) + renewed.append(job_id) + return renewed + + @locked + async def complete(self, job: Job, result: bytes | None) -> bool: + held = self._held(job) + if held is None: + return False + self._end( + held, + JobStatus.SUCCESS, + ExecutionStatus.SUCCESS, + result_payload=result, + last_error=None, + ) + return True + @locked async def fail( self, job: Job, @@ -81,114 +125,243 @@ async def fail( retry_at: float | None, execution_status: ExecutionStatus = ExecutionStatus.FAILED, ) -> bool: - async with self._lock: - current = self._held(job) - if current is None: - return False - self._jobs[job.id] = replace( - current, - status=JobStatus.QUEUED if retry_at else JobStatus.FAILED, - run_at=retry_at or current.run_at, - worker_id=None, - last_error=error.summary, - hand_backs=0, - ) - self._attempts[job.id][-1] = str(execution_status) - return True + held = self._held(job) + if held is None: + return False + if retry_at is None: + self._end(held, JobStatus.FAILED, execution_status, error, last_error=error.summary) + else: + self._requeue(held, execution_status, error, run_at=retry_at, last_error=error.summary) + return True + @locked async def release(self, job: Job, delay: float = 0.0) -> bool: - async with self._lock: - current = self._held(job) - if current is None: - return False - self._jobs[job.id] = replace( - current, status=JobStatus.QUEUED, worker_id=None, hand_backs=0 - ) - return True + held = self._held(job) + if held is None: + return False + self._requeue(held, ExecutionStatus.LOST, run_at=self._clock.now() + delay) + return True + @locked async def hand_back(self, job: Job, delay: float = 0.0) -> bool: - async with self._lock: - current = self._held(job) - if current is None: - return False - self._jobs[job.id] = replace( - current, - status=JobStatus.QUEUED, - worker_id=None, - attempts=current.attempts - 1, - hand_backs=current.hand_backs + 1, - ) - self._attempts[job.id].pop() - return True - - def _held(self, job: Job) -> Job | None: - current = self._jobs.get(job.id) - if ( - current is None - or current.status is not JobStatus.RUNNING - or current.worker_id != job.worker_id - or current.claims != job.claims - ): - return None - return current + held = self._held(job) + if held is None: + return False + self._runs[held.id].pop() + self._put( + held, + status=JobStatus.QUEUED, + run_at=self._clock.now() + delay, + attempts=held.attempts - 1, + hand_backs=held.hand_backs + 1, + worker_id=None, + lease_expires_at=None, + ) + return True + @locked async def cancel(self, job_id: str) -> bool: - async with self._lock: - self._jobs[job_id] = replace( - self._jobs[job_id], status=JobStatus.CANCELLED - ) - return True + job = self._jobs.get(job_id) + if job is None or job.status.is_terminal: + return False + if job.status is JobStatus.RUNNING: + self._close(job, ExecutionStatus.CANCELLED) + self._put( + job, + status=JobStatus.CANCELLED, + finished_at=self._clock.now(), + lease_expires_at=None, + hand_backs=0, + ) + return True + @locked async def requeue(self, job_id: str) -> bool: + job = self._jobs.get(job_id) + if job is None or job.status not in (JobStatus.FAILED, JobStatus.CANCELLED): + return False + self._put( + job, + status=JobStatus.QUEUED, + run_at=job.created_at if job.claims == 0 else self._clock.now(), + attempts=0, + last_error=None, + finished_at=None, + worker_id=None, + lease_expires_at=None, + ) return True + @locked async def get(self, job_id: str) -> Job | None: - return self._jobs.get(job_id) + job = self._jobs.get(job_id) + return None if job is None else replace(job) + @locked async def list(self, *, status=None, queue=None, limit=100, cursor=None) -> Page: - items = sorted(self._jobs.values(), key=lambda job: job.id, reverse=True) - if status is not None: - items = [job for job in items if job.status is status] - return Page(items=items[:limit]) + found = [ + job + for job in sorted(self._jobs.values(), key=lambda job: -self._order[job.id]) + if (cursor is None or self._order[job.id] < int(cursor)) + and (status is None or job.status is status) + and (queue is None or job.queue == queue) + ] + page = found[:limit] + more = len(found) > limit + return Page( + items=[replace(job) for job in page], + cursor=str(self._order[page[-1].id]) if more else None, + ) - async def executions(self, job_id: str) -> list: - return self._attempts.get(job_id, []) + @locked + async def executions(self, job_id: str) -> list[Execution]: + return [replace(run) for run in self._runs.get(job_id, [])] + @locked async def reap_expired(self) -> int: - return 0 + now = self._clock.now() + expired = [ + job + for job in self._jobs.values() + if job.status is JobStatus.RUNNING and job.lease_expires_at < now + ] + for job in expired: + if job.attempts < job.max_attempts: + self._requeue( + job, ExecutionStatus.LOST, run_at=now, last_error=LEASE_EXPIRED_MESSAGE + ) + else: + self._end( + job, + JobStatus.FAILED, + ExecutionStatus.LOST, + run_at=now, + worker_id=None, + last_error=LEASE_EXPIRED_MESSAGE, + ) + return len(expired) + @locked async def counts_by_status(self) -> dict[str, int]: - counts: dict[str, int] = {} - for job in self._jobs.values(): - counts[str(job.status)] = counts.get(str(job.status), 0) + 1 - return counts + return dict(Counter(str(job.status) for job in self._jobs.values())) + @locked async def stats(self, now: float) -> Stats: - queues: dict[str, dict[str, int]] = {} - for job in self._jobs.values(): - counts = queues.setdefault(job.queue, {}) - counts[str(job.status)] = counts.get(str(job.status), 0) + 1 - return Stats(queues=[QueueStats(queue, counts) for queue, counts in queues.items()]) + queues = [] + for queue in sorted({job.queue for job in self._jobs.values()}): + jobs = [job for job in self._jobs.values() if job.queue == queue] + due = [ + job.run_at + for job in jobs + if job.status is JobStatus.QUEUED and job.run_at <= now + ] + queues.append( + QueueStats( + queue, + dict(Counter(str(job.status) for job in jobs)), + min(due, default=None), + ) + ) + running = [job for job in self._jobs.values() if job.status is JobStatus.RUNNING] + queued = [job for job in self._jobs.values() if job.status is JobStatus.QUEUED] + return Stats( + queues=queues, + expired_leases=sum(job.lease_expires_at < now for job in running), + queued_tasks=dict(Counter(job.task for job in queued)), + ) + @locked async def purge(self, policy, now: float) -> int | None: if not policy.enabled: return None - horizons = policy.horizons(now) - async with self._lock: - doomed = [ - job.id for job in self._jobs.values() - if job.status in horizons + if self._purged_at is not None and now - self._purged_at < policy.interval: + return None + self._purged_at = now + doomed: list[Job] = [] + for status, horizon in policy.horizons(now).items(): + past = [ + job + for job in self._jobs.values() + if job.status is status and job.finished_at is not None - and job.finished_at < horizons[job.status] + and job.finished_at < horizon ] - for job_id in doomed: - del self._jobs[job_id] + past.sort(key=lambda job: job.finished_at) + doomed.extend(past[: policy.maximum_per_sweep - len(doomed)]) + for job in doomed: + del self._jobs[job.id], self._runs[job.id], self._order[job.id] return len(doomed) + def _held(self, job: Job) -> Job | None: + current = self._jobs.get(job.id) + if ( + current is None + or current.status is not JobStatus.RUNNING + or current.worker_id != job.worker_id + or current.claims != job.claims + ): + return None + return current + + def _put(self, job: Job, **changes) -> Job: + self._jobs[job.id] = replace(job, **changes) + return self._jobs[job.id] + + def _close( + self, job: Job, status: ExecutionStatus, error: ExecutionError | None = None + ) -> None: + run = self._runs[job.id][-1] + run.status = status + run.finished_at = self._clock.now() + if error is not None: + run.error_type, run.traceback = error.type, error.traceback + + def _end( + self, + job: Job, + status: JobStatus, + ran: ExecutionStatus, + error: ExecutionError | None = None, + **changes, + ) -> None: + self._close(job, ran, error) + self._put( + job, + status=status, + finished_at=self._clock.now(), + lease_expires_at=None, + hand_backs=0, + **changes, + ) + + def _requeue( + self, + job: Job, + ran: ExecutionStatus, + error: ExecutionError | None = None, + **changes, + ) -> None: + self._close(job, ran, error) + self._put( + job, + status=JobStatus.QUEUED, + worker_id=None, + lease_expires_at=None, + hand_backs=0, + **changes, + ) + register_builtin("ram", RamBackend.from_url) +class RamContractTests(BackendContract, unittest.IsolatedAsyncioTestCase): + independent_clients = False + + async def make_backend(self, clock): + return RamBackend("ram://", clock=clock) + + class NativeAsyncBackendTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.app = TinyJobs("ram://") @@ -200,7 +373,7 @@ async def asyncTearDown(self) -> None: async def test_a_pure_async_backend_satisfies_the_protocol(self): self.assertIsInstance(self.app.backend, Backend) - async def test_jobs_run_end_to_end_without_a_single_thread_hop(self): + async def test_claims_run_on_the_apps_loop_and_every_other_call_where_it_was_made(self): done = [] @self.app.task @@ -208,19 +381,35 @@ async def handler(value): done.append(value) return value - jobs = [await handler.aenqueue(n) for n in range(5)] - worker = self.app.worker(concurrency=4, reap_interval=60) - runner = asyncio.create_task(worker.start()) - for _ in range(200): - await asyncio.sleep(0.01) - statuses = [(await self.app.aget(job.id)).status for job in jobs] - if all(status.is_terminal for status in statuses): - break - worker.stop() - await asyncio.wait_for(runner, timeout=10) + threads: dict[str, set[str]] = {} + + def watched(name, method): + async def called(*args, **kwargs): + threads.setdefault(name, set()).add(threading.current_thread().name) + return await method(*args, **kwargs) + return called + + with ExitStack() as watching: + for name in ("enqueue", "claim", "complete", "get"): + watching.enter_context( + mock.patch.object(RamBackend, name, watched(name, getattr(RamBackend, name))) + ) + jobs = [await handler.aenqueue(n) for n in range(5)] + worker = self.app.worker(concurrency=4, reap_interval=60) + runner = asyncio.create_task(worker.start()) + for _ in range(200): + await asyncio.sleep(0.01) + statuses = [(await self.app.aget(job.id)).status for job in jobs] + if all(status.is_terminal for status in statuses): + break + worker.stop() + await asyncio.wait_for(runner, timeout=10) + + here = {threading.current_thread().name} self.assertEqual(sorted(done), [0, 1, 2, 3, 4]) - self.assertEqual(self.app.backend.thread_hops, 0) + self.assertEqual(threads.pop("claim"), {"tinyjobs-loop"}) + self.assertEqual(threads, {"enqueue": here, "complete": here, "get": here}) async def test_eager_mode_runs_jobs_on_a_backend_that_cannot_claim_one_by_id(self): self.assertFalse(hasattr(RamBackend, "claim_job")) diff --git a/tests/test_postgres.py b/tests/test_postgres.py index 6296ae7..ab8962e 100644 --- a/tests/test_postgres.py +++ b/tests/test_postgres.py @@ -21,7 +21,7 @@ from tinyjobs import deadlines from tinyjobs.services import lifecycle from tinyjobs import JobStatus, TinyJobs -from tinyjobs.exceptions import ConfigurationError, UnsupportedCapabilityError +from tinyjobs.exceptions import ConfigurationError, NewerStoreError, UnsupportedCapabilityError from tinyjobs.following import PARENT_ENDED_MESSAGE, PARENT_GONE_MESSAGE from tinyjobs.limits import Limit from tinyjobs.runtime import run_on @@ -356,10 +356,10 @@ async def test_a_store_from_a_newer_release_is_refused_and_left_alone(self): "WHERE key IN ('schema_version', 'compatible_since')", {"v": newer}, ) - with self.assertRaisesRegex(RuntimeError, "newer than this tinyjobs"): + with self.assertRaisesRegex(NewerStoreError, "newer than this tinyjobs"): await self.backend.setup() await sql(f"DELETE FROM \"{self.prefix}_meta\" WHERE key = 'compatible_since'") - with self.assertRaisesRegex(RuntimeError, "newer than this tinyjobs"): + with self.assertRaisesRegex(NewerStoreError, "newer than this tinyjobs"): await self.backend.setup() stored = await sql( f"SELECT value FROM \"{self.prefix}_meta\" WHERE key = 'schema_version'" diff --git a/tests/test_redis.py b/tests/test_redis.py index 8bcdbe8..1e40275 100644 --- a/tests/test_redis.py +++ b/tests/test_redis.py @@ -14,7 +14,7 @@ import tinyjobs.clock from tinyjobs import JobStatus, TinyJobs -from tinyjobs.exceptions import AppClosedError, ConfigurationError +from tinyjobs.exceptions import AppClosedError, ConfigurationError, NewerStoreError from tinyjobs.limits import Limit from tinyjobs.runtime import run_on from tinyjobs.status import ExecutionStatus @@ -225,10 +225,10 @@ async def test_a_store_from_a_newer_release_is_refused_and_left_alone(self): since = self.backend._keys("compatible_since") await self.backend._client.set(version, SCHEMA_VERSION + 1) await self.backend._client.set(since, SCHEMA_VERSION + 1) - with self.assertRaisesRegex(RuntimeError, "newer than this tinyjobs"): + with self.assertRaisesRegex(NewerStoreError, "newer than this tinyjobs"): await self.backend.setup() await self.backend._client.delete(since) - with self.assertRaisesRegex(RuntimeError, "newer than this tinyjobs"): + with self.assertRaisesRegex(NewerStoreError, "newer than this tinyjobs"): await self.backend.setup() self.assertEqual( int(await self.backend._client.get(version)), SCHEMA_VERSION + 1 diff --git a/tests/test_retention.py b/tests/test_retention.py index ec8f2c2..c387992 100644 --- a/tests/test_retention.py +++ b/tests/test_retention.py @@ -5,7 +5,7 @@ import unittest from tinyjobs import TinyJobs -from tinyjobs.exceptions import ConfigurationError +from tinyjobs.exceptions import ConfigurationError, NewerStoreError from tinyjobs.retention import KEEP_EVERYTHING, RetentionPolicy from tinyjobs.status import JobStatus @@ -516,7 +516,7 @@ async def open_it() -> None: finally: await app.aclose() - with self.assertRaisesRegex(RuntimeError, "newer than this tinyjobs"): + with self.assertRaisesRegex(NewerStoreError, "newer than this tinyjobs"): asyncio.run(open_it()) handle = sqlite3.connect(path) tables = [row[0] for row in handle.execute("SELECT name FROM sqlite_master")] diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 6b7fc8d..c0fecf9 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -17,6 +17,7 @@ from tinyjobs import deadlines, runtime from tinyjobs.flag import Flag from tinyjobs.runtime import FORK_PAUSE_SECONDS, LoopRunner +from tests.support import freed def loop_threads() -> list[threading.Thread]: @@ -245,8 +246,7 @@ def test_a_loop_paused_and_then_closed_is_let_go(self): let_go = weakref.ref(self.loop) self.runner.close() del self.loop - gc.collect() - self.assertIsNone(let_go()) + self.assertTrue(freed(let_go)) def test_a_child_refuses_a_call_to_a_loop_paused_in_its_parent(self): import os @@ -2125,8 +2125,7 @@ async def held() -> None: gone = weakref.ref(loop) with self.assertLogs("asyncio", "ERROR"): del loop, task - gc.collect() - self.assertIsNone(gone()) + self.assertTrue(freed(gone)) def test_many_open_bounds_share_one_timer(self): async def main(): diff --git a/tests/test_upgrade.py b/tests/test_upgrade.py new file mode 100644 index 0000000..3e122c4 --- /dev/null +++ b/tests/test_upgrade.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +import json +import os +import re +import sqlite3 +import subprocess +import sys +import tempfile +import unittest +import uuid +from pathlib import Path + +import tinyjobs + +PREVIOUS = os.environ.get("TINYJOBS_PREVIOUS_PYTHON") +POSTGRES_URL = os.environ.get("TINYJOBS_POSTGRES_URL") +REDIS_URL = os.environ.get("TINYJOBS_REDIS_URL") +REPOSITORY = str(Path(tinyjobs.__file__).parents[1]) +RACED = 300 +STEP_SECONDS = 120 + +DRIVER = ''' +import json +import os +import sys + +import tinyjobs +from tinyjobs import TinyJobs + +url, prefix, action = sys.argv[1:4] +app = TinyJobs(url, **({"prefix": prefix} if prefix else {})) + + +@app.task(name="upgrade.add", store_result=True) +def add(number, other): + with open(os.environ["LEDGER"], "a") as ledger: + ledger.write(f"{number}\\n") + return number + other + + +def versions(): + from importlib import import_module + + found = {} + for store, module in ( + ("sqlite", "tinyjobs.backends.sqlite.schema"), + ("postgres", "tinyjobs.backends.postgres.schema"), + ("redis", "tinyjobs.backends.redis.backend"), + ): + try: + loaded = import_module(module) + except ImportError: + continue + found[store] = [ + loaded.SCHEMA_VERSION, + getattr(loaded, "COMPATIBLE_SINCE", loaded.SCHEMA_VERSION), + ] + return {"release": tinyjobs.__version__, "source": tinyjobs.__file__, "stores": found} + + +try: + if action == "versions": + print(json.dumps(versions())) + elif action == "send": + first, last = int(sys.argv[4]), int(sys.argv[5]) + print(json.dumps([add.enqueue(number, 1).id for number in range(first, last)])) + elif action == "drain": + print(json.dumps(app.drain())) + elif action == "results": + found = [app.get(job_id) for job_id in json.loads(sys.argv[4])] + print(json.dumps([[str(job.status), job.result()] for job in found])) + elif action == "counts": + print(json.dumps(app.counts())) +finally: + app.close() +''' + + +def numbered(release: str) -> tuple[int, ...]: + return tuple(int(part) for part in re.findall(r"\d+", release)[:3]) + + +class Release: + def __init__(self, python: str, root: Path, ledger: Path, *, source: str | None) -> None: + self.python = python + self.root = root + self.environment = { + name: value for name, value in os.environ.items() if name != "PYTHONPATH" + } + self.environment["LEDGER"] = str(ledger) + self.environment["PYTHONDONTWRITEBYTECODE"] = "1" + if source is not None: + self.environment["PYTHONPATH"] = source + + def start(self, url: str, prefix: str, action: str, *arguments: object) -> subprocess.Popen: + return subprocess.Popen( + [self.python, "driver.py", url, prefix, action, *map(str, arguments)], + cwd=self.root, + env=self.environment, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + def ends(self, running: subprocess.Popen) -> tuple[int, str, str]: + try: + output, errors = running.communicate(timeout=STEP_SECONDS) + except subprocess.TimeoutExpired: + running.kill() + running.communicate() + raise + return running.returncode, output, errors + + def run(self, url: str, prefix: str, action: str, *arguments: object): + code, output, errors = self.ends(self.start(url, prefix, action, *arguments)) + if code != 0: + raise AssertionError(f"{action} left with {code}: {errors[-2000:]}") + return json.loads(output) + + def refused(self, url: str, prefix: str, action: str, *arguments: object) -> str: + code, _, errors = self.ends(self.start(url, prefix, action, *arguments)) + if code == 0: + raise AssertionError(f"{action} was expected to be refused and it ran") + return errors + + +@unittest.skipUnless( + PREVIOUS, + "set TINYJOBS_PREVIOUS_PYTHON to an interpreter with the previous release installed", +) +class UpgradeTests(unittest.TestCase): + def setUp(self) -> None: + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) + (self.root / "driver.py").write_text(DRIVER) + self.ledger = self.root / "ledger" + self.ledger.touch() + self.old = Release(PREVIOUS, self.root, self.ledger, source=None) + self.new = Release(sys.executable, self.root, self.ledger, source=REPOSITORY) + self.old_versions = self.old.run("memory://", "", "versions") + self.new_versions = self.new.run("memory://", "", "versions") + self.assertFalse(self.old_versions["source"].startswith(REPOSITORY)) + self.assertTrue(self.new_versions["source"].startswith(REPOSITORY)) + self.assertLess( + numbered(self.old_versions["release"]), numbered(self.new_versions["release"]) + ) + + def ran(self) -> list[int]: + return sorted(int(line) for line in self.ledger.read_text().split()) + + def works_beside(self, store: str) -> bool: + understood, _ = self.old_versions["stores"][store] + _, since = self.new_versions["stores"][store] + return since <= understood + + def succeeded(self, release: Release, url: str, prefix: str, ids: list[str], first: int): + found = release.run(url, prefix, "results", json.dumps(ids)) + self.assertEqual( + found, [["success", number + 1] for number in range(first, first + len(ids))] + ) + + def upgrade(self, store: str, url: str, prefix: str, stored) -> None: + old, new = self.old, self.new + before = old.run(url, prefix, "send", 0, 20) + self.assertEqual(old.run(url, prefix, "drain"), 20) + waiting = old.run(url, prefix, "send", 20, 40) + self.assertEqual(stored(), self.old_versions["stores"][store][0]) + sent = new.run(url, prefix, "send", 40, 60) + self.assertEqual(stored(), self.new_versions["stores"][store][0]) + self.succeeded(new, url, prefix, before, 0) + self.assertEqual(new.run(url, prefix, "counts"), {"queued": 40, "success": 20}) + if not self.works_beside(store): + self.assertIn("newer than this tinyjobs", old.refused(url, prefix, "counts")) + self.assertIn("newer than this tinyjobs", old.refused(url, prefix, "send", 0, 1)) + self.assertEqual(new.run(url, prefix, "drain"), 40) + self.succeeded(new, url, prefix, waiting + sent, 20) + self.assertEqual(self.ran(), list(range(60))) + return + self.assertEqual(old.run(url, prefix, "drain"), 40) + self.succeeded(new, url, prefix, waiting + sent, 20) + late = old.run(url, prefix, "send", 60, 80) + self.assertEqual(new.run(url, prefix, "drain"), 20) + self.succeeded(old, url, prefix, late, 60) + raced = new.run(url, prefix, "send", 80, 80 + RACED // 2) + raced += old.run(url, prefix, "send", 80 + RACED // 2, 80 + RACED) + racers = [release.start(url, prefix, "drain") for release in (old, new, old, new)] + drained = [] + for release, running in zip((old, new, old, new), racers): + code, output, errors = release.ends(running) + self.assertEqual(code, 0, errors[-2000:]) + drained.append(json.loads(output)) + self.assertEqual(sum(drained), RACED) + self.succeeded(old, url, prefix, raced, 80) + self.assertEqual(self.ran(), list(range(80 + RACED))) + self.assertEqual(old.run(url, prefix, "counts"), {"success": 80 + RACED}) + self.assertEqual(new.run(url, prefix, "counts"), {"success": 80 + RACED}) + + def sqlite_version(self, prefix: str) -> int: + with sqlite3.connect(self.root / "jobs.db") as connection: + [(stored,)] = connection.execute( + f"SELECT value FROM {prefix}_meta WHERE key = 'schema_version'" + ) + connection.close() + return int(stored) + + def test_a_sqlite_file_of_the_previous_release_is_upgraded_and_shared(self): + url = f"sqlite:///{self.root / 'jobs.db'}" + self.upgrade("sqlite", url, "", lambda: self.sqlite_version("tinyjobs")) + + def test_a_sqlite_file_under_a_prefix_is_upgraded_and_shared(self): + url = f"sqlite:///{self.root / 'jobs.db'}" + self.upgrade("sqlite", url, "shop", lambda: self.sqlite_version("shop")) + + @unittest.skipUnless(POSTGRES_URL, "set TINYJOBS_POSTGRES_URL to run the postgres tests") + def test_a_postgres_store_of_the_previous_release_is_upgraded_and_shared(self): + import psycopg + + prefix = f"tinyjobs_test_{uuid.uuid4().hex[:16]}" + + def drop() -> None: + with psycopg.connect(POSTGRES_URL, autocommit=True) as connection: + tables = connection.execute( + "SELECT tablename FROM pg_tables WHERE schemaname = current_schema() " + "AND left(tablename, %s) = %s", + (len(prefix) + 1, f"{prefix}_"), + ).fetchall() + for (table,) in tables: + connection.execute(f'DROP TABLE IF EXISTS "{table}" CASCADE') + + def stored() -> int: + with psycopg.connect(POSTGRES_URL, autocommit=True) as connection: + [(found,)] = connection.execute( + f"SELECT value FROM \"{prefix}_meta\" WHERE key = 'schema_version'" + ).fetchall() + return int(found) + + self.addCleanup(drop) + self.upgrade("postgres", POSTGRES_URL, prefix, stored) + + @unittest.skipUnless(REDIS_URL, "set TINYJOBS_REDIS_URL to run the redis tests") + def test_a_redis_store_of_the_previous_release_is_upgraded_and_shared(self): + import redis + + prefix = f"tinyjobs-test-{uuid.uuid4().hex}" + + def drop() -> None: + client = redis.Redis.from_url(REDIS_URL) + try: + keys = list(client.scan_iter(match=f"{{{prefix}}}:*", count=10_000)) + if keys: + client.delete(*keys) + finally: + client.close() + + def stored() -> int: + client = redis.Redis.from_url(REDIS_URL) + try: + return int(client.get(f"{{{prefix}}}:schema_version")) + finally: + client.close() + + self.addCleanup(drop) + self.upgrade("redis", REDIS_URL, prefix, stored) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_verify.py b/tests/test_verify.py index fd4e595..e23734e 100644 --- a/tests/test_verify.py +++ b/tests/test_verify.py @@ -170,6 +170,32 @@ def once_deaf(): self.assertIn("did not stop cleanly on SIGTERM", report.lines()[-1]) self.assertEqual(sum(self.counts(url).values()), 0) + def test_the_store_is_not_settled_while_a_job_moved_between_two_reads_by_status(self): + class Moving: + def __init__(self): + self.later = [JobStatus.QUEUED, JobStatus.RUNNING, JobStatus.SUCCESS] + self.cursors = [] + + def jobs(self, *, queue, limit, status=None, cursor=None): + if status is not None: + return Page([], None) + self.cursors.append(cursor) + found = JobStatus.SUCCESS if cursor is None else self.later.pop(0) + return Page([Job(id="r", task=verify.TASK, payload=b"", codec="json", + queue=queue, status=found)], None if cursor else "later") + + def sweep(self): + pass + + moving = Moving() + settings = Settings("sqlite:///jobs.db", **QUICK) + report = Report(settings) + with mock.patch("tinyjobs.verifying.SETTLE_POLL", 0.01): + Verification(settings)._settle(moving, moving, report) + self.assertTrue(report.settled) + self.assertEqual(moving.later, []) + self.assertEqual(moving.cursors, [None, "later"] * 3) + def test_the_load_keeps_the_rate_and_skips_the_kills_it_missed(self): class Fake: def __init__(self): diff --git a/tinyjobs/__init__.py b/tinyjobs/__init__.py index d83c51c..5822ddb 100644 --- a/tinyjobs/__init__.py +++ b/tinyjobs/__init__.py @@ -14,28 +14,49 @@ ) ) -__version__ = "0.17.0" +__version__ = "0.18.0.dev0" from .app import TinyJobs -from .clock import FrozenClock +from .clock import Clock, FrozenClock from .limits import Limit from .exceptions import ( Abandon, + AppClosedError, + BackendNotReadyError, ConfigurationError, + EventLoopStoppedError, + ExecutorClosedError, + ForcedInterrupt, + InterruptedCallError, IsolatedTaskError, JobNotFoundError, + NewerStoreError, ProcessDied, ResultError, ResultTooLarge, Retry, SerializationError, + SynchronousCallInsideEventLoopError, + TaskAlreadyRegisteredError, TinyJobsError, UnencodableResult, UnknownBackendError, + UnknownCodecError, UnknownTaskError, + UnsupportedCapabilityError, WaitTimeout, ) -from .models import Execution, ExecutionError, Job, JobPayload, ScheduleState +from .models import ( + Execution, + ExecutionError, + Finding, + Job, + JobPayload, + Page, + QueueStats, + ScheduleState, + Stats, +) from .retention import KEEP_EVERYTHING, RetentionPolicy from .retry import RetryPolicy from .scheduler import Scheduler @@ -46,35 +67,51 @@ __all__ = [ "Abandon", + "AppClosedError", + "BackendNotReadyError", + "Clock", "ConfigurationError", + "EventLoopStoppedError", "Execution", "ExecutionError", "ExecutionStatus", + "ExecutorClosedError", + "Finding", + "ForcedInterrupt", "FrozenClock", + "InterruptedCallError", "IsolatedTaskError", - "KEEP_EVERYTHING", "Job", "JobNotFoundError", "JobPayload", "JobStatus", + "KEEP_EVERYTHING", "Limit", "MisfirePolicy", + "NewerStoreError", + "Page", "ProcessDied", + "QueueStats", "ResultError", "ResultTooLarge", - "Retry", "RetentionPolicy", + "Retry", "RetryPolicy", "Schedule", - "ScheduleState", "Scheduler", + "ScheduleState", "SerializationError", + "Stats", + "SynchronousCallInsideEventLoopError", "Task", + "TaskAlreadyRegisteredError", "TinyJobs", "TinyJobsError", "UnencodableResult", "UnknownBackendError", + "UnknownCodecError", "UnknownTaskError", + "UnsupportedCapabilityError", "WaitTimeout", "Worker", "__version__", diff --git a/tinyjobs/backends/postgres/backend.py b/tinyjobs/backends/postgres/backend.py index 47ffa11..dcac1eb 100644 --- a/tinyjobs/backends/postgres/backend.py +++ b/tinyjobs/backends/postgres/backend.py @@ -56,6 +56,7 @@ AppClosedError, ConfigurationError, JobNotFoundError, + NewerStoreError, UnsupportedCapabilityError, ) from ...following import PARENT_MISSING_MESSAGE, refuse_parents_among @@ -833,7 +834,7 @@ async def _create(self) -> None: if stored is None and placed.tables: stored = UPGRADABLE_VERSIONS[0] if stored is not None and stored > SCHEMA_VERSION: - raise RuntimeError( + raise NewerStoreError( f"postgres schema version {stored} is newer than this tinyjobs " f"build understands ({SCHEMA_VERSION}); upgrade the package" ) diff --git a/tinyjobs/backends/redis/backend.py b/tinyjobs/backends/redis/backend.py index bd7f55e..658bdd8 100644 --- a/tinyjobs/backends/redis/backend.py +++ b/tinyjobs/backends/redis/backend.py @@ -23,7 +23,12 @@ from redis.exceptions import TimeoutError as RedisTimeoutError from ...clock import Clock, FollowingClock, elapsed -from ...exceptions import AppClosedError, ConfigurationError, JobNotFoundError +from ...exceptions import ( + AppClosedError, + ConfigurationError, + JobNotFoundError, + NewerStoreError, +) from ...following import PARENT_MISSING_MESSAGE, refuse_parents_among from ...limits import Limit from ...models import ( @@ -155,7 +160,7 @@ def from_url( async def setup(self) -> None: stored, since = map(int, await self._run("schema", SCHEMA_VERSION, COMPATIBLE_SINCE)) if stored > SCHEMA_VERSION and since > SCHEMA_VERSION: - raise RuntimeError( + raise NewerStoreError( f"redis schema version {stored} is newer than this tinyjobs " f"build understands ({SCHEMA_VERSION}); upgrade the package" ) diff --git a/tinyjobs/backends/sqlite/backend.py b/tinyjobs/backends/sqlite/backend.py index 434d3b6..354d4dd 100644 --- a/tinyjobs/backends/sqlite/backend.py +++ b/tinyjobs/backends/sqlite/backend.py @@ -27,6 +27,7 @@ AppClosedError, ConfigurationError, JobNotFoundError, + NewerStoreError, UnsupportedCapabilityError, ) from ...following import ( @@ -529,7 +530,7 @@ def _works_on(self, connection: Any, stored: int) -> bool: def _refuse_newer(self, connection: Any, stored: int) -> None: if not self._works_on(connection, stored): - raise RuntimeError( + raise NewerStoreError( f"database schema version {stored} is newer than this tinyjobs " f"build understands ({SCHEMA_VERSION}); upgrade the package" ) diff --git a/tinyjobs/exceptions.py b/tinyjobs/exceptions.py index 1bed4b2..df992e5 100644 --- a/tinyjobs/exceptions.py +++ b/tinyjobs/exceptions.py @@ -49,6 +49,10 @@ class UnsupportedCapabilityError(TinyJobsError): pass +class NewerStoreError(TinyJobsError, RuntimeError): + pass + + class JobNotFoundError(TinyJobsError): pass diff --git a/tinyjobs/testing/contract.py b/tinyjobs/testing/contract.py index da6f5f5..1b95b00 100644 --- a/tinyjobs/testing/contract.py +++ b/tinyjobs/testing/contract.py @@ -75,6 +75,13 @@ def requires(self, capability: str) -> None: if not getattr(self.backend.capabilities, capability): raise unittest.SkipTest(f"the backend does not declare {capability}") + def claims_by_id(self) -> bool: + return hasattr(self.backend, "claim_job") + + def requires_claim_by_id(self) -> None: + if not self.claims_by_id(): + raise unittest.SkipTest("the backend cannot claim a job by its id") + async def enqueue(self, **overrides) -> Job: return await self.backend.enqueue(make_job(**overrides)) @@ -99,8 +106,13 @@ async def test_claimed_jobs_keep_bytes_that_look_like_separators(self): first = await self.enqueue(payload=payload, task="tests.a=b,c", dedup_key="k=1,2\n3") second = await self.enqueue(payload=payload, task="tests.a=b,c", dedup_key="k=4,5\n6") [claimed] = await self.claim(limit=1) - by_id = await self.backend.claim_job(second.id, worker_id="worker-b", lease_seconds=60.0) - for job, expected in ((claimed, first), (by_id, second)): + if self.claims_by_id(): + other = await self.backend.claim_job( + second.id, worker_id="worker-b", lease_seconds=60.0 + ) + else: + [other] = await self.claim(limit=1, worker="worker-b") + for job, expected in ((claimed, first), (other, second)): self.assertEqual(job.id, expected.id) self.assertEqual(job.payload, payload) self.assertEqual(job.task, "tests.a=b,c") @@ -328,8 +340,7 @@ async def test_release_many_returns_the_jobs_still_held_in_one_call(self): self.assertEqual(await self.backend.release_many([]), 0) async def test_a_job_claimed_by_id_is_leased_once_and_only_when_due(self): - if not hasattr(self.backend, "claim_job"): - raise unittest.SkipTest("the backend cannot claim a job by its id") + self.requires_claim_by_id() first = await self.enqueue(run_at=START - 2) wanted = await self.enqueue(run_at=START - 1) later = await self.enqueue(run_at=START + 5) @@ -955,6 +966,7 @@ async def reap(job): async def test_a_claim_by_id_takes_room_from_the_task(self): self.requires("limits") + self.requires_claim_by_id() limits = {"tests.noop": Limit(concurrency=1)} first = await self.enqueue() second = await self.enqueue(run_at=1.0) @@ -1141,9 +1153,12 @@ async def test_a_job_after_another_waits_out_of_the_claims_way(self): self.assertEqual((stored.status, stored.after), (JobStatus.WAITING, parent.id)) self.assertEqual([job.id for job in await self.claim()], [parent.id]) self.assertEqual(await self.claim(), []) - self.assertIsNone( - await self.backend.claim_job(child.id, worker_id="worker-b", lease_seconds=60.0) - ) + if self.claims_by_id(): + self.assertIsNone( + await self.backend.claim_job( + child.id, worker_id="worker-b", lease_seconds=60.0 + ) + ) self.assertEqual((await self.backend.counts_by_status())["waiting"], 1) listed = await self.backend.list(status=JobStatus.WAITING) self.assertEqual([job.id for job in listed.items], [child.id]) diff --git a/tinyjobs/verifying.py b/tinyjobs/verifying.py index ab294d2..8f25f0e 100644 --- a/tinyjobs/verifying.py +++ b/tinyjobs/verifying.py @@ -396,7 +396,7 @@ def _settle(self, app: Any, workers: "_Workers", report: Report) -> None: deadline = time.monotonic() + self._settings.settle while not self._stopped.is_set(): workers.sweep() - if not any(self._pending(app, status) for status in PENDING): + if not self._unsettled(app): return if time.monotonic() >= deadline: report.settled = False @@ -404,6 +404,18 @@ def _settle(self, app: Any, workers: "_Workers", report: Report) -> None: self._stopped.wait(SETTLE_POLL) report.interrupted = True + def _unsettled(self, app: Any) -> bool: + if any(self._pending(app, status) for status in PENDING): + return True + cursor = None + while True: + page = app.jobs(queue=self._queue, limit=PAGE, cursor=cursor) + if any(job.status in PENDING for job in page.items): + return True + cursor = page.cursor + if not cursor or not page.items: + return False + def _pending(self, app: Any, status: JobStatus) -> bool: return bool(app.jobs(queue=self._queue, status=status, limit=1).items)