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 if !r.pending[endpoint] && (r.demand == 0 || !host.NextDue.IsZero() && now.Before(host.NextDue)) {
955 continue
956 }
957 delete(r.pending, endpoint)
958 r.attempts++
959 attempt := r.attempts
960 probeCtx, cancel := context.WithCancel(ctx)
961 r.inflight[endpoint] = &probeAttempt{token: attempt, cancel: cancel}
962 started = true
963 host.Checking = true
964 host.LastAttempt = now
965 r.hosts[endpoint] = host
966 registration := host.Registration
967 // The deadline is armed here, under r.mu and at dispatch time, so the
968 // attempt's bound starts with the attempt rather than whenever its
969 // goroutine is scheduled.
970 deadline := r.clock.NewTimer(r.probeTimeout)
971 go func(endpoint string, registration domain.RemoteRegistration, attempt uint64, probeCtx context.Context) {
972 snapshot, err := observeBounded(probeCtx, deadline, func(ctx context.Context) (ports.BrokerDaemonObservation, error) {
973 return r.probe.Probe(ctx, registration)
974 })
975 result := probeResult{endpoint: endpoint, registration: registration, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}
976 select {
977 case results <- result:
978 case <-probeCtx.Done():
979 // The attempt was retired (replaced, removed, or shutdown), so
980 // its completion is already fenced: dropping it keeps churn from
981 // queueing results the run loop would only discard.
982 }
983 }(endpoint, registration, attempt, probeCtx)
984 }
985 // Only an observation that actually started changes the projection; an
986 // idle dispatch must not bump the revision or wake subscribers. The
987 // in-flight projection is never persisted: Checking is transient.
988 if r.dispatchLocalLocked(ctx, now, results) {
989 started = true
990 }
991 if started {
992 r.publishLocked(false)
993 }
994 r.mu.Unlock()
995}
996
997// observeBounded runs one observation under its attempt deadline. retire is the
998// attempt's retirement context: a replaced, removed, or settled attempt returns
999// at once and its completion is fenced by the caller. The probe itself runs on
1000// a context derived from retire that the deadline also cancels, so a
1001// ctx-honoring probe unwinds its dial, handshake, or exchange when either
1002// fires. The attempt never waits for the probe past its deadline: a probe that
1003// ignores cancellation is abandoned (its late result is discarded) and the
1004// attempt still reports errProbeTimeout, so the endpoint's single in-flight
1005// slot is always released.
1006func observeBounded(retire context.Context, deadline ports.Timer, observe func(context.Context) (ports.BrokerDaemonObservation, error)) (ports.BrokerDaemonObservation, error) {
1007 defer deadline.Stop()
1008 probeCtx, cancel := context.WithCancelCause(retire)
1009 defer cancel(nil)
1010 type outcome struct {
1011 snapshot ports.BrokerDaemonObservation
1012 err error
1013 }
1014 done := make(chan outcome, 1)
1015 go func() {
1016 snapshot, err := observe(probeCtx)
1017 done <- outcome{snapshot: snapshot, err: err}
1018 }()
1019 select {
1020 case result := <-done:
1021 return result.snapshot, result.err
1022 case <-deadline.C():
1023 cancel(errProbeTimeout)
1024 return ports.BrokerDaemonObservation{}, errProbeTimeout
1025 case <-retire.Done():
1026 return ports.BrokerDaemonObservation{}, retire.Err()
1027 }
1028}
1029
1030func (r *Registry) apply(result probeResult) {
1031 r.mu.Lock()
1032 defer r.mu.Unlock()
1033 if result.local {
1034 r.applyLocal(result)
1035 return
1036 }
1037 current, ok := r.hosts[result.endpoint]
1038 attempt := r.inflight[result.endpoint]
1039 if !ok || !current.Registration.Equal(result.registration) || attempt == nil || attempt.token != result.attempt {
1040 // A stale completion touches neither current authority nor the
1041 // in-flight slot, which may already belong to a newer registration or
1042 // attempt.
1043 return
1044 }
1045 delete(r.inflight, result.endpoint)
1046 // The probe has returned, so releasing its derived context here only
1047 // frees the attempt's resources; the record is gone before any new
1048 // attempt can start.
1049 attempt.cancel()
1050 current.Checking = false
1051 current.LastAttempt = result.at
1052
1053 kind, availability, failed := observationOutcome(result)
1054 if !failed && availability == domain.RemoteAvailabilityNoDaemon {
1055 current.Availability = availability
1056 current.LastSuccess = result.at
1057 // An absent daemon is stable: demand does not require a fresh SSH
1058 // handshake every two seconds just to confirm it is still absent.
1059 current.NextDue = result.at.Add(r.jitter(r.freshFor, result.endpoint, result.attempt))
1060 current.ConsecutiveFailures = 0
1061 current.LastFailure = domain.RemoteFailure{}
1062 current.Identity = ""
1063 current.Incarnation = ports.BrokerDaemonIncarnation{}
1064 current.ProtocolVersion = 0
1065 current.Capabilities = 0
1066 current.InventoryKnown = false
1067 current.Sessions = nil
1068 r.hosts[result.endpoint] = current
1069 r.publishLocked(true)
1070 return
1071 }
1072 if !failed {
1073 // Configured authority comes only from the registry's own projection:
1074 // a probe result supplies observed state, never endpoint, registration,
1075 // policy, display origin, or rank.
1076 observed := stampObservationAuthority(result.snapshot.Clone(), current)
1077 observed.Checking = false
1078 observed.LastAttempt = result.at
1079 observed.LastSuccess = result.at
1080 observed.NextDue = result.at.Add(r.jitter(r.freshForLocked(false), result.endpoint, result.attempt))
1081 observed.ConsecutiveFailures = 0
1082 observed.LastFailure = domain.RemoteFailure{}
1083 observed.FailureEpisode = current.FailureEpisode
1084 // The published observation must be fully valid once authority is
1085 // stamped: a probe that invents partial identity (identity without a
1086 // protocol version) or an inventory without a known catalogue is an
1087 // invalid response, never a published observation. The durable rule is
1088 // re-checked too: a projection the offline store would reject is never
1089 // published as if it had persisted. The store applies the same rule.
1090 if err := observed.Validate(); err != nil {
1091 r.log.Debug("broker: observation is invalid", "endpoint", result.endpoint, "err", err)
1092 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
1093 } else if err := ports.ValidateDurableHostProjection(observed); err != nil {
1094 r.log.Debug("broker: observation is not persistable", "endpoint", result.endpoint, "err", err)
1095 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
1096 } else {
1097 r.hosts[result.endpoint] = observed
1098 r.publishLocked(true)
1099 return
1100 }
1101 }
1102 // Every non-reachable outcome is a typed failure on the capped retry
1103 // cadence: the last success, last-known inventory, and failure episode
1104 // counter stay authoritative.
1105 current.Availability = availability
1106 if current.ConsecutiveFailures == 0 {
1107 current.FailureEpisode++
1108 }
1109 current.ConsecutiveFailures++
1110 current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
1111 delay := r.retryDelay(current.ConsecutiveFailures)
1112 if r.demand > 0 {
1113 delay = demandRetryDelay(r.demandFreshForRemote, current.ConsecutiveFailures)
1114 }
1115 current.NextDue = result.at.Add(r.jitter(delay, result.endpoint, result.attempt))
1116 r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err, "cause", errors.Unwrap(result.err))
1117 r.hosts[result.endpoint] = current
1118 r.publishLocked(true)
1119}
1120
1121// observationOutcome classifies a completed observation. It reports
1122// failed=false only for a confirmed reachable result; every other outcome
1123// carries the typed failure kind and the availability to publish. The caller
1124// additionally checks the adopted projection against the durable rules
1125// (ports.ValidateDurableHostProjection), which cover the session bound and
1126// catalogue validity for the exact incarnation and success time.
1127func observationOutcome(result probeResult) (domain.RemoteFailureKind, domain.RemoteAvailability, bool) {
1128 if result.err != nil {
1129 kind := domain.RemoteFailureTransport
1130 var typed domain.RemoteFailure
1131 var brokerErr ports.BrokerError
1132 switch {
1133 case errors.As(result.err, &typed) && typed.Kind != domain.RemoteFailureNone:
1134 // The dialer classified the failure (for example SSH authentication).
1135 kind = typed.Kind
1136 case errors.As(result.err, &brokerErr) && (brokerErr.Code == ports.BrokerErrorConflictingPolicy || brokerErr.Code == ports.BrokerErrorIncompatible):
1137 // SSH worked; the daemon's identity or version was refused.
1138 kind = domain.RemoteFailureIncompatible
1139 case errors.Is(result.err, context.DeadlineExceeded):
1140 kind = domain.RemoteFailureTimeout
1141 }
1142 return kind, availabilityFor(kind), true
1143 }
1144 switch result.snapshot.Availability {
1145 case domain.RemoteAvailabilityReachable:
1146 return domain.RemoteFailureNone, domain.RemoteAvailabilityReachable, false
1147 case domain.RemoteAvailabilityNoDaemon:
1148 return domain.RemoteFailureNone, domain.RemoteAvailabilityNoDaemon, false
1149 case domain.RemoteAvailabilityIncompatible:
1150 return domain.RemoteFailureIncompatible, domain.RemoteAvailabilityIncompatible, true
1151 case domain.RemoteAvailabilityAuthFailed:
1152 return domain.RemoteFailureAuthentication, domain.RemoteAvailabilityAuthFailed, true
1153 case domain.RemoteAvailabilityInvalidResponse:
1154 return domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
1155 default:
1156 // Unknown or Unreachable without an error still means reachability was
1157 // not confirmed, so it retries on the typed backoff cadence.
1158 return domain.RemoteFailureTransport, domain.RemoteAvailabilityUnreachable, true
1159 }
1160}
1161
1162// availabilityFor maps a typed failure kind to the published availability.
1163// The typed kind is preserved separately for notice policy.
1164func availabilityFor(kind domain.RemoteFailureKind) domain.RemoteAvailability {
1165 switch kind {
1166 case domain.RemoteFailureAuthentication:
1167 return domain.RemoteAvailabilityAuthFailed
1168 case domain.RemoteFailureIncompatible:
1169 return domain.RemoteAvailabilityIncompatible
1170 case domain.RemoteFailureInvalidResponse:
1171 return domain.RemoteAvailabilityInvalidResponse
1172 default:
1173 return domain.RemoteAvailabilityUnreachable
1174 }
1175}
1176
1177// demandRetryDelay is the watched-host failure backoff: the demand cadence
1178// doubled per consecutive failure, capped at demandRetryMax.
1179func demandRetryDelay(base time.Duration, failures uint) time.Duration {
1180 delay := base
1181 for i := uint(1); i < failures && delay < demandRetryMax; i++ {
1182 delay *= 2
1183 }
1184 return min(delay, max(base, demandRetryMax))
1185}
1186
1187// retryDelay is the capped exponential retry cadence for a failure streak.
1188func (r *Registry) retryDelay(failures uint) time.Duration {
1189 delay := r.retryBase
1190 for i := uint(1); i < failures; i++ {
1191 if delay >= r.retryMax {
1192 return r.retryMax
1193 }
1194 delay *= 2
1195 }
1196 if delay > r.retryMax {
1197 return r.retryMax
1198 }
1199 return delay
1200}
1201
1202// jittered spreads a base interval deterministically across endpoint and
1203// attempt: identical inputs always yield the identical duration, and the
1204// result stays within 90%-110% of the base. No wall clock or global
1205// randomness participates, so retry and healthy-refresh schedules are
1206// reproducible instead of racing a shared PRNG.
1207func jittered(base time.Duration, endpoint string, attempt uint64) time.Duration {
1208 if base <= 0 {
1209 return base
1210 }
1211 h := fnv.New32a()
1212 h.Write([]byte(endpoint))
1213 var counter [8]byte
1214 binary.BigEndian.PutUint64(counter[:], attempt)
1215 h.Write(counter[:])
1216 spread := int64(h.Sum32() % 21)
1217 return base*90/100 + time.Duration(spread)*base/100
1218}
1219
1220func (r *Registry) nextDelay(now time.Time) time.Duration {
1221 r.mu.Lock()
1222 defer r.mu.Unlock()
1223 if r.demand == 0 {
1224 if r.localPending || len(r.pending) > 0 {
1225 return 0
1226 }
1227 return time.Hour
1228 }
1229 var earliest time.Time
1230 if r.local != nil && r.localAttempt == nil {
1231 if r.localHost.NextDue.IsZero() {
1232 return 0
1233 }
1234 if earliest.IsZero() || r.localHost.NextDue.Before(earliest) {
1235 earliest = r.localHost.NextDue
1236 }
1237 }
1238 for endpoint, host := range r.hosts {
1239 // An observation already in flight is completed by its result, never
1240 // by the schedule; waiting on it here would spin the run loop.
1241 if _, inflight := r.inflight[endpoint]; inflight {
1242 continue
1243 }
1244 if r.pending[endpoint] || host.NextDue.IsZero() {
1245 return 0
1246 }
1247 if earliest.IsZero() || host.NextDue.Before(earliest) {
1248 earliest = host.NextDue
1249 }
1250 }
1251 if earliest.IsZero() {
1252 return time.Hour
1253 }
1254 if delay := earliest.Sub(now); delay > 0 {
1255 return delay
1256 }
1257 return 0
1258}
1259
1260// registrationsUnchanged reports exact membership equality: the same endpoints
1261// in the same registration order, bound to the same registration identities and
1262// policy. Publication order and presentation rank are functions of durable
1263// membership order, so a reordering of identical records is a membership change
1264// that must republish; otherwise the same durable state yields different
1265// publications depending on process history. Projection fields such as
1266// availability, checking, or inventory never participate.
1267func registrationsUnchanged(current, next map[string]ports.BrokerDaemonObservation, currentOrder, nextOrder []string) bool {
1268 if len(current) != len(next) {
1269 return false
1270 }
1271 if !slices.Equal(currentOrder, nextOrder) {
1272 return false
1273 }
1274 for endpoint, candidate := range next {
1275 existing, ok := current[endpoint]
1276 if !ok || !existing.Registration.Equal(candidate.Registration) || existing.Policy != candidate.Policy {
1277 return false
1278 }
1279 }
1280 return true
1281}
1282
1283// retireLocked records the tombstone for one removed registration at the
1284// revision of the publication that carries it. Mutation paths refuse an
1285// exhausted series before retiring, so the retirement revision can never wrap
1286// to zero or fall below the last published one.
1287func (r *Registry) retireLocked(registration domain.RemoteRegistration) {
1288 retired := r.revision + 1
1289 r.tombstones[registration.Endpoint] = ports.BrokerHostTombstone{
1290 Endpoint: registration.Endpoint,
1291 Registration: registration,
1292 RetiredRevision: retired,
1293 }
1294 r.pruneTombstonesLocked()
1295}
1296
1297// revisionExhaustedLocked reports whether this epoch has no revision left to
1298// publish, latching the exhausted state on first detection. Callers must hold
1299// r.mu.
1300func (r *Registry) revisionExhaustedLocked() bool {
1301 if r.revisionExhausted {
1302 return true
1303 }
1304 if r.revision != ^ports.BrokerRevision(0) {
1305 return false
1306 }
1307 r.revisionExhausted = true
1308 r.log.Error("broker: revision series exhausted; refusing further publication", "epoch", r.epoch)
1309 return true
1310}
1311
1312// pruneTombstonesLocked keeps only the newest BrokerMaxTombstones retirements
1313// so an endpoint that churns can never grow the published set without bound.
1314func (r *Registry) pruneTombstonesLocked() {
1315 for len(r.tombstones) > ports.BrokerMaxTombstones {
1316 oldest := ""
1317 var oldestRevision ports.BrokerRevision
1318 for endpoint, tombstone := range r.tombstones {
1319 if oldest == "" || tombstone.RetiredRevision < oldestRevision ||
1320 (tombstone.RetiredRevision == oldestRevision && endpoint < oldest) {
1321 oldest, oldestRevision = endpoint, tombstone.RetiredRevision
1322 }
1323 }
1324 delete(r.tombstones, oldest)
1325 }
1326}
1327
1328// tombstonesLocked renders the bounded tombstone publication, newest first.
1329// Live endpoints are never published as tombstones.
1330func (r *Registry) tombstonesLocked() []ports.BrokerHostTombstone {
1331 if len(r.tombstones) == 0 {
1332 return nil
1333 }
1334 out := make([]ports.BrokerHostTombstone, 0, len(r.tombstones))
1335 for endpoint, tombstone := range r.tombstones {
1336 if _, live := r.hosts[endpoint]; live {
1337 continue
1338 }
1339 out = append(out, tombstone)
1340 }
1341 sort.Slice(out, func(i, j int) bool {
1342 if out[i].RetiredRevision != out[j].RetiredRevision {
1343 return out[i].RetiredRevision > out[j].RetiredRevision
1344 }
1345 return out[i].Endpoint < out[j].Endpoint
1346 })
1347 if len(out) > ports.BrokerMaxTombstones {
1348 out = out[:ports.BrokerMaxTombstones]
1349 }
1350 return out
1351}
1352
1353func (r *Registry) publishLocked(persist bool) {
1354 // Fail closed on an exhausted series: a wrapped revision would be zero, and
1355 // a wrapped series would publish revisions below the newest one, either of
1356 // which silently freezes durable persistence. The last valid snapshot stays
1357 // published, nothing is stored, and no subscriber is woken for a publication
1358 // that did not happen.
1359 if r.revisionExhaustedLocked() {
1360 return
1361 }
1362 // Remote daemons are published in registration order. The broker's own
1363 // machine daemon, when a producer is configured, is the
1364 // prepended entry at index zero; without a producer the registry produces no
1365 // local observation and never invents one.
1366 daemons := make([]ports.BrokerDaemonObservation, 0, len(r.hosts)+1)
1367 if r.local != nil {
1368 daemons = append(daemons, r.localHost.Clone())
1369 }
1370 for _, endpoint := range r.order {
1371 host, ok := r.hosts[endpoint]
1372 if !ok {
1373 continue
1374 }
1375 daemons = append(daemons, host.Clone())
1376 }
1377 snapshot := ports.BrokerSnapshot{Epoch: r.epoch, Revision: r.revision + 1, Daemons: daemons, Removed: r.tombstonesLocked()}
1378 if err := snapshot.Validate(); err != nil {
1379 // Fail closed: a publication the wire refuses aborts every client
1380 // connection on subscribe, so an invalid projection is never published.
1381 // The last valid snapshot stays current, nothing is persisted, and the
1382 // revision series does not advance past a publication that never
1383 // happened.
1384 r.log.Error("broker: refused invalid snapshot publication", "revision", r.revision+1, "err", err)
1385 return
1386 }
1387 r.revision = snapshot.Revision
1388 r.current.Store(&snapshot)
1389 if persist {
1390 // enqueue only stages an immutable publication: rendering the durable
1391 // copy and every store call happen on the writer goroutine, never
1392 // under the registry lock and never on the probe schedule.
1393 r.store.enqueue(snapshot)
1394 }
1395 for sub := range r.subs {
1396 sub.offer()
1397 }
1398}
1399
1400// durable returns the persistable copy of a publication, rendered by the
1401// writer goroutine with the shared rule (ports.BrokerSnapshot.DurableProjection).
1402// The registry may publish the process-local daemon, which is never durable,
1403// so it is dropped first; the store would refuse it otherwise. Restored
1404// observations regain policy from membership (see stampHostAuthority).
1405func durable(snapshot ports.BrokerSnapshot) ports.BrokerSnapshot {
1406 return snapshot.WithoutLocal().DurableProjection()
1407}
1408
1409// snapshotWriter serializes durable snapshot writes off the registry lock.
1410// Enqueue never blocks on store I/O and always coalesces to the newest
1411// revision; Store calls run on one goroutine, so the durable store never sees
1412// an out-of-order write. A failed write is logged and retried by the next
1413// publication, so a store fault never stalls publication. Close flushes the
1414// newest pending snapshot, stops the writer, and waits, so no write outlives
1415// shutdown.
1416type snapshotWriter struct {
1417 store ports.BrokerSnapshotStore
1418 log *slog.Logger
1419
1420 mu sync.Mutex
1421 pending *ports.BrokerSnapshot
1422 written ports.BrokerRevision
1423 started bool
1424 closed bool
1425 wake chan struct{}
1426 stop chan struct{}
1427 done chan struct{}
1428}
1429
1430func newSnapshotWriter(store ports.BrokerSnapshotStore, log *slog.Logger) *snapshotWriter {
1431 return &snapshotWriter{
1432 store: store, log: log,
1433 wake: make(chan struct{}, 1), stop: make(chan struct{}), done: make(chan struct{}),
1434 }
1435}
1436
1437// enqueue stages a snapshot for durable persistence. It coalesces to the
1438// newest revision, starts the writer on first use, and never blocks on the
1439// store or copies session payloads. Enqueue after Close is refused: nothing is
1440// written after shutdown.
1441func (w *snapshotWriter) enqueue(snapshot ports.BrokerSnapshot) {
1442 w.mu.Lock()
1443 if w.closed {
1444 w.mu.Unlock()
1445 w.log.Debug("broker: durable snapshot write refused after shutdown", "revision", snapshot.Revision)
1446 return
1447 }
1448 if w.pending != nil {
1449 if snapshot.Revision <= w.pending.Revision {
1450 w.mu.Unlock()
1451 return
1452 }
1453 } else if snapshot.Revision <= w.written {
1454 w.mu.Unlock()
1455 return
1456 }
1457 // The publication is immutable after publication, so staging it is a value
1458 // copy; the defensive durable copy is rendered on the writer goroutine.
1459 w.pending = &snapshot
1460 if !w.started {
1461 w.started = true
1462 go w.run()
1463 }
1464 w.mu.Unlock()
1465 select {
1466 case w.wake <- struct{}{}:
1467 default:
1468 }
1469}
1470
1471func (w *snapshotWriter) run() {
1472 defer close(w.done)
1473 for {
1474 select {
1475 case <-w.stop:
1476 w.flush()
1477 return
1478 case <-w.wake:
1479 w.flush()
1480 }
1481 }
1482}
1483
1484// flush writes the newest staged snapshot, skipping revisions the store
1485// already holds.
1486func (w *snapshotWriter) flush() {
1487 for {
1488 w.mu.Lock()
1489 snapshot := w.pending
1490 written := w.written
1491 w.pending = nil
1492 w.mu.Unlock()
1493 if snapshot == nil {
1494 return
1495 }
1496 if snapshot.Revision <= written {
1497 continue
1498 }
1499 if err := w.store.Store(durable(*snapshot)); err != nil {
1500 w.log.Error("broker: durable snapshot write failed", "revision", snapshot.Revision, "err", err)
1501 continue
1502 }
1503 w.mu.Lock()
1504 if snapshot.Revision > w.written {
1505 w.written = snapshot.Revision
1506 }
1507 w.mu.Unlock()
1508 }
1509}
1510
1511// close flushes the newest staged snapshot, stops the writer, and waits for
1512// the drain to finish. It never returns before the in-flight write completes.
1513func (w *snapshotWriter) close() {
1514 w.mu.Lock()
1515 if w.closed {
1516 w.mu.Unlock()
1517 return
1518 }
1519 w.closed = true
1520 started := w.started
1521 w.mu.Unlock()
1522 if !started {
1523 return
1524 }
1525 close(w.stop)
1526 <-w.done
1527}
1528
1529type subscription struct {
1530 changed chan struct{}
1531 once sync.Once
1532 close func()
1533}
1534
1535func (s *subscription) Changed() <-chan struct{} { return s.changed }
1536func (s *subscription) Close() { s.once.Do(s.close) }
1537func (s *subscription) offer() {
1538 select {
1539 case s.changed <- struct{}{}:
1540 default:
1541 }
1542}