export_test.go RawHistoryFolder

1package broker
2
3import (
4	"errors"
5	"fmt"
6	"log/slog"
7
8	"github.com/bnema/vev/internal/ports"
9)
10
11// NewRegistry restores a validated durable snapshot under a fresh broker
12// epoch with the default RegistryConfig, delegating to the same
13// NewRegistryWithConfig production composition uses. It is a test-only
14// shorthand: production always selects an explicit membership/observation
15// mode.
16func NewRegistry(epoch ports.BrokerEpoch, store ports.BrokerHostStore, probe ports.BrokerHostProbe, clock ports.Clock, log *slog.Logger) (*Registry, error) {
17	return NewRegistryWithConfig(epoch, store, probe, clock, log, RegistryConfig{})
18}
19
20// setHosts is projection-only; it is private so callers cannot bypass durable
21// membership and policy. Observation tests use it without claiming durability.
22// An unchanged membership is a no-op: it neither republishes nor restarts an
23// in-flight observation. An exhausted revision series is refused with
24// ports.ErrBrokerRevisionExhausted before the projection changes.
25// Removed registrations retire bounded tombstones so a stale completion can
26// never revive them; replaced or re-added endpoints are fenced by the
27// registry attempt token instead.
28func (r *Registry) setHosts(records []ports.BrokerHostRecord) error {
29	if len(records) > ports.BrokerMaxHosts {
30		return fmt.Errorf("broker: %d host registrations exceed the limit of %d", len(records), ports.BrokerMaxHosts)
31	}
32	if len(records) > 0 && !r.observationDisabled && nilDependency(r.probe) {
33		return errors.New("broker: remote hosts require a host probe")
34	}
35	next := make(map[string]ports.BrokerDaemonObservation, len(records))
36	order := make([]string, 0, len(records))
37	for rank, record := range records {
38		if err := record.Registration.Validate(); err != nil {
39			return err
40		}
41		if err := record.Policy.Validate(); err != nil {
42			return err
43		}
44		registration := record.Registration
45		if _, duplicate := next[registration.Endpoint]; duplicate {
46			return errors.New("broker: duplicate host registration")
47		}
48		observation := stampHostAuthority(ports.BrokerDaemonObservation{}, record, rank)
49		// Membership the registry cannot publish is a caller error, refused here
50		// rather than left to freeze every later publication: an endpoint whose
51		// derived display hint is empty after the display rule drops its runes
52		// makes every snapshot carrying it invalid.
53		if err := observation.Validate(); err != nil {
54			return fmt.Errorf("broker: host record %q is not publishable: %w", registration.Endpoint, err)
55		}
56		next[registration.Endpoint] = observation
57		order = append(order, registration.Endpoint)
58	}
59
60	r.mu.Lock()
61	defer r.mu.Unlock()
62	if r.revisionExhaustedLocked() {
63		return fmt.Errorf("broker: set hosts: %w", ports.ErrBrokerRevisionExhausted)
64	}
65	r.setHostsLocked(next, order)
66	return nil
67}