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 with a visible Checking state, so dispatch publishes it
85// exactly once; a routine refresh of a reachable daemon stays silent.
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 explicit := r.localPending
91 if !explicit && (r.demand == 0 || !r.localHost.NextDue.IsZero() && now.Before(r.localHost.NextDue)) {
92 return false
93 }
94 r.localPending = false
95 r.attempts++
96 attempt := r.attempts
97 probeCtx, cancel := context.WithCancel(ctx)
98 r.localAttempt = &probeAttempt{token: attempt, cancel: cancel}
99 visible := visibleCheck(r.localHost, explicit)
100 r.localHost.Checking = visible
101 r.localHost.LastAttempt = now
102 deadline := r.clock.NewTimer(r.probeTimeout)
103 go func(attempt uint64, probeCtx context.Context) {
104 snapshot, err := observeBounded(probeCtx, deadline, r.local.Probe.ProbeLocal)
105 result := probeResult{local: true, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}
106 select {
107 case results <- result:
108 case <-probeCtx.Done():
109 // The attempt was retired (run settled), so its completion is
110 // already fenced: dropping it keeps a settled registry from
111 // queueing a result the run loop would only discard.
112 }
113 }(attempt, probeCtx)
114 return visible
115}
116
117// applyLocal adopts one completed local probe. Callers must hold r.mu, exactly
118// as Registry.apply requires. A stale attempt touches nothing. A confirmed
119// reachable result publishes the observed state with configured authority
120// stamped on top. An explicit unknown result with no error - the local daemon
121// has not published its identity yet, so observing it is not even attempted -
122// publishes the configured authority with the unknown availability and zero
123// observed identity, and reschedules the next attempt without recording a
124// failure. Every other outcome publishes a typed failure and keeps the
125// last-known identity and inventory, so an unreachable local daemon never
126// reports an invented identity and its availability is never zero.
127func (r *Registry) applyLocal(result probeResult) {
128 if r.local == nil {
129 return
130 }
131 attempt := r.localAttempt
132 if attempt == nil || attempt.token != result.attempt {
133 return
134 }
135 r.localAttempt = nil
136 // The probe has returned, so releasing its derived context here only frees
137 // the attempt's resources; the record is gone before any new attempt can
138 // start.
139 attempt.cancel()
140
141 current := r.localHost
142 current.Checking = false
143 current.LastAttempt = result.at
144
145 // An explicit unknown result with no error means the local daemon has not
146 // published its identity yet, so observing it was not even attempted. That
147 // is not a failure: publish the configured authority with the explicit
148 // unknown availability, keep zero observed identity, and reschedule on the
149 // retry cadence so the first identity is observed promptly instead of
150 // waiting out a healthy refresh interval.
151 if result.err == nil && result.snapshot.Availability == domain.RemoteAvailabilityUnknown {
152 current.Availability = domain.RemoteAvailabilityUnknown
153 current.LastFailure = domain.RemoteFailure{}
154 current.NextDue = result.at.Add(r.jitter(r.retryDelay(1), localProbeKey, result.attempt))
155 r.localHost = current
156 r.publishLocked(true)
157 return
158 }
159
160 kind, availability, failed := observationOutcome(result)
161 if !failed {
162 observed := stampLocalAuthority(result.snapshot.Clone(), current)
163 observed.Checking = false
164 observed.LastAttempt = result.at
165 observed.LastSuccess = result.at
166 observed.NextDue = result.at.Add(r.jitter(r.freshForLocked(true), localProbeKey, result.attempt))
167 observed.ConsecutiveFailures = 0
168 observed.LastFailure = domain.RemoteFailure{}
169 observed.FailureEpisode = current.FailureEpisode
170 // The published local observation must be fully valid once authority is
171 // stamped. A probe that invents partial identity or an invalid session
172 // catalogue is an invalid response, never a published observation.
173 if err := observed.Validate(); err != nil {
174 r.log.Debug("broker: local observation is invalid", "err", err)
175 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
176 } else if err := validateLocalObservation(observed); err != nil {
177 r.log.Debug("broker: local observation is not publishable", "err", err)
178 kind, availability, failed = domain.RemoteFailureInvalidResponse, domain.RemoteAvailabilityInvalidResponse, true
179 } else {
180 r.localHost = observed
181 r.publishLocked(true)
182 return
183 }
184 }
185 // Every non-reachable outcome is a typed failure on the capped retry
186 // cadence: the observed identity, last-known inventory, and failure episode
187 // counter stay authoritative.
188 current.Availability = availability
189 if current.ConsecutiveFailures == 0 {
190 current.FailureEpisode++
191 }
192 current.ConsecutiveFailures++
193 current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
194 current.NextDue = result.at.Add(r.jitter(r.retryDelay(current.ConsecutiveFailures), localProbeKey, result.attempt))
195 r.localHost = current
196 r.publishLocked(true)
197}
198
199// stampLocalAuthority copies the configured local authority onto a fresh probe
200// result. Probe results supply observed state only, so their local flag,
201// endpoint, registration, display origin, policy, and rank fields are discarded
202// wholesale: a hostile probe can never promote itself to local authority or
203// borrow a remote endpoint.
204func stampLocalAuthority(observation, authority ports.BrokerDaemonObservation) ports.BrokerDaemonObservation {
205 observation.Local = true
206 observation.Endpoint = ""
207 observation.Registration = domain.RemoteRegistration{}
208 observation.DisplayOrigin = authority.DisplayOrigin
209 observation.Policy = authority.Policy
210 observation.Rank = authority.Rank
211 return observation
212}
213
214// validateLocalObservation re-checks the catalogue shape of a local session
215// inventory. The durable projection rule cannot apply to a local observation
216// (the local daemon is never durable and has no registration incarnation), so
217// the catalogue itself is validated directly against the current schema.
218func validateLocalObservation(observation ports.BrokerDaemonObservation) error {
219 if len(observation.Sessions) == 0 {
220 return nil
221 }
222 return catalogue.ValidateRemoteCatalog(catalogue.RemoteCatalog{
223 ProtocolVersion: protocol.Version,
224 SchemaVersion: catalogue.RemoteCatalogSchemaVersion,
225 Sessions: observation.Sessions,
226 })
227}