1package broker
2
3import (
4 "log/slog"
5 "reflect"
6 "sync"
7 "time"
8
9 "github.com/bnema/vev/internal/ports"
10)
11
12// snapshotWriter serializes durable snapshot writes off the registry lock.
13// Enqueue never blocks on store I/O and always coalesces to the newest
14// revision; Store calls run on one goroutine, so the durable store never sees
15// an out-of-order write. A failed write is logged and retried by the next
16// publication, so a store fault never stalls publication. Close flushes the
17// newest pending snapshot, stops the writer, and waits, so no write outlives
18// shutdown.
19type snapshotWriter struct {
20 store ports.BrokerSnapshotStore
21 log *slog.Logger
22
23 mu sync.Mutex
24 pending *ports.BrokerSnapshot
25 written ports.BrokerRevision
26 // last is the durable content of the newest successful write with its
27 // revision and observation timestamps cleared. Only the writer goroutine
28 // touches it.
29 last *ports.BrokerSnapshot
30 started bool
31 closed bool
32 wake chan struct{}
33 stop chan struct{}
34 done chan struct{}
35}
36
37func newSnapshotWriter(store ports.BrokerSnapshotStore, log *slog.Logger) *snapshotWriter {
38 return &snapshotWriter{
39 store: store, log: log,
40 wake: make(chan struct{}, 1), stop: make(chan struct{}), done: make(chan struct{}),
41 }
42}
43
44// enqueue stages a snapshot for durable persistence. It coalesces to the
45// newest revision, starts the writer on first use, and never blocks on the
46// store or copies session payloads. Enqueue after Close is refused: nothing is
47// written after shutdown.
48func (w *snapshotWriter) enqueue(snapshot ports.BrokerSnapshot) {
49 w.mu.Lock()
50 if w.closed {
51 w.mu.Unlock()
52 w.log.Debug("broker: durable snapshot write refused after shutdown", "revision", snapshot.Revision)
53 return
54 }
55 if w.pending != nil {
56 if snapshot.Revision <= w.pending.Revision {
57 w.mu.Unlock()
58 return
59 }
60 } else if snapshot.Revision <= w.written {
61 w.mu.Unlock()
62 return
63 }
64 // The publication is immutable after publication, so staging it is a value
65 // copy; the defensive durable copy is rendered on the writer goroutine.
66 w.pending = &snapshot
67 if !w.started {
68 w.started = true
69 go w.run()
70 }
71 w.mu.Unlock()
72 select {
73 case w.wake <- struct{}{}:
74 default:
75 }
76}
77
78func (w *snapshotWriter) run() {
79 defer close(w.done)
80 for {
81 select {
82 case <-w.stop:
83 w.flush()
84 return
85 case <-w.wake:
86 w.flush()
87 }
88 }
89}
90
91// flush writes the newest staged snapshot, skipping revisions the store
92// already holds.
93func (w *snapshotWriter) flush() {
94 for {
95 w.mu.Lock()
96 snapshot := w.pending
97 written := w.written
98 w.pending = nil
99 w.mu.Unlock()
100 if snapshot == nil {
101 return
102 }
103 if snapshot.Revision <= written {
104 continue
105 }
106 next := durable(*snapshot)
107 key := durableContent(next)
108 if w.last != nil && reflect.DeepEqual(*w.last, key) {
109 // Only observation timestamps moved: the stored copy already
110 // carries the same durable state, so skip the rewrite and fsync.
111 // A restart re-observes every host before trusting freshness.
112 continue
113 }
114 if err := w.store.Store(next); err != nil {
115 w.last = nil
116 w.log.Error("broker: durable snapshot write failed", "revision", snapshot.Revision, "err", err)
117 continue
118 }
119 w.last = &key
120 w.mu.Lock()
121 if snapshot.Revision > w.written {
122 w.written = snapshot.Revision
123 }
124 w.mu.Unlock()
125 }
126}
127
128// durableContent returns the durable projection without the fields that change
129// on every routine observation: the revision and the attempt, success, and
130// due timestamps.
131func durableContent(snapshot ports.BrokerSnapshot) ports.BrokerSnapshot {
132 out := snapshot
133 out.Revision = 0
134 out.Daemons = make([]ports.BrokerDaemonObservation, len(snapshot.Daemons))
135 for i, daemon := range snapshot.Daemons {
136 daemon.LastAttempt = time.Time{}
137 daemon.LastSuccess = time.Time{}
138 daemon.NextDue = time.Time{}
139 out.Daemons[i] = daemon
140 }
141 return out
142}
143
144// close flushes the newest staged snapshot, stops the writer, and waits for
145// the drain to finish. It never returns before the in-flight write completes.
146func (w *snapshotWriter) close() {
147 w.mu.Lock()
148 if w.closed {
149 w.mu.Unlock()
150 return
151 }
152 w.closed = true
153 started := w.started
154 w.mu.Unlock()
155 if !started {
156 return
157 }
158 close(w.stop)
159 <-w.done
160}