1package broker
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "strings"
8 "sync"
9 "testing"
10 "time"
11 "unicode/utf8"
12
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/bnema/vev/internal/protocol"
16 "github.com/stretchr/testify/require"
17)
18
19// newTestAuthority composes one offline authority over an observation-disabled
20// registry, a bounded pool, and a supervisor on a long grace.
21func newTestAuthority(t testing.TB, epoch ports.BrokerEpoch, probe ports.BrokerHostProbe, connector poolConnector) (*Authority, *Registry, *Pool, *Supervisor, *manualClock, *testStore) {
22 t.Helper()
23 clock := newManualClock(time.Unix(0, 0))
24 store := newTestStore()
25 registry, err := NewRegistryWithConfig(epoch, store, probe, clock, nil, RegistryConfig{ObservationDisabled: true})
26 require.NoError(t, err)
27 authority, pool, supervisor, _ := composeTestAuthority(t, epoch, registry, connector, clock)
28 return authority, registry, pool, supervisor, clock, store
29}
30
31// composeTestAuthority wires one existing registry into a bounded pool and a
32// supervisor on a long grace, and owns the single Registry.Run. Every dependency
33// is drained at test end so no run goroutine leaks between tests: the registry
34// run is canceled and joined (which flushes and stops its durable writer), the
35// pool is closed, and the supervisor is closed. The supervisor's own manual
36// clock is returned so a test can drive the idle lifecycle.
37func composeTestAuthority(t testing.TB, epoch ports.BrokerEpoch, registry *Registry, connector poolConnector, clock *manualClock) (*Authority, *Pool, *Supervisor, *manualClock) {
38 t.Helper()
39 resolver := poolResolver(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
40 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: r.Policy, Address: "fake", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "canonical", Bound: true}}, nil
41 })
42 pool, err := NewPool(epoch, resolver, registry, connector, clock, PoolLimits{Physical: 4, Clients: 8, Streams: 16, StreamsPerClient: 16, Warm: 4, Idle: time.Minute})
43 require.NoError(t, err)
44 t.Cleanup(func() { require.NoError(t, pool.Close()) })
45 supervisor, idleClock := newTestSupervisor(t, time.Hour)
46 authority, err := NewAuthority(epoch, registry, pool, supervisor, testSessionCodec{})
47 require.NoError(t, err)
48 // The caller owns Registry.Run (see NewAuthority): start the single run and
49 // cancel it at cleanup, so the registry's durable writer is drained and its
50 // goroutine never survives the test.
51 registryCtx, stopRegistry := context.WithCancel(context.Background())
52 registryDone := make(chan struct{})
53 go func() { defer close(registryDone); registry.Run(registryCtx) }()
54 t.Cleanup(func() { stopRegistry(); <-registryDone })
55 return authority, pool, supervisor, idleClock
56}
57
58// newServiceMembershipFixture composes one admitted authority over an
59// observation-disabled registry with the requested membership mode. The store is
60// the caller's, so a test can observe the exact CAS count and inject failures.
61func newServiceMembershipFixture(t testing.TB, store ports.BrokerHostStore, mode MembershipMode) (*Authority, *Registry, *Supervisor, *manualClock) {
62 t.Helper()
63 clock := newManualClock(time.Unix(0, 0))
64 registry, err := NewRegistryWithConfig(1, store, nil, clock, nil, RegistryConfig{
65 MembershipMode: mode,
66 ObservationDisabled: true,
67 IncarnationGenerator: (&incarnationSequencer{}).generate,
68 })
69 require.NoError(t, err)
70 registry.jitter = identityJitter
71 authority, _, supervisor, idleClock := composeTestAuthority(t, 1, registry, immediateConnector, clock)
72 return authority, registry, supervisor, idleClock
73}
74
75// requireImmutableRefusal asserts one service-level membership mutation was
76// refused with the typed immutable code 12 and the ports sentinel.
77func requireImmutableRefusal(t *testing.T, err error) {
78 t.Helper()
79 require.Error(t, err)
80 var refusal ports.BrokerError
81 require.ErrorAs(t, err, &refusal)
82 require.Equal(t, ports.BrokerErrorMembershipImmutable, refusal.Code)
83 require.Equal(t, ports.BrokerErrorCode(12), refusal.Code, "immutable membership is the exact code 12")
84 require.ErrorIs(t, err, ports.ErrBrokerMembershipImmutable)
85}
86
87// casThenGateStore commits its membership CAS and then parks the mutation until
88// release, so a test can make a caller cancellation arrive strictly after the
89// durable commit point but before the mutation returns.
90type casThenGateStore struct {
91 *testStore
92 committed chan struct{}
93 release chan struct{}
94}
95
96func newCASThenGateStore() *casThenGateStore {
97 return &casThenGateStore{testStore: newTestStore(), committed: make(chan struct{}, 1), release: make(chan struct{})}
98}
99
100func (s *casThenGateStore) ReplaceHosts(expected uint64, hosts []ports.BrokerHostRecord) error {
101 if err := s.testStore.ReplaceHosts(expected, hosts); err != nil {
102 return err
103 }
104 select {
105 case s.committed <- struct{}{}:
106 default:
107 }
108 <-s.release
109 return nil
110}
111
112// releaseCAS lets a parked post-commit mutation return.
113func (s *casThenGateStore) releaseCAS() {
114 select {
115 case <-s.release:
116 default:
117 close(s.release)
118 }
119}
120
121// immediateConnector returns one healthy physical connection for any endpoint.
122func immediateConnector(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
123 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
124}
125
126func poolClientCount(p *Pool) int {
127 p.mu.Lock()
128 defer p.mu.Unlock()
129 return len(p.clients)
130}
131
132func registrySubscriptionCount(r *Registry) int {
133 r.mu.Lock()
134 defer r.mu.Unlock()
135 return len(r.subs)
136}
137
138func registryDemandCount(r *Registry) int {
139 r.mu.Lock()
140 defer r.mu.Unlock()
141 return r.demand
142}
143
144func TestAuthorityAdmissionRollback(t *testing.T) {
145 authority, _, pool, supervisor, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
146
147 first, err := authority.AdmitClient(context.Background())
148 require.NoError(t, err)
149 var epochFromID ports.BrokerEpoch
150 for i := 0; i < 8; i++ {
151 epochFromID = epochFromID<<8 | ports.BrokerEpoch(first.ConnectionID()[i])
152 }
153 require.Equal(t, ports.BrokerEpoch(1), epochFromID, "the pool identity must carry the broker epoch")
154
155 // Exhaust the pool's connection bound: the second admission must release the
156 // supervisor client lease it already took.
157 pool.limits.Clients = 2
158 _, err = authority.AdmitClient(context.Background())
159 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
160 clients, _, _, _ := supervisorState(supervisor)
161 require.Equal(t, 1, clients, "a refused admission must release its client lease")
162 require.Equal(t, 2, poolClientCount(pool), "a refused admission must not leave a pool client")
163
164 // A cancelled admission context takes no lease and registers no client.
165 ctx, cancel := context.WithCancel(context.Background())
166 cancel()
167 _, err = authority.AdmitClient(ctx)
168 require.ErrorIs(t, err, context.Canceled)
169 clients, _, _, _ = supervisorState(supervisor)
170 require.Equal(t, 1, clients)
171
172 require.NoError(t, first.Close())
173 clients, _, _, _ = supervisorState(supervisor)
174 require.Zero(t, clients)
175 require.Zero(t, poolClientCount(pool))
176
177 // A committed supervisor refuses every later admission without taking work.
178 require.NoError(t, supervisor.Close())
179 _, err = authority.AdmitClient(context.Background())
180 requireBrokerClosed(t, err)
181 clients, _, _, _ = supervisorState(supervisor)
182 require.Zero(t, clients)
183}
184
185func TestAuthorityAdmissionContextNotRetained(t *testing.T) {
186 authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
187 ctx, cancel := context.WithCancel(context.Background())
188 service, err := authority.AdmitClient(ctx)
189 require.NoError(t, err)
190 t.Cleanup(func() { require.NoError(t, service.Close()) })
191
192 // The admission context is discarded: the connection keeps serving after the
193 // caller's setup deadline expires.
194 cancel()
195 require.NoError(t, registry.setHosts(hostRecords(registration(t, "user@host:22", 1))))
196 require.Len(t, service.Snapshot().Daemons, 1)
197
198 stream, err := service.OpenEnvelopeStream(context.Background(), poolRequest(service.ConnectionID(), 1))
199 require.NoError(t, err)
200 require.NoError(t, stream.Close())
201}
202
203func TestAuthorityValidation(t *testing.T) {
204 _, _, pool, supervisor, clock, _ := newTestAuthority(t, 1, nil, immediateConnector)
205
206 // Live membership mutation is composed in this slice, so the authority no
207 // longer refuses a registry that owns remote membership its connections can
208 // now manage: a mutable remote registry is served.
209 mutableRemote, err := NewRegistryWithConfig(1, newTestStore(), newTestProbe(1), clock, nil, RegistryConfig{MembershipMode: MembershipMutable})
210 require.NoError(t, err)
211 t.Cleanup(mutableRemote.settle)
212 require.NoError(t, mutableRemote.setHosts(hostRecords(registration(t, "user@host:22", 1))))
213 _, err = NewAuthority(1, mutableRemote, pool, supervisor, testSessionCodec{})
214 require.NoError(t, err, "a mutable remote registry must be served")
215
216 // A local-only observer is served: its entry is configured authority rather
217 // than membership, so it has nothing for the authority's connections to
218 // manage either.
219 localOnly, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{Local: &LocalObservation{
220 DisplayOrigin: "local",
221 Policy: poolPolicy(),
222 Probe: newScriptedLocalProbe(1),
223 }})
224 require.NoError(t, err)
225 t.Cleanup(localOnly.settle)
226 _, err = NewAuthority(1, localOnly, pool, supervisor, testSessionCodec{})
227 require.NoError(t, err, "a local-only observer must be served")
228
229 // Construction still rejects remote observation without a probe: a registry
230 // that would observe remotes but owns no probe cannot be constructed, so no
231 // authority ever receives one.
232 _, err = NewRegistry(1, newTestStore(), nil, clock, nil)
233 require.ErrorContains(t, err, "invalid registry dependencies", "remote observation without a probe must be refused")
234
235 disabled, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{ObservationDisabled: true})
236 require.NoError(t, err)
237 t.Cleanup(disabled.settle)
238 _, err = NewAuthority(2, disabled, pool, supervisor, testSessionCodec{})
239 require.Error(t, err, "an epoch mismatch must be refused")
240
241 _, err = NewAuthority(1, nil, pool, supervisor, testSessionCodec{})
242 require.Error(t, err, "a missing registry must be refused")
243 _, err = NewAuthority(1, disabled, nil, supervisor, testSessionCodec{})
244 require.Error(t, err, "a missing pool must be refused")
245 _, err = NewAuthority(1, disabled, pool, nil, testSessionCodec{})
246 require.Error(t, err, "a missing supervisor must be refused")
247
248 _, err = NewAuthority(1, disabled, pool, supervisor, testSessionCodec{})
249 require.NoError(t, err)
250}
251
252func TestServiceOpenStreamScopeFencing(t *testing.T) {
253 foreign := ports.BrokerConnectionID{0xff}
254 tests := []struct {
255 name string
256 build func(id ports.BrokerConnectionID) ports.BrokerOpenStreamRequest
257 wantStale bool
258 wantEpoch bool
259 }{
260 {
261 name: "a foreign connection is fenced as stale",
262 build: func(ports.BrokerConnectionID) ports.BrokerOpenStreamRequest {
263 return poolRequest(foreign, 2)
264 },
265 wantStale: true,
266 },
267 {
268 name: "a foreign epoch is fenced as a stale epoch",
269 build: func(id ports.BrokerConnectionID) ports.BrokerOpenStreamRequest {
270 request := poolRequest(id, 3)
271 request.Epoch = 2
272 return request
273 },
274 wantEpoch: true,
275 },
276 {
277 name: "an absent identity is filled from the connection",
278 build: func(ports.BrokerConnectionID) ports.BrokerOpenStreamRequest {
279 request := poolRequest(ports.BrokerConnectionID{}, 4)
280 request.Epoch = 0
281 return request
282 },
283 },
284 }
285 for _, tc := range tests {
286 t.Run(tc.name, func(t *testing.T) {
287 authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
288 service, err := authority.AdmitClient(context.Background())
289 require.NoError(t, err)
290 t.Cleanup(func() { require.NoError(t, service.Close()) })
291
292 stream, err := service.OpenEnvelopeStream(context.Background(), tc.build(service.ConnectionID()))
293 switch {
294 case tc.wantStale:
295 require.ErrorIs(t, err, ports.BrokerAdmissionStale)
296 case tc.wantEpoch:
297 var typed ports.BrokerError
298 require.ErrorAs(t, err, &typed)
299 require.Equal(t, ports.BrokerErrorStaleEpoch, typed.Code)
300 default:
301 require.NoError(t, err)
302 require.NoError(t, stream.Close())
303 }
304 })
305 }
306}
307
308// TestServiceOpenStreamRequestsLocalReprobeOnOpenAndClose pins the "Extra"
309// fix from Phase C2: OpenStream marks a local re-probe pending as soon as a
310// local control/attachment stream opens, not only once it ends, so a session
311// created or claimed over this very stream can appear in `ls --all`/the
312// picker while the client stays attached, instead of waiting for detach. Both
313// transitions are proven against the registry's actual scheduling with no
314// clock advance at all, exactly like
315// TestRegistryLocalPendingReprobeHonorsExplicitReconcileOnly proves the
316// registry-side half of the mechanism.
317func TestServiceOpenStreamRequestsLocalReprobeOnOpenAndClose(t *testing.T) {
318 reachable := func() localProbeAnswer {
319 return localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
320 Identity: "local-daemon", Incarnation: ports.BrokerDaemonIncarnation{9},
321 ProtocolVersion: protocol.Version, Availability: domain.RemoteAvailabilityReachable,
322 }}
323 }
324
325 clock := newManualClock(time.Unix(100, 0))
326 store := newTestStore()
327 local := newScriptedLocalProbe(3)
328 registry, err := NewRegistryWithConfig(1, store, nil, clock, nil, RegistryConfig{
329 Local: &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: local},
330 })
331 require.NoError(t, err)
332 registry.jitter = identityJitter
333 authority, _, _, _ := composeTestAuthority(t, 1, registry, immediateConnector, clock)
334
335 service, err := authority.AdmitClient(context.Background())
336 require.NoError(t, err)
337 t.Cleanup(func() { require.NoError(t, service.Close()) })
338
339 // Explicitly request the initial observation; admission alone does not
340 // subscribe to snapshots and must not cause background polling.
341 registry.RequestProbe("")
342 call := receiveLocalCall(t, local)
343 call.result <- reachable()
344 waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
345
346 // Opening a local control stream must itself mark a re-probe pending.
347 stream, err := service.OpenEnvelopeStream(context.Background(), poolRequest(service.ConnectionID(), 1))
348 require.NoError(t, err)
349 opened := receiveLocalCall(t, local)
350 opened.result <- reachable()
351 waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
352
353 // Closing the stream marks a second re-probe pending, exactly as it did
354 // for stream completion before this fix.
355 require.NoError(t, stream.Close())
356 closed := receiveLocalCall(t, local)
357 closed.result <- reachable()
358}
359
360func TestServiceNextStreamIDAllocation(t *testing.T) {
361 authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
362 service, err := authority.AdmitClient(context.Background())
363 require.NoError(t, err)
364 t.Cleanup(func() { require.NoError(t, service.Close()) })
365
366 t.Run("monotone, nonzero, and never reused", func(t *testing.T) {
367 seen := make(map[ports.BrokerStreamID]bool)
368 previous := ports.BrokerStreamID(0)
369 for range 64 {
370 stream, err := service.NextStreamID()
371 require.NoError(t, err)
372 require.NotZero(t, stream, "an allocated identity is never zero")
373 require.Greater(t, stream, previous, "allocation is strictly monotone")
374 require.False(t, seen[stream], "an identity is never reused")
375 seen[stream] = true
376 previous = stream
377 }
378 })
379
380 t.Run("concurrent allocation is monotone and collision-free", func(t *testing.T) {
381 const count = 256
382 ids := make([]ports.BrokerStreamID, count)
383 errs := make([]error, count)
384 var wg sync.WaitGroup
385 for i := range count {
386 wg.Add(1)
387 go func() {
388 defer wg.Done()
389 ids[i], errs[i] = service.NextStreamID()
390 }()
391 }
392 wg.Wait()
393 seen := make(map[ports.BrokerStreamID]bool, count)
394 for i := range count {
395 require.NoError(t, errs[i])
396 require.NotZero(t, ids[i])
397 require.False(t, seen[ids[i]], "two concurrent allocations never collide")
398 seen[ids[i]] = true
399 }
400 })
401
402 t.Run("exhaustion refuses rather than wrapping", func(t *testing.T) {
403 exhausted, err := authority.AdmitClient(context.Background())
404 require.NoError(t, err)
405 t.Cleanup(func() { require.NoError(t, exhausted.Close()) })
406 exhausted.(*Service).nextStream = ^ports.BrokerStreamID(0)
407 _, err = exhausted.NextStreamID()
408 require.ErrorIs(t, err, ports.BrokerAdmissionLimit, "an exhausted counter never wraps to zero")
409 })
410
411 t.Run("a closed connection refuses", func(t *testing.T) {
412 closed, err := authority.AdmitClient(context.Background())
413 require.NoError(t, err)
414 require.NoError(t, closed.Close())
415 _, err = closed.NextStreamID()
416 require.ErrorIs(t, err, ports.BrokerAdmissionClosed)
417 })
418
419 t.Run("an allocated identity is carried on the request", func(t *testing.T) {
420 stream, err := service.NextStreamID()
421 require.NoError(t, err)
422 request := poolRequest(service.ConnectionID(), 0)
423 request.Stream = stream
424 opened, err := service.OpenEnvelopeStream(context.Background(), request)
425 require.NoError(t, err)
426 require.NoError(t, opened.Close())
427 // Replaying the exact identity is refused as stale.
428 _, err = service.OpenEnvelopeStream(context.Background(), request)
429 require.ErrorIs(t, err, ports.BrokerAdmissionStale)
430 })
431}
432
433func TestServiceOpenStreamAdmissionValidatesCallerInput(t *testing.T) {
434 authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
435 service, err := authority.AdmitClient(context.Background())
436 require.NoError(t, err)
437 t.Cleanup(func() { require.NoError(t, service.Close()) })
438
439 for _, tc := range []struct {
440 name string
441 mutate func(*ports.BrokerOpenStreamRequest)
442 }{
443 {"missing stream identity", func(r *ports.BrokerOpenStreamRequest) { r.Stream = 0 }},
444 {"invalid purpose", func(r *ports.BrokerOpenStreamRequest) { r.Purpose = 0 }},
445 {"invalid policy", func(r *ports.BrokerOpenStreamRequest) { r.Policy.Transport = "" }},
446 {"observation cannot start daemon", func(r *ports.BrokerOpenStreamRequest) { r.Purpose = ports.BrokerStreamObservation }},
447 } {
448 t.Run(tc.name, func(t *testing.T) {
449 request := poolRequest(service.ConnectionID(), 1)
450 tc.mutate(&request)
451 _, err := service.OpenEnvelopeStream(context.Background(), request)
452 require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
453 })
454 }
455}
456
457func TestServiceCloseStreamScopeFencing(t *testing.T) {
458 authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
459 service, err := authority.AdmitClient(context.Background())
460 require.NoError(t, err)
461 t.Cleanup(func() { require.NoError(t, service.Close()) })
462 id := service.ConnectionID()
463
464 stream, err := service.OpenEnvelopeStream(context.Background(), poolRequest(id, 1))
465 require.NoError(t, err)
466
467 // A wrong-scope close must not retire this connection's stream.
468 require.ErrorIs(t, service.CloseStream(ports.BrokerConnectionID{0xff}, 1), ports.BrokerAdmissionStale)
469 select {
470 case <-stream.Done():
471 t.Fatal("a wrong-scope close must not retire the stream")
472 default:
473 }
474 require.NoError(t, service.CloseStream(id, 1))
475 await(t, stream.Done())
476}
477
478func TestServiceSubscriptionCleanup(t *testing.T) {
479 authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
480 service, err := authority.AdmitClient(context.Background())
481 require.NoError(t, err)
482 t.Cleanup(func() { require.NoError(t, service.Close()) })
483
484 sub, err := service.Subscribe()
485 require.NoError(t, err)
486 require.Equal(t, 1, registrySubscriptionCount(registry))
487 sub.Close()
488 require.Zero(t, registrySubscriptionCount(registry), "an explicit close must release the subscription")
489
490 sub, err = service.Subscribe()
491 require.NoError(t, err)
492 require.Equal(t, 1, registrySubscriptionCount(registry))
493
494 require.NoError(t, service.Close())
495 require.Zero(t, registrySubscriptionCount(registry), "Close must release tracked subscriptions")
496 _, err = service.Subscribe()
497 requireBrokerClosed(t, err)
498}
499
500// TestServiceSubscribeDrivesRegistryDemand pins the C2 wiring: opening a
501// service-level subscription (an attached client keeps one open the whole
502// time it watches the snapshot) increments the registry's demand count, and
503// closing it decrements it back. This exercises the actual service.go call
504// sites (Subscribe / serviceSubscription.Close), not Registry.SetDemand
505// directly, and proves the count is balanced across concurrent subscriptions
506// and idempotent Close.
507func TestServiceSubscribeDrivesRegistryDemand(t *testing.T) {
508 authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
509 service, err := authority.AdmitClient(context.Background())
510 require.NoError(t, err)
511 t.Cleanup(func() { require.NoError(t, service.Close()) })
512
513 require.Zero(t, registryDemandCount(registry))
514
515 sub, err := service.Subscribe()
516 require.NoError(t, err)
517 require.Equal(t, 1, registryDemandCount(registry))
518
519 other, err := service.Subscribe()
520 require.NoError(t, err)
521 require.Equal(t, 2, registryDemandCount(registry), "a second live subscription must add a second unit of demand")
522
523 sub.Close()
524 require.Equal(t, 1, registryDemandCount(registry), "one closed subscription must leave the other's demand live")
525
526 other.Close()
527 require.Zero(t, registryDemandCount(registry))
528
529 // Close is idempotent: a repeated close must never double-release demand.
530 other.Close()
531 require.Zero(t, registryDemandCount(registry))
532}
533
534// TestServiceSubscribeDemandByReader pins the Amendment 2 rule "reading state
535// forces nothing": a passive subscription (vev ls, vev host list) receives
536// publications but never records demand, while a watching client does.
537func TestServiceSubscribeDemandByReader(t *testing.T) {
538 tests := []struct {
539 name string
540 passive bool
541 wantDemand int
542 }{
543 {name: "watching client records demand", passive: false, wantDemand: 1},
544 {name: "passive reader records none", passive: true, wantDemand: 0},
545 }
546 for _, tt := range tests {
547 t.Run(tt.name, func(t *testing.T) {
548 authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
549 admitted, err := authority.AdmitClient(context.Background())
550 require.NoError(t, err)
551 t.Cleanup(func() { require.NoError(t, admitted.Close()) })
552 service := admitted.(*Service)
553
554 subscribe := service.Subscribe
555 if tt.passive {
556 subscribe = service.SubscribePassive
557 }
558 sub, err := subscribe()
559 require.NoError(t, err)
560 require.Equal(t, tt.wantDemand, registryDemandCount(registry))
561
562 require.NoError(t, registry.setHosts(hostRecords(registration(t, "user@host:22", 1))))
563 select {
564 case <-sub.Changed():
565 case <-time.After(time.Second):
566 t.Fatal("every subscription must observe registry publications")
567 }
568
569 sub.Close()
570 sub.Close()
571 require.Zero(t, registryDemandCount(registry), "closing must balance exactly the demand it recorded")
572 })
573 }
574}
575
576func TestServiceSnapshotDelegates(t *testing.T) {
577 authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
578 service, err := authority.AdmitClient(context.Background())
579 require.NoError(t, err)
580 t.Cleanup(func() { require.NoError(t, service.Close()) })
581
582 sub, err := service.Subscribe()
583 require.NoError(t, err)
584 defer sub.Close()
585
586 require.NoError(t, registry.setHosts(hostRecords(registration(t, "user@host:22", 1))))
587 select {
588 case <-sub.Changed():
589 case <-time.After(time.Second):
590 t.Fatal("the connection subscription must observe a registry publication")
591 }
592 require.Equal(t, registry.Snapshot(), service.Snapshot())
593 require.Len(t, service.Snapshot().Daemons, 1)
594}
595
596func TestServiceOperationLeaseLifetime(t *testing.T) {
597 entered := make(chan struct{})
598 proceed := make(chan struct{})
599 authority, _, _, supervisor, _, _ := newTestAuthority(t, 1, nil, func(ctx context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
600 close(entered)
601 select {
602 case <-ctx.Done():
603 return nil, ctx.Err()
604 case <-proceed:
605 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
606 }
607 })
608 service, err := authority.AdmitClient(context.Background())
609 require.NoError(t, err)
610 t.Cleanup(func() { require.NoError(t, service.Close()) })
611
612 type opened struct {
613 stream ports.BrokerEnvelopeStream
614 err error
615 }
616 result := make(chan opened, 1)
617 go func() {
618 stream, err := service.OpenEnvelopeStream(context.Background(), poolRequest(service.ConnectionID(), 1))
619 result <- opened{stream: stream, err: err}
620 }()
621 <-entered
622 _, operations, _, _ := supervisorState(supervisor)
623 require.Equal(t, 1, operations, "open setup must hold one operation lease")
624
625 close(proceed)
626 got := <-result
627 require.NoError(t, got.err)
628 require.NotNil(t, got.stream)
629 _, operations, _, _ = supervisorState(supervisor)
630 require.Zero(t, operations, "the operation lease must be released once setup completes")
631 select {
632 case <-got.stream.Done():
633 t.Fatal("a stream that opened successfully must outlive its setup lease")
634 default:
635 }
636 require.NoError(t, got.stream.Close())
637 await(t, got.stream.Done())
638}
639
640func TestServiceCloseOpenRace(t *testing.T) {
641 for round := 0; round < 25; round++ {
642 t.Run(fmt.Sprintf("round-%d", round), func(t *testing.T) {
643 entered := make(chan struct{})
644 proceed := make(chan struct{})
645 authority, _, pool, supervisor, _, _ := newTestAuthority(t, 1, nil, func(ctx context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
646 close(entered)
647 select {
648 case <-ctx.Done():
649 return nil, ctx.Err()
650 case <-proceed:
651 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
652 }
653 })
654 service, err := authority.AdmitClient(context.Background())
655 require.NoError(t, err)
656
657 result := make(chan error, 1)
658 go func() {
659 _, err := service.OpenEnvelopeStream(context.Background(), poolRequest(service.ConnectionID(), 1))
660 result <- err
661 }()
662 <-entered
663
664 // Close races the parked setup: whichever wins, both must settle
665 // without deadlock and leave no lease or client behind.
666 closed := make(chan error, 1)
667 go func() { closed <- service.Close() }()
668 close(proceed)
669 require.NoError(t, <-closed)
670 <-result
671
672 clients, operations, _, _ := supervisorState(supervisor)
673 require.Zero(t, clients)
674 require.Zero(t, operations)
675 require.Zero(t, poolClientCount(pool))
676 })
677 }
678}
679
680func TestServiceCloseIsIdempotentAndConcurrent(t *testing.T) {
681 authority, registry, pool, supervisor, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
682 service, err := authority.AdmitClient(context.Background())
683 require.NoError(t, err)
684 _, err = service.Subscribe()
685 require.NoError(t, err)
686 stream, err := service.OpenEnvelopeStream(context.Background(), poolRequest(service.ConnectionID(), 1))
687 require.NoError(t, err)
688
689 const closers = 8
690 errs := make(chan error, closers)
691 var wg sync.WaitGroup
692 for i := 0; i < closers; i++ {
693 wg.Add(1)
694 go func() {
695 defer wg.Done()
696 errs <- service.Close()
697 }()
698 }
699 wg.Wait()
700 for i := 0; i < closers; i++ {
701 require.NoError(t, <-errs)
702 }
703 require.NoError(t, service.Close())
704
705 require.Zero(t, registrySubscriptionCount(registry))
706 require.Zero(t, poolClientCount(pool))
707 clients, _, _, _ := supervisorState(supervisor)
708 require.Zero(t, clients)
709 await(t, stream.Done())
710
711 _, err = service.OpenEnvelopeStream(context.Background(), poolRequest(service.ConnectionID(), 2))
712 requireBrokerClosed(t, err)
713}
714
715func TestServiceDoneErrContract(t *testing.T) {
716 t.Run("local close is orderly", func(t *testing.T) {
717 authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
718 service, err := authority.AdmitClient(context.Background())
719 require.NoError(t, err)
720
721 select {
722 case <-service.Done():
723 t.Fatal("Done closed before any terminal event")
724 default:
725 }
726 require.NoError(t, service.Close())
727 await(t, service.Done())
728 require.NoError(t, service.Err(), "an orderly local Close reports no terminal cause")
729 require.NoError(t, service.Err(), "Err is stable after Done")
730 })
731
732 t.Run("broker shutdown is a typed loss", func(t *testing.T) {
733 authority, _, _, supervisor, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
734 service, err := authority.AdmitClient(context.Background())
735 require.NoError(t, err)
736
737 require.NoError(t, supervisor.Close())
738 await(t, service.Done())
739 var typed ports.BrokerError
740 require.ErrorAs(t, service.Err(), &typed)
741 require.Equal(t, ports.BrokerErrorUnavailable, typed.Code)
742 })
743}
744
745func TestServiceShutdownTerminatesOwnedWork(t *testing.T) {
746 authority, _, _, supervisor, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
747 service, err := authority.AdmitClient(context.Background())
748 require.NoError(t, err)
749
750 stream, err := service.OpenEnvelopeStream(context.Background(), poolRequest(service.ConnectionID(), 1))
751 require.NoError(t, err)
752
753 // Broker shutdown cancels the connection root: owned streams end even before
754 // the session gets to call Close.
755 require.NoError(t, supervisor.Close())
756 await(t, stream.Done())
757
758 require.NoError(t, service.Close())
759 clients, _, _, _ := supervisorState(supervisor)
760 require.Zero(t, clients)
761}
762
763// TestServiceImmutableMembershipRefusesEveryMutation proves the immutable
764// connection refuses all three membership mutations with the typed immutable
765// code 12 and the ports sentinel before any durable authority moves: no CAS is
766// attempted and no publication changes.
767func TestServiceImmutableMembershipRefusesEveryMutation(t *testing.T) {
768 store := newMembershipStore()
769 authority, registry, _, _ := newServiceMembershipFixture(t, store, MembershipImmutable)
770 service, err := authority.AdmitClient(context.Background())
771 require.NoError(t, err)
772 t.Cleanup(func() { require.NoError(t, service.Close()) })
773
774 before := registry.Snapshot()
775 _, err = service.AddHost(context.Background(), "new@host:22", poolPolicy())
776 requireImmutableRefusal(t, err)
777
778 removed, err := service.RemoveHost(context.Background(), registration(t, "new@host:22", 1))
779 require.False(t, removed)
780 requireImmutableRefusal(t, err)
781
782 updated, err := service.UpdateHostPolicy(context.Background(), registration(t, "new@host:22", 1), policyWithTrust("changed-trust"))
783 require.Equal(t, domain.RemoteRegistration{}, updated)
784 requireImmutableRefusal(t, err)
785
786 require.Equal(t, 0, store.casCalls(), "an immutable refusal never reaches the store CAS")
787 require.Equal(t, before, registry.Snapshot(), "a refused mutation must not touch authority")
788
789 service.RequestReconcile("new@host:22")
790 require.Equal(t, before, registry.Snapshot(), "an unknown reconcile hint cannot change authority")
791}
792
793func TestServiceRequestReconcileDelegatesToRegistry(t *testing.T) {
794 clock := newManualClock(time.Unix(0, 0))
795 store := newTestStore()
796 registration := registration(t, "user@host:22", 1)
797 store.hosts = ports.BrokerHosts{Revision: 1, Hosts: []ports.BrokerHostRecord{{Registration: registration, Pinned: true, Policy: poolPolicy(), Route: canonicalRoute(poolPolicy(), registration.Endpoint)}}}
798 probe := newTestProbe(2)
799 registry, err := NewRegistry(1, store, probe, clock, nil)
800 require.NoError(t, err)
801 registry.jitter = identityJitter
802 authority, _, _, _ := composeTestAuthority(t, 1, registry, immediateConnector, clock)
803 service, err := authority.AdmitClient(context.Background())
804 require.NoError(t, err)
805 t.Cleanup(func() { require.NoError(t, service.Close()) })
806
807 registry.RequestProbe(registration.Endpoint)
808 var initial probeCall
809 require.Eventually(t, func() bool {
810 select {
811 case initial = <-probe.calls:
812 return true
813 default:
814 return false
815 }
816 }, time.Second, time.Millisecond)
817 initial.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Identity: "daemon", Incarnation: ports.BrokerDaemonIncarnation{1}, ProtocolVersion: 1, Availability: domain.RemoteAvailabilityReachable}}
818 service.RequestReconcile(registration.Endpoint)
819 var reconciled probeCall
820 require.Eventually(t, func() bool {
821 select {
822 case reconciled = <-probe.calls:
823 return true
824 default:
825 return false
826 }
827 }, time.Second, time.Millisecond)
828 require.Equal(t, registration, reconciled.registration)
829}
830
831// TestServiceMutableMembershipDelegatesOnce proves an admitted mutable
832// connection delegates each mutation exactly once to durable authority: each
833// call crosses the store CAS exactly once and returns the registry's own
834// result.
835func TestServiceMutableMembershipDelegatesOnce(t *testing.T) {
836 store := newMembershipStore()
837 authority, registry, _, _ := newServiceMembershipFixture(t, store, MembershipMutable)
838 service, err := authority.AdmitClient(context.Background())
839 require.NoError(t, err)
840 t.Cleanup(func() { require.NoError(t, service.Close()) })
841
842 added, err := service.AddHost(context.Background(), "user@host:22", poolPolicy())
843 require.NoError(t, err)
844 require.False(t, added.IsZero())
845 require.Equal(t, 1, store.casCalls(), "one addition crosses the store CAS exactly once")
846
847 updated, err := service.UpdateHostPolicy(context.Background(), added, policyWithTrust("changed-trust"))
848 require.NoError(t, err)
849 require.Equal(t, added.Generation+1, updated.Generation)
850 require.Equal(t, added.Incarnation, updated.Incarnation)
851 require.Equal(t, 2, store.casCalls(), "one policy update crosses the store CAS exactly once")
852
853 removed, err := service.RemoveHost(context.Background(), updated)
854 require.NoError(t, err)
855 require.True(t, removed)
856 require.Equal(t, 3, store.casCalls(), "one removal crosses the store CAS exactly once")
857 require.Empty(t, registry.Snapshot().Daemons)
858}
859
860// TestServiceMutationLeasePinsIdleLifecycle proves an in-flight membership
861// mutation holds one operation lease: the idle timer stays disarmed while the
862// mutation is parked past the grace, and the broker expires only after it
863// returns and the lease is released.
864func TestServiceMutationLeasePinsIdleLifecycle(t *testing.T) {
865 store := newCASThenGateStore()
866 authority, _, supervisor, idleClock := newServiceMembershipFixture(t, store, MembershipMutable)
867 service, err := authority.AdmitClient(context.Background())
868 require.NoError(t, err)
869 t.Cleanup(func() { require.NoError(t, service.Close()) })
870
871 type outcome struct {
872 registration domain.RemoteRegistration
873 err error
874 }
875 result := make(chan outcome, 1)
876 go func() {
877 registration, err := service.AddHost(context.Background(), "user@host:22", poolPolicy())
878 result <- outcome{registration: registration, err: err}
879 }()
880
881 // The CAS has committed and the mutation is parked, so its operation lease is
882 // held: the idle timer is disarmed even well past the grace.
883 select {
884 case <-store.committed:
885 case <-time.After(5 * time.Second):
886 t.Fatal("the mutation never reached the store CAS")
887 }
888 _, operations, armed, _ := supervisorState(supervisor)
889 require.Equal(t, 1, operations, "the in-flight mutation holds one operation lease")
890 require.False(t, armed, "a held operation lease disarms the idle timer")
891 idleClock.Advance(time.Hour * 24)
892 requireSupervisorOpen(t, supervisor)
893
894 store.releaseCAS()
895 got := <-result
896 require.NoError(t, got.err)
897 require.False(t, got.registration.IsZero())
898
899 // The lease is released: the idle timer re-arms and the broker expires a
900 // plain grace later. The connection still holds its client lease, so drop it
901 // first.
902 require.NoError(t, service.Close())
903 _, operations, armed, _ = supervisorState(supervisor)
904 require.Zero(t, operations)
905 require.True(t, armed)
906 idleClock.Advance(time.Hour)
907 await(t, supervisor.Done())
908}
909
910// TestServiceCloseCancelsInFlightRegistryMutation proves Close cancels and
911// drains an in-flight registry mutation: the connection context reaches the
912// registry's mutation context, the caller observes the typed cancellation, and
913// Close returns only after the mutation has joined.
914func TestServiceCloseCancelsInFlightRegistryMutation(t *testing.T) {
915 store := newMembershipStore()
916 authority, registry, supervisor, _ := newServiceMembershipFixture(t, store, MembershipMutable)
917 service, err := authority.AdmitClient(context.Background())
918 require.NoError(t, err)
919
920 // Hold the registry lock so the mutation parks before it consults its
921 // context, and Close can cancel the connection context while it waits.
922 registry.mu.Lock()
923 released := false
924 t.Cleanup(func() {
925 if !released {
926 registry.mu.Unlock()
927 }
928 })
929
930 type outcome struct {
931 removed bool
932 err error
933 }
934 result := make(chan outcome, 1)
935 go func() {
936 removed, err := service.RemoveHost(context.Background(), registration(t, "user@host:22", 1))
937 result <- outcome{removed: removed, err: err}
938 }()
939
940 // Wait until the mutation holds its operation lease, which is strictly after
941 // it registered with the connection's drain group.
942 require.Eventually(t, func() bool {
943 _, operations, _, _ := supervisorState(supervisor)
944 return operations == 1
945 }, 5*time.Second, time.Millisecond, "the mutation never reached its operation lease")
946
947 // Close cancels the connection context and waits for the in-flight mutation
948 // to drain. It cannot return while the mutation is parked.
949 closed := make(chan error, 1)
950 go func() { closed <- service.Close() }()
951 select {
952 case <-closed:
953 t.Fatal("Close returned before the in-flight mutation drained")
954 case <-time.After(50 * time.Millisecond):
955 }
956
957 // Release the registry lock: the mutation now observes the cancelled
958 // connection context instead of committing.
959 registry.mu.Unlock()
960 released = true
961
962 got := <-result
963 require.ErrorIs(t, got.err, context.Canceled, "the in-flight mutation observes the connection cancellation")
964 require.False(t, got.removed)
965 select {
966 case err := <-closed:
967 require.NoError(t, err)
968 case <-time.After(5 * time.Second):
969 t.Fatal("Close never returned after the mutation drained")
970 }
971 require.Zero(t, store.casCalls(), "a cancelled mutation never reaches the store CAS")
972 require.Empty(t, registry.Snapshot().Daemons, "a cancelled mutation never mutates authority")
973}
974
975// TestServiceMutationCancellationAfterCommitStillReportsSuccess proves a caller
976// cancellation that arrives strictly after the durable CAS still reports the
977// committed success: the registry's mutation context is not consulted after
978// commit, so the value crosses the CAS and returns rather than surfacing a
979// cancellation for a mutation that did commit.
980func TestServiceMutationCancellationAfterCommitStillReportsSuccess(t *testing.T) {
981 store := newCASThenGateStore()
982 authority, _, _, _ := newServiceMembershipFixture(t, store, MembershipMutable)
983 service, err := authority.AdmitClient(context.Background())
984 require.NoError(t, err)
985 t.Cleanup(func() { require.NoError(t, service.Close()) })
986
987 ctx, cancel := context.WithCancel(context.Background())
988 type outcome struct {
989 registration domain.RemoteRegistration
990 err error
991 }
992 result := make(chan outcome, 1)
993 go func() {
994 registration, err := service.AddHost(ctx, "user@host:22", poolPolicy())
995 result <- outcome{registration: registration, err: err}
996 }()
997
998 select {
999 case <-store.committed:
1000 case <-time.After(5 * time.Second):
1001 t.Fatal("the mutation never reached the store CAS")
1002 }
1003 // The commit already happened; a cancellation now cannot un-commit it.
1004 cancel()
1005 store.releaseCAS()
1006
1007 got := <-result
1008 require.NoError(t, got.err, "a cancellation after the CAS must not turn a committed mutation into a failure")
1009 require.False(t, got.registration.IsZero())
1010}
1011
1012// TestServiceMembershipErrorMapping proves the service classifies a durable
1013// store outcome-unknown as the typed unknown code with its cause preserved, and
1014// surfaces the exact conflict code for a conflicting authority change.
1015func TestServiceMembershipErrorTextIsWireSafe(t *testing.T) {
1016 secretPath := "/home/alice/.config/vev/private-hosts.json"
1017 cause := errors.New(secretPath + "\x00\n" + strings.Repeat("é", ports.BrokerMaxErrorBytes))
1018 err := membershipError(ports.BrokerStoreOutcomeUnknownError{Err: cause})
1019 var failure ports.BrokerError
1020 require.ErrorAs(t, err, &failure)
1021 require.NoError(t, failure.Validate())
1022 require.LessOrEqual(t, len(failure.Text), ports.BrokerMaxErrorBytes)
1023 require.NotContains(t, failure.Text, "\x00")
1024 require.NotContains(t, failure.Text, "\n")
1025 require.NotContains(t, failure.Text, secretPath)
1026 require.Equal(t, "mutation outcome is unknown", failure.Text)
1027 require.ErrorIs(t, failure, cause)
1028 require.True(t, utf8.ValidString(failure.Text))
1029}
1030
1031func TestServiceMembershipErrorMapping(t *testing.T) {
1032 t.Run("outcome unknown maps to outcome-unknown with its cause", func(t *testing.T) {
1033 cause := errors.New("power loss")
1034 store := newMembershipStore()
1035 store.failCAS(ports.BrokerStoreOutcomeUnknownError{Err: cause})
1036 authority, _, _, _ := newServiceMembershipFixture(t, store, MembershipMutable)
1037 service, err := authority.AdmitClient(context.Background())
1038 require.NoError(t, err)
1039 t.Cleanup(func() { require.NoError(t, service.Close()) })
1040
1041 _, err = service.AddHost(context.Background(), "user@host:22", poolPolicy())
1042 require.Error(t, err)
1043 var failure ports.BrokerError
1044 require.ErrorAs(t, err, &failure)
1045 require.Equal(t, ports.BrokerErrorOutcomeUnknown, failure.Code)
1046 require.ErrorIs(t, err, cause, "the underlying store cause stays inspectable")
1047 var unknown ports.BrokerStoreOutcomeUnknownError
1048 require.ErrorAs(t, err, &unknown)
1049 })
1050
1051 t.Run("an exact conflict maps to code 11", func(t *testing.T) {
1052 store := newMembershipStore()
1053 store.failCAS(ports.ErrBrokerHostConflict)
1054 authority, _, _, _ := newServiceMembershipFixture(t, store, MembershipMutable)
1055 service, err := authority.AdmitClient(context.Background())
1056 require.NoError(t, err)
1057 t.Cleanup(func() { require.NoError(t, service.Close()) })
1058
1059 _, err = service.UpdateHostPolicy(context.Background(), registration(t, "user@host:22", 1), poolPolicy())
1060 require.Error(t, err)
1061 var failure ports.BrokerError
1062 require.ErrorAs(t, err, &failure)
1063 require.Equal(t, ports.BrokerErrorHostConflict, failure.Code)
1064 require.Equal(t, ports.BrokerErrorCode(11), failure.Code, "host conflict is the exact code 11")
1065 require.ErrorIs(t, err, ports.ErrBrokerHostConflict)
1066 })
1067}
1068
1069func TestRegistryObservationDisabledRestoresAndPublishes(t *testing.T) {
1070 clock := newManualClock(time.Unix(0, 0))
1071 store := newTestStore()
1072 endpoint := "user@host:22"
1073 store.loaded = ports.BrokerSnapshot{
1074 Epoch: 9,
1075 Revision: 4,
1076 Daemons: []ports.BrokerDaemonObservation{{
1077 Endpoint: endpoint,
1078 Registration: registration(t, endpoint, 1),
1079 Availability: domain.RemoteAvailabilityReachable,
1080 }},
1081 }
1082 r, err := NewRegistryWithConfig(1, store, nil, clock, nil, RegistryConfig{ObservationDisabled: true})
1083 require.NoError(t, err)
1084
1085 // The disabled registry restores durable state under its fresh epoch and
1086 // publishes it without observing anything.
1087 snapshot := r.Snapshot()
1088 require.Equal(t, ports.BrokerEpoch(1), snapshot.Epoch)
1089 require.EqualValues(t, 1, snapshot.Revision)
1090 host, ok := snapshot.Find(endpoint)
1091 require.True(t, ok)
1092 require.Equal(t, domain.RemoteAvailabilityReachable, host.Availability)
1093 require.Zero(t, clock.activeTimers())
1094}
1095
1096func TestRegistryObservationDisabledIssuesNoProbesOrTimers(t *testing.T) {
1097 clock := newManualClock(time.Unix(0, 0))
1098 store := newTestStore()
1099 probe := newTestProbe(4)
1100 r, err := NewRegistryWithConfig(1, store, probe, clock, nil, RegistryConfig{ObservationDisabled: true})
1101 require.NoError(t, err)
1102 require.NoError(t, r.setHosts(hostRecords(registration(t, "user@host:22", 1))))
1103
1104 ctx, cancel := context.WithCancel(context.Background())
1105 done := make(chan struct{})
1106 go func() { defer close(done); r.Run(ctx) }()
1107
1108 // A reconcile/probe demand must schedule nothing: no probe starts and no
1109 // timer is armed, so advancing the clock cannot produce observation work.
1110 r.RequestProbe("user@host:22")
1111 clock.Advance(time.Hour)
1112 require.Zero(t, clock.activeTimers())
1113 require.Empty(t, probe.calls)
1114
1115 cancel()
1116 await(t, done)
1117 require.Zero(t, clock.activeTimers())
1118 require.Empty(t, probe.calls)
1119 require.Equal(t, runStopped, r.running.Load())
1120}
1121
1122func TestRegistryObservationDisabledDrainsWriter(t *testing.T) {
1123 clock := newManualClock(time.Unix(0, 0))
1124 store := newBlockingStore()
1125 r, err := NewRegistryWithConfig(1, store, nil, clock, nil, RegistryConfig{ObservationDisabled: true})
1126 require.NoError(t, err)
1127
1128 ctx, cancel := context.WithCancel(context.Background())
1129 runDone := make(chan struct{})
1130 go func() { defer close(runDone); r.Run(ctx) }()
1131
1132 require.NoError(t, r.ReplaceHosts([]ports.BrokerHostRecord{{
1133 Registration: registration(t, "user@host:22", 1),
1134 Pinned: true,
1135 Policy: poolPolicy(),
1136 Route: canonicalRoute(poolPolicy(), "user@host:22"),
1137 }}))
1138 select {
1139 case <-store.entered:
1140 case <-time.After(time.Second):
1141 t.Fatal("the durable writer never attempted a write")
1142 }
1143
1144 // Run owns the writer: it must not settle while a durable write is in
1145 // flight, and it must still stop once that write drains.
1146 cancel()
1147 select {
1148 case <-runDone:
1149 t.Fatal("Run must not return before the in-flight durable write drains")
1150 case <-time.After(probeGrace):
1151 }
1152 store.releaseFirst()
1153 await(t, runDone)
1154 require.NotEmpty(t, store.writtenRevisions())
1155 require.Zero(t, clock.activeTimers())
1156}
1157
1158func TestServiceConfigValidation(t *testing.T) {
1159 valid := serviceConfig{epoch: 1, id: ports.BrokerConnectionID{1}, previewID: ports.BrokerConnectionID{2}, registry: &Registry{}, pool: &Pool{}, supervisor: &Supervisor{}, lease: &Lease{}, clock: newManualClock(time.Unix(0, 0)), codec: testSessionCodec{}}
1160 tests := []struct {
1161 name string
1162 mutate func(*serviceConfig)
1163 }{
1164 {"zero epoch", func(c *serviceConfig) { c.epoch = 0 }},
1165 {"zero id", func(c *serviceConfig) { c.id = ports.BrokerConnectionID{} }},
1166 {"zero preview", func(c *serviceConfig) { c.previewID = ports.BrokerConnectionID{} }},
1167 {"same ids", func(c *serviceConfig) { c.previewID = c.id }},
1168 {"nil registry", func(c *serviceConfig) { c.registry = nil }},
1169 {"nil pool", func(c *serviceConfig) { c.pool = nil }},
1170 {"nil supervisor", func(c *serviceConfig) { c.supervisor = nil }},
1171 {"nil lease", func(c *serviceConfig) { c.lease = nil }},
1172 {"nil clock", func(c *serviceConfig) { c.clock = nil }},
1173 {"nil codec", func(c *serviceConfig) { c.codec = nil }},
1174 }
1175 require.NoError(t, valid.validate())
1176 for _, tt := range tests {
1177 t.Run(tt.name, func(t *testing.T) {
1178 config := valid
1179 tt.mutate(&config)
1180 require.Error(t, config.validate())
1181 })
1182 }
1183}
1184
1185// testSessionCodec is used only for scripted preview streams; raw traffic is never decoded.
1186type testSessionCodec struct{}
1187
1188func (testSessionCodec) Client(stream ports.BrokerEnvelopeStream) ports.ClientConnection {
1189 if pooled, ok := stream.(*pooledStream); ok {
1190 stream = pooled.BrokerEnvelopeStream
1191 }
1192 return stream.(*previewWatchConn)
1193}