diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index 298f14760b1..e0a99ec77f5 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -3565,9 +3565,18 @@ impl CircuitThread { CheckpointRequest::Scheduled => (), CheckpointRequest::CheckpointCommand(callback) => callback(result.clone()), CheckpointRequest::SuspendCommand(callback) => { - self.controller.status.set_state(PipelineState::Terminated); - if let Err(e) = &result { - self.controller.error(e.clone(), None); + // Terminate the circuit only on a *successful* suspend. A + // failed suspend leaves the circuit intact and running; the + // `/suspend` handler reports it as `PipelinePhase::Failed` + // and stops the pipeline. Not marking `Terminated` here is + // what keeps `/status` from masking the failure as a clean + // `Suspended` during teardown: `get_status` reads the live + // controller state before the phase, so a terminated + // controller with desired status `Suspended` always reads + // back as `Suspended` (see `terminated_status`). + match &result { + Ok(_) => self.controller.status.set_state(PipelineState::Terminated), + Err(e) => self.controller.error(e.clone(), None), } callback(result.clone().map(|_| ())) } diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs index 868b5431221..e6f821fcfa5 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -1419,6 +1419,47 @@ async fn status_handler( get_status(&state) } +#[allow(clippy::result_large_err)] +/// Reports the runtime status of a controller whose circuit has reached +/// [`PipelineState::Terminated`]. +/// +/// A successful suspend terminates the circuit before the server installs +/// [`PipelinePhase::Suspended`] and deallocates the controller (see `/suspend`), +/// so a status poll landing in that window observes the still-registered +/// controller in `Terminated`. When a suspend was requested the pipeline is +/// converging to `Suspended`, so report that clean status rather than a +/// spurious `PipelineTerminated` error that the pipeline manager would record +/// as a failed execution. +/// +/// This branch is reached only for a *successful* suspend: a failed suspend +/// deliberately leaves the circuit running (see the `SuspendCommand` handler in +/// `controller.rs`) so that it never masquerades here as a clean `Suspended`; +/// the `/suspend` handler reports it as [`PipelinePhase::Failed`] instead. Any +/// termination without a suspend request is an unexpected, fatal termination +/// and stays an error. +fn terminated_status( + runtime_desired_status: RuntimeDesiredStatus, + storage_status_details: Option, +) -> Result { + if matches!(runtime_desired_status, RuntimeDesiredStatus::Suspended) { + Ok(ExtendedRuntimeStatus { + runtime_status: RuntimeStatus::Suspended, + runtime_status_details: json!(""), + runtime_desired_status, + storage_status_details, + }) + } else { + Err(ExtendedRuntimeStatusError { + status_code: StatusCode::INTERNAL_SERVER_ERROR, + error: feldera_types::error::ErrorResponse { + message: "Pipeline has been terminated.".to_string(), + error_code: Cow::from("PipelineTerminated"), + details: json!({}), + }, + }) + } +} + #[allow(clippy::result_large_err)] fn get_status(state: &ServerState) -> Result { // Runtime desired status @@ -1476,14 +1517,9 @@ fn get_status(state: &ServerState) -> Result Err(ExtendedRuntimeStatusError { - status_code: StatusCode::INTERNAL_SERVER_ERROR, - error: feldera_types::error::ErrorResponse { - message: "Pipeline has been terminated.".to_string(), - error_code: Cow::from("PipelineTerminated"), - details: json!({}), - }, - }), + PipelineState::Terminated => { + terminated_status(runtime_desired_status, storage_status_details) + } }; } Err(_) => { @@ -2135,10 +2171,12 @@ async fn suspend(state: WebData) -> Result) { + let mut suspend_error = None; loop { if let Ok(controller) = state.controller() { if let Err(error) = controller.async_suspend().await { error!("controller suspend failed ({error})"); + suspend_error = Some(error); } break; } @@ -2151,7 +2189,17 @@ async fn suspend(state: WebData) -> Result PipelinePhase::Failed(error), + None => PipelinePhase::Suspended, + }); if let Ok(controller) = state.take_controller() && let Err(error) = controller.async_stop().await { @@ -3146,7 +3194,7 @@ impl StoredStatus { /// off. #[cfg(test)] mod test_http_helpers { - use super::{ServerArgs, ServerState, bootstrap, build_app, parse_config}; + use super::{ServerArgs, ServerState, bootstrap, build_app, parse_config, terminated_status}; use crate::{ controller::ControllerBuilder, server::{InitializationState, PipelinePhase}, @@ -3181,6 +3229,26 @@ mod test_http_helpers { use tempfile::NamedTempFile; use uuid::Uuid; + /// A successful suspend terminates the circuit before the server reports + /// `PipelinePhase::Suspended`, so `terminated_status` must report a clean + /// `Suspended` while a suspend is in progress instead of the spurious + /// `PipelineTerminated` error that flaked `suspend_and_resume_demos`. + /// + /// A *failed* suspend never reaches `terminated_status`: the `SuspendCommand` + /// handler leaves the circuit running instead of terminating it, and the + /// `/suspend` handler reports the failure as `PipelinePhase::Failed`. + #[test] + fn terminated_status_reports_suspended_while_suspending() { + let status = terminated_status(RuntimeDesiredStatus::Suspended, None) + .expect("a suspending pipeline must not surface PipelineTerminated"); + assert!(matches!(status.runtime_status, RuntimeStatus::Suspended)); + + // An unexpected termination (no suspend requested) stays a fatal error. + let error = terminated_status(RuntimeDesiredStatus::Running, None) + .expect_err("an unexpected termination must remain PipelineTerminated"); + assert_eq!(error.error.error_code.as_ref(), "PipelineTerminated"); + } + pub(super) async fn get_stats(server: &TestServer) -> ExternalControllerStatus { server .get("/stats") diff --git a/python/tests/platform/test_checkpoint_suspend.py b/python/tests/platform/test_checkpoint_suspend.py index a0470b5b43e..945569140aa 100644 --- a/python/tests/platform/test_checkpoint_suspend.py +++ b/python/tests/platform/test_checkpoint_suspend.py @@ -278,3 +278,70 @@ def no_new_records_for_5s(): assert _adhoc_count(pipeline_name) == final_count, ( "Received new records after all connectors reached EOI" ) + + +@enterprise_only +@gen_pipeline_name +def test_suspend_failure_enterprise(pipeline_name): + """ + A suspend (`/stop?force=false`) whose checkpoint fails must be reported as a + failure, not masked as a clean suspend. + + This is the failure counterpart to `test_suspend_enterprise` (which covers + the happy path). It needs its own pipeline because the trigger is a + storage-disabled config: with storage off, `can_suspend` fails permanently + with `StorageRequired`, so the suspend checkpoint cannot succeed. The + manager forwards `/suspend` to the pipeline without pre-checking + suspendability, the checkpoint fails, and the pipeline must be forcefully + stopped with the error recorded in `deployment_error`. + + Regression: the adapter used to report the terminated circuit as a clean + `Suspended` whenever a suspend was desired (see `terminated_status` in + `crates/adapters/src/server.rs`), so a failed suspend reached `Stopped` + with no `deployment_error`. With the fix, a failed suspend leaves the + circuit running and the `/suspend` handler reports `PipelinePhase::Failed`, + so the failure surfaces as a `deployment_error`. + """ + sql = "CREATE TABLE t1(x int) WITH ('materialized' = 'true');" + create_pipeline(pipeline_name, sql) + + # Disable storage so the suspend checkpoint fails permanently. + resp = patch_json( + api_url(f"/pipelines/{pipeline_name}"), + { + "runtime_config": { + "workers": FELDERA_TEST_NUM_WORKERS, + "hosts": FELDERA_TEST_NUM_HOSTS, + "logging": "debug", + "storage": False, + } + }, + ) + assert resp.status_code == HTTPStatus.OK, resp.text + + start_pipeline(pipeline_name) + + # Request a suspend. The manager forwards `/suspend` to the pipeline; the + # checkpoint then fails because storage is disabled. + resp = post_no_body(api_url(f"/pipelines/{pipeline_name}/stop?force=false")) + assert resp.status_code == HTTPStatus.ACCEPTED, (resp.status_code, resp.text) + + # The failed suspend forcefully stops the pipeline. + def deployment(): + d = get(api_url(f"/pipelines/{pipeline_name}")).json() + return d.get("deployment_status"), d.get("deployment_error") + + wait_for_condition( + "pipeline stopped after failed suspend", + lambda: deployment()[0] == "Stopped", + timeout_s=60.0, + poll_interval_s=0.5, + ) + + # The failure must be recorded, not masked as a clean suspend. The old + # (buggy) code reached "Stopped" with deployment_error == None. + _, deployment_error = deployment() + assert deployment_error is not None, ( + "a failed suspend must be recorded as a deployment error, not masked " + "as a clean suspend (deployment_error is None)" + )