1package brokeripc
2
3import (
4 "context"
5 "errors"
6 "runtime"
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/adapters/ipc"
15)
16
17// Fully bounded broker setup (P3.4 slice A). These tests drive the real AF_UNIX
18// carriage, never an in-memory replacement, so the only way to interrupt a peer
19// parked in framing I/O is the behavior under test: the client closing the
20// carriage when its single setup deadline expires, and the server enforcing the
21// accept-time registration budget on the pre-Register phase.
22//
23// The client half: one Dial context/deadline bounds the Unix dial, the preamble
24// exchange, the Register send, and the wait for Registered; a peer that stalls
25// after a successful preamble is interrupted by closing the carriage with every
26// setup worker joined; and a successful connection detaches from the setup
27// context after Registered. The server half: after ports.BrokerAuthority
28// admission a client must send Register within the same accept-time handshake
29// budget, or the session, its admitted core lease, and its listener slot are
30// released, while Listener.Close stays immediate.
31
32// silentBrokerPeer is one real AF_UNIX peer that completes the broker preamble
33// and then never answers Register, so a client's registration wait is observable
34// and interruptible without sleeping on a race.
35type silentBrokerPeer struct {
36 mux ipc.MuxListener
37 preambleSeen chan struct{}
38 release chan struct{}
39 done chan struct{}
40 once sync.Once
41}
42
43// startSilentBrokerPeer binds one real same-user endpoint whose only peer
44// completes the preamble and then stays silent.
45func startSilentBrokerPeer(t *testing.T, path string) *silentBrokerPeer {
46 t.Helper()
47 mux, err := ipc.ListenMux(path)
48 require.NoError(t, err)
49 peer := &silentBrokerPeer{
50 mux: mux,
51 preambleSeen: make(chan struct{}),
52 release: make(chan struct{}),
53 done: make(chan struct{}),
54 }
55 go func() {
56 defer close(peer.done)
57 conn, err := mux.Accept()
58 if err != nil {
59 return
60 }
61 defer func() { _ = conn.Close() }()
62 if _, err := runServerPreamble(context.Background(), conn, brokerwire.DefaultCeilings()); err != nil {
63 return
64 }
65 close(peer.preambleSeen)
66 <-peer.release
67 }()
68 t.Cleanup(peer.stop)
69 return peer
70}
71
72// stop releases the silent peer and joins it. It is idempotent.
73func (p *silentBrokerPeer) stop() {
74 p.once.Do(func() { close(p.release) })
75 _ = p.mux.Close()
76 <-p.done
77}
78
79// awaitPreamble fails unless the peer observed the client's preamble.
80func (p *silentBrokerPeer) awaitPreamble(t *testing.T) {
81 t.Helper()
82 select {
83 case <-p.preambleSeen:
84 case <-time.After(5 * time.Second):
85 t.Fatal("the peer never observed the broker preamble")
86 }
87}
88
89// TestClientDialSilentAfterPreambleIsBoundedAndJoined proves a peer that
90// completes the preamble and never answers Register cannot park Dial past the
91// configured handshake budget: Dial returns a deadline failure, the carriage was
92// closed to interrupt the blocked wait, and the setup workers were joined
93// instead of being abandoned.
94func TestClientDialSilentAfterPreambleIsBoundedAndJoined(t *testing.T) {
95 path := testSocketPath(t)
96 peer := startSilentBrokerPeer(t, path)
97
98 baseline := runtime.NumGoroutine()
99 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
100 defer cancel()
101 started := time.Now()
102 client, err := Dial(ctx, path, Config{HandshakeTimeout: 200 * time.Millisecond})
103 elapsed := time.Since(started)
104 require.ErrorIs(t, err, context.DeadlineExceeded)
105 require.Nil(t, client)
106 require.Less(t, elapsed, 5*time.Second, "the setup deadline must interrupt the silent registration wait")
107 peer.awaitPreamble(t)
108
109 peer.stop()
110 require.Eventually(t, func() bool { return runtime.NumGoroutine() <= baseline+2 }, 5*time.Second, 5*time.Millisecond,
111 "a bounded Dial must not abandon a setup goroutine (baseline=%d now=%d)", baseline, runtime.NumGoroutine())
112}
113
114// TestClientDialEverySetupStepSharesTheOneDeadline proves the caller's context
115// deadline, not only HandshakeTimeout, bounds the whole client setup: a deadline
116// far tighter than HandshakeTimeout still interrupts the registration wait.
117func TestClientDialEverySetupStepSharesTheOneDeadline(t *testing.T) {
118 path := testSocketPath(t)
119 peer := startSilentBrokerPeer(t, path)
120
121 ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond)
122 defer cancel()
123 started := time.Now()
124 client, err := Dial(ctx, path, Config{HandshakeTimeout: 30 * time.Second})
125 elapsed := time.Since(started)
126 require.ErrorIs(t, err, context.DeadlineExceeded)
127 require.Nil(t, client)
128 require.Less(t, elapsed, 5*time.Second, "the caller's deadline must bound the setup")
129 peer.awaitPreamble(t)
130 peer.stop()
131}
132
133// TestClientDialCancellationInterruptsRegistrationWait proves an explicit
134// cancellation, not just a deadline, interrupts the wait for Registered and is
135// reported as cancellation rather than being masked by a carriage failure.
136func TestClientDialCancellationInterruptsRegistrationWait(t *testing.T) {
137 path := testSocketPath(t)
138 peer := startSilentBrokerPeer(t, path)
139
140 ctx, cancel := context.WithCancel(context.Background())
141 defer cancel()
142 result := make(chan error, 1)
143 go func() {
144 _, err := Dial(ctx, path, Config{HandshakeTimeout: 30 * time.Second})
145 result <- err
146 }()
147 peer.awaitPreamble(t)
148 cancel()
149
150 select {
151 case err := <-result:
152 require.ErrorIs(t, err, context.Canceled, "cancellation must interrupt the registration wait")
153 case <-time.After(5 * time.Second):
154 t.Fatal("Dial must return on cancellation instead of waiting out HandshakeTimeout")
155 }
156 peer.stop()
157}
158
159// TestClientConnectionOutlivesSetupContext proves a successful connection
160// detaches from the setup context after Registered: once Dial returns, the
161// caller's setup deadline may elapse without disturbing the established
162// connection.
163func TestClientConnectionOutlivesSetupContext(t *testing.T) {
164 e := startEndpoint(t, Config{HandshakeTimeout: 5 * time.Second})
165
166 setupCtx, cancelSetup := context.WithTimeout(context.Background(), 200*time.Millisecond)
167 defer cancelSetup()
168 client, err := Dial(setupCtx, e.path, Config{HandshakeTimeout: 5 * time.Second})
169 require.NoError(t, err)
170 t.Cleanup(func() { _ = client.Close() })
171 session := e.accept()
172 require.Equal(t, session.ConnectionID(), client.ConnectionID())
173
174 // Let the setup deadline elapse; the connection was registered in time and
175 // must keep serving on its own connection-lived context.
176 time.Sleep(400 * time.Millisecond)
177 require.ErrorIs(t, setupCtx.Err(), context.DeadlineExceeded, "the setup context must have expired")
178
179 requireConnectionDelegates(t, client)
180}
181
182// TestServerRegistrationBudgetSettlesSilentPeer proves a same-user peer cannot
183// hold an admitted core lease or a listener slot indefinitely: after
184// ports.BrokerAuthority admission the client must send Register within the
185// accept-time handshake budget, or the session closes the core service and
186// releases the slot, and the listener keeps serving.
187func TestServerRegistrationBudgetSettlesSilentPeer(t *testing.T) {
188 e := startEndpoint(t, Config{MaxClients: 1, HandshakeTimeout: 200 * time.Millisecond})
189 l := e.listener.(*listener)
190
191 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
192 defer cancel()
193 transport, err := ipc.DialMuxContext(ctx, e.path)
194 require.NoError(t, err)
195 t.Cleanup(func() { _ = transport.Close() })
196 _, err = runClientPreamble(ctx, transport, brokerwire.DefaultCeilings())
197 require.NoError(t, err)
198
199 session := e.accept()
200 core := e.authority.last()
201 require.NotNil(t, core)
202
203 require.Eventually(t, func() bool {
204 select {
205 case <-core.done:
206 return true
207 default:
208 return false
209 }
210 }, 5*time.Second, 5*time.Millisecond, "a silent unregistered peer must not hold the admitted core lease")
211 require.Eventually(t, func() bool { return l.liveSessions() == 0 }, 5*time.Second, 5*time.Millisecond,
212 "a settled session must leave the listener's live set")
213
214 // The timeout is the peer's departure, not a session failure: the settled
215 // session reports an orderly end even though its terminal cause still records
216 // the registration budget.
217 require.NoError(t, session.Close())
218 require.ErrorIs(t, session.(*serverSession).terminalErr(), ErrRegistrationTimeout)
219 require.ErrorIs(t, session.(*serverSession).terminalErr(), context.DeadlineExceeded)
220
221 // The released slot is reusable even at the bound of one: the listener
222 // admits a fresh same-user client after the timeout.
223 fresh := e.dial()
224 freshSession := e.accept()
225 require.Equal(t, fresh.ConnectionID(), freshSession.ConnectionID())
226
227 // A clean listener shutdown must not report a peer timeout as its own
228 // failure, for the settled session or for the live one it ends.
229 require.NoError(t, e.listener.Close())
230}
231
232// TestRegistrationDeadlineBoundaryMarker proves the pre-Register guard treats
233// the registered marker, not timer ordering, as the deadline boundary: a session
234// whose Register handling already began is never settled with
235// ErrRegistrationTimeout, while a session with no Register is. The deadline is
236// already in the past so the guard observes the boundary deterministically.
237func TestRegistrationDeadlineBoundaryMarker(t *testing.T) {
238 tests := []struct {
239 name string
240 registered bool
241 wantTimeout bool
242 }{
243 {name: "register handling begun at the deadline", registered: true},
244 {name: "no register by the deadline", registered: false, wantTimeout: true},
245 }
246 for _, tc := range tests {
247 t.Run(tc.name, func(t *testing.T) {
248 core := newTestCore(0x51)
249 session, err := newServerSession(0x51, newRecordingTransport(), brokerwire.DefaultCeilings(), core, Config{}, func() {}, time.Now().Add(-time.Second))
250 require.NoError(t, err)
251 session.registered.Store(tc.registered)
252
253 stop := session.armRegistrationDeadline()
254 if tc.wantTimeout {
255 require.Eventually(t, func() bool { return errors.Is(session.terminalErr(), ErrRegistrationTimeout) }, time.Second, time.Millisecond)
256 }
257 stop()
258
259 if tc.wantTimeout {
260 require.ErrorIs(t, session.terminalErr(), ErrRegistrationTimeout)
261 require.ErrorIs(t, session.terminalErr(), context.DeadlineExceeded)
262 return
263 }
264 require.NoError(t, session.terminalErr(), "a Register already being handled must not be settled as a peer timeout")
265 })
266 }
267}
268
269// TestServerRegistrationBudgetLeavesRegisteredSessionHealthy proves the
270// registration budget bounds only the pre-Register phase: a session that
271// registered in time keeps serving long after the accept-time deadline has
272// passed.
273func TestServerRegistrationBudgetLeavesRegisteredSessionHealthy(t *testing.T) {
274 e := startEndpoint(t, Config{HandshakeTimeout: 150 * time.Millisecond})
275 client, _ := e.pair()
276 core := e.authority.last()
277
278 // Outlive the accept-time registration budget.
279 time.Sleep(400 * time.Millisecond)
280 requireConnectionDelegates(t, client)
281 select {
282 case <-core.done:
283 t.Fatal("a session that registered in time must not be settled by the registration budget")
284 default:
285 }
286}
287
288// TestListenerCloseDuringRegistrationWindowIsImmediate proves Close remains
289// immediate for a session that is inside its registration window with a
290// deliberately long budget: it does not wait out HandshakeTimeout, and it
291// releases the admitted core service and the slot.
292func TestListenerCloseDuringRegistrationWindowIsImmediate(t *testing.T) {
293 e := startEndpoint(t, Config{MaxClients: 2, HandshakeTimeout: 30 * time.Second})
294 l := e.listener.(*listener)
295
296 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
297 defer cancel()
298 transport, err := ipc.DialMuxContext(ctx, e.path)
299 require.NoError(t, err)
300 t.Cleanup(func() { _ = transport.Close() })
301 _, err = runClientPreamble(ctx, transport, brokerwire.DefaultCeilings())
302 require.NoError(t, err)
303
304 _ = e.accept()
305 core := e.authority.last()
306 require.NotNil(t, core)
307
308 closed := make(chan error, 1)
309 go func() { closed <- e.listener.Close() }()
310 select {
311 case err := <-closed:
312 require.NoError(t, err)
313 case <-time.After(5 * time.Second):
314 t.Fatal("Close must not wait out the registration budget for a silent session")
315 }
316 select {
317 case <-core.done:
318 case <-time.After(5 * time.Second):
319 t.Fatal("Close must end the admitted core service")
320 }
321 require.Zero(t, l.liveSessions())
322 require.Eventually(t, func() bool { return len(l.slots) == 0 }, 5*time.Second, 5*time.Millisecond,
323 "Close must release the client slot")
324}