1package brokeripc
2
3import (
4 "context"
5 "errors"
6 "io"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/require"
12
13 "github.com/bnema/vev/internal/adapters/brokerwire"
14 "github.com/bnema/vev/internal/domain"
15 "github.com/bnema/vev/internal/ports"
16 "github.com/bnema/vev/internal/protocol/wire"
17)
18
19// Send-uncertainty taxonomy for mutating operations (GO-003). These tests drive
20// one client adapter over a channel-controlled carriage, so the exact instant
21// after a wire attempt is launched is observable without sleeping.
22//
23// The contract under test: only a request that definitely never left the process
24// fails as an ordinary, definite error; once the attempt was launched, a caller
25// cancellation, a carriage failure, or a connection teardown reports
26// outcome-unknown, so no caller replays a non-idempotent mutation blindly.
27
28// errTestCarriage is the deterministic carriage failure the gated carriage
29// reports for one released attempt.
30var errTestCarriage = errors.New("brokeripc test: carriage write failed")
31
32// gatedTransport is a channel-controlled client carriage. Every Send records its
33// envelope, announces the entered attempt, and then blocks until the test
34// releases it with an explicit result or abandons it. Close deliberately leaves
35// an in-flight Send blocked, so an attempt held inside the carriage observes
36// only the client's own terminal outcome until the test itself decides.
37type gatedTransport struct {
38 entered chan wire.Envelope
39 release chan error
40 abandon chan struct{}
41 closed chan struct{}
42
43 closeOnce sync.Once
44 abandonOnce sync.Once
45
46 mu sync.Mutex
47 sends int
48}
49
50func newGatedTransport() *gatedTransport {
51 return &gatedTransport{
52 entered: make(chan wire.Envelope, 8),
53 release: make(chan error, 8),
54 abandon: make(chan struct{}),
55 closed: make(chan struct{}),
56 }
57}
58
59// Send enters one wire attempt and holds it until the test releases it.
60func (t *gatedTransport) Send(envelope wire.Envelope) error {
61 t.mu.Lock()
62 t.sends++
63 t.mu.Unlock()
64 t.entered <- envelope
65 select {
66 case err := <-t.release:
67 return err
68 case <-t.abandon:
69 return io.EOF
70 }
71}
72
73func (t *gatedTransport) RecvBounded(uint64) (wire.Envelope, error) {
74 <-t.closed
75 return wire.Envelope{}, io.EOF
76}
77
78func (t *gatedTransport) Recv() (wire.Envelope, error) { return t.RecvBounded(0) }
79
80// Close ends the carriage without disturbing an in-flight attempt.
81func (t *gatedTransport) Close() error {
82 t.closeOnce.Do(func() { close(t.closed) })
83 return nil
84}
85
86// releaseSend hands one blocked attempt its deterministic result.
87func (t *gatedTransport) releaseSend(err error) { t.release <- err }
88
89// abandonAttempts releases every attempt still blocked in Send, so a test never
90// leaks a send goroutine.
91func (t *gatedTransport) abandonAttempts() {
92 t.abandonOnce.Do(func() { close(t.abandon) })
93}
94
95func (t *gatedTransport) sendCount() int {
96 t.mu.Lock()
97 defer t.mu.Unlock()
98 return t.sends
99}
100
101// newGatedClient builds one ready client adapter over a gated carriage. It never
102// starts the reader goroutine: the reply path is driven by the test, so every
103// observation is deterministic.
104func newGatedClient(t *testing.T, cfg Config) (*client, *gatedTransport) {
105 t.Helper()
106 transport := newGatedTransport()
107 t.Cleanup(transport.abandonAttempts)
108 scope := brokerwire.Scope{
109 Epoch: ports.BrokerEpoch(0x71),
110 Connection: ports.BrokerConnectionID{0x71, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x01},
111 }
112 conn, err := brokerwire.NewConnection(scope)
113 require.NoError(t, err)
114 require.NoError(t, conn.SignalPreamble())
115 require.NoError(t, conn.Register())
116 ctx, cancel := context.WithCancel(context.Background())
117 c := &client{
118 transport: transport,
119 ceilings: brokerwire.DefaultCeilings(),
120 scope: scope,
121 conn: conn,
122 cfg: cfg.withDefaults(),
123 ctx: ctx,
124 cancel: cancel,
125 pending: make(map[ports.BrokerOperationID]chan operationResult),
126 kinds: make(map[ports.BrokerOperationID]brokerwire.RegisterMutationKind),
127 streams: make(map[ports.BrokerStreamID]*clientStream),
128 done: make(chan struct{}),
129 }
130 t.Cleanup(func() { _ = c.Close() })
131 return c, transport
132}
133
134// mutationOutcome is one membership mutation result: the removal bool, the
135// authoritative registration (additions and policy updates), and the typed
136// error a caller observes.
137type mutationOutcome struct {
138 removed bool
139 registration domain.RemoteRegistration
140 err error
141}
142
143// mutationCall adapts one public ports.BrokerService membership method to the
144// shared table shape. Every entry drives the real public API, so the wire
145// request the carriage captures is exactly the one production sends.
146type mutationCall struct {
147 name string
148 endpoint string
149 wantRemoved bool
150 kind brokerwire.RegisterMutationKind
151 call func(ctx context.Context, c *client) (bool, domain.RemoteRegistration, error)
152 // success builds the payload of one legal wire result for this request
153 // kind; the test fills in the connection scope and operation identity.
154 success func(operation ports.BrokerOperationID) brokerwire.OperationResult
155}
156
157var mutationCalls = []mutationCall{
158 {
159 name: "AddHost",
160 endpoint: "new@host:22",
161 kind: brokerwire.MutationKindAddHost,
162 call: func(ctx context.Context, c *client) (bool, domain.RemoteRegistration, error) {
163 registration, err := c.AddHost(ctx, "new@host:22", testPolicy())
164 return false, registration, err
165 },
166 success: func(operation ports.BrokerOperationID) brokerwire.OperationResult {
167 return brokerwire.OperationResult{
168 Operation: operation, Outcome: ports.BrokerOutcomeOK,
169 Registration: mutationRegistration("new@host:22"),
170 }
171 },
172 },
173 {
174 name: "RemoveHost",
175 endpoint: "known",
176 wantRemoved: true,
177 kind: brokerwire.MutationKindRemoveHost,
178 call: func(ctx context.Context, c *client) (bool, domain.RemoteRegistration, error) {
179 removed, err := c.RemoveHost(ctx, mutationRegistration("known"))
180 return removed, domain.RemoteRegistration{}, err
181 },
182 success: func(operation ports.BrokerOperationID) brokerwire.OperationResult {
183 return brokerwire.OperationResult{Operation: operation, Outcome: ports.BrokerOutcomeOK, Removed: true}
184 },
185 },
186 {
187 name: "UpdateHostPolicy",
188 endpoint: "known",
189 kind: brokerwire.MutationKindUpdateHostPolicy,
190 call: func(ctx context.Context, c *client) (bool, domain.RemoteRegistration, error) {
191 registration, err := c.UpdateHostPolicy(ctx, mutationRegistration("known"), testPolicy())
192 return false, registration, err
193 },
194 success: func(operation ports.BrokerOperationID) brokerwire.OperationResult {
195 expected := mutationRegistration("known")
196 expected.Generation++
197 return brokerwire.OperationResult{Operation: operation, Outcome: ports.BrokerOutcomeOK, Registration: expected}
198 },
199 },
200}
201
202// mutationRegistration builds one valid exact registration for a mutation
203// fixture.
204func mutationRegistration(endpoint string) domain.RemoteRegistration {
205 registration, err := domain.NewRemoteRegistration(endpoint, [16]byte{0x11})
206 if err != nil {
207 panic(err)
208 }
209 return registration
210}
211
212// run drives one table entry through the public API.
213func (call mutationCall) run(ctx context.Context, c *client) (bool, domain.RemoteRegistration, error) {
214 return call.call(ctx, c)
215}
216
217// deliverSuccess plays one legal success result for a captured request.
218func (call mutationCall) deliverSuccess(t *testing.T, c *client, operation ports.BrokerOperationID) {
219 t.Helper()
220 result := call.success(operation)
221 result.Epoch = c.scope.Epoch
222 result.Connection = c.scope.Connection
223 require.NoError(t, c.dispatch(result))
224}
225
226// deliverResult plays one explicit broker reply for a captured request.
227func deliverResult(t *testing.T, c *client, result brokerwire.OperationResult) {
228 t.Helper()
229 require.NoError(t, c.dispatch(result))
230}
231
232// awaitResult waits for one mutation call to return.
233func awaitResult(t *testing.T, results <-chan mutationOutcome) mutationOutcome {
234 t.Helper()
235 select {
236 case got := <-results:
237 return got
238 case <-time.After(5 * time.Second):
239 t.Fatal("the mutating operation did not return")
240 return mutationOutcome{}
241 }
242}
243
244// awaitSend waits for one wire attempt to enter the carriage and decodes it.
245func awaitSend(t *testing.T, g *gatedTransport) brokerwire.ClientMessage {
246 t.Helper()
247 select {
248 case envelope := <-g.entered:
249 message, err := brokerwire.DecodeClient(envelope.Payload, brokerwire.MaxBrokerEnvelopeBytes, brokerwire.MaxStreamChunkBytes)
250 require.NoError(t, err, "the captured attempt must be a decodable client envelope")
251 return message
252 case <-time.After(5 * time.Second):
253 t.Fatal("no wire attempt was entered")
254 return nil
255 }
256}
257
258// operationOf extracts the operation identity of one captured request.
259func operationOf(t *testing.T, message brokerwire.ClientMessage) ports.BrokerOperationID {
260 t.Helper()
261 switch m := message.(type) {
262 case brokerwire.AddHost:
263 return m.Operation
264 case brokerwire.RemoveHost:
265 return m.Operation
266 case brokerwire.UpdateHostPolicy:
267 return m.Operation
268 default:
269 t.Fatalf("unexpected client message %T", message)
270 return ports.BrokerOperationID{}
271 }
272}
273
274// endpointOf extracts the target of one captured request. An AddHost names
275// its endpoint directly; a RemoveHost or UpdateHostPolicy carries exact
276// registration authority, whose endpoint is the target.
277func endpointOf(t *testing.T, message brokerwire.ClientMessage) string {
278 t.Helper()
279 switch m := message.(type) {
280 case brokerwire.AddHost:
281 return m.Endpoint
282 case brokerwire.RemoveHost:
283 return m.Registration.Endpoint
284 case brokerwire.UpdateHostPolicy:
285 return m.Registration.Endpoint
286 default:
287 t.Fatalf("unexpected client message %T", message)
288 return ""
289 }
290}
291
292// registrationOf extracts the exact expected registration of one captured
293// request. An AddHost carries no expected authority, only its policy.
294func registrationOf(t *testing.T, message brokerwire.ClientMessage) (domain.RemoteRegistration, bool) {
295 t.Helper()
296 switch m := message.(type) {
297 case brokerwire.RemoveHost:
298 return m.Registration, true
299 case brokerwire.UpdateHostPolicy:
300 return m.Registration, true
301 default:
302 return domain.RemoteRegistration{}, false
303 }
304}
305
306// policyOf extracts the policy of one captured request.
307func policyOf(t *testing.T, message brokerwire.ClientMessage) ports.BrokerPolicy {
308 t.Helper()
309 switch m := message.(type) {
310 case brokerwire.AddHost:
311 return m.Policy
312 case brokerwire.UpdateHostPolicy:
313 return m.Policy
314 default:
315 t.Fatalf("client message %T carries no policy", message)
316 return ports.BrokerPolicy{}
317 }
318}
319
320// pendingOperations snapshots the client's in-flight waiters.
321func pendingOperations(c *client) []ports.BrokerOperationID {
322 c.mu.Lock()
323 defer c.mu.Unlock()
324 out := make([]ports.BrokerOperationID, 0, len(c.pending))
325 for operation := range c.pending {
326 out = append(out, operation)
327 }
328 return out
329}
330
331func requireNoPendingMutationState(t *testing.T, c *client) {
332 t.Helper()
333 c.mu.Lock()
334 defer c.mu.Unlock()
335 require.Empty(t, c.pending, "pending replies must be released")
336 require.Empty(t, c.kinds, "pending mutation kinds must be released")
337}
338
339// TestPublicMembershipAPIDelegatesExactAuthority proves the public membership
340// surface reaches the carriage with the caller's exact authority and returns the
341// broker's authority verbatim: an addition returns the minted registration, a
342// policy update returns the generation-advanced registration, and a removal
343// returns the exact bool. Each call sends exactly one request carrying the
344// caller's policy or exact registration, and never re-derives one locally.
345func TestPublicMembershipAPIDelegatesExactAuthority(t *testing.T) {
346 policy := testPolicy()
347 cases := []struct {
348 name string
349 index int
350 }{
351 {name: "AddHost", index: 0},
352 {name: "RemoveHost", index: 1},
353 {name: "UpdateHostPolicy", index: 2},
354 }
355 for _, tc := range cases {
356 t.Run(tc.name, func(t *testing.T) {
357 call := mutationCalls[tc.index]
358 c, g := newGatedClient(t, Config{})
359 ctx, cancel := context.WithCancel(context.Background())
360 defer cancel()
361
362 results := make(chan mutationOutcome, 1)
363 go func() {
364 removed, registration, err := call.run(ctx, c)
365 results <- mutationOutcome{removed: removed, registration: registration, err: err}
366 }()
367
368 request := awaitSend(t, g)
369 operation := operationOf(t, request)
370 require.Equal(t, call.endpoint, endpointOf(t, request), "the captured request is the caller's own")
371 if expected, ok := registrationOf(t, request); ok {
372 require.Equal(t, mutationRegistration(call.endpoint), expected,
373 "the exact expected registration must travel unmodified")
374 require.NotZero(t, expected.Generation, "a removal or update never drops its fencing generation")
375 }
376 if call.kind != brokerwire.MutationKindRemoveHost {
377 require.Equal(t, policy, policyOf(t, request), "the caller's exact policy must travel")
378 }
379 kind, ok := brokerwire.MutationKindOf(request)
380 require.True(t, ok, "the captured request must be a membership mutation")
381 require.Equal(t, call.kind, kind, "the client must emit the request kind the table names")
382 g.releaseSend(nil)
383 call.deliverSuccess(t, c, operation)
384
385 got := awaitResult(t, results)
386 require.NoError(t, got.err)
387 require.Equal(t, call.wantRemoved, got.removed)
388 want := call.success(operation).Registration
389 require.Equal(t, want, got.registration, "the client returns the broker's authority verbatim")
390 require.Equal(t, 1, g.sendCount(), "one delegation is exactly one wire request")
391 require.Empty(t, pendingOperations(c))
392 })
393 }
394}
395
396// TestMutationSendUncertaintyTaxonomy pins the decision table for every mutating
397// operation: a request that definitely never left is a definite error with zero
398// sends; any failure observed after the attempt was launched is outcome-unknown
399// with exactly one send and no replay.
400func TestMutationSendUncertaintyTaxonomy(t *testing.T) {
401 scenarios := []struct {
402 name string
403 // before, when set, runs before the call starts.
404 before func(cancel context.CancelFunc)
405 // drive runs while the call is in flight.
406 drive func(t *testing.T, c *client, g *gatedTransport, cancel context.CancelFunc)
407 wantSends int
408 wantUnknown bool
409 }{
410 {
411 name: "pre-cancelled caller never dispatches and fails definitely",
412 before: func(cancel context.CancelFunc) { cancel() },
413 wantSends: 0,
414 },
415 {
416 name: "caller cancels while the attempt is blocked in Send",
417 drive: func(t *testing.T, c *client, g *gatedTransport, cancel context.CancelFunc) {
418 _ = awaitSend(t, g)
419 cancel()
420 },
421 wantSends: 1,
422 wantUnknown: true,
423 },
424 {
425 name: "connection terminates while the attempt is blocked in Send",
426 drive: func(t *testing.T, c *client, g *gatedTransport, cancel context.CancelFunc) {
427 _ = awaitSend(t, g)
428 require.NoError(t, c.Close())
429 },
430 wantSends: 1,
431 wantUnknown: true,
432 },
433 {
434 name: "reply lost after the request was sent",
435 drive: func(t *testing.T, c *client, g *gatedTransport, cancel context.CancelFunc) {
436 _ = awaitSend(t, g)
437 g.releaseSend(nil)
438 // The attempt completed, so the call now waits for its reply; only
439 // the connection's terminal outcome ends that wait.
440 require.NoError(t, c.Close())
441 },
442 wantSends: 1,
443 wantUnknown: true,
444 },
445 {
446 name: "Send fails after the request was captured",
447 drive: func(t *testing.T, c *client, g *gatedTransport, cancel context.CancelFunc) {
448 _ = awaitSend(t, g)
449 g.releaseSend(errTestCarriage)
450 },
451 wantSends: 1,
452 wantUnknown: true,
453 },
454 {
455 name: "carriage EOF after the request was captured",
456 drive: func(t *testing.T, c *client, g *gatedTransport, cancel context.CancelFunc) {
457 _ = awaitSend(t, g)
458 g.releaseSend(io.EOF)
459 },
460 wantSends: 1,
461 wantUnknown: true,
462 },
463 }
464 for _, call := range mutationCalls {
465 for _, tt := range scenarios {
466 t.Run(call.name+"/"+tt.name, func(t *testing.T) {
467 c, g := newGatedClient(t, Config{MaxPendingOperations: 1})
468 ctx, cancel := context.WithCancel(context.Background())
469 defer cancel()
470 if tt.before != nil {
471 tt.before(cancel)
472 }
473 results := make(chan mutationOutcome, 1)
474 go func() {
475 removed, registration, err := call.run(ctx, c)
476 results <- mutationOutcome{removed: removed, registration: registration, err: err}
477 }()
478 if tt.drive != nil {
479 tt.drive(t, c, g, cancel)
480 }
481 got := awaitResult(t, results)
482 require.False(t, got.removed, "no reply means the target is never reported removed")
483 require.Equal(t, domain.RemoteRegistration{}, got.registration,
484 "no reply means no registration authority is invented")
485 if tt.wantUnknown {
486 var failure ports.BrokerError
487 require.ErrorAs(t, got.err, &failure)
488 require.Equal(t, ports.BrokerErrorOutcomeUnknown, failure.Code,
489 "a launched attempt must never look like a definite failure")
490 } else {
491 require.ErrorIs(t, got.err, context.Canceled)
492 var failure ports.BrokerError
493 require.False(t, errors.As(got.err, &failure),
494 "a request that never left must not be reported as a broker outcome")
495 }
496 require.Equal(t, tt.wantSends, g.sendCount(), "the request must travel at most once")
497 requireNoPendingMutationState(t, c)
498 })
499 }
500 }
501}
502
503// TestMutationDefiniteFailureRecordsAndReleases proves a mutation whose request
504// definitely never left is recorded as failed rather than outcome-unknown, so
505// its pending slot is released for the next operation and nothing is resent.
506func TestMutationDefiniteFailureRecordsAndReleases(t *testing.T) {
507 c, g := newGatedClient(t, Config{MaxPendingOperations: 1})
508 ctx, cancel := context.WithCancel(context.Background())
509 cancel()
510
511 // Driving the public API with an already-cancelled caller is the definite
512 // no-dispatch case: no request is encoded for the wire.
513 added, err := c.AddHost(ctx, "pre@host:22", testPolicy())
514 require.ErrorIs(t, err, context.Canceled)
515 require.Equal(t, domain.RemoteRegistration{}, added)
516 require.Zero(t, g.sendCount(), "a pre-cancelled caller never enters the carriage")
517 require.Empty(t, pendingOperations(c), "the definite failure releases its pending slot")
518
519 // The bound of one proves the slot is genuinely free: a fresh mutation is
520 // admitted and completes normally.
521 fresh, cancelFresh := context.WithCancel(context.Background())
522 defer cancelFresh()
523 results := make(chan mutationOutcome, 1)
524 go func() {
525 removed, registration, callErr := mutationCalls[1].run(fresh, c)
526 results <- mutationOutcome{removed: removed, registration: registration, err: callErr}
527 }()
528 request := awaitSend(t, g)
529 operation := operationOf(t, request)
530 g.releaseSend(nil)
531 mutationCalls[1].deliverSuccess(t, c, operation)
532 got := awaitResult(t, results)
533 require.NoError(t, got.err)
534 require.True(t, got.removed)
535 require.Equal(t, 1, g.sendCount())
536}
537
538// TestMutationUnknownOutcomeReleasesWithoutResend proves one outcome-unknown
539// mutation releases its pending slot exactly once, never resends the request,
540// and ignores a late result for the same identity.
541func TestMutationUnknownOutcomeReleasesWithoutResend(t *testing.T) {
542 for _, call := range mutationCalls {
543 t.Run(call.name, func(t *testing.T) {
544 c, g := newGatedClient(t, Config{MaxPendingOperations: 1})
545 ctx, cancel := context.WithCancel(context.Background())
546 defer cancel()
547
548 results := make(chan mutationOutcome, 1)
549 go func() {
550 removed, registration, err := call.run(ctx, c)
551 results <- mutationOutcome{removed: removed, registration: registration, err: err}
552 }()
553 request := awaitSend(t, g)
554 operation := operationOf(t, request)
555 require.Equal(t, call.endpoint, endpointOf(t, request), "the captured request is the caller's own")
556 g.releaseSend(nil)
557 cancel()
558 got := awaitResult(t, results)
559 var failure ports.BrokerError
560 require.ErrorAs(t, got.err, &failure)
561 require.Equal(t, ports.BrokerErrorOutcomeUnknown, failure.Code)
562 require.False(t, got.removed)
563 require.Equal(t, 1, g.sendCount(), "an uncertain attempt is never resent")
564
565 // The identity was completed once, so its pending slot is released
566 // rather than left in flight.
567 require.ErrorIs(t, c.conn.AdmitOperation(operation), brokerwire.ErrOperationCompleted)
568 require.Empty(t, pendingOperations(c))
569
570 // A late legal result for the completed identity is dropped, and it
571 // never provokes another wire attempt.
572 late := call.success(operation)
573 late.Epoch = c.scope.Epoch
574 late.Connection = c.scope.Connection
575 deliverResult(t, c, late)
576 require.Empty(t, pendingOperations(c))
577 require.Equal(t, 1, g.sendCount())
578
579 // The connection stays usable: the next mutation is admitted under
580 // the same bound of one and completes normally.
581 fresh, cancelFresh := context.WithCancel(context.Background())
582 defer cancelFresh()
583 next := make(chan mutationOutcome, 1)
584 go func() {
585 removed, registration, err := call.run(fresh, c)
586 next <- mutationOutcome{removed: removed, registration: registration, err: err}
587 }()
588 second := awaitSend(t, g)
589 g.releaseSend(nil)
590 call.deliverSuccess(t, c, operationOf(t, second))
591 done := awaitResult(t, next)
592 require.NoError(t, done.err)
593 require.Equal(t, call.wantRemoved, done.removed)
594 require.Equal(t, 2, g.sendCount())
595 })
596 }
597}
598
599// TestMutationDeliveredReplyIsDefinite proves an ordinary delivered reply stays
600// the definite outcome for every operation, and a repeated result for the same
601// identity changes nothing.
602func TestMutationDeliveredReplyIsDefinite(t *testing.T) {
603 for _, call := range mutationCalls {
604 t.Run(call.name, func(t *testing.T) {
605 c, g := newGatedClient(t, Config{})
606 ctx, cancel := context.WithCancel(context.Background())
607 defer cancel()
608
609 results := make(chan mutationOutcome, 1)
610 go func() {
611 removed, registration, err := call.run(ctx, c)
612 results <- mutationOutcome{removed: removed, registration: registration, err: err}
613 }()
614 request := awaitSend(t, g)
615 operation := operationOf(t, request)
616 require.Equal(t, call.endpoint, endpointOf(t, request))
617 g.releaseSend(nil)
618 call.deliverSuccess(t, c, operation)
619 got := awaitResult(t, results)
620 require.NoError(t, got.err)
621 require.Equal(t, call.wantRemoved, got.removed)
622 require.Equal(t, 1, g.sendCount())
623 require.Empty(t, pendingOperations(c))
624
625 // A replayed result for an already settled identity is dropped.
626 replay := call.success(operation)
627 replay.Epoch = c.scope.Epoch
628 replay.Connection = c.scope.Connection
629 deliverResult(t, c, replay)
630 require.Empty(t, pendingOperations(c))
631 require.Equal(t, 1, g.sendCount())
632 })
633 }
634}
635
636// TestMutationInvalidResultIsOutcomeUnknownAndAbortsProtocol proves every result
637// that cannot carry authority for its request is refused rather than applied: a
638// result of the wrong kind, a success missing its registration authority, a
639// removal success that also carries a registration, and a result with a
640// malformed error detail each complete the caller as outcome-unknown and settle
641// the connection with a malformed-frame protocol violation.
642func TestMutationInvalidResultIsOutcomeUnknownAndAbortsProtocol(t *testing.T) {
643 cases := []struct {
644 name string
645 call mutationCall
646 result func(c *client, operation ports.BrokerOperationID) brokerwire.OperationResult
647 }{
648 {
649 name: "wrong kind for an addition",
650 call: mutationCalls[0],
651 result: func(c *client, operation ports.BrokerOperationID) brokerwire.OperationResult {
652 return brokerwire.OperationResult{
653 Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation,
654 Outcome: ports.BrokerOutcomeOK, Removed: true,
655 }
656 },
657 },
658 {
659 name: "addition success missing its registration",
660 call: mutationCalls[0],
661 result: func(c *client, operation ports.BrokerOperationID) brokerwire.OperationResult {
662 return brokerwire.OperationResult{
663 Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation,
664 Outcome: ports.BrokerOutcomeOK,
665 }
666 },
667 },
668 {
669 name: "policy update success missing its registration",
670 call: mutationCalls[2],
671 result: func(c *client, operation ports.BrokerOperationID) brokerwire.OperationResult {
672 return brokerwire.OperationResult{
673 Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation,
674 Outcome: ports.BrokerOutcomeOK,
675 }
676 },
677 },
678 {
679 name: "removal success carrying a registration",
680 call: mutationCalls[1],
681 result: func(c *client, operation ports.BrokerOperationID) brokerwire.OperationResult {
682 return brokerwire.OperationResult{
683 Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation,
684 Outcome: ports.BrokerOutcomeOK, Removed: true, Registration: mutationRegistration("known"),
685 }
686 },
687 },
688 {
689 name: "malformed error detail",
690 call: mutationCalls[0],
691 result: func(c *client, operation ports.BrokerOperationID) brokerwire.OperationResult {
692 return brokerwire.OperationResult{
693 Epoch: c.scope.Epoch, Connection: c.scope.Connection, Operation: operation,
694 Outcome: ports.BrokerOutcomeFailed,
695 Error: brokerwire.ErrorDetail{Code: ports.BrokerErrorCode(99), Text: "bogus"}, HasError: true,
696 }
697 },
698 },
699 }
700 for _, tc := range cases {
701 t.Run(tc.name, func(t *testing.T) {
702 c, g := newGatedClient(t, Config{})
703 ctx, cancel := context.WithCancel(context.Background())
704 defer cancel()
705
706 results := make(chan mutationOutcome, 1)
707 go func() {
708 removed, registration, err := tc.call.run(ctx, c)
709 results <- mutationOutcome{removed: removed, registration: registration, err: err}
710 }()
711 request := awaitSend(t, g)
712 operation := operationOf(t, request)
713 g.releaseSend(nil)
714
715 dispatchErr := c.dispatch(tc.result(c, operation))
716 require.Error(t, dispatchErr, "an invalid result must settle the connection")
717 require.ErrorIs(t, dispatchErr, ErrMalformedFrame, "an invalid result is a protocol violation")
718
719 got := awaitResult(t, results)
720 var failure ports.BrokerError
721 require.ErrorAs(t, got.err, &failure)
722 require.Equal(t, ports.BrokerErrorOutcomeUnknown, failure.Code,
723 "a result that cannot be applied must never look like a definite outcome")
724 require.False(t, got.removed)
725 require.Equal(t, domain.RemoteRegistration{}, got.registration)
726 require.Equal(t, 1, g.sendCount(), "an invalid result must never provoke a replay")
727 require.Empty(t, pendingOperations(c))
728 })
729 }
730}
731
732// scriptedTransport is a client carriage whose single inbound frame can be
733// injected: the reader goroutine consumes it and then blocks until the test
734// closes the carriage. It lets a test drive the client's real run loop with one
735// malformed frame and observe how a pending mutation is settled.
736type scriptedTransport struct {
737 ready chan wire.Envelope
738 done chan struct{}
739 once sync.Once
740
741 mu sync.Mutex
742 sends []wire.Envelope
743}
744
745func newScriptedTransport() *scriptedTransport {
746 return &scriptedTransport{ready: make(chan wire.Envelope, 1), done: make(chan struct{})}
747}
748
749func (t *scriptedTransport) deliver(envelope wire.Envelope) { t.ready <- envelope }
750
751func (t *scriptedTransport) Send(envelope wire.Envelope) error {
752 t.mu.Lock()
753 t.sends = append(t.sends, envelope)
754 t.mu.Unlock()
755 return nil
756}
757
758func (t *scriptedTransport) RecvBounded(uint64) (wire.Envelope, error) {
759 select {
760 case envelope := <-t.ready:
761 return envelope, nil
762 case <-t.done:
763 return wire.Envelope{}, io.EOF
764 }
765}
766
767func (t *scriptedTransport) Recv() (wire.Envelope, error) { return t.RecvBounded(0) }
768
769func (t *scriptedTransport) Close() error {
770 t.once.Do(func() { close(t.done) })
771 return nil
772}
773
774func (t *scriptedTransport) sendCount() int {
775 t.mu.Lock()
776 defer t.mu.Unlock()
777 return len(t.sends)
778}
779
780// TestMalformedResultFrameSettlesPendingMutationAsUnknown proves a malformed
781// server frame the reader refuses settles the connection and is reported to a
782// pending mutation as outcome-unknown, never as a definite failure: the request
783// was already sent, so the caller must refresh rather than replay it.
784func TestMalformedResultFrameSettlesPendingMutationAsUnknown(t *testing.T) {
785 transport := newScriptedTransport()
786 scope := brokerwire.Scope{
787 Epoch: ports.BrokerEpoch(0x72),
788 Connection: ports.BrokerConnectionID{0x72, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x01},
789 }
790 conn, err := brokerwire.NewConnection(scope)
791 require.NoError(t, err)
792 require.NoError(t, conn.SignalPreamble())
793 require.NoError(t, conn.Register())
794 ctx, cancel := context.WithCancel(context.Background())
795 defer cancel()
796 c := &client{
797 transport: transport,
798 ceilings: brokerwire.DefaultCeilings(),
799 scope: scope,
800 conn: conn,
801 cfg: Config{}.withDefaults(),
802 ctx: ctx,
803 cancel: cancel,
804 pending: make(map[ports.BrokerOperationID]chan operationResult),
805 kinds: make(map[ports.BrokerOperationID]brokerwire.RegisterMutationKind),
806 streams: make(map[ports.BrokerStreamID]*clientStream),
807 done: make(chan struct{}),
808 }
809 t.Cleanup(func() { _ = c.Close() })
810 go c.run()
811
812 results := make(chan mutationOutcome, 1)
813 go func() {
814 removed, err := c.RemoveHost(context.Background(), mutationRegistration("known"))
815 results <- mutationOutcome{removed: removed, err: err}
816 }()
817 require.Eventually(t, func() bool { return transport.sendCount() == 1 }, 5*time.Second, time.Millisecond,
818 "the mutation request must be launched before the malformed frame arrives")
819 c.mu.Lock()
820 pending := len(c.pending)
821 c.mu.Unlock()
822 require.Equal(t, 1, pending, "the mutation must still be pending when the malformed frame arrives")
823
824 // One frame the strict scanner refuses: a payload that is not a valid server
825 // envelope. The reader settles the connection on it.
826 transport.deliver(wire.Envelope{Payload: []byte{0xFF, 0xFF, 0xFF}})
827
828 got := awaitResult(t, results)
829 var failure ports.BrokerError
830 require.ErrorAs(t, got.err, &failure)
831 require.Equal(t, ports.BrokerErrorOutcomeUnknown, failure.Code,
832 "a malformed frame after dispatch must never look like a definite outcome")
833 require.False(t, got.removed)
834 require.Equal(t, domain.RemoteRegistration{}, got.registration)
835 require.Equal(t, 1, transport.sendCount(), "the request must have travelled exactly once")
836 select {
837 case <-c.Done():
838 case <-time.After(5 * time.Second):
839 t.Fatal("the malformed frame must settle the connection")
840 }
841 require.ErrorIs(t, c.Err(), ErrMalformedFrame)
842}
843
844// TestSendContextPreflightClassification pins the private helper's taxonomy
845// directly: an unencodable frame, an already-cancelled caller, and an
846// already-terminal connection are definitely not dispatched; only a launched
847// attempt is uncertain.
848func TestSendContextPreflightClassification(t *testing.T) {
849 t.Run("unencodable frame is definite", func(t *testing.T) {
850 c, g := newGatedClient(t, Config{})
851 attempted, err := c.sendContext(context.Background(), nil)
852 require.False(t, attempted)
853 require.ErrorIs(t, err, brokerwire.ErrInvalidMessage)
854 require.Zero(t, g.sendCount())
855 })
856
857 t.Run("already-cancelled caller is definite", func(t *testing.T) {
858 c, g := newGatedClient(t, Config{})
859 ctx, cancel := context.WithCancel(context.Background())
860 cancel()
861 attempted, err := c.sendContext(ctx, brokerwire.AddHost{
862 Epoch: c.scope.Epoch, Connection: c.scope.Connection,
863 Operation: ports.BrokerOperationID{0x01}, Endpoint: "pre@host:22", Policy: testPolicy(),
864 })
865 require.False(t, attempted)
866 require.ErrorIs(t, err, context.Canceled)
867 require.Zero(t, g.sendCount())
868 })
869
870 t.Run("already-terminal connection is definite", func(t *testing.T) {
871 c, g := newGatedClient(t, Config{})
872 require.NoError(t, c.Close())
873 attempted, err := c.sendContext(context.Background(), brokerwire.AddHost{
874 Epoch: c.scope.Epoch, Connection: c.scope.Connection,
875 Operation: ports.BrokerOperationID{0x01}, Endpoint: "pre@host:22", Policy: testPolicy(),
876 })
877 require.False(t, attempted)
878 require.ErrorIs(t, err, ErrConnectionClosed)
879 require.Zero(t, g.sendCount())
880 })
881
882 t.Run("launched attempt is uncertain", func(t *testing.T) {
883 c, g := newGatedClient(t, Config{})
884 type sendResult struct {
885 attempted bool
886 err error
887 }
888 done := make(chan sendResult, 1)
889 go func() {
890 attempted, err := c.sendContext(context.Background(), brokerwire.RemoveHost{
891 Epoch: c.scope.Epoch, Connection: c.scope.Connection,
892 Operation: ports.BrokerOperationID{0x02}, Registration: mutationRegistration("known"),
893 })
894 done <- sendResult{attempted: attempted, err: err}
895 }()
896 _ = awaitSend(t, g)
897 g.releaseSend(errTestCarriage)
898 select {
899 case got := <-done:
900 require.True(t, got.attempted, "a launched attempt is dispatch-uncertain")
901 require.ErrorIs(t, got.err, errTestCarriage)
902 case <-time.After(5 * time.Second):
903 t.Fatal("sendContext did not return")
904 }
905 require.Equal(t, 1, g.sendCount())
906 })
907}