1package daemonmux
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "runtime"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/require"
14
15 "github.com/bnema/vev/internal/adapters/sessionwire"
16 "github.com/bnema/vev/internal/domain"
17 "github.com/bnema/vev/internal/ports"
18 "github.com/bnema/vev/internal/protocol"
19)
20
21// errMemCarrierClosed is the synthetic carrier error an in-memory endpoint
22// returns once its shared link is closed, mirroring the adapter prompt-close
23// contract.
24var errMemCarrierClosed = errors.New("memCarrier: closed")
25
26// memCarrier is one end of a paired, in-memory FramedCarrier link. Two
27// endpoints share a closed channel and exchange complete envelopes over two
28// buffered channels, so one pump's reader is the other pump's writer.
29type memCarrier struct {
30 out chan []byte
31 in chan []byte
32 done chan struct{}
33 once *sync.Once
34}
35
36func newMemCarrierPair(buffer int) (*memCarrier, *memCarrier) {
37 aToB := make(chan []byte, buffer)
38 bToA := make(chan []byte, buffer)
39 done := make(chan struct{})
40 once := &sync.Once{}
41 a := &memCarrier{out: aToB, in: bToA, done: done, once: once}
42 b := &memCarrier{out: bToA, in: aToB, done: done, once: once}
43 return a, b
44}
45
46func (m *memCarrier) Send(ctx context.Context, payload []byte) error {
47 select {
48 case m.out <- append([]byte(nil), payload...):
49 return nil
50 case <-m.done:
51 return errMemCarrierClosed
52 case <-ctx.Done():
53 return ctx.Err()
54 }
55}
56
57func (m *memCarrier) Receive(ctx context.Context) ([]byte, error) {
58 select {
59 case payload := <-m.in:
60 return payload, nil
61 case <-m.done:
62 return nil, errMemCarrierClosed
63 case <-ctx.Done():
64 return nil, ctx.Err()
65 }
66}
67
68func (m *memCarrier) Close() error {
69 m.once.Do(func() { close(m.done) })
70 return nil
71}
72
73// newPairedPumps builds a broker-side pump (decodes server frames, emits client
74// frames) and a daemon-side pump (decodes client frames, emits server frames)
75// joined by one in-memory link, starts both, and closes them at test end.
76func newPairedPumps(t *testing.T, ceilings MuxCeilings) (*Pump, *Pump) {
77 t.Helper()
78 broker, daemon, start := newPairedPumpsUnstarted(t, ceilings)
79 start()
80 return broker, daemon
81}
82
83// newPairedPumpsUnstarted builds the same paired pump link as newPairedPumps
84// but leaves both pumps unstarted, so a daemon-side consumer can register its
85// admission observer before the pump's reader begins (the pump refuses a late
86// registration with ErrAdmissionObserverLate). The returned start function
87// starts both pumps exactly where the caller needs them running; the pumps are
88// closed at test end.
89func newPairedPumpsUnstarted(t *testing.T, ceilings MuxCeilings) (*Pump, *Pump, func()) {
90 t.Helper()
91 brokerCarrier, daemonCarrier := newMemCarrierPair(4096)
92 broker, err := NewPump(brokerCarrier, DirectionServer, ceilings)
93 require.NoError(t, err)
94 daemon, err := NewPump(daemonCarrier, DirectionClient, ceilings)
95 require.NoError(t, err)
96 ctx, cancel := context.WithCancel(context.Background())
97 t.Cleanup(func() {
98 cancel()
99 _ = broker.Close()
100 _ = daemon.Close()
101 })
102 start := func() {
103 broker.Start(ctx)
104 daemon.Start(ctx)
105 }
106 return broker, daemon, start
107}
108
109func mustConnector(t *testing.T, pump *Pump) *LogicalConnector {
110 t.Helper()
111 connector, err := NewLogicalConnector(pump)
112 require.NoError(t, err)
113 return connector
114}
115
116func logicalTestPolicy() ports.BrokerPolicy {
117 return ports.BrokerPolicy{
118 ProtocolVersion: protocol.Version,
119 CatalogSchemaVersion: 1,
120 EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned,
121 Transport: "mux",
122 Trust: "trust",
123 Launch: "launch",
124 Isolation: "isolation",
125 }
126}
127
128// controlRequest builds one valid control-purpose open request for stream.
129func controlRequest(stream int) ports.BrokerOpenStreamRequest {
130 return ports.BrokerOpenStreamRequest{
131 Epoch: 1,
132 Purpose: ports.BrokerStreamControl,
133 Local: true,
134 Connection: ports.BrokerConnectionID{1},
135 Stream: ports.BrokerStreamID(stream),
136 Policy: logicalTestPolicy(),
137 StartMode: ports.BrokerDaemonStartIfNeeded,
138 }
139}
140
141// awaitStream waits until a stream identity has been admitted on the daemon
142// side.
143func awaitStream(pump *Pump, id PhysicalStreamID, deadline time.Time) (StreamStatus, error) {
144 for time.Now().Before(deadline) {
145 if status, ok := pump.Engine().Status(id); ok && status.State != StreamTerminal {
146 return status, nil
147 }
148 time.Sleep(time.Millisecond)
149 }
150 return StreamStatus{}, fmt.Errorf("daemonmux test: stream %d was not admitted", id)
151}
152
153// sessionPeer is a test-only daemon-side typed session: it frames the same mux
154// byte stream with streamframe and wraps it with sessionwire's server
155// connection, so the client's typed connection speaks to a real typed peer.
156type sessionPeer struct {
157 raw *muxTransport
158 srv ports.ServerConnection
159 flood int
160}
161
162func newSessionPeer(pump *Pump, ref StreamRef, flood int) *sessionPeer {
163 status, _ := pump.Engine().Status(ref.Physical)
164 pipe := newMuxStreamPipe(pump, ref.Physical, pump.Ceilings().StreamChunkLimit, status.Done)
165 raw := newMuxTransport(pipe)
166 peer := &sessionPeer{raw: raw, srv: sessionwire.NewServerConnection(raw), flood: flood}
167 go peer.serve()
168 return peer
169}
170
171// serve answers each client message with p.flood Pong frames, triggering the
172// server preamble on the first call.
173func (p *sessionPeer) serve() {
174 for {
175 message, err := p.srv.ReceiveClient()
176 if err != nil {
177 return
178 }
179 switch message.(type) {
180 case protocol.Ping, protocol.Input, protocol.Resize, protocol.Detach:
181 for i := 0; i < p.flood; i++ {
182 if err := p.srv.SendServer(protocol.Pong{}); err != nil {
183 return
184 }
185 }
186 }
187 }
188}
189
190func (p *sessionPeer) close() {
191 if p.raw != nil {
192 _ = p.raw.Close()
193 }
194}
195
196// daemonAcceptor accepts mux opens on the daemon side: it confirms each one
197// with Opened (or refuses it), and optionally wraps it as a typed session
198// peer. It never uses testing.T from its goroutines.
199type daemonAcceptor struct {
200 pump *Pump
201 flood int
202
203 mu sync.Mutex
204 count int
205 peers map[PhysicalStreamID]*sessionPeer
206 errCh chan error
207}
208
209func newDaemonAcceptor(pump *Pump, flood int) *daemonAcceptor {
210 return &daemonAcceptor{pump: pump, flood: flood, peers: map[PhysicalStreamID]*sessionPeer{}, errCh: make(chan error, 256)}
211}
212
213func (a *daemonAcceptor) start(id PhysicalStreamID, deadline time.Time) {
214 go func() {
215 status, err := awaitStream(a.pump, id, deadline)
216 if err != nil {
217 a.fail(err)
218 return
219 }
220 if err := a.pump.Engine().Opened(Opened{Ref: status.Ref}); err != nil {
221 a.fail(err)
222 return
223 }
224 if err := a.pump.Send(Opened{Ref: status.Ref}); err != nil {
225 a.fail(err)
226 return
227 }
228 a.add(id, newSessionPeer(a.pump, status.Ref, a.flood))
229 }()
230}
231
232func (a *daemonAcceptor) startBare(id PhysicalStreamID, deadline time.Time) {
233 go func() {
234 status, err := awaitStream(a.pump, id, deadline)
235 if err != nil {
236 a.fail(err)
237 return
238 }
239 if err := a.pump.Engine().Opened(Opened{Ref: status.Ref}); err != nil {
240 a.fail(err)
241 return
242 }
243 if err := a.pump.Send(Opened{Ref: status.Ref}); err != nil {
244 a.fail(err)
245 return
246 }
247 a.add(id, nil)
248 }()
249}
250
251func (a *daemonAcceptor) startRefused(id PhysicalStreamID, deadline time.Time) {
252 go func() {
253 status, err := awaitStream(a.pump, id, deadline)
254 if err != nil {
255 a.fail(err)
256 return
257 }
258 refusal := ErrorDetail{Code: ports.BrokerErrorUnavailable}
259 if err := a.pump.Send(Refused{Ref: status.Ref, Error: refusal}); err != nil {
260 a.fail(err)
261 return
262 }
263 if err := a.pump.Engine().Refused(Refused{Ref: status.Ref, Error: refusal}); err != nil {
264 a.fail(err)
265 return
266 }
267 a.add(id, nil)
268 }()
269}
270
271func (a *daemonAcceptor) add(id PhysicalStreamID, peer *sessionPeer) {
272 a.mu.Lock()
273 defer a.mu.Unlock()
274 a.count++
275 if peer != nil {
276 a.peers[id] = peer
277 }
278}
279
280func (a *daemonAcceptor) fail(err error) {
281 select {
282 case a.errCh <- err:
283 default:
284 }
285}
286
287func (a *daemonAcceptor) wait(count int, deadline time.Time) error {
288 for time.Now().Before(deadline) {
289 select {
290 case err := <-a.errCh:
291 return err
292 default:
293 }
294 a.mu.Lock()
295 accepted := a.count
296 a.mu.Unlock()
297 if accepted >= count {
298 return nil
299 }
300 time.Sleep(time.Millisecond)
301 }
302 a.mu.Lock()
303 accepted := a.count
304 a.mu.Unlock()
305 select {
306 case err := <-a.errCh:
307 return err
308 default:
309 }
310 return fmt.Errorf("daemonmux test: accepted %d of %d streams", accepted, count)
311}
312
313func (a *daemonAcceptor) closePeers() {
314 a.mu.Lock()
315 defer a.mu.Unlock()
316 for _, peer := range a.peers {
317 peer.close()
318 }
319}
320
321func openWithDeadline(t *testing.T, connector *LogicalConnector, stream int) typedLogical {
322 t.Helper()
323 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
324 defer cancel()
325 connection, err := connector.Open(ctx, controlRequest(stream))
326 require.NoError(t, err)
327 return newTypedLogical(connection)
328}
329
330// watchRegistered reports whether the pump still holds an inbound watch entry
331// for one physical stream.
332func watchRegistered(pump *Pump, physical PhysicalStreamID) bool {
333 pump.watchMu.Lock()
334 defer pump.watchMu.Unlock()
335 _, ok := pump.watch[physical]
336 return ok
337}
338
339func TestNewLogicalConnectorValidation(t *testing.T) {
340 t.Run("nil pump", func(t *testing.T) {
341 connector, err := NewLogicalConnector(nil)
342 require.ErrorIs(t, err, ErrLogicalConfig)
343 require.Nil(t, connector)
344 })
345 t.Run("server-side pump is refused", func(t *testing.T) {
346 pump, _ := newTestPump(t, DirectionClient)
347 connector, err := NewLogicalConnector(pump)
348 require.ErrorIs(t, err, ErrLogicalConfig)
349 require.Nil(t, connector)
350 })
351 t.Run("broker-side pump is accepted", func(t *testing.T) {
352 pump, _ := newTestPump(t, DirectionServer)
353 connector, err := NewLogicalConnector(pump)
354 require.NoError(t, err)
355 require.NotNil(t, connector)
356 })
357}
358
359// TestMuxStreamPipeChunkAndOrder proves the byte stream chunks outbound writes
360// at the negotiated ceiling in order and drains inbound chunks in order.
361func TestMuxStreamPipeChunkAndOrder(t *testing.T) {
362 pump, carrier := newTestPump(t, DirectionClient)
363 pump.Start(context.Background())
364 require.NoError(t, pump.Engine().Open(openFor(1)))
365 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
366
367 pipe := newMuxStreamPipe(pump, 1, 4, mustStatus(t, pump.Engine(), 1).Done)
368 written, err := pipe.Write([]byte("0123456789"))
369 require.NoError(t, err)
370 require.Equal(t, 10, written)
371
372 var chunks [][]byte
373 for i := 0; i < 3; i++ {
374 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
375 require.NoError(t, err)
376 data, ok := message.(Data)
377 require.True(t, ok)
378 require.LessOrEqual(t, uint64(len(data.Data)), uint64(4))
379 chunks = append(chunks, data.Data)
380 }
381 require.Equal(t, [][]byte{[]byte("0123"), []byte("4567"), []byte("89")}, chunks)
382
383 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte("abc")})
384 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte("de")})
385 buffer := make([]byte, 3)
386 _, err = io.ReadFull(pipe, buffer)
387 require.NoError(t, err)
388 require.Equal(t, []byte("abc"), buffer)
389 buffer = make([]byte, 2)
390 _, err = io.ReadFull(pipe, buffer)
391 require.NoError(t, err)
392 require.Equal(t, []byte("de"), buffer)
393}
394
395// TestLogicalTwoConnections opens two typed logical connections over one
396// physical pair, exchanges a typed round trip on each, and closes them
397// independently.
398func TestLogicalTwoConnections(t *testing.T) {
399 const count = 2
400 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
401 connector := mustConnector(t, brokerPump)
402 deadline := time.Now().Add(30 * time.Second)
403
404 acceptor := newDaemonAcceptor(daemonPump, 1)
405 for i := 1; i <= count; i++ {
406 acceptor.start(PhysicalStreamID(i), deadline)
407 }
408
409 connections := make([]typedLogical, 0, count)
410 for i := 1; i <= count; i++ {
411 connection := openWithDeadline(t, connector, i)
412 require.NoError(t, connection.SendClient(protocol.Ping{}), "connection %d", i)
413 message, err := connection.ReceiveServer()
414 require.NoError(t, err, "connection %d", i)
415 require.Equal(t, protocol.Pong{}, message, "connection %d", i)
416 connections = append(connections, connection)
417 }
418 require.NoError(t, acceptor.wait(count, deadline))
419
420 for i, connection := range connections {
421 require.False(t, channelClosed(connection.Done()), "connection %d", i)
422 require.NoError(t, connection.Close(), "connection %d", i)
423 require.True(t, channelClosed(connection.Done()), "connection %d", i)
424 require.NoError(t, connection.Err(), "connection %d", i)
425 }
426 acceptor.closePeers()
427 require.False(t, channelClosed(brokerPump.Done()))
428 require.False(t, channelClosed(daemonPump.Done()))
429}
430
431// TestLogicalHundredConnectionsConcurrentAdmission opens 100 typed logical
432// connections concurrently over one physical pair, proves all 100 are live
433// before any traffic, interleaves a typed round trip on every stream, and then
434// drains and closes them. It fills the physical connection to its 128-stream
435// ceiling, proves the 129th stream is refused with the typed admission limit,
436// and proves a retry after a Close succeeds with a fresh monotonic identity.
437func TestLogicalHundredConnectionsConcurrentAdmission(t *testing.T) {
438 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
439 connector := mustConnector(t, brokerPump)
440 deadline := time.Now().Add(60 * time.Second)
441
442 const count = 100
443 const ceiling = int(MaxMuxStreams)
444 require.Equal(t, 128, ceiling, "the default physical connection admits 128 streams")
445
446 acceptor := newDaemonAcceptor(daemonPump, 1)
447 // Every physical identity the connector can allocate, plus the retry after
448 // the refused 129th attempt (which still consumes its monotonic identity).
449 for i := 1; i <= ceiling; i++ {
450 acceptor.start(PhysicalStreamID(i), deadline)
451 }
452 acceptor.start(PhysicalStreamID(ceiling+2), deadline)
453
454 openConcurrently := func(requests []ports.BrokerOpenStreamRequest) ([]*LogicalConnection, []error) {
455 connections := make([]*LogicalConnection, len(requests))
456 errs := make([]error, len(requests))
457 var wg sync.WaitGroup
458 for i, request := range requests {
459 wg.Add(1)
460 go func(index int, request ports.BrokerOpenStreamRequest) {
461 defer wg.Done()
462 ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
463 defer cancel()
464 connections[index], errs[index] = connector.Open(ctx, request)
465 }(i, request)
466 }
467 wg.Wait()
468 return connections, errs
469 }
470
471 requests := make([]ports.BrokerOpenStreamRequest, count)
472 for i := range requests {
473 requests[i] = controlRequest(i + 1)
474 }
475 connections, errs := openConcurrently(requests)
476 for i, err := range errs {
477 require.NoError(t, err, "connection %d", i)
478 require.NotNil(t, connections[i], "connection %d", i)
479 }
480 require.Equal(t, count, brokerPump.Engine().Live(), "all connections are open before any traffic")
481 require.NoError(t, acceptor.wait(count, deadline))
482
483 // Interleave a typed round trip across every stream, then drain it.
484 trafficErrs := make([]error, count)
485 var traffic sync.WaitGroup
486 for i := range connections {
487 traffic.Add(1)
488 go func(index int) {
489 defer traffic.Done()
490 typed := asTyped(connections[index])
491 if err := typed.SendClient(protocol.Ping{}); err != nil {
492 trafficErrs[index] = err
493 return
494 }
495 message, err := typed.ReceiveServer()
496 if err != nil {
497 trafficErrs[index] = err
498 return
499 }
500 if message != (protocol.Pong{}) {
501 trafficErrs[index] = fmt.Errorf("connection %d: received %v", index, message)
502 }
503 }(i)
504 }
505 traffic.Wait()
506 for i, err := range trafficErrs {
507 require.NoError(t, err, "connection %d", i)
508 }
509 require.Equal(t, count, brokerPump.Engine().Live(), "traffic never settles a live stream")
510
511 // Fill the physical connection to its concurrent-stream ceiling.
512 fill := make([]ports.BrokerOpenStreamRequest, ceiling-count)
513 for i := range fill {
514 fill[i] = controlRequest(count + i + 1)
515 }
516 extra, fillErrs := openConcurrently(fill)
517 for i, err := range fillErrs {
518 require.NoError(t, err, "extra connection %d", i)
519 }
520 require.Equal(t, ceiling, brokerPump.Engine().Live())
521 require.NoError(t, acceptor.wait(ceiling, deadline))
522
523 // The 129th stream is refused by the admission ceiling before it reaches
524 // the daemon.
525 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
526 defer cancel()
527 refused, err := connector.Open(ctx, controlRequest(1000))
528 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
529 require.Nil(t, refused)
530 require.Equal(t, ceiling, brokerPump.Engine().Live(), "a refused admission frees nothing")
531
532 // A Close frees one slot; the retry uses a fresh monotonic identity.
533 require.NoError(t, connections[0].Close())
534 require.True(t, channelClosed(connections[0].Done()))
535 require.Equal(t, ceiling-1, brokerPump.Engine().Live())
536 retry := openWithDeadline(t, connector, 1001)
537 require.NoError(t, retry.SendClient(protocol.Ping{}))
538 message, err := retry.ReceiveServer()
539 require.NoError(t, err)
540 require.Equal(t, protocol.Pong{}, message)
541 require.Equal(t, ceiling, brokerPump.Engine().Live())
542 require.NoError(t, acceptor.wait(ceiling+1, deadline))
543
544 for i := 1; i < len(connections); i++ {
545 require.False(t, channelClosed(connections[i].Done()), "connection %d", i)
546 require.NoError(t, connections[i].Close(), "connection %d", i)
547 require.True(t, channelClosed(connections[i].Done()), "connection %d", i)
548 require.NoError(t, connections[i].Err(), "connection %d", i)
549 }
550 for i := range extra {
551 require.NoError(t, extra[i].Close(), "extra connection %d", i)
552 }
553 require.NoError(t, retry.Close())
554 require.Zero(t, brokerPump.Engine().Live(), "every stream is retired")
555 acceptor.closePeers()
556 require.False(t, channelClosed(brokerPump.Done()))
557 require.False(t, channelClosed(daemonPump.Done()))
558}
559
560// TestLogicalBlockedReaderSiblingProgresses is the regression for a slow
561// remote consumer killing its own attachment. The daemon floods far more
562// frames than one stream window holds while the consumer is not reading: the
563// daemon's writer must wait for credit instead of resetting the stream, the
564// sibling keeps exchanging typed messages, and once the consumer reads again
565// every flooded frame arrives in order on the still-open stream.
566func TestLogicalBlockedReaderSiblingProgresses(t *testing.T) {
567 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
568 connector := mustConnector(t, brokerPump)
569 deadline := time.Now().Add(15 * time.Second)
570
571 // Each Pong is one tiny frame, so the flood is far more frames than the
572 // eight queued frames that used to reset the stream, and more credit than
573 // one window.
574 const flood = 20000
575 blockedAcceptor := newDaemonAcceptor(daemonPump, flood)
576 blockedAcceptor.start(1, deadline)
577 siblingAcceptor := newDaemonAcceptor(daemonPump, 1)
578 siblingAcceptor.start(2, deadline)
579
580 blocked := openWithDeadline(t, connector, 1)
581 sibling := openWithDeadline(t, connector, 2)
582 require.NoError(t, blockedAcceptor.wait(1, deadline))
583 require.NoError(t, siblingAcceptor.wait(1, deadline))
584
585 require.NoError(t, sibling.SendClient(protocol.Ping{}))
586 message, err := sibling.ReceiveServer()
587 require.NoError(t, err)
588 require.Equal(t, protocol.Pong{}, message)
589
590 // Trigger the flood, then do not read: the daemon fills the window and
591 // its writer waits for credit.
592 require.NoError(t, blocked.SendClient(protocol.Ping{}))
593 require.Eventually(t, func() bool {
594 status, ok := daemonPump.Engine().Status(1)
595 return ok && status.SendCredit < chunkCredit(int(MaxMuxChunkBytes))
596 }, 5*time.Second, time.Millisecond, "the daemon never exhausted its credit")
597 require.False(t, channelClosed(blocked.Done()), "a slow consumer must not reset its stream")
598
599 // The sibling still progresses and the physical connection is untouched.
600 require.NoError(t, sibling.SendClient(protocol.Ping{}))
601 message, err = sibling.ReceiveServer()
602 require.NoError(t, err)
603 require.Equal(t, protocol.Pong{}, message)
604 require.False(t, channelClosed(brokerPump.Done()))
605 require.Equal(t, StreamOpen, mustStatus(t, brokerPump.Engine(), 2).State)
606
607 // The consumer catches up: every flooded frame arrives on the same stream.
608 for i := 0; i < flood; i++ {
609 message, err := blocked.ReceiveServer()
610 require.NoError(t, err, "frame %d", i)
611 require.Equal(t, protocol.Pong{}, message)
612 }
613 require.False(t, channelClosed(blocked.Done()))
614 require.Equal(t, StreamOpen, mustStatus(t, brokerPump.Engine(), 1).State)
615}
616
617// TestLogicalMalformedInnerFrameResetsOnlyStream proves a malformed inner
618// session frame resets exactly its own stream and leaves the physical
619// connection and a sibling untouched.
620func TestLogicalMalformedInnerFrameResetsOnlyStream(t *testing.T) {
621 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
622 connector := mustConnector(t, brokerPump)
623 deadline := time.Now().Add(15 * time.Second)
624
625 bareAcceptor := newDaemonAcceptor(daemonPump, 1)
626 bareAcceptor.startBare(1, deadline)
627 siblingAcceptor := newDaemonAcceptor(daemonPump, 1)
628 siblingAcceptor.start(2, deadline)
629
630 malformed := openWithDeadline(t, connector, 1)
631 sibling := openWithDeadline(t, connector, 2)
632 require.NoError(t, bareAcceptor.wait(1, deadline))
633 require.NoError(t, siblingAcceptor.wait(1, deadline))
634
635 require.NoError(t, sibling.SendClient(protocol.Ping{}))
636 message, err := sibling.ReceiveServer()
637 require.NoError(t, err)
638 require.Equal(t, protocol.Pong{}, message)
639
640 // A zero-length inner frame is malformed: the client's first read fails and
641 // must reset only this stream.
642 require.NoError(t, daemonPump.SendData(1, []byte{0, 0, 0, 0}, nil))
643 _, err = malformed.ReceiveServer()
644 require.Error(t, err)
645 require.True(t, channelClosed(malformed.Done()))
646
647 var loss ports.BrokerStreamLost
648 require.ErrorAs(t, malformed.Err(), &loss)
649 require.Equal(t, ports.BrokerStreamID(1), loss.Stream)
650 require.Equal(t, ports.BrokerConnectionID{1}, loss.Connection)
651 require.Equal(t, ports.BrokerEpoch(1), loss.Epoch)
652 require.Equal(t, domain.RemoteFailureInvalidResponse, loss.Cause)
653
654 // The daemon observed exactly a stream-local reset.
655 require.Eventually(t, func() bool {
656 status, ok := daemonPump.Engine().Status(1)
657 return ok && status.State == StreamTerminal
658 }, 5*time.Second, time.Millisecond)
659 reset := mustStatus(t, daemonPump.Engine(), 1)
660 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.FailureKind)
661
662 // The sibling still progresses and the physical pumps stay open.
663 require.NoError(t, sibling.SendClient(protocol.Ping{}))
664 message, err = sibling.ReceiveServer()
665 require.NoError(t, err)
666 require.Equal(t, protocol.Pong{}, message)
667 require.False(t, channelClosed(brokerPump.Done()))
668 require.False(t, channelClosed(daemonPump.Done()))
669}
670
671// TestLogicalOpenCancellationReleasesReservation proves a cancelled open
672// releases its local reservation and a later open still succeeds.
673func TestLogicalOpenCancellationReleasesReservation(t *testing.T) {
674 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
675 connector := mustConnector(t, brokerPump)
676 deadline := time.Now().Add(15 * time.Second)
677
678 // No daemon acceptor: the open waits for an Opened that never arrives.
679 ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
680 defer cancel()
681 connection, err := connector.Open(ctx, controlRequest(1))
682 require.ErrorIs(t, err, context.DeadlineExceeded)
683 require.Nil(t, connection)
684
685 require.Eventually(t, func() bool {
686 status, ok := brokerPump.Engine().Status(1)
687 return ok && status.State == StreamTerminal
688 }, 5*time.Second, time.Millisecond)
689 require.False(t, watchRegistered(brokerPump, 1), "cancelled open leaked a watch entry")
690
691 // A sibling stream on the same physical connection still opens and works.
692 acceptor := newDaemonAcceptor(daemonPump, 1)
693 acceptor.start(2, deadline)
694 sibling := openWithDeadline(t, connector, 2)
695 require.NoError(t, acceptor.wait(1, deadline))
696 require.NoError(t, sibling.SendClient(protocol.Ping{}))
697 message, err := sibling.ReceiveServer()
698 require.NoError(t, err)
699 require.Equal(t, protocol.Pong{}, message)
700 require.False(t, channelClosed(brokerPump.Done()))
701}
702
703// TestLogicalOpenRefused proves a daemon refusal fails the open with the typed
704// refusal and leaves the physical connection open.
705func TestLogicalOpenRefused(t *testing.T) {
706 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
707 connector := mustConnector(t, brokerPump)
708 deadline := time.Now().Add(15 * time.Second)
709
710 acceptor := newDaemonAcceptor(daemonPump, 1)
711 acceptor.startRefused(1, deadline)
712
713 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
714 defer cancel()
715 connection, err := connector.Open(ctx, controlRequest(1))
716 require.Error(t, err)
717 require.Nil(t, connection)
718 require.NoError(t, acceptor.wait(1, deadline))
719 require.False(t, channelClosed(brokerPump.Done()))
720}
721
722// TestLogicalCloseIsolation proves closing one logical connection settles only
723// itself and leaves a sibling and the physical connection untouched.
724func TestLogicalCloseIsolation(t *testing.T) {
725 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
726 connector := mustConnector(t, brokerPump)
727 deadline := time.Now().Add(15 * time.Second)
728
729 firstAcceptor := newDaemonAcceptor(daemonPump, 1)
730 firstAcceptor.start(1, deadline)
731 secondAcceptor := newDaemonAcceptor(daemonPump, 1)
732 secondAcceptor.start(2, deadline)
733
734 first := openWithDeadline(t, connector, 1)
735 second := openWithDeadline(t, connector, 2)
736 require.NoError(t, firstAcceptor.wait(1, deadline))
737 require.NoError(t, secondAcceptor.wait(1, deadline))
738
739 require.NoError(t, first.Close())
740 require.True(t, channelClosed(first.Done()))
741 require.NoError(t, first.Err())
742 require.Equal(t, domain.RemoteFailureNone, first.FailureKind())
743 require.False(t, watchRegistered(brokerPump, 1), "closed logical connection leaked a watch entry")
744
745 // The daemon observed the orderly Close for stream 1.
746 require.Eventually(t, func() bool {
747 status, ok := daemonPump.Engine().Status(1)
748 return ok && status.State == StreamTerminal && status.Err == nil
749 }, 5*time.Second, time.Millisecond)
750
751 require.NoError(t, second.SendClient(protocol.Ping{}))
752 message, err := second.ReceiveServer()
753 require.NoError(t, err)
754 require.Equal(t, protocol.Pong{}, message)
755 require.False(t, channelClosed(brokerPump.Done()))
756 require.False(t, channelClosed(daemonPump.Done()))
757}
758
759// TestLogicalCloseDiscardsQueuedInboundReleasesAccounting proves a local Close
760// first discards the inbound chunks the engine already accepted for its stream:
761// the engine's orderly Close reaches its terminal state, releasing both the
762// stream's admission slot and its queued aggregate bytes, instead of parking
763// the stream in closing with a queue this side will never read again. Every
764// cycle closes with one queued inbound chunk under a two-stream ceiling, so a
765// leaked slot would refuse the third admission, and the peer still observes
766// exactly one orderly mux Close for each cycle.
767func TestLogicalCloseDiscardsQueuedInboundReleasesAccounting(t *testing.T) {
768 ceilings := DefaultMuxCeilings()
769 ceilings.MaxStreams = 2
770 pump, carrier := newTestPumpWithCeilings(t, DirectionServer, ceilings)
771 pump.Start(context.Background())
772
773 chunk := []byte("queued")
774 const cycles = 3
775 for i := 1; i <= cycles; i++ {
776 id := PhysicalStreamID(i)
777 require.NoError(t, pump.Engine().Open(openFor(id)), "cycle %d admission", i)
778 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
779 status := mustStatus(t, pump.Engine(), id)
780 connection := newLogicalConnection(pump, status.Ref, status.Done, pump.cause(id))
781
782 // The peer pipelined one chunk that this side never reads, so the stream
783 // holds accepted inbound data when the local Close arrives.
784 deliverServerFrame(t, carrier, Data{Physical: id, Data: chunk})
785 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.AggregateBytes() > 0 },
786 "cycle %d: the accepted chunk was never queued", i)
787
788 require.NoError(t, connection.Close())
789 require.NoError(t, connection.Close(), "Close is idempotent")
790 require.True(t, channelClosed(connection.Done()))
791 require.NoError(t, connection.Err())
792
793 closed := mustStatus(t, pump.Engine(), id)
794 require.Equal(t, StreamTerminal, closed.State, "cycle %d: the closed stream must be terminal", i)
795 require.Zero(t, closed.QueuedChunks, "cycle %d", i)
796 require.Zero(t, pump.Engine().Live(), "cycle %d: a closing stream leaked its admission slot", i)
797 require.Zero(t, pump.Engine().AggregateBytes(), "cycle %d: a closing stream leaked its queued bytes", i)
798 require.False(t, watchRegistered(pump, id), "cycle %d: the closed connection leaked a watch entry", i)
799
800 // Exactly one orderly mux Close reached the peer; the drain is local and
801 // never replaces it with a Reset.
802 message, err := DecodeClient(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
803 require.NoError(t, err)
804 closedFrame, ok := message.(Close)
805 require.True(t, ok, "cycle %d: the peer frame is not a Close", i)
806 require.Equal(t, Close{Physical: id}, closedFrame)
807 carrier.requireNoSent(t, 20*time.Millisecond)
808 }
809 require.False(t, pump.Engine().Closed())
810 require.False(t, channelClosed(pump.Done()))
811}
812
813// TestLogicalPhysicalLossFanout proves a physical loss publishes one stable
814// ports.BrokerStreamLost per affected logical stream with its exact scope.
815func TestLogicalPhysicalLossFanout(t *testing.T) {
816 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
817 connector := mustConnector(t, brokerPump)
818 deadline := time.Now().Add(15 * time.Second)
819
820 const count = 3
821 acceptor := newDaemonAcceptor(daemonPump, 1)
822 for i := 1; i <= count; i++ {
823 acceptor.start(PhysicalStreamID(i), deadline)
824 }
825 connections := make([]typedLogical, 0, count)
826 for i := 1; i <= count; i++ {
827 connection := openWithDeadline(t, connector, i)
828 require.NoError(t, connection.SendClient(protocol.Ping{}))
829 message, err := connection.ReceiveServer()
830 require.NoError(t, err)
831 require.Equal(t, protocol.Pong{}, message)
832 connections = append(connections, connection)
833 }
834 require.NoError(t, acceptor.wait(count, deadline))
835
836 // Failing the daemon-side carrier is a physical loss on the broker pump.
837 require.NoError(t, daemonPump.Close())
838 require.Eventually(t, func() bool { return channelClosed(brokerPump.Done()) }, 5*time.Second, time.Millisecond)
839
840 for i, connection := range connections {
841 require.Eventually(t, func() bool { return channelClosed(connection.Done()) }, 5*time.Second, time.Millisecond)
842 var loss ports.BrokerStreamLost
843 require.ErrorAs(t, connection.Err(), &loss, "connection %d", i)
844 require.Equal(t, ports.BrokerStreamID(i+1), loss.Stream, "connection %d", i)
845 require.Equal(t, ports.BrokerConnectionID{1}, loss.Connection, "connection %d", i)
846 require.Equal(t, domain.RemoteFailureTransport, loss.Cause, "connection %d", i)
847 var boxed ports.BrokerError
848 require.ErrorAs(t, connection.Err(), &boxed, "connection %d", i)
849 require.Equal(t, ports.BrokerErrorAttachmentLost, boxed.Code, "connection %d", i)
850 }
851 require.Error(t, brokerPump.Err())
852 require.Equal(t, domain.RemoteFailureTransport, brokerPump.FailureKind())
853}
854
855// connections leaves no goroutine behind.
856func TestLogicalNoGoroutineLeak(t *testing.T) {
857 before := runtime.NumGoroutine()
858
859 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
860 connector := mustConnector(t, brokerPump)
861 deadline := time.Now().Add(15 * time.Second)
862
863 const count = 5
864 acceptor := newDaemonAcceptor(daemonPump, 1)
865 for i := 1; i <= count; i++ {
866 acceptor.start(PhysicalStreamID(i), deadline)
867 }
868 connections := make([]typedLogical, 0, count)
869 for i := 1; i <= count; i++ {
870 connection := openWithDeadline(t, connector, i)
871 require.NoError(t, connection.SendClient(protocol.Ping{}))
872 message, err := connection.ReceiveServer()
873 require.NoError(t, err)
874 require.Equal(t, protocol.Pong{}, message)
875 connections = append(connections, connection)
876 }
877 require.NoError(t, acceptor.wait(count, deadline))
878
879 for _, connection := range connections {
880 require.NoError(t, connection.Close())
881 }
882 acceptor.closePeers()
883 require.NoError(t, brokerPump.Close())
884 require.NoError(t, daemonPump.Close())
885
886 require.Eventually(t, func() bool { return runtime.NumGoroutine() <= before+3 }, 5*time.Second, time.Millisecond,
887 "logical connection goroutines leaked")
888}
889
890// TestLogicalPeerCloseReleasesGoroutines proves a stream terminated by a peer
891// mux Close joins its watcher and framed-transport goroutines without an
892// explicit local Close.
893func TestLogicalPeerCloseReleasesGoroutines(t *testing.T) {
894 before := runtime.NumGoroutine()
895
896 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
897 connector := mustConnector(t, brokerPump)
898 deadline := time.Now().Add(15 * time.Second)
899
900 acceptor := newDaemonAcceptor(daemonPump, 1)
901 acceptor.start(1, deadline)
902 connection := openWithDeadline(t, connector, 1)
903 require.NoError(t, acceptor.wait(1, deadline))
904 require.NoError(t, connection.SendClient(protocol.Ping{}))
905 message, err := connection.ReceiveServer()
906 require.NoError(t, err)
907 require.Equal(t, protocol.Pong{}, message)
908
909 // The daemon ends the stream orderly; the logical connection must settle
910 // itself and join its goroutines without a local Close.
911 if _, err := daemonPump.Engine().Close(Close{Physical: 1}); err != nil {
912 t.Fatalf("daemon close: %v", err)
913 }
914 require.NoError(t, daemonPump.Send(Close{Physical: 1}))
915 require.Eventually(t, func() bool { return channelClosed(connection.Done()) }, 5*time.Second, time.Millisecond)
916 require.NoError(t, connection.Err())
917
918 acceptor.closePeers()
919 require.NoError(t, brokerPump.Close())
920 require.NoError(t, daemonPump.Close())
921
922 require.Eventually(t, func() bool { return runtime.NumGoroutine() <= before+3 }, 5*time.Second, time.Millisecond,
923 "peer-closed logical connection goroutines leaked")
924}
925
926// TestLogicalPhysicalLossBlockedReadReturnsLoss proves a read blocked on an
927// idle logical stream wakes when the physical connection fails and reports a
928// typed ports.BrokerStreamLost rather than a clean io.EOF. The pump publishes
929// its terminal state before the engine terminalizes each stream, so the read
930// must consult the pump's physical cause instead of assuming an orderly end.
931func TestLogicalPhysicalLossBlockedReadReturnsLoss(t *testing.T) {
932 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
933 connector := mustConnector(t, brokerPump)
934 deadline := time.Now().Add(15 * time.Second)
935
936 acceptor := newDaemonAcceptor(daemonPump, 1)
937 acceptor.start(1, deadline)
938 connection := openWithDeadline(t, connector, 1)
939 require.NoError(t, acceptor.wait(1, deadline))
940
941 require.NoError(t, connection.SendClient(protocol.Ping{}))
942 message, err := connection.ReceiveServer()
943 require.NoError(t, err)
944 require.Equal(t, protocol.Pong{}, message)
945
946 type result struct {
947 message protocol.ServerMessage
948 err error
949 }
950 blocked := make(chan result, 1)
951 go func() {
952 msg, err := connection.ReceiveServer()
953 blocked <- result{message: msg, err: err}
954 }()
955 // Let the reader reach the inner stream and block there.
956 time.Sleep(50 * time.Millisecond)
957 require.False(t, channelClosed(connection.Done()), "the read is blocked, not settled")
958
959 // Fail the daemon side; the broker observes a physical loss.
960 require.NoError(t, daemonPump.Close())
961
962 select {
963 case got := <-blocked:
964 require.Nil(t, got.message)
965 require.Error(t, got.err)
966 require.NotErrorIs(t, got.err, io.EOF, "a physical failure must not surface as a clean EOF")
967 var loss ports.BrokerStreamLost
968 require.ErrorAs(t, got.err, &loss)
969 require.Equal(t, ports.BrokerStreamID(1), loss.Stream)
970 require.Equal(t, domain.RemoteFailureTransport, loss.Cause)
971 case <-time.After(5 * time.Second):
972 t.Fatal("blocked read did not return after the physical failure")
973 }
974 require.True(t, channelClosed(connection.Done()))
975}
976
977// TestLogicalTerminalCauseSurvivesEngineEviction proves a settled logical
978// connection reports its exact stream-local cause after the engine evicts the
979// bounded record. The retained authority outlives the record, so a Reset can
980// never degrade into an orderly close for a reader that observes the end late.
981func TestLogicalTerminalCauseSurvivesEngineEviction(t *testing.T) {
982 pump, _ := newTestPump(t, DirectionServer)
983 require.NoError(t, pump.Engine().Open(openFor(1)))
984 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
985 status := mustStatus(t, pump.Engine(), 1)
986
987 connection := &LogicalConnection{
988 pump: pump,
989 ref: status.Ref,
990 terminal: newTerminalState(),
991 cause: pump.cause(status.Ref.Physical),
992 }
993 require.NotNil(t, connection.cause)
994
995 // Retire stream 1 with a stream-local Reset, then retire more streams than
996 // the engine retains so its record is evicted before any reader consults it.
997 _, err := pump.Engine().Reset(Reset{
998 Physical: 1,
999 HasError: true,
1000 Error: ErrorDetail{Code: ports.BrokerErrorUnavailable, Text: "reset", FailureKind: domain.RemoteFailureInvalidResponse},
1001 })
1002 require.NoError(t, err)
1003 for i := 2; i <= MaxRetiredStreamRecords+2; i++ {
1004 id := PhysicalStreamID(i)
1005 require.NoError(t, pump.Engine().Open(openFor(id)))
1006 _, _ = pump.Engine().Close(Close{Physical: id})
1007 }
1008 _, tracked := pump.Engine().Status(1)
1009 require.False(t, tracked, "stream 1 record was evicted")
1010
1011 connection.syncFromEngine()
1012
1013 require.True(t, channelClosed(connection.Done()))
1014 var loss ports.BrokerStreamLost
1015 require.ErrorAs(t, connection.Err(), &loss)
1016 require.Equal(t, status.Ref.Client, loss.Stream)
1017 require.Equal(t, domain.RemoteFailureInvalidResponse, loss.Cause)
1018 require.NotEqual(t, domain.RemoteFailureNone, connection.FailureKind(),
1019 "a Reset never degrades into an orderly close")
1020}
1021
1022// TestLogicalWatcherCauseSurvivesEvictionStress retires a stream and then more
1023// streams than the engine retains before the watcher is scheduled, proving a
1024// late watcher still publishes the stream-local Reset as a loss instead of an
1025// orderly close.
1026func TestLogicalWatcherCauseSurvivesEvictionStress(t *testing.T) {
1027 const rounds = 12
1028 for round := 0; round < rounds; round++ {
1029 t.Run(fmt.Sprintf("round %d", round), func(t *testing.T) {
1030 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
1031 connector := mustConnector(t, brokerPump)
1032 deadline := time.Now().Add(15 * time.Second)
1033
1034 acceptor := newDaemonAcceptor(daemonPump, 1)
1035 acceptor.start(1, deadline)
1036 connection := openWithDeadline(t, connector, 1)
1037 require.NoError(t, acceptor.wait(1, deadline))
1038
1039 // Retire the stream with a stream-local Reset, then immediately
1040 // retire more stream records than the engine retains so a watcher
1041 // scheduled late observes an evicted record.
1042 _, err := brokerPump.Engine().Reset(Reset{
1043 Physical: 1,
1044 HasError: true,
1045 Error: ErrorDetail{Code: ports.BrokerErrorUnavailable, Text: "reset", FailureKind: domain.RemoteFailureInvalidResponse},
1046 })
1047 require.NoError(t, err)
1048 for i := 2; i <= MaxRetiredStreamRecords+2; i++ {
1049 id := PhysicalStreamID(i)
1050 require.NoError(t, brokerPump.Engine().Open(openFor(id)))
1051 _, _ = brokerPump.Engine().Close(Close{Physical: id})
1052 }
1053 _, tracked := brokerPump.Engine().Status(1)
1054 require.False(t, tracked, "stream 1 record was evicted")
1055
1056 require.Eventually(t, func() bool { return channelClosed(connection.Done()) }, 5*time.Second, time.Millisecond)
1057 var loss ports.BrokerStreamLost
1058 require.ErrorAs(t, connection.Err(), &loss,
1059 "a late watcher must not degrade a Reset into an orderly close")
1060 require.Equal(t, ports.BrokerStreamID(1), loss.Stream)
1061 require.Equal(t, domain.RemoteFailureInvalidResponse, loss.Cause)
1062 require.Zero(t, brokerPump.Engine().Live())
1063 acceptor.closePeers()
1064 })
1065 }
1066}
1067
1068// TestLogicalFreshOpenRefusedByDaemonCeiling proves the daemon's fresh-refusal
1069// path reaches the broker: the 129th physical identity is refused by the
1070// daemon's stream ceiling with one typed Mux Refused, and LogicalConnector
1071// classifies it as the admission limit without handing back a connection or
1072// recording a daemon-side stream.
1073func TestLogicalFreshOpenRefusedByDaemonCeiling(t *testing.T) {
1074 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
1075 connector := mustConnector(t, brokerPump)
1076
1077 // Fill the daemon's stream ceiling directly so the broker's own admission
1078 // stays free: the connector's next allocated identity is the 129th.
1079 for i := 1; i <= int(MaxMuxStreams); i++ {
1080 require.NoError(t, daemonPump.Engine().Open(openFor(PhysicalStreamID(i))))
1081 }
1082 require.Equal(t, int(MaxMuxStreams), daemonPump.Engine().Live())
1083 connector.next = uint64(MaxMuxStreams)
1084
1085 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
1086 defer cancel()
1087 connection, err := connector.Open(ctx, controlRequest(1))
1088 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
1089 require.Nil(t, connection)
1090
1091 refusedID := PhysicalStreamID(MaxMuxStreams + 1)
1092 require.Eventually(t, func() bool {
1093 status, ok := brokerPump.Engine().Status(refusedID)
1094 return ok && status.State == StreamTerminal
1095 }, 15*time.Second, time.Millisecond)
1096 status := mustStatus(t, brokerPump.Engine(), refusedID)
1097 require.True(t, status.Refused, "the daemon's fresh refusal reached the broker")
1098 require.Equal(t, uint32(1), status.Refusal.AdmissionCode)
1099
1100 // The daemon recorded no stream for the refused identity, the broker
1101 // released its local reservation, and the physical connection stays open.
1102 _, tracked := daemonPump.Engine().Status(refusedID)
1103 require.False(t, tracked, "a refused fresh open records no daemon stream")
1104 require.Equal(t, int(MaxMuxStreams), daemonPump.Engine().Live())
1105 require.Zero(t, brokerPump.Engine().Live(), "the broker released its local reservation")
1106 require.False(t, channelClosed(brokerPump.Done()))
1107 require.False(t, channelClosed(daemonPump.Done()))
1108}