From f2138e54d80ab69ad9b4bd92740ef64d44a23550 Mon Sep 17 00:00:00 2001 From: catborise Date: Fri, 16 Oct 2026 08:52:34 +0300 Subject: [PATCH] feat(operations): jobs a request runs Request jobs (an upload's chunks) run in gunicorn: their kind is admitted running with the request as its worker, never claimed or recovered by the worker, cancelled in the request, and asked by every housekeeping pass to end its jobs that went idle. The panel can follow a job the page just started. A job a request runs does not wait for the worker, so it does not raise the worker-unavailable warning on the Jobs page or in the panel. --- operations/api/serializers.py | 7 ++- operations/jobs.py | 13 ++++++ operations/models.py | 8 +++- operations/static/operations/job_panel.js | 21 ++++++--- .../operations/job_panel_script.html | 2 +- operations/tests/fakes.py | 27 ++++++++++++ operations/tests/js/job_panel.test.js | 41 ++++++++++++++++++ operations/tests/test_api.py | 8 ++++ operations/tests/test_kinds.py | 40 ++++++++++++++++- operations/tests/test_recovery.py | 6 +++ operations/tests/test_views.py | 8 +++- operations/tests/test_worker.py | 43 ++++++++++++++++++- operations/views.py | 5 ++- operations/worker.py | 12 +++++- 14 files changed, 224 insertions(+), 17 deletions(-) diff --git a/operations/api/serializers.py b/operations/api/serializers.py index 5cb67709..39079a92 100644 --- a/operations/api/serializers.py +++ b/operations/api/serializers.py @@ -10,12 +10,13 @@ class OperationSerializer(serializers.ModelSerializer): label = serializers.SerializerMethodField() resource_label = serializers.SerializerMethodField() cancellable = serializers.SerializerMethodField() + in_request = serializers.SerializerMethodField() class Meta: model = Operation fields = ["id", "kind", "label", "username", "resource", "resource_label", "state", "phase", "progress_done", "progress_total", "result", "error", "created_at", "started_at", - "finished_at", "cancel_requested", "cancellable", "resolution", "resolution_note"] + "finished_at", "cancel_requested", "cancellable", "in_request", "resolution", "resolution_note"] read_only_fields = fields def get_label(self, op) -> str: @@ -28,6 +29,10 @@ class OperationSerializer(serializers.ModelSerializer): kind = KINDS.get(op.kind) return op.state == Operation.QUEUED or bool(kind and kind.cancellable and op.state == Operation.RUNNING) + def get_in_request(self, op) -> bool: + """A request runs it: it does not wait for the worker.""" + return op.in_request + class OperationSummarySerializer(serializers.Serializer): kind = serializers.CharField() diff --git a/operations/jobs.py b/operations/jobs.py index 9a7db501..9d019f24 100644 --- a/operations/jobs.py +++ b/operations/jobs.py @@ -73,6 +73,7 @@ class Kind: label = "" cancellable = False time_limit = None + runs_in_request = False # its requests run it: admitted running, never the worker's progress_bytes = False # progress counts bytes: the job page shows sizes def normalize(self, raw): @@ -100,6 +101,10 @@ class Kind: def reconcile(self, op): raise NotImplementedError + def expire(self, op): + """Called on every housekeeping pass for an active job a request runs; + the kind decides whether the job is idle and ends it if so.""" + def log_entry(self, op, state): """(host, instance, message) of the Logs row written with each end of the job (in the same transaction); its host and instance also go on @@ -170,6 +175,9 @@ def submit(kind_name, user, raw_params, submission_key): resource=plan.resource, resource_label=plan.label[:255], params=params, intent=plan.intent, time_limit=kind.time_limit, ) + if kind.runs_in_request: + now = timezone.now() + op.state, op.started_at, op.heartbeat_at, op.worker = Operation.RUNNING, now, now, Operation.REQUEST return claims.admit_job( op, plan.claims, existing=lambda: _existing(user, submission_key, kind.name, params), @@ -182,6 +190,8 @@ def cancel(op_id, user): op = Operation.objects.select_for_update().get(pk=op_id) if not utils.may_see(op, user): raise PermissionDenied + # read before the update below; a repeated cancel retries the request job's stop + request_job = op.in_request and op.state in (Operation.RUNNING, Operation.CANCELLING) if op.state == Operation.QUEUED: Operation.objects.filter(pk=op.pk).update(state=Operation.CANCELLED, finished_at=timezone.now()) Claim.objects.filter(operation=op).delete() @@ -194,6 +204,9 @@ def cancel(op_id, user): elif op.state != Operation.CANCELLING: raise OperationError("The job has already ended") op.refresh_from_db() + if request_job: # no worker signals it: the request stops it + get_kind(op.kind).cancel(op) + op.refresh_from_db() return op diff --git a/operations/models.py b/operations/models.py index cffb4ac9..78e46365 100644 --- a/operations/models.py +++ b/operations/models.py @@ -26,6 +26,7 @@ class Operation(models.Model): ACTIVE = (QUEUED, RUNNING, CANCELLING) TERMINAL = (SUCCEEDED, FAILED, CANCELLED, UNCERTAIN) STATES = [(s, s) for s in ACTIVE + TERMINAL] + REQUEST = "request" # the worker of a job a request runs id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) kind = models.CharField(max_length=64, db_index=True) @@ -49,7 +50,8 @@ class Operation(models.Model): started_at = models.DateTimeField(null=True) finished_at = models.DateTimeField(null=True) heartbeat_at = models.DateTimeField(null=True) - worker = models.CharField(max_length=128, blank=True) # identity of the worker that owns it + # identity of the owning worker; REQUEST for a job a request runs, which the worker never claims or recovers + worker = models.CharField(max_length=128, blank=True) time_limit = models.PositiveIntegerField(null=True) # seconds, fixed at admission cancel_requested = models.BooleanField(default=False) resolved_by = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, on_delete=models.SET_NULL, related_name="+") @@ -64,6 +66,10 @@ class Operation(models.Model): def __str__(self): return f"{self.kind} {self.id}" + @property + def in_request(self): + return self.worker == Operation.REQUEST + class Claim(models.Model): SHARED = "shared" diff --git a/operations/static/operations/job_panel.js b/operations/static/operations/job_panel.js index 51e66da8..87568282 100644 --- a/operations/static/operations/job_panel.js +++ b/operations/static/operations/job_panel.js @@ -90,8 +90,10 @@ function jobPanel(element, urls, csrfToken, initialIds) { if (last.short_operation) { html += '
A short operation is in progress.
'; } - // a job the worker has to start, run or stop (the Jobs page uses the same rule) - var waiting = last.jobs.some(function (j) { return ["queued", "running", "cancelling"].indexOf(j.state) >= 0; }); + // a job the worker has to start, run or stop, not one a request runs (the Jobs page uses the same rule) + var waiting = last.jobs.some(function (j) { + return ["queued", "running", "cancelling"].indexOf(j.state) >= 0 && !j.in_request; + }); var seen = last.worker_last_seen ? new Date(last.worker_last_seen) : null; if (waiting && (!seen || Date.now() - seen.getTime() > 60000)) { html += '
Worker unavailable, last seen ' @@ -115,10 +117,12 @@ function jobPanel(element, urls, csrfToken, initialIds) { inflight[id] = true; $.getJSON(urls.job(id)) .done(function (job) { + if (dismissed[id]) { delete pending[id]; return; } + // still active: pending until a list shows it (one older than the job may not) or it + // ends; the next poll asks for it again + if (["queued", "running", "cancelling"].indexOf(job.state) >= 0) { return; } delete pending[id]; - if (dismissed[id]) { return; } - // a just-submitted job that is still active shows in the list instead - if (["queued", "running", "cancelling"].indexOf(job.state) < 0) { outcomes[id] = job; } + outcomes[id] = job; render(); }) .fail(function (xhr) { @@ -169,5 +173,12 @@ function jobPanel(element, urls, csrfToken, initialIds) { } var poller = pollWhileVisible(3000, function () { return $.getJSON(urls.resource); }, onData, 0); + // follow a job the page just started, as a remembered one + poller.track = function (id) { + pending[id] = true; + render(); + poller.stop(); // ask for the list now + poller.start(); + }; return poller; } diff --git a/operations/templates/operations/job_panel_script.html b/operations/templates/operations/job_panel_script.html index e1d3b495..ad0258f2 100644 --- a/operations/templates/operations/job_panel_script.html +++ b/operations/templates/operations/job_panel_script.html @@ -1,7 +1,7 @@ {% load static %}