diff --git a/mooncake-transfer-engine/tent/include/tent/transport/rdma/rail_monitor.h b/mooncake-transfer-engine/tent/include/tent/transport/rdma/rail_monitor.h index 4ef853af9c..486ea82d97 100644 --- a/mooncake-transfer-engine/tent/include/tent/transport/rdma/rail_monitor.h +++ b/mooncake-transfer-engine/tent/include/tent/transport/rdma/rail_monitor.h @@ -41,6 +41,8 @@ class RailMonitor { "transports/rdma/rail_error_window_secs"; static constexpr const char *kCfgCooldownSecs = "transports/rdma/rail_cooldown_secs"; + static constexpr const char *kCfgProbeIntervalSecs = + "transports/rdma/rail_probe_interval_secs"; public: RailMonitor() = default; @@ -67,12 +69,28 @@ class RailMonitor { bool ready() { return ready_; } - bool available(int local_nic, int remote_nic); + // Pure predicate: true iff the rail is Closed/healthy and directly usable. + // Does NOT mutate state and does NOT call updateBestMapping, so it is safe + // to call from updateBestMapping (no recursion) and from any query path. + bool isAvailable(int local_nic, int remote_nic) const; + + // Mutating admit: the ONLY path that arms a probe/trial. Returns true when + // a transfer may use this rail -- Closed (no-op), or an exploratory probe / + // Half-Open trial. Sets probe_in_flight for paused rails so a second probe + // cannot race the first while it is on the wire. Callers that select a rail + // but fail to post must call cancelProbe() to clear the flag. + bool admit(int local_nic, int remote_nic); void markFailed(int local_nic, int remote_nic); void markRecovered(int local_nic, int remote_nic); + // Clear an armed probe/trial when a slice selected the rail via admit() but + // never reached the wire (e.g. endpoint creation failed, HW rejected the + // post). No-op when no probe is in flight, so callers may invoke it + // unconditionally on the re-queue path. + void cancelProbe(int local_nic, int remote_nic); + int findBestRemoteDevice(int local_nic, int remote_numa); const Topology *local() const { return local_.get(); } @@ -103,8 +121,28 @@ class RailMonitor { std::chrono::seconds cooldown{0}; std::chrono::steady_clock::time_point last_error{}; std::chrono::steady_clock::time_point resume_time{}; - - // Derived: a rail is paused iff a resume_time has been armed. + // Last time admit() armed a probe/trial. Throttles the probe rate + // during Open and defers the first probe by one probe_interval_ after a + // pause. + std::chrono::steady_clock::time_point last_probe_time{}; + // Half-Open: the cooldown expired but recovery is not yet proven. + // admit() arms exactly one trial; a trial success closes the rail + // (markRecovered), a trial failure escalates the cooldown and re-arms + // (markFailed). This replaces the old "expiry fully reopens" path, + // which slammed a still-dead peer with every slice and re-triggered the + // storm at 30s/60s/120s. + bool half_open = false; + // A probe/trial slice is selected/posted and its completion has not yet + // resolved the rail. admit() refuses to arm a second while this is set, + // so at most one probe/trial is on the wire per rail (single-threaded + // worker ownership makes a bool sufficient). Cleared by markFailed / + // markRecovered on completion, and by cancelProbe() on a pre-wire + // failure that never reaches the wire. + bool probe_in_flight = false; + + // Derived: a rail is paused iff a resume_time has been armed. The + // Half-Open sub-state also has resume_time armed so paused() stays + // true. bool paused() const { return resume_time != std::chrono::steady_clock::time_point{}; } @@ -117,6 +155,7 @@ class RailMonitor { int error_threshold_ = 3; std::chrono::seconds error_window_{10}; std::chrono::seconds cooldown_{30}; + std::chrono::seconds probe_interval_{1}; }; } // namespace tent diff --git a/mooncake-transfer-engine/tent/src/transport/rdma/rail_monitor.cpp b/mooncake-transfer-engine/tent/src/transport/rdma/rail_monitor.cpp index 236b2d1314..cd1bfd147b 100644 --- a/mooncake-transfer-engine/tent/src/transport/rdma/rail_monitor.cpp +++ b/mooncake-transfer-engine/tent/src/transport/rdma/rail_monitor.cpp @@ -77,12 +77,15 @@ Status RailMonitor::load(std::shared_ptr local, conf->get(kCfgErrorWindowSecs, (int)error_window_.count())); cooldown_ = std::chrono::seconds( conf->get(kCfgCooldownSecs, (int)cooldown_.count())); + probe_interval_ = std::chrono::seconds( + conf->get(kCfgProbeIntervalSecs, (int)probe_interval_.count())); // Config is identical on every COW snapshot refresh. Log once per // monitor so PD/e2e does not reprint the banner per slice/worker. if (first_load) { LOG(INFO) << "RailMonitor: error_threshold=" << error_threshold_ << " error_window=" << error_window_.count() << "s" - << " cooldown=" << cooldown_.count() << "s"; + << " cooldown=" << cooldown_.count() << "s" + << " probe_interval=" << probe_interval_.count() << "s"; } } if (same_layout) return Status::OK(); @@ -94,26 +97,75 @@ Status RailMonitor::load(std::shared_ptr local, return loadDefault(); } -bool RailMonitor::available(int local_nic, int remote_nic) { +bool RailMonitor::isAvailable(int local_nic, int remote_nic) const { + auto it = rail_states_.find(std::make_pair(local_nic, remote_nic)); + if (it == rail_states_.end()) return false; + // True only for Closed/healthy rails. Open (cooldown running), + // expired-Open, and Half-Open all have resume_time armed -> paused() -> + // false. They must go through admit() (which may arm a probe/trial), not be + // used directly. This is a pure predicate: it mutates nothing and never + // calls updateBestMapping, so it is safe to call from updateBestMapping + // without the recursion the old available() had. + return !it->second.paused(); +} + +bool RailMonitor::admit(int local_nic, int remote_nic) { auto it = rail_states_.find(std::make_pair(local_nic, remote_nic)); if (it == rail_states_.end()) return false; auto& st = it->second; + // Closed: healthy and directly usable. No probe, no flag -- a normal + // transfer whose completion is the fast-path markRecovered no-op. if (!st.paused()) return true; + // Half-Open: one trial is already admitted and unresolved. Block more + // traffic until the trial resolves (markRecovered closes, markFailed + // re-arms, cancelProbe reverts). resume_time stays armed so paused() + // remains true and the rail is not mistaken for Closed. + if (st.half_open) return false; auto now = std::chrono::steady_clock::now(); - if (now < st.resume_time) return false; - // Cooldown expired. Clear the pause so traffic resumes. Half-Open - // (admit one trial on expiry, escalate only on trial failure) is a - // follow-up: it needs available() split into a non-mutating predicate - // plus an admit() with in-flight tracking, so probes don't race the - // fallback/updateBestMapping callers that also call available(). - st.resume_time = {}; - st.error_count = 0; - updateBestMapping(); - LOG(INFO) << "Rail recovered: local_nic=" << local_nic - << " remote_nic=" << remote_nic - << " (cooldown expired, cooldown_retained=" << st.cooldown.count() - << "s)"; - return true; + // Open: the cooldown timer is still running. Admit ONE exploratory probe + // every probe_interval_ so a successful probe closes the rail early (defect + // B, transient faults); a failed probe is a no-op for markFailed (was + // paused, defect A). probe_in_flight gates a second probe from racing the + // first while it is still on the wire. + if (now < st.resume_time) { + if (!st.probe_in_flight && + (probe_interval_.count() == 0 || + now - st.last_probe_time >= probe_interval_)) { + st.probe_in_flight = true; + st.last_probe_time = now; + return true; + } + return false; + } + // Expired-Open: the cooldown timer fired but no probe proved the rail + // healthy (it is still paused). Transition to Half-Open and admit exactly + // ONE trial; escalation happens only if the trial FAILS (markFailed), not + // because the clock fired -- elapsed time does not prove the path healthy. + // If an Open-phase probe is still in flight at expiry, wait for it. + if (!st.probe_in_flight) { + st.half_open = true; + st.probe_in_flight = true; + st.last_probe_time = now; + LOG(INFO) << "Rail half-open: local_nic=" << local_nic + << " remote_nic=" << remote_nic + << " (cooldown=" << st.cooldown.count() + << "s retained, trial admitted)"; + return true; + } + return false; +} + +void RailMonitor::cancelProbe(int local_nic, int remote_nic) { + auto it = rail_states_.find(std::make_pair(local_nic, remote_nic)); + if (it == rail_states_.end()) return; + auto& st = it->second; + if (st.probe_in_flight) { + st.probe_in_flight = false; + // Revert a Half-Open trial to expired-Open so the next admit() re-arms + // a fresh trial. A plain Open probe just clears its flag. + st.half_open = false; + // Leave resume_time, cooldown, last_probe_time untouched. + } } void RailMonitor::markFailed(int local_nic, int remote_nic) { @@ -128,6 +180,32 @@ void RailMonitor::markFailed(int local_nic, int remote_nic) { } st.last_error = now; + // A probe/trial completion failed. Clear the in-flight flag and route by + // sub-state: a Half-Open trial failure escalates and re-arms (the ONLY + // escalation point for a paused rail); an Open-phase probe failure is a + // no-op (was paused, defect A). Either way this is not a fresh burst. + if (st.probe_in_flight) { + st.probe_in_flight = false; + if (st.half_open) { + st.half_open = false; + st.cooldown *= 2; + if (st.cooldown > kMaxCooldown) st.cooldown = kMaxCooldown; + st.error_count = 0; // fresh cycle + st.resume_time = now + st.cooldown; + st.last_probe_time = now; // defer next probe by one interval + LOG(INFO) << "Rail half-open trial failed; re-paused: local_nic=" + << local_nic << " remote_nic=" << remote_nic + << " (cooldown escalated to " << st.cooldown.count() + << "s)"; + updateBestMapping(); + } + // Open-phase probe failed: leave resume_time and cooldown untouched. + // The pause runs its course; admit() will arm the next probe after + // probe_interval_. Escalating here would let a sustained outage + // extend the pause indefinitely -- the original defect. + return; + } + const bool was_paused = st.paused(); if (st.error_count >= error_threshold_) { @@ -157,15 +235,16 @@ void RailMonitor::markFailed(int local_nic, int remote_nic) { << error_window_.count() << "s, cooldown=" << st.cooldown.count() << "s)"; st.resume_time = now + st.cooldown; + // Defer the first probe by one probe_interval_: without this, + // last_probe_time defaults to epoch and admit() would arm a probe + // the instant the pause arms, defeating the throttle. + st.last_probe_time = now; updateBestMapping(); } - // Already paused: leave resume_time and cooldown untouched. Re-arming - // or escalating here would let a sustained outage extend the pause - // indefinitely -- the original defect. The pause runs its course and - // available() reopens on cooldown expiry. - // Open-phase probing and Half-Open (admit one trial on expiry, escalate - // on trial failure) are follow-ups: they need available() split into a - // non-mutating predicate plus an admit() with in-flight tracking. + // Already paused (Open, no probe in flight): leave resume_time and + // cooldown untouched. Re-arming or escalating here would let a + // sustained outage extend the pause indefinitely -- the original + // defect. The pause runs its course; admit() probes recovery. } } @@ -173,6 +252,21 @@ void RailMonitor::markRecovered(int local_nic, int remote_nic) { auto it = rail_states_.find(std::make_pair(local_nic, remote_nic)); if (it == rail_states_.end()) return; auto& st = it->second; + // A probe/trial succeeded: clear the in-flight flag and close the rail, + // resetting all backoff memory. This is the early-recovery path for both + // an Open-phase probe and a Half-Open trial. + if (st.probe_in_flight) { + st.probe_in_flight = false; + st.half_open = false; + st.error_count = 0; + st.resume_time = {}; + st.cooldown = std::chrono::seconds(0); + st.last_probe_time = {}; + LOG(INFO) << "Rail recovered: local_nic=" << local_nic + << " remote_nic=" << remote_nic << " (probe/trial succeeded)"; + updateBestMapping(); + return; + } // Fast path: a healthy rail stays healthy. 99%+ of completions land // here, so we must not touch best_mapping_ or write any field. if (!st.paused() && st.error_count == 0 && st.cooldown.count() == 0) return; @@ -183,6 +277,7 @@ void RailMonitor::markRecovered(int local_nic, int remote_nic) { st.error_count = 0; st.resume_time = {}; st.cooldown = std::chrono::seconds(0); + st.last_probe_time = {}; if (was_paused) { LOG(INFO) << "Rail recovered: local_nic=" << local_nic << " remote_nic=" << remote_nic @@ -439,10 +534,10 @@ void RailMonitor::updateBestMapping() { remote_nic = remote_devices[remote_numa][i % remote_cnt]; } - if (!available(local_nic, remote_nic)) { + if (!isAvailable(local_nic, remote_nic)) { bool found = false; for (int cand : remote_devices[remote_numa]) { - if (available(local_nic, cand)) { + if (isAvailable(local_nic, cand)) { remote_nic = cand; found = true; break; @@ -450,7 +545,7 @@ void RailMonitor::updateBestMapping() { } if (!found) { for (int cand = 0; cand < remote_nic_count; ++cand) { - if (available(local_nic, cand)) { + if (isAvailable(local_nic, cand)) { remote_nic = cand; break; } diff --git a/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp b/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp index 4c97da4680..7132bd9a9e 100644 --- a/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp +++ b/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp @@ -772,6 +772,12 @@ void Workers::asyncPostSend() { discountFromOwner(worker, slice); } else { // The re-submit moves the count to the lane it lands on. + // No endpoint was obtained, so this slice never reached the + // wire on the rail admit() selected: clear any probe/trial + // it armed so the rail is not stranded. + if (auto* rail = slice->rail_monitor) + rail->cancelProbe(slice->source_dev_id, + slice->target_dev_id); submitFromTick(worker, slice); } } @@ -820,6 +826,11 @@ void Workers::asyncPostSend() { updateSliceStatus(slice, FAILED); discountFromOwner(worker, slice); } else { + // Rejected by hardware before reaching the wire: clear any + // probe/trial admit() armed on the selected rail. + if (auto* rail = slice->rail_monitor) + rail->cancelProbe(slice->source_dev_id, + slice->target_dev_id); submitFromTick(worker, slice); } } @@ -1566,7 +1577,7 @@ Status Workers::selectOptimalDevice(RouteHint& source, RouteHint& target, } if (gdr_excluded || - !rail.available(slice->source_dev_id, slice->target_dev_id)) { + !rail.admit(slice->source_dev_id, slice->target_dev_id)) { VLOG(1) << "Optimal device pair not available: source_dev_id " << slice->source_dev_id << ", target_dev_id " << slice->target_dev_id; @@ -1674,13 +1685,15 @@ Status Workers::selectFallbackDevice(RouteHint& source, RouteHint& target, if (strictLocalNuma() && source.topo->isCrossNuma(*source.topo_entry, sdev)) continue; - bool reachable = same_machine ? (sdev == tdev) // loopback is safe - : rail_mon->available(sdev, tdev); - // Skip NICs that cannot GPUDirect-DMA to the source/target GPU. - if (reachable && gdr_learned && + // Checked BEFORE admit() so a probe/trial is never armed on a rail + // that GDR will reject -- arming then discarding would orphan the + // in-flight flag and strand the rail. + if (!same_machine && gdr_learned && gdrPairExcluded(source, target, sdev, tdev, src_gpu, dst_gpu)) - reachable = false; + continue; + bool reachable = same_machine ? (sdev == tdev) // loopback is safe + : rail_mon->admit(sdev, tdev); if (reachable) { // A retry gets here after the failure path returned the slice's diff --git a/mooncake-transfer-engine/tent/tests/rail_monitor_test.cpp b/mooncake-transfer-engine/tent/tests/rail_monitor_test.cpp index caf7119a50..9e5a7c0a42 100644 --- a/mooncake-transfer-engine/tent/tests/rail_monitor_test.cpp +++ b/mooncake-transfer-engine/tent/tests/rail_monitor_test.cpp @@ -125,8 +125,8 @@ TEST(RailMonitorConfigTest, CustomJsonOverridesAutomaticPeerMapping) { ASSERT_TRUE(rail.load(local, remote, rail_json, nullptr).ok()); EXPECT_EQ(rail.findBestRemoteDevice(/*local_nic=*/0, /*remote_numa=*/0), 1); EXPECT_EQ(rail.findBestRemoteDevice(/*local_nic=*/1, /*remote_numa=*/0), 0); - EXPECT_TRUE(rail.available(/*local_nic=*/0, /*remote_nic=*/1)); - EXPECT_FALSE(rail.available(/*local_nic=*/0, /*remote_nic=*/0)); + EXPECT_TRUE(rail.isAvailable(/*local_nic=*/0, /*remote_nic=*/1)); + EXPECT_FALSE(rail.isAvailable(/*local_nic=*/0, /*remote_nic=*/0)); } // Build a 2-NIC topology (mlx5_a, mlx5_b) with per-NIC NUMA nodes, so the two @@ -204,19 +204,19 @@ TEST(RailMonitorRecoverTest, RecoverResetsErrorCount) { ASSERT_TRUE(rail.ready()); // Initially available - EXPECT_TRUE(rail.available(0, 0)); + EXPECT_TRUE(rail.isAvailable(0, 0)); // One failure — not yet past default threshold (3) rail.markFailed(0, 0); - EXPECT_TRUE(rail.available(0, 0)); // error_count=1, not paused + EXPECT_TRUE(rail.isAvailable(0, 0)); // error_count=1, not paused // A successful transfer — reset error_count back to 0 rail.markRecovered(0, 0); - EXPECT_TRUE(rail.available(0, 0)); + EXPECT_TRUE(rail.isAvailable(0, 0)); // Failure again — counter starts fresh from 0, one hit is not enough rail.markFailed(0, 0); - EXPECT_TRUE(rail.available(0, 0)); + EXPECT_TRUE(rail.isAvailable(0, 0)); } // --------------------------------------------------------------------------- @@ -231,13 +231,13 @@ TEST(RailMonitorRecoverTest, RecoverUnpausesPausedRail) { // Drive error_count to the default threshold (3) to trigger pause for (int i = 0; i < 3; ++i) rail.markFailed(0, 0); - EXPECT_FALSE(rail.available(0, 0)) + EXPECT_FALSE(rail.isAvailable(0, 0)) << "Rail should be paused after 3 failures"; // A successful transfer proves the path is live — should un-pause // immediately rail.markRecovered(0, 0); - EXPECT_TRUE(rail.available(0, 0)) + EXPECT_TRUE(rail.isAvailable(0, 0)) << "Rail should be available after recovery"; } @@ -277,7 +277,7 @@ TEST(RailMonitorLifetimeTest, KeepsSegmentSnapshotsAliveForFailureUpdates) { EXPECT_NO_FATAL_FAILURE({ for (int i = 0; i < 3; ++i) rail.markFailed(0, 0); }); - EXPECT_FALSE(rail.available(0, 0)); + EXPECT_FALSE(rail.isAvailable(0, 0)); } EXPECT_TRUE(weak_local.expired()); @@ -296,11 +296,11 @@ TEST(RailMonitorRecoverTest, FindBestAfterRecovery) { // Pause the only available rail for (int i = 0; i < 3; ++i) rail.markFailed(0, 0); - EXPECT_FALSE(rail.available(0, 0)); + EXPECT_FALSE(rail.isAvailable(0, 0)); // Recovery must rebuild best_mapping so findBestRemoteDevice works again rail.markRecovered(0, 0); - EXPECT_TRUE(rail.available(0, 0)); + EXPECT_TRUE(rail.isAvailable(0, 0)); int best = rail.findBestRemoteDevice(/*local_nic=*/0, /*remote_numa=*/0); EXPECT_EQ(best, 0) << "Recovered rail should be the best remote device"; } @@ -330,12 +330,12 @@ TEST(RailMonitorLoadTest, SameLayoutReloadPreservesErrorCount) { rail.markFailed(0, 0); rail.markFailed(0, 0); - EXPECT_TRUE(rail.available(0, 0)) + EXPECT_TRUE(rail.isAvailable(0, 0)) << "Two failures are below the default threshold of 3"; ASSERT_TRUE(rail.load(local2, remote2, "", &cfg).ok()); rail.markFailed(0, 0); - EXPECT_FALSE(rail.available(0, 0)) + EXPECT_FALSE(rail.isAvailable(0, 0)) << "COW snapshot refresh must not reset rail error_count"; } @@ -346,10 +346,10 @@ TEST(RailMonitorLoadTest, DifferentLayoutRebuildsMapping) { RailMonitor rail; ASSERT_TRUE(rail.load(local, remote_old).ok()); for (int i = 0; i < 3; ++i) rail.markFailed(0, 0); - EXPECT_FALSE(rail.available(0, 0)); + EXPECT_FALSE(rail.isAvailable(0, 0)); ASSERT_TRUE(rail.load(local, remote_new).ok()); - EXPECT_TRUE(rail.available(0, 0)) + EXPECT_TRUE(rail.isAvailable(0, 0)) << "A real topology change must rebuild rails from a clean state"; } @@ -358,35 +358,37 @@ TEST(RailMonitorRecoverTest, CooldownDoesNotCarryOverAfterRecovery) { auto remote = makeSingleNicTopology("mlx5_1"); Config cfg; - cfg.set(RailMonitor::kCfgErrorThreshold, 1); // pause on first failure - cfg.set(RailMonitor::kCfgErrorWindowSecs, 60); // wide: no window resets - cfg.set(RailMonitor::kCfgCooldownSecs, 1); // small initial cooldown + cfg.set(RailMonitor::kCfgErrorThreshold, 1); // pause on first failure + cfg.set(RailMonitor::kCfgErrorWindowSecs, 60); // wide: no window resets + cfg.set(RailMonitor::kCfgCooldownSecs, 1); // small initial cooldown + cfg.set(RailMonitor::kCfgProbeIntervalSecs, 60); // disable probing RailMonitor rail; ASSERT_TRUE(rail.load(local, remote, "", &cfg).ok()); // First pause cycle: single failure arms resume_time at now+1s. rail.markFailed(0, 0); - EXPECT_FALSE(rail.available(0, 0)); + EXPECT_FALSE(rail.isAvailable(0, 0)); - // Recover: must clear st.cooldown so the next pause uses 1s again, - // not the 1s left over from cycle 1 (which would double to 2s). + // Recover (regular, not a trial): must clear st.cooldown so the next + // pause uses 1s again, not the 1s left over from cycle 1 (which would + // double to 2s). rail.markRecovered(0, 0); - EXPECT_TRUE(rail.available(0, 0)); + EXPECT_TRUE(rail.isAvailable(0, 0)); // Second pause cycle: single failure must arm resume_time at now+1s. rail.markFailed(0, 0); - EXPECT_FALSE(rail.available(0, 0)); + EXPECT_FALSE(rail.isAvailable(0, 0)); // Wait 1.5s: longer than the initial 1s cooldown, shorter than the // 2s value the bug would produce. If cooldown was correctly reset on - // recovery, available() returns true; if it carried over, available() - // stays false until ~2s elapses. + // recovery, the cooldown expired and admit() admits a Half-Open trial; + // if it carried over, admit() stays false until ~2s elapses. std::this_thread::sleep_for(std::chrono::milliseconds(1500)); - EXPECT_TRUE(rail.available(0, 0)) + EXPECT_TRUE(rail.admit(0, 0)) << "After recovery, the next pause must use the initial cooldown " - "(1s); staying paused past 1.5s indicates cooldown carried over " - "from the previous cycle."; + "(1s); staying paused (admit=false) past 1.5s indicates cooldown " + "carried over from the previous cycle."; } // --------------------------------------------------------------------------- @@ -397,6 +399,12 @@ TEST(RailMonitorRecoverTest, CooldownDoesNotCarryOverAfterRecovery) { // once when a fresh pause arms; errors arriving while already paused are // no-ops. // +// Probing is disabled (probe_interval large) so a *cooling* rail's admit() +// returns false (now < resume_time, no probe) while an *expired* rail's +// admit() returns true (Half-Open trial). This discriminates the cooldown +// DURATION; the in-flight path is exercised separately by +// InFlightProbeBlocksSecondAdmit with the default probe_interval. +// // error_threshold=1, cooldown=1s. 8 rapid markFailed calls must arm resume_time // at now+1s, not now+256s. // --------------------------------------------------------------------------- @@ -409,6 +417,7 @@ TEST(RailMonitorBurstTest, BurstFailuresDoNotEscalateCooldown) { cfg.set(RailMonitor::kCfgErrorThreshold, 1); cfg.set(RailMonitor::kCfgErrorWindowSecs, 60); cfg.set(RailMonitor::kCfgCooldownSecs, 1); + cfg.set(RailMonitor::kCfgProbeIntervalSecs, 60); // disable probing RailMonitor rail; ASSERT_TRUE(rail.load(local, remote, "", &cfg).ok()); @@ -417,35 +426,30 @@ TEST(RailMonitorBurstTest, BurstFailuresDoNotEscalateCooldown) { // doubling, cooldown would be 1->2->4->...->256s (capped 300). With the // fix, only the first failure arms the pause at +1s; the rest are no-ops. for (int i = 0; i < 8; ++i) rail.markFailed(0, 0); - EXPECT_FALSE(rail.available(0, 0)); + EXPECT_FALSE(rail.isAvailable(0, 0)); + EXPECT_FALSE(rail.admit(0, 0)) << "Cooling rail must not admit (no probe)"; // 1.5s > 1s initial cooldown, far below any escalated value. If the burst - // had escalated, the rail would still be paused here. + // had escalated, the rail would still be cooling and admit()=false here. std::this_thread::sleep_for(std::chrono::milliseconds(1500)); - EXPECT_TRUE(rail.available(0, 0)) + EXPECT_TRUE(rail.admit(0, 0)) << "A single failure burst must not escalate the cooldown past the " - "initial 1s; staying paused past 1.5s indicates per-error doubling."; + "initial 1s; staying paused (admit=false) past 1.5s indicates " + "per-error doubling."; } // --------------------------------------------------------------------------- -// Defect A (cross-cycle): cooldown-expiry RETAINS the cooldown so a rail that -// fails again right after expiry backs off harder. (cooldown is reset to 0 -// only by markRecovered, a proven-healthy recovery.) This is the escalation -// path that the old "reset cooldown on expiry" behavior collapsed. -// -// Open-phase probing and Half-Open (admit one trial on expiry, escalate on -// trial failure) are out of scope for this PR: available() is a mutating -// admit used as a predicate across updateBestMapping / fallback / GDR, and -// with default probe_interval=1s there is no in-flight tracking, so a second -// probe can go out while the first is still on the wire. They land in a -// follow-up once available() is split into a non-mutating predicate plus an -// admit() with in-flight tracking. +// Half-Open: escalation happens on the trial RESULT, not because the clock +// fired. A Half-Open trial failure (markFailed while half_open) escalates the +// cooldown and re-arms; a trial success (markRecovered) closes the rail. // -// error_threshold=1, cooldown=1s. Cycle 1: pause 1s, expire. Cycle 2: re-fail -// must escalate to 2s. +// Probing disabled so a cooling rail's admit()=false while an expired rail's +// admit()=true (trial). error_threshold=1, cooldown=1s. +// Cycle 1: pause 1s, expire -> trial; trial fails -> escalate 1->2s re-arm. +// Cycle 2: 2s cooldown, expire -> trial. // --------------------------------------------------------------------------- -TEST(RailMonitorEscalationTest, ExpiryRetainsCooldownForNextCycle) { +TEST(RailMonitorHalfOpenTest, TrialFailureEscalatesCooldown) { auto local = makeSingleNicTopology("mlx5_0"); auto remote = makeSingleNicTopology("mlx5_1"); @@ -453,25 +457,168 @@ TEST(RailMonitorEscalationTest, ExpiryRetainsCooldownForNextCycle) { cfg.set(RailMonitor::kCfgErrorThreshold, 1); cfg.set(RailMonitor::kCfgErrorWindowSecs, 60); cfg.set(RailMonitor::kCfgCooldownSecs, 1); + cfg.set(RailMonitor::kCfgProbeIntervalSecs, 60); // disable probing RailMonitor rail; ASSERT_TRUE(rail.load(local, remote, "", &cfg).ok()); // Cycle 1: single failure arms a 1s pause. rail.markFailed(0, 0); - EXPECT_FALSE(rail.available(0, 0)); + EXPECT_FALSE(rail.isAvailable(0, 0)); std::this_thread::sleep_for(std::chrono::milliseconds(1100)); - ASSERT_TRUE(rail.available(0, 0)) << "Cycle 1 cooldown (1s) must expire"; + ASSERT_TRUE(rail.admit(0, 0)) << "Cycle 1 cooldown (1s) expired -> trial"; - // Cycle 2: fail again. Expiry retained cooldown=1s (not proven healthy), - // so markFailed escalates 1->2s. + // The trial fails: escalate 1->2s and re-arm. rail.markFailed(0, 0); - EXPECT_FALSE(rail.available(0, 0)); + EXPECT_FALSE(rail.isAvailable(0, 0)); std::this_thread::sleep_for(std::chrono::milliseconds(1100)); - EXPECT_FALSE(rail.available(0, 0)) + EXPECT_FALSE(rail.admit(0, 0)) << "Cycle 2 must use the escalated 2s cooldown; 1.1s is not enough."; std::this_thread::sleep_for(std::chrono::milliseconds(1400)); - EXPECT_TRUE(rail.available(0, 0)) << "Cycle 2 (2s) must expire by ~2.5s."; + EXPECT_TRUE(rail.admit(0, 0)) << "Cycle 2 (2s) expired -> trial by ~2.5s."; +} + +// --------------------------------------------------------------------------- +// A successful Half-Open trial closes the rail and resets backoff, so the +// next failure starts from the initial cooldown, not a doubled leftover. +// +// Probing disabled. Pause, expire -> trial, trial succeeds (markRecovered). +// The next pause must use 1s, not 2s. +// --------------------------------------------------------------------------- + +TEST(RailMonitorHalfOpenTest, TrialSuccessResetsBackoff) { + auto local = makeSingleNicTopology("mlx5_0"); + auto remote = makeSingleNicTopology("mlx5_1"); + + Config cfg; + cfg.set(RailMonitor::kCfgErrorThreshold, 1); + cfg.set(RailMonitor::kCfgErrorWindowSecs, 60); + cfg.set(RailMonitor::kCfgCooldownSecs, 1); + cfg.set(RailMonitor::kCfgProbeIntervalSecs, 60); + + RailMonitor rail; + ASSERT_TRUE(rail.load(local, remote, "", &cfg).ok()); + + // Pause, let it expire to Half-Open, then the trial succeeds. + rail.markFailed(0, 0); + std::this_thread::sleep_for(std::chrono::milliseconds(1100)); + ASSERT_TRUE(rail.admit(0, 0)) << "Half-Open trial admitted"; + rail.markRecovered(0, 0); + EXPECT_TRUE(rail.isAvailable(0, 0)) << "Trial success must close the rail"; + + // Next pause must use the initial 1s cooldown (backoff was reset), not 2s. + rail.markFailed(0, 0); + EXPECT_FALSE(rail.isAvailable(0, 0)); + std::this_thread::sleep_for(std::chrono::milliseconds(1500)); + EXPECT_TRUE(rail.admit(0, 0)) + << "After a successful trial, the next pause must use the initial 1s " + "cooldown; staying paused (admit=false) past 1.5s indicates backoff " + "carried over."; +} + +// --------------------------------------------------------------------------- +// Expiry admits exactly ONE trial, not a full reopen. The second caller is +// refused while the trial is in flight, so a still-dead peer is not flooded +// with every slice -- the storm the breaker exists to prevent. +// +// Probing disabled. Pause 1s, expire, admit()=true (trial), admit()=false. +// --------------------------------------------------------------------------- + +TEST(RailMonitorHalfOpenTest, ExpiryAdmitsOneTrialNotFullReopen) { + auto local = makeSingleNicTopology("mlx5_0"); + auto remote = makeSingleNicTopology("mlx5_1"); + + Config cfg; + cfg.set(RailMonitor::kCfgErrorThreshold, 1); + cfg.set(RailMonitor::kCfgErrorWindowSecs, 60); + cfg.set(RailMonitor::kCfgCooldownSecs, 1); + cfg.set(RailMonitor::kCfgProbeIntervalSecs, 60); + + RailMonitor rail; + ASSERT_TRUE(rail.load(local, remote, "", &cfg).ok()); + + rail.markFailed(0, 0); + std::this_thread::sleep_for(std::chrono::milliseconds(1100)); + EXPECT_TRUE(rail.admit(0, 0)) << "Expiry must admit one trial"; + EXPECT_FALSE(rail.admit(0, 0)) + << "After the trial is admitted, the next caller must be refused; " + "expiry must not reopen all traffic."; +} + +// --------------------------------------------------------------------------- +// In-flight tracking (the bug the reviewer flagged): with the DEFAULT +// probe_interval (1s), admit() must block a second probe while the first is +// on the wire, and clear the flag on completion so a new probe can arm. +// This test uses the default probe_interval, NOT 60s, so it exercises the +// in-flight path directly. +// --------------------------------------------------------------------------- + +TEST(RailMonitorProbeTest, InFlightProbeBlocksSecondAdmit) { + auto local = makeSingleNicTopology("mlx5_0"); + auto remote = makeSingleNicTopology("mlx5_1"); + + Config cfg; + cfg.set(RailMonitor::kCfgErrorThreshold, 1); + cfg.set(RailMonitor::kCfgErrorWindowSecs, 60); + cfg.set(RailMonitor::kCfgCooldownSecs, 10); // long cooldown + // DEFAULT probe_interval (1s) -- do NOT set kCfgProbeIntervalSecs. + + RailMonitor rail; + ASSERT_TRUE(rail.load(local, remote, "", &cfg).ok()); + + rail.markFailed(0, 0); // pause 10s; last_probe_time=now (deferral) + // Wait one probe_interval so the first probe is eligible. + std::this_thread::sleep_for(std::chrono::milliseconds(1100)); + EXPECT_TRUE(rail.admit(0, 0)) << "Open-phase probe must arm after interval"; + EXPECT_FALSE(rail.admit(0, 0)) + << "A second probe must be blocked while the first is in flight"; + + // The probe succeeds: reopen Closed and clear the in-flight flag. + rail.markRecovered(0, 0); + EXPECT_TRUE(rail.isAvailable(0, 0)); + + // Re-pause and arm another probe: the flag was cleared, so a new probe + // arms. + rail.markFailed(0, 0); + std::this_thread::sleep_for(std::chrono::milliseconds(1100)); + EXPECT_TRUE(rail.admit(0, 0)) + << "After completion clears the flag, a new probe must arm"; + EXPECT_FALSE(rail.admit(0, 0)) << "Again blocked while in flight"; +} + +// --------------------------------------------------------------------------- +// cancelProbe reverts an armed probe/trial when the slice never reaches the +// wire (pre-wire failure). After cancelProbe the rail must re-admit a trial. +// --------------------------------------------------------------------------- + +TEST(RailMonitorProbeTest, CancelProbeRevertsArmedTrial) { + auto local = makeSingleNicTopology("mlx5_0"); + auto remote = makeSingleNicTopology("mlx5_1"); + + Config cfg; + cfg.set(RailMonitor::kCfgErrorThreshold, 1); + cfg.set(RailMonitor::kCfgErrorWindowSecs, 60); + cfg.set(RailMonitor::kCfgCooldownSecs, 1); + cfg.set(RailMonitor::kCfgProbeIntervalSecs, 60); // disable probing + + RailMonitor rail; + ASSERT_TRUE(rail.load(local, remote, "", &cfg).ok()); + + rail.markFailed(0, 0); + std::this_thread::sleep_for(std::chrono::milliseconds(1100)); + ASSERT_TRUE(rail.admit(0, 0)) << "Expiry -> trial admitted"; + EXPECT_FALSE(rail.admit(0, 0)) << "Trial in flight blocks the second"; + + // The slice failed pre-wire: cancel the trial. The rail reverts to + // expired-Open and re-admits a fresh trial. + rail.cancelProbe(0, 0); + EXPECT_TRUE(rail.admit(0, 0)) << "cancelProbe must allow a fresh trial"; + EXPECT_FALSE(rail.admit(0, 0)) << "Re-armed trial is again in flight"; + + // cancelProbe is a no-op when no probe is in flight. + rail.markRecovered(0, 0); + rail.cancelProbe(0, 0); + EXPECT_TRUE(rail.isAvailable(0, 0)) << "cancelProbe no-op on a Closed rail"; } } // namespace