GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — [connectors] Report the error that stopped the circuit thread in /sta… · feldera/feldera@b843e3d

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
  • crates/connectors/src
    • controller.rs
    • controller
      • stats.rs
      • test.rs
    • server.rs
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎crates/connectors/src/controller.rs‎

Copy file name to clipboardExpand all lines: crates/connectors/src/controller.rs
+12-1Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -944,7 +944,18 @@ impl Controller {
944944
let panicked = run_result.is_err();
945945
match run_result {
946946
Ok(Ok(())) => (),
947-
Ok(Err(error)) => circuit_thread.controller.error(error, None),
947+
Ok(Err(error)) => {
948+
// The circuit cannot continue. Record the
949+
// cause before `finish` terminates the
950+
// pipeline, so that `/status` reports it.
951+
let error = Arc::new(error);
952+
circuit_thread
953+
.controller
954+
.status
955+
.global_metrics
956+
.set_fatal_error(error.clone());
957+
circuit_thread.controller.error(error, None);
958+
}
948959
Err(_) => {
949960
// The panic hook already logged the message
950961
// and backtrace. Without this report, the
Collapse file

‎crates/connectors/src/controller/stats.rs‎

Copy file name to clipboardExpand all lines: crates/connectors/src/controller/stats.rs
+22-1Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ use std::{
7979
collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque},
8080
fmt::Display,
8181
sync::{
82-
Arc, Mutex,
82+
Arc, Mutex, OnceLock,
8383
atomic::{AtomicBool, AtomicU64, Ordering},
8484
},
8585
time::{Duration, Instant},
@@ -183,6 +183,12 @@ pub struct GlobalControllerMetrics {
183183
/// for, while any other termination is a failure.
184184
suspended: AtomicBool,
185185

186+
/// The error that stopped the circuit thread, if one did.
187+
///
188+
/// Lets `/status` report the cause of a [PipelineState::Terminated]
189+
/// pipeline, instead of only the fact that it terminated.
190+
fatal_error: OnceLock<Arc<ControllerError>>,
191+
186192
/// The pipeline has been resumed from a checkpoint and is currently bootstrapping
187193
/// new and modified views.
188194
bootstrap_in_progress: AtomicBool,
@@ -341,6 +347,7 @@ impl GlobalControllerMetrics {
341347
Self {
342348
state: Atomic::new(PipelineState::Paused),
343349
suspended: AtomicBool::new(false),
350+
fatal_error: OnceLock::new(),
344351
bootstrap_in_progress: AtomicBool::new(false),
345352
concurrent_bootstrap_phase: Atomic::new(ConcurrentBootstrapPhase::Inactive),
346353
commit_progress: Mutex::new(None),
@@ -384,6 +391,20 @@ impl GlobalControllerMetrics {
384391
///
385392
/// Records the reason before the state, so that an observer that sees
386393
/// [PipelineState::Terminated] also sees why.
394+
/// The error that stopped the circuit thread, if one did.
395+
pub fn fatal_error(&self) -> Option<Arc<ControllerError>> {
396+
self.fatal_error.get().cloned()
397+
}
398+
399+
/// Records `error` as the cause of the circuit thread stopping. Only the
400+
/// first error is kept.
401+
///
402+
/// Call this before the pipeline reaches [PipelineState::Terminated], so
403+
/// that an observer that sees `Terminated` also sees why.
404+
pub(crate) fn set_fatal_error(&self, error: Arc<ControllerError>) {
405+
let _ = self.fatal_error.set(error);
406+
}
407+
387408
fn set_suspended(&self) {
388409
self.suspended.store(true, Ordering::Release);
389410
self.state
Collapse file

‎crates/connectors/src/controller/test.rs‎

Copy file name to clipboardExpand all lines: crates/connectors/src/controller/test.rs
+73Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2797,6 +2797,79 @@ pub fn replay_does_not_wait_for_endpoint_without_journaled_input(engine: &dyn Pi
27972797
controller.stop().unwrap();
27982798
}
27992799

2800+
/// A replay that reads different input than the journal recorded must stop the
2801+
/// pipeline and record the replay failure as the cause, so that `/status`
2802+
/// reports it instead of a bare `PipelineTerminated`.
2803+
#[engine_test(requires(Storage, FaultTolerance))]
2804+
pub fn replay_failure_is_recorded_as_fatal_error(engine: &dyn PipelineTestEngine) {
2805+
init_test_logger();
2806+
2807+
let tempdir = TempDir::new().unwrap();
2808+
let storage_dir = tempdir.path().join("storage");
2809+
create_dir(&storage_dir).unwrap();
2810+
let input_path = tempdir.path().join("input.csv");
2811+
let config: PipelineConfig = serde_json::from_value(json!({
2812+
"name": "test",
2813+
"workers": 4,
2814+
"storage_config": { "path": storage_dir },
2815+
"storage": true,
2816+
"fault_tolerance": {},
2817+
"clock_resolution_usecs": null,
2818+
"inputs": {
2819+
"test_input1": {
2820+
"stream": "test_input1",
2821+
"transport": {
2822+
"name": "file_input",
2823+
"config": { "path": input_path, "follow": true }
2824+
},
2825+
"format": { "name": "csv" }
2826+
}
2827+
}
2828+
}))
2829+
.unwrap();
2830+
let circuit = identity(&[Some("output")]);
2831+
2832+
// Journal three records, without a checkpoint.
2833+
std::fs::write(&input_path, "1,true,,foo\n2,true,,foo\n3,true,,foo\n").unwrap();
2834+
let controller = engine
2835+
.controller(&circuit, &config, Box::new(|e, _| panic!("error: {e}")))
2836+
.unwrap();
2837+
controller.start();
2838+
wait(
2839+
|| controller.status().num_total_processed_records() == 3,
2840+
default_timeout_ms(),
2841+
)
2842+
.expect("the input was never processed");
2843+
controller.stop().unwrap();
2844+
2845+
// Change the input, with the same length, so that the replay reads
2846+
// different records at the journaled offsets.
2847+
std::fs::write(&input_path, "4,true,,bar\n5,true,,bar\n6,true,,bar\n").unwrap();
2848+
2849+
let errors = Arc::new(Mutex::new(Vec::new()));
2850+
let controller = engine
2851+
.controller(&circuit, &config, {
2852+
let errors = errors.clone();
2853+
Box::new(move |e, _| errors.lock().unwrap().push(e.error_code()))
2854+
})
2855+
.unwrap();
2856+
controller.start();
2857+
wait(
2858+
|| controller.status().global_metrics.get_state() == crate::PipelineState::Terminated,
2859+
default_timeout_ms(),
2860+
)
2861+
.expect("the replay failure did not stop the pipeline");
2862+
2863+
let fatal_error = controller
2864+
.status()
2865+
.global_metrics
2866+
.fatal_error()
2867+
.expect("the pipeline stopped without recording the cause");
2868+
assert_eq!(fatal_error.error_code(), "ReplayFailure");
2869+
assert_eq!(errors.lock().unwrap().as_slice(), ["ReplayFailure"]);
2870+
let _ = controller.stop();
2871+
}
2872+
28002873
/// Returns the names of the ad hoc INSERT input endpoints of `controller`.
28012874
fn adhoc_input_endpoints(controller: &Controller) -> Vec<String> {
28022875
controller
Collapse file

‎crates/connectors/src/server.rs‎

Copy file name to clipboardExpand all lines: crates/connectors/src/server.rs
+43-21Lines changed: 43 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1984,6 +1984,7 @@ async fn status_handler(
19841984
/// from a checkpoint that does not exist.
19851985
fn terminated_status(
19861986
suspended: bool,
1987+
fatal_error: Option<Arc<ControllerError>>,
19871988
runtime_desired_status: RuntimeDesiredStatus,
19881989
storage_status_details: Option<StorageStatusDetails>,
19891990
) -> Result<ExtendedRuntimeStatus, ExtendedRuntimeStatusError> {
@@ -1994,6 +1995,8 @@ fn terminated_status(
19941995
runtime_desired_status,
19951996
storage_status_details,
19961997
})
1998+
} else if let Some(error) = fatal_error {
1999+
Err(controller_error_status(error))
19972000
} else {
19982001
Err(ExtendedRuntimeStatusError {
19992002
status_code: StatusCode::INTERNAL_SERVER_ERROR,
@@ -2006,6 +2009,28 @@ fn terminated_status(
20062009
}
20072010
}
20082011

2012+
/// The `/status` error for a pipeline that `error` stopped.
2013+
fn controller_error_status(error: Arc<ControllerError>) -> ExtendedRuntimeStatusError {
2014+
let mut status_code = error.status_code();
2015+
if status_code == StatusCode::SERVICE_UNAVAILABLE {
2016+
error!(
2017+
"Endpoint /status: unexpected status code {status_code} has been converted into {} for error {error}",
2018+
StatusCode::INTERNAL_SERVER_ERROR
2019+
);
2020+
status_code = StatusCode::INTERNAL_SERVER_ERROR;
2021+
}
2022+
let e = PipelineError::ControllerError { error };
2023+
let e = ErrorResponse::from_error(&e);
2024+
ExtendedRuntimeStatusError {
2025+
status_code,
2026+
error: feldera_types::error::ErrorResponse {
2027+
message: e.message,
2028+
error_code: e.error_code,
2029+
details: e.details,
2030+
},
2031+
}
2032+
}
2033+
20092034
#[allow(clippy::result_large_err)]
20102035
fn get_status(
20112036
state: &ServerState,
@@ -2101,6 +2126,7 @@ fn get_status(
21012126
)),
21022127
PipelineState::Terminated => terminated_status(
21032128
controller.status().global_metrics.suspended(),
2129+
controller.status().global_metrics.fatal_error(),
21042130
runtime_desired_status,
21052131
storage_status_details,
21062132
),
@@ -2177,24 +2203,7 @@ fn get_status(
21772203
storage_status_details,
21782204
})
21792205
} else {
2180-
let mut status_code = e.status_code();
2181-
if status_code == StatusCode::SERVICE_UNAVAILABLE {
2182-
error!(
2183-
"Endpoint /status: unexpected status code {status_code} has been converted into {} for error {e}",
2184-
StatusCode::SERVICE_UNAVAILABLE
2185-
);
2186-
status_code = StatusCode::INTERNAL_SERVER_ERROR;
2187-
}
2188-
let e = PipelineError::ControllerError { error: e };
2189-
let e = ErrorResponse::from_error(&e);
2190-
Err(ExtendedRuntimeStatusError {
2191-
status_code,
2192-
error: feldera_types::error::ErrorResponse {
2193-
message: e.message,
2194-
error_code: e.error_code,
2195-
details: e.details,
2196-
},
2197-
})
2206+
Err(controller_error_status(e))
21982207
}
21992208
}
22002209
PipelinePhase::Suspended => Ok(ExtendedRuntimeStatus {
@@ -5728,16 +5737,29 @@ mod test_status {
57285737
/// `PipelineTerminated` error that flaked `suspend_and_resume_demos`.
57295738
#[test]
57305739
fn terminated_status_reports_suspended_while_suspending() {
5731-
let status = terminated_status(true, RuntimeDesiredStatus::Suspended, None)
5740+
let status = terminated_status(true, None, RuntimeDesiredStatus::Suspended, None)
57325741
.expect("a suspending pipeline must not surface PipelineTerminated");
57335742
assert!(matches!(status.runtime_status, RuntimeStatus::Suspended));
57345743

57355744
// An unexpected termination stays a fatal error.
5736-
let error = terminated_status(false, RuntimeDesiredStatus::Running, None)
5745+
let error = terminated_status(false, None, RuntimeDesiredStatus::Running, None)
57375746
.expect_err("an unexpected termination must remain PipelineTerminated");
57385747
assert_eq!(error.error.error_code.as_ref(), "PipelineTerminated");
57395748
}
57405749

5750+
/// A pipeline that an error stopped must report that error, not only that
5751+
/// it terminated.
5752+
#[test]
5753+
fn terminated_status_reports_the_fatal_error() {
5754+
let error = Arc::new(ControllerError::ReplayFailure {
5755+
error: "replayed transaction 1 ingested different input".to_string(),
5756+
});
5757+
let status = terminated_status(false, Some(error), RuntimeDesiredStatus::Running, None)
5758+
.expect_err("a pipeline that an error stopped must report an error");
5759+
assert_eq!(status.error.error_code.as_ref(), "ReplayFailure");
5760+
assert!(status.error.message.contains("ingested different input"));
5761+
}
5762+
57415763
/// A circuit that dies while a suspend is pending must not be reported as a
57425764
/// clean suspend.
57435765
///
@@ -5750,7 +5772,7 @@ mod test_status {
57505772
/// handler may produce it.
57515773
#[test]
57525774
fn terminated_status_reports_a_death_during_suspend_as_an_error() {
5753-
let error = terminated_status(false, RuntimeDesiredStatus::Suspended, None)
5775+
let error = terminated_status(false, None, RuntimeDesiredStatus::Suspended, None)
57545776
.expect_err("a pipeline that died while suspending must not report a clean suspend");
57555777
assert_eq!(error.error.error_code.as_ref(), "PipelineTerminated");
57565778
}

0 commit comments

Comments
0 (0)
You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.