diff --git a/pkg/rclonecli/health.go b/pkg/rclonecli/health.go index 9040892c8..e07170826 100644 --- a/pkg/rclonecli/health.go +++ b/pkg/rclonecli/health.go @@ -21,9 +21,10 @@ const ( // probes are required before the rcd subprocess is considered wedged and // restarted. Prevents a single transient miss from nuking a healthy rcd. // + // Four failures span three intervals, matching the default 90s tolerance. // This is the fallback for rclone.rcd_restart_after; see // restartAfterProbeFailures, which derives the count from that setting. - maxConsecutiveProbeFailures = 3 + maxConsecutiveProbeFailures = 4 // probeTimeout bounds a single liveness probe. probeTimeout = 5 * time.Second @@ -34,8 +35,9 @@ const ( // // The knob is expressed as a duration rather than a probe count because a count // is meaningless without healthCheckInterval, and exposing both invites -// combinations that mean nothing. One duration, divided by the interval, keeps -// the two in step. +// combinations that mean nothing. The first failure starts the streak at t=0, +// so N failures span (N-1) intervals. Wait enough whole intervals to cover the +// duration, plus the initial failed probe. func (m *Manager) restartAfterProbeFailures() int { if m.cfg == nil { return maxConsecutiveProbeFailures @@ -55,20 +57,14 @@ func (m *Manager) restartAfterProbeFailures() int { // Round up by dividing first and adjusting, rather than the usual // (after + interval - 1) / interval. That form overflows for a duration near - // the maximum time.Duration, wrapping negative, and the clamp below would - // then turn the longest tolerance anyone can express into a threshold of 1: - // a restart on every failed probe, the exact opposite of what was asked for. - threshold := int(after / healthCheckInterval) + // the maximum time.Duration and could turn a long tolerance into an immediate + // restart. Even the maximum duration needs fewer than 308 million probes at + // this interval, so the initial probe and rounding fit in a 32-bit int. + threshold := int(after/healthCheckInterval) + 1 if after%healthCheckInterval != 0 { threshold++ } - // A value shorter than one interval still means "restart on the first - // sustained failure" rather than "restart immediately, every tick". - if threshold < 1 { - threshold = 1 - } - return threshold } diff --git a/pkg/rclonecli/health_test.go b/pkg/rclonecli/health_test.go index cee23f751..d0937fd50 100644 --- a/pkg/rclonecli/health_test.go +++ b/pkg/rclonecli/health_test.go @@ -8,6 +8,8 @@ import ( "io" "log/slog" "net/http" + "net/http/httptest" + "net/url" "strings" "sync/atomic" "testing" @@ -169,18 +171,21 @@ func TestRestartAfterProbeFailures_DerivesCountFromDuration(t *testing.T) { want int why string }{ - {"90s", 3, "the default, and the previous hard-coded behaviour"}, - {"30s", 1, "exactly one interval"}, - {"5m", 10, "a tolerant install riding out a long stall"}, - {"45s", 2, "rounds up rather than truncating to one interval"}, - {"1s", 1, "shorter than an interval still means one sustained failure"}, - {"", 3, "unset falls back to the built-in default"}, - {"nonsense", 3, "unparseable falls back rather than disabling the guard"}, - {"-30s", 3, "negative falls back rather than restarting every tick"}, + {"90s", 4, "three intervals elapse after the first failure"}, + {"30.000000001s", 3, "just beyond an interval needs two complete intervals"}, + {"29.999999999s", 2, "just below an interval waits one complete interval"}, + {"0s", 4, "zero uses the 90-second default"}, + {"30s", 2, "exactly one interval after the first failure"}, + {"5m", 11, "a tolerant install riding out a long stall"}, + {"45s", 3, "rounds up rather than truncating to one interval"}, + {"1s", 2, "a positive tolerance requires another probe"}, + {"", 4, "unset falls back to the built-in default"}, + {"nonsense", 4, "unparseable falls back rather than disabling the guard"}, + {"-30s", 4, "negative falls back rather than restarting every tick"}, // Near time.Duration's maximum. The obvious (x+interval-1)/interval form // overflows here and collapses to 1, turning the longest tolerance // expressible into a restart on every failed probe. - {"2562047h47m16.854775807s", 307445735, "an absurd but valid duration must not invert into no tolerance"}, + {"2562047h47m16.854775807s", 307445736, "an absurd but valid duration must not invert into no tolerance"}, } { m, _ := newHealthTestManager(t, false, time.Time{}) withRcdRestartAfter(t, m, tc.configured) @@ -194,8 +199,8 @@ func TestRestartAfterProbeFailures_DerivesCountFromDuration(t *testing.T) { func TestRestartAfterProbeFailures_NoConfigUsesDefault(t *testing.T) { m, _ := newHealthTestManager(t, false, time.Time{}) - if got := m.restartAfterProbeFailures(); got != maxConsecutiveProbeFailures { - t.Errorf("with no config wired, threshold = %d, want %d", got, maxConsecutiveProbeFailures) + if got := m.restartAfterProbeFailures(); got != 4 { + t.Errorf("with no config wired, threshold = %d, want 4 for the 90s default", got) } } @@ -205,7 +210,7 @@ func TestRestartAfterProbeFailures_NoConfigUsesDefault(t *testing.T) { // every reader. func TestPerformMountHealthCheck_HonoursConfiguredTolerance(t *testing.T) { m, restarts := newHealthTestManager(t, false, time.Now().Add(-time.Hour)) - withRcdRestartAfter(t, m, "5m") // 10 probes + withRcdRestartAfter(t, m, "5m") // 11 probes for range maxConsecutiveProbeFailures + 2 { m.performMountHealthCheck() @@ -214,7 +219,7 @@ func TestPerformMountHealthCheck_HonoursConfiguredTolerance(t *testing.T) { t.Fatalf("restarted %d times past the default threshold; the configured tolerance was ignored", got) } - for range 5 { + for range 11 - (maxConsecutiveProbeFailures + 2) { m.performMountHealthCheck() } if got := atomic.LoadInt32(restarts); got != 1 { @@ -222,6 +227,67 @@ func TestPerformMountHealthCheck_HonoursConfiguredTolerance(t *testing.T) { } } +// A regression to ceil(duration/interval) restarts at t=60s for a 90s +// tolerance. Exercise the real HTTP probe and health-check decision together, +// with each call representing a monitor tick; no wall-clock sleeps are needed. +func TestPerformMountHealthCheck_RestartToleranceWithRCServer(t *testing.T) { + for _, configured := range []string{"90s", "", "invalid", "0s", "-1s"} { + t.Run(configured, func(t *testing.T) { + var healthy atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/core/version" { + t.Errorf("unexpected probe: %s %s", r.Method, r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + if !healthy.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"version":"v1.70.0"}`)) + })) + defer server.Close() + u, err := url.Parse(server.URL) + if err != nil { + t.Fatal(err) + } + m, restarts := newHealthTestManager(t, false, afterGrace()) + withRcdRestartAfter(t, m, configured) + m.rcPort = u.Port() + m.httpClient = server.Client() + m.probe = m.pingServerWithTimeout + + for _, elapsed := range []time.Duration{0, 30 * time.Second, 60 * time.Second} { + m.performMountHealthCheck() + if got := atomic.LoadInt32(restarts); got != 0 { + t.Fatalf("restart at %s before the 90s tolerance elapsed: %d", elapsed, got) + } + } + // One successful probe breaks the streak: subsequent failures must wait + // the whole tolerance, rather than retaining the previous failures. + healthy.Store(true) + m.performMountHealthCheck() + healthy.Store(false) + for _, elapsed := range []time.Duration{0, 30 * time.Second, 60 * time.Second, 90 * time.Second} { + m.performMountHealthCheck() + want := int32(0) + if elapsed == 90*time.Second { + want = 1 + } + if got := atomic.LoadInt32(restarts); got != want { + t.Fatalf("after healthy reset, restarts at %s = %d, want %d", elapsed, got, want) + } + } + // A restart also breaks the streak; its next failure cannot restart again. + m.performMountHealthCheck() + if got := atomic.LoadInt32(restarts); got != 1 { + t.Fatalf("restart failed to reset the streak: %d restarts", got) + } + }) + } +} + // Recovery uses the real RecoverMount path, stopping at the external force // unmount boundary so no FUSE mount or subprocess is needed. func newMountHealthTestManager(t *testing.T) (*Manager, map[string]bool, chan string) { @@ -304,8 +370,8 @@ func TestPerformMountHealthCheck_SingleMountFailureDoesNotRecover(t *testing.T) func TestPerformMountHealthCheck_MountThresholdRecoversOnce(t *testing.T) { m, _, recoveries := newMountHealthTestManager(t) - withRcdRestartAfter(t, m, "2m") // Four probes, using existing rcd tolerance. - for range 3 { + withRcdRestartAfter(t, m, "2m") // Five probes, using existing rcd tolerance. + for range 4 { m.performMountHealthCheck() } assertNoMountRecovery(t, recoveries) @@ -317,7 +383,7 @@ func TestPerformMountHealthCheck_MountThresholdRecoversOnce(t *testing.T) { func TestPerformMountHealthCheck_MountSuccessResetsStreak(t *testing.T) { m, failures, recoveries := newMountHealthTestManager(t) - for range 2 { + for range 3 { m.performMountHealthCheck() } failures["altmount"] = false @@ -327,7 +393,7 @@ func TestPerformMountHealthCheck_MountSuccessResetsStreak(t *testing.T) { t.Fatalf("healthy mount retains error: %q", info.Error) } failures["altmount"] = true - for range 2 { + for range 3 { m.performMountHealthCheck() } assertNoMountRecovery(t, recoveries) @@ -338,7 +404,7 @@ func TestPerformMountHealthCheck_MountSuccessResetsStreak(t *testing.T) { func TestPerformMountHealthCheck_MountStreaksAreIndependent(t *testing.T) { m, failures, recoveries := newMountHealthTestManager(t) m.mounts["other"] = &MountInfo{Provider: "other", LocalPath: "other", Mounted: true, desired: true} - for range 2 { + for range 3 { m.performMountHealthCheck() } failures["other"] = true @@ -347,18 +413,20 @@ func TestPerformMountHealthCheck_MountStreaksAreIndependent(t *testing.T) { m.performMountHealthCheck() assertNoMountRecovery(t, recoveries) m.performMountHealthCheck() + assertNoMountRecovery(t, recoveries) + m.performMountHealthCheck() awaitMountRecovery(t, m, recoveries, "other") } func TestPerformMountHealthCheck_NewMountStartsFreshStreak(t *testing.T) { m, _, recoveries := newMountHealthTestManager(t) - for range 2 { + for range 3 { m.performMountHealthCheck() } m.mountsMutex.Lock() m.mounts["altmount"] = &MountInfo{Provider: "altmount", LocalPath: "altmount", Mounted: true, desired: true} m.mountsMutex.Unlock() - for range 2 { + for range 3 { m.performMountHealthCheck() } assertNoMountRecovery(t, recoveries) @@ -379,7 +447,7 @@ func TestCheckMountHealth_ReturnsUnderlyingError(t *testing.T) { func TestPerformMountHealthCheck_IgnoresResultForReplacedMount(t *testing.T) { m, _, recoveries := newMountHealthTestManager(t) - for range 2 { + for range 3 { m.performMountHealthCheck() } transport := m.httpClient.Transport @@ -392,7 +460,7 @@ func TestPerformMountHealthCheck_IgnoresResultForReplacedMount(t *testing.T) { }) m.performMountHealthCheck() m.httpClient.Transport = transport - for range 2 { + for range 3 { m.performMountHealthCheck() } assertNoMountRecovery(t, recoveries) @@ -402,7 +470,7 @@ func TestPerformMountHealthCheck_IgnoresResultForReplacedMount(t *testing.T) { func TestPerformMountHealthCheck_UnmountResetsStreak(t *testing.T) { m, _, recoveries := newMountHealthTestManager(t) - for range 2 { + for range 3 { m.performMountHealthCheck() } m.forceUnmount = func(string) error { return nil } @@ -417,7 +485,7 @@ func TestPerformMountHealthCheck_UnmountResetsStreak(t *testing.T) { recoveries <- path return errors.New("stop recovery before remount") } - for range 2 { + for range 3 { m.performMountHealthCheck() } assertNoMountRecovery(t, recoveries) @@ -429,7 +497,7 @@ func TestPerformMountHealthCheck_QueuedRecoveryIgnoresLifecycleChange(t *testing for _, action := range []string{"replace", "unmount"} { t.Run(action, func(t *testing.T) { m, _, recoveries := newMountHealthTestManager(t) - for range 2 { + for range 3 { m.performMountHealthCheck() } m.mountMu.Lock()