Coverage for app/backend/src/couchers/jobs/worker.py: 73%
114 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +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((now() - job.queued).total_seconds())
80 try:
81 with tracer.start_as_current_span(job.job_type) as rollspan:
82 start = perf_counter_ns()
83 job_def.handler(job_def.payload_type.FromString(job.payload))
84 finished = perf_counter_ns()
85 job.state = BackgroundJobState.completed
86 observe_in_jobs_duration_histogram(
87 job.job_type, job.state.name, job.try_count, "", (finished - start) / 1e9
88 )
89 logger.info(f"Job #{job.id} complete on try number {job.try_count}")
90 except Exception as e:
91 finished = perf_counter_ns()
92 # not sentry_sdk.set_tag: that writes to the thread's isolation scope, where the tags stick to
93 # every later report from this thread. logger.exception is in here so its event is tagged too
94 with sentry_sdk.new_scope() as scope:
95 scope.set_tag("context", "job")
96 scope.set_tag("job", job.job_type)
97 logger.exception(e)
98 sentry_sdk.capture_exception(e)
100 if job.try_count >= job.max_tries:
101 # if we already tried max_tries times, it's permanently failed
102 job.state = BackgroundJobState.failed
103 logger.info(f"Job #{job.id} failed on try number {job.try_count}")
104 else:
105 job.state = BackgroundJobState.error
106 # exponential backoff
107 job.next_attempt_after = now() + timedelta(seconds=15 * (2**job.try_count))
108 logger.info(f"Job #{job.id} error on try number {job.try_count}, next try at {job.next_attempt_after}")
109 observe_in_jobs_duration_histogram(
110 job.job_type, job.state.name, job.try_count, type(e).__name__, (finished - start) / 1e9
111 )
112 # add some info for debugging
113 job.failure_info = traceback.format_exc()
115 if config.IN_TEST:
116 raise e
118 # exiting ctx manager commits and releases the row lock
119 return True
122def service_jobs() -> None:
123 """
124 Service jobs in an infinite loop
125 """
126 while True:
127 # if no job was found, sleep for a second, otherwise query for another job straight away
128 if not process_job():
129 sleep(1)
132def _run_job_and_schedule(sched: scheduler, job_def: Job[Any], frequency: timedelta) -> None:
133 logger.info(f"Processing job of type {job_def.name}")
135 # wake ourselves up after frequency
136 sched.enter(
137 delay=frequency.total_seconds(),
138 priority=1,
139 action=_run_job_and_schedule,
140 argument=(
141 sched,
142 job_def,
143 frequency,
144 ),
145 )
147 # queue the job
148 with session_scope() as session:
149 queue_job(session, job=job_def.handler, payload=empty_pb2.Empty())
152def run_scheduler() -> None:
153 """
154 Schedules jobs according to schedule in JOBS
155 """
156 sched = scheduler(monotonic, sleep)
158 for job_type, job_def in JOBS.items():
159 if job_def.schedule is not None: 159 ↛ 158line 159 didn't jump to line 158 because the condition on line 159 was always true
160 sched.enter(
161 delay=0,
162 priority=1,
163 action=_run_job_and_schedule,
164 argument=(
165 sched,
166 job_def,
167 job_def.schedule,
168 ),
169 )
171 sched.run()
174def _per_process_init(profile_instance: str | None) -> None:
175 # Post-fork initialization: these services use threading/async internals that
176 # don't survive fork() and must be initialized fresh in each child process.
177 # Pyroscope in particular can only be initialized once per process.
178 db_post_fork()
179 setup_tracing()
180 setup_experimentation()
181 if profile_instance is not None:
182 setup_profiling(role="worker", instance=profile_instance)
185def _run_forever(func: Callable[[], None]) -> None:
186 while True:
187 try:
188 logger.info("Background worker starting")
189 func()
190 except Exception as e:
191 logger.critical("Unhandled exception in background worker", exc_info=e)
192 # cool off in case we have some programming error to not hammer the database
193 sleep(60)
196def _scheduler_process_entry() -> None:
197 _per_process_init(None)
198 _run_forever(run_scheduler)
201def _worker_process_entry(profile_instance: str, threads_per_process: int) -> None:
202 _per_process_init(profile_instance)
203 # the lru_cache doesn't hold a lock across the load, so otherwise every thread parses all the locales
204 get_main_i18next()
205 # threads rather than processes: the handlers are I/O-bound, and a process costs ~200 MB
206 threads = [
207 threading.Thread(target=_run_forever, args=(service_jobs,), name=f"jobs-thread-{i}", daemon=True)
208 for i in range(threads_per_process)
209 ]
210 for t in threads:
211 t.start()
212 # the supervisor only watches processes, so a thread dying here would silently cut our capacity: exit
213 # instead and let it restart us
214 while all(t.is_alive() for t in threads):
215 sleep(1)
216 logger.critical("A jobs thread died, exiting so the supervisor restarts us")
219def start_jobs_scheduler() -> Process:
220 scheduler = Process(target=_scheduler_process_entry)
221 scheduler.start()
222 return scheduler
225def start_jobs_worker(index: int, threads_per_process: int) -> Process:
226 worker = Process(
227 target=_worker_process_entry,
228 args=(f"worker-{index}", threads_per_process),
229 )
230 worker.start()
231 return worker