sidecar: classical resources should not be dispatched on fail/cancel

python/fluence/providers/base.py

@@ -31,6 +31,26 @@
 TAG_KEY = "fluence-pod-uid"
 
 
+def ungate_position(default=1):
+    """Queue position we ungate at. One means next in line."""
+    try:
+        return int(os.environ.get("FLUENCE_UNGATE_POSITION", default))
+    except ValueError:
+        return default
+
+
+def position_at_most(pos, threshold):
+    """True when the queue position is at or under threshold.
+
+    Braket reports anything over 2000 as the string >2000, so a position that
+    is not a number counts as far away.
+    """
+    try:
+        return int(pos) <= int(threshold)
+    except (TypeError, ValueError):
+        return False
+
+
 def log(msg: str) -> None:
     ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
     print(f"[fluence] {ts} {msg}", flush=True)
@@ -79,11 +99,21 @@ def find_my_task(self, pod_uid: str, backend: str, timeout: int) -> "Task | None
         found or timeout. Returns an opaque Task or None."""
         raise NotImplementedError
 
-    def is_ready_to_ungate(self, task: "Task") -> bool:
-        """True when the gang should be ungated — queue position == 1 or the task
-        is already RUNNING/terminal. Always implementable."""
+    def is_ready_to_ungate(self, task: "Task", position=None) -> bool:
+        """True when the gang should be ungated.
+
+        Either the queue position is at or under position, or the task already
+        left the queue and can still return a result. Ungating earlier gives
+        the gang time to start up, so raise position when it is slow.
+        """
         raise NotImplementedError
 
+    def task_failed(self, task: "Task") -> bool:
+        """True when the task ended with no result, so there is nothing to
+        fetch. Defaults to False for vendors that cannot tell.
+        """
+        return False
+
     def queue_position(self, task: "Task") -> "int | None":
         """Optional richer telemetry: integer queue position (1 == next), or None
         if the vendor does not expose one. Not required for the ungate decision."""

python/fluence/providers/braket.py

@@ -16,7 +16,15 @@
 import os
 import time
 
-from fluence.providers.base import Provider, Task, TAG_KEY, log, register
+from fluence.providers.base import (
+    Provider,
+    Task,
+    TAG_KEY,
+    log,
+    position_at_most,
+    register,
+    ungate_position,
+)
 
 
 class BraketTask(Task):
@@ -210,12 +218,22 @@ def _aws_task(self, task: BraketTask):
         from braket.aws import AwsQuantumTask
         return AwsQuantumTask(arn=task.arn)
 
-    def is_ready_to_ungate(self, task: BraketTask) -> bool:
+    def is_ready_to_ungate(self, task: BraketTask, position=None) -> bool:
         t = self._aws_task(task)
-        if t.state() in ("RUNNING", "COMPLETED", "FAILED", "CANCELLED"):
+        # a task that left the queue reports no position, so this is how we
+        # notice we missed the window
+        if t.state() in ("RUNNING", "COMPLETED"):
             return True
+        if position is None:
+            position = ungate_position()
         try:
-            return str(t.queue_position().queue_position) == "1"
+            return position_at_most(t.queue_position().queue_position, position)
+        except Exception:
+            return False
+
+    def task_failed(self, task: BraketTask) -> bool:
+        try:
+            return self._aws_task(task).state() in ("FAILED", "CANCELLED")
         except Exception:
             return False
 

python/fluence/sidecar.py

@@ -18,6 +18,8 @@
   FLUXION_BACKEND / FLUXION_VENDOR  scheduler-chosen backend / vendor
   FLUENCE_TASK_DISCOVERY_TIMEOUT  seconds to wait for discovery (default 300)
   FLUENCE_POLL_INTERVAL           seconds between polls (default 30)
+  FLUENCE_UNGATE_POSITION         ungate at this queue position or closer
+                                  (default 1, next in line)
 """
 
 from __future__ import annotations
@@ -27,20 +29,24 @@
 import time
 
 from fluence.providers import resolve_from_env
-from fluence.providers.base import log
+from fluence.providers.base import log, ungate_position
 from fluence.ungate import ungate_pods, gated_pods_from_env, namespace_from_env, wait_for_gated_pods
 
 
 
-def _poll(provider, task, poll_interval, ungate):
+def _poll(provider, task, poll_interval, ungate, position=1):
+    """Poll until the task is ready or has failed. True when ready."""
     mode = "gang" if ungate else "observe-only"
-    log(f"{mode} mode: polling queue position")
+    log(f"{mode} mode: polling queue position, ungating at {position} or closer")
     last = object()
     while True:
         try:
-            if provider.is_ready_to_ungate(task):
+            if provider.task_failed(task):
+                log("ERROR: task reached a terminal state with no result")
+                return False
+            if provider.is_ready_to_ungate(task, position):
                 log(f"task ready (position={provider.queue_position(task)})")
-                return
+                return True
             pos = provider.queue_position(task)
             if pos != last:
                 log(f"queue position: {pos}")
@@ -64,6 +70,7 @@ def main():
     discovery_timeout = int(os.environ.get("FLUENCE_TASK_DISCOVERY_TIMEOUT", 300))
     poll_interval = int(os.environ.get("FLUENCE_POLL_INTERVAL", 30))
     ungate_timeout = int(os.environ.get("FLUENCE_UNGATE_TIMEOUT", 120))
+    ungate_at = ungate_position()
 
     namespace = namespace_from_env()
 
@@ -90,11 +97,22 @@ def main():
     job_id = provider.job_id(task)
     log(f"discovered task, job_id={job_id}")
 
-    _poll(provider, task, poll_interval, ungate=not observe)
+    ready = _poll(provider, task, poll_interval, ungate=not observe,
+                  position=ungate_at)
 
     if observe:
         log("observe-only run complete")
-        return
+        sys.exit(0 if ready else 1)
+
+    if not ready:
+        # fail open like a discovery failure does, so the gang is not stranded.
+        # The pods find no result and exit, and we exit non zero so it shows
+        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)
+        sys.exit(1)
 
     # Ungate the gang: discover the gated pods in the gang group and remove their
     # gate, stamping the job-id so each can fetch results by id. The gang pods are