1package client
2
3import (
4 "context"
5 "errors"
6 "sync/atomic"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/require"
11
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/internal/protocol"
15)
16
17// Closed initial-navigation union (Plan 001 P7, task 2). These tables pin the
18// four variants locally and remotely, the zero Picker, the contradiction rules,
19// the exact-registration equality fence, the one-shot consumption, and the
20// "a reconnect never replays" contract.
21
22func validInitialNavigationTarget() protocol.ExactSessionTarget {
23 return protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(2), SessionName: "alpha"}
24}
25
26func validInitialNavigationRemoteFence() ports.BrokerEndpointFence {
27 return ports.BrokerEndpointFence{Registration: pickerTestRegistration("user@arch", 1)}
28}
29
30// TestInitialNavigationValidationClosedUnion pins the closed union's offline
31// validation: the four variants and every contradictory field combination.
32func TestInitialNavigationValidationClosedUnion(t *testing.T) {
33 target := validInitialNavigationTarget()
34 remote := validInitialNavigationRemoteFence()
35
36 tests := []struct {
37 name string
38 navigation InitialNavigation
39 wantErr bool
40 }{
41 {
42 name: "zero value is the safe picker",
43 navigation: InitialNavigation{},
44 },
45 {
46 name: "picker with a local destination contradicts",
47 navigation: InitialNavigation{Kind: InitialNavigationPicker, Destination: ports.BrokerEndpointFence{Local: true}},
48 wantErr: true,
49 },
50 {
51 name: "picker with an epoch contradicts",
52 navigation: InitialNavigation{Kind: InitialNavigationPicker, Epoch: 3},
53 wantErr: true,
54 },
55 {
56 name: "picker with a name contradicts",
57 navigation: InitialNavigation{Kind: InitialNavigationPicker, Name: "alpha"},
58 wantErr: true,
59 },
60 {
61 name: "picker with a target contradicts",
62 navigation: InitialNavigation{Kind: InitialNavigationPicker, Target: target},
63 wantErr: true,
64 },
65 {
66 name: "local ephemeral creation may omit the epoch",
67 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}},
68 },
69 {
70 name: "local named creation may omit the epoch",
71 navigation: InitialNavigation{Kind: InitialNavigationCreateNamed, Destination: ports.BrokerEndpointFence{Local: true}, Name: "fresh"},
72 },
73 {
74 name: "remote creation requires the resolving epoch",
75 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: remote},
76 wantErr: true,
77 },
78 {
79 name: "remote creation with the epoch is valid",
80 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Epoch: 3, Destination: remote},
81 },
82 {
83 name: "ephemeral creation on a bare registration contradicts",
84 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{}},
85 wantErr: true,
86 },
87 {
88 name: "creation with a target contradicts",
89 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}, Target: target},
90 wantErr: true,
91 },
92 {
93 name: "ephemeral creation with a name contradicts",
94 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}, Name: "alpha"},
95 wantErr: true,
96 },
97 {
98 name: "named creation with an invalid name contradicts",
99 navigation: InitialNavigation{Kind: InitialNavigationCreateNamed, Destination: ports.BrokerEndpointFence{Local: true}, Name: "bad name"},
100 wantErr: true,
101 },
102 {
103 name: "named creation with a target contradicts",
104 navigation: InitialNavigation{Kind: InitialNavigationCreateNamed, Destination: ports.BrokerEndpointFence{Local: true}, Name: "fresh", Target: target},
105 wantErr: true,
106 },
107 {
108 name: "local exact attach still requires an epoch",
109 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Destination: ports.BrokerEndpointFence{Local: true}, Target: target},
110 wantErr: true,
111 },
112 {
113 name: "local exact attach with an epoch is valid",
114 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true}, Target: target},
115 },
116 {
117 name: "remote exact attach with an epoch is valid",
118 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: 3, Destination: remote, Target: target},
119 },
120 {
121 name: "exact attach with a name contradicts",
122 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true}, Name: "alpha", Target: target},
123 wantErr: true,
124 },
125 {
126 name: "exact attach with an invalid target contradicts",
127 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true}, Target: protocol.ExactSessionTarget{SessionName: "alpha"}},
128 wantErr: true,
129 },
130 {
131 name: "unknown kind is refused",
132 navigation: InitialNavigation{Kind: InitialNavigationKind(9), Destination: ports.BrokerEndpointFence{Local: true}},
133 wantErr: true,
134 },
135 {
136 name: "local destination carrying a registration contradicts",
137 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true, Registration: pickerTestRegistration("user@arch", 1)}},
138 wantErr: true,
139 },
140 }
141 for _, tt := range tests {
142 t.Run(tt.name, func(t *testing.T) {
143 err := tt.navigation.Validate()
144 if tt.wantErr {
145 require.Error(t, err)
146 return
147 }
148 require.NoError(t, err)
149 })
150 }
151}
152
153// TestInitialNavigationResolveVariants pins the exact request each variant
154// produces, locally and remotely: purpose, admission, destination authority,
155// policy provenance, and the exclusive name/target fields.
156func TestInitialNavigationResolveVariants(t *testing.T) {
157 now := time.Unix(1000, 0)
158 target := validInitialNavigationTarget()
159 remote := validInitialNavigationRemoteFence()
160
161 tests := []struct {
162 name string
163 navigation InitialNavigation
164 snapshot ports.BrokerSnapshot
165 wantErr pickerCatalogueErrorCode
166 wantLocal bool
167 wantAdmission ports.BrokerStreamAdmission
168 wantName string
169 wantTarget protocol.ExactSessionTarget
170 }{
171 {
172 name: "local ephemeral creation",
173 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}},
174 snapshot: localDaemonSnapshot(3, 1),
175 wantLocal: true, wantAdmission: ports.BrokerAdmissionCreateEphemeral,
176 },
177 {
178 name: "local named creation",
179 navigation: InitialNavigation{Kind: InitialNavigationCreateNamed, Destination: ports.BrokerEndpointFence{Local: true}, Name: "fresh"},
180 snapshot: localDaemonSnapshot(3, 1),
181 wantLocal: true, wantAdmission: ports.BrokerAdmissionCreateNamed, wantName: "fresh",
182 },
183 {
184 name: "local exact attach",
185 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true}, Target: target},
186 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
187 pickerTestLocalObservation(now, pickerTestSessionWithLifecycle("alpha", target.LifecycleID)),
188 }},
189 wantLocal: true, wantAdmission: ports.BrokerAdmissionExact, wantTarget: target,
190 },
191 {
192 name: "remote ephemeral creation",
193 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Epoch: 3, Destination: remote},
194 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
195 pickerTestRemoteObservation("user@arch", 1, 1, now),
196 }},
197 wantAdmission: ports.BrokerAdmissionCreateEphemeral,
198 },
199 {
200 name: "remote named creation",
201 navigation: InitialNavigation{Kind: InitialNavigationCreateNamed, Epoch: 3, Destination: remote, Name: "fresh"},
202 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
203 pickerTestRemoteObservation("user@arch", 1, 1, now),
204 }},
205 wantAdmission: ports.BrokerAdmissionCreateNamed, wantName: "fresh",
206 },
207 {
208 name: "remote exact attach",
209 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: 3, Destination: remote, Target: target},
210 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
211 pickerTestRemoteObservation("user@arch", 1, 1, now, pickerTestSessionWithLifecycle("alpha", target.LifecycleID)),
212 }},
213 wantAdmission: ports.BrokerAdmissionExact, wantTarget: target,
214 },
215 {
216 name: "local creation on a stopped daemon stays attemptable",
217 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}},
218 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
219 func() ports.BrokerDaemonObservation {
220 observation := pickerTestUnobservedObservation(now)
221 observation.Availability = domain.RemoteAvailabilityUnreachable
222 return observation
223 }(),
224 }},
225 wantLocal: true, wantAdmission: ports.BrokerAdmissionCreateEphemeral,
226 },
227 {
228 name: "local creation on a never-observed daemon stays attemptable",
229 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}},
230 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
231 pickerTestUnobservedObservation(now),
232 }},
233 wantLocal: true, wantAdmission: ports.BrokerAdmissionCreateEphemeral,
234 },
235 {
236 name: "epoch change refuses the resolved intent",
237 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true}, Target: target},
238 snapshot: ports.BrokerSnapshot{Epoch: 4, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
239 pickerTestLocalObservation(now, pickerTestSessionWithLifecycle("alpha", target.LifecycleID)),
240 }},
241 wantErr: pickerCatalogueEpochStale,
242 },
243 {
244 name: "re-added endpoint refuses the resolved intent",
245 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Epoch: 3, Destination: remote},
246 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
247 pickerTestRemoteObservation("user@arch", 2, 2, now),
248 }},
249 wantErr: pickerCatalogueReplaced,
250 },
251 {
252 name: "advanced registration generation refuses the resolved intent",
253 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Epoch: 3, Destination: remote},
254 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
255 func() ports.BrokerDaemonObservation {
256 observation := pickerTestRemoteObservation("user@arch", 1, 1, now)
257 observation.Registration.Generation = 2
258 return observation
259 }(),
260 }},
261 wantErr: pickerCatalogueReplaced,
262 },
263 {
264 name: "same-name new lifecycle refuses the exact attach",
265 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true}, Target: target},
266 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
267 pickerTestLocalObservation(now, pickerTestSession("alpha", 5, catalogue_Up)),
268 }},
269 wantErr: pickerCatalogueReplaced,
270 },
271 {
272 name: "gone session refuses the exact attach",
273 navigation: InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true}, Target: target},
274 snapshot: localDaemonSnapshot(3, 1),
275 wantErr: pickerCatalogueGone,
276 },
277 {
278 name: "known incompatibility refuses immediately",
279 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}},
280 snapshot: ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{
281 func() ports.BrokerDaemonObservation {
282 observation := pickerTestLocalObservation(now)
283 observation.ProtocolVersion = protocol.Version + 1
284 return observation
285 }(),
286 }},
287 wantErr: pickerCatalogueIncompatible,
288 },
289 {
290 name: "unavailable snapshot refuses",
291 navigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}},
292 snapshot: ports.BrokerSnapshot{},
293 wantErr: pickerCatalogueUnavailable,
294 },
295 }
296 for _, tt := range tests {
297 t.Run(tt.name, func(t *testing.T) {
298 request, err := tt.navigation.Resolve(tt.snapshot.Clone())
299 if tt.wantErr != 0 {
300 require.Error(t, err)
301 require.True(t, pickerCatalogueErrorIs(err, tt.wantErr), "want %v, got %v", tt.wantErr, err)
302 return
303 }
304 require.NoError(t, err)
305 require.Equal(t, ports.BrokerStreamAttachment, request.Purpose)
306 require.Equal(t, tt.wantLocal, request.Local)
307 require.Equal(t, tt.wantAdmission, request.Admission)
308 require.Equal(t, tt.wantName, request.Name)
309 require.Equal(t, tt.wantTarget, request.Target)
310 require.Equal(t, ports.BrokerEpoch(tt.snapshot.Epoch), request.Epoch)
311 require.Equal(t, ports.BrokerDaemonStartIfNeeded, request.StartMode)
312 // The gabarit never owns the live connection identity.
313 require.True(t, request.Connection.IsZero())
314 require.Zero(t, request.Stream)
315 })
316 }
317}
318
319// TestInitialNavigationPickerResolveIsRefused pins that resolving the picker
320// directly is refused rather than returning a nil-but-usable request.
321func TestInitialNavigationPickerResolveIsRefused(t *testing.T) {
322 request, err := InitialNavigation{}.Resolve(localDaemonSnapshot(3, 1))
323 require.Error(t, err)
324 require.True(t, pickerCatalogueErrorIs(err, pickerCatalogueNoSelection))
325 require.Equal(t, ports.BrokerOpenStreamRequest{}, request)
326}
327
328// TestInitialNavigationResolverIsConsultedAtMostOnce pins the one-shot
329// consumption: the resolver runs exactly once for the whole run, on a
330// defensive copy of the first committed publication, and a reconnect never
331// replays the navigation.
332func TestInitialNavigationResolverIsConsultedAtMostOnce(t *testing.T) {
333 ctx, cancel := context.WithCancel(context.Background())
334 defer cancel()
335 clock := newSupervisorTestClock()
336 reader := newAttachTestReader()
337 terminal := &attachTestTerminal{in: reader}
338 picker := newAttachTestPicker()
339
340 first := newSupervisorTestService(ports.BrokerConnectionID{1})
341 first.publishSnapshot(localDaemonSnapshot(7, 1))
342 second := newSupervisorTestService(ports.BrokerConnectionID{2})
343 stream := newSessionTestStream()
344 first.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
345 return stream, nil
346 })
347 connector := newSupervisorTestConnector(func(_ context.Context, call int) (ports.BrokerNavigator, error) {
348 if call == 1 {
349 return first, nil
350 }
351 return second, nil
352 })
353
354 var calls atomic.Int64
355 var resolvedEpoch atomic.Uint64
356 var mutatedOnce atomic.Bool
357 cfg := SupervisorConfig{
358 Connector: connector,
359 Terminal: terminal,
360 Clock: clock,
361 Jitter: func() float64 { return 0 },
362 Picker: picker,
363 ResolveInitialNavigation: func(snapshot ports.BrokerSnapshot) (InitialNavigation, error) {
364 calls.Add(1)
365 resolvedEpoch.Store(uint64(snapshot.Epoch))
366 if !mutatedOnce.Swap(true) {
367 // A defensive copy: mutating the argument must not affect the
368 // supervisor's committed authority.
369 snapshot.Epoch = 0
370 snapshot.Daemons = nil
371 }
372 return InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}, nil
373 },
374 }
375 sup := mustSupervisor(t, cfg)
376 done := make(chan error, 1)
377 go func() { done <- sup.Run(ctx) }()
378
379 require.Eventually(t, func() bool { return len(first.openedRequests()) == 1 }, 5*time.Second, time.Millisecond)
380 require.Equal(t, int64(1), calls.Load(), "the resolver runs exactly once")
381 // The resolver may not mutate the supervisor's authority: the epoch it saw
382 // stays the committed one.
383 require.Equal(t, ports.BrokerEpoch(7), ports.BrokerEpoch(resolvedEpoch.Load()))
384 require.Equal(t, ports.BrokerEpoch(7), first.Snapshot().Epoch)
385 require.Len(t, first.Snapshot().Daemons, 1)
386
387 stream.deliver(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "creation failed"})
388 require.Eventually(t, func() bool { return sup.State().Presentation == PresentPicker }, 5*time.Second, time.Millisecond)
389
390 first.lose(ports.BrokerError{Code: ports.BrokerErrorUnavailable})
391 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityRetryWait }, 5*time.Second, time.Millisecond)
392 for connector.callCount() < 2 {
393 clock.mu.Lock()
394 var timer *supervisorTestTimer
395 if len(clock.timers) != 0 {
396 timer = clock.timers[len(clock.timers)-1]
397 }
398 clock.mu.Unlock()
399 if timer != nil {
400 timer.fire()
401 }
402 }
403 second.publishSnapshot(localDaemonSnapshot(8, 1))
404 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
405 require.Empty(t, second.openedRequests(), "a reconnect never replays the one-shot navigation")
406 require.Equal(t, int64(1), calls.Load(), "the resolver is never re-run on the replacement generation")
407 cancel()
408 require.ErrorIs(t, <-done, context.Canceled)
409}
410
411// TestInitialNavigationPickerIntentOpensNothing pins that a composition that
412// omits the intent, or sets the picker explicitly, opens no stream and leaves
413// the picker running.
414func TestInitialNavigationPickerIntentOpensNothing(t *testing.T) {
415 for _, tc := range []struct {
416 name string
417 configure func(*SupervisorConfig)
418 wantPicker bool
419 }{
420 {name: "omitted intent", configure: func(*SupervisorConfig) {}},
421 {name: "explicit picker", configure: func(cfg *SupervisorConfig) { cfg.InitialNavigation = InitialNavigation{} }},
422 } {
423 t.Run(tc.name, func(t *testing.T) {
424 picker := newAttachTestPicker()
425 harness := startAttachHarnessConfig(t, picker, tc.configure)
426 require.Equal(t, PresentPicker, harness.sup.State().Presentation)
427 require.Empty(t, harness.service.openedRequests(), "a picker intent opens no stream")
428 require.Zero(t, picker.resolveCount())
429 })
430 }
431}
432
433// TestNewSupervisorRefusesContradictoryInitialConfiguration pins that a
434// configuration carrying both an intent and a resolver, or an invalid intent,
435// is refused at construction instead of at the first connection.
436func TestNewSupervisorRefusesContradictoryInitialConfiguration(t *testing.T) {
437 clock := newSupervisorTestClock()
438 reader := newSupervisorTestReader()
439 t.Cleanup(reader.unblock)
440 terminal := newSupervisorTestTerminal(reader)
441 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
442 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable}
443 })
444
445 base := func() SupervisorConfig {
446 return SupervisorConfig{Connector: connector, Terminal: terminal, Clock: clock, Jitter: func() float64 { return 0 }, SessionEnvironment: SessionEnvironment{Provenance: SessionEnvironmentLocalCLI}}
447 }
448
449 both := base()
450 both.InitialNavigation = InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
451 both.ResolveInitialNavigation = func(ports.BrokerSnapshot) (InitialNavigation, error) { return InitialNavigation{}, nil }
452 _, err := NewSupervisor(both)
453 require.Error(t, err, "an intent and a resolver are mutually exclusive")
454
455 invalid := base()
456 invalid.InitialNavigation = InitialNavigation{Kind: InitialNavigationAttachExact, Destination: ports.BrokerEndpointFence{Local: true}}
457 _, err = NewSupervisor(invalid)
458 require.Error(t, err, "an exact attach without an epoch is refused at construction")
459
460 valid := base()
461 valid.InitialNavigation = InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
462 _, err = NewSupervisor(valid)
463 require.NoError(t, err)
464}
465
466// TestInitialNavigationResolverFailureIsSelectionUnavailable pins that a
467// resolver that refuses localises to the selection-unavailable notice and never
468// dials a destination or replays.
469func TestInitialNavigationResolverFailureIsSelectionUnavailable(t *testing.T) {
470 ctx, cancel := context.WithCancel(context.Background())
471 defer cancel()
472 clock := newSupervisorTestClock()
473 reader := newAttachTestReader()
474 terminal := &attachTestTerminal{in: reader}
475 picker := newAttachTestPicker()
476 service := newSupervisorTestService(ports.BrokerConnectionID{1})
477 service.publishSnapshot(localDaemonSnapshot(3, 1))
478 opened := false
479 service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
480 opened = true
481 return newSessionTestStream(), nil
482 })
483 notices := make(chan LifecycleNotice, 4)
484 var calls atomic.Int64
485 resolverErr := errors.New("no such target")
486 sup := mustSupervisor(t, SupervisorConfig{
487 Connector: newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) { return service, nil }),
488 Terminal: terminal, Clock: clock, Jitter: func() float64 { return 0 }, Picker: picker,
489 NotifyLifecycle: func(notice LifecycleNotice) { notices <- notice },
490 ResolveInitialNavigation: func(ports.BrokerSnapshot) (InitialNavigation, error) {
491 calls.Add(1)
492 return InitialNavigation{}, resolverErr
493 },
494 })
495 done := make(chan error, 1)
496 go func() { done <- sup.Run(ctx) }()
497 require.Equal(t, LifecycleNoticeSelectionUnavailable, (<-notices).Kind)
498 require.False(t, opened, "a refused resolver never dials a destination")
499 require.Equal(t, int64(1), calls.Load())
500 cancel()
501 require.ErrorIs(t, <-done, context.Canceled)
502}
503
504// TestInitialNavigationWaitsForTargetObservation pins that a resolver whose
505// target daemon is not observed yet keeps the intent armed: it resolves again
506// on the next committed publication, and only the bounded observation budget
507// turns the wait into a selection-unavailable refusal.
508func TestInitialNavigationWaitsForTargetObservation(t *testing.T) {
509 late := protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(5), SessionName: "late"}
510 tests := []struct {
511 name string
512 // observedAt is the first revision the resolver can decide on; zero
513 // means the target is never observed.
514 observedAt ports.BrokerRevision
515 // fallback is offered with the pending resolution.
516 fallback bool
517 wantOpen bool
518 }{
519 {name: "a later publication resolves the armed intent", observedAt: 2, wantOpen: true},
520 {name: "the observation budget refuses once", observedAt: 0},
521 {name: "the observation budget takes the resolver's fallback", observedAt: 0, fallback: true, wantOpen: true},
522 }
523 for _, tt := range tests {
524 t.Run(tt.name, func(t *testing.T) {
525 ctx, cancel := context.WithCancel(context.Background())
526 defer cancel()
527 clock := newSupervisorTestClock()
528 reader := newAttachTestReader()
529 terminal := &attachTestTerminal{in: reader}
530 picker := newAttachTestPicker()
531 service := newSupervisorTestService(ports.BrokerConnectionID{1})
532 service.publishSnapshot(localDaemonSnapshot(3, 1))
533 var opened atomic.Int64
534 service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
535 opened.Add(1)
536 return newSessionTestStream(), nil
537 })
538 notices := make(chan LifecycleNotice, 4)
539 var calls atomic.Int64
540 sup := mustSupervisor(t, SupervisorConfig{
541 Connector: newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) { return service, nil }),
542 Terminal: terminal, Clock: clock, Jitter: func() float64 { return 0 }, Picker: picker,
543 NotifyLifecycle: func(notice LifecycleNotice) { notices <- notice },
544 ResolveInitialNavigation: func(snapshot ports.BrokerSnapshot) (InitialNavigation, error) {
545 calls.Add(1)
546 if tt.observedAt == 0 || snapshot.Revision < tt.observedAt {
547 pending := InitialNavigationNotObserved{Endpoint: "user@host"}
548 if tt.fallback {
549 pending.Fallback = InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
550 }
551 return InitialNavigation{}, pending
552 }
553 // The target exists only in the later publication: the stream
554 // request must resolve against the snapshot that decided.
555 return InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: snapshot.Epoch, Destination: ports.BrokerEndpointFence{Local: true}, Target: late}, nil
556 },
557 })
558 done := make(chan error, 1)
559 go func() { done <- sup.Run(ctx) }()
560
561 require.Eventually(t, func() bool { return calls.Load() == 1 }, 5*time.Second, time.Millisecond)
562 var budget *supervisorTestTimer
563 require.Eventually(t, func() bool {
564 clock.mu.Lock()
565 defer clock.mu.Unlock()
566 for _, timer := range clock.timers {
567 if timer.delay == initialNavigationObservationBudget {
568 budget = timer
569 return true
570 }
571 }
572 return false
573 }, 5*time.Second, time.Millisecond, "the wait is bounded by the observation budget")
574 require.Zero(t, opened.Load(), "an unobserved target never dials")
575 service.mu.Lock()
576 reconciles := append([]string(nil), service.reconciles...)
577 service.mu.Unlock()
578 require.Equal(t, []string{"user@host"}, reconciles, "the wait asks the broker for one fresh observation")
579
580 if tt.fallback {
581 budget.fire()
582 require.Eventually(t, func() bool { return opened.Load() == 1 }, 5*time.Second, time.Millisecond)
583 require.Equal(t, protocol.ExactSessionTarget{}, service.openedRequests()[0].Target, "the fallback creation opens, not an exact attach")
584 require.Equal(t, int64(1), calls.Load())
585 } else if tt.wantOpen {
586 service.publishSnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
587 pickerTestLocalObservation(time.Unix(1000, 0), pickerTestSession("late", 5, catalogue_Up)),
588 }})
589 require.Eventually(t, func() bool { return opened.Load() == 1 }, 5*time.Second, time.Millisecond)
590 require.Equal(t, late, service.openedRequests()[0].Target)
591 require.Equal(t, int64(2), calls.Load(), "the armed intent resolves once per publication")
592 } else {
593 budget.fire()
594 select {
595 case notice := <-notices:
596 require.Equal(t, LifecycleNoticeSelectionUnavailable, notice.Kind)
597 case <-time.After(5 * time.Second):
598 t.Fatal("the budget never refused the unobserved target")
599 }
600 require.Zero(t, opened.Load())
601 service.publishSnapshot(localDaemonSnapshot(3, 2))
602 require.Never(t, func() bool { return calls.Load() > 1 }, 100*time.Millisecond, 5*time.Millisecond, "a refused intent is never re-armed")
603 }
604 cancel()
605 require.ErrorIs(t, <-done, context.Canceled)
606 })
607 }
608}
609
610// TestInitialNavigationDropsDecisionsTakenWhileHidden pins that a selection
611// recorded by the picker while it was hidden behind the Connecting
612// presentation of a pending initial navigation never replays: after the
613// initial attachment ends, the supervisor returns to the picker and opens
614// nothing the user did not see.
615func TestInitialNavigationDropsDecisionsTakenWhileHidden(t *testing.T) {
616 ctx, cancel := context.WithCancel(context.Background())
617 defer cancel()
618 clock := newSupervisorTestClock()
619 reader := newAttachTestReader()
620 terminal := &attachTestTerminal{in: reader}
621 picker := newAttachTestPicker()
622
623 service := newSupervisorTestService(ports.BrokerConnectionID{1})
624 stream := newSessionTestStream()
625 service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
626 return stream, nil
627 })
628 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) { return service, nil })
629 // A keystroke typed during the cold start: the hidden picker recorded a
630 // commit before the broker published anything.
631 picker.recordOp(pickerOp{commit: true}, "row", ports.BrokerOpenStreamRequest{})
632
633 sup := mustSupervisor(t, SupervisorConfig{
634 Connector: connector, Terminal: terminal, Clock: clock,
635 Jitter: func() float64 { return 0 }, Picker: picker,
636 InitialNavigation: InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}},
637 })
638 done := make(chan error, 1)
639 go func() { done <- sup.Run(ctx) }()
640
641 require.Eventually(t, func() bool { return sup.State().Presentation == PresentConnecting }, 5*time.Second, time.Millisecond,
642 "a pending navigation presents Connecting, not the picker")
643 service.publishSnapshot(localDaemonSnapshot(7, 1))
644 require.Eventually(t, func() bool { return len(service.openedRequests()) == 1 }, 5*time.Second, time.Millisecond)
645
646 stream.deliver(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "creation failed"})
647 require.Eventually(t, func() bool { return sup.State().Presentation == PresentPicker }, 5*time.Second, time.Millisecond)
648 require.Never(t, func() bool { return len(service.openedRequests()) > 1 }, 200*time.Millisecond, 5*time.Millisecond,
649 "a decision taken while hidden must not open a second stream")
650 require.Zero(t, picker.resolveCount(), "the hidden commit was never resolved")
651 cancel()
652 require.ErrorIs(t, <-done, context.Canceled)
653}