1package broker
2
3import (
4 "context"
5 "errors"
6 "sync/atomic"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/require"
11
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/internal/protocol"
15)
16
17// scriptedLocalProbe scripts one local observation per received call: the test
18// owns the completion, so the local probe never depends on wall time.
19type scriptedLocalProbe struct {
20 calls chan localProbeCall
21 canceled chan struct{}
22}
23
24type localProbeCall struct {
25 result chan localProbeAnswer
26}
27
28type localProbeAnswer struct {
29 snapshot ports.BrokerDaemonObservation
30 err error
31}
32
33func newScriptedLocalProbe(capacity int) *scriptedLocalProbe {
34 return &scriptedLocalProbe{calls: make(chan localProbeCall, capacity), canceled: make(chan struct{}, 1)}
35}
36
37func (p *scriptedLocalProbe) ProbeLocal(ctx context.Context) (ports.BrokerDaemonObservation, error) {
38 call := localProbeCall{result: make(chan localProbeAnswer, 1)}
39 select {
40 case p.calls <- call:
41 case <-ctx.Done():
42 p.noteCanceled()
43 return ports.BrokerDaemonObservation{}, ctx.Err()
44 }
45 select {
46 case answer := <-call.result:
47 return answer.snapshot, answer.err
48 case <-ctx.Done():
49 p.noteCanceled()
50 return ports.BrokerDaemonObservation{}, ctx.Err()
51 }
52}
53
54func (p *scriptedLocalProbe) noteCanceled() {
55 select {
56 case p.canceled <- struct{}{}:
57 default:
58 }
59}
60
61func receiveLocalCall(t *testing.T, probe *scriptedLocalProbe) localProbeCall {
62 t.Helper()
63 select {
64 case call := <-probe.calls:
65 return call
66 case <-time.After(time.Second):
67 t.Fatal("timed out waiting for local probe")
68 return localProbeCall{}
69 }
70}
71
72// requireNoLocalProbeWithin asserts a negative scheduling outcome: no local
73// probe attempt starts inside window, so the test observes "nothing happened"
74// without sleeping on the behavior under test.
75func requireNoLocalProbeWithin(t *testing.T, probe *scriptedLocalProbe, window time.Duration) {
76 t.Helper()
77 timer := time.NewTimer(window)
78 defer timer.Stop()
79 select {
80 case <-probe.calls:
81 t.Fatal("unexpected local probe attempt")
82 case <-timer.C:
83 }
84}
85
86// newLocalTestRegistry builds a registry that observes the broker's own machine
87// daemon through localProbe, beside an optional remote probe.
88func newLocalTestRegistry(t *testing.T, store ports.BrokerHostStore, remoteProbe ports.BrokerHostProbe, localProbe LocalProbe, clock *manualClock) *Registry {
89 t.Helper()
90 registry, err := NewRegistryWithConfig(1, store, remoteProbe, clock, nil, RegistryConfig{
91 Local: &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: localProbe},
92 })
93 require.NoError(t, err)
94 registry.jitter = identityJitter
95 return registry
96}
97
98// waitLocal waits for the published local entry (always index zero) to satisfy
99// predicate and returns it.
100func waitLocal(t *testing.T, r *Registry, predicate func(ports.BrokerDaemonObservation) bool) ports.BrokerDaemonObservation {
101 t.Helper()
102 snapshot := waitSnapshot(t, r, func(snapshot ports.BrokerSnapshot) bool {
103 return len(snapshot.Daemons) > 0 && snapshot.Daemons[0].Local && predicate(snapshot.Daemons[0])
104 })
105 return snapshot.Daemons[0]
106}
107
108// TestRegistryPublishesObservedLocalDaemonFirst proves the local observation is
109// published at index zero ahead of every remote, carries the configured
110// authority plus the observed identity/incarnation/version, and is never
111// durable: the store persists the remote observations only.
112func TestRegistryPublishesObservedLocalDaemonFirst(t *testing.T) {
113 clock := newManualClock(time.Unix(100, 0))
114 store := newTestStore()
115 local := newScriptedLocalProbe(1)
116 registry := newLocalTestRegistry(t, store, newTestProbe(1), local, clock)
117 reg := registration(t, "user@host:22", 1)
118 require.NoError(t, registry.setHosts(hostRecords(reg)))
119 startRegistry(t, registry)
120
121 call := receiveLocalCall(t, local)
122 call.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
123 Identity: "local-daemon",
124 Incarnation: ports.BrokerDaemonIncarnation{9},
125 ProtocolVersion: protocol.Version,
126 Availability: domain.RemoteAvailabilityReachable,
127 }}
128
129 host := waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool {
130 return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
131 })
132 require.True(t, host.Local)
133 require.Zero(t, host.Endpoint)
134 require.Equal(t, domain.RemoteRegistration{}, host.Registration)
135 require.Equal(t, "local", host.DisplayOrigin)
136 require.Equal(t, poolPolicy(), host.Policy)
137 require.Zero(t, host.Rank)
138 require.Equal(t, ports.BrokerDaemonIdentity("local-daemon"), host.Identity)
139 require.Equal(t, ports.BrokerDaemonIncarnation{9}, host.Incarnation)
140 require.Equal(t, protocol.Version, host.ProtocolVersion)
141
142 snapshot := registry.Snapshot()
143 require.True(t, snapshot.Daemons[0].Local)
144 require.Equal(t, reg.Endpoint, snapshot.Daemons[1].Endpoint)
145
146 // Never durable: the store receives remote observations only, and the local
147 // entry is dropped before persistence.
148 stored := waitStored(t, store, func(snapshot ports.BrokerSnapshot) bool {
149 return len(snapshot.Daemons) == 1 && !snapshot.Daemons[0].Local
150 })
151 require.Equal(t, reg.Endpoint, stored.Daemons[0].Endpoint)
152}
153
154// TestRegistryLocalObservationNeverInventsState proves an unreachable local
155// daemon is published with the explicit unknown availability and zero observed
156// identity, and a hostile probe result cannot supply local authority,
157// endpoint, registration, policy, or display origin.
158func TestRegistryLocalObservationNeverInventsState(t *testing.T) {
159 t.Run("failure keeps zero identity and typed availability", func(t *testing.T) {
160 clock := newManualClock(time.Unix(100, 0))
161 local := newScriptedLocalProbe(1)
162 registry := newLocalTestRegistry(t, newTestStore(), newTestProbe(1), local, clock)
163 startRegistry(t, registry)
164
165 call := receiveLocalCall(t, local)
166 call.result <- localProbeAnswer{err: errors.New("local carriage refused")}
167
168 host := waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool {
169 return !o.Checking && o.Availability == domain.RemoteAvailabilityUnreachable
170 })
171 require.True(t, host.Local)
172 require.Zero(t, host.Endpoint)
173 require.Equal(t, ports.BrokerDaemonIdentity(""), host.Identity)
174 require.True(t, host.Incarnation.IsZero())
175 require.Zero(t, host.ProtocolVersion)
176 require.Equal(t, domain.RemoteFailureTransport, host.LastFailure.Kind)
177 require.Equal(t, 1, len(registry.Snapshot().Daemons))
178 })
179
180 t.Run("hostile probe cannot supply local authority", func(t *testing.T) {
181 clock := newManualClock(time.Unix(100, 0))
182 local := newScriptedLocalProbe(1)
183 registry := newLocalTestRegistry(t, newTestStore(), newTestProbe(1), local, clock)
184 startRegistry(t, registry)
185
186 hostilePolicy := poolPolicy()
187 hostilePolicy.Trust = "attacker"
188 call := receiveLocalCall(t, local)
189 call.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
190 Local: false,
191 Endpoint: "attacker@evil",
192 DisplayOrigin: "evil",
193 Rank: 42,
194 Registration: registration(t, "attacker@evil", 9),
195 Policy: hostilePolicy,
196 Identity: "local-daemon",
197 Incarnation: ports.BrokerDaemonIncarnation{9},
198 ProtocolVersion: protocol.Version,
199 Availability: domain.RemoteAvailabilityReachable,
200 }}
201
202 host := waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool {
203 return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
204 })
205 require.True(t, host.Local)
206 require.Zero(t, host.Endpoint)
207 require.Equal(t, domain.RemoteRegistration{}, host.Registration)
208 require.Equal(t, "local", host.DisplayOrigin)
209 require.Equal(t, poolPolicy(), host.Policy)
210 require.NotEqual(t, hostilePolicy, host.Policy)
211 require.Zero(t, host.Rank)
212 })
213}
214
215// TestRegistryLocalPendingReprobeHonorsExplicitReconcileOnly proves the local
216// pending re-probe flag (set by RequestProbe("")) is the only thing that
217// admits a local attempt ahead of its scheduled NextDue: neither an explicit
218// unknown result nor a plain success schedules an early re-probe on its own,
219// but a reconcile requested right after either one is dispatched immediately,
220// on the fake clock, with no time advance at all. This is the fix for the
221// stale local catalogue after create/attach (`ls --all`/the picker showing a
222// new or newly-attached session for up to the retry/fresh interval): the
223// service now calls RequestProbe("") once a local control or attachment
224// stream completes, and this test pins the registry-side half of that fix.
225func TestRegistryLocalPendingReprobeHonorsExplicitReconcileOnly(t *testing.T) {
226 reachable := localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
227 Identity: "local-daemon",
228 Incarnation: ports.BrokerDaemonIncarnation{9},
229 ProtocolVersion: protocol.Version,
230 Availability: domain.RemoteAvailabilityReachable,
231 }}
232 unknown := localProbeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnknown}}
233
234 tests := []struct {
235 name string
236 first localProbeAnswer
237 requestReconcile bool
238 wantReprobe bool
239 }{
240 {
241 name: "after an unknown result, no reconcile never re-probes before due",
242 first: unknown,
243 wantReprobe: false,
244 },
245 {
246 name: "after a success, no reconcile never re-probes before due",
247 first: reachable,
248 wantReprobe: false,
249 },
250 {
251 name: "after an unknown result, an explicit local reconcile re-probes before due",
252 first: unknown,
253 requestReconcile: true,
254 wantReprobe: true,
255 },
256 {
257 name: "after a success, an explicit local reconcile re-probes before due",
258 first: reachable,
259 requestReconcile: true,
260 wantReprobe: true,
261 },
262 }
263
264 for _, tt := range tests {
265 t.Run(tt.name, func(t *testing.T) {
266 clock := newManualClock(time.Unix(100, 0))
267 local := newScriptedLocalProbe(2)
268 registry := newLocalTestRegistry(t, newTestStore(), nil, local, clock)
269 startRegistry(t, registry)
270
271 call := receiveLocalCall(t, local)
272 call.result <- tt.first
273 waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
274
275 if tt.requestReconcile {
276 registry.RequestProbe("")
277 }
278
279 // No clock advance at all: NextDue (freshFor after success, the retry
280 // delay after unknown) is still in the future either way, so only the
281 // pending flag - never the schedule - can explain a dispatch here.
282 if tt.wantReprobe {
283 receiveLocalCall(t, local)
284 } else {
285 requireNoLocalProbeWithin(t, local, probeGrace)
286 }
287 })
288 }
289}
290
291// TestRegistryLocalObservationValidatesProbeResults proves an invalid observed
292// projection is classified as an invalid response, never published as an
293// observation the wire refuses: a partial identity (identity without a version)
294// is rejected without inventing the missing half.
295func TestRegistryLocalObservationValidatesProbeResults(t *testing.T) {
296 clock := newManualClock(time.Unix(100, 0))
297 local := newScriptedLocalProbe(1)
298 registry := newLocalTestRegistry(t, newTestStore(), newTestProbe(1), local, clock)
299 startRegistry(t, registry)
300
301 call := receiveLocalCall(t, local)
302 call.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
303 Identity: "local-daemon", // no protocol version: partial identity
304 Incarnation: ports.BrokerDaemonIncarnation{9},
305 Availability: domain.RemoteAvailabilityReachable,
306 }}
307
308 host := waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool {
309 return !o.Checking && o.Availability == domain.RemoteAvailabilityInvalidResponse
310 })
311 require.True(t, host.Local)
312 require.Equal(t, domain.RemoteFailureInvalidResponse, host.LastFailure.Kind)
313}
314
315// TestRegistryLocalObservationConstruction pins the composition refusals: a
316// read-only registry owns no local producer, a local producer requires a probe,
317// and its policy and display origin are validated before anything is published.
318func TestRegistryLocalObservationConstruction(t *testing.T) {
319 clock := newManualClock(time.Unix(100, 0))
320 valid := &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: newScriptedLocalProbe(1)}
321
322 invalidPolicy := poolPolicy()
323 invalidPolicy.ProtocolVersion = 0
324
325 tests := []struct {
326 name string
327 cfg RegistryConfig
328 remoteProbe ports.BrokerHostProbe
329 }{
330 {
331 name: "observation-disabled registry cannot observe locally",
332 cfg: RegistryConfig{ObservationDisabled: true, Local: valid},
333 remoteProbe: newTestProbe(1),
334 },
335 {
336 name: "local producer requires a probe",
337 cfg: RegistryConfig{Local: &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy()}},
338 remoteProbe: newTestProbe(1),
339 },
340 {
341 name: "local authority policy is validated",
342 cfg: RegistryConfig{Local: &LocalObservation{DisplayOrigin: "local", Policy: invalidPolicy, Probe: newScriptedLocalProbe(1)}},
343 remoteProbe: newTestProbe(1),
344 },
345 {
346 name: "local display origin is required",
347 cfg: RegistryConfig{Local: &LocalObservation{DisplayOrigin: "", Policy: poolPolicy(), Probe: newScriptedLocalProbe(1)}},
348 remoteProbe: newTestProbe(1),
349 },
350 {
351 name: "local display origin must be normalized",
352 cfg: RegistryConfig{Local: &LocalObservation{DisplayOrigin: "ev\u202eil", Policy: poolPolicy(), Probe: newScriptedLocalProbe(1)}},
353 remoteProbe: newTestProbe(1),
354 },
355 {
356 name: "no producer at all is still refused",
357 cfg: RegistryConfig{},
358 remoteProbe: nil,
359 },
360 }
361 for _, tt := range tests {
362 t.Run(tt.name, func(t *testing.T) {
363 _, err := NewRegistryWithConfig(1, newTestStore(), tt.remoteProbe, clock, nil, tt.cfg)
364 require.Error(t, err)
365 })
366 }
367}
368
369// TestRegistryLocalOnlyRequiresNoRemoteProbe proves a composition that observes
370// only the broker's own machine daemon needs no remote probe, and that such a
371// registry refuses remote membership rather than silently never probing it.
372func TestRegistryLocalOnlyRequiresNoRemoteProbe(t *testing.T) {
373 clock := newManualClock(time.Unix(100, 0))
374 local := newScriptedLocalProbe(1)
375 registry, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{
376 Local: &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: local},
377 })
378 require.NoError(t, err)
379 registry.jitter = identityJitter
380
381 reg := registration(t, "user@host:22", 1)
382 require.Error(t, registry.setHosts(hostRecords(reg)))
383 require.Error(t, registry.ReplaceHosts(hostRecords(reg)))
384
385 startRegistry(t, registry)
386 call := receiveLocalCall(t, local)
387 call.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
388 Identity: "local-daemon",
389 Incarnation: ports.BrokerDaemonIncarnation{9},
390 ProtocolVersion: protocol.Version,
391 Availability: domain.RemoteAvailabilityReachable,
392 }}
393 host := waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool {
394 return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
395 })
396 require.True(t, host.Local)
397 require.Len(t, registry.Snapshot().Daemons, 1)
398}
399
400// localEntryCount counts the published local entries. A valid snapshot carries
401// at most one, so this is also the assertion that no second local entry is
402// ever invented.
403func localEntryCount(snapshot ports.BrokerSnapshot) int {
404 count := 0
405 for _, daemon := range snapshot.Daemons {
406 if daemon.Local {
407 count++
408 }
409 }
410 return count
411}
412
413// remoteEndpoints lists the published remote endpoints in publication order,
414// excluding the local entry.
415func remoteEndpoints(snapshot ports.BrokerSnapshot) []string {
416 var endpoints []string
417 for _, daemon := range snapshot.Daemons {
418 if daemon.Local {
419 continue
420 }
421 endpoints = append(endpoints, daemon.Endpoint)
422 }
423 return endpoints
424}
425
426// TestRegistryLocalEntryStaysFirstAcrossMembershipChanges proves the published
427// snapshot always carries exactly one local entry at index zero while the
428// remote membership churns through add, reorder, removal, re-add, and clear,
429// with the remotes in registration order behind it. The local entry stays first
430// because the registry owns it separately from durable membership: it is never
431// reordered by a membership change and never replaced by a remote.
432func TestRegistryLocalEntryStaysFirstAcrossMembershipChanges(t *testing.T) {
433 first := registration(t, "user@first:22", 1)
434 second := registration(t, "user@second:22", 2)
435 third := registration(t, "user@third:22", 3)
436
437 steps := []struct {
438 name string
439 apply func(t *testing.T, r *Registry)
440 remotes []string
441 }{
442 {
443 name: "no remotes publishes the local entry alone",
444 apply: func(*testing.T, *Registry) {},
445 },
446 {
447 name: "adding remotes keeps the local entry first",
448 apply: func(t *testing.T, r *Registry) { require.NoError(t, r.setHosts(hostRecords(first, second))) },
449 remotes: []string{first.Endpoint, second.Endpoint},
450 },
451 {
452 name: "reordering remotes keeps the local entry first",
453 apply: func(t *testing.T, r *Registry) { require.NoError(t, r.setHosts(hostRecords(second, first))) },
454 remotes: []string{second.Endpoint, first.Endpoint},
455 },
456 {
457 name: "durable replacement keeps the local entry first",
458 apply: func(t *testing.T, r *Registry) { require.NoError(t, r.ReplaceHosts(hostRecords(first))) },
459 remotes: []string{first.Endpoint},
460 },
461 {
462 name: "removing then re-adding a remote keeps the local entry first",
463 apply: func(t *testing.T, r *Registry) {
464 require.NoError(t, r.ReplaceHosts(hostRecords()))
465 require.NoError(t, r.ReplaceHosts(hostRecords(first, third)))
466 },
467 remotes: []string{first.Endpoint, third.Endpoint},
468 },
469 {
470 name: "clearing durable membership leaves the local entry alone",
471 apply: func(t *testing.T, r *Registry) { require.NoError(t, r.ReplaceHosts(hostRecords())) },
472 remotes: nil,
473 },
474 }
475 registry := newLocalTestRegistry(t, newTestStore(), newTestProbe(1), newScriptedLocalProbe(1), newManualClock(time.Unix(100, 0)))
476 for _, tt := range steps {
477 t.Run(tt.name, func(t *testing.T) {
478 tt.apply(t, registry)
479 snapshot := registry.Snapshot()
480 require.NotEmpty(t, snapshot.Daemons)
481 require.True(t, snapshot.Daemons[0].Local, "the local entry is always at index zero")
482 require.Equal(t, 1, localEntryCount(snapshot), "exactly one local entry is published")
483 require.Equal(t, tt.remotes, remoteEndpoints(snapshot), "remotes keep registration order behind the local entry")
484 })
485 }
486}
487
488// TestRegistryLocalOnlyRefusesRestoredRemoteMembership proves a local-only
489// registry (a configured local observation and no remote probe) refuses restored
490// durable remote membership at the projectMembership boundary, so it never owns
491// a remote host it could never probe. The setHosts/ReplaceHosts refusals are
492// covered beside this test.
493func TestRegistryLocalOnlyRefusesRestoredRemoteMembership(t *testing.T) {
494 clock := newManualClock(time.Unix(100, 0))
495 local := func() *LocalObservation {
496 return &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: newScriptedLocalProbe(1)}
497 }
498 reg := registration(t, "user@host:22", 1)
499
500 t.Run("construction refuses a store with restored remote membership", func(t *testing.T) {
501 store := newTestStore()
502 store.hosts = ports.BrokerHosts{Revision: 1, Hosts: hostRecords(reg)}
503 _, err := NewRegistryWithConfig(1, store, nil, clock, nil, RegistryConfig{Local: local()})
504 require.Error(t, err)
505 require.Contains(t, err.Error(), "remote hosts require a host probe")
506 })
507
508 t.Run("projectMembership refuses restored remote membership directly", func(t *testing.T) {
509 registry, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{Local: local()})
510 require.NoError(t, err)
511 // Empty restored membership is accepted: the local-only registry owns no
512 // remote host and needs no remote probe.
513 require.NoError(t, registry.projectMembership(ports.BrokerHosts{Revision: 1}))
514 err = registry.projectMembership(ports.BrokerHosts{Revision: 1, Hosts: hostRecords(reg)})
515 require.Error(t, err)
516 require.Contains(t, err.Error(), "remote hosts require a host probe")
517 require.Empty(t, remoteEndpoints(registry.Snapshot()), "the refused membership is never published")
518 })
519}
520
521// TestPoolLocalStreamsShareOnePhysicalEntry proves two local open-stream
522// requests under one broker-owned local binding resolve to the same pooled
523// (identity, policy) physical entry: the pool dials one daemonmux carriage,
524// both logical streams ride it, and closing one never closes its sibling. It
525// lives here because pool_test.go is outside this change's write scope; the
526// fake connector/resolver harness it reuses is defined there.
527func TestPoolLocalStreamsShareOnePhysicalEntry(t *testing.T) {
528 var connects atomic.Int32
529 var physical *fakePhysical
530 pool, _ := setupPool(t, func(_ context.Context, endpoint ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
531 connects.Add(1)
532 physical = &fakePhysical{endpoint: endpoint, done: make(chan struct{})}
533 return physical, nil
534 })
535 id, err := pool.RegisterClient()
536 require.NoError(t, err)
537 first, err := pool.OpenStream(context.Background(), poolRequest(id, 1))
538 require.NoError(t, err)
539 second, err := pool.OpenStream(context.Background(), poolRequest(id, 2))
540 require.NoError(t, err)
541 require.EqualValues(t, 1, connects.Load(), "both local streams share one physical carriage")
542 require.NoError(t, first.Close())
543 select {
544 case <-second.Done():
545 t.Fatal("closing one local stream closed its sibling")
546 default:
547 }
548 require.NoError(t, second.Close())
549 require.NoError(t, physical.Close())
550}
551
552// TestRegistryMutableAddHostLocalOnlyNoRemoteProbe proves a mutable registry that
553// observes only the broker's own machine daemon (a configured local observation
554// and no remote probe) refuses a remote addition outright: the refusal is
555// typed, it never reaches the durable store CAS, authority and projection stay
556// exactly as loaded, and the registry keeps serving its local entry instead of
557// panicking on the absent remote probe. A mutable registry is the interesting
558// case because the immutable default refuses before the membership mode is even
559// consulted.
560func TestRegistryMutableAddHostLocalOnlyNoRemoteProbe(t *testing.T) {
561 clock := newManualClock(time.Unix(100, 0))
562 store := newMembershipStore()
563 local := newScriptedLocalProbe(1)
564 registry, err := NewRegistryWithConfig(1, store, nil, clock, nil, RegistryConfig{
565 MembershipMode: MembershipMutable,
566 IncarnationGenerator: (&incarnationSequencer{}).generate,
567 Local: &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: local},
568 })
569 require.NoError(t, err)
570 registry.jitter = identityJitter
571
572 authority := registry.authority
573 before := registry.Snapshot()
574 require.NotPanics(t, func() {
575 _, err = registry.AddHost(context.Background(), "user@host:22", poolPolicy())
576 }, "an absent remote probe must be refused, never dereferenced")
577 require.Error(t, err)
578 require.Contains(t, err.Error(), "remote hosts require a host probe")
579 require.Equal(t, 0, store.casCalls(), "the refusal happens before the store CAS")
580 require.Equal(t, authority, registry.authority, "a refused add never moves authority")
581 require.Equal(t, before, registry.Snapshot(), "a refused add never publishes")
582
583 // The local entry stays observable: the refusal never leaves the registry
584 // unable to publish what it still owns.
585 startRegistry(t, registry)
586 call := receiveLocalCall(t, local)
587 call.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
588 Identity: "local-daemon",
589 Incarnation: ports.BrokerDaemonIncarnation{9},
590 ProtocolVersion: protocol.Version,
591 Availability: domain.RemoteAvailabilityReachable,
592 }}
593 host := waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool {
594 return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
595 })
596 require.True(t, host.Local)
597 require.Len(t, registry.Snapshot().Daemons, 1)
598}