scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-07 03:00:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Simplify pre-start worker failure handling

Browse source 
Move the "which failures can happen before Starting" classification onto
FailureReason as an exhaustive predicate and use it for every
Runnable -> Failed transition, replacing the hand-maintained allowlist.
Give the pending-cancel precedence rule a single owner shared by the
worker launch and worker exit paths.

Test cleanups: share the Notify wait loop, server record fixture, and
post-failure assertions; simplify the pre-start test runtime's hold
flag; and parameterize the slate run.failed payload helper.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Scott Werner 2026-09-03 11:45:47 -04:00
parent 54666632c8
commit c437dc012d
4 changed files with 241 additions and 225 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
lib/apps/fabro-server/src/server.rs
View file

@ -3060,18 +3060,31 @@ struct LiveWorkerProcess {
worker_ref: WorkerRef,
}
fn failure_for_incomplete_run(
/// Pick the terminal failure for a run that never produced its own terminal
/// event. A pending cancel wins over whatever failure the caller observed, so a
/// run that was cancelled while its worker was launching or dying is recorded
/// as cancelled rather than as broken.
fn failure_honoring_pending_cancel(
pending_control: Option<RunControlAction>,
terminated_message: String,
otherwise: impl FnOnce() -> (WorkflowError, FailureReason),
) -> (WorkflowError, FailureReason) {
if pending_control == Some(RunControlAction::Cancel) {
(WorkflowError::Cancelled, FailureReason::Cancelled)
} else {
otherwise()
}
}
fn failure_for_incomplete_run(
pending_control: Option<RunControlAction>,
terminated_message: String,
) -> (WorkflowError, FailureReason) {
failure_honoring_pending_cancel(pending_control, || {
(
WorkflowError::engine(terminated_message),
FailureReason::Terminated,
)
}
})
}
pub(crate) async fn reconcile_incomplete_runs_on_startup(
@ -3585,33 +3598,28 @@ async fn fail_worker_launch(
err: anyhow::Error,
) {
tracing::error!(run_id = %run_id, error = %err, "Failed to spawn worker");
let cancellation_pending = match run_store.state().await {
Ok(run_state) => run_state.pending_control == Some(RunControlAction::Cancel),
let pending_control = match run_store.state().await {
Ok(run_state) => run_state.pending_control,
Err(state_err) => {
tracing::warn!(
run_id = %run_id,
error = %render_compact_with_causes(
&state_err.to_string(),
&collect_causes(&state_err),
),
"Failed to load run state while recording worker launch failure"
error = %state_err,
"Failed to load run state after worker launch failure"
);
false
None
}
};
let (error, reason, message) = if cancellation_pending {
(
WorkflowError::Cancelled,
FailureReason::Cancelled,
"Run cancelled before worker launch completed".to_string(),
)
} else {
let message = format!("Failed to spawn worker: {err}");
let launch_message = format!("Failed to spawn worker: {err}");
let (error, reason) = failure_honoring_pending_cancel(pending_control, || {
(
WorkflowError::engine_with_anyhow("Failed to spawn worker", err),
FailureReason::LaunchFailed,
message,
)
});
let message = if reason == FailureReason::Cancelled {
"Run cancelled before worker launch completed".to_string()
} else {
launch_message
};
let failure_event = workflow_event::Event::workflow_run_failed_from_error(
&error,

256
lib/apps/fabro-server/src/server/tests.rs
View file

@ -2859,14 +2859,7 @@ allowed_usernames = ["octocat"]
.collect::<Vec<_>>()
.join(", ")
);
let runtime_directory = Storage::new(storage_dir).runtime_directory();
ServerDaemon::new(
std::process::id(),
Bind::Tcp("127.0.0.1:32276".parse::<std::net::SocketAddr>().unwrap()),
runtime_directory.log_path(),
)
.write(&runtime_directory)
.unwrap();
write_test_server_record(storage_dir);
let mut server_secret_env: HashMap<String, String> = dev_token
.map(|token| HashMap::from([("FABRO_DEV_TOKEN".to_string(), token)]))
@ -2943,6 +2936,37 @@ fn worker_token_claims(cmd: &Command, state: &AppState) -> crate::worker_token::
.claims
}
fn write_test_server_record(storage_dir: &Path) {
let runtime_directory = Storage::new(storage_dir).runtime_directory();
ServerDaemon::new(
std::process::id(),
Bind::Tcp(
"127.0.0.1:32276"
.parse::<std::net::SocketAddr>()
.expect("test bind should parse"),
),
runtime_directory.log_path(),
)
.write(&runtime_directory)
.expect("test server record should be written");
}
/// Waits up to one second for `condition` to hold, re-checking whenever
/// `notify` fires.
async fn wait_until(notify: &Notify, condition: impl Fn() -> bool, expectation: &str) {
tokio::time::timeout(std::time::Duration::from_secs(1), async {
loop {
let notified = notify.notified();
if condition() {
return;
}
notified.await;
}
})
.await
.expect(expectation);
}
#[derive(Default)]
struct RecordingWorkerRuntime {
requested: StdMutex<Vec<WorkerRef>>,
@ -2968,17 +2992,12 @@ impl RecordingWorkerRuntime {
}
async fn wait_for_forced_ref(&self, worker_ref: &WorkerRef) {
tokio::time::timeout(std::time::Duration::from_secs(1), async {
loop {
let notified = self.forced_notify.notified();
if self.forced_refs().contains(worker_ref) {
return;
}
notified.await;
}
})
.await
.expect("worker should be force-stopped after the cancellation grace period");
wait_until(
&self.forced_notify,
|| self.forced_refs().contains(worker_ref),
"worker should be force-stopped after the cancellation grace period",
)
.await;
}
}
@ -2988,12 +3007,15 @@ enum PreStartWorkerOutcome {
EarlyExit,
}
/// Test worker runtime whose `start` fails before the worker reaches
/// `Starting`, either by refusing to launch or by exiting immediately. When
/// built with `held`, `start` blocks until `release_held_start` so a test can
/// act while the launch is in flight.
struct PreStartWorkerRuntime {
outcome: PreStartWorkerOutcome,
starts: AtomicUsize,
hold_launch_failure: bool,
start_entered: Notify,
release_start: Notify,
outcome: PreStartWorkerOutcome,
starts: AtomicUsize,
start_entered: Notify,
release_start: Option<Notify>,
}
impl PreStartWorkerRuntime {
@ -3001,16 +3023,15 @@ impl PreStartWorkerRuntime {
Self {
outcome,
starts: AtomicUsize::new(0),
hold_launch_failure: false,
start_entered: Notify::new(),
release_start: Notify::new(),
release_start: None,
}
}
fn held_launch_failure() -> Self {
fn held(outcome: PreStartWorkerOutcome) -> Self {
Self {
hold_launch_failure: true,
..Self::new(PreStartWorkerOutcome::LaunchFailure)
release_start: Some(Notify::new()),
..Self::new(outcome)
}
}
@ -3019,21 +3040,19 @@ impl PreStartWorkerRuntime {
}
async fn wait_for_start(&self) {
tokio::time::timeout(std::time::Duration::from_secs(1), async {
loop {
let notified = self.start_entered.notified();
if self.start_count() > 0 {
return;
}
notified.await;
}
})
.await
.expect("test worker runtime should receive one start request");
wait_until(
&self.start_entered,
|| self.start_count() > 0,
"test worker runtime should receive one start request",
)
.await;
}
fn release_held_start(&self) {
self.release_start.notify_one();
self.release_start
.as_ref()
.expect("runtime should have been built with a held start")
.notify_one();
}
}
@ -3042,8 +3061,8 @@ impl WorkerRuntime for PreStartWorkerRuntime {
async fn start(&self, _spec: WorkerLaunchSpec) -> anyhow::Result<StartedWorker> {
self.starts.fetch_add(1, Ordering::Relaxed);
self.start_entered.notify_waiters();
if self.hold_launch_failure {
self.release_start.notified().await;
if let Some(release_start) = &self.release_start {
release_start.notified().await;
}
match self.outcome {
@ -5795,17 +5814,65 @@ fn subprocess_pre_start_failure_state(runtime: StdArc<PreStartWorkerRuntime>) ->
.vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
.worker_runtime(runtime)
.build();
let runtime_directory = Storage::new(state.server_storage_dir()).runtime_directory();
ServerDaemon::new(
std::process::id(),
Bind::Tcp("127.0.0.1:32276".parse().expect("test bind should parse")),
runtime_directory.log_path(),
)
.write(&runtime_directory)
.expect("test server record should be written");
write_test_server_record(&state.server_storage_dir());
state
}
fn run_failed_reasons(events: &[EventEnvelope]) -> Vec<FailureReason> {
events
.iter()
.filter_map(|envelope| match &envelope.event.body {
EventBody::RunFailed(props) => Some(props.failure.reason),
_ => None,
})
.collect()
}
/// Asserts that a run which failed before its worker reached `Starting`
/// recorded exactly one `run.failed` event with `expected_reason` and that the
/// durable and in-memory statuses agree. Returns the run's events for further
/// inspection.
async fn assert_run_failed_before_start(
state: &Arc<AppState>,
run_id: RunId,
expected_reason: FailureReason,
) -> Vec<EventEnvelope> {
let run_store = state
.stores
.runs
.open_run_reader(&run_id)
.await
.expect("failed run should remain readable");
let events = run_store
.list_events()
.await
.expect("failed run events should remain readable");
assert_eq!(run_failed_reasons(&events), vec![expected_reason]);
let expected_status = RunStatus::Failed {
reason: expected_reason,
};
assert_eq!(
run_store
.state()
.await
.expect("failed run state should load")
.status,
expected_status
);
assert_eq!(
state
.runs
.lock()
.expect("runs lock poisoned")
.get(&run_id)
.expect("managed run should remain present")
.status,
expected_status
);
events
}
async fn assert_subprocess_pre_start_failure(
outcome: PreStartWorkerOutcome,
expected_reason: FailureReason,
@ -5836,46 +5903,13 @@ async fn assert_subprocess_pre_start_failure(
execute_run(Arc::clone(&state), run_id).await;
assert_eq!(runtime.start_count(), 1);
let events = run_store
.list_events()
.await
.expect("failed run events should remain readable");
let events = assert_run_failed_before_start(&state, run_id, expected_reason).await;
let lifecycle_events = events
.iter()
.map(|envelope| envelope.event.event_name())
.filter(|name| matches!(*name, "run.runnable" | "run.starting" | "run.failed"))
.collect::<Vec<_>>();
assert_eq!(lifecycle_events, vec!["run.runnable", "run.failed"]);
let failure_reasons = events
.iter()
.filter_map(|envelope| match &envelope.event.body {
EventBody::RunFailed(props) => Some(props.failure.reason),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(failure_reasons, vec![expected_reason]);
let expected_status = RunStatus::Failed {
reason: expected_reason,
};
assert_eq!(
run_store
.state()
.await
.expect("failed run state should load")
.status,
expected_status
);
assert_eq!(
state
.runs
.lock()
.expect("runs lock poisoned")
.get(&run_id)
.expect("managed run should remain present")
.status,
expected_status
);
let response = app
.clone()
@ -5940,7 +5974,9 @@ async fn subprocess_pre_start_failure_persists_early_worker_exit_from_runnable()
#[tokio::test]
async fn subprocess_pre_start_failure_preserves_pending_cancellation() {
let runtime = StdArc::new(PreStartWorkerRuntime::held_launch_failure());
let runtime = StdArc::new(PreStartWorkerRuntime::held(
PreStartWorkerOutcome::LaunchFailure,
));
let state = subprocess_pre_start_failure_state(StdArc::clone(&runtime));
let app = crate::test_support::build_test_router(Arc::clone(&state));
let run_id = create_and_start_run(&app, MINIMAL_DOT)
@ -5967,53 +6003,7 @@ async fn subprocess_pre_start_failure_preserves_pending_cancellation() {
execution.await.expect("run execution task should complete");
assert_eq!(runtime.start_count(), 1);
let run_store = state
.stores
.runs
.open_run_reader(&run_id)
.await
.expect("cancelled run should remain readable");
let events = run_store
.list_events()
.await
.expect("cancelled run events should remain readable");
let failure_reasons = events
.iter()
.filter_map(|envelope| match &envelope.event.body {
EventBody::RunFailed(props) => Some(props.failure.reason),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(failure_reasons, vec![FailureReason::Cancelled]);
assert!(!events.iter().any(|envelope| {
matches!(
&envelope.event.body,
EventBody::RunFailed(props)
if props.failure.reason == FailureReason::LaunchFailed
)
}));
let expected_status = RunStatus::Failed {
reason: FailureReason::Cancelled,
};
assert_eq!(
run_store
.state()
.await
.expect("cancelled run state should load")
.status,
expected_status
);
assert_eq!(
state
.runs
.lock()
.expect("runs lock poisoned")
.get(&run_id)
.expect("managed run should remain present")
.status,
expected_status
);
assert_run_failed_before_start(&state, run_id, FailureReason::Cancelled).await;
}
async fn create_durable_run_with_events(

34
lib/components/fabro-store/src/slate/mod.rs
View file

@ -611,16 +611,16 @@ mod tests {
.unwrap();
}
fn workflow_failure_payload(label: &str) -> EventPayload {
fn failure_payload(label: &str, reason: FailureReason, message: &str) -> EventPayload {
event_payload(
label,
"2026-03-27T12:00:04Z",
"run.failed",
&serde_json::json!({
"failure": {
"reason": "workflow_error",
"reason": reason.to_string(),
"detail": {
"message": "workflow failed",
"message": message,
"category": "deterministic"
}
},
@ -634,26 +634,18 @@ mod tests {
)
}
fn workflow_failure_payload(label: &str) -> EventPayload {
failure_payload(label, FailureReason::WorkflowError, "workflow failed")
}
/// A failure that can only occur after `Starting`, so a `Runnable` run must
/// reject it.
fn sandbox_init_failure_payload(label: &str) -> EventPayload {
event_payload(
assert!(!FailureReason::SandboxInitFailed.can_occur_before_start());
failure_payload(
label,
"2026-03-27T12:00:04Z",
"run.failed",
&serde_json::json!({
"failure": {
"reason": "sandbox_init_failed",
"detail": {
"message": "sandbox initialization failed",
"category": "deterministic"
}
},
"timing": {
"wall_time_ms": 1,
"inference_time_ms": 0,
"tool_time_ms": 0,
"active_time_ms": 0
},
}),
FailureReason::SandboxInitFailed,
"sandbox initialization failed",
)
}

128
lib/foundation/fabro-types/src/status.rs
View file

@ -142,6 +142,12 @@ impl RunStatus {
if self.is_immutable() {
return false;
}
// A worker can fail before it appends `RunStarting`, while the durable run is
// still Runnable. This guards live appends only: replay synthesizes the
// missing intermediate statuses instead of rejecting the event.
if let (Self::Runnable, Self::Failed { reason }) = (self, to) {
return reason.can_occur_before_start();
}
matches!(
(self, to),
(Self::Submitted, Self::Pending { .. } | Self::Runnable)
@ -157,24 +163,10 @@ impl RunStatus {
Self::Submitted
)
| (Self::Pending { .. }, Self::Runnable)
// A worker can fail before it appends RunStarting, while the durable run is still
// Runnable. Keep this set aligned with the audited pre-Starting failure callers.
| (
Self::Runnable,
Self::Starting
| Self::Failed {
reason: FailureReason::LaunchFailed
| FailureReason::WorkflowError
| FailureReason::Terminated
| FailureReason::BootstrapFailed,
}
)
| (
Self::Submitted | Self::Pending { .. } | Self::Runnable,
Self::Failed {
reason: FailureReason::Cancelled,
}
)
| (Self::Runnable, Self::Starting)
| (Self::Submitted | Self::Pending { .. }, Self::Failed {
reason: FailureReason::Cancelled,
})
| (Self::Pending { .. }, Self::Failed {
reason: FailureReason::ApprovalDenied,
})
@ -295,7 +287,17 @@ pub enum SuccessReason {
}
#[derive(
Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Display, EnumString, IntoStaticStr,
Debug,
Clone,
Copy,
PartialEq,
Eq,
Serialize,
Deserialize,
Display,
EnumString,
IntoStaticStr,
VariantArray,
)]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
@ -312,6 +314,32 @@ pub enum FailureReason {
SandboxInitFailed,
}
impl FailureReason {
/// Whether a run can fail for this reason before its worker reaches
/// `Starting`.
///
/// Launch, bootstrap, and engine failures, a worker dying early, and
/// cancellation all happen before the worker appends `RunStarting`.
/// Every other reason implies the run already progressed past
/// `Starting` (sandbox init, publish, budget, transient infra) or
/// belongs to the approval flow.
#[must_use]
pub fn can_occur_before_start(self) -> bool {
match self {
Self::Cancelled
| Self::LaunchFailed
| Self::WorkflowError
| Self::Terminated
| Self::BootstrapFailed => true,
Self::PublishFailed
| Self::ApprovalDenied
| Self::TransientInfra
| Self::BudgetExhausted
| Self::SandboxInitFailed => false,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum TerminalStatus {
@ -360,6 +388,8 @@ pub enum RunControlAction {
mod tests {
use std::str::FromStr;
use strum::VariantArray;
use super::{
BlockedReason, FailureReason, InvalidTransition, PendingReason, RunStatus, SuccessReason,
};
@ -437,45 +467,41 @@ mod tests {
}
#[test]
fn runnable_pre_start_failures_allow_only_audited_reasons() {
for reason in [
fn runnable_accepts_only_failures_that_can_occur_before_start() {
let pre_start = [
FailureReason::Cancelled,
FailureReason::LaunchFailed,
FailureReason::WorkflowError,
FailureReason::Terminated,
FailureReason::BootstrapFailed,
] {
];
for reason in FailureReason::VARIANTS.iter().copied() {
let expected = pre_start.contains(&reason);
assert_eq!(
reason.can_occur_before_start(),
expected,
"{reason} pre-start classification"
);
let failed = RunStatus::Failed { reason };
assert!(
assert_eq!(
RunStatus::Runnable.can_transition_to(failed),
"Runnable should accept {reason} before Starting"
);
assert!(
!RunStatus::Submitted.can_transition_to(failed),
"Submitted should reject {reason}"
);
assert!(
!RunStatus::Pending {
reason: PendingReason::ApprovalRequired,
}
.can_transition_to(failed),
"Pending should reject {reason}"
);
}
assert!(RunStatus::Runnable.can_transition_to(RunStatus::Failed {
reason: FailureReason::Cancelled,
}));
for reason in [
FailureReason::SandboxInitFailed,
FailureReason::PublishFailed,
FailureReason::TransientInfra,
FailureReason::BudgetExhausted,
FailureReason::ApprovalDenied,
] {
assert!(
!RunStatus::Runnable.can_transition_to(RunStatus::Failed { reason }),
"Runnable should reject unrelated failure {reason}"
expected,
"Runnable -> {reason}"
);
if reason != FailureReason::Cancelled {
assert!(
!RunStatus::Submitted.can_transition_to(failed),
"Submitted should reject {reason}"
);
assert_eq!(
RunStatus::Pending {
reason: PendingReason::ApprovalRequired,
}
.can_transition_to(failed),
reason == FailureReason::ApprovalDenied,
"Pending -> {reason}"
);
}
}
}