1package client
2
3import (
4 "context"
5 "encoding/binary"
6 "errors"
7 "os"
8 "path/filepath"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/require"
14
15 "github.com/bnema/vev/internal/adapters/brokeripc"
16 "github.com/bnema/vev/internal/domain"
17 "github.com/bnema/vev/internal/ports"
18)
19
20// Real connector integration (Plan 001 P5.1a, GO-007). These tests wire the
21// production brokeripc Connector into the autonomous supervisor over a real
22// AF_UNIX broker endpoint, so the readiness handshake, the established-
23// connection loss, and the cancellation lifetime are pinned against the true
24// adapter and carriage rather than a scripted fake.
25
26const integrationEpoch = ports.BrokerEpoch(0x71)
27
28// integrationCore is the minimal admitted broker service one accepted
29// connection is bound to: a fixed-epoch publication and a coalescing
30// subscription, which is all the supervisor's readiness handshake needs.
31type integrationCore struct {
32 id ports.BrokerConnectionID
33 epoch ports.BrokerEpoch
34 changed chan struct{}
35 done chan struct{}
36 closed chan struct{}
37 once sync.Once
38
39 mu sync.Mutex
40 nextStream ports.BrokerStreamID
41}
42
43func (c *integrationCore) ConnectionID() ports.BrokerConnectionID { return c.id }
44func (c *integrationCore) Done() <-chan struct{} { return c.done }
45func (c *integrationCore) Err() error { return nil }
46
47// NextStreamID mirrors the production allocator so the supervisor's single
48// allocation seam is exercised over the real carriage.
49func (c *integrationCore) NextStreamID() (ports.BrokerStreamID, error) {
50 c.mu.Lock()
51 defer c.mu.Unlock()
52 c.nextStream++
53 return c.nextStream, nil
54}
55
56func (c *integrationCore) Snapshot() ports.BrokerSnapshot {
57 return ports.BrokerSnapshot{Epoch: c.epoch, Revision: 1}
58}
59
60func (c *integrationCore) Subscribe() (ports.BrokerSubscription, error) {
61 return integrationSubscription{changed: c.changed}, nil
62}
63
64func (c *integrationCore) SubscribePreview(ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
65 return nil, errors.New("integration core does not support preview")
66}
67
68func (c *integrationCore) OpenEnvelopeStream(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
69 return nil, errors.New("integration core does not support streams")
70}
71
72func (c *integrationCore) CloseStream(ports.BrokerConnectionID, ports.BrokerStreamID) error {
73 return nil
74}
75func (c *integrationCore) AddHost(context.Context, string, ports.BrokerPolicy) (domain.RemoteRegistration, error) {
76 return domain.RemoteRegistration{}, nil
77}
78func (c *integrationCore) RemoveHost(context.Context, domain.RemoteRegistration) (bool, error) {
79 return false, nil
80}
81func (c *integrationCore) UpdateHostPolicy(context.Context, domain.RemoteRegistration, ports.BrokerPolicy) (domain.RemoteRegistration, error) {
82 return domain.RemoteRegistration{}, nil
83}
84func (c *integrationCore) RequestReconcile(string) {}
85
86func (c *integrationCore) Close() error {
87 c.once.Do(func() {
88 close(c.done)
89 close(c.closed)
90 })
91 return nil
92}
93
94// integrationSubscription is one idle coalescing subscription.
95type integrationSubscription struct{ changed chan struct{} }
96
97func (s integrationSubscription) Changed() <-chan struct{} { return s.changed }
98func (s integrationSubscription) Close() {}
99
100// integrationAuthority admits one connection per accepted carriage and records
101// every admitted core, so a test can observe the broker-side end of the
102// supervisor's connection lifetime.
103type integrationAuthority struct {
104 epoch ports.BrokerEpoch
105
106 mu sync.Mutex
107 next uint64
108 cores []*integrationCore
109}
110
111var _ ports.BrokerAuthority = (*integrationAuthority)(nil)
112
113func (a *integrationAuthority) AdmitClient(ctx context.Context) (ports.BrokerCoreService, error) {
114 if err := ctx.Err(); err != nil {
115 return nil, err
116 }
117 a.mu.Lock()
118 a.next++
119 sequence := a.next
120 a.mu.Unlock()
121 var id ports.BrokerConnectionID
122 binary.BigEndian.PutUint64(id[:8], uint64(a.epoch))
123 binary.BigEndian.PutUint64(id[8:], sequence)
124 core := &integrationCore{
125 id: id,
126 epoch: a.epoch,
127 changed: make(chan struct{}),
128 done: make(chan struct{}),
129 closed: make(chan struct{}),
130 }
131 a.mu.Lock()
132 a.cores = append(a.cores, core)
133 a.mu.Unlock()
134 return core, nil
135}
136
137func (a *integrationAuthority) lastCore() *integrationCore {
138 a.mu.Lock()
139 defer a.mu.Unlock()
140 if len(a.cores) == 0 {
141 return nil
142 }
143 return a.cores[len(a.cores)-1]
144}
145
146// startIntegrationBroker binds one real broker IPC endpoint in a short private
147// directory, so the AF_UNIX path stays well within the platform limit.
148func startIntegrationBroker(t *testing.T) (*integrationAuthority, ports.BrokerListener, string) {
149 t.Helper()
150 root, err := os.MkdirTemp("/tmp", "v")
151 require.NoError(t, err)
152 t.Cleanup(func() { _ = os.RemoveAll(root) })
153 path := brokeripc.SocketPath(filepath.Join(root, "vev"))
154 authority := &integrationAuthority{epoch: integrationEpoch}
155 listener, err := brokeripc.Listen(path, integrationEpoch, authority, brokeripc.Config{})
156 require.NoError(t, err)
157 t.Cleanup(func() { _ = listener.Close() })
158 return authority, listener, path
159}
160
161// integrationSupervisor builds one autonomous supervisor over the production
162// connector and a pinned clock with a controlled terminal.
163func integrationSupervisor(t *testing.T, path string) (*Supervisor, *supervisorTestClock, *supervisorTestTerminal) {
164 t.Helper()
165 clock := newSupervisorTestClock()
166 reader := newSupervisorTestReader()
167 t.Cleanup(reader.unblock)
168 terminal := newSupervisorTestTerminal(reader)
169 sup := mustSupervisor(t, SupervisorConfig{
170 Connector: brokeripc.NewConnector(path, brokeripc.Config{}).Navigator(),
171 Terminal: terminal,
172 Clock: clock,
173 Jitter: func() float64 { return 0 },
174 })
175 return sup, clock, terminal
176}
177
178// TestSupervisorRealConnectorReadinessAndLoss pins the readiness handshake and
179// the established-connection loss against a real broker endpoint: the
180// supervisor becomes Ready only after the broker publishes the admitted
181// connection's epoch, and a retired connection is observed through Done/Err and
182// retried with a typed unavailable cause.
183func TestSupervisorRealConnectorReadinessAndLoss(t *testing.T) {
184 _, listener, path := startIntegrationBroker(t)
185 sup, _, terminal := integrationSupervisor(t, path)
186
187 ctx, cancel := context.WithCancel(context.Background())
188 defer cancel()
189 runErr := make(chan error, 1)
190 go func() { runErr <- sup.Run(ctx) }()
191
192 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
193 require.Equal(t, uint64(1), sup.State().Generation)
194 require.Equal(t, uint64(0), sup.State().Attempt)
195 require.NoError(t, sup.State().Err)
196
197 // The broker retires the accepted connection; the supervisor observes the
198 // carriage loss through Done/Err and waits the cadence before retrying.
199 server, err := listener.Accept()
200 require.NoError(t, err)
201 require.NoError(t, server.Close())
202
203 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityRetryWait }, 5*time.Second, time.Millisecond)
204 var typed ports.BrokerError
205 require.ErrorAs(t, sup.State().Err, &typed)
206 require.Equal(t, ports.BrokerErrorUnavailable, typed.Code)
207
208 cancel()
209 require.ErrorIs(t, <-runErr, context.Canceled)
210 require.Equal(t, 1, terminal.restoreCount())
211}
212
213// TestSupervisorRealConnectorCancelClosesConnection pins the cancellation
214// lifetime: cancelling the run closes the supervisor's adopted connection, ends
215// the broker-side session, and retires the admitted core exactly once.
216func TestSupervisorRealConnectorCancelClosesConnection(t *testing.T) {
217 authority, _, path := startIntegrationBroker(t)
218 sup, _, terminal := integrationSupervisor(t, path)
219
220 ctx, cancel := context.WithCancel(context.Background())
221 runErr := make(chan error, 1)
222 go func() { runErr <- sup.Run(ctx) }()
223
224 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
225 core := authority.lastCore()
226 require.NotNil(t, core, "the broker must have admitted the connection")
227
228 cancel()
229 require.ErrorIs(t, <-runErr, context.Canceled)
230 require.Equal(t, 1, terminal.restoreCount())
231
232 select {
233 case <-core.closed:
234 case <-time.After(5 * time.Second):
235 t.Fatal("cancelling the supervisor must close the broker connection")
236 }
237}