| // Copyright 2025 The Chromium Authors |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| syntax = "proto3"; |
| |
| package turboci.graph.executor.v1; |
| |
| import "turboci/graph/executor/v1/cancel_stage.proto"; |
| import "turboci/graph/executor/v1/run_stage.proto"; |
| import "turboci/graph/executor/v1/validate_stage.proto"; |
| |
| option go_package = "go.chromium.org/turboci/proto/go/graph/executor/v1/grpcpb;executorgrpcpb"; |
| option java_multiple_files = true; |
| |
| // TurboCIStageExecutor defines the interface that all stage executors must |
| // implement. |
| // |
| // The TurboCI Orchestrator can be configured to use many different services |
| // which implement TurboCIStageExecutor to execute stages of different types. |
| // |
| // TBD: Explain end-user-credential propagation. |
| service TurboCIStageExecutor { |
| // ValidateStage instructs the executor to validate the stage. |
| // |
| // It is called once (i.e. not per attempt) when the stage is being inserted |
| // into the graph. It should check the executor can run the stage at all (this |
| // includes checking ACLs if appropriate). |
| // |
| // If the stage looks good (i.e. this RPC returns OK), the executor has |
| // opportunity to adjust some of stages parameters before the stage is |
| // inserted into the graph by returning them in ValidateStageResponse |
| // (currently contains only turboci.orchestrator.v1.StageExecutionPolicy). |
| // |
| // If the stage doesn't look good (i.e. this RPC returns an error), the entire |
| // WriteNodes transaction that was attempting to insert the stage will be |
| // aborted. The graph will be left unchanged. |
| // |
| // RunStage RPC is generally expected to redo the validation, since the stage |
| // parameters may change after the stage is inserted. |
| // |
| // TBD: Formalize how the error will be propagated back to the client |
| // attempting to insert this stage. How would we allow details in the |
| // response? |
| // |
| // Sidenote - do we need to record this failed node insertion somewhere |
| // in the graph/ledger for observability? |
| // |
| // Must have no side effects, since it observes stages that potentially will |
| // never exist in the graph for real. |
| rpc ValidateStage(ValidateStageRequest) returns (ValidateStageResponse) {} |
| |
| // RunStage instructs the executor to run the given stage attempt. |
| // |
| // This is called per attempt to either run the stage to completion (for |
| // *fast* "synchronous" stages) or to quickly dispatch it to execute |
| // elsewhere (for "asynchronous" stages). |
| // |
| // How fast is *fast*? As a rule of thumb, stages which take far less than |
| // 30s can be done synchronously, and stages which take longer than that |
| // should be done asynchronously. If this is not knowable in advance, this |
| // handler format allows the executor to make this decision dynamically on a |
| // per-attempt basis. |
| // |
| // RunStage RPC is generally expected to redo the validation done by |
| // ValidateStage, since the stage parameters may change after the stage is |
| // inserted. If the stage is no longer passing the validation, RunStage should |
| // essentially finish it as failed by calling TurboCIOrchestrator.WriteNodes |
| // (see below). |
| // |
| // Expected behavior of the handler is that it will call WriteNodes at least |
| // once before the RPC deadline to update the current attempt state: |
| // * Synchronous stages - COMPLETE/INCOMPLETE. |
| // * Asynchronous stages - SCHEDULED/RUNNING/TEARING_DOWN. |
| // * Over capacity/backoff - THROTTLED. |
| // |
| // The RunStage return value and even status code are essentially ignored. |
| // The orchestrator *always* looks at the state of the stage attempt after |
| // RunStage finishes (successfully or not) to decide if the request succeeded |
| // or should be retried. This is needed to unify handling of synchronous and |
| // asynchronous stages and to simplify handling of a class of race conditions |
| // related to RPC peers (or network between them) dying midway through |
| // execution. |
| // |
| // # Interaction with stage cancellation |
| // |
| // A stage attempt can be cancelled at any moment before or during RunStage |
| // execution. |
| // |
| // If it is cancelled before RunStage switches the attempt into SCHEDULED or |
| // RUNNING state, WriteNodes RPCs (e.g. calls to switch the attempt into |
| // RUNNING state) will return FAILED_PRECONDITION error and the gRPC status |
| // details will contain StageAttemptCurrentState with INCOMPLETE `state` |
| // and `cancelled_at` populated. |
| // |
| // If the stage attempt is cancelled after RunStage switches it to SCHEDULED |
| // or RUNNING state, the orchestrator will switch the attempt into CANCELLING |
| // state and will call CancelStage RPC to notify the executor the cancellation |
| // is happening (this is most useful for SCHEDULED attempts that just sit idle |
| // in some queue). The executor can also passively check if the attempt is |
| // CANCELLING now by looking at `current_attempt_state` in WriteNodes RPC |
| // response (useful for running attempts that periodically send heartbeats by |
| // calling WriteNodes with `current_attempt` set). |
| // |
| // Either way the executor must acknowledge the cancellation by switching |
| // the attempt into TEARING_DOWN state (or any terminal states). This can |
| // happen either inside CancelStage implementation or in the executor's |
| // run loop. The orchestrator will keep calling CancelStage as long as the |
| // attempt is in CANCELLING state, up to `cancelling` timeout, after which the |
| // attempt will be marked as INCOMPLETE. |
| // |
| // If there's no `cancelling` timeout, then cancelling a stage will |
| // immediately transition the attempt into INCOMPLETE state (CancelStage will |
| // not be called). The executor can discover this happened by examining |
| // FAILED_PRECONDITION error details as explained above. |
| // |
| // |
| // Examples: |
| // |
| // # Fast Synchronous stage - single write: |
| // |
| // This shows an implementation which does some work quickly (e.g. much less |
| // than 30 seconds) and does its writes and state update in a single atomic |
| // write_nodes call. |
| // |
| // If this crashes then the orchestrator will retry the attempt. |
| // |
| // def run_stage(req): |
| // with transaction: |
| // dat = query_nodes(...) |
| // # do some work quickly |
| // write_nodes(current_attempt={COMPLETE}, <other node updates>) |
| // |
| // # Fast Synchronous stage - multiple writes: |
| // |
| // This shows a synchronous implementation which cannot do all its writes |
| // at once. |
| // |
| // If this crashes after the first write, then the orchestrator will mark |
| // this attempt as incomplete (via RUNNING timeout/heartbeat), and if |
| // the stage policy allows, create a new attempt. |
| // |
| // def run_stage(req): |
| // write_nodes(current_attempt={RUNNING}, <other node updates>) |
| // # work |
| // write_nodes(current_attempt={progress}, <other node updates>) |
| // # work |
| // write_nodes( |
| // current_attempt={progress, COMPLETE}, <other node updates>, |
| // ) |
| // |
| // # Asynchronous stage - immediate handoff: |
| // |
| // This shows an asynchronous stage implementation which must take a long |
| // time to do its work, but which will start working immediately so the |
| // run_stage handler can coordinate directly with the worker. |
| // |
| // If this crashes after the first write, then the orchestrator will mark |
| // this attempt as incomplete (via timeout/heartbeat), and if the stage |
| // policy allows, create a new attempt. |
| // |
| // (apologies for the pseudocode; `ch` should be interpreted as a Go channel) |
| // |
| // def run_stage(req): |
| // ch = start_work(req) |
| // <-ch |
| // |
| // def start_work(req): |
| // ch = Channel() |
| // def _worker: |
| // write_nodes(current_attempt={RUNNING, process_id=hostname+threadid}) |
| // |
| // # Unblock `run_stage`; it can return now that the attempt is |
| // # RUNNING. |
| // close(ch) |
| // |
| // final_update = None |
| // try: |
| // # do a bunch of work |
| // # set final_update to COMPLETE/INCOMPLETE write_nodes call. |
| // finally: |
| // # example of cleanup work |
| // write_nodes(current_attempt=TEARING_DOWN) |
| // write_nodes(final_update) |
| // |
| // run_in_background(_worker) |
| // return ch |
| // |
| // # Asynchronous stage - delayed handoff: |
| // |
| // This shows an asynchronous stage which the executor will enqueue and run |
| // later. |
| // |
| // If this crashes after the first write, then the orchestrator will mark |
| // this attempt as incomplete (via timeout/heartbeat), and if the stage |
| // policy allows, create a new attempt. |
| // |
| // def run_stage(req): |
| // send_work_to_gce(req) |
| // write_nodes(current_attempt=SCHEDULED) |
| // |
| // # sometime possibly much later |
| // def gce_worker(req): |
| // write_nodes(current_attempt={RUNNING, process_id=hostname+process_id}) |
| // # NOTE: in the case where work is multiply-enqueued, the process_id |
| // # during the transition to RUNNING should ensure that only one of |
| // # these write_nodes calls succeeds. |
| // final_update = None |
| // try: |
| // # do a bunch of work |
| // # set final_update to COMPLETE/INCOMPLETE write_nodes call. |
| // finally: |
| // # example of cleanup work |
| // write_nodes(current_attempt=TEARING_DOWN) |
| // write_nodes(final_update) |
| // |
| // # Hybrid stage: |
| // |
| // This shows a stage which is handled dynamically. |
| // |
| // def run_stage(req): |
| // cached = check_cache(req) # fast, no graph writes |
| // if cached: |
| // write_nodes(current_attempt=COMPLETE, ...) |
| // else: |
| // # See 'delayed handoff' above for send_work_to_gce. |
| // send_work_to_gce(req) |
| // write_nodes(current_attempt=SCHEDULED) |
| // |
| // # Executor over capacity |
| // |
| // This shows a stage preamble where the executor needs to implement |
| // backpressure. It can be used in conjunction with any of the other |
| // techniques above. |
| // |
| // When an attempt is THROTTLED, the orchestrator will call RunStage again |
| // later, but not until after the timestamp provided when writing the |
| // THROTTLED state. |
| // |
| // def run_stage(req): |
| // ok, next_check_time = check_capacity() |
| // if not ok: |
| // write_nodes(current_attempt={THROTTLED, until=next_check_time}) |
| // return |
| // |
| // # Any other stage implementation. |
| rpc RunStage(RunStageRequest) returns (RunStageResponse) {} |
| |
| // CancelStage instructs the executor to cancel the given stage attempt. |
| // |
| // It is called for attempts in CANCELLING state soon after they are |
| // cancelled. |
| // |
| // To handle this RPC, the executor may |
| // * Switch the attempt into COMPLETE/INCOMPLETE state if it can do so |
| // (for example, if the stage attempt has not started running yet). |
| // * Switch the attempt into TEARING_DOWN state if cancelling the attempt |
| // requires some additional work. |
| // |
| // CancelStage will be called (with exponential backoff) as long as the |
| // attempt remains in CANCELLING state. |
| // |
| // CancelStage will not be called at all if the StageAttemptExecutionPolicy of |
| // the attempt has no `cancelling` timeout. Such attempts transition into |
| // INCOMPLETE on cancellation immediately. |
| rpc CancelStage(CancelStageRequest) returns (CancelStageResponse) {} |
| } |
