fix(rclone): honor rcd restart tolerance duration

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
 }
 

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,14 +219,75 @@ 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 {
 		t.Fatalf("restarts = %d after reaching the configured threshold, want 1", got)
 	}
 }
 
+// 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()