1package broker
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/binary"
7 "errors"
8 "fmt"
9 "hash/fnv"
10 "log/slog"
11 "reflect"
12 "slices"
13 "sort"
14 "sync"
15 "sync/atomic"
16 "time"
17 "unicode/utf8"
18
19 "github.com/bnema/vev/internal/domain"
20 "github.com/bnema/vev/internal/ports"
21)
22
23// Observation cadence. No client means no scheduled observations; the
24// fifteen-second interval only applies to an absent daemon under demand.
25const (
26 defaultFreshFor = 15 * time.Second
27 defaultRetryBase = 5 * time.Second
28 defaultRetryLimit = time.Minute
29 // defaultDemandFreshForLocal and defaultDemandFreshForRemote are the
30 // built-in demand cadences RegistryConfig falls back to when
31 // DemandFreshForLocal/DemandFreshForRemote is left zero. They apply only
32 // while at least one client subscription is live (see Registry.SetDemand).
33 defaultDemandFreshForLocal = 1 * time.Second
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
39 // defaultProbeTimeout bounds one observation attempt end to end: dial,
40 // SSH/QUIC bootstrap, daemonmux preamble, and the catalogue exchange. It
41 // sits below the passive retry cap so a peer that never answers costs one
42 // bounded attempt instead of pinning the endpoint's single in-flight slot.
43 defaultProbeTimeout = 20 * time.Second
44)
45
46// errProbeTimeout is the failure an attempt reports when its ports.Clock
47// deadline fires before the probe returns. It wraps context.DeadlineExceeded
48// so observationOutcome classifies it as a typed timeout.
49var errProbeTimeout = fmt.Errorf("broker: observation exceeded its deadline: %w", context.DeadlineExceeded)
50
51// Run lifecycle states. A registry runs exactly once: runIdle is the zero
52// value, runActive marks the single admitted run, and runStopped permanently
53// refuses further runs.
54const (
55 runIdle int32 = iota
56 runActive
57 runStopped
58)
59
60// MembershipMode controls whether runtime membership mutation is admitted.
61type MembershipMode uint8
62
63const (
64 MembershipImmutable MembershipMode = iota
65 MembershipMutable
66)
67
68// RegistryConfig selects how a Registry runs. The zero value observes hosts
69// with immutable membership.
70type RegistryConfig struct {
71 MembershipMode MembershipMode
72 // IncarnationGenerator is an injectable entropy seam for mutable additions.
73 // Nil uses crypto/rand.
74 IncarnationGenerator func() ([16]byte, error)
75 // ObservationDisabled runs the registry as a read-only snapshot owner: it
76 // restores and publishes durable state, owns and drains its writer, and
77 // honors the Run lifecycle, but issues no probes, arms no timers, and never
78 // reconciles. The probe dependency is unused and may be nil. Membership is
79 // expected to be immutable for the run; a caller that needs observation
80 // keeps the zero-value config.
81 ObservationDisabled bool
82 // Local, when non-nil, enables observation of the broker's own machine
83 // daemon. The registry then publishes the local entry at index zero, ahead
84 // of every remote, and never persists it. Local is refused together with
85 // ObservationDisabled: a read-only registry owns no local producer.
86 Local *LocalObservation
87 // DemandFreshForLocal and DemandFreshForRemote are the re-probe cadences
88 // apply/applyLocal schedule into NextDue while Registry.SetDemand reports
89 // at least one live client subscription. A zero value falls back to
90 // defaultDemandFreshForLocal (about 1s) or defaultDemandFreshForRemote
91 // (about 2s). With no live subscription, no scheduled probe runs.
92 // DemandFreshForRemote is also the first delay of a watched host's
93 // failure backoff, which doubles up to max(DemandFreshForRemote, 30s).
94 DemandFreshForLocal time.Duration
95 DemandFreshForRemote time.Duration
96 // ProbeTimeout bounds every local and remote observation attempt on the
97 // registry's ports.Clock. When it fires the attempt is released and
98 // recorded as a timeout failure even if the probe never returns. A zero
99 // value selects defaultProbeTimeout.
100 ProbeTimeout time.Duration
101}
102
103// Registry owns configured hosts and their immutable observation projection.
104type Registry struct {
105 epoch ports.BrokerEpoch
106 probe ports.BrokerHostProbe
107 clock ports.Clock
108 log *slog.Logger
109 store *snapshotWriter
110 // observationDisabled is fixed at construction: a disabled registry never
111 // observes, so it has no probe, timer, or reconcile work to do.
112 observationDisabled bool
113 membershipMode MembershipMode
114 newIncarnation func() ([16]byte, error)
115
116 // hostStore and authority are the synchronous membership owner. Only this
117 // registry may mutate the store while it is live; the caller owns Close.
118 hostStore ports.BrokerHostStore
119 authority ports.BrokerHosts // guarded by mu; independent of publication revision
120
121 current atomic.Pointer[ports.BrokerSnapshot]
122 mu sync.Mutex
123 revision ports.BrokerRevision
124 // revisionExhausted records that this epoch consumed its entire revision
125 // series. The registry then fails closed: the last valid snapshot stays
126 // published, nothing is written to the store, and every mutation path
127 // reports ports.ErrBrokerRevisionExhausted instead of wrapping the series to
128 // zero or below the newest revision.
129 revisionExhausted bool
130 // attempts is the registry-wide monotonic attempt token source. Every
131 // started observation claims one token, and only the token currently
132 // recorded as in flight may complete. Tokens are never reused, so a stale
133 // completion is fenced even when the endpoint is removed and re-added with
134 // the identical registration.
135 attempts uint64
136 // hosts is the broker-native projection per endpoint. Configured
137 // authority (endpoint, registration, policy, display origin, rank) is
138 // stamped here from durable membership; probe results only supply observed
139 // state and can never override it.
140 hosts map[string]ports.BrokerDaemonObservation
141 // order is the publication order of remote hosts: registration order from
142 // durable membership. A snapshot carries the local daemon first (when a
143 // producer exists) and then these remotes in this order.
144 order []string
145 // inflight records the one admitted attempt per endpoint. The record
146 // carries the attempt's derived context cancel so a retired attempt is
147 // released the moment its registration is replaced, removed, or the run
148 // settles; a ctx-honoring probe therefore stops promptly and churn can
149 // never accumulate overlapping probes for one endpoint.
150 inflight map[string]*probeAttempt
151 pending map[string]bool
152 tombstones map[string]ports.BrokerHostTombstone
153 subs map[*subscription]struct{}
154 wake chan struct{}
155 running atomic.Int32
156
157 // local selects observation of the broker's own machine daemon. It is nil
158 // for a remote-only registry. localHost is the configured local authority
159 // stamped onto every publication (Local, DisplayOrigin, Policy, Rank) plus
160 // the observed state a probe supplies; localAttempt is the one admitted
161 // local probe. The local entry is never durable and is never invented: until
162 // a probe answers, the configured authority is published with unknown
163 // availability and zero observed identity.
164 local *LocalObservation
165 localHost ports.BrokerDaemonObservation
166 localAttempt *probeAttempt
167 // localPending marks explicit demand for a local re-probe ahead of its
168 // scheduled freshness window: RequestProbe("") and a completed local control
169 // stream or attach both set it. dispatchLocalLocked honors it the same way
170 // pending honors remote demand, admitting the attempt before NextDue and
171 // clearing the flag the moment the attempt starts.
172 localPending bool
173
174 freshFor time.Duration
175 retryBase time.Duration
176 retryMax time.Duration
177 jitter func(base time.Duration, endpoint string, attempt uint64) time.Duration
178
179 // demandFreshForLocal and demandFreshForRemote are the resolved (non-zero)
180 // demand cadences; see RegistryConfig.DemandFreshForLocal/Remote.
181 demandFreshForLocal time.Duration
182 demandFreshForRemote time.Duration
183 // probeTimeout is the resolved (non-zero) per-attempt deadline; see
184 // RegistryConfig.ProbeTimeout.
185 probeTimeout time.Duration
186 // demand counts live client subscriptions: Registry.SetDemand(true) is one
187 // subscription opening and SetDemand(false) is one closing. Guarded by mu.
188 // While demand > 0, apply and applyLocal schedule subsequent probes.
189 demand int
190}
191
192// NewRegistryWithConfig restores a validated durable snapshot under a fresh
193// broker epoch and selects the run mode from cfg. Observation-disabled mode
194// tolerates a nil probe because such a registry never probes, and a registry
195// configured with a local observation producer (Local) tolerates a nil remote
196// probe as long as it owns no remote membership. Every other dependency is
197// required exactly as NewRegistry requires it.
198func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore, probe ports.BrokerHostProbe, clock ports.Clock, log *slog.Logger, cfg RegistryConfig) (*Registry, error) {
199 if cfg.MembershipMode != MembershipImmutable && cfg.MembershipMode != MembershipMutable {
200 return nil, errors.New("broker: invalid membership mode")
201 }
202 if epoch == 0 || nilDependency(store) || nilDependency(clock) {
203 return nil, errors.New("broker: invalid registry dependencies")
204 }
205 // A remote probe is required to observe remotes, but a registry that
206 // observes only its own machine daemon (Local) may omit it. Such a registry
207 // never owns a remote host: projectMembership and the mutation paths refuse
208 // remote membership rather than silently never probing it.
209 if !cfg.ObservationDisabled && nilDependency(probe) && cfg.Local == nil {
210 return nil, errors.New("broker: invalid registry dependencies")
211 }
212 if log == nil {
213 log = slog.Default()
214 }
215 generator := cfg.IncarnationGenerator
216 if generator == nil {
217 generator = func() (id [16]byte, err error) { _, err = rand.Read(id[:]); return id, err }
218 }
219 demandFreshForLocal := cfg.DemandFreshForLocal
220 if demandFreshForLocal <= 0 {
221 demandFreshForLocal = defaultDemandFreshForLocal
222 }
223 demandFreshForRemote := cfg.DemandFreshForRemote
224 if demandFreshForRemote <= 0 {
225 demandFreshForRemote = defaultDemandFreshForRemote
226 }
227 probeTimeout := cfg.ProbeTimeout
228 if probeTimeout <= 0 {
229 probeTimeout = defaultProbeTimeout
230 }
231 r := &Registry{
232 epoch: epoch, probe: probe, clock: clock, log: log,
233 observationDisabled: cfg.ObservationDisabled,
234 membershipMode: cfg.MembershipMode, newIncarnation: generator,
235 hosts: make(map[string]ports.BrokerDaemonObservation), inflight: make(map[string]*probeAttempt),
236 pending: make(map[string]bool), tombstones: make(map[string]ports.BrokerHostTombstone),
237 subs: make(map[*subscription]struct{}), wake: make(chan struct{}, 1),
238 freshFor: defaultFreshFor, retryBase: defaultRetryBase, retryMax: defaultRetryLimit,
239 jitter: jittered,
240 demandFreshForLocal: demandFreshForLocal,
241 demandFreshForRemote: demandFreshForRemote,
242 probeTimeout: probeTimeout,
243 }
244 if cfg.ObservationDisabled && cfg.Local != nil {
245 return nil, errors.New("broker: observation-disabled registry cannot observe the local daemon")
246 }
247 if cfg.Local != nil {
248 if err := cfg.Local.validate(); err != nil {
249 return nil, err
250 }
251 // Copy the configured authority: the registry reads it from the probe
252 // and publish paths, and a caller that kept the pointer could otherwise
253 // mutate authority after construction.
254 local := *cfg.Local
255 r.local = &local
256 r.localHost = ports.BrokerDaemonObservation{Local: true, DisplayOrigin: cfg.Local.DisplayOrigin, Policy: cfg.Local.Policy, Availability: domain.RemoteAvailabilityUnknown}
257 if err := r.localHost.Validate(); err != nil {
258 return nil, fmt.Errorf("broker: invalid local observation authority: %w", err)
259 }
260 }
261 r.store = newSnapshotWriter(store, log)
262 snapshot, err := store.Load()
263 if err != nil {
264 return nil, err
265 }
266 if err := r.restore(snapshot); err != nil {
267 return nil, err
268 }
269 hosts, err := store.LoadHosts()
270 if err != nil {
271 return nil, err
272 }
273 hosts.Hosts, err = ports.UpgradeBrokerHostRoutes(hosts.Hosts)
274 if err != nil {
275 return nil, err
276 }
277 if err := r.projectMembership(hosts); err != nil {
278 return nil, err
279 }
280 r.hostStore = store
281 r.authority = ports.BrokerHosts{Revision: hosts.Revision, Hosts: ports.CloneBrokerHostRecords(hosts.Hosts)}
282 r.publishLocked(false)
283 return r, nil
284}
285
286// nilDependency reports whether a required dependency is absent, including a
287// typed nil pointer stored behind a non-nil interface. Comparing the interface
288// to nil alone would admit a typed nil and defer the failure to the first
289// method call, which panics; the reflection check keeps the dependency guard
290// total for every nil shape.
291func nilDependency(dependency any) bool {
292 if dependency == nil {
293 return true
294 }
295 value := reflect.ValueOf(dependency)
296 switch value.Kind() {
297 case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice:
298 return value.IsNil()
299 default:
300 return false
301 }
302}
303
304// projectMembership intersects durable membership with the observations a
305// previous process left behind. Durable membership is authoritative:
306// snapshots cannot resurrect removed hosts or omit never-observed
307// registrations. A restored observation is reused only when it belongs to the
308// exact same registration identity; any other registration starts unknown and
309// carries no inventory.
310func (r *Registry) projectMembership(hosts ports.BrokerHosts) error {
311 if err := hosts.Validate(); err != nil {
312 return err
313 }
314 if len(hosts.Hosts) > 0 && !r.observationDisabled && nilDependency(r.probe) {
315 return errors.New("broker: remote hosts require a host probe")
316 }
317 next := make(map[string]ports.BrokerDaemonObservation, len(hosts.Hosts))
318 order := make([]string, 0, len(hosts.Hosts))
319 for rank, record := range hosts.Hosts {
320 registration := record.Registration
321 restored, present := r.hosts[registration.Endpoint]
322 if !present || !restored.Registration.Equal(registration) {
323 restored = ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnknown}
324 }
325 projection := stampHostAuthority(restored, record, rank)
326 // A durable membership whose projection cannot be published is refused
327 // here, at the boundary that owns authority, instead of silently
328 // leaving the registry unable to publish anything.
329 if err := projection.Validate(); err != nil {
330 return fmt.Errorf("broker: host record %q is not publishable: %w", registration.Endpoint, err)
331 }
332 next[registration.Endpoint] = projection
333 order = append(order, registration.Endpoint)
334 }
335 r.hosts = next
336 r.order = order
337 return nil
338}
339
340// stampHostAuthority overwrites the configured authority fields of a daemon
341// projection with durable membership authority. Membership is the single
342// authority for endpoint, registration, policy, and the presentation hints
343// derived from them; a probe, a restored durable observation, or any other
344// producer can never supply or override them. The durable observation format
345// deliberately carries no policy (see durable), so this stamp is also what
346// gives every restored observation a valid policy.
347func stampHostAuthority(observation ports.BrokerDaemonObservation, record ports.BrokerHostRecord, rank int) ports.BrokerDaemonObservation {
348 observation.Local = false
349 observation.Endpoint = record.Registration.Endpoint
350 observation.DisplayOrigin = hostDisplayOrigin(record.Registration.Endpoint)
351 observation.Rank = rank
352 observation.Registration = record.Registration
353 observation.Policy = record.Policy
354 if observation.Availability == 0 {
355 // An unobserved daemon reports the explicit unknown availability; zero
356 // is never a valid publication.
357 observation.Availability = domain.RemoteAvailabilityUnknown
358 }
359 return observation
360}
361
362// stampObservationAuthority copies the configured authority already stamped on
363// a live projection onto a fresh probe result. Probe results supply only
364// observed state, so their endpoint, registration, policy, display origin, and
365// rank fields are discarded wholesale.
366func stampObservationAuthority(observation, authority ports.BrokerDaemonObservation) ports.BrokerDaemonObservation {
367 observation.Local = false
368 observation.Endpoint = authority.Endpoint
369 observation.DisplayOrigin = authority.DisplayOrigin
370 observation.Rank = authority.Rank
371 observation.Registration = authority.Registration
372 observation.Policy = authority.Policy
373 return observation
374}
375
376// hostDisplayOrigin derives the presentation hint for a configured host. The
377// routing endpoint stays authoritative; the origin is only the login prefix
378// stripped for display, and an endpoint that carries none is its own origin.
379// The hint is sanitized to the display-text rule before it is clamped rune- and
380// byte-safely to the ports bound: an endpoint the routing validator accepts may
381// still carry bidi or control runes, and rendering those would let a derived
382// hint reorder the picker text around it. A derived hint carries no routing
383// authority, so the refused runes are dropped rather than the host rejected. A
384// target left with no displayable rune at all yields no origin, which its
385// caller refuses as unpublishable membership instead of inventing display text.
386func hostDisplayOrigin(endpoint string) string {
387 origin := ports.SanitizeBrokerDisplayText(domain.RemoteDisplayOrigin(endpoint))
388 if len(origin) <= ports.BrokerMaxDisplayOriginBytes {
389 return origin
390 }
391 truncated := origin[:ports.BrokerMaxDisplayOriginBytes]
392 for len(truncated) > 0 && !utf8.ValidString(truncated) {
393 truncated = truncated[:len(truncated)-1]
394 }
395 return truncated
396}
397
398// restore adopts a validated durable snapshot. A zero-epoch snapshot is only
399// the empty placeholder a store returns before its first write: any record
400// without an epoch is rejected instead of silently dropped. A persisted
401// snapshot from the registry's own epoch is rejected as well: every broker
402// process samples a fresh epoch and restarts its revision series at one, so
403// adopting same-epoch revisions would make the restored series non-monotonic
404// (and a stale same-epoch record indistinguishable from current authority).
405// In-flight observation state and retirement tombstones are per-process
406// fencing state and never survive a restart.
407func (r *Registry) restore(snapshot ports.BrokerSnapshot) error {
408 if snapshot.Epoch == 0 {
409 if len(snapshot.Daemons) > 0 || len(snapshot.Removed) > 0 || snapshot.Revision != 0 {
410 return errors.New("broker: persisted snapshot has no epoch")
411 }
412 return nil
413 }
414 if snapshot.Epoch == r.epoch {
415 return errors.New("broker: persisted snapshot epoch matches the new registry epoch")
416 }
417 if err := ports.ValidateDurableSnapshot(snapshot); err != nil {
418 return err
419 }
420 for _, daemon := range snapshot.Daemons {
421 daemon = daemon.Clone()
422 daemon.Checking = false
423 r.hosts[daemon.Endpoint] = daemon
424 }
425 return nil
426}
427
428// Snapshot returns a defensive immutable copy without I/O.
429func (r *Registry) Snapshot() ports.BrokerSnapshot { return r.current.Load().Clone() }
430
431// Subscribe returns a capacity-one notification stream. Slow consumers never
432// block publication and always re-read the newest snapshot.
433func (r *Registry) Subscribe() ports.BrokerSubscription {
434 s := &subscription{changed: make(chan struct{}, 1)}
435 s.close = func() {
436 r.mu.Lock()
437 delete(r.subs, s)
438 r.mu.Unlock()
439 }
440 r.mu.Lock()
441 r.subs[s] = struct{}{}
442 r.mu.Unlock()
443 s.offer()
444 return s
445}
446
447// AddHost adds or pins endpoint under policy. Duplicate same-policy additions
448// CAS-verify authority and preserve identity; a conflicting policy is refused.
449func (r *Registry) LookupHost(ctx context.Context, endpoint string) (ports.BrokerHostRecord, bool, error) {
450 if err := ctx.Err(); err != nil {
451 return ports.BrokerHostRecord{}, false, err
452 }
453 r.mu.Lock()
454 defer r.mu.Unlock()
455 for _, record := range r.authority.Hosts {
456 if record.Registration.Endpoint == endpoint {
457 record.Route = record.Route.Clone()
458 return record, true, nil
459 }
460 }
461 return ports.BrokerHostRecord{}, false, nil
462}
463
464func (r *Registry) AddHost(ctx context.Context, endpoint string, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
465 if err := domain.ValidateRemoteHostTarget(endpoint); err != nil {
466 return domain.RemoteRegistration{}, err
467 }
468 if err := policy.Validate(); err != nil {
469 return domain.RemoteRegistration{}, err
470 }
471 r.mu.Lock()
472 defer r.mu.Unlock()
473 if err := r.mutableLocked(ctx); err != nil {
474 return domain.RemoteRegistration{}, err
475 }
476 records := append([]ports.BrokerHostRecord(nil), r.authority.Hosts...)
477 for i := range records {
478 if records[i].Registration.Endpoint != endpoint {
479 continue
480 }
481 if records[i].Policy != policy {
482 return domain.RemoteRegistration{}, ports.ErrBrokerHostConflict
483 }
484 records[i].Pinned = true
485 if err := r.commitMembershipLocked(records); err != nil {
486 return domain.RemoteRegistration{}, err
487 }
488 return records[i].Registration, nil
489 }
490 id, err := r.newIncarnation()
491 if err != nil {
492 return domain.RemoteRegistration{}, err
493 }
494 reg, err := domain.NewRemoteRegistration(endpoint, id)
495 if err != nil {
496 return domain.RemoteRegistration{}, err
497 }
498 if reg.IsZero() {
499 return domain.RemoteRegistration{}, errors.New("broker: incarnation generator returned zero")
500 }
501 route, err := ports.BrokerRouteForTransport(policy.Transport, endpoint)
502 if err != nil {
503 return domain.RemoteRegistration{}, err
504 }
505 records = append(records, ports.BrokerHostRecord{Registration: reg, Pinned: true, Policy: policy, Route: route})
506 if err := r.commitMembershipLocked(records); err != nil {
507 return domain.RemoteRegistration{}, err
508 }
509 return reg, nil
510}
511
512// RemoveHost removes only the exact expected registration.
513// BindAuthenticatedIdentity durably binds first-contact identity under the
514// exact registration and policy fence. The same binding is idempotent; a
515// different identity is refused and never overwrites authority.
516func (r *Registry) BindAuthenticatedIdentity(ctx context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
517 if err := request.Validate(); err != nil {
518 return "", err
519 }
520 if request.Fence.Local {
521 return request.Identity, nil
522 }
523 if err := ctx.Err(); err != nil {
524 return "", err
525 }
526 r.mu.Lock()
527 defer r.mu.Unlock()
528 records := append([]ports.BrokerHostRecord(nil), r.authority.Hosts...)
529 for index := range records {
530 record := &records[index]
531 if record.Identity == request.Identity && record.Policy != request.Policy {
532 return "", ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: errors.New("broker: authenticated daemon identity is already bound under an incompatible policy")}
533 }
534 }
535 for index := range records {
536 record := &records[index]
537 if !record.Registration.Equal(request.Fence.Registration) || record.Policy != request.Policy {
538 continue
539 }
540 if record.Identity != "" {
541 if record.Identity != request.Identity {
542 return "", ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: errors.New("broker: authenticated daemon identity conflicts with durable binding")}
543 }
544 return record.Identity, nil
545 }
546 record.Identity = request.Identity
547 if err := r.commitMembershipLocked(records); err != nil {
548 return "", err
549 }
550 return request.Identity, nil
551 }
552 return "", ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: errors.New("broker: stale identity binding fence")}
553}
554
555func (r *Registry) RemoveHost(ctx context.Context, expected domain.RemoteRegistration) (bool, error) {
556 if err := expected.Validate(); err != nil {
557 return false, err
558 }
559 r.mu.Lock()
560 defer r.mu.Unlock()
561 if err := r.mutableLocked(ctx); err != nil {
562 return false, err
563 }
564 records := append([]ports.BrokerHostRecord(nil), r.authority.Hosts...)
565 for i := range records {
566 if records[i].Registration.Endpoint != expected.Endpoint {
567 continue
568 }
569 if !records[i].Registration.Equal(expected) {
570 return false, ports.ErrBrokerHostConflict
571 }
572 records = append(records[:i], records[i+1:]...)
573 if err := r.commitMembershipLocked(records); err != nil {
574 return false, err
575 }
576 return true, nil
577 }
578 // A no-op still verifies the loaded revision.
579 return false, r.commitMembershipLocked(records)
580}
581
582// UpdateHostPolicy updates exact authority and advances its generation.
583func (r *Registry) UpdateHostPolicy(ctx context.Context, expected domain.RemoteRegistration, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
584 if err := expected.Validate(); err != nil {
585 return domain.RemoteRegistration{}, err
586 }
587 if err := policy.Validate(); err != nil {
588 return domain.RemoteRegistration{}, err
589 }
590 r.mu.Lock()
591 defer r.mu.Unlock()
592 if err := r.mutableLocked(ctx); err != nil {
593 return domain.RemoteRegistration{}, err
594 }
595 records := append([]ports.BrokerHostRecord(nil), r.authority.Hosts...)
596 for i := range records {
597 if records[i].Registration.Endpoint != expected.Endpoint {
598 continue
599 }
600 if !records[i].Registration.Equal(expected) {
601 return domain.RemoteRegistration{}, ports.ErrBrokerHostConflict
602 }
603 if records[i].Policy == policy {
604 if err := r.commitMembershipLocked(records); err != nil {
605 return domain.RemoteRegistration{}, err
606 }
607 return expected, nil
608 }
609 if expected.Generation == ^domain.RemoteGeneration(0) {
610 return domain.RemoteRegistration{}, ports.ErrBrokerRevisionExhausted
611 }
612 route, err := ports.BrokerRouteForTransport(policy.Transport, expected.Endpoint)
613 if err != nil {
614 return domain.RemoteRegistration{}, err
615 }
616 records[i].Registration.Generation++
617 records[i].Route = route
618 records[i].Policy = policy
619 if err := r.commitMembershipLocked(records); err != nil {
620 return domain.RemoteRegistration{}, err
621 }
622 return records[i].Registration, nil
623 }
624 return domain.RemoteRegistration{}, ports.ErrBrokerHostConflict
625}
626
627func (r *Registry) mutableLocked(ctx context.Context) error {
628 if r.membershipMode != MembershipMutable {
629 return ports.ErrBrokerMembershipImmutable
630 }
631 if err := ctx.Err(); err != nil {
632 return err
633 }
634 if r.running.Load() == runStopped {
635 return ports.ErrBrokerRegistryClosed
636 }
637 if r.revisionExhaustedLocked() {
638 return ports.ErrBrokerRevisionExhausted
639 }
640 return nil
641}
642
643// commitMembershipLocked is the sole mutable-membership durable commit path.
644// It installs projection only after CAS success.
645func (r *Registry) commitMembershipLocked(records []ports.BrokerHostRecord) error {
646 if err := ports.ValidateBrokerHostRecords(records); err != nil {
647 return err
648 }
649 if len(records) > 0 && !r.observationDisabled && nilDependency(r.probe) {
650 return errors.New("broker: remote hosts require a host probe")
651 }
652 next := make(map[string]ports.BrokerDaemonObservation, len(records))
653 order := make([]string, 0, len(records))
654 for rank, record := range records {
655 observation := stampHostAuthority(ports.BrokerDaemonObservation{}, record, rank)
656 // A membership the registry cannot publish is refused here, before the
657 // store CAS, rather than persisted and left to freeze every later
658 // publication: an endpoint whose derived display hint is empty after the
659 // display rule drops its runes makes every snapshot carrying it invalid.
660 if err := observation.Validate(); err != nil {
661 return fmt.Errorf("broker: host record %q is not publishable: %w", record.Registration.Endpoint, err)
662 }
663 next[record.Registration.Endpoint] = observation
664 order = append(order, record.Registration.Endpoint)
665 }
666 if err := r.hostStore.ReplaceHosts(r.authority.Revision, records); err != nil {
667 return fmt.Errorf("broker: replace hosts: %w", err)
668 }
669 r.authority = ports.BrokerHosts{Revision: r.authority.Revision + 1, Hosts: ports.CloneBrokerHostRecords(records)}
670 r.setHostsLocked(next, order)
671 return nil
672}
673
674// ReplaceHosts durably replaces membership and explicit policy using the loaded
675// authority revision. It delegates to commitMembershipLocked, the same
676// validated commit path the runtime mutations use, so membership whose
677// projection could never be published is refused before the store CAS instead
678// of being persisted and silently freezing every later publication. Success
679// means membership is committed, not that advisory observations have flushed.
680// Store errors (including stale authority) are returned without changing the
681// projection, probe attempts, or local token. A settled run reports
682// ports.ErrBrokerRegistryClosed and an exhausted revision series reports
683// ports.ErrBrokerRevisionExhausted; both refuse the mutation before the store
684// CAS, so a refusal never changes durable authority either. Conflicts are not
685// retried or refreshed implicitly: reconcile by reopening the owner. Callers
686// must supply fresh incarnations when removing and re-adding. Even identical
687// records perform CAS so stale authority cannot report success. The synchronous
688// write holds mu; Snapshot remains lock-free, but scheduling waits for
689// membership durability. Observation writes remain asynchronous.
690func (r *Registry) ReplaceHosts(records []ports.BrokerHostRecord) error {
691 var err error
692 records, err = ports.UpgradeBrokerHostRoutes(records)
693 if err != nil {
694 return err
695 }
696 if err := ports.ValidateBrokerHostRecords(records); err != nil {
697 return err
698 }
699 r.mu.Lock()
700 defer r.mu.Unlock()
701 if r.running.Load() == runStopped {
702 return fmt.Errorf("broker: membership replacement: %w", ports.ErrBrokerRegistryClosed)
703 }
704 if r.revisionExhaustedLocked() {
705 return fmt.Errorf("broker: membership replacement: %w", ports.ErrBrokerRevisionExhausted)
706 }
707 return r.commitMembershipLocked(records)
708}
709
710// setHostsLocked applies only a projection; the public path commits authority
711// first. The caller holds mu through both steps, fencing probe completions.
712// order is the registration order the projection was built from; it becomes
713// the publication order of remote daemons.
714func (r *Registry) setHostsLocked(next map[string]ports.BrokerDaemonObservation, order []string) {
715 if registrationsUnchanged(r.hosts, next, r.order, order) {
716 return
717 }
718 for endpoint, current := range r.hosts {
719 candidate, present := next[endpoint]
720 if present && current.Registration.Equal(candidate.Registration) && current.Policy == candidate.Policy {
721 // Unchanged authority: keep the live observation as is. Rank is the
722 // one authority field that follows membership order rather than the
723 // record, so it is re-stamped from the new projection.
724 kept := current
725 kept.Rank = candidate.Rank
726 next[endpoint] = kept
727 continue
728 }
729 // A retired registration owns no observation: cancelling its attempt
730 // releases a ctx-honoring probe immediately, and dropping its record
731 // means the retired completion can never clear or answer the
732 // replacement, and the replacement is never blocked behind it. The
733 // cancellation happens under the registry lock, so no replacement
734 // attempt can start before the retired context is cancelled.
735 r.cancelInflightLocked(endpoint)
736 delete(r.pending, endpoint)
737 if present {
738 // The endpoint stays live under a new registration, so its
739 // retirement is never published as a tombstone alongside a live
740 // host; the attempt token fences the stale completion.
741 delete(r.tombstones, endpoint)
742 continue
743 }
744 r.retireLocked(current.Registration)
745 }
746 for endpoint := range next {
747 // Re-adding an endpoint supersedes its tombstone: a live host and a
748 // tombstone for the same endpoint are mutually exclusive.
749 delete(r.tombstones, endpoint)
750 }
751 r.hosts = next
752 r.order = order
753 r.publishLocked(true)
754 r.hint()
755}
756
757// cancelInflightLocked cancels the in-flight attempt for one endpoint and
758// drops its record. Callers must hold r.mu: the retired attempt's context is
759// cancelled before the record disappears, so no replacement attempt can start
760// while the retired probe's context is still live.
761func (r *Registry) cancelInflightLocked(endpoint string) {
762 if attempt := r.inflight[endpoint]; attempt != nil {
763 attempt.cancel()
764 delete(r.inflight, endpoint)
765 }
766}
767
768// cancelAllInflightLocked cancels every in-flight attempt and drops the
769// records. Callers must hold r.mu.
770func (r *Registry) cancelAllInflightLocked() {
771 for endpoint, attempt := range r.inflight {
772 attempt.cancel()
773 delete(r.inflight, endpoint)
774 }
775}
776
777// RequestProbe is non-blocking and coalesces concurrent demand per endpoint.
778// A disabled registry never observes, so the demand is dropped without
779// scheduling anything. An empty endpoint requests a local re-probe instead of a
780// configured remote: the broker's own machine daemon is re-observed ahead of
781// its scheduled freshness window, honored by dispatchLocalLocked exactly like
782// pending honors remote demand. A registry with no local observation producer
783// drops it, just like an unknown remote endpoint is dropped.
784func (r *Registry) RequestProbe(endpoint string) {
785 if r.observationDisabled {
786 return
787 }
788 r.mu.Lock()
789 if endpoint == "" {
790 if r.local != nil {
791 r.localPending = true
792 }
793 } else if _, ok := r.hosts[endpoint]; ok {
794 r.pending[endpoint] = true
795 }
796 r.mu.Unlock()
797 r.hint()
798}
799
800// SetDemand records one live client subscription opening (active=true) or
801// closing (active=false). internal/usecase/broker/service.go calls it from
802// Service.Subscribe and serviceSubscription.Close, so the demand count tracks
803// exactly the connections a client is actively watching. The first
804// subscription requests one immediate observation per host; while demand
805// stays positive, the demand cadence schedules later observations. With no
806// subscription, only explicit RequestProbe calls dispatch. A disabled registry
807// still counts subscriptions but never observes.
808func (r *Registry) SetDemand(active bool) {
809 r.mu.Lock()
810 if active {
811 if r.demand == 0 {
812 r.localPending = r.local != nil
813 for endpoint := range r.hosts {
814 r.pending[endpoint] = true
815 }
816 }
817 r.demand++
818 } else if r.demand > 0 {
819 r.demand--
820 }
821 r.mu.Unlock()
822 r.hint()
823}
824
825// freshForLocked returns the demand cadence while subscribed. An absent
826// daemon uses the longer freshFor interval instead. Callers must hold r.mu.
827func (r *Registry) freshForLocked(local bool) time.Duration {
828 if r.demand <= 0 {
829 return r.freshFor
830 }
831 if local {
832 return r.demandFreshForLocal
833 }
834 return r.demandFreshForRemote
835}
836
837func (r *Registry) hint() {
838 select {
839 case r.wake <- struct{}{}:
840 default:
841 }
842}
843
844// Run schedules observations until cancellation. Run owns the registry
845// lifecycle and may execute exactly once: a second call is refused so a
846// concurrent or restarted caller can never duplicate scheduling. On return it
847// clears in-flight observation state, publishes the settled projection, and
848// flushes durable persistence before any further write is refused.
849//
850// An observation-disabled registry runs the same lifecycle and drain but never
851// schedules observation: it arms no timer, starts no probe, and drops every
852// reconcile. It still owns and drains its durable writer, so a shutdown never
853// returns before the newest staged publication has reached the store.
854func (r *Registry) Run(ctx context.Context) {
855 if !r.running.CompareAndSwap(runIdle, runActive) {
856 r.log.Warn("broker: registry Run refused; single-run already started or finished")
857 return
858 }
859 defer r.settle()
860 if r.observationDisabled {
861 <-ctx.Done()
862 return
863 }
864 timer := r.clock.NewTimer(0)
865 defer timer.Stop()
866 results := make(chan probeResult, ports.BrokerMaxHosts)
867 for {
868 now := r.clock.Now()
869 r.dispatch(ctx, now, results)
870 timer.Reset(r.nextDelay(now))
871 select {
872 case <-ctx.Done():
873 return
874 case <-r.wake:
875 case <-timer.C():
876 case result := <-results:
877 if ctx.Err() != nil {
878 // Shutdown wins: a cancellation is never recorded as an
879 // observation failure.
880 return
881 }
882 r.apply(result)
883 }
884 }
885}
886
887// settle is Run's cleanup: nothing stays in flight, the published projection
888// carries no transient Checking flag, durable persistence is flushed, and
889// every later Run call is refused.
890func (r *Registry) settle() {
891 r.mu.Lock()
892 r.cancelAllInflightLocked()
893 if r.localAttempt != nil {
894 r.localAttempt.cancel()
895 r.localAttempt = nil
896 }
897 cleared := false
898 if r.local != nil && r.localHost.Checking {
899 r.localHost.Checking = false
900 cleared = true
901 }
902 for endpoint, host := range r.hosts {
903 if !host.Checking {
904 continue
905 }
906 host.Checking = false
907 r.hosts[endpoint] = host
908 cleared = true
909 }
910 if cleared {
911 r.publishLocked(true)
912 }
913 r.running.Store(runStopped)
914 r.mu.Unlock()
915 r.store.close()
916}
917
918// probeAttempt is one admitted observation: the registry-wide token that
919// fences its completion, and the cancel for the context derived from the run
920// context. Cancelling the derived context retires the attempt: a probe that
921// honors cancellation returns promptly, so a replaced or removed endpoint
922// never accumulates overlapping probes, and settle releases every attempt
923// before the run loop exits.
924type probeAttempt struct {
925 token uint64
926 cancel context.CancelFunc
927}
928
929// probeResult is one completed observation bound to the exact attempt token
930// that started it.
931type probeResult struct {
932 endpoint string
933 registration domain.RemoteRegistration
934 attempt uint64
935 snapshot ports.BrokerDaemonObservation
936 err error
937 at time.Time
938 // local marks the result of one local probe attempt, which carries no
939 // endpoint or registration and is applied to the configured local entry.
940 local bool
941}
942
943func (r *Registry) dispatch(ctx context.Context, now time.Time, results chan<- probeResult) {
944 r.mu.Lock()
945 if r.demand == 0 && len(r.pending) == 0 && !r.localPending {
946 r.mu.Unlock()
947 return
948 }
949 started := false
950 for endpoint, host := range r.hosts {
951 if _, inflight := r.inflight[endpoint]; inflight {
952 continue
953 }
954 explicit := r.pending[endpoint]
955 if !explicit && (r.demand == 0 || !host.NextDue.IsZero() && now.Before(host.NextDue)) {
956 continue
957 }
958 delete(r.pending, endpoint)
959 r.attempts++
960 attempt := r.attempts
961 probeCtx, cancel := context.WithCancel(ctx)
962 r.inflight[endpoint] = &probeAttempt{token: attempt, cancel: cancel}
963 if visibleCheck(host, explicit) {
964 started = true
965 host.Checking = true
966 }
967 host.LastAttempt = now
968 r.hosts[endpoint] = host
969 registration := host.Registration
970 // The deadline is armed here, under r.mu and at dispatch time, so the
971 // attempt's bound starts with the attempt rather than whenever its
972 // goroutine is scheduled.
973 deadline := r.clock.NewTimer(r.probeTimeout)
974 go func(endpoint string, registration domain.RemoteRegistration, attempt uint64, probeCtx context.Context) {
975 snapshot, err := observeBounded(probeCtx, deadline, func(ctx context.Context) (ports.BrokerDaemonObservation, error) {
976 return r.probe.Probe(ctx, registration)
977 })
978 result := probeResult{endpoint: endpoint, registration: registration, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}
979 select {
980 case results <- result:
981 case <-probeCtx.Done():
982 // The attempt was retired (replaced, removed, or shutdown), so
983 // its completion is already fenced: dropping it keeps churn from
984 // queueing results the run loop would only discard.
985 }
986 }(endpoint, registration, attempt, probeCtx)
987 }
988 // Only an observation that visibly started changes the projection; an
989 // idle dispatch or a routine refresh must not bump the revision or wake
990 // subscribers. The in-flight projection is never persisted: Checking is
991 // transient.
992 if r.dispatchLocalLocked(ctx, now, results) {
993 started = true
994 }
995 if started {
996 r.publishLocked(false)
997 }
998 r.mu.Unlock()
999}
1000
1001// visibleCheck reports whether starting an observation publishes the transient
1002// Checking state. A routine scheduled refresh of a reachable daemon stays
1003// silent until its result lands, so an idle watched broker publishes once per
1004// observation instead of twice; an explicit request or a daemon that is not
1005// confirmed reachable still shows that it is being checked.
1006func visibleCheck(host ports.BrokerDaemonObservation, explicit bool) bool {
1007 return explicit || host.Availability != domain.RemoteAvailabilityReachable
1008}
1009
1010// observeBounded runs one observation under its attempt deadline. retire is the
1011// attempt's retirement context: a replaced, removed, or settled attempt returns
1012// at once and its completion is fenced by the caller. The probe itself runs on
1013// a context derived from retire that the deadline also cancels, so a
1014// ctx-honoring probe unwinds its dial, handshake, or exchange when either
1015// fires. The attempt never waits for the probe past its deadline: a probe that
1016// ignores cancellation is abandoned (its late result is discarded) and the
1017// attempt still reports errProbeTimeout, so the endpoint's single in-flight
1018// slot is always released.
1019func observeBounded(retire context.Context, deadline ports.Timer, observe func(context.Context) (ports.BrokerDaemonObservation, error)) (ports.BrokerDaemonObservation, error) {
1020 defer deadline.Stop()
1021 probeCtx, cancel := context.WithCancelCause(retire)
1022 defer cancel(nil)
1023 type outcome struct {
1024 snapshot ports.BrokerDaemonObservation
1025 err error
1026 }
1027 done := make(chan outcome, 1)
1028 go func() {
1029 snapshot, err := observe(probeCtx)
1030 done <- outcome{snapshot: snapshot, err: err}
1031 }()
1032 select {
1033 case result := <-done:
1034 return result.snapshot, result.err
1035 case <-deadline.C():
1036 cancel(errProbeTimeout)
1037 return ports.BrokerDaemonObservation{}, errProbeTimeout
1038 case <-retire.Done():
1039 return ports.BrokerDaemonObservation{}, retire.Err()
1040 }
1041}
1042
1043func (r *Registry) apply(result probeResult) {
1044 r.mu.Lock()
1045 defer r.mu.Unlock()
1046 if result.local {
1047 r.applyLocal(result)
1048 return
1049 }
1050 current, ok := r.hosts[result.endpoint]
1051 attempt := r.inflight[result.endpoint]
1052 if !ok || !current.Registration.Equal(result.registration) || attempt == nil || attempt.token != result.attempt {
1053 // A stale completion touches neither current authority nor the
1054 // in-flight slot, which may already belong to a newer registration or
1055 // attempt.
1056 return
1057 }
1058 delete(r.inflight, result.endpoint)
1059 // The probe has returned, so releasing its derived context here only
1060 // frees the attempt's resources; the record is gone before any new
1061 // attempt can start.
1062 attempt.cancel()
1063 current.Checking = false
1064 current.LastAttempt = result.at
1065
1066 kind, availability, failed := observationOutcome(result)
1067 if !failed && availability == domain.RemoteAvailabilityNoDaemon {
1068 current.Availability = availability
1069 current.LastSuccess = result.at
1070 // An absent daemon is stable: demand does not require a fresh SSH
1071 // handshake every two seconds just to confirm it is still absent.
1072 current.NextDue = result.at.Add(r.jitter(r.freshFor, result.endpoint, result.attempt))
1073 current.ConsecutiveFailures = 0
1074 current.LastFailure = domain.RemoteFailure{}
1075 current.Identity = ""
1076 current.Incarnation = ports.BrokerDaemonIncarnation{}
1077 current.ProtocolVersion = 0
1078 current.Capabilities = 0
1079 current.InventoryKnown = false
1080 current.Sessions = nil
1081 r.hosts[result.endpoint] = current
1082 r.publishLocked(true)
1083 return
1084 }
1085 if !failed {
1086 // Configured authority comes only from the registry's own projection:
1087 // a probe result supplies observed state, never endpoint, registration,
1088 // policy, display origin, or rank.
1089 observed := stampObservationAuthority(result.snapshot.Clone(), current)
1090 observed.Checking = false
1091 observed.LastAttempt = result.at
1092 observed.LastSuccess = result.at
1093 observed.NextDue = result.at.Add(r.jitter(r.freshForLocked(false), result.endpoint, result.attempt))
1094 observed.ConsecutiveFailures = 0
1095 observed.LastFailure = domain.RemoteFailure{}
1096 observed.FailureEpisode = current.FailureEpisode
1097 // The published observation must be fully valid once authority is
1098 // stamped: a probe that invents partial identity (identity without a
1099 // protocol version) or an inventory without a known catalogue is an
1100 // invalid response, never a published observation. The durable rule is
1101 // re-checked too: a projection the offline store would reject is never
1102 // published as if it had persisted. The store applies the same rule.
1103 if err := observed.Validate(); err != nil {
1104 r.log.Debug("broker: observation is invalid", "endpoint", result.endpoint, "err", err)
1105 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
1106 } else if err := ports.ValidateDurableHostProjection(observed); err != nil {
1107 r.log.Debug("broker: observation is not persistable", "endpoint", result.endpoint, "err", err)
1108 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
1109 } else {
1110 r.hosts[result.endpoint] = observed
1111 r.publishLocked(true)
1112 return
1113 }
1114 }
1115 // Every non-reachable outcome is a typed failure on the capped retry
1116 // cadence: the last success, last-known inventory, and failure episode
1117 // counter stay authoritative.
1118 current.Availability = availability
1119 if current.ConsecutiveFailures == 0 {
1120 current.FailureEpisode++
1121 }
1122 current.ConsecutiveFailures++
1123 current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
1124 delay := r.retryDelay(current.ConsecutiveFailures)
1125 if r.demand > 0 {
1126 delay = demandRetryDelay(r.demandFreshForRemote, current.ConsecutiveFailures)
1127 }
1128 current.NextDue = result.at.Add(r.jitter(delay, result.endpoint, result.attempt))
1129 r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err, "cause", errors.Unwrap(result.err))
1130 r.hosts[result.endpoint] = current
1131 r.publishLocked(true)
1132}
1133
1134// observationOutcome classifies a completed observation. It reports
1135// failed=false only for a confirmed reachable result; every other outcome
1136// carries the typed failure kind and the availability to publish. The caller
1137// additionally checks the adopted projection against the durable rules
1138// (ports.ValidateDurableHostProjection), which cover the session bound and
1139// catalogue validity for the exact incarnation and success time.
1140func observationOutcome(result probeResult) (domain.RemoteFailureKind, domain.RemoteAvailability, bool) {
1141 if result.err != nil {
1142 kind := domain.RemoteFailureTransport
1143 var typed domain.RemoteFailure
1144 var brokerErr ports.BrokerError
1145 switch {
1146 case errors.As(result.err, &typed) && typed.Kind != domain.RemoteFailureNone:
1147 // The dialer classified the failure (for example SSH authentication).
1148 kind = typed.Kind
1149 case errors.As(result.err, &brokerErr) && (brokerErr.Code == ports.BrokerErrorConflictingPolicy || brokerErr.Code == ports.BrokerErrorIncompatible):
1150 // SSH worked; the daemon's identity or version was refused.
1151 kind = domain.RemoteFailureIncompatible
1152 case errors.Is(result.err, context.DeadlineExceeded):
1153 kind = domain.RemoteFailureTimeout
1154 }
1155 return kind, availabilityFor(kind), true
1156 }
1157 switch result.snapshot.Availability {
1158 case domain.RemoteAvailabilityReachable:
1159 return domain.RemoteFailureNone, domain.RemoteAvailabilityReachable, false
1160 case domain.RemoteAvailabilityNoDaemon:
1161 return domain.RemoteFailureNone, domain.RemoteAvailabilityNoDaemon, false
1162 case domain.RemoteAvailabilityIncompatible:
1163 return domain.RemoteFailureIncompatible, domain.RemoteAvailabilityIncompatible, true
1164 case domain.RemoteAvailabilityAuthFailed:
1165 return domain.RemoteFailureAuthentication, domain.RemoteAvailabilityAuthFailed, true
1166 case domain.RemoteAvailabilityInvalidResponse:
1167 return domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
1168 default:
1169 // Unknown or Unreachable without an error still means reachability was
1170 // not confirmed, so it retries on the typed backoff cadence.
1171 return domain.RemoteFailureTransport, domain.RemoteAvailabilityUnreachable, true
1172 }
1173}
1174
1175// availabilityFor maps a typed failure kind to the published availability.
1176// The typed kind is preserved separately for notice policy.
1177func availabilityFor(kind domain.RemoteFailureKind) domain.RemoteAvailability {
1178 switch kind {
1179 case domain.RemoteFailureAuthentication:
1180 return domain.RemoteAvailabilityAuthFailed
1181 case domain.RemoteFailureIncompatible:
1182 return domain.RemoteAvailabilityIncompatible
1183 case domain.RemoteFailureInvalidResponse:
1184 return domain.RemoteAvailabilityInvalidResponse
1185 default:
1186 return domain.RemoteAvailabilityUnreachable
1187 }
1188}
1189
1190// demandRetryDelay is the watched-host failure backoff: the demand cadence
1191// doubled per consecutive failure, capped at demandRetryMax.
1192func demandRetryDelay(base time.Duration, failures uint) time.Duration {
1193 delay := base
1194 for i := uint(1); i < failures && delay < demandRetryMax; i++ {
1195 delay *= 2
1196 }
1197 return min(delay, max(base, demandRetryMax))
1198}
1199
1200// retryDelay is the capped exponential retry cadence for a failure streak.
1201func (r *Registry) retryDelay(failures uint) time.Duration {
1202 delay := r.retryBase
1203 for i := uint(1); i < failures; i++ {
1204 if delay >= r.retryMax {
1205 return r.retryMax
1206 }
1207 delay *= 2
1208 }
1209 if delay > r.retryMax {
1210 return r.retryMax
1211 }
1212 return delay
1213}
1214
1215// jittered spreads a base interval deterministically across endpoint and
1216// attempt: identical inputs always yield the identical duration, and the
1217// result stays within 90%-110% of the base. No wall clock or global
1218// randomness participates, so retry and healthy-refresh schedules are
1219// reproducible instead of racing a shared PRNG.
1220func jittered(base time.Duration, endpoint string, attempt uint64) time.Duration {
1221 if base <= 0 {
1222 return base
1223 }
1224 h := fnv.New32a()
1225 h.Write([]byte(endpoint))
1226 var counter [8]byte
1227 binary.BigEndian.PutUint64(counter[:], attempt)
1228 h.Write(counter[:])
1229 spread := int64(h.Sum32() % 21)
1230 return base*90/100 + time.Duration(spread)*base/100
1231}
1232
1233func (r *Registry) nextDelay(now time.Time) time.Duration {
1234 r.mu.Lock()
1235 defer r.mu.Unlock()
1236 if r.demand == 0 {
1237 if r.localPending || len(r.pending) > 0 {
1238 return 0
1239 }
1240 return time.Hour
1241 }
1242 var earliest time.Time
1243 if r.local != nil && r.localAttempt == nil {
1244 if r.localHost.NextDue.IsZero() {
1245 return 0
1246 }
1247 if earliest.IsZero() || r.localHost.NextDue.Before(earliest) {
1248 earliest = r.localHost.NextDue
1249 }
1250 }
1251 for endpoint, host := range r.hosts {
1252 // An observation already in flight is completed by its result, never
1253 // by the schedule; waiting on it here would spin the run loop.
1254 if _, inflight := r.inflight[endpoint]; inflight {
1255 continue
1256 }
1257 if r.pending[endpoint] || host.NextDue.IsZero() {
1258 return 0
1259 }
1260 if earliest.IsZero() || host.NextDue.Before(earliest) {
1261 earliest = host.NextDue
1262 }
1263 }
1264 if earliest.IsZero() {
1265 return time.Hour
1266 }
1267 if delay := earliest.Sub(now); delay > 0 {
1268 return delay
1269 }
1270 return 0
1271}
1272
1273// registrationsUnchanged reports exact membership equality: the same endpoints
1274// in the same registration order, bound to the same registration identities and
1275// policy. Publication order and presentation rank are functions of durable
1276// membership order, so a reordering of identical records is a membership change
1277// that must republish; otherwise the same durable state yields different
1278// publications depending on process history. Projection fields such as
1279// availability, checking, or inventory never participate.
1280func registrationsUnchanged(current, next map[string]ports.BrokerDaemonObservation, currentOrder, nextOrder []string) bool {
1281 if len(current) != len(next) {
1282 return false
1283 }
1284 if !slices.Equal(currentOrder, nextOrder) {
1285 return false
1286 }
1287 for endpoint, candidate := range next {
1288 existing, ok := current[endpoint]
1289 if !ok || !existing.Registration.Equal(candidate.Registration) || existing.Policy != candidate.Policy {
1290 return false
1291 }
1292 }
1293 return true
1294}
1295
1296// retireLocked records the tombstone for one removed registration at the
1297// revision of the publication that carries it. Mutation paths refuse an
1298// exhausted series before retiring, so the retirement revision can never wrap
1299// to zero or fall below the last published one.
1300func (r *Registry) retireLocked(registration domain.RemoteRegistration) {
1301 retired := r.revision + 1
1302 r.tombstones[registration.Endpoint] = ports.BrokerHostTombstone{
1303 Endpoint: registration.Endpoint,
1304 Registration: registration,
1305 RetiredRevision: retired,
1306 }
1307 r.pruneTombstonesLocked()
1308}
1309
1310// revisionExhaustedLocked reports whether this epoch has no revision left to
1311// publish, latching the exhausted state on first detection. Callers must hold
1312// r.mu.
1313func (r *Registry) revisionExhaustedLocked() bool {
1314 if r.revisionExhausted {
1315 return true
1316 }
1317 if r.revision != ^ports.BrokerRevision(0) {
1318 return false
1319 }
1320 r.revisionExhausted = true
1321 r.log.Error("broker: revision series exhausted; refusing further publication", "epoch", r.epoch)
1322 return true
1323}
1324
1325// pruneTombstonesLocked keeps only the newest BrokerMaxTombstones retirements
1326// so an endpoint that churns can never grow the published set without bound.
1327func (r *Registry) pruneTombstonesLocked() {
1328 for len(r.tombstones) > ports.BrokerMaxTombstones {
1329 oldest := ""
1330 var oldestRevision ports.BrokerRevision
1331 for endpoint, tombstone := range r.tombstones {
1332 if oldest == "" || tombstone.RetiredRevision < oldestRevision ||
1333 (tombstone.RetiredRevision == oldestRevision && endpoint < oldest) {
1334 oldest, oldestRevision = endpoint, tombstone.RetiredRevision
1335 }
1336 }
1337 delete(r.tombstones, oldest)
1338 }
1339}
1340
1341// tombstonesLocked renders the bounded tombstone publication, newest first.
1342// Live endpoints are never published as tombstones.
1343func (r *Registry) tombstonesLocked() []ports.BrokerHostTombstone {
1344 if len(r.tombstones) == 0 {
1345 return nil
1346 }
1347 out := make([]ports.BrokerHostTombstone, 0, len(r.tombstones))
1348 for endpoint, tombstone := range r.tombstones {
1349 if _, live := r.hosts[endpoint]; live {
1350 continue
1351 }
1352 out = append(out, tombstone)
1353 }
1354 sort.Slice(out, func(i, j int) bool {
1355 if out[i].RetiredRevision != out[j].RetiredRevision {
1356 return out[i].RetiredRevision > out[j].RetiredRevision
1357 }
1358 return out[i].Endpoint < out[j].Endpoint
1359 })
1360 if len(out) > ports.BrokerMaxTombstones {
1361 out = out[:ports.BrokerMaxTombstones]
1362 }
1363 return out
1364}
1365
1366func (r *Registry) publishLocked(persist bool) {
1367 // Fail closed on an exhausted series: a wrapped revision would be zero, and
1368 // a wrapped series would publish revisions below the newest one, either of
1369 // which silently freezes durable persistence. The last valid snapshot stays
1370 // published, nothing is stored, and no subscriber is woken for a publication
1371 // that did not happen.
1372 if r.revisionExhaustedLocked() {
1373 return
1374 }
1375 // Remote daemons are published in registration order. The broker's own
1376 // machine daemon, when a producer is configured, is the
1377 // prepended entry at index zero; without a producer the registry produces no
1378 // local observation and never invents one.
1379 daemons := make([]ports.BrokerDaemonObservation, 0, len(r.hosts)+1)
1380 if r.local != nil {
1381 daemons = append(daemons, r.localHost.Clone())
1382 }
1383 for _, endpoint := range r.order {
1384 host, ok := r.hosts[endpoint]
1385 if !ok {
1386 continue
1387 }
1388 daemons = append(daemons, host.Clone())
1389 }
1390 snapshot := ports.BrokerSnapshot{Epoch: r.epoch, Revision: r.revision + 1, Daemons: daemons, Removed: r.tombstonesLocked()}
1391 if err := snapshot.Validate(); err != nil {
1392 // Fail closed: a publication the wire refuses aborts every client
1393 // connection on subscribe, so an invalid projection is never published.
1394 // The last valid snapshot stays current, nothing is persisted, and the
1395 // revision series does not advance past a publication that never
1396 // happened.
1397 r.log.Error("broker: refused invalid snapshot publication", "revision", r.revision+1, "err", err)
1398 return
1399 }
1400 r.revision = snapshot.Revision
1401 r.current.Store(&snapshot)
1402 if persist {
1403 // enqueue only stages an immutable publication: rendering the durable
1404 // copy and every store call happen on the writer goroutine, never
1405 // under the registry lock and never on the probe schedule.
1406 r.store.enqueue(snapshot)
1407 }
1408 for sub := range r.subs {
1409 sub.offer()
1410 }
1411}
1412
1413// durable returns the persistable copy of a publication, rendered by the
1414// writer goroutine with the shared rule (ports.BrokerSnapshot.DurableProjection).
1415// The registry may publish the process-local daemon, which is never durable,
1416// so it is dropped first; the store would refuse it otherwise. Restored
1417// observations regain policy from membership (see stampHostAuthority).
1418func durable(snapshot ports.BrokerSnapshot) ports.BrokerSnapshot {
1419 return snapshot.WithoutLocal().DurableProjection()
1420}
1421
1422// snapshotWriter serializes durable snapshot writes off the registry lock.
1423// Enqueue never blocks on store I/O and always coalesces to the newest
1424// revision; Store calls run on one goroutine, so the durable store never sees
1425// an out-of-order write. A failed write is logged and retried by the next
1426// publication, so a store fault never stalls publication. Close flushes the
1427// newest pending snapshot, stops the writer, and waits, so no write outlives
1428// shutdown.
1429type snapshotWriter struct {
1430 store ports.BrokerSnapshotStore
1431 log *slog.Logger
1432
1433 mu sync.Mutex
1434 pending *ports.BrokerSnapshot
1435 written ports.BrokerRevision
1436 // last is the durable content of the newest successful write with its
1437 // revision and observation timestamps cleared. Only the writer goroutine
1438 // touches it.
1439 last *ports.BrokerSnapshot
1440 started bool
1441 closed bool
1442 wake chan struct{}
1443 stop chan struct{}
1444 done chan struct{}
1445}
1446
1447func newSnapshotWriter(store ports.BrokerSnapshotStore, log *slog.Logger) *snapshotWriter {
1448 return &snapshotWriter{
1449 store: store, log: log,
1450 wake: make(chan struct{}, 1), stop: make(chan struct{}), done: make(chan struct{}),
1451 }
1452}
1453
1454// enqueue stages a snapshot for durable persistence. It coalesces to the
1455// newest revision, starts the writer on first use, and never blocks on the
1456// store or copies session payloads. Enqueue after Close is refused: nothing is
1457// written after shutdown.
1458func (w *snapshotWriter) enqueue(snapshot ports.BrokerSnapshot) {
1459 w.mu.Lock()
1460 if w.closed {
1461 w.mu.Unlock()
1462 w.log.Debug("broker: durable snapshot write refused after shutdown", "revision", snapshot.Revision)
1463 return
1464 }
1465 if w.pending != nil {
1466 if snapshot.Revision <= w.pending.Revision {
1467 w.mu.Unlock()
1468 return
1469 }
1470 } else if snapshot.Revision <= w.written {
1471 w.mu.Unlock()
1472 return
1473 }
1474 // The publication is immutable after publication, so staging it is a value
1475 // copy; the defensive durable copy is rendered on the writer goroutine.
1476 w.pending = &snapshot
1477 if !w.started {
1478 w.started = true
1479 go w.run()
1480 }
1481 w.mu.Unlock()
1482 select {
1483 case w.wake <- struct{}{}:
1484 default:
1485 }
1486}
1487
1488func (w *snapshotWriter) run() {
1489 defer close(w.done)
1490 for {
1491 select {
1492 case <-w.stop:
1493 w.flush()
1494 return
1495 case <-w.wake:
1496 w.flush()
1497 }
1498 }
1499}
1500
1501// flush writes the newest staged snapshot, skipping revisions the store
1502// already holds.
1503func (w *snapshotWriter) flush() {
1504 for {
1505 w.mu.Lock()
1506 snapshot := w.pending
1507 written := w.written
1508 w.pending = nil
1509 w.mu.Unlock()
1510 if snapshot == nil {
1511 return
1512 }
1513 if snapshot.Revision <= written {
1514 continue
1515 }
1516 next := durable(*snapshot)
1517 key := durableContent(next)
1518 if w.last != nil && reflect.DeepEqual(*w.last, key) {
1519 // Only observation timestamps moved: the stored copy already
1520 // carries the same durable state, so skip the rewrite and fsync.
1521 // A restart re-observes every host before trusting freshness.
1522 continue
1523 }
1524 if err := w.store.Store(next); err != nil {
1525 w.last = nil
1526 w.log.Error("broker: durable snapshot write failed", "revision", snapshot.Revision, "err", err)
1527 continue
1528 }
1529 w.last = &key
1530 w.mu.Lock()
1531 if snapshot.Revision > w.written {
1532 w.written = snapshot.Revision
1533 }
1534 w.mu.Unlock()
1535 }
1536}
1537
1538// durableContent returns the durable projection without the fields that change
1539// on every routine observation: the revision and the attempt, success, and
1540// due timestamps.
1541func durableContent(snapshot ports.BrokerSnapshot) ports.BrokerSnapshot {
1542 out := snapshot
1543 out.Revision = 0
1544 out.Daemons = make([]ports.BrokerDaemonObservation, len(snapshot.Daemons))
1545 for i, daemon := range snapshot.Daemons {
1546 daemon.LastAttempt = time.Time{}
1547 daemon.LastSuccess = time.Time{}
1548 daemon.NextDue = time.Time{}
1549 out.Daemons[i] = daemon
1550 }
1551 return out
1552}
1553
1554// close flushes the newest staged snapshot, stops the writer, and waits for
1555// the drain to finish. It never returns before the in-flight write completes.
1556func (w *snapshotWriter) close() {
1557 w.mu.Lock()
1558 if w.closed {
1559 w.mu.Unlock()
1560 return
1561 }
1562 w.closed = true
1563 started := w.started
1564 w.mu.Unlock()
1565 if !started {
1566 return
1567 }
1568 close(w.stop)
1569 <-w.done
1570}
1571
1572type subscription struct {
1573 changed chan struct{}
1574 once sync.Once
1575 close func()
1576}
1577
1578func (s *subscription) Changed() <-chan struct{} { return s.changed }
1579func (s *subscription) Close() { s.once.Do(s.close) }
1580func (s *subscription) offer() {
1581 select {
1582 case s.changed <- struct{}{}:
1583 default:
1584 }
1585}