1package broker
2
3import (
4 "sync"
5
6 "github.com/bnema/vev/internal/ports"
7)
8
9// Snapshot returns a defensive immutable copy without I/O.
10func (r *Registry) Snapshot() ports.BrokerSnapshot { return r.current.Load().Clone() }
11
12// Subscribe returns a capacity-one notification stream. Slow consumers never
13// block publication and always re-read the newest snapshot.
14func (r *Registry) Subscribe() ports.BrokerSubscription {
15 s := &subscription{changed: make(chan struct{}, 1)}
16 s.close = func() {
17 r.mu.Lock()
18 delete(r.subs, s)
19 r.mu.Unlock()
20 }
21 r.mu.Lock()
22 r.subs[s] = struct{}{}
23 r.mu.Unlock()
24 s.offer()
25 return s
26}
27
28// revisionExhaustedLocked reports whether this epoch has no revision left to
29// publish, latching the exhausted state on first detection. Callers must hold
30// r.mu.
31func (r *Registry) revisionExhaustedLocked() bool {
32 if r.revisionExhausted {
33 return true
34 }
35 if r.revision != ^ports.BrokerRevision(0) {
36 return false
37 }
38 r.revisionExhausted = true
39 r.log.Error("broker: revision series exhausted; refusing further publication", "epoch", r.epoch)
40 return true
41}
42
43func (r *Registry) publishLocked(persist bool) {
44 // Fail closed on an exhausted series: a wrapped revision would be zero, and
45 // a wrapped series would publish revisions below the newest one, either of
46 // which silently freezes durable persistence. The last valid snapshot stays
47 // published, nothing is stored, and no subscriber is woken for a publication
48 // that did not happen.
49 if r.revisionExhaustedLocked() {
50 return
51 }
52 // Remote daemons are published in registration order. The broker's own
53 // machine daemon, when a producer is configured, is the
54 // prepended entry at index zero; without a producer the registry produces no
55 // local observation and never invents one.
56 daemons := make([]ports.BrokerDaemonObservation, 0, len(r.hosts)+1)
57 if r.local != nil {
58 daemons = append(daemons, r.localHost.Clone())
59 }
60 for _, endpoint := range r.order {
61 host, ok := r.hosts[endpoint]
62 if !ok {
63 continue
64 }
65 daemons = append(daemons, host.Clone())
66 }
67 snapshot := ports.BrokerSnapshot{Epoch: r.epoch, Revision: r.revision + 1, Daemons: daemons, Removed: r.tombstonesLocked()}
68 if err := snapshot.Validate(); err != nil {
69 // Fail closed: a publication the wire refuses aborts every client
70 // connection on subscribe, so an invalid projection is never published.
71 // The last valid snapshot stays current, nothing is persisted, and the
72 // revision series does not advance past a publication that never
73 // happened.
74 r.log.Error("broker: refused invalid snapshot publication", "revision", r.revision+1, "err", err)
75 return
76 }
77 r.revision = snapshot.Revision
78 r.current.Store(&snapshot)
79 if persist {
80 // enqueue only stages an immutable publication: rendering the durable
81 // copy and every store call happen on the writer goroutine, never
82 // under the registry lock and never on the probe schedule.
83 r.store.enqueue(snapshot)
84 }
85 for sub := range r.subs {
86 sub.offer()
87 }
88}
89
90// durable returns the persistable copy of a publication, rendered by the
91// writer goroutine with the shared rule (ports.BrokerSnapshot.DurableProjection).
92// The registry may publish the process-local daemon, which is never durable,
93// so it is dropped first; the store would refuse it otherwise. Restored
94// observations regain policy from membership (see stampHostAuthority).
95func durable(snapshot ports.BrokerSnapshot) ports.BrokerSnapshot {
96 return snapshot.WithoutLocal().DurableProjection()
97}
98
99type subscription struct {
100 changed chan struct{}
101 once sync.Once
102 close func()
103}
104
105func (s *subscription) Changed() <-chan struct{} { return s.changed }
106
107func (s *subscription) Close() { s.once.Do(s.close) }
108
109func (s *subscription) offer() {
110 select {
111 case s.changed <- struct{}{}:
112 default:
113 }
114}