fix(broker): observe hosts only while clients watch
- commit
- 4827e8f4e826be656ba60070a165c69d765f80b3
- author
- brice · · 2026-09-23 09:37 UTC
- parent
- 43e0ba0
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
6 files · +119 -66
M internal/usecase/broker/registry.go +29 -24
| @@ -20,11 +20,8 @@ import ( | |||
| 20 | 20 | "github.com/bnema/vev/internal/ports" | |
| 21 | 21 | ) | |
| 22 | 22 | ||
| 23 | - | // Observation cadence. The broker deliberately mirrors the legacy remotes | |
| 24 | - | // monitor cadence (internal/usecase/remotes) so a fleet observed through | |
| 25 | - | // either path refreshes on the same schedule. The duplication is temporary: | |
| 26 | - | // the registry is intended to replace that monitor, and these constants move | |
| 27 | - | // to a single owner once the old monitor is removed. | |
| 23 | + | // Observation cadence. No client means no scheduled observations; the | |
| 24 | + | // fifteen-second interval only applies to an absent daemon under demand. | |
| 28 | 25 | const ( | |
| 29 | 26 | defaultFreshFor = 15 * time.Second | |
| 30 | 27 | defaultRetryBase = 5 * time.Second | |
| @@ -32,9 +29,7 @@ const ( | |||
| 32 | 29 | // defaultDemandFreshForLocal and defaultDemandFreshForRemote are the | |
| 33 | 30 | // built-in demand cadences RegistryConfig falls back to when | |
| 34 | 31 | // DemandFreshForLocal/DemandFreshForRemote is left zero. They apply only | |
| 35 | - | // while at least one client subscription is live (see Registry.SetDemand); | |
| 36 | - | // otherwise defaultFreshFor governs every re-probe schedule exactly as | |
| 37 | - | // before. | |
| 32 | + | // while at least one client subscription is live (see Registry.SetDemand). | |
| 38 | 33 | defaultDemandFreshForLocal = 1 * time.Second | |
| 39 | 34 | defaultDemandFreshForRemote = 2 * time.Second | |
| 40 | 35 | // defaultProbeTimeout bounds one observation attempt end to end: dial, | |
| @@ -89,8 +84,7 @@ type RegistryConfig struct { | |||
| 89 | 84 | // apply/applyLocal schedule into NextDue while Registry.SetDemand reports | |
| 90 | 85 | // at least one live client subscription. A zero value falls back to | |
| 91 | 86 | // defaultDemandFreshForLocal (about 1s) or defaultDemandFreshForRemote | |
| 92 | - | // (about 2s). With no live subscription, NextDue keeps the passive | |
| 93 | - | // defaultFreshFor (15s) exactly as before this field existed. | |
| 87 | + | // (about 2s). With no live subscription, no scheduled probe runs. | |
| 94 | 88 | DemandFreshForLocal time.Duration | |
| 95 | 89 | DemandFreshForRemote time.Duration | |
| 96 | 90 | // ProbeTimeout bounds every local and remote observation attempt on the | |
| @@ -185,8 +179,7 @@ type Registry struct { | |||
| 185 | 179 | probeTimeout time.Duration | |
| 186 | 180 | // demand counts live client subscriptions: Registry.SetDemand(true) is one | |
| 187 | 181 | // subscription opening and SetDemand(false) is one closing. Guarded by mu. | |
| 188 | - | // While demand > 0, apply and applyLocal schedule NextDue on the faster | |
| 189 | - | // demand cadence instead of the passive freshFor. | |
| 182 | + | // While demand > 0, apply and applyLocal schedule subsequent probes. | |
| 190 | 183 | demand int | |
| 191 | 184 | } | |
| 192 | 185 | ||
| @@ -862,17 +855,20 @@ func (r *Registry) RequestProbe(endpoint string) { | |||
| 862 | 855 | // SetDemand records one live client subscription opening (active=true) or | |
| 863 | 856 | // closing (active=false). internal/usecase/broker/service.go calls it from | |
| 864 | 857 | // Service.Subscribe and serviceSubscription.Close, so the demand count tracks | |
| 865 | - | // exactly the connections a client is actively watching. While the count is | |
| 866 | - | // above zero, apply and applyLocal schedule NextDue on the faster demand | |
| 867 | - | // cadence (demandFreshForLocal/demandFreshForRemote) instead of the passive | |
| 868 | - | // defaultFreshFor. The transition takes effect for the next observation to | |
| 869 | - | // complete; hint wakes the run loop so a registry idling on its timer | |
| 870 | - | // re-evaluates promptly rather than waiting out the current cadence. A | |
| 871 | - | // disabled registry never observes, so demand is still counted (Close must | |
| 872 | - | // stay balanced) but never changes any schedule. | |
| 858 | + | // exactly the connections a client is actively watching. The first | |
| 859 | + | // subscription requests one immediate observation per host; while demand | |
| 860 | + | // stays positive, the demand cadence schedules later observations. With no | |
| 861 | + | // subscription, only explicit RequestProbe calls dispatch. A disabled registry | |
| 862 | + | // still counts subscriptions but never observes. | |
| 873 | 863 | func (r *Registry) SetDemand(active bool) { | |
| 874 | 864 | r.mu.Lock() | |
| 875 | 865 | if active { | |
| 866 | + | if r.demand == 0 { | |
| 867 | + | r.localPending = r.local != nil | |
| 868 | + | for endpoint := range r.hosts { | |
| 869 | + | r.pending[endpoint] = true | |
| 870 | + | } | |
| 871 | + | } | |
| 876 | 872 | r.demand++ | |
| 877 | 873 | } else if r.demand > 0 { | |
| 878 | 874 | r.demand-- | |
| @@ -881,9 +877,8 @@ func (r *Registry) SetDemand(active bool) { | |||
| 881 | 877 | r.hint() | |
| 882 | 878 | } | |
| 883 | 879 | ||
| 884 | - | // freshForLocked returns the re-probe cadence to stamp into NextDue: the | |
| 885 | - | // faster demand cadence while a client subscription is live, otherwise the | |
| 886 | - | // passive default. Callers must hold r.mu. | |
| 880 | + | // freshForLocked returns the demand cadence while subscribed. An absent | |
| 881 | + | // daemon uses the longer freshFor interval instead. Callers must hold r.mu. | |
| 887 | 882 | func (r *Registry) freshForLocked(local bool) time.Duration { | |
| 888 | 883 | if r.demand <= 0 { | |
| 889 | 884 | return r.freshFor | |
| @@ -1002,12 +997,16 @@ type probeResult struct { | |||
| 1002 | 997 | ||
| 1003 | 998 | func (r *Registry) dispatch(ctx context.Context, now time.Time, results chan<- probeResult) { | |
| 1004 | 999 | r.mu.Lock() | |
| 1000 | + | if r.demand == 0 && len(r.pending) == 0 && !r.localPending { | |
| 1001 | + | r.mu.Unlock() | |
| 1002 | + | return | |
| 1003 | + | } | |
| 1005 | 1004 | started := false | |
| 1006 | 1005 | for endpoint, host := range r.hosts { | |
| 1007 | 1006 | if _, inflight := r.inflight[endpoint]; inflight { | |
| 1008 | 1007 | continue | |
| 1009 | 1008 | } | |
| 1010 | - | if !r.pending[endpoint] && !host.NextDue.IsZero() && now.Before(host.NextDue) { | |
| 1009 | + | if !r.pending[endpoint] && (r.demand == 0 || !host.NextDue.IsZero() && now.Before(host.NextDue)) { | |
| 1011 | 1010 | continue | |
| 1012 | 1011 | } | |
| 1013 | 1012 | delete(r.pending, endpoint) | |
| @@ -1257,6 +1256,12 @@ func jittered(base time.Duration, endpoint string, attempt uint64) time.Duration | |||
| 1257 | 1256 | func (r *Registry) nextDelay(now time.Time) time.Duration { | |
| 1258 | 1257 | r.mu.Lock() | |
| 1259 | 1258 | defer r.mu.Unlock() | |
| 1259 | + | if r.demand == 0 { | |
| 1260 | + | if r.localPending || len(r.pending) > 0 { | |
| 1261 | + | return 0 | |
| 1262 | + | } | |
| 1263 | + | return time.Hour | |
| 1264 | + | } | |
| 1260 | 1265 | var earliest time.Time | |
| 1261 | 1266 | if r.local != nil && r.localAttempt == nil { | |
| 1262 | 1267 | if r.localHost.NextDue.IsZero() { | |
| @@ -20,11 +20,8 @@ import ( | |||||
| 20 | "github.com/bnema/vev/internal/ports" | 20 | "github.com/bnema/vev/internal/ports" | ||
| 21 | ) | 21 | ) | ||
| 22 | 22 | ||||
| 23 | - | // Observation cadence. The broker deliberately mirrors the legacy remotes | 23 | + | // Observation cadence. No client means no scheduled observations; the |
| 24 | - | // monitor cadence (internal/usecase/remotes) so a fleet observed through | 24 | + | // fifteen-second interval only applies to an absent daemon under demand. |
| 25 | - | // either path refreshes on the same schedule. The duplication is temporary: | |||
| 26 | - | // the registry is intended to replace that monitor, and these constants move | |||
| 27 | - | // to a single owner once the old monitor is removed. | |||
| 28 | const ( | 25 | const ( | ||
| 29 | defaultFreshFor = 15 * time.Second | 26 | defaultFreshFor = 15 * time.Second | ||
| 30 | defaultRetryBase = 5 * time.Second | 27 | defaultRetryBase = 5 * time.Second | ||
| @@ -32,9 +29,7 @@ const ( | |||||
| 32 | // defaultDemandFreshForLocal and defaultDemandFreshForRemote are the | 29 | // defaultDemandFreshForLocal and defaultDemandFreshForRemote are the | ||
| 33 | // built-in demand cadences RegistryConfig falls back to when | 30 | // built-in demand cadences RegistryConfig falls back to when | ||
| 34 | // DemandFreshForLocal/DemandFreshForRemote is left zero. They apply only | 31 | // DemandFreshForLocal/DemandFreshForRemote is left zero. They apply only | ||
| 35 | - | // while at least one client subscription is live (see Registry.SetDemand); | 32 | + | // while at least one client subscription is live (see Registry.SetDemand). |
| 36 | - | // otherwise defaultFreshFor governs every re-probe schedule exactly as | |||
| 37 | - | // before. | |||
| 38 | defaultDemandFreshForLocal = 1 * time.Second | 33 | defaultDemandFreshForLocal = 1 * time.Second | ||
| 39 | defaultDemandFreshForRemote = 2 * time.Second | 34 | defaultDemandFreshForRemote = 2 * time.Second | ||
| 40 | // defaultProbeTimeout bounds one observation attempt end to end: dial, | 35 | // defaultProbeTimeout bounds one observation attempt end to end: dial, | ||
| @@ -89,8 +84,7 @@ type RegistryConfig struct { | |||||
| 89 | // apply/applyLocal schedule into NextDue while Registry.SetDemand reports | 84 | // apply/applyLocal schedule into NextDue while Registry.SetDemand reports | ||
| 90 | // at least one live client subscription. A zero value falls back to | 85 | // at least one live client subscription. A zero value falls back to | ||
| 91 | // defaultDemandFreshForLocal (about 1s) or defaultDemandFreshForRemote | 86 | // defaultDemandFreshForLocal (about 1s) or defaultDemandFreshForRemote | ||
| 92 | - | // (about 2s). With no live subscription, NextDue keeps the passive | 87 | + | // (about 2s). With no live subscription, no scheduled probe runs. |
| 93 | - | // defaultFreshFor (15s) exactly as before this field existed. | |||
| 94 | DemandFreshForLocal time.Duration | 88 | DemandFreshForLocal time.Duration | ||
| 95 | DemandFreshForRemote time.Duration | 89 | DemandFreshForRemote time.Duration | ||
| 96 | // ProbeTimeout bounds every local and remote observation attempt on the | 90 | // ProbeTimeout bounds every local and remote observation attempt on the | ||
| @@ -185,8 +179,7 @@ type Registry struct { | |||||
| 185 | probeTimeout time.Duration | 179 | probeTimeout time.Duration | ||
| 186 | // demand counts live client subscriptions: Registry.SetDemand(true) is one | 180 | // demand counts live client subscriptions: Registry.SetDemand(true) is one | ||
| 187 | // subscription opening and SetDemand(false) is one closing. Guarded by mu. | 181 | // subscription opening and SetDemand(false) is one closing. Guarded by mu. | ||
| 188 | - | // While demand > 0, apply and applyLocal schedule NextDue on the faster | 182 | + | // While demand > 0, apply and applyLocal schedule subsequent probes. |
| 189 | - | // demand cadence instead of the passive freshFor. | |||
| 190 | demand int | 183 | demand int | ||
| 191 | } | 184 | } | ||
| 192 | 185 | ||||
| @@ -862,17 +855,20 @@ func (r *Registry) RequestProbe(endpoint string) { | |||||
| 862 | // SetDemand records one live client subscription opening (active=true) or | 855 | // SetDemand records one live client subscription opening (active=true) or | ||
| 863 | // closing (active=false). internal/usecase/broker/service.go calls it from | 856 | // closing (active=false). internal/usecase/broker/service.go calls it from | ||
| 864 | // Service.Subscribe and serviceSubscription.Close, so the demand count tracks | 857 | // Service.Subscribe and serviceSubscription.Close, so the demand count tracks | ||
| 865 | - | // exactly the connections a client is actively watching. While the count is | 858 | + | // exactly the connections a client is actively watching. The first |
| 866 | - | // above zero, apply and applyLocal schedule NextDue on the faster demand | 859 | + | // subscription requests one immediate observation per host; while demand |
| 867 | - | // cadence (demandFreshForLocal/demandFreshForRemote) instead of the passive | 860 | + | // stays positive, the demand cadence schedules later observations. With no |
| 868 | - | // defaultFreshFor. The transition takes effect for the next observation to | 861 | + | // subscription, only explicit RequestProbe calls dispatch. A disabled registry |
| 869 | - | // complete; hint wakes the run loop so a registry idling on its timer | 862 | + | // still counts subscriptions but never observes. |
| 870 | - | // re-evaluates promptly rather than waiting out the current cadence. A | |||
| 871 | - | // disabled registry never observes, so demand is still counted (Close must | |||
| 872 | - | // stay balanced) but never changes any schedule. | |||
| 873 | func (r *Registry) SetDemand(active bool) { | 863 | func (r *Registry) SetDemand(active bool) { | ||
| 874 | r.mu.Lock() | 864 | r.mu.Lock() | ||
| 875 | if active { | 865 | if active { | ||
| 866 | + | if r.demand == 0 { | |||
| 867 | + | r.localPending = r.local != nil | |||
| 868 | + | for endpoint := range r.hosts { | |||
| 869 | + | r.pending[endpoint] = true | |||
| 870 | + | } | |||
| 871 | + | } | |||
| 876 | r.demand++ | 872 | r.demand++ | ||
| 877 | } else if r.demand > 0 { | 873 | } else if r.demand > 0 { | ||
| 878 | r.demand-- | 874 | r.demand-- | ||
| @@ -881,9 +877,8 @@ func (r *Registry) SetDemand(active bool) { | |||||
| 881 | r.hint() | 877 | r.hint() | ||
| 882 | } | 878 | } | ||
| 883 | 879 | ||||
| 884 | - | // freshForLocked returns the re-probe cadence to stamp into NextDue: the | 880 | + | // freshForLocked returns the demand cadence while subscribed. An absent |
| 885 | - | // faster demand cadence while a client subscription is live, otherwise the | 881 | + | // daemon uses the longer freshFor interval instead. Callers must hold r.mu. |
| 886 | - | // passive default. Callers must hold r.mu. | |||
| 887 | func (r *Registry) freshForLocked(local bool) time.Duration { | 882 | func (r *Registry) freshForLocked(local bool) time.Duration { | ||
| 888 | if r.demand <= 0 { | 883 | if r.demand <= 0 { | ||
| 889 | return r.freshFor | 884 | return r.freshFor | ||
| @@ -1002,12 +997,16 @@ type probeResult struct { | |||||
| 1002 | 997 | ||||
| 1003 | func (r *Registry) dispatch(ctx context.Context, now time.Time, results chan<- probeResult) { | 998 | func (r *Registry) dispatch(ctx context.Context, now time.Time, results chan<- probeResult) { | ||
| 1004 | r.mu.Lock() | 999 | r.mu.Lock() | ||
| 1000 | + | if r.demand == 0 && len(r.pending) == 0 && !r.localPending { | |||
| 1001 | + | r.mu.Unlock() | |||
| 1002 | + | return | |||
| 1003 | + | } | |||
| 1005 | started := false | 1004 | started := false | ||
| 1006 | for endpoint, host := range r.hosts { | 1005 | for endpoint, host := range r.hosts { | ||
| 1007 | if _, inflight := r.inflight[endpoint]; inflight { | 1006 | if _, inflight := r.inflight[endpoint]; inflight { | ||
| 1008 | continue | 1007 | continue | ||
| 1009 | } | 1008 | } | ||
| 1010 | - | if !r.pending[endpoint] && !host.NextDue.IsZero() && now.Before(host.NextDue) { | 1009 | + | if !r.pending[endpoint] && (r.demand == 0 || !host.NextDue.IsZero() && now.Before(host.NextDue)) { |
| 1011 | continue | 1010 | continue | ||
| 1012 | } | 1011 | } | ||
| 1013 | delete(r.pending, endpoint) | 1012 | delete(r.pending, endpoint) | ||
| @@ -1257,6 +1256,12 @@ func jittered(base time.Duration, endpoint string, attempt uint64) time.Duration | |||||
| 1257 | func (r *Registry) nextDelay(now time.Time) time.Duration { | 1256 | func (r *Registry) nextDelay(now time.Time) time.Duration { | ||
| 1258 | r.mu.Lock() | 1257 | r.mu.Lock() | ||
| 1259 | defer r.mu.Unlock() | 1258 | defer r.mu.Unlock() | ||
| 1259 | + | if r.demand == 0 { | |||
| 1260 | + | if r.localPending || len(r.pending) > 0 { | |||
| 1261 | + | return 0 | |||
| 1262 | + | } | |||
| 1263 | + | return time.Hour | |||
| 1264 | + | } | |||
| 1260 | var earliest time.Time | 1265 | var earliest time.Time | ||
| 1261 | if r.local != nil && r.localAttempt == nil { | 1266 | if r.local != nil && r.localAttempt == nil { | ||
| 1262 | if r.localHost.NextDue.IsZero() { | 1267 | if r.localHost.NextDue.IsZero() { | ||
A internal/usecase/broker/registry_demand_only_test.go +49 -0
| @@ -0,0 +1,49 @@ | |||
| 1 | + | package broker | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "testing" | |
| 6 | + | "time" | |
| 7 | + | ||
| 8 | + | "github.com/stretchr/testify/require" | |
| 9 | + | ||
| 10 | + | "github.com/bnema/vev/internal/domain" | |
| 11 | + | "github.com/bnema/vev/internal/ports" | |
| 12 | + | ) | |
| 13 | + | ||
| 14 | + | func TestRegistryObservesOnlyUnderSubscriptionOrExplicitRequest(t *testing.T) { | |
| 15 | + | start := time.Unix(100, 0) | |
| 16 | + | clock := newManualClock(start) | |
| 17 | + | probe := newTestProbe(4) | |
| 18 | + | local := newScriptedLocalProbe(4) | |
| 19 | + | registry := newLocalTestRegistry(t, newTestStore(), probe, local, clock) | |
| 20 | + | reg := registration(t, "example.test", 1) | |
| 21 | + | require.NoError(t, registry.setHosts(hostRecords(reg))) | |
| 22 | + | // Start Run directly: the test helper's synthetic subscription must not | |
| 23 | + | // hide an idle broker probe. | |
| 24 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 25 | + | done := make(chan struct{}) | |
| 26 | + | go func() { defer close(done); registry.Run(ctx) }() | |
| 27 | + | t.Cleanup(func() { cancel(); <-done }) | |
| 28 | + | clock.Advance(time.Hour) | |
| 29 | + | requireNoProbeWithin(t, probe, probeGrace) | |
| 30 | + | requireNoLocalProbeWithin(t, local, probeGrace) | |
| 31 | + | ||
| 32 | + | registry.SetDemand(true) | |
| 33 | + | remoteCall := receiveCall(t, probe) | |
| 34 | + | localCall := receiveLocalCall(t, local) | |
| 35 | + | remoteCall.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}} | |
| 36 | + | localCall.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnknown}} | |
| 37 | + | waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | |
| 38 | + | return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable | |
| 39 | + | }) | |
| 40 | + | waitLocal(t, registry, func(host ports.BrokerDaemonObservation) bool { return !host.Checking }) | |
| 41 | + | registry.SetDemand(false) | |
| 42 | + | clock.Advance(time.Hour) | |
| 43 | + | requireNoProbeWithin(t, probe, probeGrace) | |
| 44 | + | requireNoLocalProbeWithin(t, local, probeGrace) | |
| 45 | + | ||
| 46 | + | registry.RequestProbe(reg.Endpoint) | |
| 47 | + | receiveCall(t, probe) | |
| 48 | + | requireNoLocalProbeWithin(t, local, probeGrace) | |
| 49 | + | } | |
| @@ -0,0 +1,49 @@ | |||||
| 1 | + | package broker | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "testing" | |||
| 6 | + | "time" | |||
| 7 | + | ||||
| 8 | + | "github.com/stretchr/testify/require" | |||
| 9 | + | ||||
| 10 | + | "github.com/bnema/vev/internal/domain" | |||
| 11 | + | "github.com/bnema/vev/internal/ports" | |||
| 12 | + | ) | |||
| 13 | + | ||||
| 14 | + | func TestRegistryObservesOnlyUnderSubscriptionOrExplicitRequest(t *testing.T) { | |||
| 15 | + | start := time.Unix(100, 0) | |||
| 16 | + | clock := newManualClock(start) | |||
| 17 | + | probe := newTestProbe(4) | |||
| 18 | + | local := newScriptedLocalProbe(4) | |||
| 19 | + | registry := newLocalTestRegistry(t, newTestStore(), probe, local, clock) | |||
| 20 | + | reg := registration(t, "example.test", 1) | |||
| 21 | + | require.NoError(t, registry.setHosts(hostRecords(reg))) | |||
| 22 | + | // Start Run directly: the test helper's synthetic subscription must not | |||
| 23 | + | // hide an idle broker probe. | |||
| 24 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 25 | + | done := make(chan struct{}) | |||
| 26 | + | go func() { defer close(done); registry.Run(ctx) }() | |||
| 27 | + | t.Cleanup(func() { cancel(); <-done }) | |||
| 28 | + | clock.Advance(time.Hour) | |||
| 29 | + | requireNoProbeWithin(t, probe, probeGrace) | |||
| 30 | + | requireNoLocalProbeWithin(t, local, probeGrace) | |||
| 31 | + | ||||
| 32 | + | registry.SetDemand(true) | |||
| 33 | + | remoteCall := receiveCall(t, probe) | |||
| 34 | + | localCall := receiveLocalCall(t, local) | |||
| 35 | + | remoteCall.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}} | |||
| 36 | + | localCall.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnknown}} | |||
| 37 | + | waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | |||
| 38 | + | return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable | |||
| 39 | + | }) | |||
| 40 | + | waitLocal(t, registry, func(host ports.BrokerDaemonObservation) bool { return !host.Checking }) | |||
| 41 | + | registry.SetDemand(false) | |||
| 42 | + | clock.Advance(time.Hour) | |||
| 43 | + | requireNoProbeWithin(t, probe, probeGrace) | |||
| 44 | + | requireNoLocalProbeWithin(t, local, probeGrace) | |||
| 45 | + | ||||
| 46 | + | registry.RequestProbe(reg.Endpoint) | |||
| 47 | + | receiveCall(t, probe) | |||
| 48 | + | requireNoLocalProbeWithin(t, local, probeGrace) | |||
| 49 | + | } | |||
M internal/usecase/broker/registry_local.go +1 -1
| @@ -87,7 +87,7 @@ func (r *Registry) dispatchLocalLocked(ctx context.Context, now time.Time, resul | |||
| 87 | 87 | if r.local == nil || r.localAttempt != nil { | |
| 88 | 88 | return false | |
| 89 | 89 | } | |
| 90 | - | if !r.localPending && !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue) { | |
| 90 | + | if !r.localPending && (r.demand == 0 || !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue)) { | |
| 91 | 91 | return false | |
| 92 | 92 | } | |
| 93 | 93 | r.localPending = false | |
| @@ -87,7 +87,7 @@ func (r *Registry) dispatchLocalLocked(ctx context.Context, now time.Time, resul | |||||
| 87 | if r.local == nil || r.localAttempt != nil { | 87 | if r.local == nil || r.localAttempt != nil { | ||
| 88 | return false | 88 | return false | ||
| 89 | } | 89 | } | ||
| 90 | - | if !r.localPending && !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue) { | 90 | + | if !r.localPending && (r.demand == 0 || !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue)) { |
| 91 | return false | 91 | return false | ||
| 92 | } | 92 | } | ||
| 93 | r.localPending = false | 93 | r.localPending = false | ||
M internal/usecase/broker/registry_test.go +34 -34
| @@ -573,10 +573,11 @@ func newTestRegistry(t *testing.T, epoch ports.BrokerEpoch, store ports.BrokerHo | |||
| 573 | 573 | ||
| 574 | 574 | func startRegistry(t *testing.T, r *Registry) context.CancelFunc { | |
| 575 | 575 | t.Helper() | |
| 576 | + | r.SetDemand(true) | |
| 576 | 577 | ctx, cancel := context.WithCancel(context.Background()) | |
| 577 | 578 | done := make(chan struct{}) | |
| 578 | 579 | go func() { defer close(done); r.Run(ctx) }() | |
| 579 | - | t.Cleanup(func() { cancel(); <-done }) | |
| 580 | + | t.Cleanup(func() { cancel(); <-done; r.SetDemand(false) }) | |
| 580 | 581 | return cancel | |
| 581 | 582 | } | |
| 582 | 583 | ||
| @@ -1060,22 +1061,21 @@ func TestRegistrySchedulesFreshObservationAfterSuccess(t *testing.T) { | |||
| 1060 | 1061 | }) | |
| 1061 | 1062 | require.Equal(t, start, host.LastAttempt) | |
| 1062 | 1063 | require.Equal(t, start, host.LastSuccess) | |
| 1063 | - | require.Equal(t, start.Add(defaultFreshFor), host.NextDue) | |
| 1064 | + | require.Equal(t, start.Add(defaultDemandFreshForRemote), host.NextDue) | |
| 1064 | 1065 | require.Zero(t, host.ConsecutiveFailures) | |
| 1065 | 1066 | require.Zero(t, host.FailureEpisode) | |
| 1066 | 1067 | ||
| 1067 | 1068 | // A fresh host is not observed again before its scheduled deadline. | |
| 1068 | - | clock.Advance(defaultFreshFor - time.Nanosecond) | |
| 1069 | + | clock.Advance(defaultDemandFreshForRemote - time.Nanosecond) | |
| 1069 | 1070 | requireNoProbeWithin(t, probe, probeGrace) | |
| 1070 | 1071 | clock.Advance(time.Nanosecond) | |
| 1071 | 1072 | next := receiveCall(t, probe) | |
| 1072 | 1073 | require.True(t, next.registration.Equal(reg)) | |
| 1073 | 1074 | } | |
| 1074 | 1075 | ||
| 1075 | - | // TestRegistryAppliesSubscriptionDemandFreshness pins C2: apply/applyLocal | |
| 1076 | + | // TestRegistryAppliesSubscriptionDemandFreshness checks that apply/applyLocal | |
| 1076 | 1077 | // schedule NextDue on RegistryConfig's demand cadence while Registry.SetDemand | |
| 1077 | - | // reports a live client subscription, and keep the passive defaultFreshFor | |
| 1078 | - | // once demand drops back to zero. It exercises both the remote and the local | |
| 1078 | + | // reports a live client subscription. It exercises both the remote and the local | |
| 1079 | 1079 | // observation paths on the same registry, since service.go's Subscribe/Close | |
| 1080 | 1080 | // wiring feeds one shared demand counter for both. | |
| 1081 | 1081 | func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) { | |
| @@ -1089,9 +1089,7 @@ func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) { | |||
| 1089 | 1089 | demand bool | |
| 1090 | 1090 | wantAfter time.Duration | |
| 1091 | 1091 | }{ | |
| 1092 | - | {name: "remote no demand keeps default", local: false, demand: false, wantAfter: defaultFreshFor}, | |
| 1093 | 1092 | {name: "remote demand uses faster cadence", local: false, demand: true, wantAfter: demandRemote}, | |
| 1094 | - | {name: "local no demand keeps default", local: true, demand: false, wantAfter: defaultFreshFor}, | |
| 1095 | 1093 | {name: "local demand uses faster cadence", local: true, demand: true, wantAfter: demandLocal}, | |
| 1096 | 1094 | } | |
| 1097 | 1095 | ||
| @@ -1141,10 +1139,8 @@ func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) { | |||
| 1141 | 1139 | } | |
| 1142 | 1140 | } | |
| 1143 | 1141 | ||
| 1144 | - | // TestRegistrySetDemandIsCountedNotBoolean pins that SetDemand is a balanced | |
| 1145 | - | // counter: two openings require two closings before the schedule reverts to | |
| 1146 | - | // the passive default, exactly as two concurrent client subscriptions would | |
| 1147 | - | // drive it through service.go. | |
| 1142 | + | // TestRegistrySetDemandIsCountedNotBoolean checks that subscriptions balance: | |
| 1143 | + | // observation stops after the last closes and resumes immediately on reopen. | |
| 1148 | 1144 | func TestRegistrySetDemandIsCountedNotBoolean(t *testing.T) { | |
| 1149 | 1145 | start := time.Unix(100, 0) | |
| 1150 | 1146 | clock := newManualClock(start) | |
| @@ -1167,17 +1163,18 @@ func TestRegistrySetDemandIsCountedNotBoolean(t *testing.T) { | |||
| 1167 | 1163 | }) | |
| 1168 | 1164 | require.Equal(t, start.Add(500*time.Millisecond), host.NextDue, "one remaining demand unit must still use the faster cadence") | |
| 1169 | 1165 | ||
| 1170 | - | // The second, and last, subscription closes: demand drops back to zero. | |
| 1166 | + | // The last subscription closes: no scheduled observation runs for nobody. | |
| 1171 | 1167 | registry.SetDemand(false) | |
| 1172 | - | ||
| 1173 | - | clock.Advance(500 * time.Millisecond) | |
| 1174 | - | secondSuccess := start.Add(500 * time.Millisecond) | |
| 1168 | + | registry.SetDemand(false) // balance the fixture's first subscription | |
| 1169 | + | clock.Advance(time.Hour) | |
| 1170 | + | requireNoProbeWithin(t, probe, probeGrace) | |
| 1171 | + | registry.SetDemand(true) | |
| 1175 | 1172 | next := receiveCall(t, probe) | |
| 1176 | 1173 | next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}} | |
| 1177 | 1174 | host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | |
| 1178 | - | return !host.Checking && host.LastSuccess.Equal(secondSuccess) | |
| 1175 | + | return !host.Checking && host.LastSuccess.Equal(clock.Now()) | |
| 1179 | 1176 | }) | |
| 1180 | - | require.Equal(t, secondSuccess.Add(defaultFreshFor), host.NextDue, "no live demand must revert to the passive default cadence") | |
| 1177 | + | require.Equal(t, clock.Now().Add(500*time.Millisecond), host.NextDue) | |
| 1181 | 1178 | } | |
| 1182 | 1179 | ||
| 1183 | 1180 | func TestRegistryDemandOverridesFreshSchedule(t *testing.T) { | |
| @@ -1224,7 +1221,7 @@ func TestRegistryCoalescesConcurrentDemand(t *testing.T) { | |||
| 1224 | 1221 | host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | |
| 1225 | 1222 | return host.Availability == domain.RemoteAvailabilityReachable && host.LastSuccess.Equal(start) | |
| 1226 | 1223 | }) | |
| 1227 | - | require.Equal(t, start.Add(defaultFreshFor), host.NextDue) | |
| 1224 | + | require.Equal(t, start.Add(defaultDemandFreshForRemote), host.NextDue) | |
| 1228 | 1225 | ||
| 1229 | 1226 | // The single coalesced follow-up is admitted once the first completes. | |
| 1230 | 1227 | followUp := receiveCall(t, probe) | |
| @@ -1396,8 +1393,8 @@ func TestRegistryNonReachableOutcomesUseTypedFailureBackoff(t *testing.T) { | |||
| 1396 | 1393 | require.Equal(t, uint64(1), host.FailureEpisode) | |
| 1397 | 1394 | // No observation succeeded yet, so there is no last success. | |
| 1398 | 1395 | require.True(t, host.LastSuccess.IsZero()) | |
| 1399 | - | // Every non-reachable outcome retries on the capped cadence. | |
| 1400 | - | require.Equal(t, start.Add(defaultRetryBase), host.NextDue) | |
| 1396 | + | // A watched host retries on the demand cadence. | |
| 1397 | + | require.Equal(t, start.Add(defaultDemandFreshForRemote), host.NextDue) | |
| 1401 | 1398 | } | |
| 1402 | 1399 | if tt.wantErr != nil { | |
| 1403 | 1400 | require.ErrorIs(t, host.LastFailure, tt.wantErr) | |
| @@ -1464,7 +1461,7 @@ func TestRegistryClassifiesOversizedProbeProjectionAsInvalidResponse(t *testing. | |||
| 1464 | 1461 | // replaces it. | |
| 1465 | 1462 | require.Len(t, host.Sessions, 1) | |
| 1466 | 1463 | require.Equal(t, "work", host.Sessions[0].Name) | |
| 1467 | - | require.Equal(t, clock.Now().Add(defaultRetryBase), host.NextDue) | |
| 1464 | + | require.Equal(t, clock.Now().Add(defaultDemandFreshForRemote), host.NextDue) | |
| 1468 | 1465 | } | |
| 1469 | 1466 | ||
| 1470 | 1467 | func TestRegistryAcceptsProbeProjectionAtSessionBound(t *testing.T) { | |
| @@ -1686,16 +1683,13 @@ func TestRegistryBackoffIsExponentialAndCapped(t *testing.T) { | |||
| 1686 | 1683 | startRegistry(t, registry) | |
| 1687 | 1684 | ||
| 1688 | 1685 | probeErr := errors.New("offline") | |
| 1689 | - | // The retry cadence matches the legacy remotes monitor: a 5s base doubling | |
| 1690 | - | // to a 60s cap. | |
| 1686 | + | // A watched host retries on its two-second demand cadence even as the | |
| 1687 | + | // underlying exponential failure count increases. | |
| 1691 | 1688 | want := []time.Duration{ | |
| 1692 | - | 5 * time.Second, | |
| 1693 | - | 10 * time.Second, | |
| 1694 | - | 20 * time.Second, | |
| 1695 | - | 40 * time.Second, | |
| 1696 | - | time.Minute, // 80s capped at defaultRetryLimit | |
| 1697 | - | time.Minute, | |
| 1698 | - | time.Minute, | |
| 1689 | + | defaultDemandFreshForRemote, defaultDemandFreshForRemote, | |
| 1690 | + | defaultDemandFreshForRemote, defaultDemandFreshForRemote, | |
| 1691 | + | defaultDemandFreshForRemote, defaultDemandFreshForRemote, | |
| 1692 | + | defaultDemandFreshForRemote, | |
| 1699 | 1693 | } | |
| 1700 | 1694 | for index, delay := range want { | |
| 1701 | 1695 | call := receiveCall(t, probe) | |
| @@ -1734,9 +1728,9 @@ func TestRegistryJitterAppliesToRetryAndHealthyRefresh(t *testing.T) { | |||
| 1734 | 1728 | host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | |
| 1735 | 1729 | return !host.Checking && host.Availability == domain.RemoteAvailabilityUnreachable | |
| 1736 | 1730 | }) | |
| 1737 | - | require.Equal(t, start.Add(defaultRetryBase/2), host.NextDue) | |
| 1731 | + | require.Equal(t, start.Add(defaultDemandFreshForRemote/2), host.NextDue) | |
| 1738 | 1732 | ||
| 1739 | - | clock.Advance(defaultRetryBase / 2) | |
| 1733 | + | clock.Advance(defaultDemandFreshForRemote / 2) | |
| 1740 | 1734 | next := receiveCall(t, probe) | |
| 1741 | 1735 | successAt := clock.Now() | |
| 1742 | 1736 | next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}} | |
| @@ -1745,7 +1739,7 @@ func TestRegistryJitterAppliesToRetryAndHealthyRefresh(t *testing.T) { | |||
| 1745 | 1739 | }) | |
| 1746 | 1740 | // The healthy refresh is jittered too: a fleet of hosts never converges | |
| 1747 | 1741 | // on one instant. | |
| 1748 | - | require.Equal(t, successAt.Add(defaultFreshFor/2), host.NextDue) | |
| 1742 | + | require.Equal(t, successAt.Add(defaultDemandFreshForRemote/2), host.NextDue) | |
| 1749 | 1743 | } | |
| 1750 | 1744 | ||
| 1751 | 1745 | func TestRegistryJitterIsDeterministicAndBounded(t *testing.T) { | |
| @@ -1840,6 +1834,7 @@ func TestRegistryFencesRetiredAttemptsAcrossRemovalAndIdenticalReAdd(t *testing. | |||
| 1840 | 1834 | results := make(chan probeResult, 1) | |
| 1841 | 1835 | dispatch := func() { registry.dispatch(ctx, clock.Now(), results) } | |
| 1842 | 1836 | ||
| 1837 | + | registry.RequestProbe(reg.Endpoint) | |
| 1843 | 1838 | dispatch() | |
| 1844 | 1839 | retiredCall := receiveCall(t, probe) | |
| 1845 | 1840 | require.True(t, retiredCall.registration.Equal(reg)) | |
| @@ -1864,6 +1859,7 @@ func TestRegistryFencesRetiredAttemptsAcrossRemovalAndIdenticalReAdd(t *testing. | |||
| 1864 | 1859 | // the attempt token can fence the retired completion. | |
| 1865 | 1860 | require.NoError(t, registry.setHosts(hostRecords(reg))) | |
| 1866 | 1861 | require.Empty(t, registry.Snapshot().Removed) | |
| 1862 | + | registry.RequestProbe(reg.Endpoint) | |
| 1867 | 1863 | dispatch() | |
| 1868 | 1864 | freshCall := receiveCall(t, probe) | |
| 1869 | 1865 | require.True(t, freshCall.registration.Equal(reg)) | |
| @@ -1932,6 +1928,7 @@ func TestRegistryChurnKeepsOneActiveProbePerEndpoint(t *testing.T) { | |||
| 1932 | 1928 | defer cancel() | |
| 1933 | 1929 | done := make(chan struct{}) | |
| 1934 | 1930 | go func() { defer close(done); registry.Run(ctx) }() | |
| 1931 | + | registry.RequestProbe(first.Endpoint) | |
| 1935 | 1932 | ||
| 1936 | 1933 | // Drop the initial attempt's start signal so each lockstep round below | |
| 1937 | 1934 | // consumes the start of its own replacement attempt. | |
| @@ -1953,6 +1950,7 @@ func TestRegistryChurnKeepsOneActiveProbePerEndpoint(t *testing.T) { | |||
| 1953 | 1950 | next = second | |
| 1954 | 1951 | } | |
| 1955 | 1952 | require.NoError(t, registry.setHosts(hostRecords(next))) | |
| 1953 | + | registry.RequestProbe(next.Endpoint) | |
| 1956 | 1954 | select { | |
| 1957 | 1955 | case <-probe.startedSig: | |
| 1958 | 1956 | case <-time.After(time.Second): | |
| @@ -2276,6 +2274,7 @@ func TestRegistryBoundsSlowSubscribersAndStopsOnCancellation(t *testing.T) { | |||
| 2276 | 2274 | ctx, cancel := context.WithCancel(context.Background()) | |
| 2277 | 2275 | done := make(chan struct{}) | |
| 2278 | 2276 | go func() { defer close(done); registry.Run(ctx) }() | |
| 2277 | + | registry.RequestProbe(reg.Endpoint) | |
| 2279 | 2278 | receiveCall(t, probe) | |
| 2280 | 2279 | cancel() | |
| 2281 | 2280 | select { | |
| @@ -2308,6 +2307,7 @@ func TestRegistryRunIsSingleShotAndCleansUpInFlightState(t *testing.T) { | |||
| 2308 | 2307 | ctx, cancel := context.WithCancel(context.Background()) | |
| 2309 | 2308 | done := make(chan struct{}) | |
| 2310 | 2309 | go func() { defer close(done); registry.Run(ctx) }() | |
| 2310 | + | registry.RequestProbe(reg.Endpoint) | |
| 2311 | 2311 | ||
| 2312 | 2312 | call := receiveCall(t, probe) | |
| 2313 | 2313 | require.True(t, waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | |
| @@ -573,10 +573,11 @@ func newTestRegistry(t *testing.T, epoch ports.BrokerEpoch, store ports.BrokerHo | |||||
| 573 | 573 | ||||
| 574 | func startRegistry(t *testing.T, r *Registry) context.CancelFunc { | 574 | func startRegistry(t *testing.T, r *Registry) context.CancelFunc { | ||
| 575 | t.Helper() | 575 | t.Helper() | ||
| 576 | + | r.SetDemand(true) | |||
| 576 | ctx, cancel := context.WithCancel(context.Background()) | 577 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 577 | done := make(chan struct{}) | 578 | done := make(chan struct{}) | ||
| 578 | go func() { defer close(done); r.Run(ctx) }() | 579 | go func() { defer close(done); r.Run(ctx) }() | ||
| 579 | - | t.Cleanup(func() { cancel(); <-done }) | 580 | + | t.Cleanup(func() { cancel(); <-done; r.SetDemand(false) }) |
| 580 | return cancel | 581 | return cancel | ||
| 581 | } | 582 | } | ||
| 582 | 583 | ||||
| @@ -1060,22 +1061,21 @@ func TestRegistrySchedulesFreshObservationAfterSuccess(t *testing.T) { | |||||
| 1060 | }) | 1061 | }) | ||
| 1061 | require.Equal(t, start, host.LastAttempt) | 1062 | require.Equal(t, start, host.LastAttempt) | ||
| 1062 | require.Equal(t, start, host.LastSuccess) | 1063 | require.Equal(t, start, host.LastSuccess) | ||
| 1063 | - | require.Equal(t, start.Add(defaultFreshFor), host.NextDue) | 1064 | + | require.Equal(t, start.Add(defaultDemandFreshForRemote), host.NextDue) |
| 1064 | require.Zero(t, host.ConsecutiveFailures) | 1065 | require.Zero(t, host.ConsecutiveFailures) | ||
| 1065 | require.Zero(t, host.FailureEpisode) | 1066 | require.Zero(t, host.FailureEpisode) | ||
| 1066 | 1067 | ||||
| 1067 | // A fresh host is not observed again before its scheduled deadline. | 1068 | // A fresh host is not observed again before its scheduled deadline. | ||
| 1068 | - | clock.Advance(defaultFreshFor - time.Nanosecond) | 1069 | + | clock.Advance(defaultDemandFreshForRemote - time.Nanosecond) |
| 1069 | requireNoProbeWithin(t, probe, probeGrace) | 1070 | requireNoProbeWithin(t, probe, probeGrace) | ||
| 1070 | clock.Advance(time.Nanosecond) | 1071 | clock.Advance(time.Nanosecond) | ||
| 1071 | next := receiveCall(t, probe) | 1072 | next := receiveCall(t, probe) | ||
| 1072 | require.True(t, next.registration.Equal(reg)) | 1073 | require.True(t, next.registration.Equal(reg)) | ||
| 1073 | } | 1074 | } | ||
| 1074 | 1075 | ||||
| 1075 | - | // TestRegistryAppliesSubscriptionDemandFreshness pins C2: apply/applyLocal | 1076 | + | // TestRegistryAppliesSubscriptionDemandFreshness checks that apply/applyLocal |
| 1076 | // schedule NextDue on RegistryConfig's demand cadence while Registry.SetDemand | 1077 | // schedule NextDue on RegistryConfig's demand cadence while Registry.SetDemand | ||
| 1077 | - | // reports a live client subscription, and keep the passive defaultFreshFor | 1078 | + | // reports a live client subscription. It exercises both the remote and the local |
| 1078 | - | // once demand drops back to zero. It exercises both the remote and the local | |||
| 1079 | // observation paths on the same registry, since service.go's Subscribe/Close | 1079 | // observation paths on the same registry, since service.go's Subscribe/Close | ||
| 1080 | // wiring feeds one shared demand counter for both. | 1080 | // wiring feeds one shared demand counter for both. | ||
| 1081 | func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) { | 1081 | func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) { | ||
| @@ -1089,9 +1089,7 @@ func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) { | |||||
| 1089 | demand bool | 1089 | demand bool | ||
| 1090 | wantAfter time.Duration | 1090 | wantAfter time.Duration | ||
| 1091 | }{ | 1091 | }{ | ||
| 1092 | - | {name: "remote no demand keeps default", local: false, demand: false, wantAfter: defaultFreshFor}, | |||
| 1093 | {name: "remote demand uses faster cadence", local: false, demand: true, wantAfter: demandRemote}, | 1092 | {name: "remote demand uses faster cadence", local: false, demand: true, wantAfter: demandRemote}, | ||
| 1094 | - | {name: "local no demand keeps default", local: true, demand: false, wantAfter: defaultFreshFor}, | |||
| 1095 | {name: "local demand uses faster cadence", local: true, demand: true, wantAfter: demandLocal}, | 1093 | {name: "local demand uses faster cadence", local: true, demand: true, wantAfter: demandLocal}, | ||
| 1096 | } | 1094 | } | ||
| 1097 | 1095 | ||||
| @@ -1141,10 +1139,8 @@ func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) { | |||||
| 1141 | } | 1139 | } | ||
| 1142 | } | 1140 | } | ||
| 1143 | 1141 | ||||
| 1144 | - | // TestRegistrySetDemandIsCountedNotBoolean pins that SetDemand is a balanced | 1142 | + | // TestRegistrySetDemandIsCountedNotBoolean checks that subscriptions balance: |
| 1145 | - | // counter: two openings require two closings before the schedule reverts to | 1143 | + | // observation stops after the last closes and resumes immediately on reopen. |
| 1146 | - | // the passive default, exactly as two concurrent client subscriptions would | |||
| 1147 | - | // drive it through service.go. | |||
| 1148 | func TestRegistrySetDemandIsCountedNotBoolean(t *testing.T) { | 1144 | func TestRegistrySetDemandIsCountedNotBoolean(t *testing.T) { | ||
| 1149 | start := time.Unix(100, 0) | 1145 | start := time.Unix(100, 0) | ||
| 1150 | clock := newManualClock(start) | 1146 | clock := newManualClock(start) | ||
| @@ -1167,17 +1163,18 @@ func TestRegistrySetDemandIsCountedNotBoolean(t *testing.T) { | |||||
| 1167 | }) | 1163 | }) | ||
| 1168 | require.Equal(t, start.Add(500*time.Millisecond), host.NextDue, "one remaining demand unit must still use the faster cadence") | 1164 | require.Equal(t, start.Add(500*time.Millisecond), host.NextDue, "one remaining demand unit must still use the faster cadence") | ||
| 1169 | 1165 | ||||
| 1170 | - | // The second, and last, subscription closes: demand drops back to zero. | 1166 | + | // The last subscription closes: no scheduled observation runs for nobody. |
| 1171 | registry.SetDemand(false) | 1167 | registry.SetDemand(false) | ||
| 1172 | - | 1168 | + | registry.SetDemand(false) // balance the fixture's first subscription | |
| 1173 | - | clock.Advance(500 * time.Millisecond) | 1169 | + | clock.Advance(time.Hour) |
| 1174 | - | secondSuccess := start.Add(500 * time.Millisecond) | 1170 | + | requireNoProbeWithin(t, probe, probeGrace) |
| 1171 | + | registry.SetDemand(true) | |||
| 1175 | next := receiveCall(t, probe) | 1172 | next := receiveCall(t, probe) | ||
| 1176 | next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}} | 1173 | next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}} | ||
| 1177 | host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | 1174 | host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | ||
| 1178 | - | return !host.Checking && host.LastSuccess.Equal(secondSuccess) | 1175 | + | return !host.Checking && host.LastSuccess.Equal(clock.Now()) |
| 1179 | }) | 1176 | }) | ||
| 1180 | - | require.Equal(t, secondSuccess.Add(defaultFreshFor), host.NextDue, "no live demand must revert to the passive default cadence") | 1177 | + | require.Equal(t, clock.Now().Add(500*time.Millisecond), host.NextDue) |
| 1181 | } | 1178 | } | ||
| 1182 | 1179 | ||||
| 1183 | func TestRegistryDemandOverridesFreshSchedule(t *testing.T) { | 1180 | func TestRegistryDemandOverridesFreshSchedule(t *testing.T) { | ||
| @@ -1224,7 +1221,7 @@ func TestRegistryCoalescesConcurrentDemand(t *testing.T) { | |||||
| 1224 | host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | 1221 | host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | ||
| 1225 | return host.Availability == domain.RemoteAvailabilityReachable && host.LastSuccess.Equal(start) | 1222 | return host.Availability == domain.RemoteAvailabilityReachable && host.LastSuccess.Equal(start) | ||
| 1226 | }) | 1223 | }) | ||
| 1227 | - | require.Equal(t, start.Add(defaultFreshFor), host.NextDue) | 1224 | + | require.Equal(t, start.Add(defaultDemandFreshForRemote), host.NextDue) |
| 1228 | 1225 | ||||
| 1229 | // The single coalesced follow-up is admitted once the first completes. | 1226 | // The single coalesced follow-up is admitted once the first completes. | ||
| 1230 | followUp := receiveCall(t, probe) | 1227 | followUp := receiveCall(t, probe) | ||
| @@ -1396,8 +1393,8 @@ func TestRegistryNonReachableOutcomesUseTypedFailureBackoff(t *testing.T) { | |||||
| 1396 | require.Equal(t, uint64(1), host.FailureEpisode) | 1393 | require.Equal(t, uint64(1), host.FailureEpisode) | ||
| 1397 | // No observation succeeded yet, so there is no last success. | 1394 | // No observation succeeded yet, so there is no last success. | ||
| 1398 | require.True(t, host.LastSuccess.IsZero()) | 1395 | require.True(t, host.LastSuccess.IsZero()) | ||
| 1399 | - | // Every non-reachable outcome retries on the capped cadence. | 1396 | + | // A watched host retries on the demand cadence. |
| 1400 | - | require.Equal(t, start.Add(defaultRetryBase), host.NextDue) | 1397 | + | require.Equal(t, start.Add(defaultDemandFreshForRemote), host.NextDue) |
| 1401 | } | 1398 | } | ||
| 1402 | if tt.wantErr != nil { | 1399 | if tt.wantErr != nil { | ||
| 1403 | require.ErrorIs(t, host.LastFailure, tt.wantErr) | 1400 | require.ErrorIs(t, host.LastFailure, tt.wantErr) | ||
| @@ -1464,7 +1461,7 @@ func TestRegistryClassifiesOversizedProbeProjectionAsInvalidResponse(t *testing. | |||||
| 1464 | // replaces it. | 1461 | // replaces it. | ||
| 1465 | require.Len(t, host.Sessions, 1) | 1462 | require.Len(t, host.Sessions, 1) | ||
| 1466 | require.Equal(t, "work", host.Sessions[0].Name) | 1463 | require.Equal(t, "work", host.Sessions[0].Name) | ||
| 1467 | - | require.Equal(t, clock.Now().Add(defaultRetryBase), host.NextDue) | 1464 | + | require.Equal(t, clock.Now().Add(defaultDemandFreshForRemote), host.NextDue) |
| 1468 | } | 1465 | } | ||
| 1469 | 1466 | ||||
| 1470 | func TestRegistryAcceptsProbeProjectionAtSessionBound(t *testing.T) { | 1467 | func TestRegistryAcceptsProbeProjectionAtSessionBound(t *testing.T) { | ||
| @@ -1686,16 +1683,13 @@ func TestRegistryBackoffIsExponentialAndCapped(t *testing.T) { | |||||
| 1686 | startRegistry(t, registry) | 1683 | startRegistry(t, registry) | ||
| 1687 | 1684 | ||||
| 1688 | probeErr := errors.New("offline") | 1685 | probeErr := errors.New("offline") | ||
| 1689 | - | // The retry cadence matches the legacy remotes monitor: a 5s base doubling | 1686 | + | // A watched host retries on its two-second demand cadence even as the |
| 1690 | - | // to a 60s cap. | 1687 | + | // underlying exponential failure count increases. |
| 1691 | want := []time.Duration{ | 1688 | want := []time.Duration{ | ||
| 1692 | - | 5 * time.Second, | 1689 | + | defaultDemandFreshForRemote, defaultDemandFreshForRemote, |
| 1693 | - | 10 * time.Second, | 1690 | + | defaultDemandFreshForRemote, defaultDemandFreshForRemote, |
| 1694 | - | 20 * time.Second, | 1691 | + | defaultDemandFreshForRemote, defaultDemandFreshForRemote, |
| 1695 | - | 40 * time.Second, | 1692 | + | defaultDemandFreshForRemote, |
| 1696 | - | time.Minute, // 80s capped at defaultRetryLimit | |||
| 1697 | - | time.Minute, | |||
| 1698 | - | time.Minute, | |||
| 1699 | } | 1693 | } | ||
| 1700 | for index, delay := range want { | 1694 | for index, delay := range want { | ||
| 1701 | call := receiveCall(t, probe) | 1695 | call := receiveCall(t, probe) | ||
| @@ -1734,9 +1728,9 @@ func TestRegistryJitterAppliesToRetryAndHealthyRefresh(t *testing.T) { | |||||
| 1734 | host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | 1728 | host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | ||
| 1735 | return !host.Checking && host.Availability == domain.RemoteAvailabilityUnreachable | 1729 | return !host.Checking && host.Availability == domain.RemoteAvailabilityUnreachable | ||
| 1736 | }) | 1730 | }) | ||
| 1737 | - | require.Equal(t, start.Add(defaultRetryBase/2), host.NextDue) | 1731 | + | require.Equal(t, start.Add(defaultDemandFreshForRemote/2), host.NextDue) |
| 1738 | 1732 | ||||
| 1739 | - | clock.Advance(defaultRetryBase / 2) | 1733 | + | clock.Advance(defaultDemandFreshForRemote / 2) |
| 1740 | next := receiveCall(t, probe) | 1734 | next := receiveCall(t, probe) | ||
| 1741 | successAt := clock.Now() | 1735 | successAt := clock.Now() | ||
| 1742 | next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}} | 1736 | next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}} | ||
| @@ -1745,7 +1739,7 @@ func TestRegistryJitterAppliesToRetryAndHealthyRefresh(t *testing.T) { | |||||
| 1745 | }) | 1739 | }) | ||
| 1746 | // The healthy refresh is jittered too: a fleet of hosts never converges | 1740 | // The healthy refresh is jittered too: a fleet of hosts never converges | ||
| 1747 | // on one instant. | 1741 | // on one instant. | ||
| 1748 | - | require.Equal(t, successAt.Add(defaultFreshFor/2), host.NextDue) | 1742 | + | require.Equal(t, successAt.Add(defaultDemandFreshForRemote/2), host.NextDue) |
| 1749 | } | 1743 | } | ||
| 1750 | 1744 | ||||
| 1751 | func TestRegistryJitterIsDeterministicAndBounded(t *testing.T) { | 1745 | func TestRegistryJitterIsDeterministicAndBounded(t *testing.T) { | ||
| @@ -1840,6 +1834,7 @@ func TestRegistryFencesRetiredAttemptsAcrossRemovalAndIdenticalReAdd(t *testing. | |||||
| 1840 | results := make(chan probeResult, 1) | 1834 | results := make(chan probeResult, 1) | ||
| 1841 | dispatch := func() { registry.dispatch(ctx, clock.Now(), results) } | 1835 | dispatch := func() { registry.dispatch(ctx, clock.Now(), results) } | ||
| 1842 | 1836 | ||||
| 1837 | + | registry.RequestProbe(reg.Endpoint) | |||
| 1843 | dispatch() | 1838 | dispatch() | ||
| 1844 | retiredCall := receiveCall(t, probe) | 1839 | retiredCall := receiveCall(t, probe) | ||
| 1845 | require.True(t, retiredCall.registration.Equal(reg)) | 1840 | require.True(t, retiredCall.registration.Equal(reg)) | ||
| @@ -1864,6 +1859,7 @@ func TestRegistryFencesRetiredAttemptsAcrossRemovalAndIdenticalReAdd(t *testing. | |||||
| 1864 | // the attempt token can fence the retired completion. | 1859 | // the attempt token can fence the retired completion. | ||
| 1865 | require.NoError(t, registry.setHosts(hostRecords(reg))) | 1860 | require.NoError(t, registry.setHosts(hostRecords(reg))) | ||
| 1866 | require.Empty(t, registry.Snapshot().Removed) | 1861 | require.Empty(t, registry.Snapshot().Removed) | ||
| 1862 | + | registry.RequestProbe(reg.Endpoint) | |||
| 1867 | dispatch() | 1863 | dispatch() | ||
| 1868 | freshCall := receiveCall(t, probe) | 1864 | freshCall := receiveCall(t, probe) | ||
| 1869 | require.True(t, freshCall.registration.Equal(reg)) | 1865 | require.True(t, freshCall.registration.Equal(reg)) | ||
| @@ -1932,6 +1928,7 @@ func TestRegistryChurnKeepsOneActiveProbePerEndpoint(t *testing.T) { | |||||
| 1932 | defer cancel() | 1928 | defer cancel() | ||
| 1933 | done := make(chan struct{}) | 1929 | done := make(chan struct{}) | ||
| 1934 | go func() { defer close(done); registry.Run(ctx) }() | 1930 | go func() { defer close(done); registry.Run(ctx) }() | ||
| 1931 | + | registry.RequestProbe(first.Endpoint) | |||
| 1935 | 1932 | ||||
| 1936 | // Drop the initial attempt's start signal so each lockstep round below | 1933 | // Drop the initial attempt's start signal so each lockstep round below | ||
| 1937 | // consumes the start of its own replacement attempt. | 1934 | // consumes the start of its own replacement attempt. | ||
| @@ -1953,6 +1950,7 @@ func TestRegistryChurnKeepsOneActiveProbePerEndpoint(t *testing.T) { | |||||
| 1953 | next = second | 1950 | next = second | ||
| 1954 | } | 1951 | } | ||
| 1955 | require.NoError(t, registry.setHosts(hostRecords(next))) | 1952 | require.NoError(t, registry.setHosts(hostRecords(next))) | ||
| 1953 | + | registry.RequestProbe(next.Endpoint) | |||
| 1956 | select { | 1954 | select { | ||
| 1957 | case <-probe.startedSig: | 1955 | case <-probe.startedSig: | ||
| 1958 | case <-time.After(time.Second): | 1956 | case <-time.After(time.Second): | ||
| @@ -2276,6 +2274,7 @@ func TestRegistryBoundsSlowSubscribersAndStopsOnCancellation(t *testing.T) { | |||||
| 2276 | ctx, cancel := context.WithCancel(context.Background()) | 2274 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 2277 | done := make(chan struct{}) | 2275 | done := make(chan struct{}) | ||
| 2278 | go func() { defer close(done); registry.Run(ctx) }() | 2276 | go func() { defer close(done); registry.Run(ctx) }() | ||
| 2277 | + | registry.RequestProbe(reg.Endpoint) | |||
| 2279 | receiveCall(t, probe) | 2278 | receiveCall(t, probe) | ||
| 2280 | cancel() | 2279 | cancel() | ||
| 2281 | select { | 2280 | select { | ||
| @@ -2308,6 +2307,7 @@ func TestRegistryRunIsSingleShotAndCleansUpInFlightState(t *testing.T) { | |||||
| 2308 | ctx, cancel := context.WithCancel(context.Background()) | 2307 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 2309 | done := make(chan struct{}) | 2308 | done := make(chan struct{}) | ||
| 2310 | go func() { defer close(done); registry.Run(ctx) }() | 2309 | go func() { defer close(done); registry.Run(ctx) }() | ||
| 2310 | + | registry.RequestProbe(reg.Endpoint) | |||
| 2311 | 2311 | ||||
| 2312 | call := receiveCall(t, probe) | 2312 | call := receiveCall(t, probe) | ||
| 2313 | require.True(t, waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | 2313 | require.True(t, waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool { | ||
M internal/usecase/broker/service.go +2 -3
| @@ -227,9 +227,8 @@ func (s *Service) Snapshot() ports.BrokerSnapshot { return s.registry.Snapshot() | |||
| 227 | 227 | // Close releases every subscription this connection still owns. A closed | |
| 228 | 228 | // connection refuses new subscriptions. Opening a subscription also records | |
| 229 | 229 | // one unit of registry demand (Registry.SetDemand(true)): while a client | |
| 230 | - | // keeps watching the snapshot, the registry re-probes on the faster demand | |
| 231 | - | // cadence instead of the passive default. serviceSubscription.Close balances | |
| 232 | - | // it with SetDemand(false). | |
| 230 | + | // watches the snapshot, the registry observes on the demand cadence. Without | |
| 231 | + | // a subscriber it performs no scheduled observations. Close balances demand. | |
| 233 | 232 | func (s *Service) Subscribe() (ports.BrokerSubscription, error) { | |
| 234 | 233 | s.mu.Lock() | |
| 235 | 234 | if s.closed { | |
| @@ -227,9 +227,8 @@ func (s *Service) Snapshot() ports.BrokerSnapshot { return s.registry.Snapshot() | |||||
| 227 | // Close releases every subscription this connection still owns. A closed | 227 | // Close releases every subscription this connection still owns. A closed | ||
| 228 | // connection refuses new subscriptions. Opening a subscription also records | 228 | // connection refuses new subscriptions. Opening a subscription also records | ||
| 229 | // one unit of registry demand (Registry.SetDemand(true)): while a client | 229 | // one unit of registry demand (Registry.SetDemand(true)): while a client | ||
| 230 | - | // keeps watching the snapshot, the registry re-probes on the faster demand | 230 | + | // watches the snapshot, the registry observes on the demand cadence. Without |
| 231 | - | // cadence instead of the passive default. serviceSubscription.Close balances | 231 | + | // a subscriber it performs no scheduled observations. Close balances demand. |
| 232 | - | // it with SetDemand(false). | |||
| 233 | func (s *Service) Subscribe() (ports.BrokerSubscription, error) { | 232 | func (s *Service) Subscribe() (ports.BrokerSubscription, error) { | ||
| 234 | s.mu.Lock() | 233 | s.mu.Lock() | ||
| 235 | if s.closed { | 234 | if s.closed { | ||
M internal/usecase/broker/service_test.go +4 -4
| @@ -336,10 +336,9 @@ func TestServiceOpenStreamRequestsLocalReprobeOnOpenAndClose(t *testing.T) { | |||
| 336 | 336 | require.NoError(t, err) | |
| 337 | 337 | t.Cleanup(func() { require.NoError(t, service.Close()) }) | |
| 338 | 338 | ||
| 339 | - | // The registry's first dispatch (Run start) already probes the local | |
| 340 | - | // daemon because it has never been observed: settle it so NextDue lands | |
| 341 | - | // far in the future (defaultFreshFor, no live demand yet), and only an | |
| 342 | - | // explicit reconcile can explain any further attempt. | |
| 339 | + | // Explicitly request the initial observation; admission alone does not | |
| 340 | + | // subscribe to snapshots and must not cause background polling. | |
| 341 | + | registry.RequestProbe("") | |
| 343 | 342 | call := receiveLocalCall(t, local) | |
| 344 | 343 | call.result <- reachable() | |
| 345 | 344 | waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking }) | |
| @@ -739,6 +738,7 @@ func TestServiceRequestReconcileDelegatesToRegistry(t *testing.T) { | |||
| 739 | 738 | require.NoError(t, err) | |
| 740 | 739 | t.Cleanup(func() { require.NoError(t, service.Close()) }) | |
| 741 | 740 | ||
| 741 | + | registry.RequestProbe(registration.Endpoint) | |
| 742 | 742 | var initial probeCall | |
| 743 | 743 | require.Eventually(t, func() bool { | |
| 744 | 744 | select { | |
| @@ -336,10 +336,9 @@ func TestServiceOpenStreamRequestsLocalReprobeOnOpenAndClose(t *testing.T) { | |||||
| 336 | require.NoError(t, err) | 336 | require.NoError(t, err) | ||
| 337 | t.Cleanup(func() { require.NoError(t, service.Close()) }) | 337 | t.Cleanup(func() { require.NoError(t, service.Close()) }) | ||
| 338 | 338 | ||||
| 339 | - | // The registry's first dispatch (Run start) already probes the local | 339 | + | // Explicitly request the initial observation; admission alone does not |
| 340 | - | // daemon because it has never been observed: settle it so NextDue lands | 340 | + | // subscribe to snapshots and must not cause background polling. |
| 341 | - | // far in the future (defaultFreshFor, no live demand yet), and only an | 341 | + | registry.RequestProbe("") |
| 342 | - | // explicit reconcile can explain any further attempt. | |||
| 343 | call := receiveLocalCall(t, local) | 342 | call := receiveLocalCall(t, local) | ||
| 344 | call.result <- reachable() | 343 | call.result <- reachable() | ||
| 345 | waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking }) | 344 | waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking }) | ||
| @@ -739,6 +738,7 @@ func TestServiceRequestReconcileDelegatesToRegistry(t *testing.T) { | |||||
| 739 | require.NoError(t, err) | 738 | require.NoError(t, err) | ||
| 740 | t.Cleanup(func() { require.NoError(t, service.Close()) }) | 739 | t.Cleanup(func() { require.NoError(t, service.Close()) }) | ||
| 741 | 740 | ||||
| 741 | + | registry.RequestProbe(registration.Endpoint) | |||
| 742 | var initial probeCall | 742 | var initial probeCall | ||
| 743 | require.Eventually(t, func() bool { | 743 | require.Eventually(t, func() bool { | ||
| 744 | select { | 744 | select { | ||