mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +00:00
fix(worker): a database error skips a tick instead of ending the worker
Any error in the manager loop ended the worker at once with os._exit, without the drain: a PostgreSQL restart or a SQLite write lock held past its timeout cut every running migration, block copy and backup the worker drove, and their jobs went to recovery, some ending uncertain. A database error now skips that tick and drops the broken connection; the next tick tries again. A lease lost meanwhile still ends the worker when renew sees it, and other errors still end it as before.
This commit is contained in:
parent
1d93b5c0bf
commit
4c0efe9db0
2 changed files with 32 additions and 4 deletions
|
|
@ -256,12 +256,33 @@ class WorkerTestCase(TransactionTestCase):
|
|||
self.assertIn("no context", words(Operation.objects.get(pk=op.pk).error))
|
||||
self.assertFalse(Claim.objects.filter(operation=op).exists())
|
||||
|
||||
def test_a_database_error_skips_a_tick(self):
|
||||
directory = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(directory.cleanup)
|
||||
self.worker = Worker(data=directory.name, tick=0)
|
||||
errors = [OperationalError("database is locked"), OperationalError("server closed the connection")]
|
||||
|
||||
def tick():
|
||||
if errors:
|
||||
raise errors.pop()
|
||||
self.worker.stopping.set()
|
||||
with (
|
||||
patch.object(self.worker, "tick_once", side_effect=tick),
|
||||
patch("operations.worker.signal.signal"),
|
||||
patch("operations.worker.logging.shutdown"),
|
||||
patch("operations.worker.os._exit", side_effect=SystemExit) as exit_,
|
||||
self.assertLogs("operations.worker", "ERROR"),
|
||||
):
|
||||
with self.assertRaises(SystemExit):
|
||||
self.worker.run()
|
||||
exit_.assert_called_once_with(0)
|
||||
|
||||
def test_a_failing_manager_loop_ends_the_process(self):
|
||||
directory = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(directory.cleanup)
|
||||
self.worker = Worker(data=directory.name)
|
||||
with (
|
||||
patch.object(self.worker, "tick_once", side_effect=OperationalError("database is down")),
|
||||
patch.object(self.worker, "tick_once", side_effect=KeyError("a bug")),
|
||||
patch("operations.worker.signal.signal"),
|
||||
patch("operations.worker.logging.shutdown"),
|
||||
patch("operations.worker.os._exit", side_effect=SystemExit) as exit_,
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ from importlib import import_module
|
|||
from pathlib import Path
|
||||
|
||||
from django.conf import settings
|
||||
from django.db import close_old_connections
|
||||
from django.db import DatabaseError, close_old_connections
|
||||
from django.utils import timezone
|
||||
from django.utils.translation import gettext_noop
|
||||
|
||||
|
|
@ -116,7 +116,8 @@ class Worker:
|
|||
def run(self):
|
||||
"""Until SIGTERM. Returns 1 if another worker runs on this host. Ends
|
||||
with os._exit, so job threads past the grace period end with it; a
|
||||
failing manager loop ends the process the same way (status 1)."""
|
||||
manager loop failing other than on the database ends the process the
|
||||
same way (status 1)."""
|
||||
lock = open(self.lock_path, "a")
|
||||
try:
|
||||
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
|
|
@ -133,7 +134,13 @@ class Worker:
|
|||
logger.info("Worker %s started on %s", self.identity, socket.gethostname())
|
||||
self.recovery.schedule_orphans()
|
||||
while not self.stopping.is_set():
|
||||
self.tick_once()
|
||||
try:
|
||||
self.tick_once()
|
||||
except DatabaseError as err:
|
||||
# a database restart or a long lock ends no job: the next tick tries
|
||||
# again, and a lease lost meanwhile still ends the worker (renew)
|
||||
logger.error("Worker %s tick failed: %s", self.identity, utils.safe_text(err))
|
||||
close_old_connections()
|
||||
self.stopping.wait(self.tick)
|
||||
self._drain()
|
||||
except Exception as err:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue