scout: add support for an exit marker

.github/workflows/sidecar-build-deploy.yaml

@@ -51,6 +51,7 @@ jobs:
           python tests/test_fluence.py
           python tests/test_qrmi.py
           python tests/test_hold.py
+          python tests/test_rehearsal.py
 
   build-deploy:
     runs-on: ubuntu-latest

python/fluence/handoff.py

@@ -67,20 +67,61 @@ def read(path=None) -> "dict | None":
 
 
 def wait(timeout, interval=5, path=None, log=None) -> "dict | None":
-    """Poll for the record until it appears or timeout seconds pass."""
+    """Poll for the record until it appears, the scout exits without one, or
+    timeout seconds pass."""
     deadline = time.time() + timeout
     announced = False
+    exited = os.path.join(os.path.dirname(handoff_path(path)), EXITED_FILENAME)
     while time.time() < deadline:
         record = read(path)
         if record is not None:
             return record
+        if os.path.exists(exited):
+            if log:
+                log("the scout exited without publishing a task")
+            return None
         if log and not announced:
             log(f"waiting for the scout to publish its job id at {handoff_path(path)}")
             announced = True
         time.sleep(interval)
     return None
 
 
+# the interceptor writes this when the scout's interpreter exits with no task
+# published, so a rehearsal with no credentials does not leave the workers
+# gated until the discovery timeout
+EXITED_FILENAME = "scout-exited"
+
+
+def exited_path(path=None) -> str:
+    if path:
+        return path
+    d = os.environ.get(HANDOFF_DIR_ENV, DEFAULT_HANDOFF_DIR)
+    return os.path.join(d, EXITED_FILENAME)
+
+
+def mark_scout_exited(path=None) -> bool:
+    """Write the marker, unless a task was published. Never raises."""
+    try:
+        if read() is not None:
+            return False
+        dest = exited_path(path)
+        os.makedirs(os.path.dirname(dest), exist_ok=True)
+        fd, tmp = tempfile.mkstemp(
+            dir=os.path.dirname(dest), prefix=".exited.", suffix=".tmp"
+        )
+        with os.fdopen(fd, "w") as f:
+            json.dump({"exited_at": time.time()}, f)
+        os.replace(tmp, dest)
+        return True
+    except Exception:
+        return False
+
+
+def scout_exited(path=None) -> bool:
+    return os.path.exists(exited_path(path))
+
+
 # the sidecar writes this once the workers are done and the scout's release waits for it
 DONE_FILENAME = "workers-done"
 

python/fluence/interceptor.py

@@ -12,6 +12,7 @@
 
 from __future__ import annotations
 
+import atexit
 import os
 
 
@@ -38,5 +39,17 @@ def install() -> None:
             )
 
 
+def _mark_exit() -> None:
+    # tell the sidecar the application ended, so a scout that never submitted
+    # releases its workers instead of leaving them gated
+    try:
+        from fluence import handoff
+
+        handoff.mark_scout_exited()
+    except Exception:
+        pass
+
+
 # Install on import.
 install()
+atexit.register(_mark_exit)

python/fluence/providers/braket.py

@@ -16,6 +16,7 @@
 import os
 import time
 
+from fluence import handoff
 from fluence.providers.base import (
     Provider,
     Task,
@@ -176,6 +177,9 @@ def find_my_task(self, pod_uid, backend, timeout):
             set()
         )  # regions that returned AccessDenied, skip them after the first hit
         while time.time() < deadline:
+            if handoff.scout_exited():
+                log("[braket] the scout exited without a task, giving up the search")
+                return None
             for region in regions:
                 if region in denied:
                     continue

python/fluence/sidecar.py

@@ -101,6 +101,21 @@ def main():
     log(f"resolved provider: {provider.name}")
 
     task = provider.find_my_task(pod_uid, backend, discovery_timeout)
+    if task is None and handoff.scout_exited() and not observe:
+        # a rehearsal, or an application that never submitted. Release the
+        # workers with no session and hold for them as usual
+        log(
+            "the scout finished without submitting a task, releasing the workers with no session"
+        )
+        ungate_pods(
+            wait_for_gated_pods(
+                namespace, gang_group, exclude=pod_name, timeout=ungate_timeout
+            ),
+            "",
+            namespace,
+        )
+        _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:

python/tests/test_rehearsal.py

@@ -0,0 +1,75 @@
+"""
+Tests for the scout exit marker. A scout that ends without submitting a task,
+a rehearsal with no credentials, leaves a marker that ends the sidecar's wait
+for the handoff at once, so the workers are released instead of gated until
+the discovery timeout.
+
+Run: PYTHONPATH=python python3 python/tests/test_rehearsal.py
+"""
+
+import os
+import sys
+import tempfile
+import time
+
+
+def test_marker_roundtrip():
+    from fluence import handoff
+
+    with tempfile.TemporaryDirectory() as d:
+        p = os.path.join(d, "scout-exited")
+        assert not handoff.scout_exited(p)
+        assert handoff.mark_scout_exited(p)
+        assert handoff.scout_exited(p)
+
+
+def test_no_marker_when_a_task_was_published():
+    from fluence import handoff
+
+    with tempfile.TemporaryDirectory() as d:
+        os.environ["FLUENCE_HANDOFF_DIR"] = d
+        try:
+            assert handoff.publish("job-1")
+            assert not handoff.mark_scout_exited()
+            assert not handoff.scout_exited()
+        finally:
+            os.environ.pop("FLUENCE_HANDOFF_DIR", None)
+
+
+def test_wait_returns_at_once_after_the_scout_exits():
+    from fluence import handoff
+
+    with tempfile.TemporaryDirectory() as d:
+        os.environ["FLUENCE_HANDOFF_DIR"] = d
+        try:
+            handoff.mark_scout_exited()
+            start = time.time()
+            assert handoff.wait(30, interval=1) is None
+            assert time.time() - start < 5
+        finally:
+            os.environ.pop("FLUENCE_HANDOFF_DIR", None)
+
+
+def test_wait_still_times_out_without_a_marker():
+    from fluence import handoff
+
+    with tempfile.TemporaryDirectory() as d:
+        start = time.time()
+        assert handoff.wait(1, interval=1, path=os.path.join(d, "task.json")) is None
+        assert time.time() - start >= 1
+
+
+if __name__ == "__main__":
+    fns = [
+        v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)
+    ]
+    failed = 0
+    for fn in fns:
+        try:
+            fn()
+            print(f"PASS {fn.__name__}")
+        except Exception as e:
+            failed += 1
+            print(f"FAIL {fn.__name__}: {e}")
+    print(f"\n{len(fns) - failed}/{len(fns)} passed")
+    sys.exit(1 if failed else 0)