1// Integration coverage for the real private Unix daemonmux carriage (P3.2d).
2//
3// These tests exercise the daemonmux multiplexer over an actual AF_UNIX socket
4// bound by ipc.ListenMux and dialed by ipc.DialMuxContext, not over the
5// in-memory carrier used by the unit tests. The physical connection is the one
6// real private mux carriage; every logical stream rides it. The full daemon
7// path (EndpointConnector -> ServerSupervisor -> AggregateListener -> typed
8// Listener) is exercised for the two-then-hundred admissions test, and the raw
9// pump path is exercised for the blocked-consumer and reset-isolation checks so
10// the stream accounting is asserted directly.
11package daemonmux
12
13import (
14 "context"
15 "os"
16 "path/filepath"
17 "sync"
18 "testing"
19 "time"
20
21 "github.com/stretchr/testify/require"
22
23 "github.com/bnema/vev/internal/adapters/ipc"
24 "github.com/bnema/vev/internal/domain"
25 "github.com/bnema/vev/internal/ports"
26 "github.com/bnema/vev/internal/protocol"
27 "github.com/bnema/vev/internal/protocol/wire"
28)
29
30// muxCarriagePath returns an owner-only temporary directory and one mux.sock
31// path inside it. The parent is created 0700 (os.MkdirTemp's default) and the
32// path stays well inside the cross-platform AF_UNIX limit.
33func muxCarriagePath(t *testing.T) string {
34 t.Helper()
35 root, err := os.MkdirTemp("/tmp", "m")
36 require.NoError(t, err)
37 t.Cleanup(func() { _ = os.RemoveAll(root) })
38 return filepath.Join(root, "mux.sock")
39}
40
41// unixRawCarriagePair binds a real private mux carriage at a temporary path and
42// returns its two ends as raw framed transports, joined by an actual AF_UNIX
43// SOCK_STREAM connection.
44func unixRawCarriagePair(t *testing.T) (RawFramedTransport, RawFramedTransport) {
45 t.Helper()
46 path := muxCarriagePath(t)
47 listener, err := ipc.ListenMux(path)
48 require.NoError(t, err)
49 t.Cleanup(func() { _ = listener.Close() })
50
51 type acceptOutcome struct {
52 raw RawFramedTransport
53 err error
54 }
55 accepted := make(chan acceptOutcome, 1)
56 go func() {
57 raw, acceptErr := listener.Accept()
58 if acceptErr != nil {
59 accepted <- acceptOutcome{err: acceptErr}
60 return
61 }
62 accepted <- acceptOutcome{raw: raw}
63 }()
64
65 ctx, cancel := context.WithTimeout(context.Background(), p3cTestDeadline)
66 defer cancel()
67 dialed, err := ipc.DialMuxContext(ctx, path)
68 require.NoError(t, err)
69
70 outcome := <-accepted
71 require.NoError(t, outcome.err)
72
73 client, server := dialed, outcome.raw
74 t.Cleanup(func() {
75 _ = client.Close()
76 _ = server.Close()
77 })
78 return client, server
79}
80
81// unixDialer returns a RawCarrierDialer that establishes the real mux carriage
82// at path through ipc.DialMuxContext (which also verifies same-user peer
83// credentials).
84func unixDialer(path string) RawCarrierDialer {
85 return func(ctx context.Context, _ ports.BrokerDialTarget) (RawFramedTransport, error) {
86 raw, err := ipc.DialMuxContext(ctx, path)
87 if err != nil {
88 return nil, err
89 }
90 return raw, nil
91 }
92}
93
94// serveUnix adopts every accepted real mux carriage into the test supervisor
95// until the listener closes.
96func (s *superviseServer) serveUnix(t *testing.T, listener ipc.MuxListener) {
97 t.Helper()
98 ctx, cancel := context.WithCancel(context.Background())
99 t.Cleanup(cancel)
100 go func() {
101 for {
102 raw, err := listener.Accept()
103 if err != nil {
104 return
105 }
106 s.adopt(ctx, raw)
107 }
108 }()
109}
110
111// soleChildPump returns the pump of the supervisor's one owned physical child,
112// so a test can assert the accepted stream accounting of the real carriage.
113func (s *superviseServer) soleChildPump(t *testing.T) *Pump {
114 t.Helper()
115 s.supervisor.mu.Lock()
116 defer s.supervisor.mu.Unlock()
117 require.Len(t, s.supervisor.children, 1, "expected exactly one physical mux child")
118 for child := range s.supervisor.children {
119 return child.pump
120 }
121 return nil
122}
123
124// TestUnixCarriageSupervisorTwoThenHundredTypedAdmissions drives the full
125// daemon path over one real Unix physical connection: EndpointConnector dials
126// the private carriage, ServerSupervisor adopts it, the AggregateListener
127// delivers independent typed daemon admissions, and every stream's traffic is
128// isolated. Only one physical connection and one physical child exist for all
129// 102 streams.
130func TestUnixCarriageSupervisorTwoThenHundredTypedAdmissions(t *testing.T) {
131 binding := mustServerBinding(t)
132 policy := binding.Policy()
133 path := muxCarriagePath(t)
134
135 listener, err := ipc.ListenMux(path)
136 require.NoError(t, err)
137 t.Cleanup(func() { _ = listener.Close() })
138
139 server := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
140 server.serve()
141 server.serveUnix(t, listener)
142
143 connector, err := NewEndpointConnector(unixDialer(path), DefaultMuxCeilings())
144 require.NoError(t, err)
145 physical, err := connector.Connect(context.Background(), muxEndpoint(binding.Identity(), policy, path))
146 require.NoError(t, err)
147 t.Cleanup(func() { _ = physical.Close() })
148 require.NoError(t, server.awaitAdoption(t))
149
150 // Two independent typed admissions over the one physical carriage.
151 first, err := openTyped(physical, context.Background(), muxOpenRequest(1, policy))
152 require.NoError(t, err)
153 t.Cleanup(func() { _ = first.Close() })
154 second, err := openTyped(physical, context.Background(), muxOpenRequest(2, policy))
155 require.NoError(t, err)
156 t.Cleanup(func() { _ = second.Close() })
157 server.awaitAccepted(t, 2)
158 for _, stream := range []ports.BrokerLogicalConnection{first, second} {
159 require.NoError(t, stream.SendClient(protocol.Ping{}))
160 message, err := stream.ReceiveServer()
161 require.NoError(t, err)
162 require.Equal(t, protocol.Pong{}, message)
163 }
164
165 // One hundred more independently admitted typed streams, opened
166 // concurrently over the same physical carriage.
167 const extra = 100
168 connections := make([]ports.BrokerLogicalConnection, extra)
169 openErrs := make([]error, extra)
170 var wg sync.WaitGroup
171 for i := 0; i < extra; i++ {
172 wg.Add(1)
173 go func(i int) {
174 defer wg.Done()
175 connections[i], openErrs[i] = openTyped(physical, context.Background(), muxOpenRequest(uint64(3+i), policy))
176 }(i)
177 }
178 wg.Wait()
179 for i, openErr := range openErrs {
180 require.NoError(t, openErr, "stream %d", i+3)
181 }
182 for i, connection := range connections {
183 require.NotNil(t, connection, "stream %d", i+3)
184 t.Cleanup(func() { _ = connection.Close() })
185 }
186
187 server.awaitAccepted(t, 2+extra)
188 for i, connection := range connections {
189 require.NoError(t, connection.SendClient(protocol.Ping{}), "stream %d", i+3)
190 message, err := connection.ReceiveServer()
191 require.NoError(t, err, "stream %d", i+3)
192 require.Equal(t, protocol.Pong{}, message)
193 }
194
195 // All 102 streams share exactly one physical child and one physical
196 // connection, and the daemon engine accounts for every admitted stream.
197 require.Equal(t, 1, server.supervisor.Children(), "all streams must share one physical Unix connection")
198 pump := server.soleChildPump(t)
199 require.Eventually(t, func() bool { return pump.Engine().Live() == 2+extra }, p3cTestDeadline, time.Millisecond,
200 "daemon engine admitted %d of %d streams", pump.Engine().Live(), 2+extra)
201
202 all := make([]ports.BrokerLogicalConnection, 0, 2+extra)
203 all = append(all, first, second)
204 all = append(all, connections...)
205 for _, connection := range all {
206 require.NoError(t, connection.Close())
207 }
208 require.Eventually(t, func() bool { return pump.Engine().Live() == 0 }, p3cTestDeadline, time.Millisecond,
209 "daemon engine still holds %d live streams after close", pump.Engine().Live())
210 require.False(t, channelClosed(physical.Done()), "closing every stream must leave the physical carriage open")
211}
212
213// unixDaemonPump builds a daemon-side pump over the server end of a real mux
214// carriage and returns the raw client end for direct frame injection, so a test
215// can drive deterministic inbound traffic over the actual socket.
216func unixDaemonPump(t *testing.T) (*Pump, RawFramedTransport) {
217 t.Helper()
218 client, server := unixRawCarriagePair(t)
219 carrier, err := NewPreambleCarrier(server)
220 require.NoError(t, err)
221 require.NoError(t, carrier.Negotiate(DefaultMuxCeilings()))
222 pump, err := NewPump(carrier, DirectionClient, DefaultMuxCeilings())
223 require.NoError(t, err)
224 t.Cleanup(func() { _ = pump.Close() })
225 pump.Start(context.Background())
226 return pump, client
227}
228
229// injectUnixClientFrame writes one client frame over the real Unix carriage.
230func injectUnixClientFrame(t *testing.T, client RawFramedTransport, message ClientMessage) {
231 t.Helper()
232 require.NoError(t, client.Send(wire.Envelope{Payload: mustEncodeClient(t, message)}))
233}
234
235// nextUnixServerFrame reads one daemon frame from the real Unix carriage.
236func nextUnixServerFrame(t *testing.T, client RawFramedTransport) []byte {
237 t.Helper()
238 envelope, err := client.RecvBounded(testEnvelopeCeiling)
239 require.NoError(t, err)
240 return envelope.Payload
241}
242
243// TestUnixCarriageBlockedConsumerSiblingProgress proves, over a real Unix
244// carriage, that a logical consumer which never drains one stream's inbound
245// queue neither blocks the reader nor a sibling: the stalled stream is reset by
246// its own bound while the sibling keeps receiving.
247func TestUnixCarriageBlockedConsumerSiblingProgress(t *testing.T) {
248 pump, client := unixDaemonPump(t)
249
250 const siblings = 2
251 for i := 1; i <= siblings; i++ {
252 injectUnixClientFrame(t, client, openFor(PhysicalStreamID(i)))
253 }
254 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == siblings })
255 for i := 1; i <= siblings; i++ {
256 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(PhysicalStreamID(i))}))
257 }
258
259 // A peer ignores flow control and fills stream 1's whole window while the
260 // consumer never Takes it.
261 chunk, full := floodWindow(pump.Ceilings())
262 for i := 0; i < full; i++ {
263 injectUnixClientFrame(t, client, Data{Physical: 1, Data: chunk})
264 }
265 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
266 status, ok := e.Status(1)
267 return ok && status.QueuedChunks == full
268 })
269
270 // One more chunk exceeds the granted window; a sibling chunk still lands,
271 // so neither the reader nor the sibling was blocked.
272 injectUnixClientFrame(t, client, Data{Physical: 1, Data: chunk})
273 injectUnixClientFrame(t, client, Data{Physical: 2, Data: []byte("sibling")})
274
275 taken, ok := takeEventually(t, pump, 2)
276 require.True(t, ok)
277 require.Equal(t, []byte("sibling"), taken)
278
279 stalled := mustStatus(t, pump.Engine(), 1)
280 require.Equal(t, StreamTerminal, stalled.State)
281 require.ErrorIs(t, stalled.Err, ErrStreamQueueFull)
282 require.Equal(t, domain.RemoteFailureInvalidResponse, stalled.FailureKind)
283 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 2).State)
284 require.False(t, pump.Engine().Closed())
285 require.False(t, channelClosed(pump.Done()))
286
287 reset := decodeReset(t, nextUnixServerFrame(t, client), DirectionServer)
288 require.Equal(t, PhysicalStreamID(1), reset.Physical)
289 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.Error.FailureKind)
290}
291
292// TestUnixCarriageResetIsolation proves, over a real Unix carriage, that one
293// stream-local refusal schedules exactly one Reset for its own stream and
294// leaves the physical connection and every sibling healthy.
295func TestUnixCarriageResetIsolation(t *testing.T) {
296 pump, client := unixDaemonPump(t)
297
298 for _, id := range []PhysicalStreamID{1, 2, 3} {
299 injectUnixClientFrame(t, client, openFor(id))
300 }
301 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 3 })
302 for _, id := range []PhysicalStreamID{2, 3} {
303 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
304 }
305
306 // Stream 1 is still opening: inbound data is a stream-local state refusal.
307 injectUnixClientFrame(t, client, Data{Physical: 1, Data: []byte("early")})
308 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
309 status, ok := e.Status(1)
310 return ok && status.State == StreamTerminal
311 })
312 settled := mustStatus(t, pump.Engine(), 1)
313 require.ErrorContains(t, settled.Err, "invalid stream state")
314 require.Equal(t, domain.RemoteFailureInvalidResponse, settled.FailureKind)
315
316 injectUnixClientFrame(t, client, Data{Physical: 2, Data: []byte("sibling")})
317 taken, ok := takeEventually(t, pump, 2)
318 require.True(t, ok)
319 require.Equal(t, []byte("sibling"), taken)
320
321 reset := decodeReset(t, nextUnixServerFrame(t, client), DirectionServer)
322 require.Equal(t, PhysicalStreamID(1), reset.Physical)
323 require.True(t, reset.HasError)
324 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.Error.FailureKind)
325 require.False(t, pump.Engine().Closed())
326 require.False(t, channelClosed(pump.Done()))
327}