1// Package brokerstore provides broker persistence. Legacy sources are immutable inputs:
2// the lifetime lock is the same hosts.json.lock the legacy host writer takes,
3// so a live host writer is excluded, but the legacy catalog cache source has no
4// lock at all and must be immutable or quiesced. No environment policy is
5// inferred.
6package brokerstore
7
8import (
9 "crypto/sha256"
10 "encoding/hex"
11 "encoding/json"
12 "errors"
13 "fmt"
14 "os"
15 "path/filepath"
16 "reflect"
17 "sync"
18 "syscall"
19
20 "github.com/bnema/vev/internal/ports"
21 "github.com/bnema/vev/pkg/safedir"
22)
23
24const MaxFileBytes = 16 << 20
25
26// lockFileName is the lifetime lock of a durable broker store. The name is
27// deliberately the one the legacy host writer takes beside hosts.json, so a
28// store opened on a directory that writer also uses excludes it for the
29// store's lifetime. The legacy catalog cache source is written without any
30// lock, so that source must be immutable or quiesced before migration.
31const lockFileName = "hosts.json.lock"
32
33// ErrLocked reports that another owner already holds the lifetime lock.
34var ErrLocked = fmt.Errorf("brokerstore: already owned: %w", ports.ErrBrokerStoreLocked)
35
36// ErrStale reports that the caller's authority token no longer matches durable
37// state, or that the supplied membership would regress or re-trust an existing
38// registration without a fresh identity.
39var ErrStale = fmt.Errorf("brokerstore: stale authority: %w", ports.ErrBrokerHostConflict)
40
41// ErrInvalidState reports durable state that fails validation. The caller must
42// resolve it offline from the recovery record; the store never replaces it.
43var ErrInvalidState = fmt.Errorf("brokerstore: invalid state: %w", ports.ErrBrokerStoreInvalidState)
44
45// invalid wraps one durable-state validation failure.
46func invalid(format string, args ...any) error {
47 return fmt.Errorf("%w: %s", ErrInvalidState, fmt.Sprintf(format, args...))
48}
49
50// manifestSourceLabel names one immutable recovery input by its generic role:
51// the membership source that seeds authority, and the observation source that
52// only seeds advisory inventory. A label never names a legacy file or schema.
53type manifestSourceLabel string
54
55const (
56 manifestMembership manifestSourceLabel = "membership"
57 manifestObservations manifestSourceLabel = "observations"
58)
59
60// manifestSource records one recovery input. Present distinguishes a source
61// that was supplied but carried no records from a source that was never
62// supplied: absence and emptiness can share a digest, so only presence makes
63// the recovery record an exact description of the migration input.
64type manifestSource struct {
65 Label manifestSourceLabel
66 Present bool
67 SHA256 string
68}
69
70type manifest struct {
71 Version int
72 Sources []manifestSource
73 ImportVersion int
74 ImportCompleted bool
75}
76
77// source returns the manifest entry for one generic role label.
78func (m manifest) source(label manifestSourceLabel) (manifestSource, bool) {
79 for _, source := range m.Sources {
80 if source.Label == label {
81 return source, true
82 }
83 }
84 return manifestSource{}, false
85}
86
87// Options names a private destination and immutable legacy inputs.
88// Policies must explicitly cover every imported endpoint. Fault is a test seam
89// called after each durable-write boundary; an error poisons this open handle.
90type Options struct {
91 Dir string
92 InitialHosts []ports.BrokerHostRecord
93 InitialImportProvided bool
94 Fault func(string) error
95}
96
97// migrate records the caller-provided initial membership without importing legacy files.
98func migrate(o Options) (recovery, error) {
99 r := recovery{State: state{Manifest: manifest{Version: 1, ImportVersion: 1, ImportCompleted: true, Sources: []manifestSource{{Label: manifestMembership, SHA256: digest(nil)}, {Label: manifestObservations, SHA256: digest(nil)}}}, Hosts: ports.BrokerHosts{Revision: 1}}}
100 if o.InitialImportProvided {
101 r.State.Hosts.Hosts = ports.CloneBrokerHostRecords(o.InitialHosts)
102 if err := ports.ValidateBrokerHostRecords(r.State.Hosts.Hosts); err != nil && len(r.State.Hosts.Hosts) != 0 {
103 return r, fmt.Errorf("brokerstore: initial import: %w", err)
104 }
105 }
106 return r, validate(r.State)
107}
108
109type state struct {
110 Manifest manifest
111 Hosts ports.BrokerHosts
112 Snapshot ports.BrokerSnapshot
113}
114type recovery struct {
115 State state
116 Hosts, Cache []byte
117}
118type Store struct {
119 mu sync.Mutex
120 dir string
121 lock *os.File
122 state state
123 fault func(string) error
124 failed bool
125 writeEpoch ports.BrokerEpoch // first publication binds this lifetime handle
126}
127
128var _ ports.BrokerHostStore = (*Store)(nil)
129
130// ReadHosts reads committed membership without taking the writer lock or
131// creating any files. Advisory snapshots are deliberately ignored.
132func ReadHosts(dir string) (ports.BrokerHosts, error) {
133 raw, err := readBounded(filepath.Join(dir, "state.json"))
134 if err != nil {
135 return ports.BrokerHosts{}, err
136 }
137 var st state
138 if err := strict(raw, &st); err != nil {
139 return ports.BrokerHosts{}, invalid("state.json: %v", err)
140 }
141 if err := validate(st); err != nil {
142 return ports.BrokerHosts{}, err
143 }
144 return ports.BrokerHosts{Revision: st.Hosts.Revision, Hosts: ports.CloneBrokerHostRecords(st.Hosts.Hosts)}, nil
145}
146
147// Open returns only after validated unified state is durable. The
148// recovery record is written first and retained forever as the rollback copy;
149// restart reuses its generated identities instead of migrating a second time.
150// An existing corrupt state is an error, never silently replaced by a backup.
151// Open acquires exclusive lifetime ownership, migrates immutable legacy input
152// when needed, and returns the durable broker store.
153func Open(o Options) (*Store, error) {
154 if err := safedir.EnsurePrivate(o.Dir); err != nil {
155 return nil, err
156 }
157 fd, err := syscall.Open(filepath.Join(o.Dir, lockFileName), syscall.O_CREAT|syscall.O_RDWR|syscall.O_CLOEXEC|syscall.O_NOFOLLOW, 0600)
158 if err != nil {
159 return nil, err
160 }
161 lock := os.NewFile(uintptr(fd), lockFileName)
162 info, err := lock.Stat()
163 if err != nil {
164 lock.Close()
165 return nil, err
166 }
167 if !info.Mode().IsRegular() || info.Mode().Perm() != 0600 {
168 lock.Close()
169 return nil, invalid("unsafe owner lock")
170 }
171 if err = syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
172 lock.Close()
173 return nil, fmt.Errorf("%w: %v", ErrLocked, err)
174 }
175 s := &Store{dir: o.Dir, lock: lock, fault: o.Fault}
176 ok := false
177 defer func() {
178 if !ok {
179 s.Close()
180 }
181 }()
182 raw, err := readBounded(filepath.Join(o.Dir, "state.json"))
183 if err == nil {
184 if err = strict(raw, &s.state); err != nil {
185 return nil, invalid("state.json: %v", err)
186 }
187 } else if errors.Is(err, os.ErrNotExist) {
188 var r recovery
189 raw, err = readBounded(filepath.Join(o.Dir, "recovery.json"))
190 if err == nil {
191 if err = strict(raw, &r); err != nil {
192 return nil, invalid("recovery.json: %v", err)
193 }
194 if err = r.verify(); err != nil {
195 return nil, err
196 }
197 } else if errors.Is(err, os.ErrNotExist) {
198 r, err = migrate(o)
199 if err == nil {
200 err = s.write("recovery.json", r)
201 }
202 } else {
203 return nil, invalid("recovery.json: %v", err)
204 }
205 if err == nil {
206 err = validate(r.State)
207 }
208 if err == nil {
209 err = s.write("state.json", r.State)
210 }
211 s.state = r.State
212 } else {
213 return nil, invalid("state.json: %v", err)
214 }
215 if err != nil {
216 return nil, err
217 }
218 upgraded, upgradeErr := ports.UpgradeBrokerHostRoutes(s.state.Hosts.Hosts)
219 if upgradeErr != nil {
220 return nil, upgradeErr
221 }
222 if !reflect.DeepEqual(upgraded, s.state.Hosts.Hosts) {
223 next := s.state
224 next.Hosts.Hosts = upgraded
225 next.Manifest.ImportVersion = 1
226 next.Manifest.ImportCompleted = true
227 recovery := recovery{State: next}
228 if err = s.write("recovery.json", recovery); err != nil {
229 return nil, err
230 }
231 if err = s.commit(next); err != nil {
232 return nil, err
233 }
234 } else if err = validate(s.state); err != nil {
235 return nil, err
236 }
237 ok = true
238 return s, nil
239}
240
241// verify reports whether the recovery record's retained bytes match their
242// manifest entries, including each source's recorded presence: an absent
243// source must retain no bytes, and a supplied source keeps its exact digest.
244func (r recovery) verify() error {
245 membership, ok := r.State.Manifest.source(manifestMembership)
246 if !ok || digest(r.Hosts) != membership.SHA256 {
247 return invalid("recovery digest mismatch")
248 }
249 observations, ok := r.State.Manifest.source(manifestObservations)
250 if !ok || digest(r.Cache) != observations.SHA256 {
251 return invalid("recovery digest mismatch")
252 }
253 for _, source := range []struct {
254 label manifestSourceLabel
255 present bool
256 bytes int
257 }{{manifestMembership, membership.Present, len(r.Hosts)}, {manifestObservations, observations.Present, len(r.Cache)}} {
258 if !source.present && source.bytes != 0 {
259 return invalid("recovery records an absent %s source with retained bytes", source.label)
260 }
261 }
262 return nil
263}
264func digest(b []byte) string { sum := sha256.Sum256(b); return hex.EncodeToString(sum[:]) }
265func (s *Store) ready() error {
266 if s.failed {
267 return ports.BrokerStoreOutcomeUnknownError{Err: errors.New("brokerstore: poisoned; reopen required")}
268 }
269 if s.lock == nil {
270 return errors.New("brokerstore: closed")
271 }
272 return nil
273}
274func (s *Store) Close() error {
275 s.mu.Lock()
276 defer s.mu.Unlock()
277 if s.lock == nil {
278 return nil
279 }
280 err := s.lock.Close()
281 s.lock = nil
282 return err
283}
284func (s *Store) LoadHosts() (ports.BrokerHosts, error) {
285 s.mu.Lock()
286 defer s.mu.Unlock()
287 if err := s.ready(); err != nil {
288 return ports.BrokerHosts{}, err
289 }
290 h := s.state.Hosts
291 h.Hosts = ports.CloneBrokerHostRecords(h.Hosts)
292 return h, nil
293}
294func (s *Store) Load() (ports.BrokerSnapshot, error) {
295 s.mu.Lock()
296 defer s.mu.Unlock()
297 if err := s.ready(); err != nil {
298 return ports.BrokerSnapshot{}, err
299 }
300 return s.state.Snapshot.Clone(), nil
301}
302func (s *Store) ReplaceHosts(expected uint64, hosts []ports.BrokerHostRecord) error {
303 s.mu.Lock()
304 defer s.mu.Unlock()
305 if err := s.ready(); err != nil {
306 return err
307 }
308 // Validate caller membership before the CAS: malformed registration,
309 // policy, anchoring, bound, or duplicate-endpoint input is a caller error,
310 // not corrupt durable state, so it never surfaces as ErrInvalidState.
311 if err := ports.ValidateBrokerHostRecords(hosts); err != nil {
312 return err
313 }
314 if expected != s.state.Hosts.Revision || expected == ^uint64(0) {
315 return ErrStale
316 }
317 next := s.state
318 next.Hosts = ports.BrokerHosts{Revision: expected + 1, Hosts: ports.CloneBrokerHostRecords(hosts)}
319 // Policy changes require a new registration generation as well; otherwise
320 // an already queued observation could be accepted under changed trust.
321 for _, h := range hosts {
322 for _, old := range s.state.Hosts.Hosts {
323 if h.Registration.Endpoint == old.Registration.Endpoint && h.Policy != old.Policy && h.Registration.Equal(old.Registration) {
324 return ErrStale
325 }
326 if h.Registration.Incarnation == old.Registration.Incarnation && h.Registration.Endpoint == old.Registration.Endpoint && h.Registration.Generation < old.Registration.Generation {
327 return ErrStale
328 }
329 }
330 }
331 next.Snapshot = filter(next.Snapshot, next.Hosts)
332 return s.commit(next)
333}
334func (s *Store) Store(snapshot ports.BrokerSnapshot) error {
335 s.mu.Lock()
336 defer s.mu.Unlock()
337 if err := s.ready(); err != nil {
338 return err
339 }
340 // Sanitize the caller's copy with the shared durable rule before
341 // validating the durable shape.
342 snapshot = snapshot.DurableProjection()
343 if err := ports.ValidateDurableSnapshot(snapshot); err != nil {
344 return err
345 }
346 if s.writeEpoch != 0 && s.writeEpoch != snapshot.Epoch {
347 return ErrStale
348 }
349 old := s.state.Snapshot
350 if old.Epoch == snapshot.Epoch && old.Revision >= snapshot.Revision {
351 return ErrStale
352 }
353 // Reject rather than partially adopt a queued publication from retired hosts.
354 for _, daemon := range snapshot.Daemons {
355 found := false
356 for _, a := range s.state.Hosts.Hosts {
357 if daemon.Registration.Equal(a.Registration) {
358 found = true
359 }
360 }
361 if !found {
362 return ErrStale
363 }
364 }
365 next := s.state
366 next.Snapshot = snapshot.Clone()
367 if err := s.commit(next); err != nil {
368 return err
369 }
370 s.writeEpoch = snapshot.Epoch
371 return nil
372}
373
374func filter(snapshot ports.BrokerSnapshot, hosts ports.BrokerHosts) ports.BrokerSnapshot {
375 out := snapshot.Clone()
376 out.Daemons = nil
377 out.Removed = nil
378 for _, daemon := range snapshot.Daemons {
379 for _, a := range hosts.Hosts {
380 if daemon.Registration.Equal(a.Registration) {
381 out.Daemons = append(out.Daemons, daemon.Clone())
382 }
383 }
384 }
385 return out
386}
387func validate(st state) error {
388 if st.Manifest.Version != 1 {
389 return invalid("unsupported manifest")
390 }
391 if st.Manifest.ImportVersion != 1 || !st.Manifest.ImportCompleted {
392 return invalid("initial import is incomplete")
393 }
394 labels := make(map[manifestSourceLabel]bool, len(st.Manifest.Sources))
395 for _, source := range st.Manifest.Sources {
396 if labels[source.Label] {
397 return invalid("duplicate manifest source")
398 }
399 labels[source.Label] = true
400 b, err := hex.DecodeString(source.SHA256)
401 if err != nil || len(b) != 32 {
402 return invalid("invalid digest")
403 }
404 }
405 for _, label := range []manifestSourceLabel{manifestMembership, manifestObservations} {
406 if !labels[label] {
407 return invalid("missing manifest source")
408 }
409 }
410 if err := st.Hosts.Validate(); err != nil {
411 return invalid("%v", err)
412 }
413 if st.Snapshot.Epoch == 0 {
414 if st.Snapshot.Revision != 0 || len(st.Snapshot.Daemons) != 0 || len(st.Snapshot.Removed) != 0 {
415 return invalid("invalid empty snapshot")
416 }
417 return nil
418 }
419 if err := ports.ValidateDurableSnapshot(st.Snapshot); err != nil {
420 return invalid("%v", err)
421 }
422 for _, daemon := range st.Snapshot.Daemons {
423 found := false
424 for _, a := range st.Hosts.Hosts {
425 if daemon.Registration.Equal(a.Registration) {
426 found = true
427 }
428 }
429 if !found || daemon.Checking || daemon.LastFailure.Err != nil {
430 return invalid("non-authoritative snapshot")
431 }
432 }
433 return nil
434}
435
436func (s *Store) commit(next state) error {
437 if err := validate(next); err != nil {
438 return err
439 }
440 if err := s.write("state.json", next); err != nil {
441 s.failed = true
442 return err
443 }
444 s.state = next
445 return nil
446}
447func (s *Store) boundary(name string) error {
448 if s.fault != nil {
449 return s.fault(name)
450 }
451 return nil
452}
453func (s *Store) write(name string, value any) error {
454 data, err := json.Marshal(value)
455 if err != nil {
456 return err
457 }
458 if len(data) > MaxFileBytes {
459 return errors.New("brokerstore: file too large")
460 }
461 f, err := os.CreateTemp(s.dir, ".pending-")
462 if err != nil {
463 return err
464 }
465 defer os.Remove(f.Name())
466 defer f.Close()
467 // The staged state carries rollback inputs and registration identities, so
468 // its permissions are set explicitly instead of relying on the umask.
469 if err = f.Chmod(0o600); err != nil {
470 return err
471 }
472 if _, err = f.Write(data); err != nil {
473 return err
474 }
475 if err = s.boundary(name + ":write"); err != nil {
476 return err
477 }
478 if err = f.Sync(); err != nil {
479 return err
480 }
481 if err = s.boundary(name + ":sync"); err != nil {
482 return err
483 }
484 if err = f.Close(); err != nil {
485 return err
486 }
487 if err = os.Rename(f.Name(), filepath.Join(s.dir, name)); err != nil {
488 return err
489 }
490 unknown := func(err error) error {
491 if err == nil {
492 return nil
493 }
494 return ports.BrokerStoreOutcomeUnknownError{Err: err}
495 }
496 if err = s.boundary(name + ":rename"); err != nil {
497 return unknown(err)
498 }
499 d, err := os.Open(s.dir)
500 if err != nil {
501 return unknown(err)
502 }
503 defer d.Close()
504 if err = d.Sync(); err != nil {
505 return unknown(err)
506 }
507 return unknown(s.boundary(name + ":dirsync"))
508}