huskies: merge 1175 bug Flaky tests intermittently fail merge gates

This commit is contained in:
Huskies Agent
2026-07-16 13:03:05 +00:00
parent 6a1ee8377d
commit 6f8a8ffd87
4 changed files with 48 additions and 4 deletions
@@ -32,7 +32,19 @@ impl AgentPool {
/// Called at the top of [`start_merge_agent_work`] to unblock retries, /// Called at the top of [`start_merge_agent_work`] to unblock retries,
/// and also by the periodic background reaper in the tick loop so stale /// and also by the periodic background reaper in the tick loop so stale
/// entries are cleaned up even when no new merge is triggered. /// entries are cleaned up even when no new merge is triggered.
///
/// A job's `server_start` round-trips through JSON text (see
/// [`encode_server_start_time`]/[`decode_server_start_time`]), and
/// `serde_json`'s float parser is not guaranteed bit-exact for
/// high-precision Unix timestamps — it can decode a value a couple of
/// ULPs below the original. Comparing with a bare `<` against a
/// freshly-read `current_boot` would then occasionally treat a job
/// written by *this very server instance* as belonging to a previous
/// one. Real server restarts are always seconds apart at minimum, so a
/// generous tolerance absorbs that noise without weakening genuine
/// stale-boot detection.
pub(crate) fn reap_stale_merge_jobs(&self) { pub(crate) fn reap_stale_merge_jobs(&self) {
const STALE_TOLERANCE_SECS: f64 = 1.0;
if let Some(jobs) = crate::crdt_state::read_all_merge_jobs() { if let Some(jobs) = crate::crdt_state::read_all_merge_jobs() {
let current_boot = server_start_time(); let current_boot = server_start_time();
for job in jobs { for job in jobs {
@@ -40,7 +52,7 @@ impl AgentPool {
continue; continue;
} }
let stale = match decode_server_start_time(job.error.as_deref()) { let stale = match decode_server_start_time(job.error.as_deref()) {
Some(t) => t < current_boot, Some(t) => t < current_boot - STALE_TOLERANCE_SECS,
None => true, // Legacy (pid-encoded) or malformed: stale None => true, // Legacy (pid-encoded) or malformed: stale
}; };
if stale { if stale {
@@ -113,6 +113,22 @@ async fn stale_running_merge_job_is_cleared_and_retry_succeeds() {
result.is_ok(), result.is_ok(),
"start_merge_agent_work must succeed after stale Running job is cleared; got: {result:?}" "start_merge_agent_work must succeed after stale Running job is cleared; got: {result:?}"
); );
// start_merge_agent_work spawns the actual pipeline as a background
// tokio task and returns immediately. Wait for it to reach a terminal
// state before the test ends: otherwise the task keeps running after
// `_serial` is released and can still be touching CRDT state
// (write_merge_job / delete_merge_job) while the next merge-pipeline
// test has already called init_for_test(), corrupting that test's
// thread-local state.
loop {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
if let Some(job) = pool.get_merge_status("77_story_stale")
&& !matches!(job.status, MergeJobStatus::Running)
{
break;
}
}
} }
// ── story 852: periodic background reaper ──────────────────────────────── // ── story 852: periodic background reaper ────────────────────────────────
@@ -1690,8 +1690,16 @@ mod tests {
#[test] #[test]
fn find_free_port_returns_bindable_port() { fn find_free_port_returns_bindable_port() {
let port = find_free_port(2200).expect("expected Some(port) in range 2200..2300"); // Scan from an OS-assigned ephemeral port rather than a hardcoded low
assert!((2200..2300).contains(&port)); // port. Fixed ports like 2200 are also used as scan starts elsewhere
// in production code and tests, so a second, independent bind on a
// fixed port can race against those under parallel test execution.
let reservation = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
let start = reservation.local_addr().unwrap().port();
drop(reservation);
let port = find_free_port(start).expect("expected Some(port) in scan range");
assert!((start..start.saturating_add(100)).contains(&port));
let listener = std::net::TcpListener::bind(("127.0.0.1", port)); let listener = std::net::TcpListener::bind(("127.0.0.1", port));
assert!(listener.is_ok(), "returned port {port} is not bindable"); assert!(listener.is_ok(), "returned port {port} is not bindable");
} }
+9 -1
View File
@@ -89,7 +89,15 @@ mod tests {
#[test] #[test]
fn find_free_port_returns_bindable_port() { fn find_free_port_returns_bindable_port() {
let port = find_free_port(3100); // Scan from an OS-assigned ephemeral port rather than a hardcoded low
// port. Fixed ports like 3100 are also used as scan starts elsewhere
// in production code and tests, so a second, independent bind on a
// fixed port can race against those under parallel test execution.
let reservation = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
let start = reservation.local_addr().unwrap().port();
drop(reservation);
let port = find_free_port(start);
// The returned port must be bindable. // The returned port must be bindable.
assert!( assert!(
std::net::TcpListener::bind(("127.0.0.1", port)).is_ok(), std::net::TcpListener::bind(("127.0.0.1", port)).is_ok(),