1package broker
2
3import (
4 "context"
5 "encoding/binary"
6 "errors"
7 "hash/fnv"
8 "time"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12)
13
14// cancelInflightLocked cancels the in-flight attempt for one endpoint and
15// drops its record. Callers must hold r.mu: the retired attempt's context is
16// cancelled before the record disappears, so no replacement attempt can start
17// while the retired probe's context is still live.
18func (r *Registry) cancelInflightLocked(endpoint string) {
19 if attempt := r.inflight[endpoint]; attempt != nil {
20 attempt.cancel()
21 delete(r.inflight, endpoint)
22 }
23}
24
25// cancelAllInflightLocked cancels every in-flight attempt and drops the
26// records. Callers must hold r.mu.
27func (r *Registry) cancelAllInflightLocked() {
28 for endpoint, attempt := range r.inflight {
29 attempt.cancel()
30 delete(r.inflight, endpoint)
31 }
32}
33
34// RequestProbe is non-blocking and coalesces concurrent demand per endpoint.
35// A disabled registry never observes, so the demand is dropped without
36// scheduling anything. An empty endpoint requests a local re-probe instead of a
37// configured remote: the broker's own machine daemon is re-observed ahead of
38// its scheduled freshness window, honored by dispatchLocalLocked exactly like
39// pending honors remote demand. A registry with no local observation producer
40// drops it, just like an unknown remote endpoint is dropped.
41func (r *Registry) RequestProbe(endpoint string) {
42 if r.observationDisabled {
43 return
44 }
45 r.mu.Lock()
46 if endpoint == "" {
47 if r.local != nil {
48 r.localPending = true
49 }
50 } else if _, ok := r.hosts[endpoint]; ok {
51 r.pending[endpoint] = true
52 }
53 r.mu.Unlock()
54 r.hint()
55}
56
57// SetDemand records one live client subscription opening (active=true) or
58// closing (active=false). internal/usecase/broker/service.go calls it from
59// Service.Subscribe and serviceSubscription.Close, so the demand count tracks
60// exactly the connections a client is actively watching. The first
61// subscription requests one immediate observation per host; while demand
62// stays positive, the demand cadence schedules later observations. With no
63// subscription, only explicit RequestProbe calls dispatch. A disabled registry
64// still counts subscriptions but never observes.
65func (r *Registry) SetDemand(active bool) {
66 r.mu.Lock()
67 if active {
68 if r.demand == 0 {
69 r.localPending = r.local != nil
70 for endpoint := range r.hosts {
71 r.pending[endpoint] = true
72 }
73 }
74 r.demand++
75 } else if r.demand > 0 {
76 r.demand--
77 }
78 r.mu.Unlock()
79 r.hint()
80}
81
82// freshForLocked returns the demand cadence while subscribed. An absent
83// daemon uses the longer freshFor interval instead. Callers must hold r.mu.
84func (r *Registry) freshForLocked(local bool) time.Duration {
85 if r.demand <= 0 {
86 return r.freshFor
87 }
88 if local {
89 return r.demandFreshForLocal
90 }
91 return r.demandFreshForRemote
92}
93
94func (r *Registry) hint() {
95 select {
96 case r.wake <- struct{}{}:
97 default:
98 }
99}
100
101// Run schedules observations until cancellation. Run owns the registry
102// lifecycle and may execute exactly once: a second call is refused so a
103// concurrent or restarted caller can never duplicate scheduling. On return it
104// clears in-flight observation state, publishes the settled projection, and
105// flushes durable persistence before any further write is refused.
106//
107// An observation-disabled registry runs the same lifecycle and drain but never
108// schedules observation: it arms no timer, starts no probe, and drops every
109// reconcile. It still owns and drains its durable writer, so a shutdown never
110// returns before the newest staged publication has reached the store.
111func (r *Registry) Run(ctx context.Context) {
112 if !r.running.CompareAndSwap(runIdle, runActive) {
113 r.log.Warn("broker: registry Run refused; single-run already started or finished")
114 return
115 }
116 defer r.settle()
117 if r.observationDisabled {
118 <-ctx.Done()
119 return
120 }
121 timer := r.clock.NewTimer(0)
122 defer timer.Stop()
123 results := make(chan probeResult, ports.BrokerMaxHosts)
124 for {
125 now := r.clock.Now()
126 r.dispatch(ctx, now, results)
127 timer.Reset(r.nextDelay(now))
128 select {
129 case <-ctx.Done():
130 return
131 case <-r.wake:
132 case <-timer.C():
133 case result := <-results:
134 if ctx.Err() != nil {
135 // Shutdown wins: a cancellation is never recorded as an
136 // observation failure.
137 return
138 }
139 r.apply(result)
140 }
141 }
142}
143
144// settle is Run's cleanup: nothing stays in flight, the published projection
145// carries no transient Checking flag, durable persistence is flushed, and
146// every later Run call is refused.
147func (r *Registry) settle() {
148 r.mu.Lock()
149 r.cancelAllInflightLocked()
150 if r.localAttempt != nil {
151 r.localAttempt.cancel()
152 r.localAttempt = nil
153 }
154 cleared := false
155 if r.local != nil && r.localHost.Checking {
156 r.localHost.Checking = false
157 cleared = true
158 }
159 for endpoint, host := range r.hosts {
160 if !host.Checking {
161 continue
162 }
163 host.Checking = false
164 r.hosts[endpoint] = host
165 cleared = true
166 }
167 if cleared {
168 r.publishLocked(true)
169 }
170 r.running.Store(runStopped)
171 r.mu.Unlock()
172 r.store.close()
173}
174
175// probeAttempt is one admitted observation: the registry-wide token that
176// fences its completion, and the cancel for the context derived from the run
177// context. Cancelling the derived context retires the attempt: a probe that
178// honors cancellation returns promptly, so a replaced or removed endpoint
179// never accumulates overlapping probes, and settle releases every attempt
180// before the run loop exits.
181type probeAttempt struct {
182 token uint64
183 cancel context.CancelFunc
184}
185
186// probeResult is one completed observation bound to the exact attempt token
187// that started it.
188type probeResult struct {
189 endpoint string
190 registration domain.RemoteRegistration
191 attempt uint64
192 snapshot ports.BrokerDaemonObservation
193 err error
194 at time.Time
195 // local marks the result of one local probe attempt, which carries no
196 // endpoint or registration and is applied to the configured local entry.
197 local bool
198}
199
200func (r *Registry) dispatch(ctx context.Context, now time.Time, results chan<- probeResult) {
201 r.mu.Lock()
202 if r.demand == 0 && len(r.pending) == 0 && !r.localPending {
203 r.mu.Unlock()
204 return
205 }
206 started := false
207 for endpoint, host := range r.hosts {
208 if _, inflight := r.inflight[endpoint]; inflight {
209 continue
210 }
211 explicit := r.pending[endpoint]
212 if !explicit && (r.demand == 0 || !host.NextDue.IsZero() && now.Before(host.NextDue)) {
213 continue
214 }
215 delete(r.pending, endpoint)
216 r.attempts++
217 attempt := r.attempts
218 probeCtx, cancel := context.WithCancel(ctx)
219 r.inflight[endpoint] = &probeAttempt{token: attempt, cancel: cancel}
220 if visibleCheck(host, explicit) {
221 started = true
222 host.Checking = true
223 }
224 host.LastAttempt = now
225 r.hosts[endpoint] = host
226 registration := host.Registration
227 // The deadline is armed here, under r.mu and at dispatch time, so the
228 // attempt's bound starts with the attempt rather than whenever its
229 // goroutine is scheduled.
230 deadline := r.clock.NewTimer(r.probeTimeout)
231 go func(endpoint string, registration domain.RemoteRegistration, attempt uint64, probeCtx context.Context) {
232 snapshot, err := observeBounded(probeCtx, deadline, func(ctx context.Context) (ports.BrokerDaemonObservation, error) {
233 return r.probe.Probe(ctx, registration)
234 })
235 result := probeResult{endpoint: endpoint, registration: registration, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}
236 select {
237 case results <- result:
238 case <-probeCtx.Done():
239 // The attempt was retired (replaced, removed, or shutdown), so
240 // its completion is already fenced: dropping it keeps churn from
241 // queueing results the run loop would only discard.
242 }
243 }(endpoint, registration, attempt, probeCtx)
244 }
245 // Only an observation that visibly started changes the projection; an
246 // idle dispatch or a routine refresh must not bump the revision or wake
247 // subscribers. The in-flight projection is never persisted: Checking is
248 // transient.
249 if r.dispatchLocalLocked(ctx, now, results) {
250 started = true
251 }
252 if started {
253 r.publishLocked(false)
254 }
255 r.mu.Unlock()
256}
257
258// visibleCheck reports whether starting an observation publishes the transient
259// Checking state. A routine scheduled refresh of a reachable daemon stays
260// silent until its result lands, so an idle watched broker publishes once per
261// observation instead of twice; an explicit request or a daemon that is not
262// confirmed reachable still shows that it is being checked.
263func visibleCheck(host ports.BrokerDaemonObservation, explicit bool) bool {
264 return explicit || host.Availability != domain.RemoteAvailabilityReachable
265}
266
267// observeBounded runs one observation under its attempt deadline. retire is the
268// attempt's retirement context: a replaced, removed, or settled attempt returns
269// at once and its completion is fenced by the caller. The probe itself runs on
270// a context derived from retire that the deadline also cancels, so a
271// ctx-honoring probe unwinds its dial, handshake, or exchange when either
272// fires. The attempt never waits for the probe past its deadline: a probe that
273// ignores cancellation is abandoned (its late result is discarded) and the
274// attempt still reports errProbeTimeout, so the endpoint's single in-flight
275// slot is always released.
276func observeBounded(retire context.Context, deadline ports.Timer, observe func(context.Context) (ports.BrokerDaemonObservation, error)) (ports.BrokerDaemonObservation, error) {
277 defer deadline.Stop()
278 probeCtx, cancel := context.WithCancelCause(retire)
279 defer cancel(nil)
280 type outcome struct {
281 snapshot ports.BrokerDaemonObservation
282 err error
283 }
284 done := make(chan outcome, 1)
285 go func() {
286 snapshot, err := observe(probeCtx)
287 done <- outcome{snapshot: snapshot, err: err}
288 }()
289 select {
290 case result := <-done:
291 return result.snapshot, result.err
292 case <-deadline.C():
293 cancel(errProbeTimeout)
294 return ports.BrokerDaemonObservation{}, errProbeTimeout
295 case <-retire.Done():
296 return ports.BrokerDaemonObservation{}, retire.Err()
297 }
298}
299
300func (r *Registry) apply(result probeResult) {
301 r.mu.Lock()
302 defer r.mu.Unlock()
303 if result.local {
304 r.applyLocal(result)
305 return
306 }
307 current, ok := r.hosts[result.endpoint]
308 attempt := r.inflight[result.endpoint]
309 if !ok || !current.Registration.Equal(result.registration) || attempt == nil || attempt.token != result.attempt {
310 // A stale completion touches neither current authority nor the
311 // in-flight slot, which may already belong to a newer registration or
312 // attempt.
313 return
314 }
315 delete(r.inflight, result.endpoint)
316 // The probe has returned, so releasing its derived context here only
317 // frees the attempt's resources; the record is gone before any new
318 // attempt can start.
319 attempt.cancel()
320 current.Checking = false
321 current.LastAttempt = result.at
322
323 kind, availability, failed := observationOutcome(result)
324 if !failed && availability == domain.RemoteAvailabilityNoDaemon {
325 current.Availability = availability
326 current.LastSuccess = result.at
327 // An absent daemon is stable: demand does not require a fresh SSH
328 // handshake every two seconds just to confirm it is still absent.
329 current.NextDue = result.at.Add(r.jitter(r.freshFor, result.endpoint, result.attempt))
330 current.ConsecutiveFailures = 0
331 current.LastFailure = domain.RemoteFailure{}
332 current.Identity = ""
333 current.Incarnation = ports.BrokerDaemonIncarnation{}
334 current.ProtocolVersion = 0
335 current.Capabilities = 0
336 current.InventoryKnown = false
337 current.Sessions = nil
338 r.hosts[result.endpoint] = current
339 r.publishLocked(true)
340 return
341 }
342 if !failed {
343 // Configured authority comes only from the registry's own projection:
344 // a probe result supplies observed state, never endpoint, registration,
345 // policy, display origin, or rank.
346 observed := stampObservationAuthority(result.snapshot.Clone(), current)
347 observed.Checking = false
348 observed.LastAttempt = result.at
349 observed.LastSuccess = result.at
350 observed.NextDue = result.at.Add(r.jitter(r.freshForLocked(false), result.endpoint, result.attempt))
351 observed.ConsecutiveFailures = 0
352 observed.LastFailure = domain.RemoteFailure{}
353 observed.FailureEpisode = current.FailureEpisode
354 // The published observation must be fully valid once authority is
355 // stamped: a probe that invents partial identity (identity without a
356 // protocol version) or an inventory without a known catalogue is an
357 // invalid response, never a published observation. The durable rule is
358 // re-checked too: a projection the offline store would reject is never
359 // published as if it had persisted. The store applies the same rule.
360 if err := observed.Validate(); err != nil {
361 r.log.Debug("broker: observation is invalid", "endpoint", result.endpoint, "err", err)
362 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
363 } else if err := ports.ValidateDurableHostProjection(observed); err != nil {
364 r.log.Debug("broker: observation is not persistable", "endpoint", result.endpoint, "err", err)
365 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
366 } else {
367 r.hosts[result.endpoint] = observed
368 r.publishLocked(true)
369 return
370 }
371 }
372 // Every non-reachable outcome is a typed failure on the capped retry
373 // cadence: the last success, last-known inventory, and failure episode
374 // counter stay authoritative.
375 current.Availability = availability
376 if current.ConsecutiveFailures == 0 {
377 current.FailureEpisode++
378 }
379 current.ConsecutiveFailures++
380 current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
381 delay := r.retryDelay(current.ConsecutiveFailures)
382 if r.demand > 0 {
383 delay = demandRetryDelay(r.demandFreshForRemote, current.ConsecutiveFailures)
384 }
385 current.NextDue = result.at.Add(r.jitter(delay, result.endpoint, result.attempt))
386 r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err, "cause", errors.Unwrap(result.err))
387 r.hosts[result.endpoint] = current
388 r.publishLocked(true)
389}
390
391// observationOutcome classifies a completed observation. It reports
392// failed=false only for a confirmed reachable result; every other outcome
393// carries the typed failure kind and the availability to publish. The caller
394// additionally checks the adopted projection against the durable rules
395// (ports.ValidateDurableHostProjection), which cover the session bound and
396// catalogue validity for the exact incarnation and success time.
397func observationOutcome(result probeResult) (domain.RemoteFailureKind, domain.RemoteAvailability, bool) {
398 if result.err != nil {
399 kind := domain.RemoteFailureTransport
400 var typed domain.RemoteFailure
401 var brokerErr ports.BrokerError
402 switch {
403 case errors.As(result.err, &typed) && typed.Kind != domain.RemoteFailureNone:
404 // The dialer classified the failure (for example SSH authentication).
405 kind = typed.Kind
406 case errors.As(result.err, &brokerErr) && (brokerErr.Code == ports.BrokerErrorConflictingPolicy || brokerErr.Code == ports.BrokerErrorIncompatible):
407 // SSH worked; the daemon's identity or version was refused.
408 kind = domain.RemoteFailureIncompatible
409 case errors.Is(result.err, context.DeadlineExceeded):
410 kind = domain.RemoteFailureTimeout
411 }
412 return kind, availabilityFor(kind), true
413 }
414 switch result.snapshot.Availability {
415 case domain.RemoteAvailabilityReachable:
416 return domain.RemoteFailureNone, domain.RemoteAvailabilityReachable, false
417 case domain.RemoteAvailabilityNoDaemon:
418 return domain.RemoteFailureNone, domain.RemoteAvailabilityNoDaemon, false
419 case domain.RemoteAvailabilityIncompatible:
420 return domain.RemoteFailureIncompatible, domain.RemoteAvailabilityIncompatible, true
421 case domain.RemoteAvailabilityAuthFailed:
422 return domain.RemoteFailureAuthentication, domain.RemoteAvailabilityAuthFailed, true
423 case domain.RemoteAvailabilityInvalidResponse:
424 return domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
425 default:
426 // Unknown or Unreachable without an error still means reachability was
427 // not confirmed, so it retries on the typed backoff cadence.
428 return domain.RemoteFailureTransport, domain.RemoteAvailabilityUnreachable, true
429 }
430}
431
432// availabilityFor maps a typed failure kind to the published availability.
433// The typed kind is preserved separately for notice policy.
434func availabilityFor(kind domain.RemoteFailureKind) domain.RemoteAvailability {
435 switch kind {
436 case domain.RemoteFailureAuthentication:
437 return domain.RemoteAvailabilityAuthFailed
438 case domain.RemoteFailureIncompatible:
439 return domain.RemoteAvailabilityIncompatible
440 case domain.RemoteFailureInvalidResponse:
441 return domain.RemoteAvailabilityInvalidResponse
442 default:
443 return domain.RemoteAvailabilityUnreachable
444 }
445}
446
447// demandRetryDelay is the watched-host failure backoff: the demand cadence
448// doubled per consecutive failure, capped at demandRetryMax.
449func demandRetryDelay(base time.Duration, failures uint) time.Duration {
450 delay := base
451 for i := uint(1); i < failures && delay < demandRetryMax; i++ {
452 delay *= 2
453 }
454 return min(delay, max(base, demandRetryMax))
455}
456
457// retryDelay is the capped exponential retry cadence for a failure streak.
458func (r *Registry) retryDelay(failures uint) time.Duration {
459 delay := r.retryBase
460 for i := uint(1); i < failures; i++ {
461 if delay >= r.retryMax {
462 return r.retryMax
463 }
464 delay *= 2
465 }
466 if delay > r.retryMax {
467 return r.retryMax
468 }
469 return delay
470}
471
472// jittered spreads a base interval deterministically across endpoint and
473// attempt: identical inputs always yield the identical duration, and the
474// result stays within 90%-110% of the base. No wall clock or global
475// randomness participates, so retry and healthy-refresh schedules are
476// reproducible instead of racing a shared PRNG.
477func jittered(base time.Duration, endpoint string, attempt uint64) time.Duration {
478 if base <= 0 {
479 return base
480 }
481 h := fnv.New32a()
482 h.Write([]byte(endpoint))
483 var counter [8]byte
484 binary.BigEndian.PutUint64(counter[:], attempt)
485 h.Write(counter[:])
486 spread := int64(h.Sum32() % 21)
487 return base*90/100 + time.Duration(spread)*base/100
488}
489
490func (r *Registry) nextDelay(now time.Time) time.Duration {
491 r.mu.Lock()
492 defer r.mu.Unlock()
493 if r.demand == 0 {
494 if r.localPending || len(r.pending) > 0 {
495 return 0
496 }
497 return time.Hour
498 }
499 var earliest time.Time
500 if r.local != nil && r.localAttempt == nil {
501 if r.localHost.NextDue.IsZero() {
502 return 0
503 }
504 if earliest.IsZero() || r.localHost.NextDue.Before(earliest) {
505 earliest = r.localHost.NextDue
506 }
507 }
508 for endpoint, host := range r.hosts {
509 // An observation already in flight is completed by its result, never
510 // by the schedule; waiting on it here would spin the run loop.
511 if _, inflight := r.inflight[endpoint]; inflight {
512 continue
513 }
514 if r.pending[endpoint] || host.NextDue.IsZero() {
515 return 0
516 }
517 if earliest.IsZero() || host.NextDue.Before(earliest) {
518 earliest = host.NextDue
519 }
520 }
521 if earliest.IsZero() {
522 return time.Hour
523 }
524 if delay := earliest.Sub(now); delay > 0 {
525 return delay
526 }
527 return 0
528}