1// Integration coverage for the real private QUIC daemonmux carriage (P3.2e).
2//
3// These tests exercise the daemonmux multiplexer over an actual loopback QUIC
4// connection authenticated by quic.NewServer's ephemeral certificate plus
5// one-time token/nonce and dialed by quic.DialMuxContext, not over the
6// in-memory carrier used by the unit tests or the Unix socket of P3.2d. One
7// QUIC connection is one physical daemonmux connection: every logical stream
8// rides its single bidirectional stream, and no logical attachment is ever
9// mapped to a native QUIC stream.
10//
11// The full daemon path (EndpointConnector -> ServerSupervisor ->
12// AggregateListener -> typed Listener) is exercised for the two-then-hundred
13// admissions and physical-failure tests, and the raw pump path is exercised for
14// the blocked-consumer and reset-isolation checks so the stream accounting is
15// asserted directly.
16package daemonmux
17
18import (
19 "context"
20 "net"
21 "strconv"
22 "sync"
23 "testing"
24 "time"
25
26 "github.com/stretchr/testify/require"
27
28 "github.com/bnema/vev/internal/adapters/quic"
29 "github.com/bnema/vev/internal/domain"
30 "github.com/bnema/vev/internal/ports"
31 "github.com/bnema/vev/internal/protocol"
32 "github.com/bnema/vev/internal/protocol/wire"
33)
34
35// quicCarriageDeadline bounds every real loopback QUIC mux carriage wait. QUIC
36// pays a TLS handshake and a pinned dial per physical carriage, so the parity
37// waits stay deliberately generous.
38const quicCarriageDeadline = 10 * time.Second
39
40// quicCarriageEndpoint mints one ephemeral bootstrap endpoint and returns it
41// with its readiness record and the dial address composed from loopback plus
42// the readiness port. The one-time token and nonce stay in the readiness
43// record: daemonmux only ever sees the host-independent address.
44func quicCarriageEndpoint(t *testing.T) (*quic.Server, quic.Readiness, string) {
45 t.Helper()
46 server, readiness, err := quic.NewServer()
47 require.NoError(t, err)
48 t.Cleanup(func() { _ = server.Close() })
49 return server, readiness, net.JoinHostPort("127.0.0.1", strconv.Itoa(readiness.Port))
50}
51
52// quicDialer returns a RawCarrierDialer that establishes the real QUIC mux
53// carriage through quic.DialMuxContext - exact ephemeral certificate pin plus
54// single-use token/nonce - and hands the raw bounded transport to the connector
55// unmodified.
56func quicDialer(readiness quic.Readiness) RawCarrierDialer {
57 return func(ctx context.Context, target ports.BrokerDialTarget) (RawFramedTransport, error) {
58 raw, err := quic.DialMuxContext(ctx, target.Address, readiness, quic.Config{}, quicCarriageDeadline)
59 if err != nil {
60 return nil, err
61 }
62 return raw, nil
63 }
64}
65
66// quicDialCarriage establishes the broker end of one real QUIC mux carriage.
67func quicDialCarriage(t *testing.T, ctx context.Context, addr string, readiness quic.Readiness) RawFramedTransport {
68 t.Helper()
69 raw, err := quicDialer(readiness)(ctx, ports.BrokerDialTarget{Address: addr})
70 require.NoError(t, err)
71 t.Cleanup(func() { _ = raw.Close() })
72 return raw
73}
74
75// serveQUIC adopts every carriage the bootstrap endpoint admits into the test
76// supervisor until the endpoint closes. There is exactly one: the credential is
77// one-time.
78func (s *superviseServer) serveQUIC(t *testing.T, server *quic.Server) {
79 t.Helper()
80 ctx, cancel := context.WithCancel(context.Background())
81 t.Cleanup(cancel)
82 go func() {
83 for {
84 raw, err := server.AcceptMux(ctx)
85 if err != nil {
86 return
87 }
88 s.adopt(ctx, raw)
89 }
90 }()
91}
92
93// serveQUICCapturing adopts the bootstrap endpoint's one authenticated carriage
94// and publishes the accepted daemon carriage, so a test can fail that physical
95// connection deterministically.
96func (s *superviseServer) serveQUICCapturing(t *testing.T, server *quic.Server) <-chan RawFramedTransport {
97 t.Helper()
98 ctx, cancel := context.WithCancel(context.Background())
99 t.Cleanup(cancel)
100 captured := make(chan RawFramedTransport, 1)
101 go func() {
102 raw, err := server.AcceptMux(ctx)
103 if err != nil {
104 return
105 }
106 captured <- raw
107 s.adopt(ctx, raw)
108 }()
109 return captured
110}
111
112// awaitQUICAccepted waits for the aggregate to deliver count typed admissions.
113func (s *superviseServer) awaitQUICAccepted(t *testing.T, count int) {
114 t.Helper()
115 require.Eventually(t, func() bool { return s.acceptedCount() >= count }, quicCarriageDeadline, time.Millisecond,
116 "server accepted %d of %d streams", s.acceptedCount(), count)
117}
118
119// quicDaemonPump builds a daemon-side pump over the daemon end of a real QUIC
120// mux carriage and returns the raw broker end for direct frame injection, so a
121// test can drive deterministic inbound traffic over the actual QUIC stream.
122func quicDaemonPump(t *testing.T) (*Pump, RawFramedTransport) {
123 t.Helper()
124 server, readiness, addr := quicCarriageEndpoint(t)
125 ctx, cancel := context.WithTimeout(context.Background(), quicCarriageDeadline)
126 t.Cleanup(cancel)
127
128 type acceptOutcome struct {
129 raw RawFramedTransport
130 err error
131 }
132 accepted := make(chan acceptOutcome, 1)
133 go func() {
134 raw, err := server.AcceptMux(ctx)
135 accepted <- acceptOutcome{raw: raw, err: err}
136 }()
137
138 client := quicDialCarriage(t, ctx, addr, readiness)
139
140 var daemon RawFramedTransport
141 select {
142 case outcome := <-accepted:
143 require.NoError(t, outcome.err)
144 require.NotNil(t, outcome.raw)
145 daemon = outcome.raw
146 t.Cleanup(func() { _ = daemon.Close() })
147 case <-time.After(quicCarriageDeadline):
148 t.Fatal("AcceptMux did not return")
149 }
150
151 carrier, err := NewPreambleCarrier(daemon)
152 require.NoError(t, err)
153 require.NoError(t, carrier.Negotiate(DefaultMuxCeilings()))
154 pump, err := NewPump(carrier, DirectionClient, DefaultMuxCeilings())
155 require.NoError(t, err)
156 t.Cleanup(func() { _ = pump.Close() })
157 pump.Start(context.Background())
158 return pump, client
159}
160
161// injectQUICClientFrame writes one client frame over the real QUIC carriage.
162func injectQUICClientFrame(t *testing.T, client RawFramedTransport, message ClientMessage) {
163 t.Helper()
164 require.NoError(t, client.Send(wire.Envelope{Payload: mustEncodeClient(t, message)}))
165}
166
167// nextQUICServerFrame reads one daemon frame from the real QUIC carriage.
168func nextQUICServerFrame(t *testing.T, client RawFramedTransport) []byte {
169 t.Helper()
170 envelope, err := client.RecvBounded(testEnvelopeCeiling)
171 require.NoError(t, err)
172 return envelope.Payload
173}
174
175// TestQUICCarriageSupervisorTwoThenHundredTypedAdmissions drives the full daemon
176// path over one real QUIC physical connection: EndpointConnector dials the
177// authenticated carriage, ServerSupervisor adopts it, the AggregateListener
178// delivers independent typed daemon admissions, and every stream's traffic is
179// isolated. Only one physical connection, one physical child, and one QUIC
180// stream exist for all 102 streams.
181func TestQUICCarriageSupervisorTwoThenHundredTypedAdmissions(t *testing.T) {
182 binding := mustServerBinding(t)
183 policy := binding.Policy()
184 server, readiness, addr := quicCarriageEndpoint(t)
185
186 daemon := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
187 daemon.serve()
188 daemon.serveQUIC(t, server)
189
190 connector, err := NewEndpointConnector(quicDialer(readiness), DefaultMuxCeilings())
191 require.NoError(t, err)
192 physical, err := connector.Connect(context.Background(), muxEndpoint(binding.Identity(), policy, addr))
193 require.NoError(t, err)
194 t.Cleanup(func() { _ = physical.Close() })
195 require.NoError(t, daemon.awaitAdoption(t))
196
197 // Two independent typed admissions over the one physical carriage.
198 first, err := openTyped(physical, context.Background(), muxOpenRequest(1, policy))
199 require.NoError(t, err)
200 t.Cleanup(func() { _ = first.Close() })
201 second, err := openTyped(physical, context.Background(), muxOpenRequest(2, policy))
202 require.NoError(t, err)
203 t.Cleanup(func() { _ = second.Close() })
204 daemon.awaitQUICAccepted(t, 2)
205 for _, stream := range []ports.BrokerLogicalConnection{first, second} {
206 require.NoError(t, stream.SendClient(protocol.Ping{}))
207 message, err := stream.ReceiveServer()
208 require.NoError(t, err)
209 require.Equal(t, protocol.Pong{}, message)
210 }
211
212 // One hundred more independently admitted typed streams, opened
213 // concurrently over the same physical carriage.
214 const extra = 100
215 connections := make([]ports.BrokerLogicalConnection, extra)
216 openErrs := make([]error, extra)
217 var wg sync.WaitGroup
218 for i := 0; i < extra; i++ {
219 wg.Add(1)
220 go func(i int) {
221 defer wg.Done()
222 connections[i], openErrs[i] = openTyped(physical, context.Background(), muxOpenRequest(uint64(3+i), policy))
223 }(i)
224 }
225 wg.Wait()
226 for i, openErr := range openErrs {
227 require.NoError(t, openErr, "stream %d", i+3)
228 }
229 for i, connection := range connections {
230 require.NotNil(t, connection, "stream %d", i+3)
231 t.Cleanup(func() { _ = connection.Close() })
232 }
233
234 daemon.awaitQUICAccepted(t, 2+extra)
235 for i, connection := range connections {
236 require.NoError(t, connection.SendClient(protocol.Ping{}), "stream %d", i+3)
237 message, err := connection.ReceiveServer()
238 require.NoError(t, err, "stream %d", i+3)
239 require.Equal(t, protocol.Pong{}, message)
240 }
241
242 // All 102 streams share exactly one physical child and one physical QUIC
243 // connection, and the daemon engine accounts for every admitted stream.
244 require.Equal(t, 1, daemon.supervisor.Children(), "all streams must share one physical QUIC connection")
245 pump := daemon.soleChildPump(t)
246 require.Eventually(t, func() bool { return pump.Engine().Live() == 2+extra }, quicCarriageDeadline, time.Millisecond,
247 "daemon engine admitted %d of %d streams", pump.Engine().Live(), 2+extra)
248
249 all := make([]ports.BrokerLogicalConnection, 0, 2+extra)
250 all = append(all, first, second)
251 all = append(all, connections...)
252 for _, connection := range all {
253 require.NoError(t, connection.Close())
254 }
255 require.Eventually(t, func() bool { return pump.Engine().Live() == 0 }, quicCarriageDeadline, time.Millisecond,
256 "daemon engine still holds %d live streams after close", pump.Engine().Live())
257 require.False(t, channelClosed(physical.Done()), "closing every stream must leave the physical carriage open")
258}
259
260// TestQUICCarriageBlockedConsumerSiblingProgress proves, over a real QUIC
261// carriage, that a logical consumer which never drains one stream's inbound
262// queue neither blocks the reader nor a sibling: the stalled stream is reset by
263// its own bound while the sibling keeps receiving.
264func TestQUICCarriageBlockedConsumerSiblingProgress(t *testing.T) {
265 pump, client := quicDaemonPump(t)
266
267 const siblings = 2
268 for i := 1; i <= siblings; i++ {
269 injectQUICClientFrame(t, client, openFor(PhysicalStreamID(i)))
270 }
271 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == siblings })
272 for i := 1; i <= siblings; i++ {
273 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(PhysicalStreamID(i))}))
274 }
275
276 // A peer ignores flow control and fills stream 1's whole window while the
277 // consumer never Takes it.
278 chunk, full := floodWindow(pump.Ceilings())
279 for i := 0; i < full; i++ {
280 injectQUICClientFrame(t, client, Data{Physical: 1, Data: chunk})
281 }
282 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
283 status, ok := e.Status(1)
284 return ok && status.QueuedChunks == full
285 })
286
287 // One more chunk exceeds the granted window; a sibling chunk still lands,
288 // so neither the reader nor the sibling was blocked.
289 injectQUICClientFrame(t, client, Data{Physical: 1, Data: chunk})
290 injectQUICClientFrame(t, client, Data{Physical: 2, Data: []byte("sibling")})
291
292 taken, ok := takeEventually(t, pump, 2)
293 require.True(t, ok)
294 require.Equal(t, []byte("sibling"), taken)
295
296 stalled := mustStatus(t, pump.Engine(), 1)
297 require.Equal(t, StreamTerminal, stalled.State)
298 require.ErrorIs(t, stalled.Err, ErrStreamQueueFull)
299 require.Equal(t, domain.RemoteFailureInvalidResponse, stalled.FailureKind)
300 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 2).State)
301 require.False(t, pump.Engine().Closed())
302 require.False(t, channelClosed(pump.Done()))
303
304 reset := decodeReset(t, nextQUICServerFrame(t, client), DirectionServer)
305 require.Equal(t, PhysicalStreamID(1), reset.Physical)
306 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.Error.FailureKind)
307}
308
309// TestQUICCarriageResetIsolation proves, over a real QUIC carriage, that one
310// stream-local refusal schedules exactly one Reset for its own stream and
311// leaves the physical connection and every sibling healthy.
312func TestQUICCarriageResetIsolation(t *testing.T) {
313 pump, client := quicDaemonPump(t)
314
315 for _, id := range []PhysicalStreamID{1, 2, 3} {
316 injectQUICClientFrame(t, client, openFor(id))
317 }
318 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 3 })
319 for _, id := range []PhysicalStreamID{2, 3} {
320 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
321 }
322
323 // Stream 1 is still opening: inbound data is a stream-local state refusal.
324 injectQUICClientFrame(t, client, Data{Physical: 1, Data: []byte("early")})
325 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
326 status, ok := e.Status(1)
327 return ok && status.State == StreamTerminal
328 })
329 settled := mustStatus(t, pump.Engine(), 1)
330 require.ErrorContains(t, settled.Err, "invalid stream state")
331 require.Equal(t, domain.RemoteFailureInvalidResponse, settled.FailureKind)
332
333 injectQUICClientFrame(t, client, Data{Physical: 2, Data: []byte("sibling")})
334 taken, ok := takeEventually(t, pump, 2)
335 require.True(t, ok)
336 require.Equal(t, []byte("sibling"), taken)
337
338 reset := decodeReset(t, nextQUICServerFrame(t, client), DirectionServer)
339 require.Equal(t, PhysicalStreamID(1), reset.Physical)
340 require.True(t, reset.HasError)
341 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.Error.FailureKind)
342 require.False(t, pump.Engine().Closed())
343 require.False(t, channelClosed(pump.Done()))
344}
345
346// TestQUICCarriagePhysicalFailureFansOutAndIsIsolated proves a lost QUIC
347// physical carriage publishes its terminal outcome before fanning out to every
348// logical stream it carried, while a sibling physical QUIC connection on the
349// same supervisor is untouched and a fresh stream on the lost physical is
350// refused.
351func TestQUICCarriagePhysicalFailureFansOutAndIsIsolated(t *testing.T) {
352 binding := mustServerBinding(t)
353 policy := binding.Policy()
354 serverA, readinessA, addrA := quicCarriageEndpoint(t)
355 serverB, readinessB, addrB := quicCarriageEndpoint(t)
356
357 daemon := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
358 daemon.serve()
359 carriageA := daemon.serveQUICCapturing(t, serverA)
360 daemon.serveQUIC(t, serverB)
361
362 connectorA, err := NewEndpointConnector(quicDialer(readinessA), DefaultMuxCeilings())
363 require.NoError(t, err)
364 physicalA, err := connectorA.Connect(context.Background(), muxEndpoint(binding.Identity(), policy, addrA))
365 require.NoError(t, err)
366 t.Cleanup(func() { _ = physicalA.Close() })
367
368 connectorB, err := NewEndpointConnector(quicDialer(readinessB), DefaultMuxCeilings())
369 require.NoError(t, err)
370 physicalB, err := connectorB.Connect(context.Background(), muxEndpoint(binding.Identity(), policy, addrB))
371 require.NoError(t, err)
372 t.Cleanup(func() { _ = physicalB.Close() })
373
374 require.NoError(t, daemon.awaitAdoption(t))
375 require.NoError(t, daemon.awaitAdoption(t))
376 require.Equal(t, 2, daemon.supervisor.Children())
377
378 lost := make([]ports.BrokerLogicalConnection, 3)
379 for i := range lost {
380 lost[i], err = openTyped(physicalA, context.Background(), muxOpenRequest(uint64(i+1), policy))
381 require.NoError(t, err, "stream %d", i+1)
382 t.Cleanup(func() { _ = lost[i].Close() })
383 }
384 survivor, err := openTyped(physicalB, context.Background(), muxOpenRequest(1, policy))
385 require.NoError(t, err)
386 t.Cleanup(func() { _ = survivor.Close() })
387 daemon.awaitQUICAccepted(t, 4)
388
389 // The physical terminal outcome is ordered before its logical fan-out.
390 ordered := make(chan bool, 1)
391 go func() {
392 select {
393 case <-lost[0].Done():
394 case <-time.After(quicCarriageDeadline):
395 ordered <- false
396 return
397 }
398 select {
399 case <-physicalA.Done():
400 ordered <- true
401 case <-time.After(quicCarriageDeadline):
402 ordered <- false
403 }
404 }()
405
406 // Lose the daemon end of physical A's QUIC carriage entirely.
407 rawA := <-carriageA
408 require.NoError(t, rawA.Close())
409
410 require.True(t, <-ordered, "the logical Done fired before the physical Done")
411 require.True(t, channelClosed(physicalA.Done()))
412 require.Error(t, physicalA.Err())
413 require.NotEqual(t, domain.RemoteFailureNone, physicalA.FailureKind())
414 // The physical outcome fans out to every logical stream it carried. Each
415 // stream publishes on its own watcher goroutine, so the fan-out is
416 // asynchronous even though it can never precede the physical Done.
417 for i, stream := range lost {
418 require.Eventually(t, func() bool {
419 return channelClosed(stream.Done()) && stream.Err() != nil
420 }, quicCarriageDeadline, time.Millisecond, "lost stream %d never received the physical failure", i+1)
421 }
422
423 // The sibling physical QUIC connection is unaffected.
424 require.False(t, channelClosed(physicalB.Done()))
425 require.False(t, channelClosed(survivor.Done()))
426 require.NoError(t, survivor.SendClient(protocol.Ping{}))
427 message, err := survivor.ReceiveServer()
428 require.NoError(t, err)
429 require.Equal(t, protocol.Pong{}, message)
430 require.Eventually(t, func() bool { return daemon.supervisor.Children() == 1 }, quicCarriageDeadline, time.Millisecond,
431 "the lost QUIC physical child was never reaped")
432
433 // The lost physical refuses a fresh stream.
434 _, err = openTyped(physicalA, context.Background(), muxOpenRequest(9, policy))
435 require.Error(t, err)
436}
437
438// TestQUICCarriageSetupContextDetachedFromPhysicalLifetime proves the setup
439// context covers only setup: once EndpointConnector.Connect returned over a real
440// QUIC carriage, cancelling the setup context never stops the pooled physical
441// connection, which Close alone owns.
442func TestQUICCarriageSetupContextDetachedFromPhysicalLifetime(t *testing.T) {
443 binding := mustServerBinding(t)
444 server, readiness, addr := quicCarriageEndpoint(t)
445
446 daemon := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
447 daemon.serve()
448 daemon.serveQUIC(t, server)
449
450 connector, err := NewEndpointConnector(quicDialer(readiness), DefaultMuxCeilings())
451 require.NoError(t, err)
452 setupCtx, cancel := context.WithCancel(context.Background())
453 physical, err := connector.Connect(setupCtx, muxEndpoint(binding.Identity(), binding.Policy(), addr))
454 require.NoError(t, err)
455 require.NoError(t, daemon.awaitAdoption(t))
456 t.Cleanup(func() { _ = physical.Close() })
457
458 cancel()
459 require.False(t, channelClosed(physical.Done()))
460 logical, err := openTyped(physical, context.Background(), muxOpenRequest(1, binding.Policy()))
461 require.NoError(t, err)
462 t.Cleanup(func() { _ = logical.Close() })
463 daemon.awaitQUICAccepted(t, 1)
464 require.NoError(t, logical.SendClient(protocol.Ping{}))
465 message, err := logical.ReceiveServer()
466 require.NoError(t, err)
467 require.Equal(t, protocol.Pong{}, message)
468 require.False(t, channelClosed(physical.Done()))
469}
470
471// TestQUICCarriageCanceledSetupAdmitsNothing proves one setup context covers
472// bootstrap, dial, auth, and the mux handshake: a canceled setup admits no
473// physical connection, no child, and no stream.
474func TestQUICCarriageCanceledSetupAdmitsNothing(t *testing.T) {
475 binding := mustServerBinding(t)
476 server, readiness, addr := quicCarriageEndpoint(t)
477
478 daemon := newSuperviseServer(t, binding, DefaultMuxCeilings(), 0, nil)
479 daemon.serveQUIC(t, server)
480
481 connector, err := NewEndpointConnector(quicDialer(readiness), DefaultMuxCeilings())
482 require.NoError(t, err)
483 setupCtx, cancel := context.WithCancel(context.Background())
484 cancel()
485 physical, err := connector.Connect(setupCtx, muxEndpoint(binding.Identity(), binding.Policy(), addr))
486 require.ErrorIs(t, err, context.Canceled)
487 require.Nil(t, physical)
488 require.Zero(t, daemon.supervisor.Children())
489 require.Zero(t, daemon.acceptedCount())
490}
491
492// TestQUICCarriageSetupDeadlineBoundsTheMuxHandshake proves the setup deadline
493// covers the daemonmux physical handshake over a real QUIC carriage: a daemon
494// that never answers the preamble fails the whole Connect within the setup
495// deadline and publishes no physical connection.
496func TestQUICCarriageSetupDeadlineBoundsTheMuxHandshake(t *testing.T) {
497 binding := mustServerBinding(t)
498 _, readiness, addr := quicCarriageEndpoint(t)
499
500 connector, err := NewEndpointConnector(quicDialer(readiness), DefaultMuxCeilings())
501 require.NoError(t, err)
502 setupCtx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
503 defer cancel()
504 physical, err := connector.Connect(setupCtx, muxEndpoint(binding.Identity(), binding.Policy(), addr))
505 require.ErrorIs(t, err, context.DeadlineExceeded)
506 require.Nil(t, physical)
507}