registry_test.go RawHistoryFolder

1package broker
2
3import (
4	"bytes"
5	"context"
6	"crypto/sha256"
7	"errors"
8	"fmt"
9	"log/slog"
10	"sort"
11	"strings"
12	"sync"
13	"testing"
14	"time"
15	"unicode/utf8"
16
17	"github.com/bnema/vev/internal/domain"
18	"github.com/bnema/vev/internal/ports"
19	"github.com/bnema/vev/internal/protocol"
20	"github.com/bnema/vev/internal/protocol/catalogue"
21	"github.com/stretchr/testify/require"
22)
23
24// probeGrace bounds a negative scheduling assertion: a probe may not start
25// inside this window, so the test observes "nothing happened" without
26// sleeping on the behavior under test.
27const probeGrace = 25 * time.Millisecond
28
29// manualClock is a deterministic ports.Clock/ports.Timer pair: time only moves
30// when Advance is called, and timers with equal deadlines fire in creation
31// order. Scheduling assertions are therefore exact instead of racing wall time.
32type manualClock struct {
33	mu     sync.Mutex
34	now    time.Time
35	order  uint64
36	timers map[*manualTimer]struct{}
37}
38
39func newManualClock(start time.Time) *manualClock {
40	return &manualClock{now: start, timers: make(map[*manualTimer]struct{})}
41}
42
43func (c *manualClock) Now() time.Time {
44	c.mu.Lock()
45	defer c.mu.Unlock()
46	return c.now
47}
48
49func (c *manualClock) NewTimer(d time.Duration) ports.Timer {
50	timer := &manualTimer{clock: c, ch: make(chan time.Time, 1)}
51	c.mu.Lock()
52	defer c.mu.Unlock()
53	timer.arm(d)
54	return timer
55}
56
57// Advance moves time forward and fires every timer that came due.
58func (c *manualClock) Advance(d time.Duration) {
59	if d < 0 {
60		panic("broker: manualClock.Advance requires a non-negative duration")
61	}
62	c.mu.Lock()
63	defer c.mu.Unlock()
64	c.now = c.now.Add(d)
65	due := make([]*manualTimer, 0, len(c.timers))
66	for timer := range c.timers {
67		if !timer.when.After(c.now) {
68			due = append(due, timer)
69		}
70	}
71	sort.Slice(due, func(i, j int) bool {
72		if due[i].when.Equal(due[j].when) {
73			return due[i].order < due[j].order
74		}
75		return due[i].when.Before(due[j].when)
76	})
77	for _, timer := range due {
78		timer.active = false
79		delete(c.timers, timer)
80		select {
81		case timer.ch <- c.now:
82		default:
83		}
84	}
85}
86
87type manualTimer struct {
88	clock  *manualClock
89	ch     chan time.Time
90	when   time.Time
91	order  uint64
92	active bool
93}
94
95func (t *manualTimer) C() <-chan time.Time { return t.ch }
96
97func (t *manualTimer) Stop() bool {
98	t.clock.mu.Lock()
99	defer t.clock.mu.Unlock()
100	if !t.active {
101		return false
102	}
103	t.active = false
104	delete(t.clock.timers, t)
105	return true
106}
107
108func (t *manualTimer) Reset(d time.Duration) bool {
109	t.clock.mu.Lock()
110	defer t.clock.mu.Unlock()
111	wasActive := t.active
112	if t.active {
113		t.active = false
114		delete(t.clock.timers, t)
115	}
116	select {
117	case <-t.ch:
118	default:
119	}
120	t.arm(d)
121	return wasActive
122}
123
124// arm schedules the timer relative to the clock's current time. The caller
125// must hold clock.mu.
126func (t *manualTimer) arm(d time.Duration) {
127	t.when = t.clock.now.Add(d)
128	t.order = t.clock.order
129	t.clock.order++
130	if d <= 0 {
131		t.active = false
132		select {
133		case t.ch <- t.clock.now:
134		default:
135		}
136		return
137	}
138	t.active = true
139	t.clock.timers[t] = struct{}{}
140}
141
142// identityJitter removes scheduling jitter so a test can assert exact
143// deadlines; the production jitter is covered separately.
144func identityJitter(d time.Duration, _ string, _ uint64) time.Duration { return d }
145
146// syncBuffer captures log output written from the registry's persistence
147// goroutine as well as the test goroutine.
148type syncBuffer struct {
149	mu sync.Mutex
150	b  bytes.Buffer
151}
152
153func (w *syncBuffer) Write(p []byte) (int, error) {
154	w.mu.Lock()
155	defer w.mu.Unlock()
156	return w.b.Write(p)
157}
158
159func (w *syncBuffer) String() string {
160	w.mu.Lock()
161	defer w.mu.Unlock()
162	return w.b.String()
163}
164
165func testLogger() (*slog.Logger, *syncBuffer) {
166	logs := &syncBuffer{}
167	return slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug})), logs
168}
169
170// testStore is a race-safe in-memory durable store whose writes signal a
171// capacity-one channel so tests wait on a write instead of on wall time. It
172// implements the whole BrokerHostStore seam: authority defaults to the hosts of
173// the restored snapshot, so a fixture that only names a durable snapshot still
174// keeps exercising the rule that observations intersect authoritative
175// membership.
176type testStore struct {
177	mu       sync.Mutex
178	loaded   ports.BrokerSnapshot
179	loadErr  error
180	hosts    ports.BrokerHosts
181	hostsErr error
182	stored   ports.BrokerSnapshot
183	storeErr error
184	writes   int
185	changes  chan struct{}
186}
187
188func (s *testStore) Load() (ports.BrokerSnapshot, error) {
189	s.mu.Lock()
190	defer s.mu.Unlock()
191	if s.loadErr != nil {
192		return ports.BrokerSnapshot{}, s.loadErr
193	}
194	return s.loaded.Clone(), nil
195}
196
197func (s *testStore) LoadHosts() (ports.BrokerHosts, error) {
198	s.mu.Lock()
199	defer s.mu.Unlock()
200	if s.hostsErr != nil {
201		return ports.BrokerHosts{}, s.hostsErr
202	}
203	if s.hosts.Revision != 0 || len(s.hosts.Hosts) > 0 {
204		return s.hosts, nil
205	}
206	authority := ports.BrokerHosts{Revision: 1}
207	for _, host := range s.loaded.Daemons {
208		authority.Hosts = append(authority.Hosts, ports.BrokerHostRecord{Registration: host.Registration, Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), host.Registration.Endpoint)})
209	}
210	return authority, nil
211}
212
213func (s *testStore) ReplaceHosts(expected uint64, hosts []ports.BrokerHostRecord) error {
214	s.mu.Lock()
215	defer s.mu.Unlock()
216	if s.hostsErr != nil {
217		return s.hostsErr
218	}
219	if s.hosts.Revision == 0 {
220		s.hosts.Revision = 1
221	}
222	if expected != s.hosts.Revision {
223		return ports.ErrBrokerHostConflict
224	}
225	s.hosts.Hosts = append([]ports.BrokerHostRecord(nil), hosts...)
226	s.hosts.Revision++
227	return nil
228}
229
230func (s *testStore) Close() error { return nil }
231
232func (s *testStore) Store(snapshot ports.BrokerSnapshot) error {
233	s.mu.Lock()
234	s.stored = snapshot.Clone()
235	s.writes++
236	err := s.storeErr
237	s.mu.Unlock()
238	s.signal()
239	return err
240}
241
242func (s *testStore) storedSnapshot() ports.BrokerSnapshot {
243	s.mu.Lock()
244	defer s.mu.Unlock()
245	return s.stored.Clone()
246}
247
248func (s *testStore) writeCount() int {
249	s.mu.Lock()
250	defer s.mu.Unlock()
251	return s.writes
252}
253
254func (s *testStore) setStoreErr(err error) {
255	s.mu.Lock()
256	defer s.mu.Unlock()
257	s.storeErr = err
258}
259
260func (s *testStore) signal() {
261	s.mu.Lock()
262	if s.changes == nil {
263		s.changes = make(chan struct{}, 1)
264	}
265	changes := s.changes
266	s.mu.Unlock()
267	select {
268	case changes <- struct{}{}:
269	default:
270	}
271}
272
273func (s *testStore) changesChan() chan struct{} {
274	s.mu.Lock()
275	defer s.mu.Unlock()
276	if s.changes == nil {
277		s.changes = make(chan struct{}, 1)
278	}
279	return s.changes
280}
281
282// blockingStore holds the first durable write inside Store so a test can prove
283// that publication never waits behind persistence.
284type blockingStore struct {
285	mu        sync.Mutex
286	revisions []ports.BrokerRevision
287	entered   chan ports.BrokerRevision
288	released  chan struct{}
289	first     sync.Once
290}
291
292func newBlockingStore() *blockingStore {
293	return &blockingStore{entered: make(chan ports.BrokerRevision, 1), released: make(chan struct{})}
294}
295
296func (s *blockingStore) Load() (ports.BrokerSnapshot, error) { return ports.BrokerSnapshot{}, nil }
297
298func (s *blockingStore) LoadHosts() (ports.BrokerHosts, error) {
299	return ports.BrokerHosts{Revision: 1}, nil
300}
301
302func (s *blockingStore) ReplaceHosts(uint64, []ports.BrokerHostRecord) error { return nil }
303
304func (s *blockingStore) Close() error { return nil }
305
306func (s *blockingStore) Store(snapshot ports.BrokerSnapshot) error {
307	s.mu.Lock()
308	s.revisions = append(s.revisions, snapshot.Revision)
309	s.mu.Unlock()
310	s.first.Do(func() {
311		select {
312		case s.entered <- snapshot.Revision:
313		default:
314		}
315		<-s.released
316	})
317	return nil
318}
319
320func (s *blockingStore) releaseFirst() { close(s.released) }
321
322func (s *blockingStore) writtenRevisions() []ports.BrokerRevision {
323	s.mu.Lock()
324	defer s.mu.Unlock()
325	return append([]ports.BrokerRevision(nil), s.revisions...)
326}
327
328// blockingCASStore parks the first durable membership CAS until it is released,
329// so a test can prove publication and Snapshot never wait behind membership
330// durability. Load, LoadHosts, and Store keep the race-safe testStore behavior,
331// and every later CAS is admitted without blocking.
332type blockingCASStore struct {
333	testStore
334	entered  chan struct{}
335	released chan struct{}
336	block    sync.Once
337	open     sync.Once
338}
339
340func newBlockingCASStore() *blockingCASStore {
341	return &blockingCASStore{entered: make(chan struct{}, 1), released: make(chan struct{})}
342}
343
344func (s *blockingCASStore) ReplaceHosts(expected uint64, hosts []ports.BrokerHostRecord) error {
345	s.block.Do(func() {
346		s.entered <- struct{}{}
347		<-s.released
348	})
349	return s.testStore.ReplaceHosts(expected, hosts)
350}
351
352// releaseCAS lets the parked CAS commit. It is idempotent so a failing test can
353// unblock the parked mutator from cleanup.
354func (s *blockingCASStore) releaseCAS() { s.open.Do(func() { close(s.released) }) }
355
356// testProbe scripts one observation per received call: the test owns the
357// completion, so a probe never depends on wall time.
358type testProbe struct {
359	calls    chan probeCall
360	canceled chan struct{}
361}
362
363func newTestProbe(capacity int) *testProbe {
364	return &testProbe{calls: make(chan probeCall, capacity), canceled: make(chan struct{}, 1)}
365}
366
367type probeCall struct {
368	registration domain.RemoteRegistration
369	result       chan probeAnswer
370}
371
372type probeAnswer struct {
373	snapshot ports.BrokerDaemonObservation
374	err      error
375}
376
377func (p *testProbe) Probe(ctx context.Context, registration domain.RemoteRegistration) (ports.BrokerDaemonObservation, error) {
378	call := probeCall{registration: registration, result: make(chan probeAnswer, 1)}
379	select {
380	case p.calls <- call:
381	case <-ctx.Done():
382		p.noteCanceled()
383		return ports.BrokerDaemonObservation{}, ctx.Err()
384	}
385	select {
386	case answer := <-call.result:
387		return answer.snapshot, answer.err
388	case <-ctx.Done():
389		p.noteCanceled()
390		return ports.BrokerDaemonObservation{}, ctx.Err()
391	}
392}
393
394func (p *testProbe) noteCanceled() {
395	select {
396	case p.canceled <- struct{}{}:
397	default:
398	}
399}
400
401// churnProbe is a ctx-honoring probe used by the churn regression test. It
402// tracks, per endpoint, how many probes were simultaneously live with an
403// uncancelled context: an attempt that is retired without cancelling its
404// context stays live and shows up as an overlap with its replacement.
405type churnProbe struct {
406	release    chan struct{}
407	released   sync.Once
408	startedSig chan struct{}
409
410	mu       sync.Mutex
411	contexts map[context.Context]string
412	maxLive  map[string]int
413	started  int
414}
415
416func newChurnProbe() *churnProbe {
417	return &churnProbe{
418		release:    make(chan struct{}),
419		startedSig: make(chan struct{}, 256),
420		contexts:   make(map[context.Context]string),
421		maxLive:    make(map[string]int),
422	}
423}
424
425func (p *churnProbe) Probe(ctx context.Context, registration domain.RemoteRegistration) (ports.BrokerDaemonObservation, error) {
426	p.mu.Lock()
427	live := 0
428	for other, endpoint := range p.contexts {
429		if endpoint == registration.Endpoint && other.Err() == nil {
430			live++
431		}
432	}
433	// The attempt is registered even when its context is already cancelled (a
434	// probe launched into a settling run), but only an uncancelled context is
435	// counted live: an attempt retired without cancelling its context stays in
436	// the map with Err()==nil and still overlaps its replacement.
437	p.contexts[ctx] = registration.Endpoint
438	if ctx.Err() == nil {
439		live++
440	}
441	if live > p.maxLive[registration.Endpoint] {
442		p.maxLive[registration.Endpoint] = live
443	}
444	p.started++
445	p.mu.Unlock()
446	select {
447	case p.startedSig <- struct{}{}:
448	default:
449	}
450	defer func() {
451		p.mu.Lock()
452		delete(p.contexts, ctx)
453		p.mu.Unlock()
454	}()
455	select {
456	case <-ctx.Done():
457	case <-p.release:
458	}
459	return ports.BrokerDaemonObservation{}, ctx.Err()
460}
461
462// openRelease lets every still-blocked probe return, so a failing test never
463// leaks probe goroutines.
464func (p *churnProbe) openRelease() { p.released.Do(func() { close(p.release) }) }
465
466func (p *churnProbe) startedCount() int {
467	p.mu.Lock()
468	defer p.mu.Unlock()
469	return p.started
470}
471
472func (p *churnProbe) maxLiveFor(endpoint string) int {
473	p.mu.Lock()
474	defer p.mu.Unlock()
475	return p.maxLive[endpoint]
476}
477
478func (p *churnProbe) observedEndpoints() []string {
479	p.mu.Lock()
480	defer p.mu.Unlock()
481	endpoints := make([]string, 0, len(p.maxLive))
482	for endpoint := range p.maxLive {
483		endpoints = append(endpoints, endpoint)
484	}
485	sort.Strings(endpoints)
486	return endpoints
487}
488
489func (p *churnProbe) inflight() int {
490	p.mu.Lock()
491	defer p.mu.Unlock()
492	return len(p.contexts)
493}
494
495func registration(t *testing.T, endpoint string, marker byte) domain.RemoteRegistration {
496	t.Helper()
497	var incarnation [16]byte
498	incarnation[0] = marker
499	result, err := domain.NewRemoteRegistration(endpoint, incarnation)
500	require.NoError(t, err)
501	return result
502}
503
504// canonicalRoute derives the exact route production would write for one test
505// policy and endpoint. Test policies only use the closed remote transport
506// vocabulary, so the mapping cannot fail; a fixture that hand-builds membership
507// therefore matches what AddHost or UpgradeBrokerHostRoutes would persist.
508func canonicalRoute(policy ports.BrokerPolicy, endpoint string) ports.BrokerRouteSpec {
509	route, err := ports.BrokerRouteForTransport(policy.Transport, endpoint)
510	if err != nil {
511		panic(err)
512	}
513	return route
514}
515
516// hostRecord builds one pinned authority record with the shared test policy and
517// the canonical route production derives from it. The projection-only setHosts
518// seam takes records now, so membership carries the same configured authority
519// (registration, policy, and route) as production.
520func hostRecord(reg domain.RemoteRegistration) ports.BrokerHostRecord {
521	return ports.BrokerHostRecord{Registration: reg, Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), reg.Endpoint)}
522}
523
524func hostRecords(registrations ...domain.RemoteRegistration) []ports.BrokerHostRecord {
525	records := make([]ports.BrokerHostRecord, 0, len(registrations))
526	for _, reg := range registrations {
527		records = append(records, hostRecord(reg))
528	}
529	return records
530}
531
532// hostSession builds one catalogue-valid session. The registry applies the
533// same durable projection rules as the store before adopting an observation, so
534// a fixture session must carry a non-zero lifecycle identity and a non-nil tab
535// list to count as durable.
536func hostSession(name string) catalogue.RemoteCatalogSession {
537	sum := sha256.Sum256([]byte("lifecycle:" + name))
538	lifecycle := domain.SessionLifecycleID{}
539	copy(lifecycle[:], sum[:])
540	return catalogue.RemoteCatalogSession{
541		LifecycleID: lifecycle,
542		Name:        name,
543		State:       catalogue.RemoteCatalogSessionUp,
544		Tabs:        []catalogue.RemoteCatalogTab{},
545	}
546}
547
548// hostSessions builds count minimal catalogue-valid sessions for the projection
549// session bound tests.
550func hostSessions(count int) []catalogue.RemoteCatalogSession {
551	sessions := make([]catalogue.RemoteCatalogSession, 0, count)
552	for index := 0; index < count; index++ {
553		sessions = append(sessions, hostSession(fmt.Sprintf("session-%d", index)))
554	}
555	return sessions
556}
557
558// newTestStore returns the minimal authoritative durable store: a valid
559// revision series over empty membership. NewRegistry requires a
560// BrokerHostStore, so every registry fixture names this store explicitly
561// instead of relying on an optional seam.
562func newTestStore() *testStore { return &testStore{} }
563
564// newTestRegistry builds a registry with exact scheduling and no logger so a
565// failure never depends on jitter.
566func newTestRegistry(t *testing.T, epoch ports.BrokerEpoch, store ports.BrokerHostStore, probe ports.BrokerHostProbe, clock *manualClock) *Registry {
567	t.Helper()
568	registry, err := NewRegistry(epoch, store, probe, clock, nil)
569	require.NoError(t, err)
570	registry.jitter = identityJitter
571	return registry
572}
573
574func startRegistry(t *testing.T, r *Registry) context.CancelFunc {
575	t.Helper()
576	r.SetDemand(true)
577	ctx, cancel := context.WithCancel(context.Background())
578	done := make(chan struct{})
579	go func() { defer close(done); r.Run(ctx) }()
580	t.Cleanup(func() { cancel(); <-done; r.SetDemand(false) })
581	return cancel
582}
583
584// registryTestWait bounds how long a test waits for an expected probe or
585// snapshot. It only delays a failing test, so it is generous for loaded
586// race-enabled CI runners.
587const registryTestWait = 5 * time.Second
588
589func receiveCall(t *testing.T, probe *testProbe) probeCall {
590	t.Helper()
591	select {
592	case call := <-probe.calls:
593		return call
594	case <-time.After(registryTestWait):
595		t.Fatal("timed out waiting for probe")
596		return probeCall{}
597	}
598}
599
600func requireNoProbeWithin(t *testing.T, probe *testProbe, window time.Duration) {
601	t.Helper()
602	timer := time.NewTimer(window)
603	defer timer.Stop()
604	select {
605	case call := <-probe.calls:
606		t.Fatalf("unexpected probe for %q", call.registration.Endpoint)
607	case <-timer.C:
608	}
609}
610
611func waitSnapshot(t *testing.T, r *Registry, predicate func(ports.BrokerSnapshot) bool) ports.BrokerSnapshot {
612	t.Helper()
613	sub := r.Subscribe()
614	defer sub.Close()
615	deadline := time.After(registryTestWait)
616	for {
617		snapshot := r.Snapshot()
618		if predicate(snapshot) {
619			return snapshot
620		}
621		select {
622		case <-sub.Changed():
623		case <-deadline:
624			t.Fatal("timed out waiting for snapshot")
625		}
626	}
627}
628
629func waitHost(t *testing.T, r *Registry, endpoint string, predicate func(ports.BrokerDaemonObservation) bool) ports.BrokerDaemonObservation {
630	t.Helper()
631	snapshot := waitSnapshot(t, r, func(snapshot ports.BrokerSnapshot) bool {
632		host, ok := snapshot.Find(endpoint)
633		return ok && predicate(host)
634	})
635	host, _ := snapshot.Find(endpoint)
636	return host
637}
638
639// requireHostStableWithin asserts a stale completion has no delayed effect:
640// the published host stays exactly as observed for a bounded window, and any
641// intervening publication fails immediately.
642func requireHostStableWithin(t *testing.T, r *Registry, endpoint string, want ports.BrokerDaemonObservation, window time.Duration) {
643	t.Helper()
644	sub := r.Subscribe()
645	defer sub.Close()
646	deadline := time.After(window)
647	for {
648		got, ok := r.Snapshot().Find(endpoint)
649		require.True(t, ok)
650		require.Equal(t, want, got)
651		select {
652		case <-sub.Changed():
653		case <-deadline:
654			return
655		}
656	}
657}
658
659// waitStored waits for a durable write satisfying predicate.
660func waitStored(t *testing.T, store *testStore, predicate func(ports.BrokerSnapshot) bool) ports.BrokerSnapshot {
661	t.Helper()
662	deadline := time.After(2 * time.Second)
663	for {
664		if snapshot := store.storedSnapshot(); predicate(snapshot) {
665			return snapshot
666		}
667		select {
668		case <-store.changesChan():
669		case <-deadline:
670			t.Fatal("timed out waiting for durable snapshot")
671			return ports.BrokerSnapshot{}
672		}
673	}
674}
675
676// waitWriterRevision waits until the serializer reports the revision durable.
677func waitWriterRevision(t *testing.T, r *Registry, revision ports.BrokerRevision) {
678	t.Helper()
679	deadline := time.Now().Add(2 * time.Second)
680	for {
681		r.store.mu.Lock()
682		written := r.store.written
683		r.store.mu.Unlock()
684		if written >= revision {
685			return
686		}
687		if time.Now().After(deadline) {
688			t.Fatalf("timed out waiting for durable revision %d", revision)
689		}
690		time.Sleep(time.Millisecond)
691	}
692}
693
694// waitForLog waits until captured log output contains want, so a persistence
695// fault is proven observable instead of silently dropped.
696func waitForLog(t *testing.T, logs *syncBuffer, want string) {
697	t.Helper()
698	deadline := time.Now().Add(2 * time.Second)
699	for !strings.Contains(logs.String(), want) {
700		if time.Now().After(deadline) {
701			t.Fatalf("timed out waiting for log %q in %q", want, logs.String())
702		}
703		time.Sleep(time.Millisecond)
704	}
705}
706
707// attemptFor returns the attempt token currently in flight for an endpoint,
708// or zero when no attempt is admitted.
709func attemptFor(r *Registry, endpoint string) uint64 {
710	r.mu.Lock()
711	defer r.mu.Unlock()
712	if attempt := r.inflight[endpoint]; attempt != nil {
713		return attempt.token
714	}
715	return 0
716}
717
718func TestRegistryRestoresPersistedSnapshotUnderFreshEpoch(t *testing.T) {
719	reg := registration(t, "example.test", 1)
720	sessions := []catalogue.RemoteCatalogSession{{
721		LifecycleID: domain.SessionLifecycleID{1},
722		Name:        "work",
723		State:       catalogue.RemoteCatalogSessionUp,
724		Tabs:        []catalogue.RemoteCatalogTab{{ID: "tab-1", Name: "shell"}},
725	}}
726	stored := ports.BrokerSnapshot{
727		Epoch:    7,
728		Revision: 9,
729		Daemons: []ports.BrokerDaemonObservation{{
730			Endpoint:       reg.Endpoint,
731			Registration:   reg,
732			Availability:   domain.RemoteAvailabilityReachable,
733			Checking:       true,
734			LastSuccess:    time.Unix(50, 0),
735			NextDue:        time.Unix(80, 0),
736			InventoryKnown: true,
737			Sessions:       sessions,
738		}},
739	}
740	store := &testStore{loaded: stored}
741	registry, err := NewRegistry(11, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
742	require.NoError(t, err)
743
744	snapshot := registry.Snapshot()
745	require.Equal(t, ports.BrokerEpoch(11), snapshot.Epoch)
746	// A restored broker process starts a fresh revision series: revisions
747	// never compare across epochs.
748	require.Equal(t, ports.BrokerRevision(1), snapshot.Revision)
749	require.Len(t, snapshot.Daemons, 1)
750	require.Empty(t, snapshot.Removed)
751
752	host := snapshot.Daemons[0]
753	require.True(t, host.Registration.Equal(reg))
754	require.Equal(t, domain.RemoteAvailabilityReachable, host.Availability)
755	require.Equal(t, time.Unix(80, 0), host.NextDue)
756	require.Equal(t, sessions, host.Sessions)
757	// In-flight observation state never survives a restart.
758	require.False(t, host.Checking)
759
760	// Restoring is read-only: the registry republishes under its own epoch
761	// without rewriting the durable store.
762	require.Equal(t, ports.BrokerRevision(0), store.storedSnapshot().Revision)
763	require.Zero(t, store.writeCount())
764}
765
766func TestRegistryRestoreRejectsInvalidInputs(t *testing.T) {
767	clock := newManualClock(time.Unix(100, 0))
768	reg := registration(t, "example.test", 1)
769
770	_, err := NewRegistry(0, newTestStore(), newTestProbe(1), clock, nil)
771	require.Error(t, err)
772	_, err = NewRegistry(1, newTestStore(), nil, clock, nil)
773	require.Error(t, err)
774	_, err = NewRegistry(1, newTestStore(), newTestProbe(1), nil, nil)
775	require.Error(t, err)
776
777	// The store is a required dependency: a literal nil store is rejected, and
778	// a typed nil behind the interface is rejected without panicking.
779	_, err = NewRegistry(1, nil, newTestProbe(1), clock, nil)
780	require.Error(t, err)
781	var typedNilStore *testStore
782	_, err = NewRegistry(1, typedNilStore, newTestProbe(1), clock, nil)
783	require.Error(t, err)
784
785	loadErr := errors.New("load failed")
786	_, err = NewRegistry(1, &testStore{loadErr: loadErr}, newTestProbe(1), clock, nil)
787	require.ErrorIs(t, err, loadErr)
788
789	_, err = NewRegistry(1, &testStore{loaded: ports.BrokerSnapshot{Epoch: 7}}, newTestProbe(1), clock, nil)
790	require.Error(t, err)
791
792	// A durable record without an epoch is inconsistent, never silently empty.
793	_, err = NewRegistry(1, &testStore{loaded: ports.BrokerSnapshot{Epoch: 0, Daemons: []ports.BrokerDaemonObservation{{Endpoint: reg.Endpoint, Registration: reg}}}}, newTestProbe(1), clock, nil)
794	require.Error(t, err)
795	_, err = NewRegistry(1, &testStore{loaded: ports.BrokerSnapshot{Epoch: 0, Removed: []ports.BrokerHostTombstone{{Endpoint: reg.Endpoint, Registration: reg, RetiredRevision: 3}}}}, newTestProbe(1), clock, nil)
796	require.Error(t, err)
797	_, err = NewRegistry(1, &testStore{loaded: ports.BrokerSnapshot{Epoch: 0, Revision: 4}}, newTestProbe(1), clock, nil)
798	require.Error(t, err)
799
800	empty, err := NewRegistry(1, &testStore{}, newTestProbe(1), clock, nil)
801	require.NoError(t, err)
802	require.Equal(t, ports.BrokerRevision(1), empty.Snapshot().Revision)
803	require.Empty(t, empty.Snapshot().Daemons)
804}
805
806func TestRegistryRestoreRejectsPersistedSnapshotFromItsOwnEpoch(t *testing.T) {
807	clock := newManualClock(time.Unix(100, 0))
808	reg := registration(t, "example.test", 1)
809
810	// A fresh broker process samples a fresh epoch and restarts its revision
811	// series at one, so a durable record from its own epoch can never be
812	// adopted: the restored series would not be monotonic and a stale record
813	// would be indistinguishable from current authority.
814	currentEpoch := ports.BrokerSnapshot{
815		Epoch:    7,
816		Revision: 9,
817		Daemons: []ports.BrokerDaemonObservation{{
818			Endpoint:     reg.Endpoint,
819			Registration: reg,
820			Availability: domain.RemoteAvailabilityReachable,
821		}},
822	}
823	_, err := NewRegistry(7, &testStore{loaded: currentEpoch}, newTestProbe(1), clock, nil)
824	require.Error(t, err)
825	require.Contains(t, err.Error(), "matches the new registry epoch")
826
827	// The same rejection holds for a tombstone-only and for an empty-hosts
828	// record from the current epoch.
829	_, err = NewRegistry(7, &testStore{loaded: ports.BrokerSnapshot{Epoch: 7, Revision: 1, Removed: []ports.BrokerHostTombstone{{
830		Endpoint: reg.Endpoint, Registration: reg, RetiredRevision: 1,
831	}}}}, newTestProbe(1), clock, nil)
832	require.Error(t, err)
833	_, err = NewRegistry(7, &testStore{loaded: ports.BrokerSnapshot{Epoch: 7, Revision: 1}}, newTestProbe(1), clock, nil)
834	require.Error(t, err)
835
836	// A different epoch is still adopted normally.
837	registry, err := NewRegistry(8, &testStore{loaded: currentEpoch}, newTestProbe(1), clock, nil)
838	require.NoError(t, err)
839	require.Len(t, registry.Snapshot().Daemons, 1)
840}
841
842func TestRegistrySetHostsRejectsInvalidRegistrations(t *testing.T) {
843	registry := newTestRegistry(t, 1, newTestStore(), newTestProbe(1), newManualClock(time.Unix(100, 0)))
844
845	reg := registration(t, "example.test", 1)
846	require.Error(t, registry.setHosts(hostRecords(reg, reg)))
847	require.Error(t, registry.setHosts(hostRecords(domain.RemoteRegistration{Endpoint: reg.Endpoint})))
848	require.Empty(t, registry.Snapshot().Daemons)
849}
850
851func TestRegistrySetHostsRejectsHostLimitWithoutMutation(t *testing.T) {
852	registry := newTestRegistry(t, 1, newTestStore(), newTestProbe(1), newManualClock(time.Unix(100, 0)))
853	reg := registration(t, "example.test", 1)
854	require.NoError(t, registry.setHosts(hostRecords(reg)))
855
856	oversized := make([]ports.BrokerHostRecord, 0, ports.BrokerMaxHosts+1)
857	for index := 0; index <= ports.BrokerMaxHosts; index++ {
858		oversized = append(oversized, hostRecord(registration(t, fmt.Sprintf("host-%d.test", index), byte(index+1))))
859	}
860	before := registry.Snapshot()
861	require.Error(t, registry.setHosts(oversized))
862	// The rejected call mutates nothing: same membership, same revision.
863	require.Equal(t, before, registry.Snapshot())
864
865	// The boundary itself is accepted.
866	require.NoError(t, registry.setHosts(oversized[:ports.BrokerMaxHosts]))
867	require.Len(t, registry.Snapshot().Daemons, ports.BrokerMaxHosts)
868}
869
870func TestRegistryUnchangedSetHostsIsNoOpAndReorderRepublishes(t *testing.T) {
871	clock := newManualClock(time.Unix(100, 0))
872	probe := newTestProbe(2)
873	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
874	first := registration(t, "example.test", 1)
875	second := registration(t, "other.test", 2)
876	require.NoError(t, registry.setHosts(hostRecords(first, second)))
877	startRegistry(t, registry)
878
879	// Every configured host is observed exactly once.
880	calls := make(map[string]probeCall, 2)
881	for range 2 {
882		call := receiveCall(t, probe)
883		calls[call.registration.Endpoint] = call
884	}
885	require.Len(t, calls, 2)
886	// Dispatch starts each probe while it holds the registry lock and publishes
887	// the in-flight projection at the end of the same critical section, so a
888	// received call is not yet a visible Checking state. Wait for the
889	// publication instead of assuming it has landed.
890	for endpoint := range calls {
891		waitHost(t, registry, endpoint, func(host ports.BrokerDaemonObservation) bool { return host.Checking })
892	}
893	before := registry.Snapshot()
894	for _, host := range before.Daemons {
895		require.Truef(t, host.Checking, "host %q is not checking: %+v", host.Endpoint, before.Daemons)
896	}
897
898	// Identical membership in the same order never republishes and never
899	// restarts an in-flight observation.
900	require.NoError(t, registry.setHosts(hostRecords(first, second)))
901	require.Equal(t, before, registry.Snapshot())
902	requireNoProbeWithin(t, probe, probeGrace)
903
904	// A reordering of identical records is a membership change: publication
905	// order and the presentation rank stamped from it follow durable membership
906	// authority, so the registry republishes both without cancelling the
907	// in-flight observation.
908	require.NoError(t, registry.setHosts(hostRecords(second, first)))
909	reordered := registry.Snapshot()
910	require.Greater(t, reordered.Revision, before.Revision)
911	require.Equal(t, []string{"other.test", "example.test"}, []string{reordered.Daemons[0].Endpoint, reordered.Daemons[1].Endpoint})
912	require.Equal(t, 0, reordered.Daemons[0].Rank)
913	require.Equal(t, 1, reordered.Daemons[1].Rank)
914	for _, host := range reordered.Daemons {
915		require.True(t, host.Checking)
916	}
917	requireNoProbeWithin(t, probe, probeGrace)
918
919	// Restoring the configured order restores the published order and ranks.
920	require.NoError(t, registry.setHosts(hostRecords(first, second)))
921	require.Equal(t, before.Daemons, registry.Snapshot().Daemons)
922	requireNoProbeWithin(t, probe, probeGrace)
923
924	// The in-flight observations still complete normally.
925	for endpoint, call := range calls {
926		call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
927		waitHost(t, registry, endpoint, func(host ports.BrokerDaemonObservation) bool {
928			return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
929		})
930	}
931}
932
933// TestRegistrySanitizesDerivedDisplayOrigin pins the boundary between routing
934// authority and presentation: a target the routing validator accepts may carry
935// bidi controls or line separators, and a hint the registry derives from it must
936// never render them. The sanitized hint is published, satisfies the contract
937// gate, and leaves the host observable.
938func TestRegistrySanitizesDerivedDisplayOrigin(t *testing.T) {
939	tests := []struct {
940		name     string
941		endpoint string
942		want     string
943	}{
944		{name: "login prefix strips the bidi rune from the origin", endpoint: "user@ho\u202est.test", want: "host.test"},
945		{name: "bare target drops the bidi rune", endpoint: "ho\u202est.test", want: "host.test"},
946		{name: "right-to-left mark dropped", endpoint: "user@ho\u200fst.test", want: "host.test"},
947		{name: "ordinary target unchanged", endpoint: "user@host.test", want: "host.test"},
948	}
949	for _, tc := range tests {
950		t.Run(tc.name, func(t *testing.T) {
951			clock := newManualClock(time.Unix(100, 0))
952			probe := newTestProbe(1)
953			registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
954			require.NoError(t, registry.setHosts(hostRecords(registration(t, tc.endpoint, 1))))
955
956			host := registry.Snapshot().Daemons[0]
957			require.Equal(t, tc.endpoint, host.Endpoint, "the routing target stays authoritative")
958			require.Equal(t, tc.want, host.DisplayOrigin)
959			require.NoError(t, host.Validate())
960			require.NoError(t, registry.Snapshot().Validate())
961
962			// Sanitizing the hint never makes the host unobservable.
963			startRegistry(t, registry)
964			call := receiveCall(t, probe)
965			call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
966			waitHost(t, registry, tc.endpoint, func(host ports.BrokerDaemonObservation) bool {
967				return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
968			})
969		})
970	}
971}
972
973// TestRegistryRefusesUnpublishableMembership pins the fail-closed membership
974// boundary: a target whose derived hint holds only refused runes has no
975// displayable origin, and membership that cannot be published is refused where
976// authority is owned instead of freezing every later publication.
977func TestRegistryRefusesUnpublishableMembership(t *testing.T) {
978	registry := newTestRegistry(t, 1, newTestStore(), newTestProbe(1), newManualClock(time.Unix(100, 0)))
979	err := registry.setHosts(hostRecords(registration(t, "user@\u202e", 1)))
980	require.ErrorContains(t, err, "not publishable")
981	require.Empty(t, registry.Snapshot().Daemons)
982}
983
984// TestRegistryRefusesInvalidPublication pins the last-resort publication guard:
985// a projection that reached the live set unpublished is never published, because
986// a snapshot the wire refuses aborts every client connection on subscribe, and a
987// refused publication never advances the revision series.
988func TestRegistryRefusesInvalidPublication(t *testing.T) {
989	registry := newTestRegistry(t, 1, newTestStore(), newTestProbe(1), newManualClock(time.Unix(100, 0)))
990	require.NoError(t, registry.setHosts(hostRecords(registration(t, "example.test", 1))))
991	before := registry.Snapshot()
992
993	// Inject behind the membership boundary, which now refuses such a record.
994	registry.mu.Lock()
995	broken := before.Daemons[0].Clone()
996	broken.DisplayOrigin = ""
997	registry.hosts[broken.Endpoint] = broken
998	registry.publishLocked(false)
999	registry.mu.Unlock()
1000
1001	require.Equal(t, before, registry.Snapshot())
1002}
1003
1004func TestRegistrySnapshotIsImmutable(t *testing.T) {
1005	clock := newManualClock(time.Unix(100, 0))
1006	probe := newTestProbe(1)
1007	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1008	reg := registration(t, "example.test", 1)
1009	require.NoError(t, registry.setHosts(hostRecords(reg)))
1010	startRegistry(t, registry)
1011
1012	call := receiveCall(t, probe)
1013	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1014		Availability:   domain.RemoteAvailabilityReachable,
1015		InventoryKnown: true,
1016		Sessions: []catalogue.RemoteCatalogSession{{
1017			LifecycleID: domain.SessionLifecycleID{1},
1018			Name:        "work",
1019			State:       catalogue.RemoteCatalogSessionUp,
1020			Tabs:        []catalogue.RemoteCatalogTab{{ID: "tab-1", Name: "shell"}},
1021		}},
1022	}}
1023	observed := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1024		return !host.Checking && len(host.Sessions) == 1
1025	})
1026	require.Equal(t, "work", observed.Sessions[0].Name)
1027
1028	// Mutating a returned projection never reaches the registry.
1029	observed.Endpoint = "mutated"
1030	observed.Sessions[0].Name = "mutated"
1031	observed.Sessions[0].Tabs[0].Name = "mutated"
1032	current := registry.Snapshot()
1033	require.Equal(t, reg.Endpoint, current.Daemons[0].Endpoint)
1034	require.Equal(t, "work", current.Daemons[0].Sessions[0].Name)
1035	require.Equal(t, "shell", current.Daemons[0].Sessions[0].Tabs[0].Name)
1036
1037	// A later publication never mutates a snapshot already handed out.
1038	earlier := registry.Snapshot()
1039	registry.RequestProbe(reg.Endpoint)
1040	next := receiveCall(t, probe)
1041	next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1042		Availability:   domain.RemoteAvailabilityReachable,
1043		InventoryKnown: true,
1044		Sessions:       []catalogue.RemoteCatalogSession{hostSession("other")},
1045	}}
1046	waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1047		return !host.Checking && len(host.Sessions) == 1 && host.Sessions[0].Name == "other"
1048	})
1049	require.Equal(t, "work", earlier.Daemons[0].Sessions[0].Name)
1050}
1051
1052func TestRegistrySchedulesFreshObservationAfterSuccess(t *testing.T) {
1053	start := time.Unix(100, 0)
1054	clock := newManualClock(start)
1055	probe := newTestProbe(1)
1056	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1057	reg := registration(t, "example.test", 1)
1058	require.NoError(t, registry.setHosts(hostRecords(reg)))
1059	startRegistry(t, registry)
1060
1061	call := receiveCall(t, probe)
1062	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1063
1064	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1065		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1066	})
1067	require.Equal(t, start, host.LastAttempt)
1068	require.Equal(t, start, host.LastSuccess)
1069	require.Equal(t, start.Add(defaultDemandFreshForRemote), host.NextDue)
1070	require.Zero(t, host.ConsecutiveFailures)
1071	require.Zero(t, host.FailureEpisode)
1072
1073	// A fresh host is not observed again before its scheduled deadline.
1074	clock.Advance(defaultDemandFreshForRemote - time.Nanosecond)
1075	requireNoProbeWithin(t, probe, probeGrace)
1076	clock.Advance(time.Nanosecond)
1077	next := receiveCall(t, probe)
1078	require.True(t, next.registration.Equal(reg))
1079}
1080
1081// TestRegistryAppliesSubscriptionDemandFreshness checks that apply/applyLocal
1082// schedule NextDue on RegistryConfig's demand cadence while Registry.SetDemand
1083// reports a live client subscription. It exercises both the remote and the local
1084// observation paths on the same registry, since service.go's Subscribe/Close
1085// wiring feeds one shared demand counter for both.
1086func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) {
1087	const (
1088		demandLocal  = 250 * time.Millisecond
1089		demandRemote = 500 * time.Millisecond
1090	)
1091	tests := []struct {
1092		name      string
1093		local     bool
1094		demand    bool
1095		wantAfter time.Duration
1096	}{
1097		{name: "remote demand uses faster cadence", local: false, demand: true, wantAfter: demandRemote},
1098		{name: "local demand uses faster cadence", local: true, demand: true, wantAfter: demandLocal},
1099	}
1100
1101	for _, tt := range tests {
1102		t.Run(tt.name, func(t *testing.T) {
1103			start := time.Unix(100, 0)
1104			clock := newManualClock(start)
1105			remoteProbe := newTestProbe(1)
1106			localProbe := newScriptedLocalProbe(1)
1107			registry, err := NewRegistryWithConfig(1, newTestStore(), remoteProbe, clock, nil, RegistryConfig{
1108				Local:                &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: localProbe},
1109				DemandFreshForLocal:  demandLocal,
1110				DemandFreshForRemote: demandRemote,
1111			})
1112			require.NoError(t, err)
1113			registry.jitter = identityJitter
1114			reg := registration(t, "example.test", 1)
1115			require.NoError(t, registry.setHosts(hostRecords(reg)))
1116			startRegistry(t, registry)
1117
1118			if tt.demand {
1119				registry.SetDemand(true)
1120				t.Cleanup(func() { registry.SetDemand(false) })
1121			}
1122
1123			if tt.local {
1124				call := receiveLocalCall(t, localProbe)
1125				call.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
1126					Identity:        "local-daemon",
1127					Incarnation:     ports.BrokerDaemonIncarnation{9},
1128					ProtocolVersion: protocol.Version,
1129					Availability:    domain.RemoteAvailabilityReachable,
1130				}}
1131				host := waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool {
1132					return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
1133				})
1134				require.Equal(t, start.Add(tt.wantAfter), host.NextDue)
1135				return
1136			}
1137			call := receiveCall(t, remoteProbe)
1138			call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1139			host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1140				return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1141			})
1142			require.Equal(t, start.Add(tt.wantAfter), host.NextDue)
1143		})
1144	}
1145}
1146
1147// TestRegistrySetDemandIsCountedNotBoolean checks that subscriptions balance:
1148// observation stops after the last closes and resumes immediately on reopen.
1149func TestRegistrySetDemandIsCountedNotBoolean(t *testing.T) {
1150	start := time.Unix(100, 0)
1151	clock := newManualClock(start)
1152	probe := newTestProbe(2)
1153	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1154	registry.demandFreshForRemote = 500 * time.Millisecond
1155	reg := registration(t, "example.test", 1)
1156	require.NoError(t, registry.setHosts(hostRecords(reg)))
1157	startRegistry(t, registry)
1158
1159	// Two subscriptions open; only one closes, so demand must stay active.
1160	registry.SetDemand(true)
1161	registry.SetDemand(true)
1162	registry.SetDemand(false)
1163
1164	call := receiveCall(t, probe)
1165	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1166	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1167		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1168	})
1169	require.Equal(t, start.Add(500*time.Millisecond), host.NextDue, "one remaining demand unit must still use the faster cadence")
1170
1171	// The last subscription closes: no scheduled observation runs for nobody.
1172	registry.SetDemand(false)
1173	registry.SetDemand(false) // balance the fixture's first subscription
1174	clock.Advance(time.Hour)
1175	requireNoProbeWithin(t, probe, probeGrace)
1176	registry.SetDemand(true)
1177	next := receiveCall(t, probe)
1178	next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1179	host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1180		return !host.Checking && host.LastSuccess.Equal(clock.Now())
1181	})
1182	require.Equal(t, clock.Now().Add(500*time.Millisecond), host.NextDue)
1183}
1184
1185func TestRegistryDemandOverridesFreshSchedule(t *testing.T) {
1186	start := time.Unix(100, 0)
1187	clock := newManualClock(start)
1188	probe := newTestProbe(1)
1189	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1190	reg := registration(t, "example.test", 1)
1191	require.NoError(t, registry.setHosts(hostRecords(reg)))
1192	startRegistry(t, registry)
1193
1194	call := receiveCall(t, probe)
1195	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1196	waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1197		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1198	})
1199
1200	// Explicit demand is admitted immediately, before the fresh deadline.
1201	registry.RequestProbe(reg.Endpoint)
1202	forced := receiveCall(t, probe)
1203	require.True(t, forced.registration.Equal(reg))
1204	require.Equal(t, start, clock.Now())
1205}
1206
1207func TestRegistryCoalescesConcurrentDemand(t *testing.T) {
1208	start := time.Unix(100, 0)
1209	clock := newManualClock(start)
1210	probe := newTestProbe(2)
1211	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1212	reg := registration(t, "example.test", 1)
1213	require.NoError(t, registry.setHosts(hostRecords(reg)))
1214	startRegistry(t, registry)
1215
1216	call := receiveCall(t, probe)
1217	for range 10 {
1218		registry.RequestProbe(reg.Endpoint)
1219	}
1220	select {
1221	case duplicate := <-probe.calls:
1222		t.Fatalf("concurrent demand started a duplicate probe for %q", duplicate.registration.Endpoint)
1223	default:
1224	}
1225	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1226	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1227		return host.Availability == domain.RemoteAvailabilityReachable && host.LastSuccess.Equal(start)
1228	})
1229	require.Equal(t, start.Add(defaultDemandFreshForRemote), host.NextDue)
1230
1231	// The single coalesced follow-up is admitted once the first completes.
1232	followUp := receiveCall(t, probe)
1233	require.True(t, followUp.registration.Equal(reg))
1234	followUp.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1235	waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1236		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1237	})
1238}
1239
1240func TestRegistryIgnoresDemandForUnknownEndpoint(t *testing.T) {
1241	clock := newManualClock(time.Unix(100, 0))
1242	probe := newTestProbe(1)
1243	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1244	reg := registration(t, "example.test", 1)
1245	require.NoError(t, registry.setHosts(hostRecords(reg)))
1246	registry.RequestProbe("unknown.test")
1247	startRegistry(t, registry)
1248
1249	call := receiveCall(t, probe)
1250	require.True(t, call.registration.Equal(reg))
1251	// No probe is ever scheduled for an endpoint outside configured membership.
1252	requireNoProbeWithin(t, probe, probeGrace)
1253}
1254
1255func TestRegistryPublishesReachabilityAndPersists(t *testing.T) {
1256	start := time.Unix(100, 0)
1257	clock := newManualClock(start)
1258	probe := newTestProbe(1)
1259	store := &testStore{}
1260	registry := newTestRegistry(t, 3, store, probe, clock)
1261	reg := registration(t, "example.test", 1)
1262	require.NoError(t, registry.setHosts(hostRecords(reg)))
1263	startRegistry(t, registry)
1264
1265	call := receiveCall(t, probe)
1266	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1267		Availability:   domain.RemoteAvailabilityReachable,
1268		InventoryKnown: true,
1269		Sessions:       []catalogue.RemoteCatalogSession{hostSession("work")},
1270	}}
1271	snapshot := waitSnapshot(t, registry, func(snapshot ports.BrokerSnapshot) bool {
1272		host, ok := snapshot.Find(reg.Endpoint)
1273		return ok && !host.Checking && host.Availability == domain.RemoteAvailabilityReachable && len(host.Sessions) == 1
1274	})
1275
1276	stored := waitStored(t, store, func(stored ports.BrokerSnapshot) bool {
1277		return stored.Revision >= snapshot.Revision && len(stored.Daemons) == 1
1278	})
1279	require.Equal(t, ports.BrokerEpoch(3), stored.Epoch)
1280	require.Len(t, stored.Daemons, 1)
1281	require.Equal(t, "work", stored.Daemons[0].Sessions[0].Name)
1282	require.Equal(t, reg.Endpoint, stored.Daemons[0].Endpoint)
1283	require.False(t, stored.Daemons[0].Checking)
1284}
1285
1286func TestRegistryNonReachableOutcomesUseTypedFailureBackoff(t *testing.T) {
1287	dialErr := errors.New("dial failed")
1288	tests := []struct {
1289		name             string
1290		answer           probeAnswer
1291		wantAvailability domain.RemoteAvailability
1292		wantKind         domain.RemoteFailureKind
1293		wantErr          error
1294	}{
1295		{
1296			name:             "unreachable without error",
1297			answer:           probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}},
1298			wantAvailability: domain.RemoteAvailabilityUnreachable,
1299			wantKind:         domain.RemoteFailureTransport,
1300		},
1301		{
1302			name:             "unknown outcome is not reachability",
1303			answer:           probeAnswer{snapshot: ports.BrokerDaemonObservation{}},
1304			wantAvailability: domain.RemoteAvailabilityUnreachable,
1305			wantKind:         domain.RemoteFailureTransport,
1306		},
1307		{
1308			name:             "incompatible",
1309			answer:           probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityIncompatible}},
1310			wantAvailability: domain.RemoteAvailabilityIncompatible,
1311			wantKind:         domain.RemoteFailureIncompatible,
1312		},
1313		{
1314			name:             "authentication",
1315			answer:           probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityAuthFailed}},
1316			wantAvailability: domain.RemoteAvailabilityAuthFailed,
1317			wantKind:         domain.RemoteFailureAuthentication,
1318		},
1319		{
1320			name:             "invalid response",
1321			answer:           probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityInvalidResponse}},
1322			wantAvailability: domain.RemoteAvailabilityInvalidResponse,
1323			wantKind:         domain.RemoteFailureInvalidResponse,
1324		},
1325		{
1326			name:             "no daemon after a prior success",
1327			answer:           probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityNoDaemon}},
1328			wantAvailability: domain.RemoteAvailabilityNoDaemon,
1329			wantKind:         domain.RemoteFailureNone,
1330		},
1331		{
1332			name:             "transport error",
1333			answer:           probeAnswer{err: dialErr},
1334			wantAvailability: domain.RemoteAvailabilityUnreachable,
1335			wantKind:         domain.RemoteFailureTransport,
1336			wantErr:          dialErr,
1337		},
1338		{
1339			name:             "deadline exceeded",
1340			answer:           probeAnswer{err: context.DeadlineExceeded},
1341			wantAvailability: domain.RemoteAvailabilityUnreachable,
1342			wantKind:         domain.RemoteFailureTimeout,
1343			wantErr:          context.DeadlineExceeded,
1344		},
1345	}
1346	for _, tt := range tests {
1347		t.Run(tt.name, func(t *testing.T) {
1348			start := time.Unix(100, 0)
1349			clock := newManualClock(start)
1350			probe := newTestProbe(1)
1351			registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1352			reg := registration(t, "example.test", 1)
1353			require.NoError(t, registry.setHosts(hostRecords(reg)))
1354			startRegistry(t, registry)
1355			if tt.wantAvailability == domain.RemoteAvailabilityNoDaemon {
1356				reachable := probeAnswer{snapshot: ports.BrokerDaemonObservation{
1357					Availability: domain.RemoteAvailabilityReachable, Identity: ports.BrokerDaemonIdentity("remote-daemon"),
1358					Incarnation: ports.BrokerDaemonIncarnation{1}, ProtocolVersion: protocol.Version,
1359					InventoryKnown: true, Sessions: []catalogue.RemoteCatalogSession{hostSession("old")},
1360				}}
1361				call := receiveCall(t, probe)
1362				call.result <- reachable
1363				waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1364					return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable && host.Identity == ports.BrokerDaemonIdentity("remote-daemon")
1365				})
1366				clock.Advance(5 * time.Second)
1367				registry.RequestProbe(reg.Endpoint)
1368				call = receiveCall(t, probe)
1369				call.result <- reachable
1370				waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1371					return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1372				})
1373				clock.Advance(20 * time.Second)
1374				registry.RequestProbe(reg.Endpoint)
1375			}
1376
1377			call := receiveCall(t, probe)
1378			call.result <- tt.answer
1379			host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1380				return !host.Checking && (host.ConsecutiveFailures == 1 || tt.wantAvailability == domain.RemoteAvailabilityNoDaemon && host.Availability == domain.RemoteAvailabilityNoDaemon)
1381			})
1382			require.Equal(t, tt.wantAvailability, host.Availability)
1383			require.Equal(t, tt.wantKind, host.LastFailure.Kind)
1384			if tt.wantAvailability == domain.RemoteAvailabilityNoDaemon {
1385				require.Zero(t, host.ConsecutiveFailures)
1386				require.Zero(t, host.FailureEpisode)
1387				require.Equal(t, start.Add(25*time.Second), host.LastSuccess)
1388				require.Equal(t, start.Add(40*time.Second), host.NextDue)
1389				require.Empty(t, host.Identity)
1390				require.True(t, host.Incarnation.IsZero())
1391				require.Zero(t, host.ProtocolVersion)
1392				require.Zero(t, host.Capabilities)
1393				require.False(t, host.InventoryKnown)
1394				require.Empty(t, host.Sessions)
1395				require.Equal(t, domain.RemoteFailureNone, host.LastFailure.Kind)
1396			} else {
1397				require.Equal(t, uint(1), host.ConsecutiveFailures)
1398				require.Equal(t, uint64(1), host.FailureEpisode)
1399				// No observation succeeded yet, so there is no last success.
1400				require.True(t, host.LastSuccess.IsZero())
1401				// A watched host retries on the demand cadence.
1402				require.Equal(t, start.Add(defaultDemandFreshForRemote), host.NextDue)
1403			}
1404			if tt.wantErr != nil {
1405				require.ErrorIs(t, host.LastFailure, tt.wantErr)
1406			} else {
1407				require.Nil(t, host.LastFailure.Err)
1408			}
1409		})
1410	}
1411}
1412
1413func TestRegistryClassifiesOversizedProbeProjectionAsInvalidResponse(t *testing.T) {
1414	start := time.Unix(100, 0)
1415	clock := newManualClock(start)
1416	probe := newTestProbe(1)
1417	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1418	reg := registration(t, "example.test", 1)
1419	require.NoError(t, registry.setHosts(hostRecords(reg)))
1420	startRegistry(t, registry)
1421	sub := registry.Subscribe()
1422	defer sub.Close()
1423
1424	// A reachable observation establishes the last-known inventory.
1425	call := receiveCall(t, probe)
1426	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1427		Availability:   domain.RemoteAvailabilityReachable,
1428		InventoryKnown: true,
1429		Sessions:       []catalogue.RemoteCatalogSession{hostSession("work")},
1430	}}
1431	waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1432		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1433	})
1434
1435	// A projection carrying one session more than the durable session bound is
1436	// a malformed response: it is classified as an invalid response and never
1437	// reaches publication, because the store enforces the same catalogue rule.
1438	// The loop checks every observed publication, so an invalid snapshot can
1439	// never pass unnoticed.
1440	clock.Advance(defaultFreshFor)
1441	call = receiveCall(t, probe)
1442	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1443		Availability:   domain.RemoteAvailabilityReachable,
1444		InventoryKnown: true,
1445		Sessions:       hostSessions(ports.BrokerMaxSessionsPerHost + 1),
1446	}}
1447	deadline := time.After(time.Second)
1448	var host ports.BrokerDaemonObservation
1449	for {
1450		snapshot := registry.Snapshot()
1451		require.NoError(t, snapshot.Validate())
1452		current, ok := snapshot.Find(reg.Endpoint)
1453		if ok && !current.Checking && current.ConsecutiveFailures == 1 {
1454			host = current
1455			break
1456		}
1457		select {
1458		case <-sub.Changed():
1459		case <-deadline:
1460			t.Fatal("timed out waiting for the invalid-response failure")
1461		}
1462	}
1463	require.Equal(t, domain.RemoteAvailabilityInvalidResponse, host.Availability)
1464	require.Equal(t, domain.RemoteFailureInvalidResponse, host.LastFailure.Kind)
1465	// The last-known inventory is retained: a malformed projection never
1466	// replaces it.
1467	require.Len(t, host.Sessions, 1)
1468	require.Equal(t, "work", host.Sessions[0].Name)
1469	require.Equal(t, clock.Now().Add(defaultDemandFreshForRemote), host.NextDue)
1470}
1471
1472func TestRegistryAcceptsProbeProjectionAtSessionBound(t *testing.T) {
1473	start := time.Unix(100, 0)
1474	clock := newManualClock(start)
1475	probe := newTestProbe(1)
1476	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1477	reg := registration(t, "example.test", 1)
1478	require.NoError(t, registry.setHosts(hostRecords(reg)))
1479	startRegistry(t, registry)
1480
1481	// The bound itself is valid: exactly BrokerMaxSessionsPerHost sessions are
1482	// accepted and published, and the resulting snapshot validates.
1483	call := receiveCall(t, probe)
1484	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1485		Availability:   domain.RemoteAvailabilityReachable,
1486		InventoryKnown: true,
1487		Sessions:       hostSessions(ports.BrokerMaxSessionsPerHost),
1488	}}
1489	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1490		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1491	})
1492	require.Len(t, host.Sessions, ports.BrokerMaxSessionsPerHost)
1493	require.Zero(t, host.ConsecutiveFailures)
1494	snapshot := registry.Snapshot()
1495	require.NoError(t, snapshot.Validate())
1496	require.Len(t, snapshot.Daemons[0].Sessions, ports.BrokerMaxSessionsPerHost)
1497}
1498
1499// TestRegistryRejectsNonDurableObservation proves the registry and the durable
1500// store agree on what may be persisted: a reachable projection the store would
1501// reject is classified as an invalid response, so the last-known inventory
1502// stays authoritative instead of a snapshot that can never be written.
1503func TestRegistryRejectsNonDurableObservation(t *testing.T) {
1504	start := time.Unix(100, 0)
1505	clock := newManualClock(start)
1506	probe := newTestProbe(1)
1507	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1508	reg := registration(t, "example.test", 1)
1509	require.NoError(t, registry.setHosts(hostRecords(reg)))
1510	startRegistry(t, registry)
1511
1512	call := receiveCall(t, probe)
1513	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1514		Availability:   domain.RemoteAvailabilityReachable,
1515		InventoryKnown: true,
1516		Sessions:       []catalogue.RemoteCatalogSession{hostSession("work")},
1517	}}
1518	waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1519		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1520	})
1521
1522	// A probe answer that violates the catalogue rules the store enforces (here
1523	// a zero lifecycle identity and an absent tab list) is a malformed response.
1524	clock.Advance(defaultFreshFor)
1525	call = receiveCall(t, probe)
1526	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1527		Availability:   domain.RemoteAvailabilityReachable,
1528		InventoryKnown: true,
1529		Sessions:       []catalogue.RemoteCatalogSession{{Name: "work", State: catalogue.RemoteCatalogSessionUp}},
1530	}}
1531	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1532		return !host.Checking && host.ConsecutiveFailures == 1
1533	})
1534	require.Equal(t, domain.RemoteAvailabilityInvalidResponse, host.Availability)
1535	require.Equal(t, domain.RemoteFailureInvalidResponse, host.LastFailure.Kind)
1536	require.Len(t, host.Sessions, 1)
1537	require.Equal(t, "work", host.Sessions[0].Name)
1538	require.Equal(t, start, host.LastSuccess)
1539	snapshot := registry.Snapshot()
1540	require.NoError(t, snapshot.Validate())
1541	require.NoError(t, ports.ValidateDurableHostProjection(snapshot.Daemons[0]))
1542}
1543
1544// TestRegistryRevisionOverflowFailsClosed pins the revision series: zero is not
1545// a valid revision and revisions never regress, so an exhausted counter fails
1546// closed instead of wrapping. The last valid snapshot stays published, the
1547// registry is poisoned, and every mutation path reports a typed error rather
1548// than emitting revision zero or a revision below the newest one, either of
1549// which would silently freeze durable persistence.
1550func TestRegistryRevisionOverflowFailsClosed(t *testing.T) {
1551	logger, logs := testLogger()
1552	registry, err := NewRegistry(1, newTestStore(), newTestProbe(1), newManualClock(time.Unix(100, 0)), logger)
1553	require.NoError(t, err)
1554	first := registration(t, "example.test", 1)
1555	require.NoError(t, registry.setHosts(hostRecords(first)))
1556	before := registry.Snapshot()
1557	require.NoError(t, before.Validate())
1558	require.Equal(t, ports.BrokerRevision(2), before.Revision)
1559	authority := registry.authority
1560
1561	// The next publication would overflow the series to zero.
1562	registry.revision = ^ports.BrokerRevision(0)
1563
1564	// Both mutation paths refuse the change before touching the store or the
1565	// projection, so neither durable authority nor the publication series moves.
1566	require.ErrorIs(t, registry.setHosts(hostRecords(registration(t, "other.test", 2))), ports.ErrBrokerRevisionExhausted)
1567	require.ErrorIs(t, registry.ReplaceHosts([]ports.BrokerHostRecord{{Registration: first, Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), first.Endpoint)}}), ports.ErrBrokerRevisionExhausted)
1568	require.Equal(t, authority, registry.authority)
1569
1570	// The publication path itself fails closed too: a direct attempt neither
1571	// wraps the counter nor replaces the last valid snapshot.
1572	registry.mu.Lock()
1573	registry.publishLocked(false)
1574	registry.mu.Unlock()
1575
1576	after := registry.Snapshot()
1577	require.Equal(t, before, after)
1578	require.NoError(t, after.Validate())
1579	require.Greater(t, after.Revision, ports.BrokerRevision(0))
1580	require.Equal(t, ^ports.BrokerRevision(0), registry.revision)
1581	require.Equal(t, first, after.Daemons[0].Registration)
1582	// The refusal is observable: an exhausted series logs instead of failing
1583	// silently behind a frozen snapshot.
1584	require.Contains(t, logs.String(), "revision series exhausted")
1585}
1586
1587func TestRegistryPreservesLastSuccessAndOwnsFailureEpisode(t *testing.T) {
1588	start := time.Unix(100, 0)
1589	clock := newManualClock(start)
1590	probe := newTestProbe(1)
1591	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1592	reg := registration(t, "example.test", 1)
1593	require.NoError(t, registry.setHosts(hostRecords(reg)))
1594	startRegistry(t, registry)
1595
1596	// A successful observation owns the success time and opens no episode.
1597	call := receiveCall(t, probe)
1598	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1599		Availability:   domain.RemoteAvailabilityReachable,
1600		InventoryKnown: true,
1601		Sessions:       []catalogue.RemoteCatalogSession{hostSession("work")},
1602	}}
1603	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1604		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1605	})
1606	require.Equal(t, start, host.LastSuccess)
1607	require.Zero(t, host.FailureEpisode)
1608
1609	// A failure preserves the last success and opens exactly one episode.
1610	clock.Advance(defaultFreshFor)
1611	call = receiveCall(t, probe)
1612	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
1613		Availability:   domain.RemoteAvailabilityIncompatible,
1614		InventoryKnown: true,
1615		Sessions:       []catalogue.RemoteCatalogSession{{Name: "stale", State: catalogue.RemoteCatalogSessionUp}},
1616	}}
1617	host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1618		return !host.Checking && host.ConsecutiveFailures == 1
1619	})
1620	require.Equal(t, start, host.LastSuccess)
1621	require.Equal(t, uint64(1), host.FailureEpisode)
1622	// The last-known inventory is retained: a failed observation never
1623	// replaces it.
1624	require.Equal(t, "work", host.Sessions[0].Name)
1625
1626	// Further failures in the same streak preserve the episode identity so
1627	// notice policy does not re-notify.
1628	clock.Advance(defaultRetryBase)
1629	call = receiveCall(t, probe)
1630	call.result <- probeAnswer{err: errors.New("offline")}
1631	host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1632		return !host.Checking && host.ConsecutiveFailures == 2
1633	})
1634	require.Equal(t, start, host.LastSuccess)
1635	require.Equal(t, uint64(1), host.FailureEpisode)
1636
1637	// Recovery clears the streak, keeps the episode, and advances the success.
1638	clock.Advance(2 * defaultRetryBase)
1639	recovered := clock.Now()
1640	call = receiveCall(t, probe)
1641	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1642	host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1643		return !host.Checking && host.ConsecutiveFailures == 0
1644	})
1645	require.Equal(t, recovered, host.LastSuccess)
1646	require.Equal(t, uint64(1), host.FailureEpisode)
1647	require.Equal(t, domain.RemoteFailureNone, host.LastFailure.Kind)
1648	require.Nil(t, host.LastFailure.Err)
1649}
1650
1651func TestRegistryDemandCapsFailureBackoff(t *testing.T) {
1652	start := time.Unix(100, 0)
1653	clock := newManualClock(start)
1654	probe := newTestProbe(1)
1655	reg := registration(t, "demand.test", 1)
1656	store := newTestStore()
1657	require.NoError(t, store.ReplaceHosts(1, hostRecords(reg)))
1658	registry := newTestRegistry(t, 1, store, probe, clock)
1659	startRegistry(t, registry)
1660	registry.SetDemand(true)
1661	t.Cleanup(func() { registry.SetDemand(false) })
1662
1663	for failure := uint(1); failure <= 9; failure++ {
1664		call := receiveCall(t, probe)
1665		call.result <- probeAnswer{err: errors.New("daemon not available")}
1666		host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1667			return !host.Checking && host.ConsecutiveFailures == failure
1668		})
1669		require.LessOrEqual(t, host.NextDue.Sub(clock.Now()), demandRetryMax, "demanded retry gap must never exceed the demand retry cap")
1670		clock.Advance(host.NextDue.Sub(clock.Now()))
1671		registry.RequestProbe(reg.Endpoint)
1672	}
1673	call := receiveCall(t, probe)
1674	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1675	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1676		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1677	})
1678	require.Zero(t, host.ConsecutiveFailures)
1679}
1680
1681func TestRegistryBackoffIsExponentialAndCapped(t *testing.T) {
1682	start := time.Unix(100, 0)
1683	clock := newManualClock(start)
1684	probe := newTestProbe(1)
1685	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1686	reg := registration(t, "example.test", 1)
1687	require.NoError(t, registry.setHosts(hostRecords(reg)))
1688	startRegistry(t, registry)
1689
1690	probeErr := errors.New("offline")
1691	// A watched host backs off from its two-second demand cadence, doubling
1692	// per failure up to the demand retry cap.
1693	want := []time.Duration{
1694		2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second,
1695		demandRetryMax, demandRetryMax, demandRetryMax,
1696	}
1697	for index, delay := range want {
1698		call := receiveCall(t, probe)
1699		at := clock.Now()
1700		call.result <- probeAnswer{err: probeErr}
1701		host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1702			return !host.Checking && host.ConsecutiveFailures == uint(index+1)
1703		})
1704		require.Equal(t, domain.RemoteAvailabilityUnreachable, host.Availability)
1705		require.Equal(t, domain.RemoteFailureTransport, host.LastFailure.Kind)
1706		require.ErrorIs(t, host.LastFailure, probeErr)
1707		require.Equal(t, at.Add(delay), host.NextDue)
1708
1709		if index == len(want)-1 {
1710			break
1711		}
1712		// The retry is not admitted early: only the backoff deadline wakes it.
1713		clock.Advance(delay - time.Nanosecond)
1714		requireNoProbeWithin(t, probe, probeGrace)
1715		clock.Advance(time.Nanosecond)
1716	}
1717}
1718
1719func TestRegistryJitterAppliesToRetryAndHealthyRefresh(t *testing.T) {
1720	start := time.Unix(100, 0)
1721	clock := newManualClock(start)
1722	probe := newTestProbe(1)
1723	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1724	registry.jitter = func(base time.Duration, _ string, _ uint64) time.Duration { return base / 2 }
1725	reg := registration(t, "example.test", 1)
1726	require.NoError(t, registry.setHosts(hostRecords(reg)))
1727	startRegistry(t, registry)
1728
1729	call := receiveCall(t, probe)
1730	call.result <- probeAnswer{err: errors.New("offline")}
1731	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1732		return !host.Checking && host.Availability == domain.RemoteAvailabilityUnreachable
1733	})
1734	require.Equal(t, start.Add(defaultDemandFreshForRemote/2), host.NextDue)
1735
1736	clock.Advance(defaultDemandFreshForRemote / 2)
1737	next := receiveCall(t, probe)
1738	successAt := clock.Now()
1739	next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
1740	host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1741		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
1742	})
1743	// The healthy refresh is jittered too: a fleet of hosts never converges
1744	// on one instant.
1745	require.Equal(t, successAt.Add(defaultDemandFreshForRemote/2), host.NextDue)
1746}
1747
1748func TestRegistryJitterIsDeterministicAndBounded(t *testing.T) {
1749	bases := []time.Duration{defaultRetryBase, defaultFreshFor, defaultRetryLimit}
1750	endpoints := []string{"example.test", "other.test", "user@host:2222"}
1751	for _, base := range bases {
1752		minimum, maximum := base*9/10, base*11/10
1753		for _, endpoint := range endpoints {
1754			spread := make(map[time.Duration]struct{})
1755			for attempt := uint64(1); attempt <= 64; attempt++ {
1756				got := jittered(base, endpoint, attempt)
1757				require.GreaterOrEqual(t, got, minimum)
1758				require.LessOrEqual(t, got, maximum)
1759				// Identical endpoint and attempt always yield the identical
1760				// duration: no wall clock and no shared PRNG participate.
1761				require.Equal(t, got, jittered(base, endpoint, attempt))
1762				spread[got] = struct{}{}
1763			}
1764			require.Greater(t, len(spread), 1, "jitter must vary across attempts")
1765		}
1766	}
1767	// A non-positive base has no spread to apply.
1768	require.Zero(t, jittered(0, "example.test", 1))
1769}
1770
1771func TestRegistryFencesStaleCompletionAcrossRegistrationChanges(t *testing.T) {
1772	replacementAnswer := func(name string) probeAnswer {
1773		return probeAnswer{snapshot: ports.BrokerDaemonObservation{
1774			Availability:   domain.RemoteAvailabilityReachable,
1775			InventoryKnown: true,
1776			Sessions:       []catalogue.RemoteCatalogSession{hostSession(name)},
1777		}}
1778	}
1779	assertFenced := func(t *testing.T, current, replacement domain.RemoteRegistration) {
1780		t.Helper()
1781		clock := newManualClock(time.Unix(100, 0))
1782		probe := newTestProbe(2)
1783		registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1784		require.NoError(t, registry.setHosts(hostRecords(current)))
1785		startRegistry(t, registry)
1786		currentCall := receiveCall(t, probe)
1787		require.True(t, currentCall.registration.Equal(current))
1788
1789		require.NoError(t, registry.setHosts(hostRecords(replacement)))
1790		replacementCall := receiveCall(t, probe)
1791		require.True(t, replacementCall.registration.Equal(replacement))
1792
1793		// The replacement completes and becomes authoritative...
1794		replacementCall.result <- replacementAnswer("replacement")
1795		applied := waitHost(t, registry, replacement.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1796			return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable && len(host.Sessions) == 1
1797		})
1798		require.True(t, applied.Registration.Equal(replacement))
1799		require.Equal(t, "replacement", applied.Sessions[0].Name)
1800
1801		// ...and the completion bound to the retired registration, arriving
1802		// afterwards, never touches it.
1803		currentCall.result <- replacementAnswer("retired")
1804		requireHostStableWithin(t, registry, replacement.Endpoint, applied, probeGrace)
1805
1806		// The retired completion also never holds the replacement's slot: a
1807		// further observation is still admitted and still applies.
1808		registry.RequestProbe(replacement.Endpoint)
1809		next := receiveCall(t, probe)
1810		require.True(t, next.registration.Equal(replacement))
1811		next.result <- replacementAnswer("next")
1812		waitHost(t, registry, replacement.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1813			return !host.Checking && len(host.Sessions) == 1 && host.Sessions[0].Name == "next"
1814		})
1815	}
1816
1817	t.Run("incarnation", func(t *testing.T) {
1818		assertFenced(t, registration(t, "example.test", 1), registration(t, "example.test", 2))
1819	})
1820	t.Run("generation", func(t *testing.T) {
1821		current := registration(t, "example.test", 1)
1822		replacement := current
1823		replacement.Generation++
1824		assertFenced(t, current, replacement)
1825	})
1826}
1827
1828func TestRegistryFencesRetiredAttemptsAcrossRemovalAndIdenticalReAdd(t *testing.T) {
1829	clock := newManualClock(time.Unix(100, 0))
1830	probe := newTestProbe(2)
1831	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1832	reg := registration(t, "example.test", 1)
1833	require.NoError(t, registry.setHosts(hostRecords(reg)))
1834
1835	ctx, cancel := context.WithCancel(context.Background())
1836	defer cancel()
1837	results := make(chan probeResult, 1)
1838	dispatch := func() { registry.dispatch(ctx, clock.Now(), results) }
1839
1840	registry.RequestProbe(reg.Endpoint)
1841	dispatch()
1842	retiredCall := receiveCall(t, probe)
1843	require.True(t, retiredCall.registration.Equal(reg))
1844	retired := attemptFor(registry, reg.Endpoint)
1845	require.NotZero(t, retired)
1846
1847	// Removal retires a bounded tombstone and drops the in-flight token.
1848	require.NoError(t, registry.setHosts(nil))
1849	removed := registry.Snapshot()
1850	require.Empty(t, removed.Daemons)
1851	require.Len(t, removed.Removed, 1)
1852	require.Equal(t, removed.Revision, removed.Removed[0].RetiredRevision)
1853	require.True(t, removed.Removed[0].Fences(reg))
1854	require.NoError(t, removed.Validate())
1855
1856	// The retired attempt can never revive a removed endpoint.
1857	registry.apply(probeResult{endpoint: reg.Endpoint, registration: reg, attempt: retired, snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}, at: clock.Now()})
1858	require.Empty(t, registry.Snapshot().Daemons)
1859	require.Len(t, registry.Snapshot().Removed, 1)
1860
1861	// Re-adding the identical registration supersedes the tombstone, so only
1862	// the attempt token can fence the retired completion.
1863	require.NoError(t, registry.setHosts(hostRecords(reg)))
1864	require.Empty(t, registry.Snapshot().Removed)
1865	registry.RequestProbe(reg.Endpoint)
1866	dispatch()
1867	freshCall := receiveCall(t, probe)
1868	require.True(t, freshCall.registration.Equal(reg))
1869	fresh := attemptFor(registry, reg.Endpoint)
1870	require.Greater(t, fresh, retired)
1871
1872	registry.apply(probeResult{endpoint: reg.Endpoint, registration: reg, attempt: retired, snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}, at: clock.Now()})
1873	stale, ok := registry.Snapshot().Find(reg.Endpoint)
1874	require.True(t, ok)
1875	require.True(t, stale.Checking)
1876	require.Equal(t, domain.RemoteAvailabilityUnknown, stale.Availability)
1877
1878	registry.apply(probeResult{endpoint: reg.Endpoint, registration: reg, attempt: fresh, snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}, at: clock.Now()})
1879	applied, ok := registry.Snapshot().Find(reg.Endpoint)
1880	require.True(t, ok)
1881	require.False(t, applied.Checking)
1882	require.Equal(t, domain.RemoteAvailabilityReachable, applied.Availability)
1883}
1884
1885func TestRegistryRetiresInflightAttemptsOnReplacementAndRemoval(t *testing.T) {
1886	current := registration(t, "example.test", 1)
1887	replacement := registration(t, "example.test", 2)
1888	tests := []struct {
1889		name string
1890		next []ports.BrokerHostRecord
1891	}{
1892		{name: "replacement", next: []ports.BrokerHostRecord{hostRecord(replacement)}},
1893		{name: "removal", next: nil},
1894	}
1895	for _, tt := range tests {
1896		t.Run(tt.name, func(t *testing.T) {
1897			clock := newManualClock(time.Unix(100, 0))
1898			probe := newTestProbe(2)
1899			registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1900			require.NoError(t, registry.setHosts(hostRecords(current)))
1901			startRegistry(t, registry)
1902
1903			call := receiveCall(t, probe)
1904			require.True(t, call.registration.Equal(current))
1905
1906			// Replacing or removing the registration retires the in-flight
1907			// attempt: the probe's derived context is cancelled before the
1908			// replacement can start, so a ctx-honoring probe stops instead of
1909			// overlapping its successor.
1910			require.NoError(t, registry.setHosts(tt.next))
1911			select {
1912			case <-probe.canceled:
1913			case <-time.After(time.Second):
1914				t.Fatal("retired probe did not observe cancellation")
1915			}
1916		})
1917	}
1918}
1919
1920func TestRegistryChurnKeepsOneActiveProbePerEndpoint(t *testing.T) {
1921	clock := newManualClock(time.Unix(100, 0))
1922	probe := newChurnProbe()
1923	defer probe.openRelease()
1924	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
1925
1926	first := registration(t, "example.test", 1)
1927	second := registration(t, "example.test", 2)
1928	require.NoError(t, registry.setHosts(hostRecords(first)))
1929
1930	ctx, cancel := context.WithCancel(context.Background())
1931	defer cancel()
1932	done := make(chan struct{})
1933	go func() { defer close(done); registry.Run(ctx) }()
1934	registry.RequestProbe(first.Endpoint)
1935
1936	// Drop the initial attempt's start signal so each lockstep round below
1937	// consumes the start of its own replacement attempt.
1938	select {
1939	case <-probe.startedSig:
1940	case <-time.After(time.Second):
1941		t.Fatal("timed out waiting for the initial probe")
1942	}
1943	require.Equal(t, 1, probe.startedCount())
1944
1945	// Lockstep churn: every round replaces the registration, which must cancel
1946	// the retired attempt before the replacement starts. An uncancelled
1947	// retirement would stay live and overlap its successor, and the exact start
1948	// count catches an attempt that never restarts.
1949	const rounds = 8
1950	for round := 0; round < rounds; round++ {
1951		next := first
1952		if round%2 == 0 {
1953			next = second
1954		}
1955		require.NoError(t, registry.setHosts(hostRecords(next)))
1956		registry.RequestProbe(next.Endpoint)
1957		select {
1958		case <-probe.startedSig:
1959		case <-time.After(time.Second):
1960			t.Fatal("timed out waiting for the replacement attempt to start")
1961		}
1962		require.Equal(t, round+2, probe.startedCount())
1963	}
1964
1965	// Concurrent churn and demand: the same invariant must hold under the race
1966	// detector while membership and demand churn from several goroutines.
1967	failures := make([]error, 4)
1968	var wg sync.WaitGroup
1969	for worker := 0; worker < len(failures); worker++ {
1970		wg.Add(1)
1971		go func(worker int) {
1972			defer wg.Done()
1973			for round := 0; round < 32; round++ {
1974				next := first
1975				if (worker+round)%2 == 0 {
1976					next = second
1977				}
1978				if err := registry.setHosts(hostRecords(next)); err != nil {
1979					failures[worker] = err
1980					return
1981				}
1982				registry.RequestProbe(first.Endpoint)
1983				registry.Snapshot()
1984			}
1985		}(worker)
1986	}
1987	wg.Wait()
1988	for _, err := range failures {
1989		require.NoError(t, err)
1990	}
1991
1992	endpoints := probe.observedEndpoints()
1993	require.NotEmpty(t, endpoints)
1994	for _, endpoint := range endpoints {
1995		require.Equal(t, 1, probe.maxLiveFor(endpoint), "endpoint %q had overlapping live probes", endpoint)
1996	}
1997	require.GreaterOrEqual(t, probe.startedCount(), rounds+1)
1998
1999	// Settling releases every attempt: no probe stays live once Run returns.
2000	cancel()
2001	select {
2002	case <-done:
2003	case <-time.After(time.Second):
2004		t.Fatal("registry did not stop after cancellation")
2005	}
2006	deadline := time.Now().Add(time.Second)
2007	for probe.inflight() != 0 {
2008		if time.Now().After(deadline) {
2009			t.Fatalf("%d probes still live after settle", probe.inflight())
2010		}
2011		time.Sleep(time.Millisecond)
2012	}
2013}
2014
2015func TestRegistryPublishesBoundedTombstones(t *testing.T) {
2016	registry := newTestRegistry(t, 1, newTestStore(), newTestProbe(1), newManualClock(time.Unix(100, 0)))
2017
2018	// Two full membership rounds with distinct endpoints: without pruning the
2019	// retired set would double.
2020	for round := 0; round < 2; round++ {
2021		registrations := make([]ports.BrokerHostRecord, 0, ports.BrokerMaxHosts)
2022		for index := 0; index < ports.BrokerMaxHosts; index++ {
2023			registrations = append(registrations, hostRecord(registration(t, fmt.Sprintf("round-%d-host-%d.test", round, index), byte(round*ports.BrokerMaxHosts+index+1))))
2024		}
2025		require.NoError(t, registry.setHosts(registrations))
2026		require.NoError(t, registry.setHosts(nil))
2027	}
2028
2029	snapshot := registry.Snapshot()
2030	require.NoError(t, snapshot.Validate())
2031	require.Len(t, snapshot.Removed, ports.BrokerMaxTombstones)
2032	for index, tombstone := range snapshot.Removed {
2033		require.NoError(t, tombstone.Validate())
2034		// Only the newest retirements survive the bound.
2035		require.Equal(t, snapshot.Revision, tombstone.RetiredRevision)
2036		if index > 0 {
2037			require.GreaterOrEqual(t, snapshot.Removed[index-1].RetiredRevision, tombstone.RetiredRevision)
2038		}
2039	}
2040}
2041
2042func TestRegistryPersistenceRunsOffLockAndCoalescesWrites(t *testing.T) {
2043	clock := newManualClock(time.Unix(100, 0))
2044	probe := newTestProbe(1)
2045	store := newBlockingStore()
2046	registry, err := NewRegistry(1, store, probe, clock, nil)
2047	require.NoError(t, err)
2048	defer registry.store.close()
2049
2050	reg := registration(t, "example.test", 1)
2051	require.NoError(t, registry.setHosts(hostRecords(reg)))
2052	// The writer is now blocked inside the first durable write.
2053	first := <-store.entered
2054	require.Equal(t, ports.BrokerRevision(2), first)
2055
2056	// Publication is never blocked behind persistence and never holds the
2057	// registry lock across a store call.
2058	var setErr error
2059	published := make(chan struct{})
2060	go func() {
2061		defer close(published)
2062		for generation := domain.RemoteGeneration(2); generation <= 6; generation++ {
2063			reg.Generation = generation
2064			if err := registry.setHosts(hostRecords(reg)); err != nil {
2065				setErr = err
2066				return
2067			}
2068		}
2069	}()
2070	select {
2071	case <-published:
2072	case <-time.After(time.Second):
2073		t.Fatal("publication blocked behind durable persistence")
2074	}
2075	require.NoError(t, setErr)
2076
2077	last := registry.Snapshot().Revision
2078	store.releaseFirst()
2079	waitWriterRevision(t, registry, last)
2080
2081	// Writes stay ordered and coalesced: only the first and the newest
2082	// revision ever reach the store.
2083	require.Equal(t, []ports.BrokerRevision{2, last}, store.writtenRevisions())
2084}
2085
2086// TestRegistryReplaceHostsBlockedCASKeepsSnapshotLockFree proves membership
2087// durability and publication stay independent: a ReplaceHosts parked inside the
2088// durable CAS may hold the registry lock, but Snapshot is served from the atomic
2089// publication, no observation may be applied behind it, and the run loop
2090// resumes scheduling as soon as the CAS commits.
2091func TestRegistryReplaceHostsBlockedCASKeepsSnapshotLockFree(t *testing.T) {
2092	clock := newManualClock(time.Unix(100, 0))
2093	probe := newTestProbe(2)
2094	store := newBlockingCASStore()
2095	registry, err := NewRegistry(1, store, probe, clock, nil)
2096	require.NoError(t, err)
2097
2098	reg := registration(t, "example.test", 1)
2099	require.NoError(t, registry.setHosts(hostRecords(reg)))
2100	startRegistry(t, registry)
2101	call := receiveCall(t, probe)
2102	require.True(t, waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2103		return host.Checking
2104	}).Checking)
2105
2106	// The replacement takes the registry lock and parks inside the durable CAS.
2107	replaced := make(chan error, 1)
2108	go func() {
2109		replaced <- registry.ReplaceHosts([]ports.BrokerHostRecord{{Registration: reg, Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), reg.Endpoint)}})
2110	}()
2111	<-store.entered
2112	t.Cleanup(store.releaseCAS)
2113
2114	// Snapshot stays lock-free: the atomic publication is readable while the
2115	// mutator is parked inside the store, and it still carries the pre-CAS
2116	// projection.
2117	observed := make(chan ports.BrokerSnapshot, 1)
2118	go func() { observed <- registry.Snapshot() }()
2119	select {
2120	case snapshot := <-observed:
2121		require.NoError(t, snapshot.Validate())
2122		require.Len(t, snapshot.Daemons, 1)
2123		require.Equal(t, reg, snapshot.Daemons[0].Registration)
2124	case <-time.After(time.Second):
2125		t.Fatal("Snapshot blocked behind the durable membership CAS")
2126	}
2127
2128	// Wake the run loop and complete the observation while the CAS is still
2129	// parked: applying the result needs the registry lock, so the in-flight
2130	// projection must stay exactly as published until membership commits.
2131	registry.hint()
2132	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
2133	pending, ok := registry.Snapshot().Find(reg.Endpoint)
2134	require.True(t, ok)
2135	require.True(t, pending.Checking)
2136	stable := time.NewTimer(probeGrace)
2137	defer stable.Stop()
2138
2139window:
2140	for {
2141		current, found := registry.Snapshot().Find(reg.Endpoint)
2142		require.True(t, found)
2143		require.Equal(t, pending, current)
2144		select {
2145		case <-stable.C:
2146			break window
2147		case <-time.After(time.Millisecond):
2148		}
2149	}
2150
2151	// Committing membership releases the mutator and unparks the run loop, which
2152	// applies the observation and resumes scheduling the next one.
2153	store.releaseCAS()
2154	require.NoError(t, <-replaced)
2155	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2156		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
2157	})
2158	require.Equal(t, clock.Now(), host.LastSuccess)
2159	clock.Advance(defaultFreshFor)
2160	receiveCall(t, probe)
2161}
2162
2163func TestRegistryStoreErrorsAreObservableAndDoNotStallPublication(t *testing.T) {
2164	clock := newManualClock(time.Unix(100, 0))
2165	probe := newTestProbe(1)
2166	logger, logs := testLogger()
2167	storeErr := errors.New("disk full")
2168	store := &testStore{storeErr: storeErr}
2169	registry, err := NewRegistry(1, store, probe, clock, logger)
2170	require.NoError(t, err)
2171	defer registry.store.close()
2172
2173	reg := registration(t, "example.test", 1)
2174	require.NoError(t, registry.setHosts(hostRecords(reg)))
2175	waitForLog(t, logs, "durable snapshot write failed")
2176	require.Contains(t, logs.String(), storeErr.Error())
2177
2178	// A failed write is not fatal and never blocks the next publication.
2179	store.setStoreErr(nil)
2180	reg.Generation++
2181	require.NoError(t, registry.setHosts(hostRecords(reg)))
2182	stored := waitStored(t, store, func(snapshot ports.BrokerSnapshot) bool {
2183		return snapshot.Revision == registry.Snapshot().Revision
2184	})
2185	require.Len(t, stored.Daemons, 1)
2186	require.True(t, stored.Daemons[0].Registration.Equal(reg))
2187}
2188
2189func TestRegistryPersistedCopyClearsChecking(t *testing.T) {
2190	clock := newManualClock(time.Unix(100, 0))
2191	probe := newTestProbe(1)
2192	store := &testStore{}
2193	registry := newTestRegistry(t, 3, store, probe, clock)
2194	reg := registration(t, "example.test", 1)
2195	require.NoError(t, registry.setHosts(hostRecords(reg)))
2196	startRegistry(t, registry)
2197
2198	// The observation is in flight: the published projection carries Checking,
2199	// the durable copy never does.
2200	require.True(t, waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2201		return host.Checking
2202	}).Checking)
2203	stored := waitStored(t, store, func(snapshot ports.BrokerSnapshot) bool {
2204		return len(snapshot.Daemons) == 1
2205	})
2206	require.False(t, stored.Daemons[0].Checking)
2207
2208	// Completing the observation persists the settled projection.
2209	call := receiveCall(t, probe)
2210	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
2211	settled := waitStored(t, store, func(snapshot ports.BrokerSnapshot) bool {
2212		return len(snapshot.Daemons) == 1 && snapshot.Daemons[0].Availability == domain.RemoteAvailabilityReachable
2213	})
2214	require.False(t, settled.Daemons[0].Checking)
2215}
2216
2217// TestRegistryNeverPersistsRetirementTombstones proves retirement tombstones
2218// are process-local fencing state: the durable copy carries no Removed, and a
2219// fresh registry reopened over that snapshot neither resurrects the retired
2220// host nor sees a stale tombstone.
2221func TestRegistryNeverPersistsRetirementTombstones(t *testing.T) {
2222	clock := newManualClock(time.Unix(100, 0))
2223	store := &testStore{}
2224	registry := newTestRegistry(t, 1, store, newTestProbe(1), clock)
2225	defer registry.store.close()
2226
2227	reg := registration(t, "retired.test", 1)
2228	require.NoError(t, registry.setHosts(hostRecords(reg)))
2229	require.NoError(t, registry.setHosts(nil))
2230
2231	// Removal publishes a bounded tombstone and drops the live host in memory.
2232	require.Empty(t, registry.Snapshot().Daemons)
2233	require.Len(t, registry.Snapshot().Removed, 1)
2234
2235	// The durable copy drops the tombstone before the store ever sees it.
2236	stored := waitStored(t, store, func(snapshot ports.BrokerSnapshot) bool {
2237		return len(snapshot.Daemons) == 0 && snapshot.Revision == registry.Snapshot().Revision
2238	})
2239	require.Empty(t, stored.Removed)
2240
2241	// Reopening over the durable snapshot adopts empty membership without the
2242	// stale tombstone.
2243	reopened, err := NewRegistry(2, &testStore{loaded: stored}, newTestProbe(1), clock, nil)
2244	require.NoError(t, err)
2245	require.Empty(t, reopened.Snapshot().Daemons)
2246	require.Empty(t, reopened.Snapshot().Removed)
2247}
2248
2249func TestRegistryBoundsSlowSubscribersAndStopsOnCancellation(t *testing.T) {
2250	clock := newManualClock(time.Unix(100, 0))
2251	probe := newTestProbe(1)
2252	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
2253
2254	sub := registry.Subscribe()
2255	reg := registration(t, "example.test", 1)
2256	for generation := domain.RemoteGeneration(1); generation <= 10; generation++ {
2257		reg.Generation = generation
2258		require.NoError(t, registry.setHosts(hostRecords(reg)))
2259	}
2260	// A subscriber that never drains is bounded to one pending wake and
2261	// never blocks publication.
2262	require.Len(t, sub.Changed(), 1)
2263
2264	// Drain the pending wake, then close: a closed subscriber never receives
2265	// another notification, and closing is idempotent.
2266	<-sub.Changed()
2267	sub.Close()
2268	sub.Close()
2269	reg.Generation++
2270	require.NoError(t, registry.setHosts(hostRecords(reg)))
2271	select {
2272	case <-sub.Changed():
2273		t.Fatal("closed subscription received a notification")
2274	default:
2275	}
2276
2277	ctx, cancel := context.WithCancel(context.Background())
2278	done := make(chan struct{})
2279	go func() { defer close(done); registry.Run(ctx) }()
2280	registry.RequestProbe(reg.Endpoint)
2281	receiveCall(t, probe)
2282	cancel()
2283	select {
2284	case <-done:
2285	case <-time.After(time.Second):
2286		t.Fatal("registry did not stop after cancellation")
2287	}
2288	select {
2289	case <-probe.canceled:
2290	case <-time.After(time.Second):
2291		t.Fatal("in-flight probe did not observe cancellation")
2292	}
2293	// Cleanup settles the projection: no host stays in flight.
2294	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2295		return !host.Checking
2296	})
2297	require.Equal(t, domain.RemoteAvailabilityUnknown, host.Availability)
2298}
2299
2300func TestRegistryRunIsSingleShotAndCleansUpInFlightState(t *testing.T) {
2301	clock := newManualClock(time.Unix(100, 0))
2302	probe := newTestProbe(1)
2303	logger, logs := testLogger()
2304	store := &testStore{}
2305	registry, err := NewRegistry(1, store, probe, clock, logger)
2306	require.NoError(t, err)
2307	reg := registration(t, "example.test", 1)
2308	require.NoError(t, registry.setHosts(hostRecords(reg)))
2309
2310	ctx, cancel := context.WithCancel(context.Background())
2311	done := make(chan struct{})
2312	go func() { defer close(done); registry.Run(ctx) }()
2313	registry.RequestProbe(reg.Endpoint)
2314
2315	call := receiveCall(t, probe)
2316	require.True(t, waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2317		return host.Checking
2318	}).Checking)
2319
2320	// A concurrent Run is refused instead of duplicating scheduling.
2321	refused := make(chan struct{})
2322	go func() { defer close(refused); registry.Run(context.Background()) }()
2323	select {
2324	case <-refused:
2325	case <-time.After(time.Second):
2326		t.Fatal("second Run did not return")
2327	}
2328	require.Contains(t, logs.String(), "Run refused")
2329
2330	cancel()
2331	select {
2332	case <-done:
2333	case <-time.After(time.Second):
2334		t.Fatal("registry did not stop after cancellation")
2335	}
2336	// Cleanup publishes the settled projection: nothing stays in flight.
2337	require.False(t, waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2338		return !host.Checking
2339	}).Checking)
2340
2341	// The retired completion can never be applied once the run is over.
2342	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
2343	settled := registry.Snapshot()
2344	require.Equal(t, domain.RemoteAvailabilityUnknown, settled.Daemons[0].Availability)
2345	require.Zero(t, settled.Daemons[0].ConsecutiveFailures)
2346
2347	// Persistence is flushed before shutdown and no write outlives it.
2348	stored := waitStored(t, store, func(snapshot ports.BrokerSnapshot) bool {
2349		return snapshot.Revision >= settled.Revision
2350	})
2351	require.Equal(t, settled.Revision, stored.Revision)
2352	require.False(t, stored.Daemons[0].Checking)
2353	writes := store.writeCount()
2354	require.NoError(t, registry.setHosts(nil))
2355	require.Equal(t, writes, store.writeCount())
2356	require.Contains(t, logs.String(), "refused after shutdown")
2357
2358	// A restart is refused rather than racing a stale scheduler.
2359	restarted := make(chan struct{})
2360	go func() { defer close(restarted); registry.Run(context.Background()) }()
2361	select {
2362	case <-restarted:
2363	case <-time.After(time.Second):
2364		t.Fatal("restarted Run did not return")
2365	}
2366	requireNoProbeWithin(t, probe, probeGrace)
2367}
2368
2369func TestRegistryRevisionsIncreaseMonotonicallyWithinEpoch(t *testing.T) {
2370	start := time.Unix(100, 0)
2371	clock := newManualClock(start)
2372	probe := newTestProbe(2)
2373	registry := newTestRegistry(t, 42, newTestStore(), probe, clock)
2374
2375	var last ports.BrokerRevision
2376	observe := func() {
2377		t.Helper()
2378		snapshot := registry.Snapshot()
2379		require.Equal(t, ports.BrokerEpoch(42), snapshot.Epoch)
2380		require.Greater(t, snapshot.Revision, last)
2381		last = snapshot.Revision
2382	}
2383	observe()
2384
2385	first := registration(t, "a.test", 1)
2386	second := registration(t, "b.test", 2)
2387	require.NoError(t, registry.setHosts(hostRecords(first)))
2388	observe()
2389	require.NoError(t, registry.setHosts(hostRecords(first, second)))
2390	observe()
2391
2392	startRegistry(t, registry)
2393	for range 2 {
2394		call := receiveCall(t, probe)
2395		call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
2396		waitHost(t, registry, call.registration.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2397			return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
2398		})
2399		observe()
2400	}
2401}
2402
2403// TestRegistryProbeCannotOverrideConfiguredAuthority proves the ownership rule:
2404// a probe supplies observed state only. Configured authority (local flag,
2405// endpoint, registration, policy, display origin, rank) is stamped by the
2406// registry from its own host records and a hostile probe result can never
2407// override it.
2408func TestRegistryProbeCannotOverrideConfiguredAuthority(t *testing.T) {
2409	clock := newManualClock(time.Unix(100, 0))
2410	probe := newTestProbe(1)
2411	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
2412	reg := registration(t, "user@host:22", 1)
2413	require.NoError(t, registry.setHosts(hostRecords(reg)))
2414	startRegistry(t, registry)
2415
2416	call := receiveCall(t, probe)
2417	require.True(t, call.registration.Equal(reg))
2418
2419	// Every authority field is invented by the probe, including a local flag,
2420	// a foreign endpoint and registration, a different policy, and display/rank
2421	// hints. None of it may reach the published projection.
2422	hostilePolicy := poolPolicy()
2423	hostilePolicy.Trust = "attacker"
2424	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
2425		Local:         true,
2426		Endpoint:      "attacker@evil",
2427		DisplayOrigin: "evil",
2428		Rank:          42,
2429		Registration:  registration(t, "attacker@evil", 9),
2430		Policy:        hostilePolicy,
2431		Availability:  domain.RemoteAvailabilityReachable,
2432	}}
2433	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2434		return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
2435	})
2436	require.False(t, host.Local)
2437	require.Equal(t, reg.Endpoint, host.Endpoint)
2438	require.Equal(t, domain.RemoteDisplayOrigin(reg.Endpoint), host.DisplayOrigin)
2439	require.Zero(t, host.Rank)
2440	require.True(t, host.Registration.Equal(reg))
2441	require.Equal(t, poolPolicy(), host.Policy)
2442	require.NotEqual(t, hostilePolicy, host.Policy)
2443	require.Len(t, registry.Snapshot().Daemons, 1)
2444}
2445
2446// TestRegistryPublishesObservedDaemonIdentity proves observed identity,
2447// incarnation, protocol version, and capabilities flow from a probe result into
2448// the published snapshot; configured authority is stamped alongside them.
2449func TestRegistryPublishesObservedDaemonIdentity(t *testing.T) {
2450	clock := newManualClock(time.Unix(100, 0))
2451	probe := newTestProbe(1)
2452	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
2453	reg := registration(t, "user@host:22", 1)
2454	require.NoError(t, registry.setHosts(hostRecords(reg)))
2455	startRegistry(t, registry)
2456
2457	call := receiveCall(t, probe)
2458	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
2459		Identity:        "authed-daemon",
2460		Incarnation:     ports.BrokerDaemonIncarnation{7, 11},
2461		ProtocolVersion: protocol.Version,
2462		Capabilities:    protocol.CapabilityResume | protocol.CapabilityUDP,
2463		Availability:    domain.RemoteAvailabilityReachable,
2464	}}
2465	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2466		return !host.Checking && host.Identity != ""
2467	})
2468	require.Equal(t, ports.BrokerDaemonIdentity("authed-daemon"), host.Identity)
2469	require.Equal(t, ports.BrokerDaemonIncarnation{7, 11}, host.Incarnation)
2470	require.Equal(t, protocol.Version, host.ProtocolVersion)
2471	require.Equal(t, protocol.CapabilityResume|protocol.CapabilityUDP, host.Capabilities)
2472	require.Equal(t, poolPolicy(), host.Policy)
2473	require.NoError(t, registry.Snapshot().Validate())
2474}
2475
2476// TestRegistryRejectsObservedIdentityWithoutProtocolVersion pins the partial
2477// identity rule: an observation that reports identity and incarnation without a
2478// protocol version fails ports validation and is classified as an invalid
2479// response, so it never publishes a half-observed daemon.
2480func TestRegistryRejectsObservedIdentityWithoutProtocolVersion(t *testing.T) {
2481	// The contract itself rejects partial identity.
2482	require.Error(t, (ports.BrokerDaemonObservation{
2483		DisplayOrigin: "host",
2484		Policy:        poolPolicy(),
2485		Identity:      "authed-daemon",
2486		Incarnation:   ports.BrokerDaemonIncarnation{7},
2487		Availability:  domain.RemoteAvailabilityReachable,
2488	}).Validate())
2489
2490	clock := newManualClock(time.Unix(100, 0))
2491	probe := newTestProbe(1)
2492	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
2493	reg := registration(t, "user@host:22", 1)
2494	require.NoError(t, registry.setHosts(hostRecords(reg)))
2495	startRegistry(t, registry)
2496
2497	call := receiveCall(t, probe)
2498	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{
2499		Identity:     "authed-daemon",
2500		Incarnation:  ports.BrokerDaemonIncarnation{7},
2501		Availability: domain.RemoteAvailabilityReachable,
2502	}}
2503	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
2504		return !host.Checking && host.ConsecutiveFailures == 1
2505	})
2506	require.Equal(t, domain.RemoteAvailabilityInvalidResponse, host.Availability)
2507	require.Equal(t, domain.RemoteFailureInvalidResponse, host.LastFailure.Kind)
2508	require.Empty(t, host.Identity, "partial identity must never be published")
2509	require.True(t, host.Incarnation.IsZero())
2510}
2511
2512// TestRegistryPublishesRemoteDaemonsInRegistrationOrder pins the publication
2513// invariant: remote daemons appear in durable membership registration order (no
2514// local producer exists yet, so the snapshot carries remotes only).
2515func TestRegistryPublishesRemoteDaemonsInRegistrationOrder(t *testing.T) {
2516	registry := newTestRegistry(t, 1, newTestStore(), newTestProbe(1), newManualClock(time.Unix(100, 0)))
2517	first := registration(t, "a.test", 1)
2518	second := registration(t, "b.test", 2)
2519	third := registration(t, "c.test", 3)
2520	require.NoError(t, registry.setHosts(hostRecords(first, second)))
2521	require.Equal(t, []string{"a.test", "b.test"}, daemonEndpoints(registry.Snapshot()))
2522
2523	// Membership changes publish in the new registration order, not sorted and
2524	// not map order.
2525	require.NoError(t, registry.setHosts(hostRecords(third, second, first)))
2526	require.Equal(t, []string{"c.test", "b.test", "a.test"}, daemonEndpoints(registry.Snapshot()))
2527}
2528
2529func daemonEndpoints(snapshot ports.BrokerSnapshot) []string {
2530	endpoints := make([]string, 0, len(snapshot.Daemons))
2531	for _, daemon := range snapshot.Daemons {
2532		endpoints = append(endpoints, daemon.Endpoint)
2533	}
2534	return endpoints
2535}
2536
2537// TestDurableProjectionDropsLocalObservationAndPolicy pins the durable format
2538// decision: the persistable copy carries remote observations only (a local
2539// daemon observation is process-local) and omits policy, because membership is
2540// the single policy authority and the loader re-stamps it.
2541func TestDurableProjectionDropsLocalObservationAndPolicy(t *testing.T) {
2542	reg := registration(t, "user@host:22", 1)
2543	snapshot := ports.BrokerSnapshot{Epoch: 1, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
2544		{Local: true, DisplayOrigin: "local", Policy: poolPolicy(), Availability: domain.RemoteAvailabilityReachable},
2545		{Endpoint: reg.Endpoint, DisplayOrigin: domain.RemoteDisplayOrigin(reg.Endpoint), Registration: reg, Policy: poolPolicy(), Availability: domain.RemoteAvailabilityReachable},
2546	}}
2547	out := durable(snapshot)
2548	require.Len(t, out.Daemons, 1)
2549	require.False(t, out.Daemons[0].Local)
2550	require.Equal(t, reg.Endpoint, out.Daemons[0].Endpoint)
2551	require.Equal(t, ports.BrokerPolicy{}, out.Daemons[0].Policy)
2552	require.Nil(t, out.Removed)
2553	// The published snapshot the caller handed in is untouched.
2554	require.Len(t, snapshot.Daemons, 2)
2555	require.Equal(t, poolPolicy(), snapshot.Daemons[0].Policy)
2556}
2557
2558// TestHostDisplayOriginIsBounded pins the presentation-hint bound: a derived
2559// origin is clamped rune-safely at the ports limit so an over-long configured
2560// endpoint still yields a valid publication instead of an unobservable host.
2561func TestHostDisplayOriginIsBounded(t *testing.T) {
2562	require.Equal(t, "host:22", hostDisplayOrigin("user@host:22"))
2563	require.Equal(t, "plain.test", hostDisplayOrigin("plain.test"))
2564
2565	origin := hostDisplayOrigin("user@" + strings.Repeat("€", 100))
2566	require.LessOrEqual(t, len(origin), ports.BrokerMaxDisplayOriginBytes)
2567	require.True(t, utf8.ValidString(origin))
2568	require.NotEmpty(t, origin)
2569}