1package brokerwire
2
3// Membership wire contract (Phase C1).
4//
5// AddHost carries a required policy, RemoveHost carries exact registration
6// authority instead of an endpoint name, UpdateHostPolicy carries exact
7// authority plus the replacement policy, and every result carries the
8// presence-carrying outcome of the request it completes. These tests pin the
9// stateless codec's half of that contract: byte round trips, direction
10// closure, strict scan before unmarshal, exact generation preservation, the
11// request/result presence rules, and the typed error codes 11/12.
12
13import (
14 "math"
15 "testing"
16
17 "google.golang.org/protobuf/encoding/protowire"
18 "google.golang.org/protobuf/proto"
19
20 "github.com/bnema/vev/internal/domain"
21 "github.com/bnema/vev/internal/ports"
22 "github.com/bnema/vev/internal/protocol/wire"
23 "github.com/stretchr/testify/require"
24)
25
26// membershipRequests are the three mutating client requests with their frozen
27// envelope tags.
28func membershipRequests(connection ports.BrokerConnectionID, operation ports.BrokerOperationID) []struct {
29 name string
30 message ClientMessage
31 tag protowire.Number
32} {
33 return []struct {
34 name string
35 message ClientMessage
36 tag protowire.Number
37 }{
38 {"add_host", AddHost{Epoch: 7, Connection: connection, Operation: operation, Endpoint: "dev@host:22", Policy: testPolicy()}, 105},
39 {"remove_host", RemoveHost{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration()}, 106},
40 {"update_host_policy", UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration(), Policy: testPolicy()}, 111},
41 }
42}
43
44// TestMembershipRequestByteRoundTrip proves each mutating request travels on
45// its frozen tag and survives an encode/decode/re-encode cycle byte for byte,
46// including the generation that fences a removal and the full policy.
47func TestMembershipRequestByteRoundTrip(t *testing.T) {
48 connection := testConnectionID(0x21)
49 operation := testOperationID(0x31)
50 for _, tc := range membershipRequests(connection, operation) {
51 t.Run(tc.name, func(t *testing.T) {
52 raw := mustEncodeClient(t, tc.message)
53 number, wireType, consumed := protowire.ConsumeTag(raw)
54 require.Greater(t, consumed, 0)
55 require.Equal(t, tc.tag, number)
56 require.Equal(t, protowire.BytesType, wireType)
57
58 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
59 require.NoError(t, err)
60 require.Equal(t, tc.message, decoded)
61
62 again, err := EncodeClient(decoded, testEnvelopeCeiling, testChunkCeiling)
63 require.NoError(t, err)
64 require.Equal(t, raw, again, "re-encode must be byte-for-byte identical")
65 })
66 }
67}
68
69// TestMembershipResultByteRoundTrip proves every legal result shape round-trips
70// byte for byte: an authoritative success carrying its registration, an
71// idempotent removal success carrying none, and a failed result carrying only
72// its typed error.
73func TestMembershipResultByteRoundTrip(t *testing.T) {
74 connection := testConnectionID(0x22)
75 operation := testOperationID(0x32)
76 messages := []struct {
77 name string
78 message OperationResult
79 }{
80 {"add success carries registration", OperationResult{
81 Epoch: 7, Connection: connection, Operation: operation,
82 Outcome: ports.BrokerOutcomeOK, Registration: testRegistration(),
83 }},
84 {"update success carries advanced registration", OperationResult{
85 Epoch: 7, Connection: connection, Operation: operation,
86 Outcome: ports.BrokerOutcomeOK, Registration: domain.RemoteRegistration{
87 Endpoint: "dev@host:22", Incarnation: testRegistration().Incarnation, Generation: 10,
88 },
89 }},
90 {"remove success reports removal only", OperationResult{
91 Epoch: 7, Connection: connection, Operation: operation,
92 Outcome: ports.BrokerOutcomeOK, Removed: true,
93 }},
94 {"idempotent remove reports not removed", OperationResult{
95 Epoch: 7, Connection: connection, Operation: operation,
96 Outcome: ports.BrokerOutcomeOK,
97 }},
98 {"failed carries only its error", OperationResult{
99 Epoch: 7, Connection: connection, Operation: operation,
100 Outcome: ports.BrokerOutcomeFailed, Error: testErrorDetail(), HasError: true,
101 }},
102 {"unknown carries only its error", OperationResult{
103 Epoch: 7, Connection: connection, Operation: operation,
104 Outcome: ports.BrokerOutcomeUnknown,
105 Error: ErrorDetail{Code: ports.BrokerErrorOutcomeUnknown}, HasError: true,
106 }},
107 }
108 for _, tc := range messages {
109 t.Run(tc.name, func(t *testing.T) {
110 raw := mustEncodeServer(t, tc.message)
111 number, wireType, consumed := protowire.ConsumeTag(raw)
112 require.Greater(t, consumed, 0)
113 require.Equal(t, protowire.Number(203), number)
114 require.Equal(t, protowire.BytesType, wireType)
115
116 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
117 require.NoError(t, err)
118 require.Equal(t, tc.message, decoded)
119
120 again, err := EncodeServer(decoded, testEnvelopeCeiling, testChunkCeiling)
121 require.NoError(t, err)
122 require.Equal(t, raw, again, "re-encode must be byte-for-byte identical")
123 })
124 }
125}
126
127// TestMembershipExactGenerationPreserved proves the generation that fences a
128// remove/re-add race survives the wire exactly, including its extremes: a
129// generation is never widened, truncated, or re-derived from time.
130func TestMembershipExactGenerationPreserved(t *testing.T) {
131 connection := testConnectionID(0x23)
132 operation := testOperationID(0x33)
133 base := testRegistration()
134 for name, generation := range map[string]domain.RemoteGeneration{
135 "one": 1,
136 "large": 1 << 40,
137 "max uint64": math.MaxUint64,
138 } {
139 t.Run(name, func(t *testing.T) {
140 registration := base
141 registration.Generation = generation
142
143 request := RemoveHost{Epoch: 7, Connection: connection, Operation: operation, Registration: registration}
144 decoded, err := DecodeClient(mustEncodeClient(t, request), testEnvelopeCeiling, testChunkCeiling)
145 require.NoError(t, err)
146 require.Equal(t, request, decoded)
147 require.Equal(t, generation, decoded.(RemoveHost).Registration.Generation)
148
149 update := UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: operation, Registration: registration, Policy: testPolicy()}
150 decodedUpdate, err := DecodeClient(mustEncodeClient(t, update), testEnvelopeCeiling, testChunkCeiling)
151 require.NoError(t, err)
152 require.Equal(t, update, decodedUpdate)
153 require.Equal(t, generation, decodedUpdate.(UpdateHostPolicy).Registration.Generation)
154
155 result := OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Registration: registration}
156 decodedResult, err := DecodeServer(mustEncodeServer(t, result), testEnvelopeCeiling, testChunkCeiling)
157 require.NoError(t, err)
158 require.Equal(t, result, decodedResult)
159 require.Equal(t, generation, decodedResult.(OperationResult).Registration.Generation)
160 })
161 }
162}
163
164// TestMembershipWrongDirection proves direction stays structural: a mutating
165// request never decodes as server output and a result never decodes as a
166// request. Misuse is refused by the strict scanner (the tag is unknown to the
167// other envelope) before generated unmarshal; the typed direction refusal is
168// already covered by the closed directional interfaces and
169// TestBrokerWrongDirection.
170func TestMembershipWrongDirection(t *testing.T) {
171 connection := testConnectionID(0x24)
172 operation := testOperationID(0x34)
173 for _, tc := range membershipRequests(connection, operation) {
174 t.Run("request/"+tc.name, func(t *testing.T) {
175 raw := mustEncodeClient(t, tc.message)
176 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
177 require.Error(t, err)
178 require.Nil(t, decoded, "no server message may be decoded from a client frame")
179 })
180 }
181 result := OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()}
182 raw := mustEncodeServer(t, result)
183 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
184 require.Error(t, err)
185 require.Nil(t, decoded, "no client message may be decoded from a server frame")
186}
187
188// TestMembershipMissingAuthorityRefused proves an absent policy or an absent
189// registration is not authority: the decode path refuses both rather than
190// inventing a default, on encode and on receive.
191func TestMembershipMissingAuthorityRefused(t *testing.T) {
192 connection := testConnectionID(0x25)
193 operation := testOperationID(0x35)
194
195 clearFields := func(t *testing.T, message ClientMessage, clear func(*wire.BrokerClientEnvelope)) []byte {
196 t.Helper()
197 raw := mustEncodeClient(t, message)
198 envelope := &wire.BrokerClientEnvelope{}
199 require.NoError(t, wire.ScanEnvelope(envelope, raw))
200 require.NoError(t, proto.Unmarshal(raw, envelope))
201 clear(envelope)
202 mutated, err := proto.Marshal(envelope)
203 require.NoError(t, err)
204 return mutated
205 }
206
207 t.Run("add_host without policy", func(t *testing.T) {
208 _, err := EncodeClient(AddHost{Epoch: 7, Connection: connection, Operation: operation, Endpoint: "dev@host:22"}, testEnvelopeCeiling, testChunkCeiling)
209 require.ErrorIs(t, err, ErrInvalidMessage)
210 mutated := clearFields(t, AddHost{Epoch: 7, Connection: connection, Operation: operation, Endpoint: "dev@host:22", Policy: testPolicy()}, func(envelope *wire.BrokerClientEnvelope) {
211 envelope.GetAddHost().Policy = nil
212 })
213 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
214 require.ErrorIs(t, err, ErrInvalidMessage)
215 })
216
217 t.Run("add_host with invalid policy", func(t *testing.T) {
218 _, err := EncodeClient(AddHost{Epoch: 7, Connection: connection, Operation: operation, Endpoint: "dev@host:22", Policy: ports.BrokerPolicy{}}, testEnvelopeCeiling, testChunkCeiling)
219 require.ErrorIs(t, err, ErrInvalidMessage)
220 mutated := clearFields(t, AddHost{Epoch: 7, Connection: connection, Operation: operation, Endpoint: "dev@host:22", Policy: testPolicy()}, func(envelope *wire.BrokerClientEnvelope) {
221 envelope.GetAddHost().GetPolicy().EnvironmentPolicy = 99
222 })
223 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
224 require.ErrorIs(t, err, ErrInvalidMessage)
225 })
226
227 t.Run("remove_host without registration", func(t *testing.T) {
228 _, err := EncodeClient(RemoveHost{Epoch: 7, Connection: connection, Operation: operation}, testEnvelopeCeiling, testChunkCeiling)
229 require.ErrorIs(t, err, ErrInvalidMessage)
230 mutated := clearFields(t, RemoveHost{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration()}, func(envelope *wire.BrokerClientEnvelope) {
231 envelope.GetRemoveHost().Registration = nil
232 })
233 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
234 require.ErrorIs(t, err, ErrInvalidMessage)
235 })
236
237 t.Run("remove_host with partial registration", func(t *testing.T) {
238 partial := testRegistration()
239 partial.Incarnation = [16]byte{}
240 _, err := EncodeClient(RemoveHost{Epoch: 7, Connection: connection, Operation: operation, Registration: partial}, testEnvelopeCeiling, testChunkCeiling)
241 require.ErrorIs(t, err, ErrInvalidMessage)
242
243 shortIncarnation := clearFields(t, RemoveHost{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration()}, func(envelope *wire.BrokerClientEnvelope) {
244 envelope.GetRemoveHost().GetRegistration().Incarnation = []byte{1, 2, 3}
245 })
246 _, err = DecodeClient(shortIncarnation, testEnvelopeCeiling, testChunkCeiling)
247 require.ErrorIs(t, err, ErrInvalidMessage)
248
249 zeroGeneration := clearFields(t, RemoveHost{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration()}, func(envelope *wire.BrokerClientEnvelope) {
250 envelope.GetRemoveHost().GetRegistration().Generation = 0
251 })
252 _, err = DecodeClient(zeroGeneration, testEnvelopeCeiling, testChunkCeiling)
253 require.ErrorIs(t, err, ErrInvalidMessage)
254
255 badEndpoint := clearFields(t, RemoveHost{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration()}, func(envelope *wire.BrokerClientEnvelope) {
256 envelope.GetRemoveHost().GetRegistration().Endpoint = "two words"
257 })
258 _, err = DecodeClient(badEndpoint, testEnvelopeCeiling, testChunkCeiling)
259 require.ErrorIs(t, err, ErrInvalidMessage)
260 })
261
262 t.Run("update_host_policy missing registration or policy", func(t *testing.T) {
263 _, err := EncodeClient(UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: operation, Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
264 require.ErrorIs(t, err, ErrInvalidMessage)
265 _, err = EncodeClient(UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration()}, testEnvelopeCeiling, testChunkCeiling)
266 require.ErrorIs(t, err, ErrInvalidMessage)
267
268 update := UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration(), Policy: testPolicy()}
269 missingPolicy := clearFields(t, update, func(envelope *wire.BrokerClientEnvelope) {
270 envelope.GetUpdateHostPolicy().Policy = nil
271 })
272 _, err = DecodeClient(missingPolicy, testEnvelopeCeiling, testChunkCeiling)
273 require.ErrorIs(t, err, ErrInvalidMessage)
274 missingRegistration := clearFields(t, update, func(envelope *wire.BrokerClientEnvelope) {
275 envelope.GetUpdateHostPolicy().Registration = nil
276 })
277 _, err = DecodeClient(missingRegistration, testEnvelopeCeiling, testChunkCeiling)
278 require.ErrorIs(t, err, ErrInvalidMessage)
279 })
280}
281
282// TestMembershipResultPresenceMismatchRefused proves the result presence rules
283// fail closed in both directions of the codec: removal never travels with a
284// registration, a failed or outcome-unknown result carries neither authority
285// and must carry its error, and a success carries no error.
286func TestMembershipResultPresenceMismatchRefused(t *testing.T) {
287 connection := testConnectionID(0x26)
288 operation := testOperationID(0x36)
289 clearRegistration := func(t *testing.T) []byte {
290 t.Helper()
291 raw := mustEncodeServer(t, OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()})
292 envelope := &wire.BrokerServerEnvelope{}
293 require.NoError(t, wire.ScanEnvelope(envelope, raw))
294 require.NoError(t, proto.Unmarshal(raw, envelope))
295 envelope.GetOperationResult().Registration.Generation = 0
296 mutated, err := proto.Marshal(envelope)
297 require.NoError(t, err)
298 return mutated
299 }
300
301 for name, message := range map[string]OperationResult{
302 "removed and registration together": {
303 Epoch: 7, Connection: connection, Operation: operation,
304 Outcome: ports.BrokerOutcomeOK, Removed: true, Registration: testRegistration(),
305 },
306 "failed with registration": {
307 Epoch: 7, Connection: connection, Operation: operation,
308 Outcome: ports.BrokerOutcomeFailed, Registration: testRegistration(), Error: testErrorDetail(), HasError: true,
309 },
310 "failed with removal": {
311 Epoch: 7, Connection: connection, Operation: operation,
312 Outcome: ports.BrokerOutcomeFailed, Removed: true, Error: testErrorDetail(), HasError: true,
313 },
314 "failed without error": {
315 Epoch: 7, Connection: connection, Operation: operation,
316 Outcome: ports.BrokerOutcomeFailed,
317 },
318 "unknown without error": {
319 Epoch: 7, Connection: connection, Operation: operation,
320 Outcome: ports.BrokerOutcomeUnknown,
321 },
322 "success with error": {
323 Epoch: 7, Connection: connection, Operation: operation,
324 Outcome: ports.BrokerOutcomeOK, Registration: testRegistration(), Error: testErrorDetail(), HasError: true,
325 },
326 "success with invalid registration": {
327 Epoch: 7, Connection: connection, Operation: operation,
328 Outcome: ports.BrokerOutcomeOK, Registration: domain.RemoteRegistration{Endpoint: "dev@host:22"},
329 },
330 } {
331 t.Run("encode/"+name, func(t *testing.T) {
332 _, err := EncodeServer(message, testEnvelopeCeiling, testChunkCeiling)
333 require.ErrorIs(t, err, ErrInvalidMessage)
334 })
335 }
336
337 t.Run("decode refuses a registration that cannot fence", func(t *testing.T) {
338 _, err := DecodeServer(clearRegistration(t), testEnvelopeCeiling, testChunkCeiling)
339 require.ErrorIs(t, err, ErrInvalidMessage)
340 })
341}
342
343// TestMembershipResultAuthorityPerMutationKind proves the per-kind rule names
344// the right authority for each request, so a removal success can never be read
345// as an addition and an addition success can never be read as a removal.
346func TestMembershipResultAuthorityPerMutationKind(t *testing.T) {
347 connection := testConnectionID(0x27)
348 operation := testOperationID(0x37)
349 add := OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()}
350 remove := OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Removed: true}
351 removeNoop := OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK}
352 failed := OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeFailed, Error: testErrorDetail(), HasError: true}
353
354 require.NoError(t, add.ValidateForMutation(MutationKindAddHost))
355 require.NoError(t, add.ValidateForMutation(MutationKindUpdateHostPolicy))
356 require.ErrorIs(t, add.ValidateForMutation(MutationKindRemoveHost), ErrInvalidMessage)
357
358 require.NoError(t, remove.ValidateForMutation(MutationKindRemoveHost))
359 require.NoError(t, removeNoop.ValidateForMutation(MutationKindRemoveHost), "an absent host is an idempotent no-op success")
360 require.ErrorIs(t, remove.ValidateForMutation(MutationKindAddHost), ErrInvalidMessage)
361 require.ErrorIs(t, removeNoop.ValidateForMutation(MutationKindAddHost), ErrInvalidMessage, "an addition success must carry its registration")
362 require.ErrorIs(t, removeNoop.ValidateForMutation(MutationKindUpdateHostPolicy), ErrInvalidMessage)
363
364 for _, kind := range []RegisterMutationKind{MutationKindAddHost, MutationKindRemoveHost, MutationKindUpdateHostPolicy} {
365 require.NoError(t, failed.ValidateForMutation(kind))
366 }
367 require.ErrorIs(t, add.ValidateForMutation(RegisterMutationKind(0)), errConvertRange)
368 require.ErrorIs(t, add.ValidateForMutation(RegisterMutationKind(9)), errConvertRange)
369}
370
371// TestMembershipErrorCodes proves the two membership codes travel and return
372// as the exact closed enumeration: 11 is host conflict and 12 is immutable
373// membership, on both the encode and decode paths.
374func TestMembershipErrorCodes(t *testing.T) {
375 connection := testConnectionID(0x28)
376 operation := testOperationID(0x38)
377 require.Equal(t, ports.BrokerErrorCode(11), ports.BrokerErrorHostConflict)
378 require.Equal(t, ports.BrokerErrorCode(12), ports.BrokerErrorMembershipImmutable)
379
380 for name, code := range map[string]ports.BrokerErrorCode{
381 "host_conflict": ports.BrokerErrorHostConflict,
382 "membership_immutable": ports.BrokerErrorMembershipImmutable,
383 } {
384 t.Run(name, func(t *testing.T) {
385 message := OperationResult{
386 Epoch: 7, Connection: connection, Operation: operation,
387 Outcome: ports.BrokerOutcomeFailed,
388 Error: ErrorDetail{Code: code, Text: "refused"}, HasError: true,
389 }
390 raw := mustEncodeServer(t, message)
391 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
392 require.NoError(t, err)
393 require.Equal(t, message, decoded)
394 require.Equal(t, code, decoded.(OperationResult).Error.Code)
395
396 envelope := &wire.BrokerServerEnvelope{}
397 require.NoError(t, proto.Unmarshal(raw, envelope))
398 require.Equal(t, uint32(code), envelope.GetOperationResult().GetError().GetCode())
399 })
400 }
401}
402
403// TestMembershipMalformedTruncatedTrailing proves the strict scanner runs
404// before generated unmarshal on the new variants too: unknown fields,
405// truncation, trailing garbage, and an oversize envelope are all refused.
406func TestMembershipMalformedTruncatedTrailing(t *testing.T) {
407 connection := testConnectionID(0x29)
408 operation := testOperationID(0x39)
409 requests := membershipRequests(connection, operation)
410
411 for _, tc := range requests {
412 t.Run("truncated/"+tc.name, func(t *testing.T) {
413 raw := mustEncodeClient(t, tc.message)
414 for size := range len(raw) {
415 _, err := DecodeClient(raw[:size], testEnvelopeCeiling, testChunkCeiling)
416 require.Error(t, err, "prefix[:%d] accepted", size)
417 }
418 })
419 t.Run("trailing/"+tc.name, func(t *testing.T) {
420 raw := mustEncodeClient(t, tc.message)
421 _, err := DecodeClient(append(append([]byte(nil), raw...), 0xFF), testEnvelopeCeiling, testChunkCeiling)
422 require.Error(t, err)
423 concatenated := append(append([]byte(nil), raw...), raw...)
424 _, err = DecodeClient(concatenated, testEnvelopeCeiling, testChunkCeiling)
425 require.ErrorIs(t, err, wire.ErrScanDuplicate)
426 })
427 t.Run("oversize/"+tc.name, func(t *testing.T) {
428 raw := mustEncodeClient(t, tc.message)
429 _, err := DecodeClient(raw, uint64(len(raw)-1), testChunkCeiling)
430 require.ErrorIs(t, err, wire.ErrScanLength)
431 })
432 }
433
434 t.Run("unknown nested field", func(t *testing.T) {
435 // UpdateHostPolicy → RemoteRegistration with an undeclared field.
436 registration := protowire.AppendTag(nil, 1, protowire.BytesType)
437 registration = protowire.AppendBytes(registration, []byte("dev@host:22"))
438 registration = protowire.AppendTag(registration, 9, protowire.VarintType)
439 registration = protowire.AppendVarint(registration, 1)
440 update := protowire.AppendTag(nil, 3, protowire.BytesType) // UpdateHostPolicy.registration
441 update = protowire.AppendBytes(update, registration)
442 outer := protowire.AppendTag(nil, 111, protowire.BytesType) // BrokerClientEnvelope.update_host_policy
443 outer = protowire.AppendBytes(outer, update)
444 _, err := DecodeClient(outer, testEnvelopeCeiling, testChunkCeiling)
445 require.ErrorIs(t, err, wire.ErrScanUnknown)
446 })
447
448 t.Run("unknown request field", func(t *testing.T) {
449 raw := mustEncodeClient(t, AddHost{Epoch: 7, Connection: connection, Operation: operation, Endpoint: "dev@host:22", Policy: testPolicy()})
450 envelope := &wire.BrokerClientEnvelope{}
451 require.NoError(t, wire.ScanEnvelope(envelope, raw))
452 require.NoError(t, proto.Unmarshal(raw, envelope))
453 envelope.GetAddHost().Policy = nil
454 mutated, err := proto.Marshal(envelope)
455 require.NoError(t, err)
456 mutated = protowire.AppendTag(mutated, 9, protowire.VarintType)
457 mutated = protowire.AppendVarint(mutated, 1)
458 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
459 require.ErrorIs(t, err, wire.ErrScanUnknown)
460 })
461
462 t.Run("wrong wire type on the new tag", func(t *testing.T) {
463 wrongType := protowire.AppendTag(nil, 111, protowire.VarintType)
464 wrongType = protowire.AppendVarint(wrongType, 1)
465 _, err := DecodeClient(wrongType, testEnvelopeCeiling, testChunkCeiling)
466 require.ErrorIs(t, err, wire.ErrScanWireType)
467 })
468
469 t.Run("result truncated", func(t *testing.T) {
470 raw := mustEncodeServer(t, OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()})
471 for size := range len(raw) {
472 _, err := DecodeServer(raw[:size], testEnvelopeCeiling, testChunkCeiling)
473 require.Error(t, err, "server prefix[:%d] accepted", size)
474 }
475 })
476}
477
478// TestMembershipUnknownTaxonomyRefused proves an out-of-range request tag or a
479// result outcome outside the closed taxonomy is refused rather than coerced.
480func TestMembershipUnknownTaxonomyRefused(t *testing.T) {
481 connection := testConnectionID(0x2A)
482 operation := testOperationID(0x3A)
483 raw := mustEncodeServer(t, OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeFailed, Error: testErrorDetail(), HasError: true})
484 envelope := &wire.BrokerServerEnvelope{}
485 require.NoError(t, wire.ScanEnvelope(envelope, raw))
486 require.NoError(t, proto.Unmarshal(raw, envelope))
487 envelope.GetOperationResult().Outcome = 9
488 mutated, err := proto.Marshal(envelope)
489 require.NoError(t, err)
490 _, err = DecodeServer(mutated, testEnvelopeCeiling, testChunkCeiling)
491 require.ErrorIs(t, err, ErrInvalidMessage)
492
493 // A malformed error code on a result is refused like one on an error frame.
494 envelope.GetOperationResult().Outcome = 2
495 envelope.GetOperationResult().GetError().Code = 99
496 mutated, err = proto.Marshal(envelope)
497 require.NoError(t, err)
498 _, err = DecodeServer(mutated, testEnvelopeCeiling, testChunkCeiling)
499 require.ErrorIs(t, err, ErrInvalidMessage)
500}