From 87436fe33e6ded4f9c1e88d0c38d1961bab9c0ba Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 9 Jul 2026 23:56:53 -0700 Subject: [PATCH 1/2] [adapters] Fix a bug + a race in suspend processing. This commit fixes a severe bug and a less severe race. The bug: when we fail to make a checkpoint while suspending, we log the error, but still set pipeline status to Suspended, making it look like the Suspend succeeded. There are two options for handling this better: (1) go back to the Running state, giving the user a chance to fix the issue and retry (not sure what they can do if the disk runs out of space though...), (2) stop the pipeline and set state to Failed. Like with any other failure, some progress is lost, but at least we don't mask the error. The race: when the checkpoint succeeded, there was a brief window after the checkpoint completes and before we set phase to Suspended when the pipeline's state was Stopped. This caused tests waiting for the Suspended state to fail, since they don't expect the Stopped status. This flaked the cloud `suspend_and_resume_demos` test: a demo reached `Stopped` carrying `PipelineTerminated` instead of suspending cleanly. Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/controller.rs | 15 ++++-- crates/adapters/src/server.rs | 88 +++++++++++++++++++++++++++---- 2 files changed, 90 insertions(+), 13 deletions(-) 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") From d0af8e3df9e51bb9783c5a200144236de0ee63ab Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 10 Jul 2026 14:47:06 -0700 Subject: [PATCH 2/2] [py] Add a python test for a failed suspend operation. Make sure a failed suspend is reported as a pipeline failure. Signed-off-by: Leonid Ryzhyk --- .../tests/platform/test_checkpoint_suspend.py | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) 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)" + )