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

Preserve committed outcomes during host failure cleanup

Browse source
This commit is contained in:
Scott Werner 2026-10-05 17:08:57 -04:00
parent ba215111f4
commit 372018cd87
7 changed files with 149 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
lib/apps/fabro-cli/src/commands/run/attach.rs
View file

@ -1034,6 +1034,7 @@ mod tests {
let server = MockServer::start_async().await;
let mut state = terminal_run_state_response(run_id);
state["status"] = serde_json::json!({"kind": "running"});
state["conclusion"] = serde_json::Value::Null;
let state: server_client::RunProjection = serde_json::from_value(state).unwrap();
let executed = serde_json::json!({
"run_id": run_id, "stream_seq": 1, "kind": "petri", "id": "executed",
@ -1097,7 +1098,6 @@ mod tests {
!task.is_finished(),
"successful execution does not end attach while publication is pending"
);
waiting.delete_async().await;
let finished = serde_json::json!({
"run_id": run_id, "stream_seq": 2, "kind": "petri", "id": "finished",
"recorded_at": 2000,
@ -1114,6 +1114,7 @@ mod tests {
.body(format!("data: {finished}\n\n"));
})
.await;
waiting.delete_async().await;
assert_eq!(
tokio::time::timeout(Duration::from_secs(5), task)
.await

68
lib/apps/fabro-server/src/petri_runs.rs
View file

@ -1086,6 +1086,51 @@ mod tests {
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_rejected_platform_failure_does_not_settle_the_run() {
let runtime = Arc::new(HeldWorkerRuntime::default());
let (state, app, run_id, token) = held_worker_run(&runtime).await;
run_to_running_as_worker(&app, run_id, &token).await;
let pool = state.stores.run_summaries.pool();
sqlx::query(
"CREATE TRIGGER reject_platform_record BEFORE INSERT ON platform_records \
BEGIN SELECT RAISE(FAIL, 'scripted append failure'); END",
)
.execute(&pool)
.await
.unwrap();
let response = append_lifecycle_as_worker(
&app,
run_id,
&token,
RunLifecycleKind::Failed,
RunStatus::Failed {
reason: FailureReason::LaunchFailed,
},
)
.await;
fabro_test::assert_axum_status(
response,
StatusCode::INTERNAL_SERVER_ERROR,
"rejected platform terminal append",
)
.await;
assert_public_result(&state, &app, run_id, RunStatus::Running, None).await;
crate::server::persist_run_failure(
&state,
run_id,
FailureReason::LaunchFailed,
"worker launch failed".into(),
)
.await;
assert_public_result(&state, &app, run_id, RunStatus::Running, None).await;
sqlx::query("DROP TRIGGER reject_platform_record")
.execute(&pool)
.await
.unwrap();
runtime.end_worker();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn required_publication_result_agrees_across_worker_api_projection_and_cleanup() {
use fabro_petri::petri::LogId;
@ -1177,6 +1222,29 @@ mod tests {
}
assert_eq!(concluded_runs(&app).await, 1);
assert_public_result(&state, &app, run_id, expected, rejection).await;
// A host error arriving after cleanup must neither append a
// competing terminal record nor replace the useful failure.
let platform_count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM platform_records WHERE run_id = ?")
.bind(run_id.to_string())
.fetch_one(&state.stores.run_summaries.pool())
.await
.unwrap();
crate::server::persist_run_failure(
&state,
run_id,
FailureReason::Terminated,
"worker wait failed during teardown".into(),
)
.await;
assert_public_result(&state, &app, run_id, expected, rejection).await;
let after_count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM platform_records WHERE run_id = ?")
.bind(run_id.to_string())
.fetch_one(&state.stores.run_summaries.pool())
.await
.unwrap();
assert_eq!(after_count, platform_count);
let (rebuilt, _, _) =
fabro_petri::test_support::rebuild(&state.db_pool, &state.db_pool, run_id)
.await

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

@ -3503,13 +3503,55 @@ async fn fail_run_before_execution(
reason: FailureReason,
message: String,
) {
if let Err(err) =
run_records::lifecycle(state, run_id, run_records::failed(reason, message.clone())).await
{
error!(run_id = %run_id, error = %err, "Failed to persist run failure status");
}
persist_run_failure(state, run_id, reason, message).await;
}
fail_managed_run(state, run_id, reason, message);
/// Record a host failure only while no terminal result is committed. This
/// also handles a worker wait/launch error racing its durable Petri finish.
pub(crate) async fn persist_run_failure(
state: &Arc<AppState>,
run_id: RunId,
reason: FailureReason,
message: String,
) {
match run_records::projection(state, run_id).await {
Ok(Some(projection)) if projection.status.is_terminal() => {
let failure = projection
.conclusion
.as_ref()
.and_then(|conclusion| conclusion.failure.as_ref())
.map(|failure| failure.detail.message.clone());
settle_managed_run_at_finish(state, run_id, projection.status, failure);
}
Ok(Some(_)) => {
match run_records::lifecycle(
state,
run_id,
run_records::failed(reason, message.clone()),
)
.await
{
Ok(_) => fail_managed_run(state, run_id, reason, message),
Err(err) => {
error!(run_id = %run_id, error = %err, "Failed to persist run failure status");
}
}
}
Ok(None) => {
error!(run_id = %run_id, "Run missing when recording a host failure");
}
Err(err) => {
error!(run_id = %run_id, error = %err, "Failed to read the committed run result");
}
}
// Resource cleanup is operational; it cannot substitute for a committed
// outcome if storage is unavailable.
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
clear_live_run_state(managed_run);
}
drop(runs);
cleanup_worker_control_bus_for_run(state, run_id);
state.scheduler_notify.notify_one();
}
@ -3764,26 +3806,13 @@ async fn fail_worker_launch(state: &Arc<AppState>, run_id: RunId, err: anyhow::E
None
}
};
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,
)
});
let message = if reason == FailureReason::Cancelled {
"Run cancelled before worker launch completed".to_string()
} else {
launch_message
};
let _ = run_records::lifecycle(
state,
run_id,
run_records::failed(reason, error.to_string()),
)
.await;
fail_managed_run(state, run_id, reason, message);
state.scheduler_notify.notify_one();
persist_run_failure(state, run_id, reason, error.to_string()).await;
}
/// A worker that exited without recording the run's end left it failed.
@ -4249,15 +4278,17 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
Err(err) => {
tracing::error!(run_id = %run_id, error = %err, "Failed while waiting on worker");
let message = format!("Worker wait failed: {err}");
let superseded = {
let runs = state.runs.lock().expect("runs lock poisoned");
runs.get(&run_id)
.is_some_and(|run| run.worker_ref.as_ref() != Some(&worker_ref))
};
if superseded {
return;
}
state.worker_runtime.force_stop(&worker_ref).await;
let _ = run_records::lifecycle(
&state,
run_id,
run_records::failed(FailureReason::Terminated, message.clone()),
)
.await;
fail_managed_run(&state, run_id, FailureReason::Terminated, message);
state.scheduler_notify.notify_one();
state.petri_runs.worker_exited(run_id);
persist_run_failure(&state, run_id, FailureReason::Terminated, message).await;
return;
}
};

14
lib/apps/fabro-server/src/server/petri_runs.rs
View file

@ -488,8 +488,8 @@ pub(crate) async fn execute(state: Arc<AppState>, run_id: RunId) {
run_id: run_id.to_string(),
run_dir: run_dir.join("petri"),
execution,
// The projector's signal follows each durable append; the managed
// run's settle at Petri's finish precedes it.
// The coordinator finish is stored before managed status settles;
// the projector also reads only durable records.
store: Arc::new(SettlingStore {
inner: state
.petri_projector
@ -534,7 +534,7 @@ pub(crate) async fn execute(state: Arc<AppState>, run_id: RunId) {
}
};
match run_records::lifecycle(&state, run_id, record).await {
Ok(()) => finish(&state, run_id, status, error),
Ok(_) => finish(&state, run_id, status, error),
Err(err) => {
error!(run_id = %run_id, error = %err, "Failed to persist run outcome");
release_live_state(&state, run_id);
@ -669,7 +669,7 @@ async fn fail_before_execution(state: &Arc<AppState>, run_id: RunId, message: &s
error!(run_id = %run_id, error = message, "Petri run cannot start");
let (status, error, record) = failed(FailureReason::WorkflowError, message.to_string());
match run_records::lifecycle(state, run_id, record).await {
Ok(()) => finish(state, run_id, status, error),
Ok(_) => finish(state, run_id, status, error),
Err(err) => {
error!(run_id = %run_id, error = %err, "Failed to persist run failure status");
release_live_state(state, run_id);
@ -679,9 +679,9 @@ async fn fail_before_execution(state: &Arc<AppState>, run_id: RunId, message: &s
/// Settle the managed run at its terminal record and release its
/// scheduler slot. A run that Petri finished settled already, at the
/// `run.finished` record ([`SettlingStore`]); this refines its status and
/// error and ends its live state. A run deleted since is gone from the map
/// and stays gone.
/// `run.finished` record ([`SettlingStore`]); this preserves that status,
/// fills any missing failure detail and ends its live state. A run deleted
/// since is gone from the map and stays gone.
fn finish(state: &Arc<AppState>, run_id: RunId, status: RunStatus, error: Option<String>) {
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {

6
lib/components/fabro-petri/src/engine.rs
View file

@ -137,7 +137,7 @@ pub enum RunStatus {
#[derive(Clone, Debug)]
pub struct RunOutcome {
pub status: RunStatus,
/// The root invocation's failure message, when it failed.
/// Required-finalization failure detail, or the root execution's failure.
pub failure: Option<String>,
/// Whether the record is whole: the run recorded its finish and every
/// log replays byte for byte.
@ -365,8 +365,8 @@ pub async fn outcome_of(store: &dyn RunStore, run_id: &str) -> Result<RunOutcome
/// How Fabro reports what [`run`] returned. A cancelled run is a failure
/// with the cancelled reason, as the legacy executor reports one; every
/// other shortfall is a workflow error whose message says what the record,
/// or the host, said.
/// publication rejection keeps its publish_failed reason. Other shortfalls
/// are workflow errors whose message says what the record, or the host, said.
#[must_use]
pub fn conclusion(result: &Result<RunOutcome, RunError>) -> Conclusion {
match result {

10
lib/components/fabro-petri/src/hooks.rs
View file

@ -53,16 +53,18 @@
//!
//! # Operation identities
//!
//! Every external effect here is keyed on `(run key, execution, DecisionId,
//! effect kind)` from the hook context and deduplicated on retry: the
//! checkpoint's key is the attempt's decision in its execution, effect
//! Checkpoint and artifact effects are keyed on `(run key, execution,
//! DecisionId, effect kind)` from the hook context and deduplicated on retry:
//! the checkpoint's key is the attempt's decision in its execution, effect
//! `checkpoint`; an artifact's is the same decision, effect `artifact`, with
//! the file's path and content digest as the identity within it. A
//! re-dispatched attempt whose commit already landed reuses it when the
//! workspace still sits on it unchanged (see [`RunWorkspaces::commit`]); a
//! reissued routing decision finds the record, or the commit by its
//! trailers, and writes nothing twice; a file already collected under the
//! same path and digest is not collected again.
//! same path and digest is not collected again. Publication keeps the
//! publisher's reconciliation policy; required finalization adds no independent
//! effect ledger or guarantee of deduplication across every external crash.
//!
//! # Where the workspace is
//!

7
lib/components/fabro-petri/src/test_support/finalization.rs
View file

@ -10,6 +10,7 @@ use petri_runtime::frontend::CompileInputs;
use petri_runtime::ir::FinalizationFailure;
use petri_runtime::{RunOptions, Runtime};
use petri_store::{Access, LogId, MemoryRunStore, Record, RunKey, RunStore as _};
use tokio::fs;
use tokio::sync::{Notify, Semaphore};
use crate::providers::{self, SandboxProviderConfig};
@ -76,7 +77,7 @@ pub async fn test_run_records(
) -> TestRunRecords {
let root = tempfile::tempdir().expect("the fixture has an isolated directory");
let workflow = root.path().join("workflow.fabro");
tokio::fs::write(
fs::write(
&workflow,
r#"digraph Finalization {
graph [goal="Check required publication"]
@ -88,7 +89,7 @@ pub async fn test_run_records(
)
.await
.expect("the fixture workflow writes");
tokio::fs::write(
fs::write(
root.path().join("workflow.toml"),
"_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n",
)
@ -129,7 +130,7 @@ pub async fn test_run_records(
)];
for execution in inspection.executions {
let id = LogId::Execution(execution.execution);
records.push((id.clone(), logs.read(&id).await.expect("execution reads")));
records.push((id, logs.read(&id).await.expect("execution reads")));
}
let mut blobs = Vec::new();
for graph in inspection.graphs {