bug: do not release when scout not ready

Signed-off-by: vsoch <vsoch@users.noreply.github.com>

Author: vsoch

docs/quantum-scheduling.md

@@ -63,7 +63,7 @@ The quantum handler also enforces admission control, which is always on unless t
 5. When the task is next in line, or already running, annotate each worker with the job id and the session, then remove the gate.
 6. Wait for the workers to reach Succeeded or Failed, write the done marker so the scout's release proceeds, and exit.
 
-If the task is never found or fails, the sidecar ungates anyway so the workers are not stranded, and exits non zero so it shows.
+The sidecar ungates the workers only when the scout's task is running or has completed. If the task fails, is cancelled, or is never found, the workers stay gated, the sidecar tells the scout to release its session at once, and it exits non zero so it shows. Delete the job to clear the gated workers. The one exception is a rehearsal, where the scout exits without submitting anything, for example with no credentials. Then the workers are released with no session.
 
 ## Priority and preemption
 

python/fluence/providers/qrmi.py

@@ -88,16 +88,10 @@ def _status_name(status) -> str:
     return s.rsplit(".", 1)[-1].upper()
 
 
-# the task has left the queue or already finished
-_READY_STATES = {
-    "RUNNING",
-    "COMPLETED",
-    "COMPLETE",
-    "DONE",
-    "FAILED",
-    "CANCELLED",
-    "CANCELED",
-}
+# the task is running on the device, or ran to the end
+_READY_STATES = {"RUNNING", "COMPLETED", "COMPLETE", "DONE"}
+# the task ended with nothing to use
+_FAILED_STATES = {"FAILED", "CANCELLED", "CANCELED"}
 
 
 class QRMIProvider(Provider):
@@ -255,10 +249,15 @@ def _status(self, task: QRMITask) -> "str | None":
     def is_ready_to_ungate(self, task: QRMITask, position=None) -> bool:
         # QRMI reports no queue position, so position is accepted to match the
         # interface and the sidecar's call, and ignored. Ready means the task
-        # has left the queue.
+        # is running on the device or has completed. Failed and cancelled are
+        # not ready, task_failed reports them.
         name = self._status(task)
         return name in _READY_STATES if name else False
 
+    def task_failed(self, task: QRMITask) -> bool:
+        name = self._status(task)
+        return name in _FAILED_STATES if name else False
+
     def queue_position(self, task: QRMITask):
         # QRMI has no queue position, so report 1 once the task runs and None before
         name = self._status(task)

python/fluence/sidecar.py

@@ -50,10 +50,6 @@ def _poll(provider, task, poll_interval, ungate, position=1):
     log(f"{mode} mode: polling queue position, ungating at {position} or closer")
     last = object()
     # A vendor call can fail for a while and recover, so those are retried.
-    # A bug in our own code fails the same way every time and never
-    # recovers, and while it retries a session that already started keeps
-    # billing with nothing using it. So programming errors stop the poll at
-    # once, and so does any error that keeps happening.
     max_errors = int(os.environ.get("FLUENCE_POLL_MAX_ERRORS", 10))
     errors = 0
     while True:
@@ -143,19 +139,13 @@ def main():
         _hold_for_workers(namespace, gang_group, pod_name)
         sys.exit(0)
     if task is None:
-        log("ERROR: could not discover quantum task")
-        if not observe:
-            # fail open so the workers are not stranded
-            ungate_pods(
-                wait_for_gated_pods(
-                    namespace, gang_group, exclude=pod_name, timeout=ungate_timeout
-                ),
-                "",
-                namespace,
-                backend=placed_backend,
-                attributes=placed_attrs,
-            )
-            _hold_for_workers(namespace, gang_group, pod_name)
+        # The scout neither published a task nor exited, so something is
+        # wrong with it. Releasing the workers would start them on a run that
+        # has no quantum task, so they stay gated. Delete the job to clear them.
+        log(
+            "ERROR: could not discover the quantum task. The workers stay gated, "
+            "delete the job to clear them"
+        )
         sys.exit(1)
 
     job_id = provider.job_id(task)
@@ -168,25 +158,16 @@ def main():
         sys.exit(0 if ready else 1)
 
     if not ready:
-        # fail open so the workers are not stranded. The session goes with
-        # them: the scout is still holding it, and workers that ran outside
-        # it would leave it open and idle, which a dedicated session bills
+        # The task failed, was cancelled, or could not be followed. There is
+        # nothing for the workers to use, so they stay gated, and the scout is
+        # told to stop holding its session so it closes now instead of when
+        # the hold times out. Delete the job to clear the workers.
         log(
-            "ERROR: ungating anyway so the gang is not stranded, but the task "
-            "produced no result"
-        )
-        ungate_pods(
-            gated_pods_from_env()
-            or wait_for_gated_pods(
-                namespace, gang_group, exclude=pod_name, timeout=ungate_timeout
-            ),
-            job_id,
-            namespace,
-            session_id=provider.session_id(task),
-            backend=placed_backend,
-            attributes=placed_attrs,
+            "ERROR: the quantum task is not usable. The workers stay gated and "
+            "the scout releases its session now. Delete the job to clear them"
         )
