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 "github.com/bnema/vev/internal/usecase/broker"
14)
15
16// p3cResolver is a test BrokerRouteAuthority: it returns one authenticated
17// endpoint per request, keyed on the request's registration endpoint so two
18// compatible aliases resolve to the same (identity, policy) physical key with
19// distinct adapter routes.
20type p3cResolver func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error)
21
22func (f p3cResolver) ResolveDialTarget(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
23 return f(ctx, request)
24}
25
26type p3cBinder struct{}
27
28func (p3cBinder) BindAuthenticatedIdentity(_ context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
29 return request.Identity, nil
30}
31
32// p3cRemoteRequest builds one valid remote control-purpose request.
33func p3cRemoteRequest(connection ports.BrokerConnectionID, endpoint string, stream uint64, policy ports.BrokerPolicy) ports.BrokerOpenStreamRequest {
34 registration := domain.RemoteRegistration{
35 Endpoint: endpoint,
36 Incarnation: [16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16},
37 Generation: 1,
38 }
39 return ports.BrokerOpenStreamRequest{
40 Epoch: 1,
41 Purpose: ports.BrokerStreamControl,
42 Local: false,
43 Connection: connection,
44 Stream: ports.BrokerStreamID(stream),
45 Endpoint: endpoint,
46 Registration: registration,
47 Policy: policy,
48 StartMode: ports.BrokerDaemonStartIfNeeded,
49 }
50}
51
52// p3cPoolLimits bounds one pool with room for two aliases and two streams.
53func p3cPoolLimits() broker.PoolLimits {
54 return broker.PoolLimits{Physical: 2, Clients: 2, Streams: 4, StreamsPerClient: 4, Warm: 2, Idle: time.Hour}
55}
56
57// TestPoolSharesCompatibleAliasesOverOnePhysical proves the pool pools two
58// compatible aliases over exactly one daemonmux physical connection: only one
59// carriage is dialed, and both logical streams flow over it.
60func TestPoolSharesCompatibleAliasesOverOnePhysical(t *testing.T) {
61 binding := mustRemoteServerBinding(t)
62 policy := binding.Policy()
63 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
64 server.serve()
65 resolver := p3cResolver(func(_ context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
66 return muxEndpoint(binding.Identity(), policy, "raw://"+request.Endpoint), nil
67 })
68
69 connector, err := NewEndpointConnector(server.dial(), DefaultMuxCeilings())
70 require.NoError(t, err)
71 pool, err := broker.NewPool(1, resolver, p3cBinder{}, connector, newListenerClock(time.Now()), p3cPoolLimits())
72 require.NoError(t, err)
73 t.Cleanup(func() { _ = pool.Close() })
74
75 client, err := pool.RegisterClient()
76 require.NoError(t, err)
77
78 first, err := openTyped(pool, context.Background(), p3cRemoteRequest(client, "aliasA@host:22", 1, policy))
79 require.NoError(t, err)
80 require.NotNil(t, first)
81 t.Cleanup(func() { _ = first.Close() })
82
83 second, err := openTyped(pool, context.Background(), p3cRemoteRequest(client, "aliasB@host:22", 2, policy))
84 require.NoError(t, err)
85 require.NotNil(t, second)
86 t.Cleanup(func() { _ = second.Close() })
87
88 require.Equal(t, int64(1), server.dials.Load(), "compatible aliases must share one physical carriage")
89 server.awaitAccepted(t, 2)
90
91 for _, stream := range []ports.BrokerLogicalConnection{first, second} {
92 require.NoError(t, stream.SendClient(protocol.Ping{}))
93 message, err := stream.ReceiveServer()
94 require.NoError(t, err)
95 require.Equal(t, protocol.Pong{}, message)
96 }
97}
98
99// TestPoolRefusesConflictingPolicyWithoutDial proves the pool rejects a request
100// whose policy conflicts with the resolved endpoint policy before any carriage
101// is dialed.
102func TestPoolRefusesConflictingPolicyWithoutDial(t *testing.T) {
103 binding := mustServerBinding(t)
104 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
105 resolver := p3cResolver(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
106 return muxEndpoint(binding.Identity(), binding.Policy(), "raw://conflict"), nil
107 })
108
109 connector, err := NewEndpointConnector(server.dial(), DefaultMuxCeilings())
110 require.NoError(t, err)
111 pool, err := broker.NewPool(1, resolver, p3cBinder{}, connector, newListenerClock(time.Now()), p3cPoolLimits())
112 require.NoError(t, err)
113 t.Cleanup(func() { _ = pool.Close() })
114
115 client, err := pool.RegisterClient()
116 require.NoError(t, err)
117 _, err = openTyped(pool, context.Background(), p3cRemoteRequest(client, "aliasA@host:22", 1, alternatePolicy(binding.Policy())))
118 require.Error(t, err)
119 var typed ports.BrokerError
120 require.ErrorAs(t, err, &typed)
121 require.Equal(t, ports.BrokerErrorConflictingPolicy, typed.Code)
122 require.Zero(t, server.dials.Load(), "a conflicting policy never dials a carriage")
123}
124
125// TestPoolSurfacesPhysicalLossPerStream proves a carriage loss is published
126// per affected logical stream through the pool: the pooled stream terminates
127// with a broker stream-lost error once the accepted physical carriage dies.
128func TestPoolSurfacesPhysicalLossPerStream(t *testing.T) {
129 binding := mustRemoteServerBinding(t)
130 policy := binding.Policy()
131 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
132 server.serve()
133 resolver := p3cResolver(func(_ context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
134 return muxEndpoint(binding.Identity(), policy, "raw://"+request.Endpoint), nil
135 })
136
137 connector, err := NewEndpointConnector(server.dial(), DefaultMuxCeilings())
138 require.NoError(t, err)
139 pool, err := broker.NewPool(1, resolver, p3cBinder{}, connector, newListenerClock(time.Now()), p3cPoolLimits())
140 require.NoError(t, err)
141 t.Cleanup(func() { _ = pool.Close() })
142
143 client, err := pool.RegisterClient()
144 require.NoError(t, err)
145 stream, err := openTyped(pool, context.Background(), p3cRemoteRequest(client, "aliasA@host:22", 1, policy))
146 require.NoError(t, err)
147 server.awaitAccepted(t, 1)
148
149 // Lose the daemon carriage: the supervised child is closed, so the broker
150 // pump publishes the physical loss and the pool settles the stream.
151 server.supervisor.mu.Lock()
152 children := make([]*supervisedPhysical, 0, len(server.supervisor.children))
153 for child := range server.supervisor.children {
154 children = append(children, child)
155 }
156 server.supervisor.mu.Unlock()
157 require.Len(t, children, 1)
158 require.NoError(t, children[0].Close())
159
160 select {
161 case <-stream.Done():
162 case <-time.After(p3cTestDeadline):
163 t.Fatal("the pooled logical stream never observed the physical loss")
164 }
165 var lost ports.BrokerStreamLost
166 require.ErrorAs(t, stream.Err(), &lost)
167 require.Equal(t, ports.BrokerStreamID(1), lost.Stream)
168 require.NotEqual(t, domain.RemoteFailureNone, lost.Cause)
169}