1package daemonmux
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "net"
9 "os/exec"
10 "runtime"
11 "sync"
12 "sync/atomic"
13 "testing"
14 "time"
15
16 "github.com/stretchr/testify/require"
17
18 "github.com/bnema/vev/internal/adapters/clock"
19 "github.com/bnema/vev/internal/adapters/ipc"
20 "github.com/bnema/vev/internal/adapters/sshstdio"
21 "github.com/bnema/vev/internal/domain"
22 "github.com/bnema/vev/internal/ports"
23 "github.com/bnema/vev/internal/protocol"
24)
25
26// p3cTestDeadline bounds every p3c integration wait. It is deliberately
27// generous wall time: the only timing under test is a real carriage loss or an
28// admission handshake, never a handshake budget.
29const p3cTestDeadline = 5 * time.Second
30
31func TestDialErrorClassifiesMissingDaemonSeparately(t *testing.T) {
32 if runtime.GOOS == "windows" {
33 t.Skip("test uses the POSIX shell")
34 }
35 exitError := func(t *testing.T, code int) error {
36 t.Helper()
37 cmd := exec.Command("sh", "-c", fmt.Sprintf("exit %d", code))
38 err := cmd.Run()
39 var exit *exec.ExitError
40 require.ErrorAs(t, err, &exit)
41 return errors.Join(sshstdio.ErrMuxSSHExit, exit)
42 }
43
44 tests := []struct {
45 name string
46 err error
47 want ports.BrokerErrorCode
48 }{
49 {name: "reserved helper status means no daemon", err: exitError(t, sshstdio.MuxExitNoDaemon), want: ports.BrokerErrorNoDaemon},
50 {name: "ssh transport status remains unavailable", err: exitError(t, 255), want: ports.BrokerErrorUnavailable},
51 {name: "other helper status remains unavailable", err: exitError(t, 127), want: ports.BrokerErrorUnavailable},
52 {name: "zero exit after carriage end remains unavailable", err: errors.Join(sshstdio.ErrMuxSSHExit, io.EOF), want: ports.BrokerErrorUnavailable},
53 {name: "other mux failure remains unavailable", err: errors.New("socket timeout"), want: ports.BrokerErrorUnavailable},
54 }
55 for _, tt := range tests {
56 t.Run(tt.name, func(t *testing.T) {
57 for _, classify := range []struct {
58 name string
59 fn func(error) error
60 }{
61 {name: "dial", fn: dialError},
62 {name: "handshake", fn: handshakeError},
63 } {
64 t.Run(classify.name, func(t *testing.T) {
65 var brokerErr ports.BrokerError
66 require.ErrorAs(t, classify.fn(tt.err), &brokerErr)
67 require.Equal(t, tt.want, brokerErr.Code)
68 })
69 }
70 })
71 }
72}
73
74// rawCarriagePair returns two independent real raw framed transports joined by
75// an in-memory duplex pipe. Each end frames through the shared streamframe
76// carriage via the IPC adapter, so the bridge, physical preamble, pump, and
77// sessionwire all run over real framing rather than an in-memory envelope
78// channel.
79func rawCarriagePair(t *testing.T) (RawFramedTransport, RawFramedTransport) {
80 t.Helper()
81 client, server := net.Pipe()
82 t.Cleanup(func() {
83 _ = client.Close()
84 _ = server.Close()
85 })
86 return rawFramedEnd(t, client), rawFramedEnd(t, server)
87}
88
89// rawFramedEnd adapts one net.Conn to the raw framed carriage the bridge and
90// supervisor consume.
91func rawFramedEnd(t *testing.T, conn net.Conn) RawFramedTransport {
92 t.Helper()
93 raw, ok := ipc.NewTransport(conn).(RawFramedTransport)
94 require.True(t, ok, "ipc transport must satisfy the raw framed carriage")
95 return raw
96}
97
98// alternatePolicy returns a valid policy that differs from base in exactly one
99// field, so a mismatch is never a validation failure.
100func alternatePolicy(base ports.BrokerPolicy) ports.BrokerPolicy {
101 other := base
102 other.Transport = base.Transport + "-alt"
103 return other
104}
105
106// muxOpenRequest builds one valid local control-purpose open request under
107// policy, so a client can open a stream without a remote registration.
108func muxOpenRequest(stream uint64, policy ports.BrokerPolicy) ports.BrokerOpenStreamRequest {
109 return ports.BrokerOpenStreamRequest{
110 Epoch: 1,
111 Purpose: ports.BrokerStreamControl,
112 Local: true,
113 Connection: ports.BrokerConnectionID{1},
114 Stream: ports.BrokerStreamID(stream),
115 Policy: policy,
116 StartMode: ports.BrokerDaemonStartIfNeeded,
117 }
118}
119
120// muxEndpoint builds the resolved endpoint the broker verifies against.
121func muxEndpoint(identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy, address string) ports.BrokerDialTarget {
122 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: policy, Address: address, StartMode: ports.BrokerDaemonExistingOnly, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: identity, Bound: true}}
123}
124
125// superviseServer is a test daemon: an AggregateListener plus a
126// ServerSupervisor, with an optional accept loop that records and serves every
127// typed connection the aggregate delivers. Its dial method produces the broker
128// end of one real carriage and hands the daemon end to the supervisor.
129type superviseServer struct {
130 t *testing.T
131 aggregate *AggregateListener
132 supervisor *ServerSupervisor
133 dials atomic.Int64
134 adoptions chan error
135
136 mu sync.Mutex
137 accepted []ports.ServerConnection
138 serveOne sync.Once
139}
140
141func newSuperviseServer(t *testing.T, binding ServerBinding, ceilings MuxCeilings, limit int, timeSource ports.Clock) *superviseServer {
142 t.Helper()
143 if ceilings == (MuxCeilings{}) {
144 ceilings = DefaultMuxCeilings()
145 }
146 if timeSource == nil {
147 timeSource = clock.New()
148 }
149 aggregate := NewAggregateListener()
150 supervisor, err := newServerSupervisor(aggregate, binding, ceilings, limit, timeSource)
151 require.NoError(t, err)
152 t.Cleanup(func() {
153 _ = supervisor.Close()
154 _ = aggregate.Close()
155 })
156 return &superviseServer{
157 t: t,
158 aggregate: aggregate,
159 supervisor: supervisor,
160 adoptions: make(chan error, 16),
161 }
162}
163
164// dial returns a RawCarrierDialer that produces one fresh real carriage per
165// call and adopts its daemon end.
166func (s *superviseServer) dial() RawCarrierDialer {
167 return func(_ context.Context, _ ports.BrokerDialTarget) (RawFramedTransport, error) {
168 s.dials.Add(1)
169 client, server := rawCarriagePair(s.t)
170 go func() { s.adoptions <- s.supervisor.Adopt(context.Background(), server) }()
171 return client, nil
172 }
173}
174
175// adopt hands one already built carriage to the supervisor and reports the
176// outcome through the shared adoption channel.
177func (s *superviseServer) adopt(ctx context.Context, server RawFramedTransport) {
178 go func() { s.adoptions <- s.supervisor.Adopt(ctx, server) }()
179}
180
181// awaitAdoption returns the next Adopt outcome.
182func (s *superviseServer) awaitAdoption(t *testing.T) error {
183 t.Helper()
184 select {
185 case err := <-s.adoptions:
186 return err
187 case <-time.After(p3cTestDeadline):
188 t.Fatal("supervisor: Adopt did not return")
189 return nil
190 }
191}
192
193// serve starts the accept loop: every delivered typed connection is recorded
194// and answered with Pong per Ping until its stream ends.
195func (s *superviseServer) serve() {
196 s.serveOne.Do(func() {
197 go func() {
198 for {
199 connection, err := s.aggregate.Accept()
200 if err != nil {
201 return
202 }
203 s.mu.Lock()
204 s.accepted = append(s.accepted, connection)
205 s.mu.Unlock()
206 go func() { _ = servePongs(connection) }()
207 }
208 }()
209 })
210}
211
212// acceptedCount reports how many typed connections the aggregate delivered.
213func (s *superviseServer) acceptedCount() int {
214 s.mu.Lock()
215 defer s.mu.Unlock()
216 return len(s.accepted)
217}
218
219func (s *superviseServer) awaitAccepted(t *testing.T, count int) {
220 t.Helper()
221 require.Eventually(t, func() bool { return s.acceptedCount() >= count }, p3cTestDeadline, time.Millisecond,
222 "server accepted %d of %d streams", s.acceptedCount(), count)
223}
224
225// TestNewEndpointConnectorValidation proves a connector refuses a nil dial
226// function or an invalid ceiling advertisement, and Connect refuses an invalid
227// resolved endpoint.
228func TestNewEndpointConnectorValidation(t *testing.T) {
229 _, err := NewEndpointConnector(nil, DefaultMuxCeilings())
230 require.ErrorIs(t, err, ErrConnectorConfig)
231
232 _, err = NewEndpointConnector(func(context.Context, ports.BrokerDialTarget) (RawFramedTransport, error) { return nil, nil }, MuxCeilings{})
233 require.ErrorIs(t, err, ErrConnectorConfig)
234
235 connector, err := NewEndpointConnector(func(context.Context, ports.BrokerDialTarget) (RawFramedTransport, error) {
236 t.Fatal("dial must not run for an invalid endpoint")
237 return nil, nil
238 }, DefaultMuxCeilings())
239 require.NoError(t, err)
240
241 _, err = connector.Connect(context.Background(), ports.BrokerDialTarget{})
242 require.Error(t, err)
243 var typed ports.BrokerError
244 require.ErrorAs(t, err, &typed)
245 require.Equal(t, ports.BrokerErrorIncompatible, typed.Code)
246}
247
248// TestEndpointConnectorRefusesPolicyMismatch proves the resolved endpoint stays
249// authoritative: a daemon that does not accept the endpoint policy is refused,
250// no physical connection is published, and the daemon admits no child.
251func TestEndpointConnectorRefusesPolicyMismatch(t *testing.T) {
252 binding := mustServerBinding(t)
253 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
254
255 connector, err := NewEndpointConnector(server.dial(), DefaultMuxCeilings())
256 require.NoError(t, err)
257 physical, err := connector.Connect(context.Background(), muxEndpoint(binding.Identity(), alternatePolicy(binding.Policy()), "raw://mismatch"))
258 require.Error(t, err)
259 require.Nil(t, physical)
260 require.ErrorIs(t, err, ErrHandshakeRejected)
261 var typed ports.BrokerError
262 require.ErrorAs(t, err, &typed)
263 require.Equal(t, ports.BrokerErrorConflictingPolicy, typed.Code)
264
265 require.ErrorIs(t, server.awaitAdoption(t), ErrHandshakeRejected)
266 require.Zero(t, server.supervisor.Children(), "a refused handshake admits no physical child")
267 require.Equal(t, int64(1), server.dials.Load())
268}
269
270// TestEndpointConnectorExposesAuthorityAndPolicyGuard proves one successful
271// connection exposes the immutable accepted authority, delegates its terminal
272// results to the pump, refuses a mismatched Open without touching the carriage,
273// and admits a matching Open over the real carriage.
274func TestEndpointConnectorExposesAuthorityAndPolicyGuard(t *testing.T) {
275 binding := mustServerBinding(t)
276 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
277 server.serve()
278
279 connector, err := NewEndpointConnector(server.dial(), DefaultMuxCeilings())
280 require.NoError(t, err)
281 physical, err := connector.Connect(context.Background(), muxEndpoint(binding.Identity(), binding.Policy(), "raw://authority"))
282 require.NoError(t, err)
283 require.NotNil(t, physical)
284 t.Cleanup(func() { _ = physical.Close() })
285 require.NoError(t, server.awaitAdoption(t))
286
287 require.Equal(t, binding.Identity(), physical.Identity())
288 require.Equal(t, binding.Incarnation(), physical.Incarnation())
289 require.Equal(t, binding.Policy(), physical.Policy())
290 require.False(t, channelClosed(physical.Done()))
291 require.NoError(t, physical.Err())
292 require.Equal(t, domain.RemoteFailureNone, physical.FailureKind())
293
294 // A mismatched request policy is refused before the wire is touched.
295 _, err = openTyped(physical, context.Background(), muxOpenRequest(1, alternatePolicy(binding.Policy())))
296 require.Error(t, err)
297 var typed ports.BrokerError
298 require.ErrorAs(t, err, &typed)
299 require.Equal(t, ports.BrokerErrorConflictingPolicy, typed.Code)
300 require.Zero(t, server.acceptedCount(), "a policy-mismatched open is never admitted")
301
302 // The matching open is admitted and usable over the real carriage.
303 logical, err := openTyped(physical, context.Background(), muxOpenRequest(1, binding.Policy()))
304 require.NoError(t, err)
305 server.awaitAccepted(t, 1)
306 require.NoError(t, logical.SendClient(protocol.Ping{}))
307 message, err := logical.ReceiveServer()
308 require.NoError(t, err)
309 require.Equal(t, protocol.Pong{}, message)
310
311 require.NoError(t, logical.Close())
312 require.NoError(t, physical.Close())
313 require.True(t, channelClosed(physical.Done()))
314 require.NoError(t, physical.Err(), "an orderly local Close is not a failure")
315 require.Equal(t, domain.RemoteFailureNone, physical.FailureKind())
316}
317
318// TestEndpointConnectorDetachesSetupContext proves a successful connection is
319// not coupled to the caller's setup context: cancelling it after Connect
320// returned does not stop the pooled physical connection.
321func TestEndpointConnectorDetachesSetupContext(t *testing.T) {
322 binding := mustServerBinding(t)
323 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
324 server.serve()
325
326 connector, err := NewEndpointConnector(server.dial(), DefaultMuxCeilings())
327 require.NoError(t, err)
328 setupCtx, cancel := context.WithCancel(context.Background())
329 physical, err := connector.Connect(setupCtx, muxEndpoint(binding.Identity(), binding.Policy(), "raw://detach"))
330 require.NoError(t, err)
331 require.NoError(t, server.awaitAdoption(t))
332 t.Cleanup(func() { _ = physical.Close() })
333
334 cancel()
335 // The setup context ended; the pooled connection must still work.
336 require.False(t, channelClosed(physical.Done()))
337 logical, err := openTyped(physical, context.Background(), muxOpenRequest(1, binding.Policy()))
338 require.NoError(t, err)
339 server.awaitAccepted(t, 1)
340 require.NoError(t, logical.SendClient(protocol.Ping{}))
341 message, err := logical.ReceiveServer()
342 require.NoError(t, err)
343 require.Equal(t, protocol.Pong{}, message)
344 require.NoError(t, logical.Close())
345}
346
347// TestPhysicalDoneBeforeLogicalLoss proves a carriage loss publishes the
348// physical terminal outcome before the affected logical stream is terminalized:
349// by the time the logical Done fires, the physical Done is already closed, and
350// both carry the loss cause.
351func TestPhysicalDoneBeforeLogicalLoss(t *testing.T) {
352 binding := mustServerBinding(t)
353 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
354 server.serve()
355
356 connector, err := NewEndpointConnector(server.dial(), DefaultMuxCeilings())
357 require.NoError(t, err)
358 physical, err := connector.Connect(context.Background(), muxEndpoint(binding.Identity(), binding.Policy(), "raw://loss"))
359 require.NoError(t, err)
360 t.Cleanup(func() { _ = physical.Close() })
361 require.NoError(t, server.awaitAdoption(t))
362
363 logical, err := openTyped(physical, context.Background(), muxOpenRequest(1, binding.Policy()))
364 require.NoError(t, err)
365 server.awaitAccepted(t, 1)
366 require.False(t, channelClosed(physical.Done()))
367
368 ordered := make(chan bool, 1)
369 go func() {
370 select {
371 case <-logical.Done():
372 case <-time.After(p3cTestDeadline):
373 ordered <- false
374 return
375 }
376 select {
377 case <-physical.Done():
378 ordered <- true
379 case <-time.After(p3cTestDeadline):
380 ordered <- false
381 }
382 }()
383
384 // Kill the daemon's carriage: the broker pump observes the carrier error.
385 server.supervisor.mu.Lock()
386 children := make([]*supervisedPhysical, 0, len(server.supervisor.children))
387 for child := range server.supervisor.children {
388 children = append(children, child)
389 }
390 server.supervisor.mu.Unlock()
391 require.Len(t, children, 1)
392 require.NoError(t, children[0].Close())
393
394 require.True(t, <-ordered, "the logical Done fired before the physical Done")
395 require.True(t, channelClosed(physical.Done()))
396 require.Error(t, physical.Err())
397 require.NotEqual(t, domain.RemoteFailureNone, physical.FailureKind())
398 require.Error(t, logical.Err())
399}
400
401// TestEndpointConnectorStableIdentityDifferentIncarnationRestart proves a
402// restart behind a stable authenticated identity surfaces a fresh incarnation:
403// the identity and policy stay equal, the incarnation changes, and each
404// connection reports its own captured authority.
405func TestEndpointConnectorStableIdentityDifferentIncarnationRestart(t *testing.T) {
406 first, err := NewServerBinding(testIdentity(), testIncarnation(), testPolicy())
407 require.NoError(t, err)
408 second, err := NewServerBinding(testIdentity(), ports.BrokerDaemonIncarnation{9, 8, 7, 6, 5, 4, 3, 2, 1, 0, 1, 2, 3, 4, 5, 6}, testPolicy())
409 require.NoError(t, err)
410 require.Equal(t, first.Identity(), second.Identity())
411 require.NotEqual(t, first.Incarnation(), second.Incarnation())
412
413 serverA := newSuperviseServer(t, first, DefaultMuxCeilings(), 0, nil)
414 serverB := newSuperviseServer(t, second, DefaultMuxCeilings(), 0, nil)
415
416 connectorA, err := NewEndpointConnector(serverA.dial(), DefaultMuxCeilings())
417 require.NoError(t, err)
418 physicalA, err := connectorA.Connect(context.Background(), muxEndpoint(first.Identity(), first.Policy(), "raw://restart-a"))
419 require.NoError(t, err)
420 require.NoError(t, serverA.awaitAdoption(t))
421 require.Equal(t, first.Incarnation(), physicalA.Incarnation())
422 require.NoError(t, physicalA.Close())
423
424 connectorB, err := NewEndpointConnector(serverB.dial(), DefaultMuxCeilings())
425 require.NoError(t, err)
426 physicalB, err := connectorB.Connect(context.Background(), muxEndpoint(second.Identity(), second.Policy(), "raw://restart-b"))
427 require.NoError(t, err)
428 require.NoError(t, serverB.awaitAdoption(t))
429 t.Cleanup(func() { _ = physicalB.Close() })
430
431 require.Equal(t, physicalA.Identity(), physicalB.Identity(), "the authenticated identity is stable across a restart")
432 require.Equal(t, physicalA.Policy(), physicalB.Policy())
433 require.NotEqual(t, physicalA.Incarnation(), physicalB.Incarnation(), "a restart carries a fresh incarnation")
434}