1package broker
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "slices"
8 "sort"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12)
13
14// AddHost adds or pins endpoint under policy. Duplicate same-policy additions
15// CAS-verify authority and preserve identity; a conflicting policy is refused.
16func (r *Registry) LookupHost(ctx context.Context, endpoint string) (ports.BrokerHostRecord, bool, error) {
17 if err := ctx.Err(); err != nil {
18 return ports.BrokerHostRecord{}, false, err
19 }
20 r.mu.Lock()
21 defer r.mu.Unlock()
22 for _, record := range r.authority.Hosts {
23 if record.Registration.Endpoint == endpoint {
24 record.Route = record.Route.Clone()
25 return record, true, nil
26 }
27 }
28 return ports.BrokerHostRecord{}, false, nil
29}
30
31func (r *Registry) AddHost(ctx context.Context, endpoint string, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
32 if err := domain.ValidateRemoteHostTarget(endpoint); err != nil {
33 return domain.RemoteRegistration{}, err
34 }
35 if err := policy.Validate(); err != nil {
36 return domain.RemoteRegistration{}, err
37 }
38 r.mu.Lock()
39 defer r.mu.Unlock()
40 if err := r.mutableLocked(ctx); err != nil {
41 return domain.RemoteRegistration{}, err
42 }
43 records := append([]ports.BrokerHostRecord(nil), r.authority.Hosts...)
44 for i := range records {
45 if records[i].Registration.Endpoint != endpoint {
46 continue
47 }
48 if records[i].Policy != policy {
49 return domain.RemoteRegistration{}, ports.ErrBrokerHostConflict
50 }
51 records[i].Pinned = true
52 if err := r.commitMembershipLocked(records); err != nil {
53 return domain.RemoteRegistration{}, err
54 }
55 return records[i].Registration, nil
56 }
57 id, err := r.newIncarnation()
58 if err != nil {
59 return domain.RemoteRegistration{}, err
60 }
61 reg, err := domain.NewRemoteRegistration(endpoint, id)
62 if err != nil {
63 return domain.RemoteRegistration{}, err
64 }
65 if reg.IsZero() {
66 return domain.RemoteRegistration{}, errors.New("broker: incarnation generator returned zero")
67 }
68 route, err := ports.BrokerRouteForTransport(policy.Transport, endpoint)
69 if err != nil {
70 return domain.RemoteRegistration{}, err
71 }
72 records = append(records, ports.BrokerHostRecord{Registration: reg, Pinned: true, Policy: policy, Route: route})
73 if err := r.commitMembershipLocked(records); err != nil {
74 return domain.RemoteRegistration{}, err
75 }
76 return reg, nil
77}
78
79// RemoveHost removes only the exact expected registration.
80// BindAuthenticatedIdentity durably binds first-contact identity under the
81// exact registration and policy fence. The same binding is idempotent; a
82// different identity is refused and never overwrites authority.
83func (r *Registry) BindAuthenticatedIdentity(ctx context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
84 if err := request.Validate(); err != nil {
85 return "", err
86 }
87 if request.Fence.Local {
88 return request.Identity, nil
89 }
90 if err := ctx.Err(); err != nil {
91 return "", err
92 }
93 r.mu.Lock()
94 defer r.mu.Unlock()
95 records := append([]ports.BrokerHostRecord(nil), r.authority.Hosts...)
96 for index := range records {
97 record := &records[index]
98 if record.Identity == request.Identity && record.Policy != request.Policy {
99 return "", ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: errors.New("broker: authenticated daemon identity is already bound under an incompatible policy")}
100 }
101 }
102 for index := range records {
103 record := &records[index]
104 if !record.Registration.Equal(request.Fence.Registration) || record.Policy != request.Policy {
105 continue
106 }
107 if record.Identity != "" {
108 if record.Identity != request.Identity {
109 return "", ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: errors.New("broker: authenticated daemon identity conflicts with durable binding")}
110 }
111 return record.Identity, nil
112 }
113 record.Identity = request.Identity
114 if err := r.commitMembershipLocked(records); err != nil {
115 return "", err
116 }
117 return request.Identity, nil
118 }
119 return "", ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: errors.New("broker: stale identity binding fence")}
120}
121
122func (r *Registry) RemoveHost(ctx context.Context, expected domain.RemoteRegistration) (bool, error) {
123 if err := expected.Validate(); err != nil {
124 return false, err
125 }
126 r.mu.Lock()
127 defer r.mu.Unlock()
128 if err := r.mutableLocked(ctx); err != nil {
129 return false, err
130 }
131 records := append([]ports.BrokerHostRecord(nil), r.authority.Hosts...)
132 for i := range records {
133 if records[i].Registration.Endpoint != expected.Endpoint {
134 continue
135 }
136 if !records[i].Registration.Equal(expected) {
137 return false, ports.ErrBrokerHostConflict
138 }
139 records = append(records[:i], records[i+1:]...)
140 if err := r.commitMembershipLocked(records); err != nil {
141 return false, err
142 }
143 return true, nil
144 }
145 // A no-op still verifies the loaded revision.
146 return false, r.commitMembershipLocked(records)
147}
148
149// UpdateHostPolicy updates exact authority and advances its generation.
150func (r *Registry) UpdateHostPolicy(ctx context.Context, expected domain.RemoteRegistration, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
151 if err := expected.Validate(); err != nil {
152 return domain.RemoteRegistration{}, err
153 }
154 if err := policy.Validate(); err != nil {
155 return domain.RemoteRegistration{}, err
156 }
157 r.mu.Lock()
158 defer r.mu.Unlock()
159 if err := r.mutableLocked(ctx); err != nil {
160 return domain.RemoteRegistration{}, err
161 }
162 records := append([]ports.BrokerHostRecord(nil), r.authority.Hosts...)
163 for i := range records {
164 if records[i].Registration.Endpoint != expected.Endpoint {
165 continue
166 }
167 if !records[i].Registration.Equal(expected) {
168 return domain.RemoteRegistration{}, ports.ErrBrokerHostConflict
169 }
170 if records[i].Policy == policy {
171 if err := r.commitMembershipLocked(records); err != nil {
172 return domain.RemoteRegistration{}, err
173 }
174 return expected, nil
175 }
176 if expected.Generation == ^domain.RemoteGeneration(0) {
177 return domain.RemoteRegistration{}, ports.ErrBrokerRevisionExhausted
178 }
179 route, err := ports.BrokerRouteForTransport(policy.Transport, expected.Endpoint)
180 if err != nil {
181 return domain.RemoteRegistration{}, err
182 }
183 records[i].Registration.Generation++
184 records[i].Route = route
185 records[i].Policy = policy
186 if err := r.commitMembershipLocked(records); err != nil {
187 return domain.RemoteRegistration{}, err
188 }
189 return records[i].Registration, nil
190 }
191 return domain.RemoteRegistration{}, ports.ErrBrokerHostConflict
192}
193
194func (r *Registry) mutableLocked(ctx context.Context) error {
195 if r.membershipMode != MembershipMutable {
196 return ports.ErrBrokerMembershipImmutable
197 }
198 if err := ctx.Err(); err != nil {
199 return err
200 }
201 if r.running.Load() == runStopped {
202 return ports.ErrBrokerRegistryClosed
203 }
204 if r.revisionExhaustedLocked() {
205 return ports.ErrBrokerRevisionExhausted
206 }
207 return nil
208}
209
210// commitMembershipLocked is the sole mutable-membership durable commit path.
211// It installs projection only after CAS success.
212func (r *Registry) commitMembershipLocked(records []ports.BrokerHostRecord) error {
213 if err := ports.ValidateBrokerHostRecords(records); err != nil {
214 return err
215 }
216 if len(records) > 0 && !r.observationDisabled && nilDependency(r.probe) {
217 return errors.New("broker: remote hosts require a host probe")
218 }
219 next := make(map[string]ports.BrokerDaemonObservation, len(records))
220 order := make([]string, 0, len(records))
221 for rank, record := range records {
222 observation := stampHostAuthority(ports.BrokerDaemonObservation{}, record, rank)
223 // A membership the registry cannot publish is refused here, before the
224 // store CAS, rather than persisted and left to freeze every later
225 // publication: an endpoint whose derived display hint is empty after the
226 // display rule drops its runes makes every snapshot carrying it invalid.
227 if err := observation.Validate(); err != nil {
228 return fmt.Errorf("broker: host record %q is not publishable: %w", record.Registration.Endpoint, err)
229 }
230 next[record.Registration.Endpoint] = observation
231 order = append(order, record.Registration.Endpoint)
232 }
233 if err := r.hostStore.ReplaceHosts(r.authority.Revision, records); err != nil {
234 return fmt.Errorf("broker: replace hosts: %w", err)
235 }
236 r.authority = ports.BrokerHosts{Revision: r.authority.Revision + 1, Hosts: ports.CloneBrokerHostRecords(records)}
237 r.setHostsLocked(next, order)
238 return nil
239}
240
241// ReplaceHosts durably replaces membership and explicit policy using the loaded
242// authority revision. It delegates to commitMembershipLocked, the same
243// validated commit path the runtime mutations use, so membership whose
244// projection could never be published is refused before the store CAS instead
245// of being persisted and silently freezing every later publication. Success
246// means membership is committed, not that advisory observations have flushed.
247// Store errors (including stale authority) are returned without changing the
248// projection, probe attempts, or local token. A settled run reports
249// ports.ErrBrokerRegistryClosed and an exhausted revision series reports
250// ports.ErrBrokerRevisionExhausted; both refuse the mutation before the store
251// CAS, so a refusal never changes durable authority either. Conflicts are not
252// retried or refreshed implicitly: reconcile by reopening the owner. Callers
253// must supply fresh incarnations when removing and re-adding. Even identical
254// records perform CAS so stale authority cannot report success. The synchronous
255// write holds mu; Snapshot remains lock-free, but scheduling waits for
256// membership durability. Observation writes remain asynchronous.
257func (r *Registry) ReplaceHosts(records []ports.BrokerHostRecord) error {
258 var err error
259 records, err = ports.UpgradeBrokerHostRoutes(records)
260 if err != nil {
261 return err
262 }
263 if err := ports.ValidateBrokerHostRecords(records); err != nil {
264 return err
265 }
266 r.mu.Lock()
267 defer r.mu.Unlock()
268 if r.running.Load() == runStopped {
269 return fmt.Errorf("broker: membership replacement: %w", ports.ErrBrokerRegistryClosed)
270 }
271 if r.revisionExhaustedLocked() {
272 return fmt.Errorf("broker: membership replacement: %w", ports.ErrBrokerRevisionExhausted)
273 }
274 return r.commitMembershipLocked(records)
275}
276
277// setHostsLocked applies only a projection; the public path commits authority
278// first. The caller holds mu through both steps, fencing probe completions.
279// order is the registration order the projection was built from; it becomes
280// the publication order of remote daemons.
281func (r *Registry) setHostsLocked(next map[string]ports.BrokerDaemonObservation, order []string) {
282 if registrationsUnchanged(r.hosts, next, r.order, order) {
283 return
284 }
285 for endpoint, current := range r.hosts {
286 candidate, present := next[endpoint]
287 if present && current.Registration.Equal(candidate.Registration) && current.Policy == candidate.Policy {
288 // Unchanged authority: keep the live observation as is. Rank is the
289 // one authority field that follows membership order rather than the
290 // record, so it is re-stamped from the new projection.
291 kept := current
292 kept.Rank = candidate.Rank
293 next[endpoint] = kept
294 continue
295 }
296 // A retired registration owns no observation: cancelling its attempt
297 // releases a ctx-honoring probe immediately, and dropping its record
298 // means the retired completion can never clear or answer the
299 // replacement, and the replacement is never blocked behind it. The
300 // cancellation happens under the registry lock, so no replacement
301 // attempt can start before the retired context is cancelled.
302 r.cancelInflightLocked(endpoint)
303 delete(r.pending, endpoint)
304 if present {
305 // The endpoint stays live under a new registration, so its
306 // retirement is never published as a tombstone alongside a live
307 // host; the attempt token fences the stale completion.
308 delete(r.tombstones, endpoint)
309 continue
310 }
311 r.retireLocked(current.Registration)
312 }
313 for endpoint := range next {
314 // Re-adding an endpoint supersedes its tombstone: a live host and a
315 // tombstone for the same endpoint are mutually exclusive.
316 delete(r.tombstones, endpoint)
317 }
318 r.hosts = next
319 r.order = order
320 r.publishLocked(true)
321 r.hint()
322}
323
324// registrationsUnchanged reports exact membership equality: the same endpoints
325// in the same registration order, bound to the same registration identities and
326// policy. Publication order and presentation rank are functions of durable
327// membership order, so a reordering of identical records is a membership change
328// that must republish; otherwise the same durable state yields different
329// publications depending on process history. Projection fields such as
330// availability, checking, or inventory never participate.
331func registrationsUnchanged(current, next map[string]ports.BrokerDaemonObservation, currentOrder, nextOrder []string) bool {
332 if len(current) != len(next) {
333 return false
334 }
335 if !slices.Equal(currentOrder, nextOrder) {
336 return false
337 }
338 for endpoint, candidate := range next {
339 existing, ok := current[endpoint]
340 if !ok || !existing.Registration.Equal(candidate.Registration) || existing.Policy != candidate.Policy {
341 return false
342 }
343 }
344 return true
345}
346
347// retireLocked records the tombstone for one removed registration at the
348// revision of the publication that carries it. Mutation paths refuse an
349// exhausted series before retiring, so the retirement revision can never wrap
350// to zero or fall below the last published one.
351func (r *Registry) retireLocked(registration domain.RemoteRegistration) {
352 retired := r.revision + 1
353 r.tombstones[registration.Endpoint] = ports.BrokerHostTombstone{
354 Endpoint: registration.Endpoint,
355 Registration: registration,
356 RetiredRevision: retired,
357 }
358 r.pruneTombstonesLocked()
359}
360
361// pruneTombstonesLocked keeps only the newest BrokerMaxTombstones retirements
362// so an endpoint that churns can never grow the published set without bound.
363func (r *Registry) pruneTombstonesLocked() {
364 for len(r.tombstones) > ports.BrokerMaxTombstones {
365 oldest := ""
366 var oldestRevision ports.BrokerRevision
367 for endpoint, tombstone := range r.tombstones {
368 if oldest == "" || tombstone.RetiredRevision < oldestRevision ||
369 (tombstone.RetiredRevision == oldestRevision && endpoint < oldest) {
370 oldest, oldestRevision = endpoint, tombstone.RetiredRevision
371 }
372 }
373 delete(r.tombstones, oldest)
374 }
375}
376
377// tombstonesLocked renders the bounded tombstone publication, newest first.
378// Live endpoints are never published as tombstones.
379func (r *Registry) tombstonesLocked() []ports.BrokerHostTombstone {
380 if len(r.tombstones) == 0 {
381 return nil
382 }
383 out := make([]ports.BrokerHostTombstone, 0, len(r.tombstones))
384 for endpoint, tombstone := range r.tombstones {
385 if _, live := r.hosts[endpoint]; live {
386 continue
387 }
388 out = append(out, tombstone)
389 }
390 sort.Slice(out, func(i, j int) bool {
391 if out[i].RetiredRevision != out[j].RetiredRevision {
392 return out[i].RetiredRevision > out[j].RetiredRevision
393 }
394 return out[i].Endpoint < out[j].Endpoint
395 })
396 if len(out) > ports.BrokerMaxTombstones {
397 out = out[:ports.BrokerMaxTombstones]
398 }
399 return out
400}