huskies: merge 1170 bug Full tokio runtime stall after unblock → merge auto-assign
This commit is contained in:
@@ -65,9 +65,22 @@ impl AgentPool {
|
|||||||
// AC6: Detect empty-diff stories before starting the merge pipeline.
|
// AC6: Detect empty-diff stories before starting the merge pipeline.
|
||||||
// If the worktree has no commits on the feature branch, block the
|
// If the worktree has no commits on the feature branch, block the
|
||||||
// story immediately via the state machine — no merge job needed.
|
// story immediately via the state machine — no merge job needed.
|
||||||
if let Some(wt_path) = worktree::find_worktree_path(project_root, story_id)
|
//
|
||||||
&& !crate::agents::gates::worktree_has_committed_work(&wt_path)
|
// Bug 1170: worktree_has_committed_work shells out to `git log`
|
||||||
{
|
// synchronously. assign_merge_stage runs on the shared tokio
|
||||||
|
// runtime (it's invoked reactively on every CRDT transition, incl.
|
||||||
|
// unblock), so calling it inline here blocked a runtime worker
|
||||||
|
// thread for the duration of the git subprocess — with a story
|
||||||
|
// whose worktree/agent had crashed, that call could hang
|
||||||
|
// indefinitely and stall /health and the liveness heartbeat along
|
||||||
|
// with it. Run it on the blocking-thread pool instead.
|
||||||
|
if let Some(wt_path) = worktree::find_worktree_path(project_root, story_id) {
|
||||||
|
let has_commits = tokio::task::spawn_blocking(move || {
|
||||||
|
crate::agents::gates::worktree_has_committed_work(&wt_path)
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap_or(false);
|
||||||
|
if !has_commits {
|
||||||
let empty_diff_reason = "Feature branch has no code changes — the coder agent \
|
let empty_diff_reason = "Feature branch has no code changes — the coder agent \
|
||||||
did not produce any commits.";
|
did not produce any commits.";
|
||||||
slog_warn!(
|
slog_warn!(
|
||||||
@@ -77,7 +90,9 @@ impl AgentPool {
|
|||||||
if let Err(e) =
|
if let Err(e) =
|
||||||
crate::agents::lifecycle::transition_to_blocked(story_id, empty_diff_reason)
|
crate::agents::lifecycle::transition_to_blocked(story_id, empty_diff_reason)
|
||||||
{
|
{
|
||||||
slog_error!("[auto-assign] Failed to transition '{story_id}' to Blocked: {e}");
|
slog_error!(
|
||||||
|
"[auto-assign] Failed to transition '{story_id}' to Blocked: {e}"
|
||||||
|
);
|
||||||
}
|
}
|
||||||
let _ = self
|
let _ = self
|
||||||
.watcher_tx
|
.watcher_tx
|
||||||
@@ -87,6 +102,7 @@ impl AgentPool {
|
|||||||
});
|
});
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Skip if a merge job is already running for this story (e.g. triggered
|
// Skip if a merge job is already running for this story (e.g. triggered
|
||||||
// by a previous auto-assign pass or by pipeline advancement).
|
// by a previous auto-assign pass or by pipeline advancement).
|
||||||
@@ -111,3 +127,89 @@ impl AgentPool {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::super::super::AgentPool;
|
||||||
|
use crate::config::ProjectConfig;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
|
|
||||||
|
/// Bug 1170 regression: `assign_merge_stage` used to call
|
||||||
|
/// `worktree_has_committed_work` (which shells out to `git`) directly on
|
||||||
|
/// the async runtime. For a story with a crashed/unassignable agent
|
||||||
|
/// sitting in `4_merge/`, that synchronous subprocess call had no yield
|
||||||
|
/// point, so on a runtime with few worker threads it starved every other
|
||||||
|
/// task — including the liveness heartbeat and `/health` — for the whole
|
||||||
|
/// scan. After wrapping the call in `spawn_blocking`, the executor stays
|
||||||
|
/// free to interleave other work while the git subprocess runs
|
||||||
|
/// off-runtime.
|
||||||
|
///
|
||||||
|
/// This reproduces the unblock → merge-auto-assign path: a story sits in
|
||||||
|
/// `4_merge/` with no active agent entry (the crashed/unassignable case)
|
||||||
|
/// and a worktree directory that isn't a real git repo, forcing every
|
||||||
|
/// `git` invocation in the scan to fail — but only after paying the
|
||||||
|
/// process fork/exec cost, which is what stalls a non-yielding runtime.
|
||||||
|
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
|
||||||
|
async fn assign_merge_stage_does_not_stall_liveness_heartbeat() {
|
||||||
|
crate::db::ensure_content_store();
|
||||||
|
|
||||||
|
let tmp = tempfile::tempdir().unwrap();
|
||||||
|
let root = tmp.path().to_path_buf();
|
||||||
|
std::fs::create_dir_all(root.join(".huskies")).unwrap();
|
||||||
|
std::fs::write(root.join(".huskies/project.toml"), "").unwrap();
|
||||||
|
|
||||||
|
let worktrees_dir = root.join(".huskies/worktrees");
|
||||||
|
std::fs::create_dir_all(&worktrees_dir).unwrap();
|
||||||
|
|
||||||
|
// Simulate several stories stuck in 4_merge/ with a crashed/unassignable
|
||||||
|
// agent: each has a worktree directory (so find_worktree_path succeeds
|
||||||
|
// and the git-shelling check runs) but is not a real git repo and has
|
||||||
|
// no active agent entry in the pool.
|
||||||
|
const STORY_COUNT: usize = 25;
|
||||||
|
for i in 0..STORY_COUNT {
|
||||||
|
let story_id = format!("11700_merge_{i:03}");
|
||||||
|
std::fs::create_dir_all(worktrees_dir.join(&story_id)).unwrap();
|
||||||
|
crate::db::write_item_with_content(
|
||||||
|
&story_id,
|
||||||
|
"4_merge",
|
||||||
|
"---\nname: Crashed Merge\n---\n",
|
||||||
|
crate::db::ItemMeta::named("Crashed Merge"),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
let pool = AgentPool::new_test(3200);
|
||||||
|
let config = ProjectConfig::load(&root).unwrap_or_default();
|
||||||
|
|
||||||
|
// Stand in for the liveness heartbeat (tick_loop.rs's
|
||||||
|
// spawn_liveness_tick) and /health polling: a tight-interval task
|
||||||
|
// racing the merge scan on the single-worker-thread runtime.
|
||||||
|
let ticks = Arc::new(AtomicU64::new(0));
|
||||||
|
let ticks_clone = Arc::clone(&ticks);
|
||||||
|
let heartbeat = tokio::spawn(async move {
|
||||||
|
loop {
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(2)).await;
|
||||||
|
ticks_clone.fetch_add(1, Ordering::SeqCst);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
let start = std::time::Instant::now();
|
||||||
|
pool.assign_merge_stage(&root, &config).await;
|
||||||
|
let elapsed = start.elapsed();
|
||||||
|
|
||||||
|
heartbeat.abort();
|
||||||
|
let observed_ticks = ticks.load(Ordering::SeqCst);
|
||||||
|
|
||||||
|
// With a 2ms heartbeat cadence, an unstalled runtime should have
|
||||||
|
// fired roughly elapsed/2ms ticks. Require at least a quarter of that
|
||||||
|
// as a generous floor — a stalled runtime (pre-fix) produces ~0 ticks
|
||||||
|
// because the single worker thread never yields during the scan.
|
||||||
|
let expected_min_ticks = (elapsed.as_millis() / 2 / 4) as u64;
|
||||||
|
assert!(
|
||||||
|
observed_ticks >= expected_min_ticks,
|
||||||
|
"liveness heartbeat stalled during assign_merge_stage: {observed_ticks} tick(s) \
|
||||||
|
over {elapsed:?} (expected at least ~{expected_min_ticks}); the merge scan likely \
|
||||||
|
blocked the tokio runtime instead of yielding via spawn_blocking"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user