1package brokeripc
2
3import (
4 "bytes"
5 "context"
6 "encoding/binary"
7 "errors"
8 "fmt"
9 "io"
10 "os"
11 "path/filepath"
12 "sync"
13 "testing"
14 "time"
15
16 "github.com/stretchr/testify/require"
17
18 "github.com/bnema/vev/internal/adapters/brokerwire"
19 "github.com/bnema/vev/internal/adapters/sessionwire"
20 "github.com/bnema/vev/internal/domain"
21 "github.com/bnema/vev/internal/ports"
22 "github.com/bnema/vev/internal/protocol"
23 "github.com/bnema/vev/internal/protocol/catalogue"
24 "github.com/bnema/vev/internal/protocol/wire"
25)
26
27// testSocketPath returns a short, owner-only parent directory and one broker
28// endpoint inside it, so a real AF_UNIX path is exercised well within the
29// cross-platform pathname limit.
30func testSocketPath(t *testing.T) string {
31 t.Helper()
32 root, err := os.MkdirTemp("/tmp", "v")
33 require.NoError(t, err)
34 t.Cleanup(func() { _ = os.RemoveAll(root) })
35 return SocketPath(filepath.Join(root, "vev"))
36}
37
38// testPolicy builds one valid broker policy.
39func testPolicy() ports.BrokerPolicy {
40 return ports.BrokerPolicy{
41 ProtocolVersion: protocol.Version,
42 CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,
43 EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned,
44 Transport: "unix",
45 Trust: "trusted",
46 Launch: "explicit",
47 Isolation: "per-user",
48 }
49}
50
51// openRequest builds one valid control-purpose open request for exactly stream,
52// with this connection's start-if-needed authorization. After the cutover the
53// calling service allocates every identity with NextStreamID, so a test that
54// names a stream always states it explicitly.
55func openRequest(stream ports.BrokerStreamID) ports.BrokerOpenStreamRequest {
56 return ports.BrokerOpenStreamRequest{
57 Purpose: ports.BrokerStreamControl,
58 Local: true,
59 Stream: stream,
60 Policy: testPolicy(),
61 StartMode: ports.BrokerDaemonStartIfNeeded,
62 }
63}
64
65// attachmentRequest builds one valid attachment request for exactly stream with
66// the given admission variant and creation name.
67func attachmentRequest(stream ports.BrokerStreamID, admission ports.BrokerStreamAdmission, name string) ports.BrokerOpenStreamRequest {
68 request := openRequest(stream)
69 request.Purpose = ports.BrokerStreamAttachment
70 request.Admission = admission
71 request.Name = name
72 return request
73}
74
75// nextStreamID reads one fresh identity from the service's only allocator,
76// exactly as the composed supervisor and broker operations do.
77func nextStreamID(t *testing.T, service ports.BrokerSession) ports.BrokerStreamID {
78 t.Helper()
79 stream, err := service.NextStreamID()
80 require.NoError(t, err)
81 return stream
82}
83
84// membershipProbeEndpoint is the endpoint every liveness probe adds through the
85// real membership delegation path.
86const membershipProbeEndpoint = "probe@host:22"
87
88// requireConnectionDelegates proves one connection still admits a full
89// membership round trip after a stream-local or subscriber-local event: the
90// probe adds one host through the same delegated path a real caller uses and
91// asserts the registration comes back for exactly the requested endpoint.
92func requireConnectionDelegates(t *testing.T, service ports.BrokerSession) {
93 t.Helper()
94 require.NoError(t, connectionDelegates(context.Background(), service))
95}
96
97// connectionDelegates is the error-returning form of requireConnectionDelegates,
98// safe to call from a test goroutine where FailNow is not.
99func connectionDelegates(ctx context.Context, service ports.BrokerSession) error {
100 ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
101 defer cancel()
102 registration, err := service.AddHost(ctx, membershipProbeEndpoint, testPolicy())
103 if err != nil {
104 return err
105 }
106 if registration.Endpoint != membershipProbeEndpoint {
107 return fmt.Errorf("AddHost answered registration for %q, want %q", registration.Endpoint, membershipProbeEndpoint)
108 }
109 return nil
110}
111
112// recvWithin reads one server message under an explicit bound, so a session that
113// never answers settles as a bounded test failure instead of hanging the package
114// on a read that can never complete.
115func (r *rawCarriage) recvWithin(t *testing.T, timeout time.Duration) brokerwire.ServerMessage {
116 t.Helper()
117 type result struct {
118 message brokerwire.ServerMessage
119 err error
120 }
121 done := make(chan result, 1)
122 go func() {
123 envelope, err := r.transport.RecvBounded(r.ceilings.MaxReceiveEnvelopeBytes)
124 if err != nil {
125 done <- result{nil, err}
126 return
127 }
128 message, err := brokerwire.DecodeServer(envelope.Payload, r.ceilings.MaxReceiveEnvelopeBytes, r.ceilings.StreamChunkLimit)
129 done <- result{message, err}
130 }()
131 select {
132 case got := <-done:
133 require.NoError(t, got.err, "the broker must answer with a decodable server frame")
134 return got.message
135 case <-time.After(timeout):
136 t.Fatal("the broker did not answer within the bound")
137 return nil
138 }
139}
140
141// awaitReadError proves one raw read fails under an explicit bound, so a session
142// that never settles is a bounded test failure rather than a hang.
143func (r *rawCarriage) awaitReadError(t *testing.T, timeout time.Duration) error {
144 t.Helper()
145 done := make(chan error, 1)
146 go func() {
147 _, err := r.transport.RecvBounded(r.ceilings.MaxReceiveEnvelopeBytes)
148 done <- err
149 }()
150 select {
151 case err := <-done:
152 return err
153 case <-time.After(timeout):
154 t.Fatal("the connection must settle within the bound")
155 return nil
156 }
157}
158
159// testDaemonAt builds one catalogue-valid remote daemon projection for index i.
160func testDaemonAt(i int) ports.BrokerDaemonObservation {
161 endpoint := fmt.Sprintf("user%d@host%d:22", i, i)
162 return ports.BrokerDaemonObservation{
163 Endpoint: endpoint,
164 DisplayOrigin: endpoint,
165 Registration: domain.RemoteRegistration{Endpoint: endpoint, Incarnation: [16]byte{0x10, byte(i + 1), 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xf0, 0x11}, Generation: domain.RemoteGeneration(i + 1)},
166 Policy: testPolicy(),
167 Availability: domain.RemoteAvailabilityReachable,
168 LastSuccess: time.Unix(1700000000+int64(i), 0).UTC(),
169 InventoryKnown: true,
170 Sessions: []catalogue.RemoteCatalogSession{},
171 }
172}
173
174// testSessionAt builds one catalogue-valid session for a host and index.
175func testSessionAt(host, index int) catalogue.RemoteCatalogSession {
176 return catalogue.RemoteCatalogSession{
177 LifecycleID: domain.SessionLifecycleID{byte(host + 1), byte(index >> 8), byte(index), 0x77},
178 Name: fmt.Sprintf("s%02d-%03d", host, index),
179 State: catalogue.RemoteCatalogSessionDown,
180 Tabs: []catalogue.RemoteCatalogTab{},
181 }
182}
183
184// testSnapshot builds one valid publication for an epoch and revision.
185func testSnapshot(epoch ports.BrokerEpoch, revision ports.BrokerRevision) ports.BrokerSnapshot {
186 daemon := testDaemonAt(0)
187 daemon.Sessions = []catalogue.RemoteCatalogSession{testSessionAt(0, 0), testSessionAt(0, 1)}
188 return ports.BrokerSnapshot{Epoch: epoch, Revision: revision, Daemons: []ports.BrokerDaemonObservation{daemon}}
189}
190
191// testLocalSnapshot builds one valid publication whose first daemon is local:
192// the local daemon with one session followed by one remote daemon with one.
193func testLocalSnapshot(epoch ports.BrokerEpoch, revision ports.BrokerRevision) ports.BrokerSnapshot {
194 local := ports.BrokerDaemonObservation{
195 Local: true,
196 DisplayOrigin: "local",
197 Policy: testPolicy(),
198 Availability: domain.RemoteAvailabilityReachable,
199 InventoryKnown: true,
200 Sessions: []catalogue.RemoteCatalogSession{testSessionAt(0, 0)},
201 }
202 remote := testDaemonAt(0)
203 remote.Sessions = []catalogue.RemoteCatalogSession{testSessionAt(1, 0)}
204 return ports.BrokerSnapshot{Epoch: epoch, Revision: revision, Daemons: []ports.BrokerDaemonObservation{local, remote}}
205}
206
207// snapshotHub is a multi-subscriber publication hub standing in for the broker
208// registry: one immutable current snapshot, coalescing capacity-one subscribers.
209type snapshotHub struct {
210 mu sync.Mutex
211 snapshot ports.BrokerSnapshot
212 subs map[*fakeSubscription]struct{}
213}
214
215func newSnapshotHub(snapshot ports.BrokerSnapshot) *snapshotHub {
216 return &snapshotHub{snapshot: snapshot, subs: make(map[*fakeSubscription]struct{})}
217}
218
219func (h *snapshotHub) current() ports.BrokerSnapshot {
220 h.mu.Lock()
221 defer h.mu.Unlock()
222 return h.snapshot.Clone()
223}
224
225func (h *snapshotHub) subscribe() *fakeSubscription {
226 sub := &fakeSubscription{hub: h, changed: make(chan struct{}, 1)}
227 h.mu.Lock()
228 h.subs[sub] = struct{}{}
229 h.mu.Unlock()
230 return sub
231}
232
233func (h *snapshotHub) publish(snapshot ports.BrokerSnapshot) {
234 h.mu.Lock()
235 h.snapshot = snapshot.Clone()
236 subs := make([]*fakeSubscription, 0, len(h.subs))
237 for sub := range h.subs {
238 subs = append(subs, sub)
239 }
240 h.mu.Unlock()
241 for _, sub := range subs {
242 sub.notify()
243 }
244}
245
246func (h *snapshotHub) remove(sub *fakeSubscription) {
247 h.mu.Lock()
248 delete(h.subs, sub)
249 h.mu.Unlock()
250}
251
252// fakeSubscription is one coalescing capacity-one subscription.
253type fakeSubscription struct {
254 hub *snapshotHub
255 changed chan struct{}
256 once sync.Once
257 mu sync.Mutex
258 closed bool
259}
260
261func (s *fakeSubscription) Changed() <-chan struct{} { return s.changed }
262
263func (s *fakeSubscription) Close() {
264 s.once.Do(func() {
265 s.mu.Lock()
266 s.closed = true
267 s.mu.Unlock()
268 s.hub.remove(s)
269 })
270}
271
272func (s *fakeSubscription) notify() {
273 s.mu.Lock()
274 defer s.mu.Unlock()
275 if s.closed {
276 return
277 }
278 select {
279 case s.changed <- struct{}{}:
280 default:
281 }
282}
283
284// fakeLogicalConn is one admitted core logical connection: it records messages
285// the broker relays toward the daemon and lets a test inject daemon replies.
286type fakeLogicalConn struct {
287 fromClient chan protocol.ClientMessage
288 toClient chan protocol.ServerMessage
289 // preamble carries the canned session preamble acceptance: the raw relay
290 // forwards the client's preamble request like any other envelope.
291 preamble chan []byte
292 done chan struct{}
293 once sync.Once
294 mu sync.Mutex
295 err error
296 closed bool
297}
298
299func newFakeLogicalConn() *fakeLogicalConn {
300 return &fakeLogicalConn{
301 fromClient: make(chan protocol.ClientMessage, 64),
302 toClient: make(chan protocol.ServerMessage, 64),
303 preamble: make(chan []byte, 1),
304 done: make(chan struct{}),
305 }
306}
307
308func (c *fakeLogicalConn) SendEnvelope(payload []byte) error {
309 if request, err := sessionwire.EncodePreambleRequestForTest(); err == nil && bytes.Equal(payload, request) {
310 response, err := sessionwire.EncodePreambleResponseForTest()
311 if err != nil {
312 return err
313 }
314 c.preamble <- response
315 return nil
316 }
317 message, err := sessionwire.DecodeClientEnvelope(payload)
318 if err != nil {
319 return err
320 }
321 select {
322 case c.fromClient <- message:
323 return nil
324 case <-c.done:
325 return ports.BrokerAdmissionClosed
326 }
327}
328
329func (c *fakeLogicalConn) RecvEnvelope() ([]byte, error) {
330 // A message already queued when Close ran must still be observed first: a
331 // real carriage never reports the connection done before a message the peer
332 // sent ahead of an orderly close has been delivered. Draining non-blocking
333 // before the select keeps that ordering deterministic instead of racing a
334 // buffered send against an already-closed done channel.
335 select {
336 case response := <-c.preamble:
337 return response, nil
338 case message := <-c.toClient:
339 return sessionwire.EncodeServerMessage(message)
340 default:
341 }
342 select {
343 case response := <-c.preamble:
344 return response, nil
345 case message := <-c.toClient:
346 return sessionwire.EncodeServerMessage(message)
347 case <-c.done:
348 if err := c.Err(); err != nil {
349 return nil, err
350 }
351 return nil, io.EOF
352 }
353}
354
355func (c *fakeLogicalConn) Capabilities() protocol.ConnectionCapabilities {
356 return protocol.ConnectionCapabilities{OutputDataLimit: protocol.MaxOutputDataLen}
357}
358
359func (c *fakeLogicalConn) LinkState() ports.LinkState { return ports.LinkStateConnected }
360func (c *fakeLogicalConn) LinkEvents() <-chan ports.LinkEvent { return nil }
361func (c *fakeLogicalConn) Done() <-chan struct{} { return c.done }
362func (c *fakeLogicalConn) Err() error { c.mu.Lock(); defer c.mu.Unlock(); return c.err }
363
364func (c *fakeLogicalConn) Close() error {
365 c.once.Do(func() {
366 c.mu.Lock()
367 c.closed = true
368 c.mu.Unlock()
369 close(c.done)
370 })
371 return nil
372}
373
374// fail publishes one terminal core failure.
375func (c *fakeLogicalConn) fail(err error) {
376 c.mu.Lock()
377 c.err = err
378 c.mu.Unlock()
379 _ = c.Close()
380}
381
382// addHostCall records one delegated membership addition: the exact arguments
383// the admitted core received and the exact result it returned. Recording both
384// is what lets a test prove the adapter delegated once with the caller's own
385// authority instead of re-deriving a policy or registration on the way.
386type addHostCall struct {
387 Endpoint string
388 Policy ports.BrokerPolicy
389 Result domain.RemoteRegistration
390 Err error
391}
392
393// removeHostCall records one delegated exact-registration removal.
394type removeHostCall struct {
395 Expected domain.RemoteRegistration
396 Removed bool
397 Err error
398}
399
400// updateHostPolicyCall records one delegated exact-registration policy update.
401type updateHostPolicyCall struct {
402 Expected domain.RemoteRegistration
403 Policy ports.BrokerPolicy
404 Result domain.RemoteRegistration
405 Err error
406}
407
408// fakeCore is one admitted per-connection broker service.
409type fakeCore struct {
410 mu sync.Mutex
411 id ports.BrokerConnectionID
412 hub *snapshotHub
413 streams map[ports.BrokerStreamID]*fakeLogicalConn
414 opens []ports.BrokerOpenStreamRequest
415 closedIDs []ports.BrokerStreamID
416 added []addHostCall
417 removed []removeHostCall
418 updated []updateHostPolicyCall
419 reconcile []string
420 openErr error
421 // subscribes records each core subscription's reader kind (true = passive).
422 subscribes []bool
423 closed bool
424 done chan struct{}
425 once sync.Once
426
427 // Membership result hooks. Their zero values keep the deterministic
428 // defaults that mirror the real registry: a fresh generation-1 registration
429 // for a new endpoint, a generation advanced by exactly one for a policy
430 // update, and removal reported from removeRemoved.
431 addErr error
432 removeErr error
433 updateErr error
434 addResult *domain.RemoteRegistration
435 updateResult *domain.RemoteRegistration
436 removeRemoved bool
437
438 nextStream ports.BrokerStreamID
439}
440
441func (c *fakeCore) ConnectionID() ports.BrokerConnectionID { return c.id }
442
443// NextStreamID is the admitted core connection's only stream allocator, exactly
444// like the production service: the listener delegates it unchanged.
445func (c *fakeCore) NextStreamID() (ports.BrokerStreamID, error) {
446 c.mu.Lock()
447 defer c.mu.Unlock()
448 c.nextStream++
449 return c.nextStream, nil
450}
451
452func (c *fakeCore) Done() <-chan struct{} { return c.done }
453
454func (c *fakeCore) Err() error { return nil }
455
456func (c *fakeCore) Snapshot() ports.BrokerSnapshot { return c.hub.current() }
457
458func (c *fakeCore) Subscribe() (ports.BrokerSubscription, error) {
459 return c.recordSubscribe(false), nil
460}
461
462// SubscribePassive is the optional passive-reader capability the server uses
463// for a Subscribe flagged passive.
464func (c *fakeCore) SubscribePassive() (ports.BrokerSubscription, error) {
465 return c.recordSubscribe(true), nil
466}
467
468func (c *fakeCore) recordSubscribe(passive bool) *fakeSubscription {
469 c.mu.Lock()
470 c.subscribes = append(c.subscribes, passive)
471 c.mu.Unlock()
472 return c.hub.subscribe()
473}
474
475func (c *fakeCore) subscribeKinds() []bool {
476 c.mu.Lock()
477 defer c.mu.Unlock()
478 return append([]bool(nil), c.subscribes...)
479}
480
481// fakePreviewSubscription is one server-side preview subscription whose
482// Changed channel a test drives explicitly.
483type fakePreviewSubscription struct {
484 changed chan struct{}
485 once sync.Once
486 mu sync.Mutex
487 closed bool
488}
489
490func newFakePreviewSubscription() *fakePreviewSubscription {
491 return &fakePreviewSubscription{changed: make(chan struct{}, 1)}
492}
493
494func (s *fakePreviewSubscription) Changed() <-chan struct{} { return s.changed }
495
496func (s *fakePreviewSubscription) Latest() ports.BrokerPreviewPublication {
497 return ports.BrokerPreviewPublication{}
498}
499
500func (s *fakePreviewSubscription) Close() {
501 s.once.Do(func() {
502 s.mu.Lock()
503 s.closed = true
504 s.mu.Unlock()
505 })
506}
507
508func (c *fakeCore) SubscribePreview(ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
509 return newFakePreviewSubscription(), nil
510}
511
512func (c *fakeCore) OpenEnvelopeStream(_ context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
513 c.mu.Lock()
514 defer c.mu.Unlock()
515 if c.openErr != nil {
516 return nil, c.openErr
517 }
518 if c.closed {
519 return nil, ports.BrokerAdmissionClosed
520 }
521 conn := newFakeLogicalConn()
522 c.streams[request.Stream] = conn
523 c.opens = append(c.opens, request)
524 return conn, nil
525}
526
527func (c *fakeCore) CloseStream(connection ports.BrokerConnectionID, stream ports.BrokerStreamID) error {
528 if connection != c.id {
529 return ports.BrokerAdmissionStale
530 }
531 c.mu.Lock()
532 conn, ok := c.streams[stream]
533 if ok {
534 delete(c.streams, stream)
535 }
536 c.closedIDs = append(c.closedIDs, stream)
537 c.mu.Unlock()
538 if !ok {
539 return ports.BrokerAdmissionStale
540 }
541 return conn.Close()
542}
543
544// mapMembershipError mirrors the admitted core's own mapping of a registry
545// refusal onto the closed broker error taxonomy. The real broker.Service already
546// returns these typed codes, so the fake applies the same rule: a caller of the
547// adapter must never see a bare registry sentinel as if it were an untyped
548// failure.
549func mapMembershipError(err error) error {
550 if err == nil {
551 return nil
552 }
553 code := ports.BrokerErrorCode(0)
554 switch {
555 case errors.Is(err, ports.ErrBrokerHostConflict):
556 code = ports.BrokerErrorHostConflict
557 case errors.Is(err, ports.ErrBrokerMembershipImmutable):
558 code = ports.BrokerErrorMembershipImmutable
559 default:
560 var unknown ports.BrokerStoreOutcomeUnknownError
561 var unknownPointer *ports.BrokerStoreOutcomeUnknownError
562 if errors.As(err, &unknown) || errors.As(err, &unknownPointer) {
563 code = ports.BrokerErrorOutcomeUnknown
564 }
565 }
566 if code == 0 {
567 return err
568 }
569 return ports.BrokerError{Code: code, Cause: err}
570}
571
572// AddHost records one delegated membership addition and returns its exact
573// result. The default is a fresh generation-1 registration for the requested
574// endpoint, exactly as the real registry mints one for a new endpoint.
575func (c *fakeCore) AddHost(ctx context.Context, endpoint string, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
576 if err := ctx.Err(); err != nil {
577 return domain.RemoteRegistration{}, err
578 }
579 c.mu.Lock()
580 defer c.mu.Unlock()
581 registration := domain.RemoteRegistration{}
582 var err error
583 switch {
584 case c.addErr != nil:
585 err = mapMembershipError(c.addErr)
586 c.addErr = nil
587 case c.addResult != nil:
588 registration = *c.addResult
589 default:
590 registration, err = domain.NewRemoteRegistration(endpoint, [16]byte{0x01})
591 }
592 c.added = append(c.added, addHostCall{Endpoint: endpoint, Policy: policy, Result: registration, Err: err})
593 return registration, err
594}
595
596// RemoveHost records one delegated exact-registration removal and reports
597// whether the host is present. The fake owns no membership state, so the
598// answer is the configured removeRemoved value for every exact registration.
599func (c *fakeCore) RemoveHost(ctx context.Context, expected domain.RemoteRegistration) (bool, error) {
600 if err := ctx.Err(); err != nil {
601 return false, err
602 }
603 c.mu.Lock()
604 defer c.mu.Unlock()
605 removed := false
606 var err error
607 if c.removeErr != nil {
608 err = mapMembershipError(c.removeErr)
609 c.removeErr = nil
610 } else {
611 removed = c.removeRemoved
612 }
613 c.removed = append(c.removed, removeHostCall{Expected: expected, Removed: removed, Err: err})
614 return removed, err
615}
616
617// UpdateHostPolicy records one delegated exact-registration policy update. The
618// default advances the generation by exactly one, mirroring the real registry's
619// replacement-policy path, and preserves the expected incarnation.
620func (c *fakeCore) UpdateHostPolicy(ctx context.Context, expected domain.RemoteRegistration, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
621 if err := ctx.Err(); err != nil {
622 return domain.RemoteRegistration{}, err
623 }
624 c.mu.Lock()
625 defer c.mu.Unlock()
626 registration := domain.RemoteRegistration{}
627 var err error
628 switch {
629 case c.updateErr != nil:
630 err = mapMembershipError(c.updateErr)
631 c.updateErr = nil
632 case c.updateResult != nil:
633 registration = *c.updateResult
634 default:
635 registration = expected
636 if registration.Generation != ^domain.RemoteGeneration(0) {
637 registration.Generation++
638 }
639 }
640 c.updated = append(c.updated, updateHostPolicyCall{Expected: expected, Policy: policy, Result: registration, Err: err})
641 return registration, err
642}
643
644// addCalls snapshots the delegated additions.
645func (c *fakeCore) addCalls() []addHostCall {
646 c.mu.Lock()
647 defer c.mu.Unlock()
648 return append([]addHostCall(nil), c.added...)
649}
650
651// removeCalls snapshots the delegated removals.
652func (c *fakeCore) removeCalls() []removeHostCall {
653 c.mu.Lock()
654 defer c.mu.Unlock()
655 return append([]removeHostCall(nil), c.removed...)
656}
657
658// updateCalls snapshots the delegated policy updates.
659func (c *fakeCore) updateCalls() []updateHostPolicyCall {
660 c.mu.Lock()
661 defer c.mu.Unlock()
662 return append([]updateHostPolicyCall(nil), c.updated...)
663}
664
665// refuseAddHost makes the next delegated addition fail with err. The refusal is
666// one-shot: the fake owns no membership state, so a later call must be free to
667// succeed rather than pin the whole connection as immutable forever.
668func (c *fakeCore) refuseAddHost(err error) { c.mu.Lock(); c.addErr = err; c.mu.Unlock() }
669
670// refuseRemoveHost makes the next delegated removal fail with err, one-shot.
671func (c *fakeCore) refuseRemoveHost(err error) { c.mu.Lock(); c.removeErr = err; c.mu.Unlock() }
672
673// reportRemoved configures the answer a delegated removal returns.
674func (c *fakeCore) reportRemoved(removed bool) { c.mu.Lock(); c.removeRemoved = removed; c.mu.Unlock() }
675
676// zeroAddRegistration makes a delegated addition succeed without returning any
677// registration authority, so a test can prove the server never publishes an
678// addition success that cannot be fenced.
679func (c *fakeCore) zeroAddRegistration() {
680 c.mu.Lock()
681 zero := domain.RemoteRegistration{}
682 c.addResult = &zero
683 c.mu.Unlock()
684}
685
686func (c *fakeCore) RequestReconcile(endpoint string) {
687 c.mu.Lock()
688 c.reconcile = append(c.reconcile, endpoint)
689 c.mu.Unlock()
690}
691
692func (c *fakeCore) Close() error {
693 c.once.Do(func() {
694 c.mu.Lock()
695 c.closed = true
696 conns := make([]*fakeLogicalConn, 0, len(c.streams))
697 for _, conn := range c.streams {
698 conns = append(conns, conn)
699 }
700 c.streams = make(map[ports.BrokerStreamID]*fakeLogicalConn)
701 c.mu.Unlock()
702 for _, conn := range conns {
703 _ = conn.Close()
704 }
705 close(c.done)
706 })
707 return nil
708}
709
710// logicalConn returns one admitted core connection by stream identity.
711func (c *fakeCore) logicalConn(stream ports.BrokerStreamID) *fakeLogicalConn {
712 c.mu.Lock()
713 defer c.mu.Unlock()
714 return c.streams[stream]
715}
716
717// newTestCore builds one admitted core service for a session test: a fresh hub
718// and a connection identity bound to epoch.
719func newTestCore(epoch ports.BrokerEpoch) *fakeCore {
720 var id ports.BrokerConnectionID
721 binary.BigEndian.PutUint64(id[:8], uint64(epoch))
722 id[15] = 1
723 return &fakeCore{
724 id: id,
725 hub: newSnapshotHub(ports.BrokerSnapshot{}),
726 streams: make(map[ports.BrokerStreamID]*fakeLogicalConn),
727 done: make(chan struct{}),
728 }
729}
730
731// fakeAuthority admits clients for one listener and hands out per-connection
732// core services.
733type fakeAuthority struct {
734 mu sync.Mutex
735 hub *snapshotHub
736 epoch ports.BrokerEpoch
737 next uint64
738 clients []*fakeCore
739 admitErr error
740}
741
742func newFakeAuthority(epoch ports.BrokerEpoch, snapshot ports.BrokerSnapshot) *fakeAuthority {
743 return &fakeAuthority{hub: newSnapshotHub(snapshot), epoch: epoch}
744}
745
746func (a *fakeAuthority) AdmitClient(ctx context.Context) (ports.BrokerCoreService, error) {
747 if err := ctx.Err(); err != nil {
748 return nil, err
749 }
750 a.mu.Lock()
751 defer a.mu.Unlock()
752 if a.admitErr != nil {
753 return nil, a.admitErr
754 }
755 a.next++
756 var id ports.BrokerConnectionID
757 binary.BigEndian.PutUint64(id[:8], uint64(a.epoch))
758 binary.BigEndian.PutUint64(id[8:], a.next)
759 core := &fakeCore{id: id, hub: a.hub, streams: make(map[ports.BrokerStreamID]*fakeLogicalConn), done: make(chan struct{})}
760 a.clients = append(a.clients, core)
761 return core, nil
762}
763
764func (a *fakeAuthority) last() *fakeCore {
765 a.mu.Lock()
766 defer a.mu.Unlock()
767 if len(a.clients) == 0 {
768 return nil
769 }
770 return a.clients[len(a.clients)-1]
771}
772
773func (a *fakeAuthority) publish(snapshot ports.BrokerSnapshot) { a.hub.publish(snapshot) }
774
775// endpoint is one running broker IPC listener with its authority.
776type endpoint struct {
777 t *testing.T
778 path string
779 listener ports.BrokerListener
780 authority *fakeAuthority
781 epoch ports.BrokerEpoch
782}
783
784func startEndpoint(t *testing.T, cfg Config) *endpoint {
785 t.Helper()
786 epoch := ports.BrokerEpoch(0x51)
787 authority := newFakeAuthority(epoch, ports.BrokerSnapshot{})
788 listener, err := Listen(testSocketPath(t), epoch, authority, cfg)
789 require.NoError(t, err)
790 t.Cleanup(func() { _ = listener.Close() })
791 return &endpoint{t: t, path: listener.Addr(), listener: listener, authority: authority, epoch: epoch}
792}
793
794// accept returns the next admitted session.
795func (e *endpoint) accept() ports.BrokerCoreService {
796 e.t.Helper()
797 type result struct {
798 service ports.BrokerCoreService
799 err error
800 }
801 out := make(chan result, 1)
802 go func() {
803 service, err := e.listener.Accept()
804 out <- result{service, err}
805 }()
806 select {
807 case got := <-out:
808 require.NoError(e.t, got.err)
809 return got.service
810 case <-time.After(5 * time.Second):
811 e.t.Fatal("listener.Accept did not return")
812 return nil
813 }
814}
815
816// dial connects one client adapter with the default test configuration.
817func (e *endpoint) dial() ports.BrokerService { return e.dialWith(Config{}) }
818
819// dialWith connects one client adapter with an explicit configuration.
820func (e *endpoint) dialWith(cfg Config) ports.BrokerService {
821 e.t.Helper()
822 cfg.HandshakeTimeout = 2 * time.Second
823 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
824 defer cancel()
825 client, err := Dial(ctx, e.path, cfg)
826 require.NoError(e.t, err)
827 e.t.Cleanup(func() { _ = client.Close() })
828 return client
829}
830
831// pair dials one client and accepts its session.
832func (e *endpoint) pair() (ports.BrokerService, ports.BrokerCoreService) {
833 e.t.Helper()
834 client := e.dial()
835 session := e.accept()
836 return client, session
837}
838
839// publish makes one snapshot current and wakes every subscriber.
840func (e *endpoint) publish(epoch ports.BrokerEpoch, revision ports.BrokerRevision) ports.BrokerSnapshot {
841 e.t.Helper()
842 snapshot := testSnapshot(epoch, revision)
843 e.authority.publish(snapshot)
844 return snapshot
845}
846
847// recordingTransport is one in-memory server carriage for a session: it decodes
848// and records every complete snapshot transfer the session writes and lets a
849// test wait for a generation to publish. It never yields an inbound frame, so
850// only the code under test drives publication.
851type recordingTransport struct {
852 mu sync.Mutex
853 ends map[brokerwire.SubscriptionGeneration]map[ports.BrokerRevision]int
854 changed chan struct{}
855 closedC chan struct{}
856 once sync.Once
857}
858
859func newRecordingTransport() *recordingTransport {
860 return &recordingTransport{
861 ends: make(map[brokerwire.SubscriptionGeneration]map[ports.BrokerRevision]int),
862 changed: make(chan struct{}, 1),
863 closedC: make(chan struct{}),
864 }
865}
866
867// Send decodes one server frame and records a completed snapshot transfer.
868func (t *recordingTransport) Send(envelope wire.Envelope) error {
869 message, err := brokerwire.DecodeServer(envelope.Payload, brokerwire.MaxBrokerEnvelopeBytes, brokerwire.MaxStreamChunkBytes)
870 if err != nil {
871 return err
872 }
873 part, ok := message.(brokerwire.SnapshotPart)
874 if !ok {
875 return nil
876 }
877 if _, ok := part.Part.(brokerwire.SnapshotEnd); !ok {
878 return nil
879 }
880 t.mu.Lock()
881 if t.ends[part.Generation] == nil {
882 t.ends[part.Generation] = make(map[ports.BrokerRevision]int)
883 }
884 t.ends[part.Generation][part.Revision]++
885 t.mu.Unlock()
886 select {
887 case t.changed <- struct{}{}:
888 default:
889 }
890 return nil
891}
892
893func (t *recordingTransport) RecvBounded(uint64) (wire.Envelope, error) {
894 <-t.closedC
895 return wire.Envelope{}, io.EOF
896}
897
898func (t *recordingTransport) Recv() (wire.Envelope, error) { return t.RecvBounded(0) }
899
900func (t *recordingTransport) Close() error {
901 t.once.Do(func() { close(t.closedC) })
902 return nil
903}
904
905// transferCount reports how many complete transfers of one generation committed
906// one revision.
907func (t *recordingTransport) transferCount(generation brokerwire.SubscriptionGeneration, revision ports.BrokerRevision) int {
908 t.mu.Lock()
909 defer t.mu.Unlock()
910 return t.ends[generation][revision]
911}
912
913// waitForTransfer blocks until generation committed revision at least once.
914func (t *recordingTransport) waitForTransfer(generation brokerwire.SubscriptionGeneration, revision ports.BrokerRevision, timeout time.Duration) bool {
915 return t.waitForTransferCount(generation, revision, 1, timeout)
916}
917
918// waitForTransferCount blocks until generation committed revision at least
919// minimum times, or the timeout elapses.
920func (t *recordingTransport) waitForTransferCount(generation brokerwire.SubscriptionGeneration, revision ports.BrokerRevision, minimum int, timeout time.Duration) bool {
921 deadline := time.NewTimer(timeout)
922 defer deadline.Stop()
923 for {
924 if t.transferCount(generation, revision) >= minimum {
925 return true
926 }
927 select {
928 case <-t.changed:
929 case <-deadline.C:
930 return t.transferCount(generation, revision) >= minimum
931 }
932 }
933}