-        _hold_for_workers(namespace, gang_group, pod_name)
+        if not handoff.mark_workers_done():
+            log("WARNING: could not tell the scout to release its session")
         sys.exit(1)
 
     # ungate the workers, stamping the job id and the session first

python/tests/test_qrmi.py

@@ -470,39 +470,101 @@ def queue_position(self, task):
     assert sidecar._poll(Recovering(), object(), 0, ungate=True) is True
 
 
-def test_fail_open_hands_the_workers_the_session(monkeypatch):
-    """When the poll fails the workers are released anyway. They must get
-    the session id, or they run outside a session the scout still holds."""
+def _sidecar_with(monkeypatch, task, ready, scout_exited=False):
+    """Run the sidecar's main with a fake provider. Returns what it did."""
     from fluence import sidecar
 
-    calls = {}
-    monkeypatch.setattr(sidecar, "_poll", lambda *a, **k: False)
+    did = {"ungated": None, "marked": False}
+    monkeypatch.setattr(sidecar, "_poll", lambda *a, **k: ready)
     monkeypatch.setattr(sidecar, "_hold_for_workers", lambda *a, **k: None)
     monkeypatch.setattr(sidecar, "gated_pods_from_env", lambda: ["worker-0"])
+    monkeypatch.setattr(sidecar, "wait_for_gated_pods", lambda *a, **k: ["worker-0"])
     monkeypatch.setattr(
         sidecar,
         "ungate_pods",
-        lambda pods, job_id, ns, **kw: calls.update(kw) or len(pods),
+        lambda pods, job_id, ns, **kw: did.update(ungated={"job_id": job_id, **kw})
+        or len(pods),
     )
     monkeypatch.setattr(sidecar, "placement_from_env", lambda: ("ibm_x", {}))
     monkeypatch.setattr(sidecar, "namespace_from_env", lambda: "default")
+    monkeypatch.setattr(
+        sidecar.handoff,
+        "mark_workers_done",
+        lambda *a, **k: did.update(marked=True) or True,
+    )
+    monkeypatch.setattr(sidecar.handoff, "scout_exited", lambda *a, **k: scout_exited)
 
     class P:
         name = "fake"
 
         def find_my_task(self, uid, backend, timeout):
-            return "task"
+            return task
 
-        def job_id(self, task):
+        def job_id(self, t):
             return "job-1"
 
-        def session_id(self, task):
+        def session_id(self, t):
             return "sess-1"
 
     monkeypatch.setattr(sidecar, "resolve_from_env", lambda: P())
     monkeypatch.setenv("FLUENCE_OBSERVE", "false")
+    code = 0
     try:
         sidecar.main()
     except SystemExit as e:
-        assert e.code == 1
-    assert calls.get("session_id") == "sess-1"
+        code = e.code
+    did["code"] = code
+    return did
+
+
+def test_a_task_that_is_not_ready_leaves_the_workers_gated(monkeypatch):
+    """A failed or cancelled task has nothing for the workers. Releasing them
+    would only spend money on a run that already failed."""
+    did = _sidecar_with(monkeypatch, task="task", ready=False)
+    assert did["ungated"] is None
+    assert did["code"] == 1
+
+
+def test_a_task_that_is_not_ready_tells_the_scout_to_release(monkeypatch):
+    """The scout holds its session until the workers are done. They will
+    never run, so the marker is written now and the session closes."""
+    did = _sidecar_with(monkeypatch, task="task", ready=False)
+    assert did["marked"] is True
+
+
+def test_a_task_that_cannot_be_found_leaves_the_workers_gated(monkeypatch):
+    did = _sidecar_with(monkeypatch, task=None, ready=False, scout_exited=False)
+    assert did["ungated"] is None
+    assert did["code"] == 1
+
+
+def test_a_rehearsal_still_releases_the_workers(monkeypatch):
+    """The scout exited with nothing submitted, no credentials. That is the
+    mock run, and the workers go with no session."""
+    did = _sidecar_with(monkeypatch, task=None, ready=False, scout_exited=True)
+    assert did["ungated"] is not None
+    assert did["ungated"]["job_id"] == ""
+    assert not did["ungated"].get("session_id")
+    assert did["code"] == 0
+
+
+def test_a_ready_task_releases_the_workers_with_the_session(monkeypatch):
+    did = _sidecar_with(monkeypatch, task="task", ready=True)
+    assert did["ungated"]["job_id"] == "job-1"
+    assert did["ungated"]["session_id"] == "sess-1"
+
+
+def test_qrmi_ready_means_running_or_completed(monkeypatch):
+    from fluence.providers import qrmi as q
+
+    prov = q.QRMIProvider()
+    for name, ready, failed in [
+        ("QUEUED", False, False),
+        ("RUNNING", True, False),
+        ("COMPLETED", True, False),
+        ("FAILED", False, True),
+        ("CANCELLED", False, True),
+    ]:
+        monkeypatch.setattr(prov, "_status", lambda task, n=name: n)
+        assert prov.is_ready_to_ungate(object()) is ready, name
+        assert prov.task_failed(object()) is failed, name