From 0c8aecd2eec00e2c21bb0b7942c64f1bcde4ff36 Mon Sep 17 00:00:00 2001 From: vsoch Date: Tue, 18 Aug 2026 20:39:22 -0700 Subject: [PATCH] sidecar: classical resources should not be dispatched on fail/cancel Signed-off-by: vsoch --- python/fluence/providers/base.py | 36 +++++++++++++++++++++++++++--- python/fluence/providers/braket.py | 26 +++++++++++++++++---- python/fluence/sidecar.py | 32 ++++++++++++++++++++------ 3 files changed, 80 insertions(+), 14 deletions(-) diff --git a/python/fluence/providers/base.py b/python/fluence/providers/base.py index 561bca2..527dd45 100644 --- a/python/fluence/providers/base.py +++ b/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.""" diff --git a/python/fluence/providers/braket.py b/python/fluence/providers/braket.py index d6e6ea9..bd11e9d 100644 --- a/python/fluence/providers/braket.py +++ b/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 diff --git a/python/fluence/sidecar.py b/python/fluence/sidecar.py index d0724e5..1924d8a 100644 --- a/python/fluence/sidecar.py +++ b/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