1package daemonmux
2
3import (
4 "context"
5 "testing"
6 "time"
7
8 "github.com/stretchr/testify/require"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/bnema/vev/internal/protocol"
13)
14
15// connectRaw builds one broker PhysicalConnection over an already established
16// raw carriage whose daemon end was handed to a supervisor.
17func connectRaw(t *testing.T, raw RawFramedTransport, endpoint ports.BrokerDialTarget) ports.BrokerPhysicalConnection {
18 t.Helper()
19 connector, err := NewEndpointConnector(func(context.Context, ports.BrokerDialTarget) (RawFramedTransport, error) {
20 return raw, nil
21 }, DefaultMuxCeilings())
22 require.NoError(t, err)
23 physical, err := connector.Connect(context.Background(), endpoint)
24 require.NoError(t, err)
25 require.NotNil(t, physical)
26 t.Cleanup(func() { _ = physical.Close() })
27 return physical
28}
29
30// awaitHandshakeBridge performs the broker side of the daemonmux physical
31// preamble over raw and returns the negotiated carrier, ready for a pump.
32func awaitHandshakeBridge(t *testing.T, raw RawFramedTransport, endpoint ports.BrokerDialTarget) (FramedCarrier, ClientHandshakeResult) {
33 t.Helper()
34 bridge, err := NewPreambleCarrier(raw)
35 require.NoError(t, err)
36 result, err := RunClientHandshake(context.Background(), bridge, endpoint, DefaultMuxCeilings())
37 require.NoError(t, err)
38 require.NoError(t, bridge.Negotiate(result.Ceilings))
39 return bridge, result
40}
41
42// TestNewServerSupervisorValidation proves the supervisor refuses a nil
43// aggregate, an invalid binding, an invalid ceiling advertisement, and a nil
44// carriage.
45func TestNewServerSupervisorValidation(t *testing.T) {
46 binding := mustServerBinding(t)
47
48 _, err := NewServerSupervisor(nil, binding, DefaultMuxCeilings(), 0)
49 require.ErrorIs(t, err, ErrSupervisorConfig)
50
51 _, err = NewServerSupervisor(NewAggregateListener(), ServerBinding{}, DefaultMuxCeilings(), 0)
52 require.ErrorIs(t, err, ErrSupervisorConfig)
53
54 _, err = NewServerSupervisor(NewAggregateListener(), binding, MuxCeilings{}, 0)
55 require.ErrorIs(t, err, ErrSupervisorConfig)
56
57 supervisor, err := NewServerSupervisor(NewAggregateListener(), binding, DefaultMuxCeilings(), 0)
58 require.NoError(t, err)
59 t.Cleanup(func() { _ = supervisor.Close() })
60 require.ErrorIs(t, supervisor.Adopt(context.Background(), nil), ErrSupervisorConfig)
61}
62
63// TestServerSupervisorFailedHandshakeRegistersNoChild proves a preamble that
64// the binding refuses closes the carriage, admits no stream, and registers no
65// owned child with the aggregate.
66func TestServerSupervisorFailedHandshakeRegistersNoChild(t *testing.T) {
67 binding := mustServerBinding(t)
68 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
69
70 clientRaw, serverRaw := rawCarriagePair(t)
71 server.adopt(context.Background(), serverRaw)
72
73 bridge, err := NewPreambleCarrier(clientRaw)
74 require.NoError(t, err)
75 t.Cleanup(func() { _ = bridge.Close() })
76 _, err = RunClientHandshake(context.Background(), bridge, muxEndpoint(binding.Identity(), alternatePolicy(binding.Policy()), "raw://refused"), DefaultMuxCeilings())
77 require.ErrorIs(t, err, ErrHandshakeRejected)
78
79 require.ErrorIs(t, server.awaitAdoption(t), ErrHandshakeRejected)
80 require.Zero(t, server.supervisor.Children(), "a refused handshake admits no physical child")
81 require.Zero(t, server.acceptedCount(), "a refused handshake admits no stream")
82}
83
84// TestServerSupervisorEnforcesAcceptedPolicyBeforeAdmission proves the accepted
85// physical policy is enforced on every inbound Open before fresh stream
86// admission: a broker that completed the handshake at the accepted policy but
87// names a different policy in an Open is refused on that stream alone, while a
88// matching Open is admitted to the daemon's accept stream.
89func TestServerSupervisorEnforcesAcceptedPolicyBeforeAdmission(t *testing.T) {
90 binding := mustServerBinding(t)
91 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
92 server.serve()
93 endpoint := muxEndpoint(binding.Identity(), binding.Policy(), "raw://stream-policy")
94
95 clientRaw, serverRaw := rawCarriagePair(t)
96 server.adopt(context.Background(), serverRaw)
97 carrier, result := awaitHandshakeBridge(t, clientRaw, endpoint)
98 require.NoError(t, server.awaitAdoption(t))
99
100 pump, err := NewPump(carrier, DirectionServer, result.Ceilings)
101 require.NoError(t, err)
102 t.Cleanup(func() { _ = pump.Close() })
103 pump.Start(context.Background())
104 logical, err := NewLogicalConnector(pump)
105 require.NoError(t, err)
106
107 _, err = logical.Open(context.Background(), muxOpenRequest(1, alternatePolicy(binding.Policy())))
108 require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
109 require.Zero(t, server.acceptedCount(), "a policy-mismatched open never reaches the accept stream")
110
111 connection, err := logical.Open(context.Background(), muxOpenRequest(2, binding.Policy()))
112 require.NoError(t, err)
113 require.NotNil(t, connection)
114 server.awaitAccepted(t, 1)
115 typed := asTyped(connection)
116 require.NoError(t, typed.SendClient(protocol.Ping{}))
117 message, err := typed.ReceiveServer()
118 require.NoError(t, err)
119 require.Equal(t, protocol.Pong{}, message)
120}
121
122// TestServerSupervisorDeliversFirstHelloDataFrame proves the complete first
123// application-frame path without timing probes: LogicalConnection.SendClient
124// writes Hello through sessionwire, the physical pump emits Data, the
125// supervisor-owned daemon pump dispatches it, and the accepted logical server
126// connection's sessionwire ReceiveClient decodes that same Hello. This guards
127// the Opened-to-first-Data boundary where losing a wakeup or using the wrong
128// direction/physical ID would otherwise leave the daemon blocked forever.
129func TestServerSupervisorDeliversFirstHelloDataFrame(t *testing.T) {
130 binding := mustServerBinding(t)
131 aggregate := NewAggregateListener()
132 supervisor, err := NewServerSupervisor(aggregate, binding, DefaultMuxCeilings(), 1)
133 require.NoError(t, err)
134 t.Cleanup(func() {
135 _ = supervisor.Close()
136 _ = aggregate.Close()
137 })
138
139 clientRaw, serverRaw := rawCarriagePair(t)
140 adopted := make(chan error, 1)
141 go func() { adopted <- supervisor.Adopt(context.Background(), serverRaw) }()
142 physical := connectRaw(t, clientRaw, muxEndpoint(binding.Identity(), binding.Policy(), "raw://first-hello"))
143 require.NoError(t, <-adopted)
144
145 accepted := make(chan ports.ServerConnection, 1)
146 acceptErr := make(chan error, 1)
147 go func() {
148 connection, err := aggregate.Accept()
149 if err != nil {
150 acceptErr <- err
151 return
152 }
153 accepted <- connection
154 }()
155
156 logical, err := openTyped(physical, context.Background(), muxOpenRequest(1, binding.Policy()))
157 require.NoError(t, err)
158 var server ports.ServerConnection
159 select {
160 case server = <-accepted:
161 case err := <-acceptErr:
162 require.NoError(t, err)
163 case <-time.After(p3cTestDeadline):
164 t.Fatal("supervisor did not dispatch the admitted logical stream")
165 }
166
167 hello := protocol.Hello{Version: protocol.Version, Intent: protocol.IntentEphemeral, Size: domain.Size{Cols: 80, Rows: 24}, EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned}
168 sent := make(chan error, 1)
169 received := make(chan protocol.ClientMessage, 1)
170 receiveErr := make(chan error, 1)
171 go func() { sent <- logical.SendClient(hello) }()
172 go func() {
173 message, err := server.ReceiveClient()
174 if err != nil {
175 receiveErr <- err
176 return
177 }
178 received <- message
179 }()
180 select {
181 case message := <-received:
182 require.Equal(t, hello, message)
183 case err := <-receiveErr:
184 require.NoError(t, err)
185 case <-time.After(p3cTestDeadline):
186 t.Fatal("first Hello Data frame did not reach sessionwire ReceiveClient")
187 }
188 require.NoError(t, <-sent)
189}
190
191// TestServerSupervisorStampsAcceptedOriginThroughAggregate proves the accepted
192// physical origin the supervisor resolved from the binding's provisioned member
193// reaches the daemon end-to-end: the aggregate hands back the same typed
194// connection the child listener produced, and its admission provider reports the
195// exact accepted policy and origin. It guards the wrapper chain so a later
196// wrapper can never strip a child's admission metadata.
197func TestServerSupervisorStampsAcceptedOriginThroughAggregate(t *testing.T) {
198 binding := mustServerBinding(t)
199 aggregate := NewAggregateListener()
200 supervisor, err := NewServerSupervisor(aggregate, binding, DefaultMuxCeilings(), 1)
201 require.NoError(t, err)
202 t.Cleanup(func() {
203 _ = supervisor.Close()
204 _ = aggregate.Close()
205 })
206
207 clientRaw, serverRaw := rawCarriagePair(t)
208 adopted := make(chan error, 1)
209 go func() { adopted <- supervisor.Adopt(context.Background(), serverRaw) }()
210 physical := connectRaw(t, clientRaw, muxEndpoint(binding.Identity(), binding.Policy(), "raw://origin"))
211 require.NoError(t, <-adopted)
212
213 accepted := make(chan ports.ServerConnection, 1)
214 acceptErr := make(chan error, 1)
215 go func() {
216 connection, err := aggregate.Accept()
217 if err != nil {
218 acceptErr <- err
219 return
220 }
221 accepted <- connection
222 }()
223
224 logical, err := openTyped(physical, context.Background(), muxOpenRequest(1, binding.Policy()))
225 require.NoError(t, err)
226 t.Cleanup(func() { _ = logical.Close() })
227
228 var server ports.ServerConnection
229 select {
230 case server = <-accepted:
231 case err := <-acceptErr:
232 require.NoError(t, err)
233 case <-time.After(p3cTestDeadline):
234 t.Fatal("supervisor did not dispatch the admitted logical stream")
235 }
236
237 provider, ok := server.(ports.SessionAdmissionProvider)
238 require.True(t, ok, "the accepted connection preserves the admission provider through the aggregate")
239 admission, ok := provider.SessionAdmission()
240 require.True(t, ok)
241 require.Equal(t, ports.SessionOriginLocal, admission.Origin)
242 require.Equal(t, binding.Policy(), admission.Policy)
243 require.Equal(t, ports.BrokerStreamControl, admission.Purpose)
244 require.NoError(t, admission.Validate())
245}
246
247// TestServerSupervisorRefusesLiarRegistration proves the supervisor's accepted
248// origin reaches the listener's admission check: a client that lies about its
249// locality - a local registration on a local carriage - is refused on its own
250// stream and never delivered to the daemon.
251func TestServerSupervisorRefusesLiarRegistration(t *testing.T) {
252 binding := mustServerBinding(t)
253 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
254 server.serve()
255
256 clientRaw, serverRaw := rawCarriagePair(t)
257 server.adopt(context.Background(), serverRaw)
258 carrier, result := awaitHandshakeBridge(t, clientRaw, muxEndpoint(binding.Identity(), binding.Policy(), "raw://liar"))
259 require.NoError(t, server.awaitAdoption(t))
260
261 pump, err := NewPump(carrier, DirectionServer, result.Ceilings)
262 require.NoError(t, err)
263 t.Cleanup(func() { _ = pump.Close() })
264 pump.Start(context.Background())
265 logical, err := NewLogicalConnector(pump)
266 require.NoError(t, err)
267
268 // A remote-shaped Open (a local carriage with a registration) contradicts
269 // the accepted local origin: the listener refuses it rather than stamping a
270 // provenance it did not provision.
271 liar := p3cRemoteRequest(ports.BrokerConnectionID{1}, "dev@host:22", 1, binding.Policy())
272 _, err = logical.Open(context.Background(), liar)
273 require.Error(t, err)
274 require.Zero(t, server.acceptedCount(), "a locality-contradicting open never reaches the accept stream")
275}
276
277// TestServerSupervisorServerLossIsolated proves one lost physical child never
278// stops the aggregate: a surviving child keeps accepting while the lost child
279// is reaped and removed.
280func TestServerSupervisorServerLossIsolated(t *testing.T) {
281 binding := mustServerBinding(t)
282 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
283 server.serve()
284 endpoint := muxEndpoint(binding.Identity(), binding.Policy(), "raw://isolation")
285
286 clientA, serverA := rawCarriagePair(t)
287 server.adopt(context.Background(), serverA)
288 physicalA := connectRaw(t, clientA, endpoint)
289 require.NoError(t, server.awaitAdoption(t))
290
291 clientB, serverB := rawCarriagePair(t)
292 server.adopt(context.Background(), serverB)
293 physicalB := connectRaw(t, clientB, endpoint)
294 require.NoError(t, server.awaitAdoption(t))
295
296 logicalA, err := openTyped(physicalA, context.Background(), muxOpenRequest(1, binding.Policy()))
297 require.NoError(t, err)
298 logicalB, err := openTyped(physicalB, context.Background(), muxOpenRequest(1, binding.Policy()))
299 require.NoError(t, err)
300 server.awaitAccepted(t, 2)
301 require.Equal(t, 2, server.supervisor.Children())
302
303 // Lose child A's carriage entirely.
304 _ = serverA.Close()
305 require.Eventually(t, func() bool { return channelClosed(physicalA.Done()) }, p3cTestDeadline, time.Millisecond,
306 "the lost physical connection never reached its terminal outcome")
307 require.Eventually(t, func() bool { return server.supervisor.Children() == 1 }, p3cTestDeadline, time.Millisecond,
308 "the lost child was never reaped")
309
310 // The surviving child still accepts and the aggregate never failed.
311 logicalB2, err := openTyped(physicalB, context.Background(), muxOpenRequest(2, binding.Policy()))
312 require.NoError(t, err)
313 require.NotNil(t, logicalB2)
314 server.awaitAccepted(t, 3)
315
316 // The lost physical refuses a fresh stream while its published logical
317 // stream reports the physical loss.
318 _, err = openTyped(physicalA, context.Background(), muxOpenRequest(2, binding.Policy()))
319 require.Error(t, err)
320 require.True(t, channelClosed(logicalA.Done()))
321 require.False(t, channelClosed(logicalB.Done()))
322}
323
324// TestServerSupervisorCloseTearsDownOwnedChildren proves Close removes and
325// closes every owned child, refuses later carriage, and leaves the aggregate
326// open.
327func TestServerSupervisorCloseTearsDownOwnedChildren(t *testing.T) {
328 binding := mustServerBinding(t)
329 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
330 endpoint := muxEndpoint(binding.Identity(), binding.Policy(), "raw://close")
331
332 clientA, serverA := rawCarriagePair(t)
333 server.adopt(context.Background(), serverA)
334 physicalA := connectRaw(t, clientA, endpoint)
335 require.NoError(t, server.awaitAdoption(t))
336
337 clientB, serverB := rawCarriagePair(t)
338 server.adopt(context.Background(), serverB)
339 physicalB := connectRaw(t, clientB, endpoint)
340 require.NoError(t, server.awaitAdoption(t))
341 require.Equal(t, 2, server.supervisor.Children())
342
343 require.NoError(t, server.supervisor.Close())
344 require.Zero(t, server.supervisor.Children())
345 require.Eventually(t, func() bool {
346 return channelClosed(physicalA.Done()) && channelClosed(physicalB.Done())
347 }, p3cTestDeadline, time.Millisecond, "Close did not tear down every owned carriage")
348
349 // A closed supervisor refuses later carriage and closes it.
350 lateClient, lateServer := rawCarriagePair(t)
351 t.Cleanup(func() { _ = lateClient.Close() })
352 require.ErrorIs(t, server.supervisor.Adopt(context.Background(), lateServer), ErrSupervisorClosed)
353
354 // Close is idempotent.
355 require.NoError(t, server.supervisor.Close())
356}
357
358// TestServerSupervisorBoundsConcurrentAdmissions proves admission is bounded: a
359// second carriage waits while the first holds the only admission slot in its
360// handshake, and proceeds once the first finishes.
361func TestServerSupervisorBoundsConcurrentAdmissions(t *testing.T) {
362 binding := mustServerBinding(t)
363 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 1, nil)
364 endpoint := muxEndpoint(binding.Identity(), binding.Policy(), "raw://bounds")
365
366 clientA, serverA := rawCarriagePair(t)
367 server.adopt(context.Background(), serverA)
368 require.Eventually(t, func() bool { return len(server.supervisor.slots) == 1 }, p3cTestDeadline, time.Millisecond,
369 "the first admission never took the only slot")
370
371 clientB, serverB := rawCarriagePair(t)
372 admittedB := make(chan error, 1)
373 go func() { admittedB <- server.supervisor.Adopt(context.Background(), serverB) }()
374 select {
375 case err := <-admittedB:
376 t.Fatalf("the admission bound did not hold the second carriage: %v", err)
377 case <-time.After(150 * time.Millisecond):
378 }
379
380 // Finish the first handshake; the slot is released and the second proceeds.
381 bridgeA, _ := awaitHandshakeBridge(t, clientA, endpoint)
382 t.Cleanup(func() { _ = bridgeA.Close() })
383 require.NoError(t, server.awaitAdoption(t))
384
385 bridgeB, _ := awaitHandshakeBridge(t, clientB, endpoint)
386 t.Cleanup(func() { _ = bridgeB.Close() })
387 select {
388 case err := <-admittedB:
389 require.NoError(t, err)
390 case <-time.After(p3cTestDeadline):
391 t.Fatal("the second carriage never proceeded after the first released its slot")
392 }
393 require.Equal(t, 2, server.supervisor.Children())
394}