Recover jobs stuck in Cancelling when no worker acknowledges the cancel

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()

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'))

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']