1package broker
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "time"
8
9 "github.com/bnema/vev/internal/domain"
10 "github.com/bnema/vev/internal/ports"
11 "github.com/bnema/vev/internal/protocol"
12 "github.com/bnema/vev/internal/protocol/catalogue"
13)
14
15// Broker-owned local observation.
16//
17// The broker's own machine daemon is not a configured host: it has no
18// endpoint, no registration, and no durable membership. It is observed through
19// the same probe lifecycle as a remote host, but its configured authority
20// (display origin and policy) is supplied by the composition and its entry is
21// always published first, at index zero, and never persisted. Observing the
22// local daemon only reads: the probe dials the broker-owned local route, and
23// the shared observation implementation behind it opens exactly one
24// BrokerStreamObservation stream, asks for the daemon's own catalogue through
25// the existing typed control protocol, and validates the answer. It therefore
26// never attaches, never claims a geometry, never requests an admission, never
27// creates a session or process, and never starts a stopped daemon.
28
29// localProbeKey seeds the deterministic jitter for the local probe schedule.
30// It is a fixed token, so local retry and refresh jitter is reproducible and
31// independent of the configured remotes. A remote endpoint spelled "local"
32// shares the seed, which affects only the jitter spread and never identity.
33const localProbeKey = "local"
34
35// LocalProbe observes the broker's own machine daemon without attaching to it
36// or starting a stopped daemon. It reports only observed state: the observed
37// daemon identity, incarnation, protocol version, capabilities, availability,
38// failure, freshness, and session inventory. Configured authority (the local
39// flag, display origin, policy, and rank) is stamped by the registry and never
40// trusted from the probe; unknown identity, version, or inventory is zero and
41// is never invented. The returned observation must satisfy
42// ports.BrokerDaemonObservation.Validate once the registry stamps authority.
43//
44// ProbeLocal must honor cancellation: the registry cancels the attempt context
45// when the run settles, and a probe that ignores cancellation delays shutdown.
46type LocalProbe interface {
47 ProbeLocal(ctx context.Context) (ports.BrokerDaemonObservation, error)
48}
49
50// LocalObservation configures observation of the broker's own machine daemon.
51// The display origin and policy are configured authority: the registry stamps
52// them onto every local publication and never reads them from a probe. The
53// probe is the sole observed-state source.
54type LocalObservation struct {
55 // DisplayOrigin is the presentation hint published for the local entry. It
56 // is a display value only and never routing authority.
57 DisplayOrigin string
58 // Policy is the exact connection policy of the local binding.
59 Policy ports.BrokerPolicy
60 // Probe observes the local daemon. It is required.
61 Probe LocalProbe
62}
63
64// validate refuses a local observation the registry could not publish or fence.
65func (l *LocalObservation) validate() error {
66 if l == nil {
67 return nil
68 }
69 if nilDependency(l.Probe) {
70 return errors.New("broker: local observation has no probe")
71 }
72 if err := l.Policy.Validate(); err != nil {
73 return fmt.Errorf("broker: local observation authority: %w", err)
74 }
75 if l.DisplayOrigin == "" || ports.SanitizeBrokerDisplayText(l.DisplayOrigin) != l.DisplayOrigin {
76 return errors.New("broker: local observation has an invalid display origin")
77 }
78 return nil
79}
80
81// dispatchLocalLocked admits at most one local probe attempt when its schedule
82// says it is due, or immediately when localPending marks explicit demand for a
83// re-probe ahead of schedule. Callers must hold r.mu. It reports whether an
84// attempt started, so dispatch publishes the transient Checking state exactly
85// once.
86func (r *Registry) dispatchLocalLocked(ctx context.Context, now time.Time, results chan<- probeResult) bool {
87 if r.local == nil || r.localAttempt != nil {
88 return false
89 }
90 if !r.localPending && (r.demand == 0 || !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue)) {
91 return false
92 }
93 r.localPending = false
94 r.attempts++
95 attempt := r.attempts
96 probeCtx, cancel := context.WithCancel(ctx)
97 r.localAttempt = &probeAttempt{token: attempt, cancel: cancel}
98 r.localHost.Checking = true
99 r.localHost.LastAttempt = now
100 deadline := r.clock.NewTimer(r.probeTimeout)
101 go func(attempt uint64, probeCtx context.Context) {
102 snapshot, err := observeBounded(probeCtx, deadline, r.local.Probe.ProbeLocal)
103 result := probeResult{local: true, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}
104 select {
105 case results <- result:
106 case <-probeCtx.Done():
107 // The attempt was retired (run settled), so its completion is
108 // already fenced: dropping it keeps a settled registry from
109 // queueing a result the run loop would only discard.
110 }
111 }(attempt, probeCtx)
112 return true
113}
114
115// applyLocal adopts one completed local probe. Callers must hold r.mu, exactly
116// as Registry.apply requires. A stale attempt touches nothing. A confirmed
117// reachable result publishes the observed state with configured authority
118// stamped on top. An explicit unknown result with no error - the local daemon
119// has not published its identity yet, so observing it is not even attempted -
120// publishes the configured authority with the unknown availability and zero
121// observed identity, and reschedules the next attempt without recording a
122// failure. Every other outcome publishes a typed failure and keeps the
123// last-known identity and inventory, so an unreachable local daemon never
124// reports an invented identity and its availability is never zero.
125func (r *Registry) applyLocal(result probeResult) {
126 if r.local == nil {
127 return
128 }
129 attempt := r.localAttempt
130 if attempt == nil || attempt.token != result.attempt {
131 return
132 }
133 r.localAttempt = nil
134 // The probe has returned, so releasing its derived context here only frees
135 // the attempt's resources; the record is gone before any new attempt can
136 // start.
137 attempt.cancel()
138
139 current := r.localHost
140 current.Checking = false
141 current.LastAttempt = result.at
142
143 // An explicit unknown result with no error means the local daemon has not
144 // published its identity yet, so observing it was not even attempted. That
145 // is not a failure: publish the configured authority with the explicit
146 // unknown availability, keep zero observed identity, and reschedule on the
147 // retry cadence so the first identity is observed promptly instead of
148 // waiting out a healthy refresh interval.
149 if result.err == nil && result.snapshot.Availability == domain.RemoteAvailabilityUnknown {
150 current.Availability = domain.RemoteAvailabilityUnknown
151 current.LastFailure = domain.RemoteFailure{}
152 current.NextDue = result.at.Add(r.jitter(r.retryDelay(1), localProbeKey, result.attempt))
153 r.localHost = current
154 r.publishLocked(true)
155 return
156 }
157
158 kind, availability, failed := observationOutcome(result)
159 if !failed {
160 observed := stampLocalAuthority(result.snapshot.Clone(), current)
161 observed.Checking = false
162 observed.LastAttempt = result.at
163 observed.LastSuccess = result.at
164 observed.NextDue = result.at.Add(r.jitter(r.freshForLocked(true), localProbeKey, result.attempt))
165 observed.ConsecutiveFailures = 0
166 observed.LastFailure = domain.RemoteFailure{}
167 observed.FailureEpisode = current.FailureEpisode
168 // The published local observation must be fully valid once authority is
169 // stamped. A probe that invents partial identity or an invalid session
170 // catalogue is an invalid response, never a published observation.
171 if err := observed.Validate(); err != nil {
172 r.log.Debug("broker: local observation is invalid", "err", err)
173 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
174 } else if err := validateLocalObservation(observed); err != nil {
175 r.log.Debug("broker: local observation is not publishable", "err", err)
176 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
177 } else {
178 r.localHost = observed
179 r.publishLocked(true)
180 return
181 }
182 }
183 // Every non-reachable outcome is a typed failure on the capped retry
184 // cadence: the observed identity, last-known inventory, and failure episode
185 // counter stay authoritative.
186 current.Availability = availability
187 if current.ConsecutiveFailures == 0 {
188 current.FailureEpisode++
189 }
190 current.ConsecutiveFailures++
191 current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
192 current.NextDue = result.at.Add(r.jitter(r.retryDelay(current.ConsecutiveFailures), localProbeKey, result.attempt))
193 r.localHost = current
194 r.publishLocked(true)
195}
196
197// stampLocalAuthority copies the configured local authority onto a fresh probe
198// result. Probe results supply observed state only, so their local flag,
199// endpoint, registration, display origin, policy, and rank fields are discarded
200// wholesale: a hostile probe can never promote itself to local authority or
201// borrow a remote endpoint.
202func stampLocalAuthority(observation, authority ports.BrokerDaemonObservation) ports.BrokerDaemonObservation {
203 observation.Local = true
204 observation.Endpoint = ""
205 observation.Registration = domain.RemoteRegistration{}
206 observation.DisplayOrigin = authority.DisplayOrigin
207 observation.Policy = authority.Policy
208 observation.Rank = authority.Rank
209 return observation
210}
211
212// validateLocalObservation re-checks the catalogue shape of a local session
213// inventory. The durable projection rule cannot apply to a local observation
214// (the local daemon is never durable and has no registration incarnation), so
215// the catalogue itself is validated directly against the current schema.
216func validateLocalObservation(observation ports.BrokerDaemonObservation) error {
217 if len(observation.Sessions) == 0 {
218 return nil
219 }
220 return catalogue.ValidateRemoteCatalog(catalogue.RemoteCatalog{
221 ProtocolVersion: protocol.Version,
222 SchemaVersion: catalogue.RemoteCatalogSchemaVersion,
223 Sessions: observation.Sessions,
224 })
225}