1package broker
2
3import (
4 "context"
5 "crypto/rand"
6 "errors"
7 "fmt"
8 "log/slog"
9 "reflect"
10 "sync"
11 "sync/atomic"
12 "time"
13 "unicode/utf8"
14
15 "github.com/bnema/vev/internal/domain"
16 "github.com/bnema/vev/internal/ports"
17)
18
19// Observation cadence. No client means no scheduled observations; the
20// fifteen-second interval only applies to an absent daemon under demand.
21const (
22 defaultFreshFor = 15 * time.Second
23 defaultRetryBase = 5 * time.Second
24 defaultRetryLimit = time.Minute
25 // defaultDemandFreshForLocal and defaultDemandFreshForRemote are the
26 // built-in demand cadences RegistryConfig falls back to when
27 // DemandFreshForLocal/DemandFreshForRemote is left zero. They apply only
28 // while at least one client subscription is live (see Registry.SetDemand).
29 defaultDemandFreshForLocal = 1 * time.Second
30 defaultDemandFreshForRemote = 2 * time.Second
31 // demandRetryMax caps the failure backoff while a client watches: retries
32 // start at the demand cadence and double up to this bound, so an
33 // unreachable or refusing host is not redialed every two seconds.
34 demandRetryMax = 30 * time.Second
35 // defaultProbeTimeout bounds one observation attempt end to end: dial,
36 // SSH/QUIC bootstrap, daemonmux preamble, and the catalogue exchange. It
37 // sits below the passive retry cap so a peer that never answers costs one
38 // bounded attempt instead of pinning the endpoint's single in-flight slot.
39 defaultProbeTimeout = 20 * time.Second
40)
41
42// errProbeTimeout is the failure an attempt reports when its ports.Clock
43// deadline fires before the probe returns. It wraps context.DeadlineExceeded
44// so observationOutcome classifies it as a typed timeout.
45var errProbeTimeout = fmt.Errorf("broker: observation exceeded its deadline: %w", context.DeadlineExceeded)
46
47// Run lifecycle states. A registry runs exactly once: runIdle is the zero
48// value, runActive marks the single admitted run, and runStopped permanently
49// refuses further runs.
50const (
51 runIdle int32 = iota
52 runActive
53 runStopped
54)
55
56// MembershipMode controls whether runtime membership mutation is admitted.
57type MembershipMode uint8
58
59const (
60 MembershipImmutable MembershipMode = iota
61 MembershipMutable
62)
63
64// RegistryConfig selects how a Registry runs. The zero value observes hosts
65// with immutable membership.
66type RegistryConfig struct {
67 MembershipMode MembershipMode
68 // IncarnationGenerator is an injectable entropy seam for mutable additions.
69 // Nil uses crypto/rand.
70 IncarnationGenerator func() ([16]byte, error)
71 // ObservationDisabled runs the registry as a read-only snapshot owner: it
72 // restores and publishes durable state, owns and drains its writer, and
73 // honors the Run lifecycle, but issues no probes, arms no timers, and never
74 // reconciles. The probe dependency is unused and may be nil. Membership is
75 // expected to be immutable for the run; a caller that needs observation
76 // keeps the zero-value config.
77 ObservationDisabled bool
78 // Local, when non-nil, enables observation of the broker's own machine
79 // daemon. The registry then publishes the local entry at index zero, ahead
80 // of every remote, and never persists it. Local is refused together with
81 // ObservationDisabled: a read-only registry owns no local producer.
82 Local *LocalObservation
83 // DemandFreshForLocal and DemandFreshForRemote are the re-probe cadences
84 // apply/applyLocal schedule into NextDue while Registry.SetDemand reports
85 // at least one live client subscription. A zero value falls back to
86 // defaultDemandFreshForLocal (about 1s) or defaultDemandFreshForRemote
87 // (about 2s). With no live subscription, no scheduled probe runs.
88 // DemandFreshForRemote is also the first delay of a watched host's
89 // failure backoff, which doubles up to max(DemandFreshForRemote, 30s).
90 DemandFreshForLocal time.Duration
91 DemandFreshForRemote time.Duration
92 // ProbeTimeout bounds every local and remote observation attempt on the
93 // registry's ports.Clock. When it fires the attempt is released and
94 // recorded as a timeout failure even if the probe never returns. A zero
95 // value selects defaultProbeTimeout.
96 ProbeTimeout time.Duration
97}
98
99// Registry owns configured hosts and their immutable observation projection.
100type Registry struct {
101 epoch ports.BrokerEpoch
102 probe ports.BrokerHostProbe
103 clock ports.Clock
104 log *slog.Logger
105 store *snapshotWriter
106 // observationDisabled is fixed at construction: a disabled registry never
107 // observes, so it has no probe, timer, or reconcile work to do.
108 observationDisabled bool
109 membershipMode MembershipMode
110 newIncarnation func() ([16]byte, error)
111
112 // hostStore and authority are the synchronous membership owner. Only this
113 // registry may mutate the store while it is live; the caller owns Close.
114 hostStore ports.BrokerHostStore
115 authority ports.BrokerHosts // guarded by mu; independent of publication revision
116
117 current atomic.Pointer[ports.BrokerSnapshot]
118 mu sync.Mutex
119 revision ports.BrokerRevision
120 // revisionExhausted records that this epoch consumed its entire revision
121 // series. The registry then fails closed: the last valid snapshot stays
122 // published, nothing is written to the store, and every mutation path
123 // reports ports.ErrBrokerRevisionExhausted instead of wrapping the series to
124 // zero or below the newest revision.
125 revisionExhausted bool
126 // attempts is the registry-wide monotonic attempt token source. Every
127 // started observation claims one token, and only the token currently
128 // recorded as in flight may complete. Tokens are never reused, so a stale
129 // completion is fenced even when the endpoint is removed and re-added with
130 // the identical registration.
131 attempts uint64
132 // hosts is the broker-native projection per endpoint. Configured
133 // authority (endpoint, registration, policy, display origin, rank) is
134 // stamped here from durable membership; probe results only supply observed
135 // state and can never override it.
136 hosts map[string]ports.BrokerDaemonObservation
137 // order is the publication order of remote hosts: registration order from
138 // durable membership. A snapshot carries the local daemon first (when a
139 // producer exists) and then these remotes in this order.
140 order []string
141 // inflight records the one admitted attempt per endpoint. The record
142 // carries the attempt's derived context cancel so a retired attempt is
143 // released the moment its registration is replaced, removed, or the run
144 // settles; a ctx-honoring probe therefore stops promptly and churn can
145 // never accumulate overlapping probes for one endpoint.
146 inflight map[string]*probeAttempt
147 pending map[string]bool
148 tombstones map[string]ports.BrokerHostTombstone
149 subs map[*subscription]struct{}
150 wake chan struct{}
151 running atomic.Int32
152
153 // local selects observation of the broker's own machine daemon. It is nil
154 // for a remote-only registry. localHost is the configured local authority
155 // stamped onto every publication (Local, DisplayOrigin, Policy, Rank) plus
156 // the observed state a probe supplies; localAttempt is the one admitted
157 // local probe. The local entry is never durable and is never invented: until
158 // a probe answers, the configured authority is published with unknown
159 // availability and zero observed identity.
160 local *LocalObservation
161 localHost ports.BrokerDaemonObservation
162 localAttempt *probeAttempt
163 // localPending marks explicit demand for a local re-probe ahead of its
164 // scheduled freshness window: RequestProbe("") and a completed local control
165 // stream or attach both set it. dispatchLocalLocked honors it the same way
166 // pending honors remote demand, admitting the attempt before NextDue and
167 // clearing the flag the moment the attempt starts.
168 localPending bool
169
170 freshFor time.Duration
171 retryBase time.Duration
172 retryMax time.Duration
173 jitter func(base time.Duration, endpoint string, attempt uint64) time.Duration
174
175 // demandFreshForLocal and demandFreshForRemote are the resolved (non-zero)
176 // demand cadences; see RegistryConfig.DemandFreshForLocal/Remote.
177 demandFreshForLocal time.Duration
178 demandFreshForRemote time.Duration
179 // probeTimeout is the resolved (non-zero) per-attempt deadline; see
180 // RegistryConfig.ProbeTimeout.
181 probeTimeout time.Duration
182 // demand counts live client subscriptions: Registry.SetDemand(true) is one
183 // subscription opening and SetDemand(false) is one closing. Guarded by mu.
184 // While demand > 0, apply and applyLocal schedule subsequent probes.
185 demand int
186}
187
188// NewRegistryWithConfig restores a validated durable snapshot under a fresh
189// broker epoch and selects the run mode from cfg. Observation-disabled mode
190// tolerates a nil probe because such a registry never probes, and a registry
191// configured with a local observation producer (Local) tolerates a nil remote
192// probe as long as it owns no remote membership. Every other dependency is
193// required exactly as NewRegistry requires it.
194func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore, probe ports.BrokerHostProbe, clock ports.Clock, log *slog.Logger, cfg RegistryConfig) (*Registry, error) {
195 if cfg.MembershipMode != MembershipImmutable && cfg.MembershipMode != MembershipMutable {
196 return nil, errors.New("broker: invalid membership mode")
197 }
198 if epoch == 0 || nilDependency(store) || nilDependency(clock) {
199 return nil, errors.New("broker: invalid registry dependencies")
200 }
201 // A remote probe is required to observe remotes, but a registry that
202 // observes only its own machine daemon (Local) may omit it. Such a registry
203 // never owns a remote host: projectMembership and the mutation paths refuse
204 // remote membership rather than silently never probing it.
205 if !cfg.ObservationDisabled && nilDependency(probe) && cfg.Local == nil {
206 return nil, errors.New("broker: invalid registry dependencies")
207 }
208 if log == nil {
209 log = slog.Default()
210 }
211 generator := cfg.IncarnationGenerator
212 if generator == nil {
213 generator = func() (id [16]byte, err error) { _, err = rand.Read(id[:]); return id, err }
214 }
215 demandFreshForLocal := cfg.DemandFreshForLocal
216 if demandFreshForLocal <= 0 {
217 demandFreshForLocal = defaultDemandFreshForLocal
218 }
219 demandFreshForRemote := cfg.DemandFreshForRemote
220 if demandFreshForRemote <= 0 {
221 demandFreshForRemote = defaultDemandFreshForRemote
222 }
223 probeTimeout := cfg.ProbeTimeout
224 if probeTimeout <= 0 {
225 probeTimeout = defaultProbeTimeout
226 }
227 r := &Registry{
228 epoch: epoch, probe: probe, clock: clock, log: log,
229 observationDisabled: cfg.ObservationDisabled,
230 membershipMode: cfg.MembershipMode, newIncarnation: generator,
231 hosts: make(map[string]ports.BrokerDaemonObservation), inflight: make(map[string]*probeAttempt),
232 pending: make(map[string]bool), tombstones: make(map[string]ports.BrokerHostTombstone),
233 subs: make(map[*subscription]struct{}), wake: make(chan struct{}, 1),
234 freshFor: defaultFreshFor, retryBase: defaultRetryBase, retryMax: defaultRetryLimit,
235 jitter: jittered,
236 demandFreshForLocal: demandFreshForLocal,
237 demandFreshForRemote: demandFreshForRemote,
238 probeTimeout: probeTimeout,
239 }
240 if cfg.ObservationDisabled && cfg.Local != nil {
241 return nil, errors.New("broker: observation-disabled registry cannot observe the local daemon")
242 }
243 if cfg.Local != nil {
244 if err := cfg.Local.validate(); err != nil {
245 return nil, err
246 }
247 // Copy the configured authority: the registry reads it from the probe
248 // and publish paths, and a caller that kept the pointer could otherwise
249 // mutate authority after construction.
250 local := *cfg.Local
251 r.local = &local
252 r.localHost = ports.BrokerDaemonObservation{Local: true, DisplayOrigin: cfg.Local.DisplayOrigin, Policy: cfg.Local.Policy, Availability: domain.RemoteAvailabilityUnknown}
253 if err := r.localHost.Validate(); err != nil {
254 return nil, fmt.Errorf("broker: invalid local observation authority: %w", err)
255 }
256 }
257 r.store = newSnapshotWriter(store, log)
258 snapshot, err := store.Load()
259 if err != nil {
260 return nil, err
261 }
262 if err := r.restore(snapshot); err != nil {
263 return nil, err
264 }
265 hosts, err := store.LoadHosts()
266 if err != nil {
267 return nil, err
268 }
269 hosts.Hosts, err = ports.UpgradeBrokerHostRoutes(hosts.Hosts)
270 if err != nil {
271 return nil, err
272 }
273 if err := r.projectMembership(hosts); err != nil {
274 return nil, err
275 }
276 r.hostStore = store
277 r.authority = ports.BrokerHosts{Revision: hosts.Revision, Hosts: ports.CloneBrokerHostRecords(hosts.Hosts)}
278 r.publishLocked(false)
279 return r, nil
280}
281
282// nilDependency reports whether a required dependency is absent, including a
283// typed nil pointer stored behind a non-nil interface. Comparing the interface
284// to nil alone would admit a typed nil and defer the failure to the first
285// method call, which panics; the reflection check keeps the dependency guard
286// total for every nil shape.
287func nilDependency(dependency any) bool {
288 if dependency == nil {
289 return true
290 }
291 value := reflect.ValueOf(dependency)
292 switch value.Kind() {
293 case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice:
294 return value.IsNil()
295 default:
296 return false
297 }
298}
299
300// projectMembership intersects durable membership with the observations a
301// previous process left behind. Durable membership is authoritative:
302// snapshots cannot resurrect removed hosts or omit never-observed
303// registrations. A restored observation is reused only when it belongs to the
304// exact same registration identity; any other registration starts unknown and
305// carries no inventory.
306func (r *Registry) projectMembership(hosts ports.BrokerHosts) error {
307 if err := hosts.Validate(); err != nil {
308 return err
309 }
310 if len(hosts.Hosts) > 0 && !r.observationDisabled && nilDependency(r.probe) {
311 return errors.New("broker: remote hosts require a host probe")
312 }
313 next := make(map[string]ports.BrokerDaemonObservation, len(hosts.Hosts))
314 order := make([]string, 0, len(hosts.Hosts))
315 for rank, record := range hosts.Hosts {
316 registration := record.Registration
317 restored, present := r.hosts[registration.Endpoint]
318 if !present || !restored.Registration.Equal(registration) {
319 restored = ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnknown}
320 }
321 projection := stampHostAuthority(restored, record, rank)
322 // A durable membership whose projection cannot be published is refused
323 // here, at the boundary that owns authority, instead of silently
324 // leaving the registry unable to publish anything.
325 if err := projection.Validate(); err != nil {
326 return fmt.Errorf("broker: host record %q is not publishable: %w", registration.Endpoint, err)
327 }
328 next[registration.Endpoint] = projection
329 order = append(order, registration.Endpoint)
330 }
331 r.hosts = next
332 r.order = order
333 return nil
334}
335
336// stampHostAuthority overwrites the configured authority fields of a daemon
337// projection with durable membership authority. Membership is the single
338// authority for endpoint, registration, policy, and the presentation hints
339// derived from them; a probe, a restored durable observation, or any other
340// producer can never supply or override them. The durable observation format
341// deliberately carries no policy (see durable), so this stamp is also what
342// gives every restored observation a valid policy.
343func stampHostAuthority(observation ports.BrokerDaemonObservation, record ports.BrokerHostRecord, rank int) ports.BrokerDaemonObservation {
344 observation.Local = false
345 observation.Endpoint = record.Registration.Endpoint
346 observation.DisplayOrigin = hostDisplayOrigin(record.Registration.Endpoint)
347 observation.Rank = rank
348 observation.Registration = record.Registration
349 observation.Policy = record.Policy
350 if observation.Availability == 0 {
351 // An unobserved daemon reports the explicit unknown availability; zero
352 // is never a valid publication.
353 observation.Availability = domain.RemoteAvailabilityUnknown
354 }
355 return observation
356}
357
358// stampObservationAuthority copies the configured authority already stamped on
359// a live projection onto a fresh probe result. Probe results supply only
360// observed state, so their endpoint, registration, policy, display origin, and
361// rank fields are discarded wholesale.
362func stampObservationAuthority(observation, authority ports.BrokerDaemonObservation) ports.BrokerDaemonObservation {
363 observation.Local = false
364 observation.Endpoint = authority.Endpoint
365 observation.DisplayOrigin = authority.DisplayOrigin
366 observation.Rank = authority.Rank
367 observation.Registration = authority.Registration
368 observation.Policy = authority.Policy
369 return observation
370}
371
372// hostDisplayOrigin derives the presentation hint for a configured host. The
373// routing endpoint stays authoritative; the origin is only the login prefix
374// stripped for display, and an endpoint that carries none is its own origin.
375// The hint is sanitized to the display-text rule before it is clamped rune- and
376// byte-safely to the ports bound: an endpoint the routing validator accepts may
377// still carry bidi or control runes, and rendering those would let a derived
378// hint reorder the picker text around it. A derived hint carries no routing
379// authority, so the refused runes are dropped rather than the host rejected. A
380// target left with no displayable rune at all yields no origin, which its
381// caller refuses as unpublishable membership instead of inventing display text.
382func hostDisplayOrigin(endpoint string) string {
383 origin := ports.SanitizeBrokerDisplayText(domain.RemoteDisplayOrigin(endpoint))
384 if len(origin) <= ports.BrokerMaxDisplayOriginBytes {
385 return origin
386 }
387 truncated := origin[:ports.BrokerMaxDisplayOriginBytes]
388 for len(truncated) > 0 && !utf8.ValidString(truncated) {
389 truncated = truncated[:len(truncated)-1]
390 }
391 return truncated
392}
393
394// restore adopts a validated durable snapshot. A zero-epoch snapshot is only
395// the empty placeholder a store returns before its first write: any record
396// without an epoch is rejected instead of silently dropped. A persisted
397// snapshot from the registry's own epoch is rejected as well: every broker
398// process samples a fresh epoch and restarts its revision series at one, so
399// adopting same-epoch revisions would make the restored series non-monotonic
400// (and a stale same-epoch record indistinguishable from current authority).
401// In-flight observation state and retirement tombstones are per-process
402// fencing state and never survive a restart.
403func (r *Registry) restore(snapshot ports.BrokerSnapshot) error {
404 if snapshot.Epoch == 0 {
405 if len(snapshot.Daemons) > 0 || len(snapshot.Removed) > 0 || snapshot.Revision != 0 {
406 return errors.New("broker: persisted snapshot has no epoch")
407 }
408 return nil
409 }
410 if snapshot.Epoch == r.epoch {
411 return errors.New("broker: persisted snapshot epoch matches the new registry epoch")
412 }
413 if err := ports.ValidateDurableSnapshot(snapshot); err != nil {
414 return err
415 }
416 for _, daemon := range snapshot.Daemons {
417 daemon = daemon.Clone()
418 daemon.Checking = false
419 r.hosts[daemon.Endpoint] = daemon
420 }
421 return nil
422}