fix(broker): back off watched-host retries up to 30 seconds

commit: 9cfdc13ed07685424aeedb8c8331491445bc76ff
author: brice · 29d ago · 2026-09-23 19:48 UTC
parent: 4eaa45d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +21 -9

M internal/usecase/broker/registry.go +16 -2


@@ -32,6 +32,10 @@ const (
32	32		// while at least one client subscription is live (see Registry.SetDemand).
33	33		defaultDemandFreshForLocal = 1 * time.Second
34	34		defaultDemandFreshForRemote = 2 * time.Second
	35	+	// demandRetryMax caps the failure backoff while a client watches: retries
	36	+	// start at the demand cadence and double up to this bound, so an
	37	+	// unreachable or refusing host is not redialed every two seconds.
	38	+	demandRetryMax = 30 * time.Second
35	39		// defaultProbeTimeout bounds one observation attempt end to end: dial,
36	40		// SSH/QUIC bootstrap, daemonmux preamble, and the catalogue exchange. It
37	41		// sits below the passive retry cap so a peer that never answers costs one

@@ -1103,8 +1107,8 @@ func (r *Registry) apply(result probeResult) {
1103	1107		current.ConsecutiveFailures++
1104	1108		current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
1105	1109		delay := r.retryDelay(current.ConsecutiveFailures)
1106		-	if r.demand > 0 && delay > r.demandFreshForRemote {
1107		-	delay = r.demandFreshForRemote
	1110	+	if r.demand > 0 {
	1111	+	delay = demandRetryDelay(r.demandFreshForRemote, current.ConsecutiveFailures)
1108	1112		}
1109	1113		current.NextDue = result.at.Add(r.jitter(delay, result.endpoint, result.attempt))
1110	1114		r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err, "cause", errors.Unwrap(result.err))

@@ -1159,6 +1163,16 @@ func availabilityFor(kind domain.RemoteFailureKind) domain.RemoteAvailability {
1159	1163		}
1160	1164		}
1161	1165
	1166	+	// demandRetryDelay is the watched-host failure backoff: the demand cadence
	1167	+	// doubled per consecutive failure, capped at demandRetryMax.
	1168	+	func demandRetryDelay(base time.Duration, failures uint) time.Duration {
	1169	+	delay := base
	1170	+	for i := uint(1); i < failures && delay < demandRetryMax; i++ {
	1171	+	delay *= 2
	1172	+	}
	1173	+	return min(delay, demandRetryMax)
	1174	+	}
	1175	+
1162	1176		// retryDelay is the capped exponential retry cadence for a failure streak.
1163	1177		func (r *Registry) retryDelay(failures uint) time.Duration {
1164	1178		delay := r.retryBase

 	// while at least one client subscription is live (see Registry.SetDemand).
 	defaultDemandFreshForLocal  = 1 * time.Second
 	defaultDemandFreshForRemote = 2 * time.Second
+	// demandRetryMax caps the failure backoff while a client watches: retries
+	// start at the demand cadence and double up to this bound, so an
+	// unreachable or refusing host is not redialed every two seconds.
+	demandRetryMax = 30 * time.Second
 	// defaultProbeTimeout bounds one observation attempt end to end: dial,
 	// SSH/QUIC bootstrap, daemonmux preamble, and the catalogue exchange. It
 	// sits below the passive retry cap so a peer that never answers costs one
 	current.ConsecutiveFailures++
 	current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
 	delay := r.retryDelay(current.ConsecutiveFailures)
-	if r.demand > 0 && delay > r.demandFreshForRemote {
+	if r.demand > 0 {
-		delay = r.demandFreshForRemote
+		delay = demandRetryDelay(r.demandFreshForRemote, current.ConsecutiveFailures)
 	}
 	current.NextDue = result.at.Add(r.jitter(delay, result.endpoint, result.attempt))
 	r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err, "cause", errors.Unwrap(result.err))
 	}
 }
+// demandRetryDelay is the watched-host failure backoff: the demand cadence
+// doubled per consecutive failure, capped at demandRetryMax.
+func demandRetryDelay(base time.Duration, failures uint) time.Duration {
+	delay := base
+	for i := uint(1); i < failures && delay < demandRetryMax; i++ {
+		delay *= 2
+	}
+	return min(delay, demandRetryMax)
+}
 // retryDelay is the capped exponential retry cadence for a failure streak.
 func (r *Registry) retryDelay(failures uint) time.Duration {
 	delay := r.retryBase

M internal/usecase/broker/registry_test.go +5 -7


@@ -1661,7 +1661,7 @@ func TestRegistryDemandCapsFailureBackoff(t *testing.T) {
1661	1661		host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1662	1662		return !host.Checking && host.ConsecutiveFailures == failure
1663	1663		})
1664		-	require.LessOrEqual(t, host.NextDue.Sub(clock.Now()), defaultDemandFreshForRemote, "demanded retry gap must never exceed the freshness interval")
	1664	+	require.LessOrEqual(t, host.NextDue.Sub(clock.Now()), demandRetryMax, "demanded retry gap must never exceed the demand retry cap")
1665	1665		clock.Advance(host.NextDue.Sub(clock.Now()))
1666	1666		registry.RequestProbe(reg.Endpoint)
1667	1667		}

@@ -1683,13 +1683,11 @@ func TestRegistryBackoffIsExponentialAndCapped(t *testing.T) {
1683	1683		startRegistry(t, registry)
1684	1684
1685	1685		probeErr := errors.New("offline")
1686		-	// A watched host retries on its two-second demand cadence even as the
1687		-	// underlying exponential failure count increases.
	1686	+	// A watched host backs off from its two-second demand cadence, doubling
	1687	+	// per failure up to the demand retry cap.
1688	1688		want := []time.Duration{
1689		-	defaultDemandFreshForRemote, defaultDemandFreshForRemote,
1690		-	defaultDemandFreshForRemote, defaultDemandFreshForRemote,
1691		-	defaultDemandFreshForRemote, defaultDemandFreshForRemote,
1692		-	defaultDemandFreshForRemote,
	1689	+	2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second,
	1690	+	demandRetryMax, demandRetryMax, demandRetryMax,
1693	1691		}
1694	1692		for index, delay := range want {
1695	1693		call := receiveCall(t, probe)

 		host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
 			return !host.Checking && host.ConsecutiveFailures == failure
 		})
-		require.LessOrEqual(t, host.NextDue.Sub(clock.Now()), defaultDemandFreshForRemote, "demanded retry gap must never exceed the freshness interval")
+		require.LessOrEqual(t, host.NextDue.Sub(clock.Now()), demandRetryMax, "demanded retry gap must never exceed the demand retry cap")
 		clock.Advance(host.NextDue.Sub(clock.Now()))
 		registry.RequestProbe(reg.Endpoint)
 	}
 	startRegistry(t, registry)
 	probeErr := errors.New("offline")
-	// A watched host retries on its two-second demand cadence even as the
+	// A watched host backs off from its two-second demand cadence, doubling
-	// underlying exponential failure count increases.
+	// per failure up to the demand retry cap.
 	want := []time.Duration{
-		defaultDemandFreshForRemote, defaultDemandFreshForRemote,
+		2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second,
-		defaultDemandFreshForRemote, defaultDemandFreshForRemote,
+		demandRetryMax, demandRetryMax, demandRetryMax,
-		defaultDemandFreshForRemote, defaultDemandFreshForRemote,
-		defaultDemandFreshForRemote,
 	}
 	for index, delay := range want {
 		call := receiveCall(t, probe)