detached ce90900

refactor(broker): keep one durable snapshot validator

commit
ce909004ac56a4521cdeb290a5259d87d3ba7dc6
author
brice · · 2026-09-23 12:14 UTC
parent
2f0d2b4
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +41 -101
Unified
UnifiedSplit
M internal/adapters/brokerstore/store.go +2 -39
@@ -359,7 +359,7 @@ func (s *Store) Store(snapshot ports.BrokerSnapshot) error {
359359 }
360360 }
361361 }
362- if err := validateDurableSnapshot(snapshot); err != nil {
362+ if err := ports.ValidateDurableSnapshot(snapshot); err != nil {
363363 return err
364364 }
365365 if s.writeEpoch != 0 && s.writeEpoch != snapshot.Epoch {
@@ -435,7 +435,7 @@ func validate(st state) error {
435435 }
436436 return nil
437437 }
438- if err := validateDurableSnapshot(st.Snapshot); err != nil {
438+ if err := ports.ValidateDurableSnapshot(st.Snapshot); err != nil {
439439 return invalid("%v", err)
440440 }
441441 for _, daemon := range st.Snapshot.Daemons {
@@ -452,43 +452,6 @@ func validate(st state) error {
452452 return nil
453453 }
454454
455-func validateDurableSnapshot(snapshot ports.BrokerSnapshot) error {
456- if snapshot.Epoch == 0 {
457- if snapshot.Revision != 0 || len(snapshot.Daemons) != 0 || len(snapshot.Removed) != 0 {
458- return errors.New("invalid empty snapshot")
459- }
460- return nil
461- }
462- if snapshot.Revision == 0 {
463- return errors.New("snapshot has no revision")
464- }
465- if len(snapshot.Daemons) > ports.BrokerMaxDaemonsPerSnapshot {
466- return errors.New("snapshot has too many daemons")
467- }
468- seen := make(map[string]struct{}, len(snapshot.Daemons))
469- for _, daemon := range snapshot.Daemons {
470- if err := ports.ValidateDurableObservation(daemon); err != nil {
471- return err
472- }
473- if _, duplicate := seen[daemon.Endpoint]; duplicate {
474- return errors.New("snapshot has duplicate host")
475- }
476- seen[daemon.Endpoint] = struct{}{}
477- }
478- if len(snapshot.Removed) > ports.BrokerMaxTombstones {
479- return errors.New("snapshot has too many tombstones")
480- }
481- for _, tombstone := range snapshot.Removed {
482- if err := tombstone.Validate(); err != nil {
483- return err
484- }
485- if _, live := seen[tombstone.Endpoint]; live {
486- return errors.New("snapshot carries a live host as tombstone")
487- }
488- }
489- return nil
490-}
491-
492455 func (s *Store) commit(next state) error {
493456 if err := validate(next); err != nil {
494457 return err
@@ -359,7 +359,7 @@ func (s *Store) Store(snapshot ports.BrokerSnapshot) error {
359 }359 }
360 }360 }
361 }361 }
362- if err := validateDurableSnapshot(snapshot); err != nil {362+ if err := ports.ValidateDurableSnapshot(snapshot); err != nil {
363 return err363 return err
364 }364 }
365 if s.writeEpoch != 0 && s.writeEpoch != snapshot.Epoch {365 if s.writeEpoch != 0 && s.writeEpoch != snapshot.Epoch {
@@ -435,7 +435,7 @@ func validate(st state) error {
435 }435 }
436 return nil436 return nil
437 }437 }
438- if err := validateDurableSnapshot(st.Snapshot); err != nil {438+ if err := ports.ValidateDurableSnapshot(st.Snapshot); err != nil {
439 return invalid("%v", err)439 return invalid("%v", err)
440 }440 }
441 for _, daemon := range st.Snapshot.Daemons {441 for _, daemon := range st.Snapshot.Daemons {
@@ -452,43 +452,6 @@ func validate(st state) error {
452 return nil452 return nil
453 }453 }
454 454
455-func validateDurableSnapshot(snapshot ports.BrokerSnapshot) error {
456- if snapshot.Epoch == 0 {
457- if snapshot.Revision != 0 || len(snapshot.Daemons) != 0 || len(snapshot.Removed) != 0 {
458- return errors.New("invalid empty snapshot")
459- }
460- return nil
461- }
462- if snapshot.Revision == 0 {
463- return errors.New("snapshot has no revision")
464- }
465- if len(snapshot.Daemons) > ports.BrokerMaxDaemonsPerSnapshot {
466- return errors.New("snapshot has too many daemons")
467- }
468- seen := make(map[string]struct{}, len(snapshot.Daemons))
469- for _, daemon := range snapshot.Daemons {
470- if err := ports.ValidateDurableObservation(daemon); err != nil {
471- return err
472- }
473- if _, duplicate := seen[daemon.Endpoint]; duplicate {
474- return errors.New("snapshot has duplicate host")
475- }
476- seen[daemon.Endpoint] = struct{}{}
477- }
478- if len(snapshot.Removed) > ports.BrokerMaxTombstones {
479- return errors.New("snapshot has too many tombstones")
480- }
481- for _, tombstone := range snapshot.Removed {
482- if err := tombstone.Validate(); err != nil {
483- return err
484- }
485- if _, live := seen[tombstone.Endpoint]; live {
486- return errors.New("snapshot carries a live host as tombstone")
487- }
488- }
489- return nil
490-}
491-
492 func (s *Store) commit(next state) error {455 func (s *Store) commit(next state) error {
493 if err := validate(next); err != nil {456 if err := validate(next); err != nil {
494 return err457 return err
M internal/ports/broker_store.go +38 -0
@@ -272,3 +272,41 @@ func ValidateDurableObservation(daemon BrokerDaemonObservation) error {
272272 }
273273 return ValidateDurableHostProjection(daemon)
274274 }
275+
276+// ValidateDurableSnapshot checks the shared durable snapshot shape.
277+func ValidateDurableSnapshot(snapshot BrokerSnapshot) error {
278+ if snapshot.Epoch == 0 {
279+ if snapshot.Revision != 0 || len(snapshot.Daemons) != 0 || len(snapshot.Removed) != 0 {
280+ return errors.New("invalid empty snapshot")
281+ }
282+ return nil
283+ }
284+ if snapshot.Revision == 0 {
285+ return errors.New("snapshot has no revision")
286+ }
287+ if len(snapshot.Daemons) > BrokerMaxDaemonsPerSnapshot {
288+ return errors.New("snapshot has too many daemons")
289+ }
290+ seen := make(map[string]struct{}, len(snapshot.Daemons))
291+ for _, daemon := range snapshot.Daemons {
292+ if err := ValidateDurableObservation(daemon); err != nil {
293+ return err
294+ }
295+ if _, duplicate := seen[daemon.Endpoint]; duplicate {
296+ return errors.New("snapshot has duplicate host")
297+ }
298+ seen[daemon.Endpoint] = struct{}{}
299+ }
300+ if len(snapshot.Removed) > BrokerMaxTombstones {
301+ return errors.New("snapshot has too many tombstones")
302+ }
303+ for _, tombstone := range snapshot.Removed {
304+ if err := tombstone.Validate(); err != nil {
305+ return err
306+ }
307+ if _, live := seen[tombstone.Endpoint]; live {
308+ return errors.New("snapshot carries a live host as tombstone")
309+ }
310+ }
311+ return nil
312+}
@@ -272,3 +272,41 @@ func ValidateDurableObservation(daemon BrokerDaemonObservation) error {
272 }272 }
273 return ValidateDurableHostProjection(daemon)273 return ValidateDurableHostProjection(daemon)
274 }274 }
275+
276+// ValidateDurableSnapshot checks the shared durable snapshot shape.
277+func ValidateDurableSnapshot(snapshot BrokerSnapshot) error {
278+ if snapshot.Epoch == 0 {
279+ if snapshot.Revision != 0 || len(snapshot.Daemons) != 0 || len(snapshot.Removed) != 0 {
280+ return errors.New("invalid empty snapshot")
281+ }
282+ return nil
283+ }
284+ if snapshot.Revision == 0 {
285+ return errors.New("snapshot has no revision")
286+ }
287+ if len(snapshot.Daemons) > BrokerMaxDaemonsPerSnapshot {
288+ return errors.New("snapshot has too many daemons")
289+ }
290+ seen := make(map[string]struct{}, len(snapshot.Daemons))
291+ for _, daemon := range snapshot.Daemons {
292+ if err := ValidateDurableObservation(daemon); err != nil {
293+ return err
294+ }
295+ if _, duplicate := seen[daemon.Endpoint]; duplicate {
296+ return errors.New("snapshot has duplicate host")
297+ }
298+ seen[daemon.Endpoint] = struct{}{}
299+ }
300+ if len(snapshot.Removed) > BrokerMaxTombstones {
301+ return errors.New("snapshot has too many tombstones")
302+ }
303+ for _, tombstone := range snapshot.Removed {
304+ if err := tombstone.Validate(); err != nil {
305+ return err
306+ }
307+ if _, live := seen[tombstone.Endpoint]; live {
308+ return errors.New("snapshot carries a live host as tombstone")
309+ }
310+ }
311+ return nil
312+}
M internal/usecase/broker/registry.go +1 -62
@@ -408,7 +408,7 @@ func (r *Registry) restore(snapshot ports.BrokerSnapshot) error {
408408 if snapshot.Epoch == r.epoch {
409409 return errors.New("broker: persisted snapshot epoch matches the new registry epoch")
410410 }
411- if err := validateRestoredSnapshot(snapshot); err != nil {
411+ if err := ports.ValidateDurableSnapshot(snapshot); err != nil {
412412 return err
413413 }
414414 for _, daemon := range snapshot.Daemons {
@@ -419,67 +419,6 @@ func (r *Registry) restore(snapshot ports.BrokerSnapshot) error {
419419 return nil
420420 }
421421
422-// validateRestoredSnapshot applies the durable shape rules to a snapshot read
423-// back from a store: every observation must be a remote, non-local projection
424-// of an exact valid registration whose endpoint matches, with a closed
425-// availability/failure range and a catalogue-valid inventory. Policy and
426-// display authority are deliberately not checked here because the durable
427-// format omits policy and projectMembership stamps both from membership, so a
428-// restored observation is always remote-only and always carries authority once
429-// membership is projected.
430-func validateRestoredSnapshot(snapshot ports.BrokerSnapshot) error {
431- if snapshot.Revision == 0 {
432- return errors.New("broker: persisted snapshot has no revision")
433- }
434- if len(snapshot.Daemons) > ports.BrokerMaxDaemonsPerSnapshot {
435- return errors.New("broker: persisted snapshot has too many daemons")
436- }
437- seen := make(map[string]struct{}, len(snapshot.Daemons))
438- for _, daemon := range snapshot.Daemons {
439- if err := validateDurableObservation(daemon); err != nil {
440- return err
441- }
442- if _, duplicate := seen[daemon.Endpoint]; duplicate {
443- return fmt.Errorf("broker: persisted snapshot has duplicate host %q", daemon.Endpoint)
444- }
445- seen[daemon.Endpoint] = struct{}{}
446- }
447- if len(snapshot.Removed) > ports.BrokerMaxTombstones {
448- return errors.New("broker: persisted snapshot has too many tombstones")
449- }
450- for _, tombstone := range snapshot.Removed {
451- if err := tombstone.Validate(); err != nil {
452- return err
453- }
454- if _, live := seen[tombstone.Endpoint]; live {
455- return fmt.Errorf("broker: persisted snapshot carries live host %q as tombstone", tombstone.Endpoint)
456- }
457- }
458- return nil
459-}
460-
461-// validateDurableObservation reports whether one observation satisfies the
462-// durable shape shared with the offline store: remote-only, an exact valid
463-// registration matching its endpoint, and the ports durable projection rules.
464-func validateDurableObservation(daemon ports.BrokerDaemonObservation) error {
465- if daemon.Local {
466- return errors.New("broker: persisted local daemon observation")
467- }
468- if err := domain.ValidateRemoteHostTarget(daemon.Endpoint); err != nil {
469- return fmt.Errorf("broker: persisted observation: %w", err)
470- }
471- if err := daemon.Registration.Validate(); err != nil {
472- return fmt.Errorf("broker: persisted observation: %w", err)
473- }
474- if daemon.Endpoint != daemon.Registration.Endpoint {
475- return errors.New("broker: persisted observation endpoint does not match registration")
476- }
477- if len(daemon.Sessions) > 0 && !daemon.InventoryKnown {
478- return errors.New("broker: persisted observation carries sessions without known inventory")
479- }
480- return ports.ValidateDurableHostProjection(daemon)
481-}
482-
483422 // Snapshot returns a defensive immutable copy without I/O.
484423 func (r *Registry) Snapshot() ports.BrokerSnapshot { return r.current.Load().Clone() }
485424
@@ -408,7 +408,7 @@ func (r *Registry) restore(snapshot ports.BrokerSnapshot) error {
408 if snapshot.Epoch == r.epoch {408 if snapshot.Epoch == r.epoch {
409 return errors.New("broker: persisted snapshot epoch matches the new registry epoch")409 return errors.New("broker: persisted snapshot epoch matches the new registry epoch")
410 }410 }
411- if err := validateRestoredSnapshot(snapshot); err != nil {411+ if err := ports.ValidateDurableSnapshot(snapshot); err != nil {
412 return err412 return err
413 }413 }
414 for _, daemon := range snapshot.Daemons {414 for _, daemon := range snapshot.Daemons {
@@ -419,67 +419,6 @@ func (r *Registry) restore(snapshot ports.BrokerSnapshot) error {
419 return nil419 return nil
420 }420 }
421 421
422-// validateRestoredSnapshot applies the durable shape rules to a snapshot read
423-// back from a store: every observation must be a remote, non-local projection
424-// of an exact valid registration whose endpoint matches, with a closed
425-// availability/failure range and a catalogue-valid inventory. Policy and
426-// display authority are deliberately not checked here because the durable
427-// format omits policy and projectMembership stamps both from membership, so a
428-// restored observation is always remote-only and always carries authority once
429-// membership is projected.
430-func validateRestoredSnapshot(snapshot ports.BrokerSnapshot) error {
431- if snapshot.Revision == 0 {
432- return errors.New("broker: persisted snapshot has no revision")
433- }
434- if len(snapshot.Daemons) > ports.BrokerMaxDaemonsPerSnapshot {
435- return errors.New("broker: persisted snapshot has too many daemons")
436- }
437- seen := make(map[string]struct{}, len(snapshot.Daemons))
438- for _, daemon := range snapshot.Daemons {
439- if err := validateDurableObservation(daemon); err != nil {
440- return err
441- }
442- if _, duplicate := seen[daemon.Endpoint]; duplicate {
443- return fmt.Errorf("broker: persisted snapshot has duplicate host %q", daemon.Endpoint)
444- }
445- seen[daemon.Endpoint] = struct{}{}
446- }
447- if len(snapshot.Removed) > ports.BrokerMaxTombstones {
448- return errors.New("broker: persisted snapshot has too many tombstones")
449- }
450- for _, tombstone := range snapshot.Removed {
451- if err := tombstone.Validate(); err != nil {
452- return err
453- }
454- if _, live := seen[tombstone.Endpoint]; live {
455- return fmt.Errorf("broker: persisted snapshot carries live host %q as tombstone", tombstone.Endpoint)
456- }
457- }
458- return nil
459-}
460-
461-// validateDurableObservation reports whether one observation satisfies the
462-// durable shape shared with the offline store: remote-only, an exact valid
463-// registration matching its endpoint, and the ports durable projection rules.
464-func validateDurableObservation(daemon ports.BrokerDaemonObservation) error {
465- if daemon.Local {
466- return errors.New("broker: persisted local daemon observation")
467- }
468- if err := domain.ValidateRemoteHostTarget(daemon.Endpoint); err != nil {
469- return fmt.Errorf("broker: persisted observation: %w", err)
470- }
471- if err := daemon.Registration.Validate(); err != nil {
472- return fmt.Errorf("broker: persisted observation: %w", err)
473- }
474- if daemon.Endpoint != daemon.Registration.Endpoint {
475- return errors.New("broker: persisted observation endpoint does not match registration")
476- }
477- if len(daemon.Sessions) > 0 && !daemon.InventoryKnown {
478- return errors.New("broker: persisted observation carries sessions without known inventory")
479- }
480- return ports.ValidateDurableHostProjection(daemon)
481-}
482-
483 // Snapshot returns a defensive immutable copy without I/O.422 // Snapshot returns a defensive immutable copy without I/O.
484 func (r *Registry) Snapshot() ports.BrokerSnapshot { return r.current.Load().Clone() }423 func (r *Registry) Snapshot() ports.BrokerSnapshot { return r.current.Load().Clone() }
485 424

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close