1package broker
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 portsmocks "github.com/bnema/vev/internal/ports/mocks"
13 "github.com/bnema/vev/internal/protocol"
14 "github.com/stretchr/testify/mock"
15 "github.com/stretchr/testify/require"
16)
17
18// TestBrokerRemoteProbeBorrowsWarmTransport proves observation reuses the
19// broker pool's attached or warm transport instead of bootstrapping SSH/QUIC
20// again, and redials only when the borrowed transport has retired.
21func TestBrokerRemoteProbeBorrowsWarmTransport(t *testing.T) {
22 registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
23 policy := observationPolicy()
24 const identity ports.BrokerDaemonIdentity = "remote-daemon"
25 dialErr := errors.New("dial refused")
26 tests := []struct {
27 name string
28 bound bool
29 shared bool
30 openErr error
31 retired bool
32 wantDials int
33 wantErr error
34 }{
35 {name: "warm transport observes without dialing", bound: true, shared: true},
36 {name: "unbound identity dials", bound: false, shared: true, wantDials: 1, wantErr: dialErr},
37 {name: "no pooled transport dials", bound: true, wantDials: 1, wantErr: dialErr},
38 {name: "live transport refusal is reported without redial", bound: true, shared: true, openErr: errors.New("stream refused")},
39 {name: "retired transport redials", bound: true, shared: true, openErr: errors.New("physical closed"), retired: true, wantDials: 1, wantErr: dialErr},
40 }
41 for _, tt := range tests {
42 t.Run(tt.name, func(t *testing.T) {
43 record := ports.BrokerHostRecord{Registration: registration, Policy: policy}
44 if tt.bound {
45 record.Identity = identity
46 }
47 routes := portsmocks.NewMockBrokerRouteAuthority(t)
48 routes.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).Return(ports.BrokerDialTarget{
49 Fence: ports.BrokerEndpointFence{Registration: registration}, Policy: policy, Address: "opaque", StartMode: ports.BrokerDaemonExistingOnly,
50 ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: record.Identity, Bound: tt.bound},
51 }, nil)
52 connector := portsmocks.NewMockBrokerEndpointConnector(t)
53 if tt.wantDials > 0 {
54 connector.EXPECT().Connect(mock.Anything, mock.Anything).Return(nil, dialErr).Times(tt.wantDials)
55 }
56 physical := portsmocks.NewMockBrokerPhysicalConnection(t)
57 codec := portsmocks.NewMockSessionCodec(t)
58 done := make(chan struct{})
59 if tt.retired {
60 close(done)
61 }
62 if tt.bound && tt.shared {
63 if tt.openErr != nil {
64 if !tt.retired {
65 physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(nil, tt.openErr)
66 physical.EXPECT().Done().Return(done)
67 }
68 } else {
69 logical := portsmocks.NewMockBrokerEnvelopeStream(t)
70 typed := portsmocks.NewMockClientConnection(t)
71 codec.EXPECT().Client(logical).Return(typed)
72 var sent protocol.CommandRequest
73 typed.EXPECT().SendClient(mock.Anything).RunAndReturn(func(message protocol.ClientMessage) error {
74 sent = message.(protocol.CommandRequest)
75 return nil
76 })
77 typed.EXPECT().ReceiveServer().RunAndReturn(func() (protocol.ServerMessage, error) {
78 return protocol.CommandResult{RequestID: sent.RequestID, Outcome: protocol.CommandSucceeded, Output: observationCatalogue(t)}, nil
79 })
80 logical.EXPECT().Close().Return(nil)
81 physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(logical, nil)
82 physical.EXPECT().Identity().Return(identity)
83 physical.EXPECT().Incarnation().Return(ports.BrokerDaemonIncarnation{1})
84 }
85 }
86 probe := &RemoteProbe{
87 Epoch: 7, Routes: routes, Binder: portsmocks.NewMockBrokerIdentityBinder(t), Connector: connector,
88 Hosts: observationHosts(t, record), Codec: codec,
89 }
90 // Use the real pool for borrowing. Stateful pool retirement is
91 // covered with a fake physical in broker's worker tests.
92 pool, err := NewPool(7, routes, probe.Binder, connector, newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 1, Streams: 2, StreamsPerClient: 2, Warm: 1})
93 require.NoError(t, err)
94 t.Cleanup(func() { require.NoError(t, pool.Close()) })
95 if tt.shared && tt.bound && !tt.retired {
96 physical.EXPECT().Policy().Return(policy).Maybe()
97 physical.EXPECT().Identity().Return(identity).Maybe()
98 physical.EXPECT().Incarnation().Return(ports.BrokerDaemonIncarnation{1}).Maybe()
99 physical.EXPECT().Done().Return(done).Maybe()
100 physical.EXPECT().Close().Return(nil).Maybe()
101 require.True(t, pool.AdoptPhysical(physical, policy))
102 }
103 probe.SharePool(pool)
104
105 observation, err := probe.Probe(context.Background(), registration)
106 switch {
107 case tt.wantErr != nil:
108 require.ErrorIs(t, err, tt.wantErr)
109 case tt.openErr != nil:
110 require.ErrorIs(t, err, tt.openErr)
111 default:
112 require.NoError(t, err)
113 require.Equal(t, identity, observation.Identity)
114 require.True(t, observation.InventoryKnown)
115 }
116 })
117 }
118}
119
120// TestBrokerRemoteProbeSilentBorrowedTransportNeverRedials proves a probe on a
121// healthy borrowed transport whose daemon never answers is bounded by the
122// attempt context (the registry's ports.Clock deadline cancels it) and returns
123// without bootstrapping a second transport: a live pooled physical is never
124// abandoned for a redial just because one observation went unanswered.
125func TestBrokerRemoteProbeSilentBorrowedTransportNeverRedials(t *testing.T) {
126 registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
127 policy := observationPolicy()
128 const identity ports.BrokerDaemonIdentity = "remote-daemon"
129 tests := []struct {
130 name string
131 // silentOpen: the daemon never answers the observation open; otherwise
132 // the stream opens and the catalogue command is never answered.
133 silentOpen bool
134 }{
135 {name: "unanswered observation open", silentOpen: true},
136 {name: "unanswered catalogue command"},
137 }
138 for _, tt := range tests {
139 t.Run(tt.name, func(t *testing.T) {
140 record := ports.BrokerHostRecord{Registration: registration, Policy: policy, Identity: identity}
141 routes := portsmocks.NewMockBrokerRouteAuthority(t)
142 routes.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).Return(ports.BrokerDialTarget{
143 Fence: ports.BrokerEndpointFence{Registration: registration}, Policy: policy, Address: "opaque", StartMode: ports.BrokerDaemonExistingOnly,
144 ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: identity, Bound: true},
145 }, nil)
146 // No Connect expectation: any redial fails the mock.
147 connector := portsmocks.NewMockBrokerEndpointConnector(t)
148 physical := portsmocks.NewMockBrokerPhysicalConnection(t)
149 codec := portsmocks.NewMockSessionCodec(t)
150 healthy := make(chan struct{})
151 physical.EXPECT().Policy().Return(policy).Maybe()
152 physical.EXPECT().Identity().Return(identity).Maybe()
153 physical.EXPECT().Incarnation().Return(ports.BrokerDaemonIncarnation{1}).Maybe()
154 physical.EXPECT().Done().Return(healthy).Maybe()
155 physical.EXPECT().Close().Return(nil).Maybe()
156
157 ctx, cancel := context.WithCancel(context.Background())
158 defer cancel()
159 // waiting signals that the probe is parked on the silent daemon; the
160 // worker only signals, the test goroutine cancels and asserts.
161 waiting := make(chan struct{}, 1)
162 if tt.silentOpen {
163 physical.EXPECT().OpenStream(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
164 waiting <- struct{}{}
165 <-ctx.Done()
166 return nil, ctx.Err()
167 })
168 } else {
169 logical := portsmocks.NewMockBrokerEnvelopeStream(t)
170 typed := portsmocks.NewMockClientConnection(t)
171 codec.EXPECT().Client(logical).Return(typed)
172 closed := make(chan struct{})
173 var once sync.Once
174 typed.EXPECT().SendClient(mock.Anything).Return(nil)
175 typed.EXPECT().ReceiveServer().RunAndReturn(func() (protocol.ServerMessage, error) {
176 waiting <- struct{}{}
177 <-closed
178 return nil, errors.New("stream closed")
179 })
180 logical.EXPECT().Close().RunAndReturn(func() error { once.Do(func() { close(closed) }); return nil })
181 physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(logical, nil)
182 }
183 probe := &RemoteProbe{
184 Epoch: 7, Routes: routes, Binder: portsmocks.NewMockBrokerIdentityBinder(t), Connector: connector,
185 Hosts: observationHosts(t, record), Codec: codec,
186 }
187 pool, err := NewPool(7, routes, probe.Binder, connector, newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 1, Streams: 2, StreamsPerClient: 2, Warm: 1})
188 require.NoError(t, err)
189 t.Cleanup(func() { require.NoError(t, pool.Close()) })
190 require.True(t, pool.AdoptPhysical(physical, policy))
191 probe.SharePool(pool)
192
193 type outcome struct{ err error }
194 done := make(chan outcome, 1)
195 go func() {
196 _, err := probe.Probe(ctx, registration)
197 done <- outcome{err: err}
198 }()
199 select {
200 case <-waiting:
201 case <-time.After(2 * time.Second):
202 t.Fatal("probe never reached the silent daemon")
203 }
204 cancel()
205 select {
206 case result := <-done:
207 require.ErrorIs(t, result.err, context.Canceled)
208 case <-time.After(2 * time.Second):
209 t.Fatal("probe outlived its attempt context")
210 }
211 })
212 }
213}