1package client
2
3import (
4 "context"
5 "sync"
6 "testing"
7 "time"
8
9 "github.com/stretchr/testify/require"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/internal/protocol"
14)
15
16func TestP54LifecycleActionsAreDistinct(t *testing.T) {
17 for _, tc := range []struct {
18 name string
19 action AttachmentLifecycleActionKind
20 wantNotice LifecycleNoticeKind
21 wantExit bool
22 }{
23 {name: "detach to picker", action: AttachmentDetachToPicker, wantNotice: LifecycleNoticeDetachToPicker},
24 {name: "detach and exit", action: AttachmentDetachAndExit, wantNotice: LifecycleNoticeDetachAndExit, wantExit: true},
25 } {
26 t.Run(tc.name, func(t *testing.T) {
27 actions := make(chan AttachmentLifecycleAction, 1)
28 notices := make(chan LifecycleNotice, 4)
29 picker := newAttachTestPicker()
30 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
31 cfg.LifecycleActions = actions
32 cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice }
33 })
34 stream := newSessionTestStream()
35 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
36 return stream, nil
37 })
38 picker.commit(sessionTestRequest(true))
39 awaitStreamHello(t, &sync.Mutex{}, &stream)
40 deliverReadyStream(t, stream)
41 awaitAttachedState(t, harness.sup)
42 actions <- AttachmentLifecycleAction{Token: authorityToken(&harness.sup.attachments.authority), Kind: tc.action}
43 require.Equal(t, tc.wantNotice, (<-notices).Kind)
44 if tc.wantExit {
45 require.NoError(t, harness.waitRun(t))
46 } else {
47 awaitPickerState(t, harness.sup)
48 require.True(t, picker.owns())
49 require.Len(t, harness.service.openedRequests(), 1, "detach must not auto-reattach")
50 }
51 })
52 }
53}
54
55func TestP54AttachmentOutcomesReturnToPicker(t *testing.T) {
56 for _, tc := range []struct {
57 name string
58 settle func(*sessionTestStream)
59 notice LifecycleNoticeKind
60 }{
61 // ReasonDetach (the explicit user detach) is intentionally excluded here:
62 // TestSupervisorDetachedReasonDeterminesExitOrPicker pins that it exits
63 // the client instead of returning to the picker.
64 {name: "server shutdown", settle: func(s *sessionTestStream) { s.deliver(protocol.Detached{Reason: protocol.ReasonServerShutdown}) }, notice: LifecycleNoticeSessionEnded},
65 {name: "session deletion", settle: func(s *sessionTestStream) { s.deliver(protocol.Detached{Reason: protocol.ReasonSessionKilled}) }, notice: LifecycleNoticeSessionEnded},
66 // A stream loss after attachment resumes the same session instead:
67 // TestSupervisorAttachmentLossResumesSameSession pins that path.
68 } {
69 t.Run(tc.name, func(t *testing.T) {
70 picker := newAttachTestPicker()
71 notices := make(chan LifecycleNotice, 4)
72 harness := startAttachHarness(t, picker)
73 harness.sup.cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice }
74 stream := newSessionTestStream()
75 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
76 return stream, nil
77 })
78 picker.commit(sessionTestRequest(true))
79 require.Eventually(t, func() bool { return len(stream.messages()) > 0 }, 5*time.Second, time.Millisecond)
80 deliverReadyStream(t, stream)
81 awaitAttachedState(t, harness.sup)
82 tc.settle(stream)
83 awaitPickerState(t, harness.sup)
84 require.Equal(t, tc.notice, (<-notices).Kind)
85 require.Len(t, harness.service.openedRequests(), 1)
86 })
87 }
88}
89
90func TestP54LocalRefusalIsNotReportedAsDestinationFailure(t *testing.T) {
91 picker := newAttachTestPicker()
92 notices := make(chan LifecycleNotice, 4)
93 harness := startAttachHarness(t, picker)
94 harness.sup.cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice }
95
96 // The committed selection is refused locally, before any destination exists:
97 // reporting that as a destination failure would name the wrong cause.
98 picker.mu.Lock()
99 picker.resolveErr = pickerCatalogueError{Code: pickerCatalogueGone, Text: "the committed session is gone"}
100 picker.mu.Unlock()
101 picker.commit(sessionTestRequest(true))
102
103 awaitPickerState(t, harness.sup)
104 require.Equal(t, LifecycleNoticeSelectionUnavailable, (<-notices).Kind)
105 require.Empty(t, harness.service.openedRequests(), "a refused selection never dials a destination")
106}
107
108// TestP54ResolveInitialUsesRealInitialNavigationResolve proves the closed union
109// resolves the no-argument ephemeral creation through a real catalogue: the
110// resolution uses the same rules as the interactive picker and applies no
111// privileged initial semantics.
112func TestP54ResolveInitialUsesRealInitialNavigationResolve(t *testing.T) {
113 clock := newSupervisorTestClock()
114 catalogue, _ := pickerTestCatalogue(t)
115 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{pickerTestLocalObservation(clock.Now())}}))
116 navigation := InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
117 request, err := navigation.Resolve(catalogue.Snapshot())
118 require.NoError(t, err)
119 require.Equal(t, ports.BrokerAdmissionCreateEphemeral, request.Admission)
120 require.True(t, request.Local)
121 require.Empty(t, request.Name)
122 require.Equal(t, protocol.ExactSessionTarget{}, request.Target)
123}
124
125func TestP54InitialNavigationCreatesEphemeralThroughBroker(t *testing.T) {
126 picker := newAttachTestPicker()
127 navigation := InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
128 ctx, cancel := context.WithCancel(context.Background())
129 defer cancel()
130 clock := newSupervisorTestClock()
131 reader := newAttachTestReader()
132 terminal := &attachTestTerminal{in: reader}
133 service := newSupervisorTestService(ports.BrokerConnectionID{1})
134 service.publishSnapshot(localDaemonSnapshot(3, 1))
135 stream := newSessionTestStream()
136 service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
137 return stream, nil
138 })
139 sup := mustSupervisor(t, SupervisorConfig{Connector: newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) { return service, nil }), Terminal: terminal, Clock: clock, Jitter: func() float64 { return 0 }, Picker: picker, InitialNavigation: navigation})
140 done := make(chan error, 1)
141 go func() { done <- sup.Run(ctx) }()
142 require.Eventually(t, func() bool { return len(service.openedRequests()) == 1 }, 5*time.Second, time.Millisecond)
143 request := service.openedRequests()[0]
144 require.Equal(t, ports.BrokerAdmissionCreateEphemeral, request.Admission)
145 require.Equal(t, sessionTestRequest(true).Local, request.Local)
146 require.Zero(t, picker.resolveCount(), "initial navigation resolves through the union, never through the picker")
147 stream.deliver(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "creation failed"})
148 require.Eventually(t, func() bool { return sup.State().Presentation == PresentPicker }, 5*time.Second, time.Millisecond)
149 cancel()
150 require.ErrorIs(t, <-done, context.Canceled)
151}
152
153func TestP54InitialNavigationIsNotReplayedAfterReconnect(t *testing.T) {
154 ctx, cancel := context.WithCancel(context.Background())
155 defer cancel()
156 clock := newSupervisorTestClock()
157 reader := newAttachTestReader()
158 terminal := &attachTestTerminal{in: reader}
159 picker := newAttachTestPicker()
160 navigation := InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
161 first := newSupervisorTestService(ports.BrokerConnectionID{1})
162 first.publishSnapshot(localDaemonSnapshot(1, 1))
163 second := newSupervisorTestService(ports.BrokerConnectionID{2})
164 stream := newSessionTestStream()
165 first.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
166 return stream, nil
167 })
168 connector := newSupervisorTestConnector(func(_ context.Context, call int) (ports.BrokerNavigator, error) {
169 if call == 1 {
170 return first, nil
171 }
172 return second, nil
173 })
174 sup := mustSupervisor(t, SupervisorConfig{Connector: connector, Terminal: terminal, Clock: clock, Jitter: func() float64 { return 0 }, Picker: picker, InitialNavigation: navigation})
175 done := make(chan error, 1)
176 go func() { done <- sup.Run(ctx) }()
177 require.Equal(t, 1, connector.awaitStart(t))
178 require.Eventually(t, func() bool { return len(first.openedRequests()) == 1 }, 5*time.Second, time.Millisecond)
179 stream.deliver(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "creation failed"})
180 require.Eventually(t, func() bool { return sup.State().Presentation == PresentPicker }, 5*time.Second, time.Millisecond)
181 first.lose(ports.BrokerError{Code: ports.BrokerErrorUnavailable})
182 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityRetryWait }, 5*time.Second, time.Millisecond)
183 for connector.callCount() < 2 {
184 clock.mu.Lock()
185 var timer *supervisorTestTimer
186 if len(clock.timers) != 0 {
187 timer = clock.timers[len(clock.timers)-1]
188 }
189 clock.mu.Unlock()
190 if timer != nil {
191 timer.fire()
192 }
193 }
194 second.publishSnapshot(localDaemonSnapshot(2, 1))
195 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
196 require.Empty(t, second.openedRequests(), "one-shot navigation must not replay on the replacement generation")
197 cancel()
198 require.ErrorIs(t, <-done, context.Canceled)
199}
200
201func TestP54SingleReaderDeliversAttachedByteAfterRealSupervisorHandoff(t *testing.T) {
202 picker := newAttachTestPicker()
203 harness := startAttachHarness(t, picker)
204 stream := newSessionTestStream()
205 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
206 return stream, nil
207 })
208 picker.commit(sessionTestRequest(true))
209 require.Eventually(t, func() bool { return len(stream.messages()) > 0 }, 5*time.Second, time.Millisecond)
210 deliverReadyStream(t, stream)
211 awaitAttachedState(t, harness.sup)
212 harness.reader.push([]byte("x"))
213 require.Eventually(t, func() bool {
214 for _, message := range stream.messages() {
215 if input, ok := message.(protocol.Input); ok && string(input.Data) == "x" {
216 return true
217 }
218 }
219 return false
220 }, 5*time.Second, time.Millisecond, "one typed byte must survive the picker-to-attachment claim handoff")
221}
222
223func TestP54BrokerRestartDoesNotReattachOrReplayUnknownMutation(t *testing.T) {
224 ctx, cancel := context.WithCancel(context.Background())
225 defer cancel()
226 clock := newSupervisorTestClock()
227 reader := newSupervisorTestReader()
228 t.Cleanup(reader.unblock)
229 first := newSupervisorTestService(ports.BrokerConnectionID{1})
230 first.publish(1, 1)
231 second := newSupervisorTestService(ports.BrokerConnectionID{2})
232 connector := newSupervisorTestConnector(func(_ context.Context, call int) (ports.BrokerNavigator, error) {
233 if call == 1 {
234 return first, nil
235 }
236 return second, nil
237 })
238 notices := make(chan LifecycleNotice, 4)
239 sup := mustSupervisor(t, SupervisorConfig{Connector: connector, Terminal: newSupervisorTestTerminal(reader), Clock: clock, Jitter: func() float64 { return 0 }, NotifyLifecycle: func(n LifecycleNotice) { notices <- n }})
240 done := make(chan error, 1)
241 go func() { done <- sup.Run(ctx) }()
242 require.Equal(t, 1, connector.awaitStart(t))
243 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
244 first.lose(ports.BrokerError{Code: ports.BrokerErrorOutcomeUnknown, Text: "reply lost after mutation"})
245 require.Equal(t, LifecycleNoticeBrokerLost, (<-notices).Kind)
246 retry := clock.awaitTimer(t)
247 retry.fire()
248 require.Equal(t, 2, connector.awaitStart(t))
249 second.publish(2, 1)
250 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
251 require.Equal(t, LifecycleNoticeBrokerReconnected, (<-notices).Kind)
252 require.Empty(t, first.openedRequests())
253 require.Empty(t, second.openedRequests(), "restart must not auto-reattach or replay a mutation")
254 cancel()
255 require.ErrorIs(t, <-done, context.Canceled)
256}
257
258func TestP54ReadyMapPrunedAcrossClaims(t *testing.T) {
259 pump := newTerminalInputPump(nil)
260 for i := 0; i < 100; i++ {
261 consumer, ok := pump.tryClaim()
262 require.True(t, ok)
263 pump.revoke(consumer)
264 }
265 pump.readyMu.Lock()
266 defer pump.readyMu.Unlock()
267 require.Empty(t, pump.ready, "revoked monotonic claim IDs must not retain readiness channels")
268}
269
270// TestP54PickerCommittedReadIsNotReplayedToAttachment proves a picker read the
271// user already acted on is committed at the handoff and never re-delivered as
272// session input.
273func TestP54PickerCommittedReadIsNotReplayedToAttachment(t *testing.T) {
274 reader := newAttachTestReader()
275 consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
276 lifetime := startTerminalInputLifetime(reader, consumer)
277 defer lifetime.stop()
278 reader.push([]byte("1\r"))
279 var committed []byte
280 require.Eventually(t, func() bool {
281 select {
282 case committed = <-consumer.reads:
283 return true
284 default:
285 return false
286 }
287 }, 5*time.Second, time.Millisecond)
288 require.Equal(t, []byte("1\r"), committed)
289 lifetime.releasePicker()
290 attachment, claimed := lifetime.pump.tryClaim()
291 require.True(t, claimed)
292 _, ok := lifetime.pump.take(context.Background(), attachment)
293 require.False(t, ok, "committed picker read must not cross the handoff")
294 lifetime.pump.revoke(attachment)
295}
296
297func TestP54AttachmentFinalizeDropsUnleasedPendingBeforePicker(t *testing.T) {
298 reader := newAttachTestReader()
299 consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
300 lifetime := startTerminalInputLifetime(reader, consumer)
301 defer lifetime.stop()
302 lifetime.releasePicker()
303 host := newAttachmentHost(attachmentHostConfig{Input: lifetime.pump})
304 host.ownerBoundary = true
305 fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
306 require.True(t, host.authority.grant(fg))
307 reader.push([]byte("ls\r"))
308 require.Eventually(t, func() bool {
309 lifetime.pump.mu.Lock()
310 defer lifetime.pump.mu.Unlock()
311 return lifetime.pump.pending != nil && lifetime.pump.delivering == 0
312 }, 5*time.Second, time.Millisecond)
313 fg.finish()
314 lifetime.acquirePicker()
315 assertNoPickerRead(t, consumer.reads)
316}
317
318func TestP54AttachmentFinalizeDropsPreservedResidualBeforePicker(t *testing.T) {
319 reader := newAttachTestReader()
320 consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
321 lifetime := startTerminalInputLifetime(reader, consumer)
322 defer lifetime.stop()
323 lifetime.releasePicker()
324 host := newAttachmentHost(attachmentHostConfig{Input: lifetime.pump})
325 host.ownerBoundary = true
326 fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
327 require.True(t, host.authority.grant(fg))
328 reader.push([]byte("\r"))
329 event, ok := fg.Input(context.Background())
330 require.True(t, ok)
331 fg.PreserveInput(event.Data)
332 fg.finish()
333 lifetime.acquirePicker()
334 assertNoPickerRead(t, consumer.reads)
335}
336
337func assertNoPickerRead(t *testing.T, reads <-chan []byte) {
338 t.Helper()
339 select {
340 case got := <-reads:
341 t.Fatalf("attachment bytes crossed into picker: %q", got)
342 case <-time.After(100 * time.Millisecond):
343 }
344}
345
346func TestP54UndecidedAttachmentReadIsDroppedBeforePicker(t *testing.T) {
347 reader := newAttachTestReader()
348 consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
349 lifetime := startTerminalInputLifetime(reader, consumer)
350 defer lifetime.stop()
351 lifetime.releasePicker()
352 attachment, claimed := lifetime.pump.tryClaim()
353 require.True(t, claimed)
354 reader.push([]byte("ls\r"))
355 var delivery terminalReadResult
356 require.Eventually(t, func() bool {
357 var ok bool
358 delivery, ok = lifetime.pump.take(context.Background(), attachment)
359 return ok
360 }, 5*time.Second, time.Millisecond)
361 require.Equal(t, []byte("ls\r"), delivery.data)
362 lifetime.pump.dropOwned(attachment)
363 lifetime.pump.revoke(attachment)
364 lifetime.acquirePicker()
365 select {
366 case got := <-consumer.reads:
367 t.Fatalf("attachment bytes crossed into picker: %q", got)
368 case <-time.After(100 * time.Millisecond):
369 }
370}
371
372type recordingPickerConsumer struct{ reads chan []byte }
373
374func (c *recordingPickerConsumer) ConsumeTerminalRead(data []byte) bool {
375 c.reads <- append([]byte(nil), data...)
376 return true
377}
378
379func TestP54InputDecisionProtocol(t *testing.T) {
380 pump := newTerminalInputPump(nil)
381 <-pump.space
382 host := newAttachmentHost(attachmentHostConfig{Input: pump})
383 fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
384 require.True(t, host.authority.grant(fg))
385 pump.mu.Lock()
386 pump.pending = &terminalReadResult{data: []byte("once")}
387 pump.mu.Unlock()
388 pump.signalReady(fg.consumer)
389 event, ok := fg.Input(context.Background())
390 require.True(t, ok)
391 require.Equal(t, []byte("once"), event.Data)
392 fg.PreserveInput(event.Data)
393 fg.AckInput() // refused: already decided
394 host.authority.revoke(fg)
395 next := host.newForeground(AttachmentToken{Generation: 2}, newSessionTestStream())
396 require.True(t, host.authority.grant(next))
397 replayed, ok := next.Input(context.Background())
398 require.True(t, ok)
399 require.Equal(t, []byte("once"), replayed.Data)
400 next.AckInput()
401 next.PreserveInput([]byte("duplicate")) // refused: no outstanding delivery
402 require.Empty(t, pump.residual)
403}
404
405func TestP54GeometryIsPerAttachmentAndCarriesAcrossPicker(t *testing.T) {
406 resizes := make(chan domain.Geometry, 4)
407 host := newAttachmentHost(attachmentHostConfig{})
408 host.resizes = resizes
409 host.ownerBoundary = true
410 ctx, cancel := context.WithCancel(context.Background())
411 defer cancel()
412 host.startGeometry(ctx)
413 defer host.stopGeometry()
414 first := domain.Geometry{Size: domain.Size{Cols: 90, Rows: 30}}
415 latest := domain.Geometry{Size: domain.Size{Cols: 120, Rows: 40}}
416 resizes <- first
417 resizes <- latest
418 require.Eventually(t, func() bool { geometry, ok := host.latestGeometry(); return ok && geometry.Size == latest.Size }, 5*time.Second, time.Millisecond)
419 fg1 := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
420 require.True(t, host.authority.grant(fg1))
421 geometry, ok := fg1.Resize(context.Background())
422 require.True(t, ok)
423 require.Equal(t, latest.Size, geometry.Size)
424 host.authority.revoke(fg1)
425 fg2 := host.newForeground(AttachmentToken{Generation: 2}, newSessionTestStream())
426 require.True(t, host.authority.grant(fg2))
427 geometry, ok = fg2.Resize(context.Background())
428 require.True(t, ok)
429 require.Equal(t, latest.Size, geometry.Size)
430 _, stale := fg1.Resize(context.Background())
431 require.False(t, stale)
432}
433
434func TestP54LifecycleSkipsStaleTokenAndDeliversLiveDetach(t *testing.T) {
435 actions := make(chan AttachmentLifecycleAction, 2)
436 host := newAttachmentHost(attachmentHostConfig{Actions: actions})
437 fg := host.newForeground(AttachmentToken{Generation: 2, Attempt: 3}, newSessionTestStream())
438 require.True(t, host.authority.grant(fg))
439 actions <- AttachmentLifecycleAction{Token: AttachmentToken{Generation: 1, Attempt: 9}, Kind: AttachmentDetachAndExit}
440 actions <- AttachmentLifecycleAction{Token: fg.token, Kind: AttachmentDetachToPicker}
441 action, ok := fg.Lifecycle(context.Background())
442 require.True(t, ok)
443 require.Equal(t, AttachmentDetachToPicker, action)
444 fg.finish()
445}
446
447func TestP54LifecycleInterleavingsRace(t *testing.T) {
448 for i := 0; i < 50; i++ {
449 pump := newTerminalInputPump(nil)
450 resizes := make(chan domain.Geometry, 1)
451 actions := make(chan AttachmentLifecycleAction, 1)
452 host := newAttachmentHost(attachmentHostConfig{Input: pump, Actions: actions})
453 host.resizes = resizes
454 ctx, cancel := context.WithCancel(context.Background())
455 host.startGeometry(ctx)
456 fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
457 require.True(t, host.authority.grant(fg))
458 var wg sync.WaitGroup
459 wg.Add(3)
460 go func() { defer wg.Done(); resizes <- domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}} }()
461 go func() {
462 defer wg.Done()
463 actions <- AttachmentLifecycleAction{Token: fg.token, Kind: AttachmentDetachToPicker}
464 }()
465 go func() { defer wg.Done(); fg.finish() }()
466 wg.Wait()
467 cancel()
468 host.stopGeometry()
469 }
470}