commit 86a47e36edae3bc773cfc7468c3ab9836dcc6721 equwal <truex@equwal.com> 2026-09-22 16:45:08 -0700 Put paid and free server jobs on separate queues A burst worker must take paid jobs only, so that a free credit never starts a machine that costs money. With the Redis queue, a paid job now goes to subplz-jobs-paid and a free job to subplz-jobs. billing.is_paid tells them apart: a bought credit, a book of a monthly plan and an unlimited plan pay; a free credit, the daily credit and a browser job do not. worker.py takes the queues in SUBPLZ_WEB_WORKER_QUEUES, paid jobs first by default. With Redis, a restart of the web server no longer puts running jobs back on the queue. A worker on another machine can still run them, and a second copy would convert the book twice. The in-process queue keeps the old behaviour, because its jobs stop with the process. The dev requirements add rq, redis and fakeredis for the queue tests, and hypothesis for the property tests of the next commits.
backend/api.py | 2 +- backend/billing.py | 7 +++ backend/main.py | 13 +++-- backend/queue.py | 22 ++++++--- requirements-dev.txt | 7 +++ tests/conftest.py | 2 +- tests/test_local_jobs.py | 2 +- tests/test_paid_queue.py | 123 +++++++++++++++++++++++++++++++++++++++++++++++ worker.py | 27 +++++++++-- 9 files changed, 186 insertions(+), 19 deletions(-)
diff --git a/backend/api.py b/backend/api.py index 08c31e0..5211e66 100644 --- a/backend/api.py +++ b/backend/api.py @@ -675,7 +675,7 @@ def start_job( job.error = None session.commit() - queue.enqueue(job.id) + queue.enqueue(job.id, paid=billing.is_paid(job)) return _job_out(job, []) diff --git a/backend/billing.py b/backend/billing.py index e3b7bab..48fab13 100644 --- a/backend/billing.py +++ b/backend/billing.py @@ -290,3 +290,10 @@ def refund_job(job_id: str) -> None: if job is not None: refund(session, job) session.commit() + + +def is_paid(job: Job) -> bool: + """True for a server job that money pays for: a bought credit, a book of a + monthly plan, or an unlimited plan. False for a free credit, the daily + credit, and a browser job. Only paid jobs may start a burst worker.""" + return not (job.local or job.free_credit_spent or job.daily_credit_on is not None) diff --git a/backend/main.py b/backend/main.py index ffe9c0e..11d98c7 100644 --- a/backend/main.py +++ b/backend/main.py @@ -19,7 +19,7 @@ from fastapi import FastAPI from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles -from . import captions, subrep +from . import billing, captions, subrep from .api import router from .db import Job, JobStatus, SessionLocal, init_db from .languages import all_languages @@ -61,7 +61,12 @@ def _requeue_interrupted() -> None: An in-process queue dies with the process, so anything left `running` is orphaned. Put it back in the queue rather than leaving a stuck progress bar. + + A Redis queue keeps its jobs, and a worker on another machine may still run + one. Putting that job on the queue again would convert the book twice. """ + if settings.queue_backend != "memory": + return with SessionLocal() as s: stale = ( s.query(Job) @@ -73,10 +78,10 @@ def _requeue_interrupted() -> None: job.stage = "Queued (resumed after restart)" job.progress = 0.0 s.commit() - ids = [j.id for j in stale] + ids = [(j.id, billing.is_paid(j)) for j in stale] - for job_id in ids: - queue.enqueue(job_id) + for job_id, paid in ids: + queue.enqueue(job_id, paid=paid) if ids: log.info("re-queued %d interrupted job(s)", len(ids)) diff --git a/backend/queue.py b/backend/queue.py index dc0bf82..2283d12 100644 --- a/backend/queue.py +++ b/backend/queue.py @@ -3,6 +3,10 @@ Localhost runs jobs on a small thread pool inside the API process. The public deployment sets SUBPLZ_WEB_QUEUE_BACKEND=redis and runs `worker.py` on separate machines; the API side then only enqueues. Same `enqueue(job_id)` call either way. + +A paid job and a free job go to different Redis queues. The web server's own +worker takes both, paid jobs first. A burst worker takes paid jobs only (see +infra/burst/README.md), so a free credit never starts a machine that costs money. """ from __future__ import annotations @@ -16,11 +20,14 @@ from .settings import settings log = logging.getLogger(__name__) +PAID_QUEUE = "subplz-jobs-paid" +FREE_QUEUE = "subplz-jobs" + class JobQueue(ABC): @abstractmethod - def enqueue(self, job_id: str) -> None: - ... + def enqueue(self, job_id: str, paid: bool = True) -> None: + """Run `job_id`. `paid` is False for a job that a free credit pays for.""" @abstractmethod def depth(self) -> int: @@ -44,7 +51,7 @@ class InProcessQueue(JobQueue): self._lock = threading.Lock() self._pending: set[str] = set() - def enqueue(self, job_id: str) -> None: + def enqueue(self, job_id: str, paid: bool = True) -> None: with self._lock: if job_id in self._pending: return @@ -79,10 +86,11 @@ class RedisQueue(JobQueue): from rq import Queue as RQQueue self._conn = Redis.from_url(url) - self._q = RQQueue("subplz-jobs", connection=self._conn) + self._paid = RQQueue(PAID_QUEUE, connection=self._conn) + self._free = RQQueue(FREE_QUEUE, connection=self._conn) - def enqueue(self, job_id: str) -> None: - self._q.enqueue( + def enqueue(self, job_id: str, paid: bool = True) -> None: + (self._paid if paid else self._free).enqueue( "backend.runner.run_job", job_id, job_timeout=settings.job_timeout_seconds, @@ -90,7 +98,7 @@ class RedisQueue(JobQueue): ) def depth(self) -> int: - return self._q.count + self._q.started_job_registry.count + return sum(q.count + q.started_job_registry.count for q in (self._paid, self._free)) def build_queue() -> JobQueue: diff --git a/requirements-dev.txt b/requirements-dev.txt index e3fcd49..66de55e 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,3 +1,10 @@ -r requirements.txt pytest>=8 httpx>=0.27 + +# Property tests (tests/test_burst_controller.py, tests/test_copy_database.py). +hypothesis>=6.100 +# The burst controller reads the rq queue. Its tests use a fake Redis. +rq>=2.0 +redis>=5.0 +fakeredis>=2.20 diff --git a/tests/conftest.py b/tests/conftest.py index 8ed33c7..9b362ea 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -43,7 +43,7 @@ init_db() @pytest.fixture(autouse=True) def _no_real_work(monkeypatch): """Starting a job must not actually launch an alignment.""" - monkeypatch.setattr(api.queue, "enqueue", lambda job_id: None) + monkeypatch.setattr(api.queue, "enqueue", lambda job_id, paid=True: None) @pytest.fixture(autouse=True) diff --git a/tests/test_local_jobs.py b/tests/test_local_jobs.py index 6ca0daf..4cffb6a 100644 --- a/tests/test_local_jobs.py +++ b/tests/test_local_jobs.py @@ -101,7 +101,7 @@ def test_restart_does_not_queue_browser_jobs(client, monkeypatch): job = begin(client).json() queued = [] - monkeypatch.setattr(main.queue, "enqueue", queued.append) + monkeypatch.setattr(main.queue, "enqueue", lambda job_id, paid=True: queued.append(job_id)) main._requeue_interrupted() assert job["id"] not in queued assert get_job_row(job["id"]).status == JobStatus.running diff --git a/tests/test_paid_queue.py b/tests/test_paid_queue.py new file mode 100644 index 0000000..e7ca08a --- /dev/null +++ b/tests/test_paid_queue.py @@ -0,0 +1,123 @@ +"""Burst workers take paid jobs only. + +Each server job goes to the paid queue or to the free queue. The web server's +own worker takes both, paid jobs first. A burst worker takes paid jobs only, so +free credits never start a machine that costs money. +""" + +from __future__ import annotations + +import time +from datetime import date + +import fakeredis +import pytest +from rq import Queue + +import worker +from backend import api, billing, main +from backend import queue as jobqueue +from backend.db import Job, JobStatus, SessionLocal +from backend.settings import settings + +from .conftest import account_id, checkout_event, make_job, post_webhook + + +@pytest.fixture +def cloud(monkeypatch): + monkeypatch.setattr(settings, "cloud_enabled", True) + + +@pytest.fixture +def sent(monkeypatch): + """What the API put on the queue: (job id, paid).""" + calls: list[tuple[str, bool]] = [] + monkeypatch.setattr(api.queue, "enqueue", lambda job_id, paid=True: calls.append((job_id, paid))) + return calls + + +def start(client, job_id: str): + return client.post(f"/api/jobs/{job_id}/start", json={"language": "en"}) + + +@pytest.mark.parametrize("fields, paid", [ + ({"credit_spent": 1}, True), # a bought credit + ({"plan_credit_spent": 1}, True), # a book of a monthly plan + ({}, True), # an unlimited plan spends no credit + ({"free_credit_spent": 1}, False), + ({"daily_credit_on": date(2026, 9, 22)}, False), + ({"local": 1}, False), # a browser job never goes to a worker +]) +def test_is_paid(fields, paid): + job = Job(**{"local": 0, "credit_spent": 0, "plan_credit_spent": 0, + "free_credit_spent": 0, "daily_credit_on": None, **fields}) + assert billing.is_paid(job) is paid + + +def test_a_free_credit_job_goes_to_the_free_queue(client, cloud, sent): + job_id = make_job(client, with_files=True) + assert start(client, job_id).status_code == 200 + assert sent == [(job_id, False)] + + +def test_a_bought_credit_job_goes_to_the_paid_queue(client, cloud, sent): + assert start(client, make_job(client, with_files=True)).status_code == 200 # the free one + event = checkout_event(f"cs_{time.time_ns()}", account_id(client), "pack10", + email=f"paid-{time.time_ns()}@example.com") + assert post_webhook(client, event).status_code == 200 + job_id = make_job(client, with_files=True) + assert start(client, job_id).status_code == 200 + assert sent[-1] == (job_id, True) + + +def test_the_redis_queue_keeps_paid_and_free_jobs_apart(monkeypatch): + conn = fakeredis.FakeRedis() + monkeypatch.setattr("redis.Redis.from_url", lambda url: conn) + q = jobqueue.RedisQueue("redis://unused") + q.enqueue("job_paid", paid=True) + q.enqueue("job_free", paid=False) + assert [j.args for j in Queue(jobqueue.PAID_QUEUE, connection=conn).jobs] == [("job_paid",)] + assert [j.args for j in Queue(jobqueue.FREE_QUEUE, connection=conn).jobs] == [("job_free",)] + assert q.depth() == 2 + + +def test_the_web_server_worker_takes_paid_jobs_first(monkeypatch): + monkeypatch.delenv("SUBPLZ_WEB_WORKER_QUEUES", raising=False) + assert worker.queue_names() == [jobqueue.PAID_QUEUE, jobqueue.FREE_QUEUE] + + +def test_a_burst_worker_takes_paid_jobs_only(monkeypatch): + monkeypatch.setenv("SUBPLZ_WEB_WORKER_QUEUES", "paid") + assert worker.queue_names() == [jobqueue.PAID_QUEUE] + + +def test_an_unknown_queue_is_refused(monkeypatch): + monkeypatch.setenv("SUBPLZ_WEB_WORKER_QUEUES", "paid,vip") + with pytest.raises(ValueError, match="vip"): + worker.queue_names() + + +def _running_job(client) -> str: + job_id = make_job(client, JobStatus.running) + with SessionLocal() as s: + s.get(Job, job_id).credit_spent = 1 + s.commit() + return job_id + + +def test_a_restart_with_the_redis_queue_leaves_running_jobs_alone(client, sent, monkeypatch): + # With Redis, a worker on another machine still runs the job. Putting it on + # the queue again would convert the book twice. + job_id = _running_job(client) + monkeypatch.setattr(settings, "queue_backend", "redis") + main._requeue_interrupted() + assert sent == [] + with SessionLocal() as s: + assert s.get(Job, job_id).status == JobStatus.running + + +def test_a_restart_with_the_memory_queue_runs_interrupted_jobs_again(client, sent, monkeypatch): + job_id = _running_job(client) + monkeypatch.setattr(settings, "queue_backend", "memory") + main._requeue_interrupted() + assert (job_id, True) in sent diff --git a/worker.py b/worker.py index 6a1e0fd..3103f15 100644 --- a/worker.py +++ b/worker.py @@ -16,11 +16,15 @@ Workers and the API must share the database and the storage bucket. They do not need to share a filesystem: staged uploads are the one thing that is local to whoever received them, which is why the API writes uploads to shared storage before enqueuing when the queue backend is redis. + +SUBPLZ_WEB_WORKER_QUEUES says which jobs this worker takes, in order: "paid,free" +(the default: paid jobs first) or "paid" (a burst worker, see infra/burst). """ from __future__ import annotations import logging +import os import sys from backend.settings import settings @@ -32,6 +36,18 @@ logging.basicConfig( log = logging.getLogger("subplz.worker") +def queue_names() -> list[str]: + """The Redis queues of this worker, the first one first.""" + from backend.queue import FREE_QUEUE, PAID_QUEUE + + known = {"paid": PAID_QUEUE, "free": FREE_QUEUE} + wanted = [n.strip() for n in os.environ.get("SUBPLZ_WEB_WORKER_QUEUES", "paid,free").split(",")] + unknown = [n for n in wanted if n not in known] + if unknown or not wanted: + raise ValueError(f"SUBPLZ_WEB_WORKER_QUEUES: unknown queue {unknown}; use paid and free") + return [known[n] for n in wanted] + + def main() -> int: if settings.queue_backend != "redis": log.error( @@ -49,15 +65,16 @@ def main() -> int: return 2 conn = Redis.from_url(settings.redis_url) - queue = Queue("subplz-jobs", connection=conn) + queues = [Queue(name, connection=conn) for name in queue_names()] log.info( - "worker starting | redis=%s | device=%s model=%s | storage=%s", - settings.redis_url, settings.device, settings.model, + "worker starting | queues=%s | device=%s model=%s | storage=%s", + ",".join(q.name for q in queues), settings.device, settings.model, settings.storage_backend, ) - # burst=False: stay alive and keep taking jobs. - Worker([queue], connection=conn).work(with_scheduler=False) + # burst=False: stay alive and keep taking jobs. rq takes a job from the + # first queue that has one, so the order of the queues is their priority. + Worker(queues, connection=conn).work(with_scheduler=False) return 0