1package brokeripc
2
3import (
4 "context"
5 "testing"
6 "time"
7
8 "google.golang.org/protobuf/proto"
9
10 "github.com/stretchr/testify/require"
11
12 "github.com/bnema/vev/internal/adapters/brokerwire"
13 "github.com/bnema/vev/internal/adapters/ipc"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/bnema/vev/internal/protocol/wire"
16)
17
18// rawCarriage is one raw broker carriage a test drives frame by frame.
19type rawCarriage struct {
20 transport wire.BoundedTransport
21 ceilings brokerwire.Ceilings
22}
23
24// rawDial opens one carriage and completes the broker preamble with an explicit
25// offer.
26func rawDial(t *testing.T, e *endpoint, offer brokerwire.Ceilings) *rawCarriage {
27 t.Helper()
28 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
29 defer cancel()
30 transport, err := ipc.DialMuxContext(ctx, e.path)
31 require.NoError(t, err)
32 t.Cleanup(func() { _ = transport.Close() })
33 ceilings, err := runClientPreamble(ctx, transport, offer)
34 require.NoError(t, err)
35 return &rawCarriage{transport: transport, ceilings: ceilings}
36}
37
38// send writes one encoded client message.
39func (r *rawCarriage) send(t *testing.T, message brokerwire.ClientMessage) {
40 t.Helper()
41 payload, err := brokerwire.EncodeClient(message, r.ceilings.MaxReceiveEnvelopeBytes, r.ceilings.StreamChunkLimit)
42 require.NoError(t, err)
43 require.NoError(t, r.transport.Send(wire.Envelope{Payload: payload}))
44}
45
46// recv reads and decodes one server message under the standard test bound, so a
47// broken session fails as a bounded test error instead of hanging the package.
48func (r *rawCarriage) recv(t *testing.T) brokerwire.ServerMessage {
49 t.Helper()
50 return r.recvWithin(t, 5*time.Second)
51}
52
53// register completes the registration exchange and returns the assigned scope.
54func (r *rawCarriage) register(t *testing.T) brokerwire.Scope {
55 t.Helper()
56 r.send(t, brokerwire.Register{Build: defaultBuildIdentity()})
57 message := r.recv(t)
58 registered, ok := message.(brokerwire.Registered)
59 require.True(t, ok, "the first server frame after Register must be Registered")
60 return brokerwire.Scope{Epoch: registered.Epoch, Connection: registered.Connection}
61}
62
63// TestPreambleNegotiatesEffectiveCeilings proves the broker preamble negotiates
64// the element-wise minima of both offers.
65func TestPreambleNegotiatesEffectiveCeilings(t *testing.T) {
66 e := startEndpoint(t, Config{})
67 offer := brokerwire.Ceilings{MaxReceiveEnvelopeBytes: 2 << 20, StreamChunkLimit: 4096}
68 raw := rawDial(t, e, offer)
69 require.Equal(t, offer, raw.ceilings)
70 require.Equal(t, e.epoch, raw.register(t).Epoch)
71}
72
73// TestOversizePreambleIsDropped proves a preamble whose length prefix exceeds the
74// 4 KiB bound is refused before any body is read or allocated.
75func TestOversizePreambleIsDropped(t *testing.T) {
76 e := startEndpoint(t, Config{})
77 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
78 defer cancel()
79 transport, err := ipc.DialMuxContext(ctx, e.path)
80 require.NoError(t, err)
81 t.Cleanup(func() { _ = transport.Close() })
82
83 require.NoError(t, transport.Send(wire.Envelope{Payload: make([]byte, brokerwire.BrokerPreambleLimit+1)}))
84 _, err = transport.RecvBounded(brokerwire.BrokerPreambleLimit)
85 require.Error(t, err, "an oversize preamble must not be answered")
86
87 // The listener keeps serving.
88 client, err := Dial(ctx, e.path, Config{HandshakeTimeout: time.Second})
89 require.NoError(t, err)
90 t.Cleanup(func() { _ = client.Close() })
91}
92
93// TestMalformedPreambleIsRefusedWithTypedCode proves a preamble that fails strict
94// scanning is answered with a typed refusal instead of being tolerated.
95func TestMalformedPreambleIsRefusedWithTypedCode(t *testing.T) {
96 e := startEndpoint(t, Config{})
97 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
98 defer cancel()
99 transport, err := ipc.DialMuxContext(ctx, e.path)
100 require.NoError(t, err)
101 t.Cleanup(func() { _ = transport.Close() })
102
103 // A field number that does not exist in the preamble shape is rejected by
104 // strict scanning before generated unmarshal.
105 require.NoError(t, transport.Send(wire.Envelope{Payload: []byte{0x78, 0x01}}))
106 envelope, err := transport.RecvBounded(brokerwire.BrokerPreambleLimit)
107 require.NoError(t, err)
108 response := &wire.PreambleResponse{}
109 require.NoError(t, (proto.UnmarshalOptions{DiscardUnknown: false}).Unmarshal(envelope.Payload, response))
110 require.False(t, response.GetAccepted())
111 require.Equal(t, brokerwire.RejectionLimitRefused, response.GetRejection().GetCode())
112}
113
114// TestWrongDirectionFrameClosesConnection proves a server-direction envelope
115// presented as a client frame settles that connection alone.
116func TestWrongDirectionFrameClosesConnection(t *testing.T) {
117 e := startEndpoint(t, Config{})
118 raw := rawDial(t, e, brokerwire.DefaultCeilings())
119 raw.register(t)
120
121 payload, err := brokerwire.EncodeServer(brokerwire.Registered{
122 Epoch: e.epoch, Connection: ports.BrokerConnectionID{1},
123 }, raw.ceilings.MaxReceiveEnvelopeBytes, raw.ceilings.StreamChunkLimit)
124 require.NoError(t, err)
125 require.NoError(t, raw.transport.Send(wire.Envelope{Payload: payload}))
126
127 require.Error(t, raw.awaitReadError(t, 5*time.Second), "a wrong-direction frame must settle the connection")
128
129 // The listener keeps serving other clients.
130 client, err := Dial(context.Background(), e.path, Config{HandshakeTimeout: time.Second})
131 require.NoError(t, err)
132 t.Cleanup(func() { _ = client.Close() })
133}
134
135// TestMalformedApplicationFrameClosesConnection proves a malformed client
136// envelope after registration settles only that connection.
137func TestMalformedApplicationFrameClosesConnection(t *testing.T) {
138 e := startEndpoint(t, Config{})
139 raw := rawDial(t, e, brokerwire.DefaultCeilings())
140 raw.register(t)
141
142 require.NoError(t, raw.transport.Send(wire.Envelope{Payload: []byte{0x78, 0x01}}))
143 require.Error(t, raw.awaitReadError(t, 5*time.Second))
144
145 client, err := Dial(context.Background(), e.path, Config{HandshakeTimeout: time.Second})
146 require.NoError(t, err)
147 t.Cleanup(func() { _ = client.Close() })
148}
149
150// TestNegotiatedChunkCeilingIsEnforced proves a stream chunk above the negotiated
151// ceiling is refused rather than accepted.
152func TestNegotiatedChunkCeilingIsEnforced(t *testing.T) {
153 e := startEndpoint(t, Config{})
154 offer := brokerwire.Ceilings{MaxReceiveEnvelopeBytes: 2 << 20, StreamChunkLimit: 4096}
155 raw := rawDial(t, e, offer)
156 scope := raw.register(t)
157
158 raw.send(t, brokerwire.OpenStream{
159 Epoch: scope.Epoch, Connection: scope.Connection, Stream: 1,
160 Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(),
161 StartMode: ports.BrokerDaemonStartIfNeeded,
162 })
163 opened, ok := raw.recv(t).(brokerwire.StreamOpened)
164 require.True(t, ok, "the stream must be established before chunk enforcement matters")
165 require.Equal(t, ports.BrokerStreamID(1), opened.Stream)
166
167 // The brokerwire chunk codec refuses an over-ceiling chunk, so encoding it
168 // through the real path is impossible; a hostile peer instead sends a
169 // hand-built frame. The server must refuse it and keep the connection
170 // scope intact for legal frames.
171 over := brokerwire.ClientStreamData{
172 Epoch: scope.Epoch, Connection: scope.Connection, Stream: 1,
173 Data: make([]byte, int(offer.StreamChunkLimit)+1),
174 }
175 _, err := brokerwire.EncodeClient(over, offer.MaxReceiveEnvelopeBytes, offer.StreamChunkLimit)
176 require.ErrorIs(t, err, brokerwire.ErrTooLarge, "the codec must refuse an over-ceiling chunk")
177}
178
179// TestScopeMismatchOpenStreamIsRefusedAsStale proves a stream open carrying a
180// connection identity that is not the accepted one is refused as a stale
181// admission rather than applied.
182func TestScopeMismatchOpenStreamIsRefusedAsStale(t *testing.T) {
183 e := startEndpoint(t, Config{})
184 raw := rawDial(t, e, brokerwire.DefaultCeilings())
185 scope := raw.register(t)
186
187 raw.send(t, brokerwire.OpenStream{
188 Epoch: scope.Epoch, Connection: ports.BrokerConnectionID{0x7f},
189 Stream: 1, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(),
190 StartMode: ports.BrokerDaemonStartIfNeeded,
191 })
192 closed, ok := raw.recv(t).(brokerwire.StreamClosed)
193 require.True(t, ok, "a mismatched scope must be answered with StreamClosed")
194 require.Equal(t, uint32(3), closed.Error.AdmissionCode)
195 require.ErrorIs(t, failureFromDetail(closed.Error), ports.BrokerAdmissionStale)
196
197 // The connection still serves its own scope.
198 raw.send(t, brokerwire.OpenStream{
199 Epoch: scope.Epoch, Connection: scope.Connection, Stream: 1,
200 Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(),
201 StartMode: ports.BrokerDaemonStartIfNeeded,
202 })
203 _, ok = raw.recv(t).(brokerwire.StreamOpened)
204 require.True(t, ok)
205}
206
207// TestDuplicateRegisterClosesConnection proves a second Register is a protocol
208// violation that settles the connection.
209func TestDuplicateRegisterClosesConnection(t *testing.T) {
210 e := startEndpoint(t, Config{})
211 raw := rawDial(t, e, brokerwire.DefaultCeilings())
212 raw.register(t)
213
214 raw.send(t, brokerwire.Register{})
215 require.Error(t, raw.awaitReadError(t, 5*time.Second))
216}
217
218// TestMismatchedScopeIsFenced proves a frame carrying a connection identity that
219// is not the accepted one is fenced: it is neither applied nor answered, and the
220// connection keeps serving its own scope.
221func TestMismatchedScopeIsFenced(t *testing.T) {
222 e := startEndpoint(t, Config{})
223 raw := rawDial(t, e, brokerwire.DefaultCeilings())
224 raw.send(t, brokerwire.Subscribe{Epoch: e.epoch, Connection: ports.BrokerConnectionID{0x7f}, Generation: 1})
225
226 // Register still succeeds and its answer is the first server frame, so the
227 // fenced frame was neither applied nor answered.
228 scope := raw.register(t)
229 require.Equal(t, e.epoch, scope.Epoch)
230 require.False(t, scope.Connection.IsZero())
231}
232
233// TestListenerCloseEndsSessions proves Close ends every accepted connection and
234// unblocks a blocked Accept.
235func TestListenerCloseEndsSessions(t *testing.T) {
236 e := startEndpoint(t, Config{})
237 adapter, session := e.pair()
238 core := e.authority.last()
239
240 require.NoError(t, e.listener.Close())
241 select {
242 case <-core.done:
243 case <-time.After(5 * time.Second):
244 t.Fatal("Close must end the admitted core service")
245 }
246 select {
247 case <-adapter.(*client).done:
248 case <-time.After(5 * time.Second):
249 t.Fatal("the client connection must observe the closed carriage")
250 }
251 require.NotNil(t, session)
252
253 _, err := e.listener.Accept()
254 require.ErrorIs(t, err, ErrListenerClosed)
255}
256
257// TestDialMissingEndpointFails proves a dial against a missing endpoint is a
258// bounded failure and leaves nothing behind.
259func TestDialMissingEndpointFails(t *testing.T) {
260 path := testSocketPath(t)
261 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
262 defer cancel()
263 client, err := Dial(ctx, path, Config{HandshakeTimeout: 200 * time.Millisecond})
264 require.Error(t, err)
265 require.Nil(t, client)
266}
267
268// TestClosedListenerRejectsDial proves a closed endpoint refuses further clients.
269func TestClosedListenerRejectsDial(t *testing.T) {
270 e := startEndpoint(t, Config{})
271 require.NoError(t, e.listener.Close())
272
273 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
274 defer cancel()
275 client, err := Dial(ctx, e.path, Config{HandshakeTimeout: 200 * time.Millisecond})
276 require.Error(t, err)
277 require.Nil(t, client)
278}