feat(broker): serve a local-only observer through the authority
An authority's connections expose immutable membership: they never add or remove hosts. That is why NewAuthority refused any registry that observes, since an observer would probe configured hosts the service refuses to manage. That reasoning does not hold for an observer limited to the broker's own daemon. A local-only observer has no remote membership to manage, and its local entry is configured authority supplied by the composition rather than membership, so an authority can publish and serve it. Accept exactly that case, and keep refusing an observer that owns remote membership. This is what an offline composition needs to observe this machine's daemon and serve clients from the same publication.
- commit
- 70034fb09e350e5dbe27e90088a4aa0b8c3b0b8f
- author
- brice · · 2026-09-19 12:26 UTC
- parent
- eef2c8e
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +42 -6
M internal/usecase/broker/registry_local.go +11 -0
| @@ -75,6 +75,17 @@ func (l *LocalObservation) validate() error { | |||
| 75 | 75 | return nil | |
| 76 | 76 | } | |
| 77 | 77 | ||
| 78 | + | // observesLocalOnly reports whether this registry observes only the broker's | |
| 79 | + | // own daemon: observation is enabled, a local binding is configured, and no | |
| 80 | + | // remote membership is projected. It is the registry-side answer an authority | |
| 81 | + | // needs, because an authority whose connections expose immutable membership may | |
| 82 | + | // serve an observer only when there is no remote membership to manage. | |
| 83 | + | func (r *Registry) observesLocalOnly() bool { | |
| 84 | + | r.mu.Lock() | |
| 85 | + | defer r.mu.Unlock() | |
| 86 | + | return !r.observationDisabled && r.local != nil && len(r.hosts) == 0 | |
| 87 | + | } | |
| 88 | + | ||
| 78 | 89 | // dispatchLocalLocked admits at most one local probe attempt when its schedule | |
| 79 | 90 | // says it is due. Callers must hold r.mu. It reports whether an attempt | |
| 80 | 91 | // started, so dispatch publishes the transient Checking state exactly once. | |
| @@ -75,6 +75,17 @@ func (l *LocalObservation) validate() error { | |||||
| 75 | return nil | 75 | return nil | ||
| 76 | } | 76 | } | ||
| 77 | 77 | ||||
| 78 | + | // observesLocalOnly reports whether this registry observes only the broker's | |||
| 79 | + | // own daemon: observation is enabled, a local binding is configured, and no | |||
| 80 | + | // remote membership is projected. It is the registry-side answer an authority | |||
| 81 | + | // needs, because an authority whose connections expose immutable membership may | |||
| 82 | + | // serve an observer only when there is no remote membership to manage. | |||
| 83 | + | func (r *Registry) observesLocalOnly() bool { | |||
| 84 | + | r.mu.Lock() | |||
| 85 | + | defer r.mu.Unlock() | |||
| 86 | + | return !r.observationDisabled && r.local != nil && len(r.hosts) == 0 | |||
| 87 | + | } | |||
| 88 | + | ||||
| 78 | // dispatchLocalLocked admits at most one local probe attempt when its schedule | 89 | // dispatchLocalLocked admits at most one local probe attempt when its schedule | ||
| 79 | // says it is due. Callers must hold r.mu. It reports whether an attempt | 90 | // says it is due. Callers must hold r.mu. It reports whether an attempt | ||
| 80 | // started, so dispatch publishes the transient Checking state exactly once. | 91 | // started, so dispatch publishes the transient Checking state exactly once. | ||
M internal/usecase/broker/service.go +10 -6
| @@ -58,10 +58,14 @@ type Authority struct { | |||
| 58 | 58 | var _ ports.BrokerAuthority = (*Authority)(nil) | |
| 59 | 59 | ||
| 60 | 60 | // NewAuthority composes one connection-scoped broker authority over an existing | |
| 61 | - | // registry, pool, and supervisor. All three must be live and share epoch. The | |
| 62 | - | // registry must be observation-disabled: this authority's sessions expose | |
| 63 | - | // immutable offline membership, so an observing registry would probe hosts the | |
| 64 | - | // service refuses to manage. | |
| 61 | + | // registry, pool, and supervisor. All three must be live and share epoch. | |
| 62 | + | // | |
| 63 | + | // The authority's connections expose immutable membership: they neither add nor | |
| 64 | + | // remove hosts. It therefore accepts an observation-disabled registry, or an | |
| 65 | + | // observer whose observation is limited to the broker's own daemon, and refuses | |
| 66 | + | // a registry that observes configured hosts the service would refuse to manage. | |
| 67 | + | // A local-only observer has no remote membership to manage: its local entry is | |
| 68 | + | // configured authority supplied by the composition, not membership. | |
| 65 | 69 | // | |
| 66 | 70 | // The caller owns the dependencies' lifecycles. In particular it must start the | |
| 67 | 71 | // registry's single Registry.Run to own and drain the durable writer, and must | |
| @@ -76,8 +80,8 @@ func NewAuthority(epoch ports.BrokerEpoch, registry *Registry, pool *Pool, super | |||
| 76 | 80 | if registry.epoch != epoch || pool.epoch != epoch { | |
| 77 | 81 | return nil, errors.New("broker: authority epoch mismatch") | |
| 78 | 82 | } | |
| 79 | - | if !registry.observationDisabled { | |
| 80 | - | return nil, errors.New("broker: authority requires an observation-disabled registry") | |
| 83 | + | if !registry.observationDisabled && !registry.observesLocalOnly() { | |
| 84 | + | return nil, errors.New("broker: authority requires an observation-disabled registry or a local-only observer") | |
| 81 | 85 | } | |
| 82 | 86 | return &Authority{epoch: epoch, registry: registry, pool: pool, supervisor: supervisor}, nil | |
| 83 | 87 | } | |
| @@ -58,10 +58,14 @@ type Authority struct { | |||||
| 58 | var _ ports.BrokerAuthority = (*Authority)(nil) | 58 | var _ ports.BrokerAuthority = (*Authority)(nil) | ||
| 59 | 59 | ||||
| 60 | // NewAuthority composes one connection-scoped broker authority over an existing | 60 | // NewAuthority composes one connection-scoped broker authority over an existing | ||
| 61 | - | // registry, pool, and supervisor. All three must be live and share epoch. The | 61 | + | // registry, pool, and supervisor. All three must be live and share epoch. |
| 62 | - | // registry must be observation-disabled: this authority's sessions expose | 62 | + | // |
| 63 | - | // immutable offline membership, so an observing registry would probe hosts the | 63 | + | // The authority's connections expose immutable membership: they neither add nor |
| 64 | - | // service refuses to manage. | 64 | + | // remove hosts. It therefore accepts an observation-disabled registry, or an |
| 65 | + | // observer whose observation is limited to the broker's own daemon, and refuses | |||
| 66 | + | // a registry that observes configured hosts the service would refuse to manage. | |||
| 67 | + | // A local-only observer has no remote membership to manage: its local entry is | |||
| 68 | + | // configured authority supplied by the composition, not membership. | |||
| 65 | // | 69 | // | ||
| 66 | // The caller owns the dependencies' lifecycles. In particular it must start the | 70 | // The caller owns the dependencies' lifecycles. In particular it must start the | ||
| 67 | // registry's single Registry.Run to own and drain the durable writer, and must | 71 | // registry's single Registry.Run to own and drain the durable writer, and must | ||
| @@ -76,8 +80,8 @@ func NewAuthority(epoch ports.BrokerEpoch, registry *Registry, pool *Pool, super | |||||
| 76 | if registry.epoch != epoch || pool.epoch != epoch { | 80 | if registry.epoch != epoch || pool.epoch != epoch { | ||
| 77 | return nil, errors.New("broker: authority epoch mismatch") | 81 | return nil, errors.New("broker: authority epoch mismatch") | ||
| 78 | } | 82 | } | ||
| 79 | - | if !registry.observationDisabled { | 83 | + | if !registry.observationDisabled && !registry.observesLocalOnly() { |
| 80 | - | return nil, errors.New("broker: authority requires an observation-disabled registry") | 84 | + | return nil, errors.New("broker: authority requires an observation-disabled registry or a local-only observer") |
| 81 | } | 85 | } | ||
| 82 | return &Authority{epoch: epoch, registry: registry, pool: pool, supervisor: supervisor}, nil | 86 | return &Authority{epoch: epoch, registry: registry, pool: pool, supervisor: supervisor}, nil | ||
| 83 | } | 87 | } | ||
M internal/usecase/broker/service_test.go +21 -0
| @@ -126,6 +126,27 @@ func TestAuthorityValidation(t *testing.T) { | |||
| 126 | 126 | _, err = NewAuthority(1, observing, pool, supervisor) | |
| 127 | 127 | require.Error(t, err, "an observing registry must be refused by an offline authority") | |
| 128 | 128 | ||
| 129 | + | // An observer that is limited to the broker's own daemon is served: it has | |
| 130 | + | // no remote membership for the authority's immutable-membership connections | |
| 131 | + | // to manage, and its local entry is configured authority rather than | |
| 132 | + | // membership. | |
| 133 | + | localOnly, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{Local: &LocalObservation{ | |
| 134 | + | DisplayOrigin: "local", | |
| 135 | + | Policy: poolPolicy(), | |
| 136 | + | Probe: newScriptedLocalProbe(1), | |
| 137 | + | }}) | |
| 138 | + | require.NoError(t, err) | |
| 139 | + | _, err = NewAuthority(1, localOnly, pool, supervisor) | |
| 140 | + | require.NoError(t, err, "a local-only observer must be served") | |
| 141 | + | ||
| 142 | + | // The same observer stops being served as soon as it owns remote membership | |
| 143 | + | // the authority's connections could never manage. | |
| 144 | + | observerWithMembership, err := NewRegistry(1, newTestStore(), newTestProbe(1), clock, nil) | |
| 145 | + | require.NoError(t, err) | |
| 146 | + | require.NoError(t, observerWithMembership.setHosts(hostRecords(registration(t, "user@host:22", 1)))) | |
| 147 | + | _, err = NewAuthority(1, observerWithMembership, pool, supervisor) | |
| 148 | + | require.Error(t, err, "an observer that owns remote membership must be refused") | |
| 149 | + | ||
| 129 | 150 | disabled, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{ObservationDisabled: true}) | |
| 130 | 151 | require.NoError(t, err) | |
| 131 | 152 | _, err = NewAuthority(2, disabled, pool, supervisor) | |
| @@ -126,6 +126,27 @@ func TestAuthorityValidation(t *testing.T) { | |||||
| 126 | _, err = NewAuthority(1, observing, pool, supervisor) | 126 | _, err = NewAuthority(1, observing, pool, supervisor) | ||
| 127 | require.Error(t, err, "an observing registry must be refused by an offline authority") | 127 | require.Error(t, err, "an observing registry must be refused by an offline authority") | ||
| 128 | 128 | ||||
| 129 | + | // An observer that is limited to the broker's own daemon is served: it has | |||
| 130 | + | // no remote membership for the authority's immutable-membership connections | |||
| 131 | + | // to manage, and its local entry is configured authority rather than | |||
| 132 | + | // membership. | |||
| 133 | + | localOnly, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{Local: &LocalObservation{ | |||
| 134 | + | DisplayOrigin: "local", | |||
| 135 | + | Policy: poolPolicy(), | |||
| 136 | + | Probe: newScriptedLocalProbe(1), | |||
| 137 | + | }}) | |||
| 138 | + | require.NoError(t, err) | |||
| 139 | + | _, err = NewAuthority(1, localOnly, pool, supervisor) | |||
| 140 | + | require.NoError(t, err, "a local-only observer must be served") | |||
| 141 | + | ||||
| 142 | + | // The same observer stops being served as soon as it owns remote membership | |||
| 143 | + | // the authority's connections could never manage. | |||
| 144 | + | observerWithMembership, err := NewRegistry(1, newTestStore(), newTestProbe(1), clock, nil) | |||
| 145 | + | require.NoError(t, err) | |||
| 146 | + | require.NoError(t, observerWithMembership.setHosts(hostRecords(registration(t, "user@host:22", 1)))) | |||
| 147 | + | _, err = NewAuthority(1, observerWithMembership, pool, supervisor) | |||
| 148 | + | require.Error(t, err, "an observer that owns remote membership must be refused") | |||
| 149 | + | ||||
| 129 | disabled, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{ObservationDisabled: true}) | 150 | disabled, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{ObservationDisabled: true}) | ||
| 130 | require.NoError(t, err) | 151 | require.NoError(t, err) | ||
| 131 | _, err = NewAuthority(2, disabled, pool, supervisor) | 152 | _, err = NewAuthority(2, disabled, pool, supervisor) | ||