scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix(server): skip stale worker cleanup after resume

Browse source 
Guard server worker cleanup against superseded subprocesses so rewind and
resume flows do not append a synthetic failure from an older worker. Update
CLI snapshots for the current interview events and give the shared test
session lock more time to cover daemon startup and shutdown.
This commit is contained in:
Bryan Helmkamp 2026-04-07 20:11:17 -04:00
parent 326e0c27fa
commit c5f68d4eda
No known key found for this signature in database
4 changed files with 78 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -878,6 +878,31 @@ fn attach_json_errors_without_prompting_for_human_input() {
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "interview.started",
"id": "[EVENT_ID]",
"node_id": "approve",
"node_label": "approve",
"properties": {
"allow_freeform": false,
"options": [
{
"key": "A",
"label": "[A] Approve"
},
{
"key": "R",
"label": "[R] Revise"
}
],
"question": "Approve?",
"question_id": "[ULID]",
"question_type": "multiple_choice",
"stage": "approve"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
}
]
"#);

37
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -955,6 +955,43 @@ fn json_run_implies_auto_approve_for_human_gates() {
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "interview.started",
"id": "[EVENT_ID]",
"node_id": "approve",
"node_label": "approve",
"properties": {
"allow_freeform": false,
"options": [
{
"key": "A",
"label": "[A] Approve"
},
{
"key": "R",
"label": "[R] Revise"
}
],
"question": "Approve?",
"question_id": "[ULID]",
"question_type": "multiple_choice",
"stage": "approve"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "interview.completed",
"id": "[EVENT_ID]",
"properties": {
"answer": "A",
"duration_ms": "[DURATION_MS]",
"question": "Approve?",
"question_id": "[ULID]"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.completed",
"id": "[EVENT_ID]",

14
lib/crates/fabro-server/src/server.rs
View file

@ -3796,6 +3796,20 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
}
}
let superseded = {
let runs = state.runs.lock().expect("runs lock poisoned");
runs.get(&run_id)
.is_some_and(|managed_run| managed_run.worker_pid != Some(worker_pid))
};
if superseded {
tracing::info!(
run_id = %run_id,
worker_pid,
"Skipping stale worker cleanup for superseded run execution"
);
return;
}
append_worker_exit_failure(&run_store, run_id, &wait_status).await;
let final_state = match run_store.state().await {

3
lib/crates/fabro-test/src/lib.rs
View file

@ -46,6 +46,7 @@ static INSTA_FILTERS: &[(&str, &str)] = &[
const MANAGED_STORAGE_MARKER: &str = "# fabro-test managed storage_dir";
const TEST_IN_MEMORY_STORE_ENV: &str = "FABRO_TEST_IN_MEMORY_STORE";
const SESSION_LOCK_TIMEOUT: Duration = Duration::from_secs(20);
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum TestMode {
@ -267,7 +268,7 @@ fn with_session_lock<T>(root: &Path, f: impl FnOnce() -> T) -> T {
ensure_parent_dir(&lock_path);
let lock_file = File::create(&lock_path)
.unwrap_or_else(|err| panic!("failed to create {}: {err}", lock_path.display()));
let deadline = std::time::Instant::now() + Duration::from_secs(5);
let deadline = std::time::Instant::now() + SESSION_LOCK_TIMEOUT;
while !fabro_proc::try_flock_exclusive(&lock_file)
.unwrap_or_else(|err| panic!("failed to lock {}: {err}", lock_path.display()))
{