diff --git a/server/dive_server/__init__.py b/server/dive_server/__init__.py index 22af9f2a8..b69acde3f 100644 --- a/server/dive_server/__init__.py +++ b/server/dive_server/__init__.py @@ -2,6 +2,8 @@ import os from pathlib import Path +import cherrypy +from cherrypy.process.plugins import Monitor from girder import events, plugin from girder.constants import AccessType from girder.models.user import User @@ -14,6 +16,7 @@ from .crud_annotation import GroupItem, RevisionLogItem, TrackItem from .event import send_new_user_email +from .stale_cancellation import SWEEP_INTERVAL_SECONDS, reap_stale_canceling_jobs from .views_annotation import AnnotationResource from .views_configuration import ConfigurationResource from .views_dataset import DatasetResource @@ -73,3 +76,12 @@ def load(self, info): 'send_new_user_email', send_new_user_email, ) + + # Jobs canceled while no worker is listening otherwise stay in + # "Cancelling" forever; see stale_cancellation.py. + Monitor( + cherrypy.engine, + reap_stale_canceling_jobs, + frequency=SWEEP_INTERVAL_SECONDS, + name='DIVE stale cancellation sweep', + ).subscribe() diff --git a/server/dive_server/stale_cancellation.py b/server/dive_server/stale_cancellation.py new file mode 100644 index 000000000..2b7a09891 --- /dev/null +++ b/server/dive_server/stale_cancellation.py @@ -0,0 +1,83 @@ +"""Finalize jobs whose cancellation was never acknowledged by a worker. + +Canceling a job moves it to CANCELING (824) and broadcasts a Celery revoke +(girder_plugin_worker.event_handlers.cancel). The job only reaches CANCELED +once a live worker acknowledges: either the running task notices the revoke +and stops, or Celery discards the revoked message when it is delivered. If no +worker is connected to the job's queue (a private standalone queue whose +worker is offline, or a worker that crashed or was redeployed), or a worker +restart wiped its in-memory revoked-task list, nothing ever acknowledges and +the job shows "Cancelling" indefinitely. + +The Girder server periodically sweeps for jobs that have sat in CANCELING +with no update for STALE_CANCELING_TIMEOUT and moves them to CANCELED. The +revoke is also re-broadcast so a worker that reconnects later discards the +still-queued task message instead of running it. +""" + +from datetime import datetime, timedelta, timezone +import logging + +from celery.result import AsyncResult +from girder_jobs.constants import JobStatus +from girder_jobs.models.job import Job +from girder_plugin_worker.status import CustomJobStatus +from girder_worker.app import app + +logger = logging.getLogger(__name__) + +# A live worker acknowledges a cancel within about a minute (the subprocess +# monitor polls every 30 seconds), and any job update (including log writes) +# refreshes `updated`, keeping actively-handled jobs out of the sweep. A job +# idle in CANCELING past this timeout has no worker acting on it. Finalizing +# early is safe even if a worker is in a long non-polling phase (media +# download/upload): its own later CANCELED acknowledgment becomes a no-op. +STALE_CANCELING_TIMEOUT = timedelta(seconds=60) +SWEEP_INTERVAL_SECONDS = 30 + + +def stale_canceling_query(now: datetime) -> dict: + return { + 'status': CustomJobStatus.CANCELING, + 'updated': {'$lt': now - STALE_CANCELING_TIMEOUT}, + } + + +def rebroadcast_revoke(celery_task_id: str): + """Re-revoke so a worker that reconnects discards the queued task message.""" + AsyncResult(celery_task_id, app=app).revoke() + + +def reap_stale_canceling_jobs(job_model=None): + # Never raise: cherrypy's Monitor permanently stops its background thread + # on the first uncaught exception. + try: + _reap_stale_canceling_jobs(job_model) + except Exception: + logger.exception('Stale cancellation sweep failed') + + +def _reap_stale_canceling_jobs(job_model=None): + job_model = job_model if job_model is not None else Job() + now = datetime.now(timezone.utc) + for job in job_model.find(stale_canceling_query(now)): + try: + logger.info( + 'Job %s stuck in CANCELING since %s; marking CANCELED', + job['_id'], + job.get('updated'), + ) + celery_task_id = job.get('celeryTaskId') + if celery_task_id: + try: + rebroadcast_revoke(celery_task_id) + except Exception: + logger.exception('Could not re-revoke Celery task %s', celery_task_id) + job_model.updateJob( + job, + log='Cancellation was not acknowledged by any worker; ' + 'the job has been marked as canceled.\n', + status=JobStatus.CANCELED, + ) + except Exception: + logger.exception('Failed to finalize canceled job %s', job.get('_id')) diff --git a/server/tests/test_stale_cancellation.py b/server/tests/test_stale_cancellation.py new file mode 100644 index 000000000..b565b0d19 --- /dev/null +++ b/server/tests/test_stale_cancellation.py @@ -0,0 +1,68 @@ +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock + +from girder_jobs.constants import JobStatus +from girder_plugin_worker.status import CustomJobStatus + +from dive_server import stale_cancellation +from dive_server.stale_cancellation import ( + STALE_CANCELING_TIMEOUT, + reap_stale_canceling_jobs, + stale_canceling_query, +) + + +def test_query_matches_only_stale_canceling_jobs(): + now = datetime.now(timezone.utc) + query = stale_canceling_query(now) + assert query['status'] == CustomJobStatus.CANCELING + cutoff = query['updated']['$lt'] + assert cutoff == now - STALE_CANCELING_TIMEOUT + # A job updated after the cutoff (still being acknowledged) is untouched + assert not (now - timedelta(seconds=30)) < cutoff + + +def test_reap_moves_stale_jobs_to_canceled(): + job = {'_id': 'job1', 'status': CustomJobStatus.CANCELING, + 'updated': datetime.now(timezone.utc) - timedelta(days=2)} + job_model = MagicMock() + job_model.find.return_value = [job] + + reap_stale_canceling_jobs(job_model=job_model) + + job_model.updateJob.assert_called_once() + _, kwargs = job_model.updateJob.call_args + assert kwargs['status'] == JobStatus.CANCELED + + +def test_reap_rebroadcasts_revoke_for_queued_task(monkeypatch): + revoked = [] + monkeypatch.setattr(stale_cancellation, 'rebroadcast_revoke', revoked.append) + job = {'_id': 'job1', 'celeryTaskId': 'task-abc', 'status': CustomJobStatus.CANCELING, + 'updated': datetime.now(timezone.utc) - timedelta(days=2)} + job_model = MagicMock() + job_model.find.return_value = [job] + + reap_stale_canceling_jobs(job_model=job_model) + + assert revoked == ['task-abc'] + job_model.updateJob.assert_called_once() + + +def test_reap_continues_after_a_failing_job(): + good_job = {'_id': 'job2', 'status': CustomJobStatus.CANCELING, + 'updated': datetime.now(timezone.utc) - timedelta(days=2)} + job_model = MagicMock() + job_model.find.return_value = [{'_id': 'job1'}, good_job] + calls = [] + + def update_job(job, **kwargs): + calls.append(job['_id']) + if job['_id'] == 'job1': + raise RuntimeError('validation failed') + + job_model.updateJob.side_effect = update_job + + reap_stale_canceling_jobs(job_model=job_model) + + assert calls == ['job1', 'job2']