1package brokeripc
2
3import (
4 "context"
5 "testing"
6 "time"
7
8 "github.com/bnema/vev/internal/adapters/brokerwire"
9 "github.com/bnema/vev/internal/adapters/ipc"
10 "github.com/bnema/vev/internal/ports"
11 "github.com/stretchr/testify/require"
12)
13
14func TestBuildRegistration(t *testing.T) {
15 for _, tc := range []struct {
16 name string
17 server, client string
18 livePeer bool
19 retire bool
20 stale string
21 }{
22 {name: "same build", server: "build-a", client: "build-a"},
23 {name: "sole different build", server: "build-a", client: "build-b", retire: true},
24 {name: "different build with live peer", server: "build-a", client: "build-b", livePeer: true, stale: "build-a"},
25 } {
26 t.Run(tc.name, func(t *testing.T) {
27 retired := make(chan struct{}, 1)
28 e := startEndpoint(t, Config{Build: tc.server, OnRetire: func() { retired <- struct{}{} }})
29 if tc.livePeer {
30 peer, err := Dial(context.Background(), e.path, Config{Build: tc.server})
31 require.NoError(t, err)
32 defer peer.Close()
33 }
34 service, err := Dial(context.Background(), e.path, Config{Build: tc.client})
35 if tc.retire {
36 require.ErrorIs(t, err, ErrBrokerRetired)
37 select {
38 case <-retired:
39 case <-time.After(5 * time.Second):
40 t.Fatal("retire callback not called")
41 }
42 _, err = Dial(context.Background(), e.path, Config{Build: tc.client})
43 require.Error(t, err)
44 return
45 }
46 require.NoError(t, err)
47 defer service.Close()
48 stale, ok := StaleBuild(service)
49 require.Equal(t, tc.stale != "", ok)
50 require.Equal(t, tc.stale, stale)
51 select {
52 case <-retired:
53 t.Fatal("unexpected retirement")
54 default:
55 }
56 })
57 }
58}
59
60func TestRetiringRefusesConcurrentAdmission(t *testing.T) {
61 const epoch = ports.BrokerEpoch(0x51)
62 path := testSocketPath(t)
63 gated := &gatedAuthority{
64 inner: newFakeAuthority(epoch, ports.BrokerSnapshot{}),
65 entered: make(chan struct{}), gate: make(chan struct{}),
66 }
67 bound, err := listen(path, epoch, gated, Config{Build: "build-a"}, ipc.SameUserPeerVerifier())
68 require.NoError(t, err)
69 defer bound.Close()
70 l := bound.(*listener)
71 raw := rawDial(t, &endpoint{t: t, path: path, epoch: epoch}, brokerwire.DefaultCeilings())
72 select {
73 case <-gated.entered:
74 case <-time.After(5 * time.Second):
75 t.Fatal("admission not reached")
76 }
77 // The registration decision and the admission gate share l.mu. A pending
78 // carriage prevents retirement; once retiring, the in-flight admission
79 // cannot publish a new session even if its authority returns success.
80 l.mu.Lock()
81 require.Len(t, l.pending, 1)
82 l.mu.Unlock()
83 // Commit retirement under the listener lock after admission has entered
84 // the authority, then release the authority gate.
85 l.mu.Lock()
86 l.retiring = true
87 l.mu.Unlock()
88 close(gated.gate)
89 result := make(chan error, 1)
90 go func() { _, err := raw.transport.RecvBounded(raw.ceilings.MaxReceiveEnvelopeBytes); result <- err }()
91 select {
92 case err := <-result:
93 require.Error(t, err)
94 case <-time.After(5 * time.Second):
95 t.Fatal("retiring listener admitted a pending session")
96 }
97}