Skip to content

feat(orchestrator): a seam for taking queued executions off the launch path #346

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
yuechao-qin wants to merge 4 commits into
base: master
Choose a base branch
from
+430 −14
Open

feat(orchestrator): a seam for taking queued executions off the launch path #346

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
49b2780
feat(orchestrator): a seam for taking queued executions off the launc…
yuechao-qin Aug 26, 2026
2564bdd
Merge remote-tracking branch 'origin/master' into ycq/gsd-proton-quot…
yuechao-qin Sep 9, 2026
0ee981c
fix(orchestrator): fail open when the queued-execution interceptor ra…
yuechao-qin Sep 11, 2026
53f398e
test(orchestrator): pin the interceptor seam's position, not just its…
yuechao-qin Sep 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion cloud_pipelines_backend/backend_types_sql.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,15 @@


class ContainerExecutionStatus(str, enum.Enum):
"""The lifecycle status of an execution node.

`UNINITIALIZED` is the parked state: an execution a `QueuedExecutionInterceptor`
took off the launch path. The queued sweep deliberately does not select it, so a
parked execution stays invisible until whoever parked it puts it back to `QUEUED`.
"""

INVALID = "INVALID" # Compatibility with Vertex AI CustomJob
UNINITIALIZED = "UNINITIALIZED" # Remove
UNINITIALIZED = "UNINITIALIZED" # Parked by an interceptor; not swept
QUEUED = "QUEUED" # Before WAITING_FOR_UPSTREAM or STARTING
# READY_TO_START = "READY_TO_START" # Input artifacts ready, but no job ID
WAITING_FOR_UPSTREAM = "WAITING_FOR_UPSTREAM"
Expand Down
47 changes: 41 additions & 6 deletions cloud_pipelines_backend/orchestrator_sql.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,21 @@ class OrchestratorError(RuntimeError):
pass


class QueuedExecutionInterceptor(typing.Protocol):
"""Given a chance to take a queued execution off the launch path.

Implemented downstream. Called on the orchestrator's session once the execution is
known to be launchable -- inputs present, not conditionally skipped, no cache hit, not
cancelled. An implementation that returns True owns the execution from that point: it
sets whatever status it wants and commits. The orchestrator makes no assumption about
which status that is.
"""
Comment thread
yuechao-qin marked this conversation as resolved.

def intercept(self, *, session: orm.Session, execution: bts.ExecutionNode) -> bool:
"""True if this execution was taken over and must not launch; False to continue."""
...


class OrchestratorService_Sql:
def __init__(
self,
Expand All @@ -57,6 +72,7 @@ def __init__(
_max_container_execution_refresh_error_retries: int = 3,
_max_queue_batch_size: int = 1,
_max_queue_batch_duration: datetime.timedelta = datetime.timedelta(),
queued_execution_interceptor: QueuedExecutionInterceptor | None = None,
):
self._session_factory = session_factory
self._launcher = launcher
Expand All @@ -75,6 +91,7 @@ def __init__(

self._max_queue_batch_size = _max_queue_batch_size
self._max_queue_batch_duration = _max_queue_batch_duration
self._queued_execution_interceptor = queued_execution_interceptor

def run_loop(self):
while True:
Expand Down Expand Up @@ -124,12 +141,8 @@ def internal_process_queued_executions_queue(self, session: orm.Session):
query_start_timestamp = time.monotonic_ns()
query = (
sql.select(bts.ExecutionNode).where(
bts.ExecutionNode.container_execution_status.in_(
(
bts.ContainerExecutionStatus.UNINITIALIZED,
bts.ContainerExecutionStatus.QUEUED,
)
)
bts.ExecutionNode.container_execution_status
== bts.ContainerExecutionStatus.QUEUED
Comment on lines 142 to +145

@Volv-G Volv-G Sep 2, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(AI-assisted)

Overloading UNINITIALIZED as the parked state is the part of this design I'd push back on — I think this wants a dedicated status, which is where the design discussion landed too ("look into creating a new state", left unresolved).

Three reasons the reuse bothers me:

  1. The two meanings are opposites. "Never initialized" and "deliberately taken off the launch path by a gate that intends to put it back" are different facts about a node, and after this PR nothing distinguishes them. A reader of a row — or of a dashboard — cannot tell which one they're looking at.

  2. Legacy rows change meaning retroactively. 3a2173b API server - Changed the initial status from UNINITIALIZED to QUEUED means this was the initial status for new nodes. Any surviving row at UNINITIALIZED is drained by the sweep today and becomes permanently invisible after this change. Probably zero rows in practice — SELECT COUNT(*) FROM execution_node WHERE container_execution_status = 'UNINITIALIZED' settles it — but with a distinct state the question wouldn't arise at all.

  3. It's user-visible. A quota-parked node will render as UNINITIALIZED in the UI, which is meaningless to the person whose pipeline it is. That label mapping does not live in this repo, so nothing here can soften it.

The honest counter-argument, which I don't think is fatal: container_execution_status compiles to a MySQL ENUM(...), so a new member is ALTER TABLE execution_node MODIFY COLUMN … on a very hot table — and at least one downstream consumer builds its schema with metadata.create_all() and no migration framework, so it would need a hand-written migration there too. That is real cost. But it is a one-time cost paid at the bottom of a six-PR stack, and it only gets more expensive once parked rows exist in production and the ambiguity is load-bearing.

So: either add the state now, or — if the cost wins — please record the decision in the enum comment (backend_types_sql.py:17-23) as an explicit, priced trade-off rather than a reuse of a spare member, so the next person to touch this knows it was chosen and not inherited.

)
# TODO: Maybe add last_processed_at
# .order_by(bts.ExecutionNode.last_processed_at)
Expand Down Expand Up @@ -610,6 +623,28 @@ def internal_process_one_queued_execution(
session.commit()
return

# Give the interceptor a chance to take this execution off the launch path.
# If it returns True it has taken ownership: it decided what state the execution is
Comment thread
yuechao-qin marked this conversation as resolved.
# in and committed that itself. We stop here and do not launch.
if self._queued_execution_interceptor is not None:
try:
intercepted = self._queued_execution_interceptor.intercept(
session=session, execution=execution
)
except Exception:
# Fail open. An optional gate must not be able to stop the fleet: the
# failure mode of a broken interceptor is no gating, not no launches.
# No rollback here on purpose: a gate that raised mid-write leaves the
# session dirty, and the launch path below opens a new transaction before
# it writes anything, which discards it.
_logger.exception(
f"Queued-execution interceptor raised on execution {execution.id}; "
f"launching ungated."
)
intercepted = False
if intercepted:
return

# Creating new container execution
container_execution_uuid = _generate_random_id()

Expand Down
Loading
Loading