From ec3444e836f0f16f18dd428d8549ab82f1800cd5 Mon Sep 17 00:00:00 2001 From: Piergiuseppe D'Abbraccio Date: Thu, 12 Nov 2026 15:02:54 +0100 Subject: [PATCH 1/6] Refuse a store too new with NewerStoreError, which commands report as an error instead of a traceback, and export from the package the exceptions and return types a caller meets --- tests/test_after.py | 24 ++++++++++++-- tests/test_app.py | 27 +++++++++++++++ tests/test_postgres.py | 6 ++-- tests/test_redis.py | 6 ++-- tests/test_retention.py | 4 +-- tinyjobs/__init__.py | 47 ++++++++++++++++++++++++--- tinyjobs/backends/postgres/backend.py | 3 +- tinyjobs/backends/redis/backend.py | 9 +++-- tinyjobs/backends/sqlite/backend.py | 3 +- tinyjobs/exceptions.py | 4 +++ 10 files changed, 114 insertions(+), 19 deletions(-) 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..191e4b3 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -456,6 +456,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://"): 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/tinyjobs/__init__.py b/tinyjobs/__init__.py index d83c51c..b1a36b0 100644 --- a/tinyjobs/__init__.py +++ b/tinyjobs/__init__.py @@ -17,25 +17,46 @@ __version__ = "0.17.0" 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 From 24b563c2a6c6bd61fc44cdc42c030647e4f75e2f Mon Sep 17 00:00:00 2001 From: Piergiuseppe D'Abbraccio Date: Thu, 12 Nov 2026 15:02:54 +0100 Subject: [PATCH 2/6] Let a store without claim_job pass the contract, and make the in-memory template backend pass it: leases, executions, paging, the clock it is given and a lock for the threads that call it --- tests/test_async_backend.py | 459 ++++++++++++++++++++++++----------- tinyjobs/testing/contract.py | 29 ++- 2 files changed, 346 insertions(+), 142 deletions(-) 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/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]) From 46f410ccb7878399a53747cd5adfb609f0326b37 Mon Sep 17 00:00:00 2001 From: Piergiuseppe D'Abbraccio Date: Thu, 12 Nov 2026 15:02:55 +0100 Subject: [PATCH 3/6] Check before every release that a store of the release before is upgraded and then shared by both versions, on SQLite, Postgres and Redis, and publish only from a pushed version tag --- .github/workflows/workflow.yaml | 57 ++++++- tests/test_upgrade.py | 269 ++++++++++++++++++++++++++++++++ tinyjobs/__init__.py | 2 +- 3 files changed, 324 insertions(+), 4 deletions(-) create mode 100644 tests/test_upgrade.py 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/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/tinyjobs/__init__.py b/tinyjobs/__init__.py index b1a36b0..5822ddb 100644 --- a/tinyjobs/__init__.py +++ b/tinyjobs/__init__.py @@ -14,7 +14,7 @@ ) ) -__version__ = "0.17.0" +__version__ = "0.18.0.dev0" from .app import TinyJobs from .clock import Clock, FrozenClock From b66178f9e07d95ec565f03cbba9b3bef3123c5e9 Mon Sep 17 00:00:00 2001 From: Piergiuseppe D'Abbraccio Date: Thu, 12 Nov 2026 15:02:55 +0100 Subject: [PATCH 4/6] Add a changelog for every release and a compatibility page that says what is public, what a release may change and what 1.0 waits for, and bring the pages on the backend protocol, the metrics, the flags, the codec and the early roadmap in line with the code --- .gitignore | 1 + CHANGELOG.md | 407 +++++++++++++++++++++++++++++++++++++++++ README.md | 2 + docs/compatibility.md | 82 +++++++++ docs/design.md | 195 ++++++++++++-------- docs/fastapi.md | 23 +-- docs/guarantees.md | 2 +- docs/limits.md | 2 +- docs/other-backends.md | 57 ++++-- docs/status.md | 3 + docs/tests.md | 16 ++ docs/watching-it.md | 2 + mkdocs.yml | 1 + pyproject.toml | 1 + 14 files changed, 693 insertions(+), 101 deletions(-) create mode 100644 CHANGELOG.md create mode 100644 docs/compatibility.md 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..be830e9 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,407 @@ +# 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. + +### 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" From 3e2caad78da971ac1a3a2717b31fa8437c1aec59 Mon Sep 17 00:00:00 2001 From: Piergiuseppe D'Abbraccio Date: Thu, 12 Nov 2026 16:06:00 +0100 Subject: [PATCH 5/6] Give a free-threaded build time to free an object before three tests ask whether it is gone --- tests/support.py | 8 ++++++++ tests/test_app.py | 4 ++-- tests/test_runtime.py | 7 +++---- 3 files changed, 13 insertions(+), 6 deletions(-) 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_app.py b/tests/test_app.py index 191e4b3..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, @@ -956,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_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(): From f84d27aab4828d7c4273e3d58e5f8b8bc3ba2135 Mon Sep 17 00:00:00 2001 From: Piergiuseppe D'Abbraccio Date: Sat, 14 Nov 2026 15:13:20 +0100 Subject: [PATCH 6/6] Confirm the store is settled with one listing of the queue, so verify no longer calls unfinished the jobs that went back to the queue between two reads by status --- CHANGELOG.md | 5 +++++ tests/test_verify.py | 26 ++++++++++++++++++++++++++ tinyjobs/verifying.py | 14 +++++++++++++- 3 files changed, 44 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index be830e9..d7ad38c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,11 @@ versions are the `SCHEMA_VERSION` of each backend. - `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 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/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)