supervisor_picker_parity_test.go RawHistoryFolder
1package client
2
3import (
4 "context"
5 "errors"
6 "strings"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/require"
12
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/bnema/vev/internal/protocol"
16)
17
18// Supervisor halves of picker parity (Plan 003 B1/B2): the in-place tab switch
19// over a live attachment and the `x` kill through the broker operations seam.
20// They drive the real supervisor, host, foreground, and worker with bounded
21// waits and no sleeps.
22
23// parityTestPicker adds the optional kill and notice seams of the real picker
24// to the scripted picker. The fields are guarded by attachTestPicker.mu.
25type parityTestPicker struct {
26 *attachTestPicker
27 kill pickerKillTarget
28 notices []string
29}
30
31func newParityTestPicker() *parityTestPicker {
32 return &parityTestPicker{attachTestPicker: newAttachTestPicker()}
33}
34
35func (p *parityTestPicker) ResolveKill(string) (pickerKillTarget, error) {
36 p.mu.Lock()
37 defer p.mu.Unlock()
38 return p.kill, nil
39}
40
41func (p *parityTestPicker) Notify(n domain.Notification) {
42 p.mu.Lock()
43 p.notices = append(p.notices, n.Message)
44 p.mu.Unlock()
45}
46
47func (p *parityTestPicker) invalidatePresentation() {}
48
49func (p *parityTestPicker) noticeList() []string {
50 p.mu.Lock()
51 defer p.mu.Unlock()
52 return append([]string(nil), p.notices...)
53}
54
55// TestPickerOverlayCommitAnotherTabSwitchesInPlace pins the tab-aware same
56// target: committing another tab of the attached session switches the live
57// attachment's view with SelectTab and never reconnects; the attached tab or
58// the session itself just closes the overlay.
59func TestPickerOverlayCommitAnotherTabSwitchesInPlace(t *testing.T) {
60 tests := []struct {
61 name string
62 tab attachmentTab
63 inPlace bool
64 }{
65 {name: "another tab", tab: attachmentTab{preferred: "t_other"}, inPlace: true},
66 {name: "the attached tab", tab: attachmentTab{preferred: "t_abc123"}},
67 {name: "the session row", tab: attachmentTab{}},
68 }
69 for _, tt := range tests {
70 t.Run(tt.name, func(t *testing.T) {
71 picker := newParityTestPicker()
72 harness := startAttachHarness(t, picker)
73 stream := attachLiveSession(t, harness, picker.attachTestPicker)
74 foreground := harness.sup.attachments.authority.foreground()
75 stream.deliver(navigationOffer(1))
76 awaitPresentation(t, harness.sup, PresentAttachedPicker)
77
78 picker.mu.Lock()
79 picker.tab = tt.tab
80 picker.mu.Unlock()
81 picker.recordOp(pickerOp{commit: true}, "row", sessionTestRequest(true))
82
83 awaitPresentation(t, harness.sup, PresentAttached)
84 awaitSent(t, stream, "OutputResetRequest", isSent[protocol.OutputResetRequest])
85 if tt.inPlace {
86 selected := awaitSent(t, stream, "SelectTab", isSent[protocol.SelectTab]).(protocol.SelectTab)
87 require.Equal(t, tt.tab.preferred, selected.TabID)
88 } else {
89 require.Zero(t, countSent[protocol.SelectTab](stream))
90 }
91 require.Same(t, foreground, harness.sup.attachments.authority.foreground(), "the attachment is never replaced")
92 require.Len(t, harness.service.openedRequests(), 1, "a tab switch never reconnects")
93 require.Zero(t, countSent[protocol.Detach](stream))
94 require.False(t, stream.closedNow())
95 })
96 }
97}
98
99// TestSupervisorPickerKill pins B1 end to end: `x` resolves the row, sends one
100// typed Kill on its own local control stream, reports the typed outcome as a
101// notice, and asks the broker to re-observe the local daemon. An unknown
102// outcome is reported once and never replayed.
103func TestSupervisorPickerKill(t *testing.T) {
104 tests := []struct {
105 name string
106 stopped bool
107 overlay bool
108 reply func(*brokerOpsTestStream)
109 wantNotice string
110 }{
111 {name: "live session", reply: func(s *brokerOpsTestStream) {
112 s.deliver(protocol.KillResult{RequestID: brokerOperationRequestID, Outcome: protocol.KillSucceeded})
113 }, wantNotice: "killed session alpha"},
114 {name: "stopped session history", stopped: true, reply: func(s *brokerOpsTestStream) {
115 s.deliver(protocol.KillResult{RequestID: brokerOperationRequestID, Outcome: protocol.KillSucceeded})
116 }, wantNotice: "deleted stopped session alpha"},
117 {name: "daemon refusal", reply: func(s *brokerOpsTestStream) {
118 s.deliver(protocol.KillResult{RequestID: brokerOperationRequestID, Outcome: protocol.KillFailed, Code: protocol.ErrNoSuchSession, Text: "no such session: alpha"})
119 }, wantNotice: "couldn't kill alpha: no such session: alpha"},
120 {name: "outcome unknown is never replayed", reply: func(s *brokerOpsTestStream) {
121 s.failReceive(errors.New("carriage lost"))
122 _ = s.Close()
123 }, wantNotice: "kill alpha: outcome unknown"},
124 {name: "from the overlay", overlay: true, reply: func(s *brokerOpsTestStream) {
125 s.deliver(protocol.KillResult{RequestID: brokerOperationRequestID, Outcome: protocol.KillSucceeded})
126 }, wantNotice: "killed session alpha"},
127 }
128 for _, tt := range tests {
129 t.Run(tt.name, func(t *testing.T) {
130 picker := newParityTestPicker()
131 harness := startAttachHarness(t, picker)
132 harness.service.publishSnapshot(localDaemonSnapshot(3, 2))
133
134 var mu sync.Mutex
135 controls := make([]*brokerOpsTestStream, 0, 1)
136 session := newSessionTestStream()
137 harness.service.setOpenStream(func(_ context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
138 if request.Purpose != ports.BrokerStreamControl {
139 return session, nil
140 }
141 control := newBrokerOpsTestStream()
142 mu.Lock()
143 controls = append(controls, control)
144 mu.Unlock()
145 return control, nil
146 })
147 if tt.overlay {
148 picker.commit(sessionTestRequest(true))
149 awaitStreamHello(t, &sync.Mutex{}, &session)
150 deliverReadyStream(t, session)
151 awaitAttachedState(t, harness.sup)
152 session.deliver(navigationOffer(1))
153 awaitPresentation(t, harness.sup, PresentAttachedPicker)
154 }
155
156 picker.mu.Lock()
157 picker.kill = pickerKillTarget{route: BrokerOperationRoute{Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true}}, name: "alpha", stopped: tt.stopped}
158 picker.mu.Unlock()
159 picker.recordOp(pickerOp{kill: true}, "row", ports.BrokerOpenStreamRequest{})
160
161 var control *brokerOpsTestStream
162 require.Eventually(t, func() bool {
163 mu.Lock()
164 defer mu.Unlock()
165 if len(controls) == 0 {
166 return false
167 }
168 control = controls[0]
169 return true
170 }, 5*time.Second, time.Millisecond, "the kill never opened a control stream")
171 control.awaitDispatch(t)
172 tt.reply(control)
173
174 require.Eventually(t, func() bool {
175 for _, notice := range picker.noticeList() {
176 if notice == tt.wantNotice {
177 return true
178 }
179 }
180 return false
181 }, 5*time.Second, time.Millisecond, "notice %q never shown (got %v)", tt.wantNotice, picker.noticeList())
182 require.Eventually(t, func() bool { return len(harness.service.reconcileHints()) > 0 }, 5*time.Second, time.Millisecond)
183 for _, hint := range harness.service.reconcileHints() {
184 require.Empty(t, hint, "the local daemon is re-observed")
185 }
186
187 mu.Lock()
188 require.Len(t, controls, 1, "a kill is never replayed")
189 mu.Unlock()
190 sent := control.messages()
191 require.Len(t, sent, 1)
192 kill, ok := sent[0].(protocol.Kill)
193 require.True(t, ok)
194 require.Equal(t, protocol.Kill{RequestID: brokerOperationRequestID, Name: "alpha", Scope: protocol.KillSession}, kill)
195 var controlOpens int
196 for _, request := range harness.service.openedRequests() {
197 if request.Purpose == ports.BrokerStreamControl {
198 controlOpens++
199 require.True(t, request.Local)
200 require.Equal(t, ports.BrokerEpoch(3), request.Epoch)
201 }
202 }
203 require.Equal(t, 1, controlOpens)
204 if tt.overlay {
205 require.Equal(t, PresentAttachedPicker, harness.sup.State().Presentation, "a kill keeps the overlay open")
206 require.False(t, session.closedNow())
207 }
208 require.False(t, strings.Contains(strings.Join(picker.noticeList(), "\n"), "already in progress"))
209 })
210 }
211}