Coverage for app/backend/src/couchers/jobs/worker.py: 73%
114 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1"""
2Background job workers
3"""
5import logging
6import threading
7import traceback
8from collections.abc import Callable
9from datetime import timedelta
10from multiprocessing import Process
11from sched import scheduler
12from time import monotonic, perf_counter_ns, sleep
13from typing import Any
15import sentry_sdk
16from google.protobuf import empty_pb2
17from opentelemetry import trace
18from sqlalchemy import select
20from couchers.config import config
21from couchers.db import db_post_fork, session_scope
22from couchers.experimentation import setup_experimentation
23from couchers.i18n.locales import get_main_i18next
24from couchers.jobs.definitions import JOBS, Job
25from couchers.jobs.enqueue import queue_job
26from couchers.metrics import (
27 background_jobs_got_job_counter,
28 background_jobs_no_jobs_counter,
29 jobs_queued_histogram,
30 observe_in_jobs_duration_histogram,
31)
32from couchers.models import BackgroundJob, BackgroundJobState
33from couchers.profiling import setup_profiling
34from couchers.tracing import setup_tracing
35from couchers.utils import now
37logger = logging.getLogger(__name__)
38tracer = trace.get_tracer(__name__)
41def process_job() -> bool:
42 """
43 Attempt to process one job from the job queue. Returns False if no job was found, True if a job was processed,
44 regardless of failure/success.
45 """
46 logger.debug("Looking for a job")
48 with session_scope() as session:
49 # SELECT ... FOR UPDATE is what makes sure only one worker handles a given job: no two transactions can hold
50 # the row lock at once. SKIP LOCKED means an already-claimed job is passed over rather than waited on, so
51 # workers spread out across the queue. This must run at READ COMMITTED (the default): a stricter isolation
52 # level can't follow the update chain of a row another worker just committed, and aborts the whole dequeue
53 # with a serialization error instead of skipping the row
54 job = (
55 session.execute(
56 select(BackgroundJob)
57 .where(BackgroundJob.ready_for_retry)
58 .order_by(BackgroundJob.priority.desc(), BackgroundJob.next_attempt_after.asc())
59 .limit(1)
60 .with_for_update(skip_locked=True)
61 )
62 .scalars()
63 .one_or_none()
64 )
66 if not job:
67 background_jobs_no_jobs_counter.inc()
68 logger.debug("No pending jobs")
69 return False
71 background_jobs_got_job_counter.inc()
73 # we've got a lock for a job now, it's "pending" until we commit or the lock is gone
74 logger.info(f"Job #{job.id} of type {job.job_type} grabbed")
75 job.try_count += 1
77 job_def = JOBS[job.job_type]
79 jobs_queued_histogram.labels(str(job.priority)).observe(
80 max((now() - job.next_attempt_after).total_seconds(), 0.0)
81 )
82 try:
83 with tracer.start_as_current_span(job.job_type) as rollspan:
84 start = perf_counter_ns()
85 job_def.handler(job_def.payload_type.FromString(job.payload))
86 finished = perf_counter_ns()
87 job.state = BackgroundJobState.completed
88 observe_in_jobs_duration_histogram(
89 job.job_type, job.state.name, job.try_count, "", (finished - start) / 1e9
90 )
91 logger.info(f"Job #{job.id} complete on try number {job.try_count}")
92 except Exception as e:
93 finished = perf_counter_ns()
94 logger.exception(e)
96 if job.try_count >= job.max_tries:
97 # if we already tried max_tries times, it's permanently failed
98 job.state = BackgroundJobState.failed
99 logger.info(f"Job #{job.id} failed on try number {job.try_count}")
100 # a new scope keeps these tags on this report only, not on every later one from this thread
101 with sentry_sdk.new_scope() as scope:
102 scope.set_tag("context", "job")
103 scope.set_tag("job", job.job_type)
104 sentry_sdk.capture_exception(e)
105 else:
106 job.state = BackgroundJobState.error
107 # exponential backoff
108 job.next_attempt_after = now() + timedelta(seconds=15 * (2**job.try_count))
109 logger.info(f"Job #{job.id} error on try number {job.try_count}, next try at {job.next_attempt_after}")
110 observe_in_jobs_duration_histogram(
111 job.job_type, job.state.name, job.try_count, type(e).__name__, (finished - start) / 1e9
112 )
113 # add some info for debugging
114 job.failure_info = traceback.format_exc()
116 if config.IN_TEST:
117 raise e
119 # exiting ctx manager commits and releases the row lock
120 return True
123def service_jobs() -> None:
124 """
125 Service jobs in an infinite loop
126 """
127 while True:
128 # if no job was found, sleep for a second, otherwise query for another job straight away
129 if not process_job():
130 sleep(1)
133def _run_job_and_schedule(sched: scheduler, job_def: Job[Any], frequency: timedelta) -> None:
134 logger.info(f"Processing job of type {job_def.name}")
136 # wake ourselves up after frequency
137 sched.enter(
138 delay=frequency.total_seconds(),
139 priority=1,
140 action=_run_job_and_schedule,
141 argument=(
142 sched,
143 job_def,
144 frequency,
145 ),
146 )
148 # queue the job
149 with session_scope() as session:
150 queue_job(session, job=job_def.handler, payload=empty_pb2.Empty())
153def run_scheduler() -> None:
154 """
155 Schedules jobs according to schedule in JOBS
156 """
157 sched = scheduler(monotonic, sleep)
159 for job_type, job_def in JOBS.items():
160 if job_def.schedule is not None: 160 ↛ 159line 160 didn't jump to line 159 because the condition on line 160 was always true
161 sched.enter(
162 delay=0,
163 priority=1,
164 action=_run_job_and_schedule,
165 argument=(
166 sched,
167 job_def,
168 job_def.schedule,
169 ),
170 )
172 sched.run()
175def _per_process_init(profile_instance: str | None) -> None:
176 # Post-fork initialization: these services use threading/async internals that
177 # don't survive fork() and must be initialized fresh in each child process.
178 # Pyroscope in particular can only be initialized once per process.
179 db_post_fork()
180 setup_tracing()
181 setup_experimentation()
182 if profile_instance is not None:
183 setup_profiling(role="worker", instance=profile_instance)
186def _run_forever(func: Callable[[], None]) -> None:
187 while True:
188 try:
189 logger.info("Background worker starting")
190 func()
191 except Exception as e:
192 logger.critical("Unhandled exception in background worker", exc_info=e)
193 # cool off in case we have some programming error to not hammer the database
194 sleep(60)
197def _scheduler_process_entry() -> None:
198 _per_process_init(None)
199 _run_forever(run_scheduler)
202def _worker_process_entry(profile_instance: str, threads_per_process: int) -> None:
203 _per_process_init(profile_instance)
204 # the lru_cache doesn't hold a lock across the load, so otherwise every thread parses all the locales
205 get_main_i18next()
206 # threads rather than processes: the handlers are I/O-bound, and a process costs ~200 MB
207 threads = [
208 threading.Thread(target=_run_forever, args=(service_jobs,), name=f"jobs-thread-{i}", daemon=True)
209 for i in range(threads_per_process)
210 ]
211 for t in threads:
212 t.start()
213 # the supervisor only watches processes, so a thread dying here would silently cut our capacity: exit
214 # instead and let it restart us
215 while all(t.is_alive() for t in threads):
216 sleep(1)
217 logger.critical("A jobs thread died, exiting so the supervisor restarts us")
220def start_jobs_scheduler() -> Process:
221 scheduler = Process(target=_scheduler_process_entry)
222 scheduler.start()
223 return scheduler
226def start_jobs_worker(index: int, threads_per_process: int) -> Process:
227 worker = Process(
228 target=_worker_process_entry,
229 args=(f"worker-{index}", threads_per_process),
230 )
231 worker.start()
232 return worker