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 // The scheduler only sends for an admitted stream, so the local
264 // settlement follows the send. The broker may react to the refusal
265 // with a Reset that terminalizes the stream first; that is a settled
266 // stream, not a failure.
267 err = a.pump.Engine().Refused(Refused{Ref: status.Ref, Error: refusal})
268 if errors.Is(err, ErrStreamState) {
269 if current, ok := a.pump.Engine().Status(id); ok && current.State == StreamTerminal {
270 err = nil
271 }
272 }
273 if err != nil {
274 a.fail(err)
275 return
276 }
277 a.add(id, nil)
278 }()
279}
280
281func (a *daemonAcceptor) add(id PhysicalStreamID, peer *sessionPeer) {
282 a.mu.Lock()
283 defer a.mu.Unlock()
284 a.count++
285 if peer != nil {
286 a.peers[id] = peer
287 }
288}
289
290func (a *daemonAcceptor) fail(err error) {
291 select {
292 case a.errCh <- err:
293 default:
294 }
295}
296
297func (a *daemonAcceptor) wait(count int, deadline time.Time) error {
298 for time.Now().Before(deadline) {
299 select {
300 case err := <-a.errCh:
301 return err
302 default:
303 }
304 a.mu.Lock()
305 accepted := a.count
306 a.mu.Unlock()
307 if accepted >= count {
308 return nil
309 }
310 time.Sleep(time.Millisecond)
311 }
312 a.mu.Lock()
313 accepted := a.count
314 a.mu.Unlock()
315 select {
316 case err := <-a.errCh:
317 return err
318 default:
319 }
320 return fmt.Errorf("daemonmux test: accepted %d of %d streams", accepted, count)
321}
322
323func (a *daemonAcceptor) closePeers() {
324 a.mu.Lock()
325 defer a.mu.Unlock()
326 for _, peer := range a.peers {
327 peer.close()
328 }
329}
330
331func openWithDeadline(t *testing.T, connector *LogicalConnector, stream int) typedLogical {
332 t.Helper()
333 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
334 defer cancel()
335 connection, err := connector.Open(ctx, controlRequest(stream))
336 require.NoError(t, err)
337 return newTypedLogical(connection)
338}
339
340// watchRegistered reports whether the pump still holds an inbound watch entry
341// for one physical stream.
342func watchRegistered(pump *Pump, physical PhysicalStreamID) bool {
343 pump.watchMu.Lock()
344 defer pump.watchMu.Unlock()
345 _, ok := pump.watch[physical]
346 return ok
347}
348
349func TestNewLogicalConnectorValidation(t *testing.T) {
350 t.Run("nil pump", func(t *testing.T) {
351 connector, err := NewLogicalConnector(nil)
352 require.ErrorIs(t, err, ErrLogicalConfig)
353 require.Nil(t, connector)
354 })
355 t.Run("server-side pump is refused", func(t *testing.T) {
356 pump, _ := newTestPump(t, DirectionClient)
357 connector, err := NewLogicalConnector(pump)
358 require.ErrorIs(t, err, ErrLogicalConfig)
359 require.Nil(t, connector)
360 })
361 t.Run("broker-side pump is accepted", func(t *testing.T) {
362 pump, _ := newTestPump(t, DirectionServer)
363 connector, err := NewLogicalConnector(pump)
364 require.NoError(t, err)
365 require.NotNil(t, connector)
366 })
367}
368
369// TestMuxStreamPipeChunkAndOrder proves the byte stream chunks outbound writes
370// at the negotiated ceiling in order and drains inbound chunks in order.
371func TestMuxStreamPipeChunkAndOrder(t *testing.T) {
372 pump, carrier := newTestPump(t, DirectionClient)
373 pump.Start(context.Background())
374 require.NoError(t, pump.Engine().Open(openFor(1)))
375 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
376
377 pipe := newMuxStreamPipe(pump, 1, 4, mustStatus(t, pump.Engine(), 1).Done)
378 written, err := pipe.Write([]byte("0123456789"))
379 require.NoError(t, err)
380 require.Equal(t, 10, written)
381
382 var chunks [][]byte
383 for i := 0; i < 3; i++ {
384 message, err := DecodeServer(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
385 require.NoError(t, err)
386 data, ok := message.(Data)
387 require.True(t, ok)
388 require.LessOrEqual(t, uint64(len(data.Data)), uint64(4))
389 chunks = append(chunks, data.Data)
390 }
391 require.Equal(t, [][]byte{[]byte("0123"), []byte("4567"), []byte("89")}, chunks)
392
393 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte("abc")})
394 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte("de")})
395 buffer := make([]byte, 3)
396 _, err = io.ReadFull(pipe, buffer)
397 require.NoError(t, err)
398 require.Equal(t, []byte("abc"), buffer)
399 buffer = make([]byte, 2)
400 _, err = io.ReadFull(pipe, buffer)
401 require.NoError(t, err)
402 require.Equal(t, []byte("de"), buffer)
403}
404
405// TestLogicalTwoConnections opens two typed logical connections over one
406// physical pair, exchanges a typed round trip on each, and closes them
407// independently.
408func TestLogicalTwoConnections(t *testing.T) {
409 const count = 2
410 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
411 connector := mustConnector(t, brokerPump)
412 deadline := time.Now().Add(30 * time.Second)
413
414 acceptor := newDaemonAcceptor(daemonPump, 1)
415 for i := 1; i <= count; i++ {
416 acceptor.start(PhysicalStreamID(i), deadline)
417 }
418
419 connections := make([]typedLogical, 0, count)
420 for i := 1; i <= count; i++ {
421 connection := openWithDeadline(t, connector, i)
422 require.NoError(t, connection.SendClient(protocol.Ping{}), "connection %d", i)
423 message, err := connection.ReceiveServer()
424 require.NoError(t, err, "connection %d", i)
425 require.Equal(t, protocol.Pong{}, message, "connection %d", i)
426 connections = append(connections, connection)
427 }
428 require.NoError(t, acceptor.wait(count, deadline))
429
430 for i, connection := range connections {
431 require.False(t, channelClosed(connection.Done()), "connection %d", i)
432 require.NoError(t, connection.Close(), "connection %d", i)
433 require.True(t, channelClosed(connection.Done()), "connection %d", i)
434 require.NoError(t, connection.Err(), "connection %d", i)
435 }
436 acceptor.closePeers()
437 require.False(t, channelClosed(brokerPump.Done()))
438 require.False(t, channelClosed(daemonPump.Done()))
439}
440
441// TestLogicalHundredConnectionsConcurrentAdmission opens 100 typed logical
442// connections concurrently over one physical pair, proves all 100 are live
443// before any traffic, interleaves a typed round trip on every stream, and then
444// drains and closes them. It fills the physical connection to its 128-stream
445// ceiling, proves the 129th stream is refused with the typed admission limit,
446// and proves a retry after a Close succeeds with a fresh monotonic identity.
447func TestLogicalHundredConnectionsConcurrentAdmission(t *testing.T) {
448 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
449 connector := mustConnector(t, brokerPump)
450 deadline := time.Now().Add(60 * time.Second)
451
452 const count = 100
453 const ceiling = int(MaxMuxStreams)
454 require.Equal(t, 128, ceiling, "the default physical connection admits 128 streams")
455
456 acceptor := newDaemonAcceptor(daemonPump, 1)
457 // Every physical identity the connector can allocate, plus the retry after
458 // the refused 129th attempt (which still consumes its monotonic identity).
459 for i := 1; i <= ceiling; i++ {
460 acceptor.start(PhysicalStreamID(i), deadline)
461 }
462 acceptor.start(PhysicalStreamID(ceiling+2), deadline)
463
464 openConcurrently := func(requests []ports.BrokerOpenStreamRequest) ([]*LogicalConnection, []error) {
465 connections := make([]*LogicalConnection, len(requests))
466 errs := make([]error, len(requests))
467 var wg sync.WaitGroup
468 for i, request := range requests {
469 wg.Add(1)
470 go func(index int, request ports.BrokerOpenStreamRequest) {
471 defer wg.Done()
472 ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
473 defer cancel()
474 connections[index], errs[index] = connector.Open(ctx, request)
475 }(i, request)
476 }
477 wg.Wait()
478 return connections, errs
479 }
480
481 requests := make([]ports.BrokerOpenStreamRequest, count)
482 for i := range requests {
483 requests[i] = controlRequest(i + 1)
484 }
485 connections, errs := openConcurrently(requests)
486 for i, err := range errs {
487 require.NoError(t, err, "connection %d", i)
488 require.NotNil(t, connections[i], "connection %d", i)
489 }
490 require.Equal(t, count, brokerPump.Engine().Live(), "all connections are open before any traffic")
491 require.NoError(t, acceptor.wait(count, deadline))
492
493 // Interleave a typed round trip across every stream, then drain it.
494 trafficErrs := make([]error, count)
495 var traffic sync.WaitGroup
496 for i := range connections {
497 traffic.Add(1)
498 go func(index int) {
499 defer traffic.Done()
500 typed := asTyped(connections[index])
501 if err := typed.SendClient(protocol.Ping{}); err != nil {
502 trafficErrs[index] = err
503 return
504 }
505 message, err := typed.ReceiveServer()
506 if err != nil {
507 trafficErrs[index] = err
508 return
509 }
510 if message != (protocol.Pong{}) {
511 trafficErrs[index] = fmt.Errorf("connection %d: received %v", index, message)
512 }
513 }(i)
514 }
515 traffic.Wait()
516 for i, err := range trafficErrs {
517 require.NoError(t, err, "connection %d", i)
518 }
519 require.Equal(t, count, brokerPump.Engine().Live(), "traffic never settles a live stream")
520
521 // Fill the physical connection to its concurrent-stream ceiling.
522 fill := make([]ports.BrokerOpenStreamRequest, ceiling-count)
523 for i := range fill {
524 fill[i] = controlRequest(count + i + 1)
525 }
526 extra, fillErrs := openConcurrently(fill)
527 for i, err := range fillErrs {
528 require.NoError(t, err, "extra connection %d", i)
529 }
530 require.Equal(t, ceiling, brokerPump.Engine().Live())
531 require.NoError(t, acceptor.wait(ceiling, deadline))
532
533 // The 129th stream is refused by the admission ceiling before it reaches
534 // the daemon.
535 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
536 defer cancel()
537 refused, err := connector.Open(ctx, controlRequest(1000))
538 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
539 require.Nil(t, refused)
540 require.Equal(t, ceiling, brokerPump.Engine().Live(), "a refused admission frees nothing")
541
542 // A Close frees one slot; the retry uses a fresh monotonic identity.
543 require.NoError(t, connections[0].Close())
544 require.True(t, channelClosed(connections[0].Done()))
545 require.Equal(t, ceiling-1, brokerPump.Engine().Live())
546 retry := openWithDeadline(t, connector, 1001)
547 require.NoError(t, retry.SendClient(protocol.Ping{}))
548 message, err := retry.ReceiveServer()
549 require.NoError(t, err)
550 require.Equal(t, protocol.Pong{}, message)
551 require.Equal(t, ceiling, brokerPump.Engine().Live())
552 require.NoError(t, acceptor.wait(ceiling+1, deadline))
553
554 for i := 1; i < len(connections); i++ {
555 require.False(t, channelClosed(connections[i].Done()), "connection %d", i)
556 require.NoError(t, connections[i].Close(), "connection %d", i)
557 require.True(t, channelClosed(connections[i].Done()), "connection %d", i)
558 require.NoError(t, connections[i].Err(), "connection %d", i)
559 }
560 for i := range extra {
561 require.NoError(t, extra[i].Close(), "extra connection %d", i)
562 }
563 require.NoError(t, retry.Close())
564 require.Zero(t, brokerPump.Engine().Live(), "every stream is retired")
565 acceptor.closePeers()
566 require.False(t, channelClosed(brokerPump.Done()))
567 require.False(t, channelClosed(daemonPump.Done()))
568}
569
570// TestLogicalBlockedReaderSiblingProgresses is the regression for a slow
571// remote consumer killing its own attachment. The daemon floods far more
572// frames than one stream window holds while the consumer is not reading: the
573// daemon's writer must wait for credit instead of resetting the stream, the
574// sibling keeps exchanging typed messages, and once the consumer reads again
575// every flooded frame arrives in order on the still-open stream.
576func TestLogicalBlockedReaderSiblingProgresses(t *testing.T) {
577 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
578 connector := mustConnector(t, brokerPump)
579 deadline := time.Now().Add(15 * time.Second)
580
581 // Each Pong is one tiny frame, so the flood is far more frames than the
582 // eight queued frames that used to reset the stream, and more credit than
583 // one window.
584 const flood = 20000
585 blockedAcceptor := newDaemonAcceptor(daemonPump, flood)
586 blockedAcceptor.start(1, deadline)
587 siblingAcceptor := newDaemonAcceptor(daemonPump, 1)
588 siblingAcceptor.start(2, deadline)
589
590 blocked := openWithDeadline(t, connector, 1)
591 sibling := openWithDeadline(t, connector, 2)
592 require.NoError(t, blockedAcceptor.wait(1, deadline))
593 require.NoError(t, siblingAcceptor.wait(1, deadline))
594
595 require.NoError(t, sibling.SendClient(protocol.Ping{}))
596 message, err := sibling.ReceiveServer()
597 require.NoError(t, err)
598 require.Equal(t, protocol.Pong{}, message)
599
600 // Trigger the flood, then do not read: the daemon fills the window and
601 // its writer waits for credit.
602 require.NoError(t, blocked.SendClient(protocol.Ping{}))
603 require.Eventually(t, func() bool {
604 status, ok := daemonPump.Engine().Status(1)
605 return ok && status.SendCredit < chunkCredit(int(MaxMuxChunkBytes))
606 }, 5*time.Second, time.Millisecond, "the daemon never exhausted its credit")
607 require.False(t, channelClosed(blocked.Done()), "a slow consumer must not reset its stream")
608
609 // The sibling still progresses and the physical connection is untouched.
610 require.NoError(t, sibling.SendClient(protocol.Ping{}))
611 message, err = sibling.ReceiveServer()
612 require.NoError(t, err)
613 require.Equal(t, protocol.Pong{}, message)
614 require.False(t, channelClosed(brokerPump.Done()))
615 require.Equal(t, StreamOpen, mustStatus(t, brokerPump.Engine(), 2).State)
616
617 // The consumer catches up: every flooded frame arrives on the same stream.
618 for i := 0; i < flood; i++ {
619 message, err := blocked.ReceiveServer()
620 require.NoError(t, err, "frame %d", i)
621 require.Equal(t, protocol.Pong{}, message)
622 }
623 require.False(t, channelClosed(blocked.Done()))
624 require.Equal(t, StreamOpen, mustStatus(t, brokerPump.Engine(), 1).State)
625}
626
627// TestLogicalMalformedInnerFrameResetsOnlyStream proves a malformed inner
628// session frame resets exactly its own stream and leaves the physical
629// connection and a sibling untouched.
630func TestLogicalMalformedInnerFrameResetsOnlyStream(t *testing.T) {
631 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
632 connector := mustConnector(t, brokerPump)
633 deadline := time.Now().Add(15 * time.Second)
634
635 bareAcceptor := newDaemonAcceptor(daemonPump, 1)
636 bareAcceptor.startBare(1, deadline)
637 siblingAcceptor := newDaemonAcceptor(daemonPump, 1)
638 siblingAcceptor.start(2, deadline)
639
640 malformed := openWithDeadline(t, connector, 1)
641 sibling := openWithDeadline(t, connector, 2)
642 require.NoError(t, bareAcceptor.wait(1, deadline))
643 require.NoError(t, siblingAcceptor.wait(1, deadline))
644
645 require.NoError(t, sibling.SendClient(protocol.Ping{}))
646 message, err := sibling.ReceiveServer()
647 require.NoError(t, err)
648 require.Equal(t, protocol.Pong{}, message)
649
650 // A zero-length inner frame is malformed: the client's first read fails and
651 // must reset only this stream.
652 require.NoError(t, daemonPump.SendData(1, []byte{0, 0, 0, 0}, nil))
653 _, err = malformed.ReceiveServer()
654 require.Error(t, err)
655 require.True(t, channelClosed(malformed.Done()))
656
657 var loss ports.BrokerStreamLost
658 require.ErrorAs(t, malformed.Err(), &loss)
659 require.Equal(t, ports.BrokerStreamID(1), loss.Stream)
660 require.Equal(t, ports.BrokerConnectionID{1}, loss.Connection)
661 require.Equal(t, ports.BrokerEpoch(1), loss.Epoch)
662 require.Equal(t, domain.RemoteFailureInvalidResponse, loss.Cause)
663
664 // The daemon observed exactly a stream-local reset.
665 require.Eventually(t, func() bool {
666 status, ok := daemonPump.Engine().Status(1)
667 return ok && status.State == StreamTerminal
668 }, 5*time.Second, time.Millisecond)
669 reset := mustStatus(t, daemonPump.Engine(), 1)
670 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.FailureKind)
671
672 // The sibling still progresses and the physical pumps stay open.
673 require.NoError(t, sibling.SendClient(protocol.Ping{}))
674 message, err = sibling.ReceiveServer()
675 require.NoError(t, err)
676 require.Equal(t, protocol.Pong{}, message)
677 require.False(t, channelClosed(brokerPump.Done()))
678 require.False(t, channelClosed(daemonPump.Done()))
679}
680
681// TestLogicalOpenCancellationReleasesReservation proves a cancelled open
682// releases its local reservation and a later open still succeeds.
683func TestLogicalOpenCancellationReleasesReservation(t *testing.T) {
684 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
685 connector := mustConnector(t, brokerPump)
686 deadline := time.Now().Add(15 * time.Second)
687
688 // No daemon acceptor: the open waits for an Opened that never arrives.
689 ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
690 defer cancel()
691 connection, err := connector.Open(ctx, controlRequest(1))
692 require.ErrorIs(t, err, context.DeadlineExceeded)
693 require.Nil(t, connection)
694
695 require.Eventually(t, func() bool {
696 status, ok := brokerPump.Engine().Status(1)
697 return ok && status.State == StreamTerminal
698 }, 5*time.Second, time.Millisecond)
699 require.False(t, watchRegistered(brokerPump, 1), "cancelled open leaked a watch entry")
700
701 // A sibling stream on the same physical connection still opens and works.
702 acceptor := newDaemonAcceptor(daemonPump, 1)
703 acceptor.start(2, deadline)
704 sibling := openWithDeadline(t, connector, 2)
705 require.NoError(t, acceptor.wait(1, deadline))
706 require.NoError(t, sibling.SendClient(protocol.Ping{}))
707 message, err := sibling.ReceiveServer()
708 require.NoError(t, err)
709 require.Equal(t, protocol.Pong{}, message)
710 require.False(t, channelClosed(brokerPump.Done()))
711}
712
713// TestLogicalOpenRefused proves a daemon refusal fails the open with the typed
714// refusal and leaves the physical connection open.
715func TestLogicalOpenRefused(t *testing.T) {
716 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
717 connector := mustConnector(t, brokerPump)
718 deadline := time.Now().Add(15 * time.Second)
719
720 acceptor := newDaemonAcceptor(daemonPump, 1)
721 acceptor.startRefused(1, deadline)
722
723 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
724 defer cancel()
725 connection, err := connector.Open(ctx, controlRequest(1))
726 require.Error(t, err)
727 require.Nil(t, connection)
728 require.NoError(t, acceptor.wait(1, deadline))
729 require.False(t, channelClosed(brokerPump.Done()))
730}
731
732// TestLogicalCloseIsolation proves closing one logical connection settles only
733// itself and leaves a sibling and the physical connection untouched.
734func TestLogicalCloseIsolation(t *testing.T) {
735 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
736 connector := mustConnector(t, brokerPump)
737 deadline := time.Now().Add(15 * time.Second)
738
739 firstAcceptor := newDaemonAcceptor(daemonPump, 1)
740 firstAcceptor.start(1, deadline)
741 secondAcceptor := newDaemonAcceptor(daemonPump, 1)
742 secondAcceptor.start(2, deadline)
743
744 first := openWithDeadline(t, connector, 1)
745 second := openWithDeadline(t, connector, 2)
746 require.NoError(t, firstAcceptor.wait(1, deadline))
747 require.NoError(t, secondAcceptor.wait(1, deadline))
748
749 require.NoError(t, first.Close())
750 require.True(t, channelClosed(first.Done()))
751 require.NoError(t, first.Err())
752 require.Equal(t, domain.RemoteFailureNone, first.FailureKind())
753 require.False(t, watchRegistered(brokerPump, 1), "closed logical connection leaked a watch entry")
754
755 // The daemon observed the orderly Close for stream 1.
756 require.Eventually(t, func() bool {
757 status, ok := daemonPump.Engine().Status(1)
758 return ok && status.State == StreamTerminal && status.Err == nil
759 }, 5*time.Second, time.Millisecond)
760
761 require.NoError(t, second.SendClient(protocol.Ping{}))
762 message, err := second.ReceiveServer()
763 require.NoError(t, err)
764 require.Equal(t, protocol.Pong{}, message)
765 require.False(t, channelClosed(brokerPump.Done()))
766 require.False(t, channelClosed(daemonPump.Done()))
767}
768
769// TestLogicalCloseDiscardsQueuedInboundReleasesAccounting proves a local Close
770// first discards the inbound chunks the engine already accepted for its stream:
771// the engine's orderly Close reaches its terminal state, releasing both the
772// stream's admission slot and its queued aggregate bytes, instead of parking
773// the stream in closing with a queue this side will never read again. Every
774// cycle closes with one queued inbound chunk under a two-stream ceiling, so a
775// leaked slot would refuse the third admission, and the peer still observes
776// exactly one orderly mux Close for each cycle.
777func TestLogicalCloseDiscardsQueuedInboundReleasesAccounting(t *testing.T) {
778 ceilings := DefaultMuxCeilings()
779 ceilings.MaxStreams = 2
780 pump, carrier := newTestPumpWithCeilings(t, DirectionServer, ceilings)
781 pump.Start(context.Background())
782
783 chunk := []byte("queued")
784 const cycles = 3
785 for i := 1; i <= cycles; i++ {
786 id := PhysicalStreamID(i)
787 require.NoError(t, pump.Engine().Open(openFor(id)), "cycle %d admission", i)
788 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
789 status := mustStatus(t, pump.Engine(), id)
790 connection := newLogicalConnection(pump, status.Ref, status.Done, pump.cause(id))
791
792 // The peer pipelined one chunk that this side never reads, so the stream
793 // holds accepted inbound data when the local Close arrives.
794 deliverServerFrame(t, carrier, Data{Physical: id, Data: chunk})
795 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.AggregateBytes() > 0 },
796 "cycle %d: the accepted chunk was never queued", i)
797
798 require.NoError(t, connection.Close())
799 require.NoError(t, connection.Close(), "Close is idempotent")
800 require.True(t, channelClosed(connection.Done()))
801 require.NoError(t, connection.Err())
802
803 closed := mustStatus(t, pump.Engine(), id)
804 require.Equal(t, StreamTerminal, closed.State, "cycle %d: the closed stream must be terminal", i)
805 require.Zero(t, closed.QueuedChunks, "cycle %d", i)
806 require.Zero(t, pump.Engine().Live(), "cycle %d: a closing stream leaked its admission slot", i)
807 require.Zero(t, pump.Engine().AggregateBytes(), "cycle %d: a closing stream leaked its queued bytes", i)
808 require.False(t, watchRegistered(pump, id), "cycle %d: the closed connection leaked a watch entry", i)
809
810 // Exactly one orderly mux Close reached the peer; the drain is local and
811 // never replaces it with a Reset.
812 message, err := DecodeClient(carrier.nextSent(t), testEnvelopeCeiling, testChunkCeiling)
813 require.NoError(t, err)
814 closedFrame, ok := message.(Close)
815 require.True(t, ok, "cycle %d: the peer frame is not a Close", i)
816 require.Equal(t, Close{Physical: id}, closedFrame)
817 carrier.requireNoSent(t, 20*time.Millisecond)
818 }
819 require.False(t, pump.Engine().Closed())
820 require.False(t, channelClosed(pump.Done()))
821}
822
823// TestLogicalPhysicalLossFanout proves a physical loss publishes one stable
824// ports.BrokerStreamLost per affected logical stream with its exact scope.
825func TestLogicalPhysicalLossFanout(t *testing.T) {
826 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
827 connector := mustConnector(t, brokerPump)
828 deadline := time.Now().Add(15 * time.Second)
829
830 const count = 3
831 acceptor := newDaemonAcceptor(daemonPump, 1)
832 for i := 1; i <= count; i++ {
833 acceptor.start(PhysicalStreamID(i), deadline)
834 }
835 connections := make([]typedLogical, 0, count)
836 for i := 1; i <= count; i++ {
837 connection := openWithDeadline(t, connector, i)
838 require.NoError(t, connection.SendClient(protocol.Ping{}))
839 message, err := connection.ReceiveServer()
840 require.NoError(t, err)
841 require.Equal(t, protocol.Pong{}, message)
842 connections = append(connections, connection)
843 }
844 require.NoError(t, acceptor.wait(count, deadline))
845
846 // Failing the daemon-side carrier is a physical loss on the broker pump.
847 require.NoError(t, daemonPump.Close())
848 require.Eventually(t, func() bool { return channelClosed(brokerPump.Done()) }, 5*time.Second, time.Millisecond)
849
850 for i, connection := range connections {
851 require.Eventually(t, func() bool { return channelClosed(connection.Done()) }, 5*time.Second, time.Millisecond)
852 var loss ports.BrokerStreamLost
853 require.ErrorAs(t, connection.Err(), &loss, "connection %d", i)
854 require.Equal(t, ports.BrokerStreamID(i+1), loss.Stream, "connection %d", i)
855 require.Equal(t, ports.BrokerConnectionID{1}, loss.Connection, "connection %d", i)
856 require.Equal(t, domain.RemoteFailureTransport, loss.Cause, "connection %d", i)
857 var boxed ports.BrokerError
858 require.ErrorAs(t, connection.Err(), &boxed, "connection %d", i)
859 require.Equal(t, ports.BrokerErrorAttachmentLost, boxed.Code, "connection %d", i)
860 }
861 require.Error(t, brokerPump.Err())
862 require.Equal(t, domain.RemoteFailureTransport, brokerPump.FailureKind())
863}
864
865// connections leaves no goroutine behind.
866func TestLogicalNoGoroutineLeak(t *testing.T) {
867 before := runtime.NumGoroutine()
868
869 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
870 connector := mustConnector(t, brokerPump)
871 deadline := time.Now().Add(15 * time.Second)
872
873 const count = 5
874 acceptor := newDaemonAcceptor(daemonPump, 1)
875 for i := 1; i <= count; i++ {
876 acceptor.start(PhysicalStreamID(i), deadline)
877 }
878 connections := make([]typedLogical, 0, count)
879 for i := 1; i <= count; i++ {
880 connection := openWithDeadline(t, connector, i)
881 require.NoError(t, connection.SendClient(protocol.Ping{}))
882 message, err := connection.ReceiveServer()
883 require.NoError(t, err)
884 require.Equal(t, protocol.Pong{}, message)
885 connections = append(connections, connection)
886 }
887 require.NoError(t, acceptor.wait(count, deadline))
888
889 for _, connection := range connections {
890 require.NoError(t, connection.Close())
891 }
892 acceptor.closePeers()
893 require.NoError(t, brokerPump.Close())
894 require.NoError(t, daemonPump.Close())
895
896 require.Eventually(t, func() bool { return runtime.NumGoroutine() <= before+3 }, 5*time.Second, time.Millisecond,
897 "logical connection goroutines leaked")
898}
899
900// TestLogicalPeerCloseReleasesGoroutines proves a stream terminated by a peer
901// mux Close joins its watcher and framed-transport goroutines without an
902// explicit local Close.
903func TestLogicalPeerCloseReleasesGoroutines(t *testing.T) {
904 before := runtime.NumGoroutine()
905
906 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
907 connector := mustConnector(t, brokerPump)
908 deadline := time.Now().Add(15 * time.Second)
909
910 acceptor := newDaemonAcceptor(daemonPump, 1)
911 acceptor.start(1, deadline)
912 connection := openWithDeadline(t, connector, 1)
913 require.NoError(t, acceptor.wait(1, deadline))
914 require.NoError(t, connection.SendClient(protocol.Ping{}))
915 message, err := connection.ReceiveServer()
916 require.NoError(t, err)
917 require.Equal(t, protocol.Pong{}, message)
918
919 // The daemon ends the stream orderly; the logical connection must settle
920 // itself and join its goroutines without a local Close.
921 if _, err := daemonPump.Engine().Close(Close{Physical: 1}); err != nil {
922 t.Fatalf("daemon close: %v", err)
923 }
924 require.NoError(t, daemonPump.Send(Close{Physical: 1}))
925 require.Eventually(t, func() bool { return channelClosed(connection.Done()) }, 5*time.Second, time.Millisecond)
926 require.NoError(t, connection.Err())
927
928 acceptor.closePeers()
929 require.NoError(t, brokerPump.Close())
930 require.NoError(t, daemonPump.Close())
931
932 require.Eventually(t, func() bool { return runtime.NumGoroutine() <= before+3 }, 5*time.Second, time.Millisecond,
933 "peer-closed logical connection goroutines leaked")
934}
935
936// TestLogicalPhysicalLossBlockedReadReturnsLoss proves a read blocked on an
937// idle logical stream wakes when the physical connection fails and reports a
938// typed ports.BrokerStreamLost rather than a clean io.EOF. The pump publishes
939// its terminal state before the engine terminalizes each stream, so the read
940// must consult the pump's physical cause instead of assuming an orderly end.
941func TestLogicalPhysicalLossBlockedReadReturnsLoss(t *testing.T) {
942 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
943 connector := mustConnector(t, brokerPump)
944 deadline := time.Now().Add(15 * time.Second)
945
946 acceptor := newDaemonAcceptor(daemonPump, 1)
947 acceptor.start(1, deadline)
948 connection := openWithDeadline(t, connector, 1)
949 require.NoError(t, acceptor.wait(1, deadline))
950
951 require.NoError(t, connection.SendClient(protocol.Ping{}))
952 message, err := connection.ReceiveServer()
953 require.NoError(t, err)
954 require.Equal(t, protocol.Pong{}, message)
955
956 type result struct {
957 message protocol.ServerMessage
958 err error
959 }
960 blocked := make(chan result, 1)
961 go func() {
962 msg, err := connection.ReceiveServer()
963 blocked <- result{message: msg, err: err}
964 }()
965 // Let the reader reach the inner stream and block there.
966 time.Sleep(50 * time.Millisecond)
967 require.False(t, channelClosed(connection.Done()), "the read is blocked, not settled")
968
969 // Fail the daemon side; the broker observes a physical loss.
970 require.NoError(t, daemonPump.Close())
971
972 select {
973 case got := <-blocked:
974 require.Nil(t, got.message)
975 require.Error(t, got.err)
976 require.NotErrorIs(t, got.err, io.EOF, "a physical failure must not surface as a clean EOF")
977 var loss ports.BrokerStreamLost
978 require.ErrorAs(t, got.err, &loss)
979 require.Equal(t, ports.BrokerStreamID(1), loss.Stream)
980 require.Equal(t, domain.RemoteFailureTransport, loss.Cause)
981 case <-time.After(5 * time.Second):
982 t.Fatal("blocked read did not return after the physical failure")
983 }
984 require.True(t, channelClosed(connection.Done()))
985}
986
987// TestLogicalTerminalCauseSurvivesEngineEviction proves a settled logical
988// connection reports its exact stream-local cause after the engine evicts the
989// bounded record. The retained authority outlives the record, so a Reset can
990// never degrade into an orderly close for a reader that observes the end late.
991func TestLogicalTerminalCauseSurvivesEngineEviction(t *testing.T) {
992 pump, _ := newTestPump(t, DirectionServer)
993 require.NoError(t, pump.Engine().Open(openFor(1)))
994 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
995 status := mustStatus(t, pump.Engine(), 1)
996
997 connection := &LogicalConnection{
998 pump: pump,
999 ref: status.Ref,
1000 terminal: newTerminalState(),
1001 cause: pump.cause(status.Ref.Physical),
1002 }
1003 require.NotNil(t, connection.cause)
1004
1005 // Retire stream 1 with a stream-local Reset, then retire more streams than
1006 // the engine retains so its record is evicted before any reader consults it.
1007 _, err := pump.Engine().Reset(Reset{
1008 Physical: 1,
1009 HasError: true,
1010 Error: ErrorDetail{Code: ports.BrokerErrorUnavailable, Text: "reset", FailureKind: domain.RemoteFailureInvalidResponse},
1011 })
1012 require.NoError(t, err)
1013 for i := 2; i <= MaxRetiredStreamRecords+2; i++ {
1014 id := PhysicalStreamID(i)
1015 require.NoError(t, pump.Engine().Open(openFor(id)))
1016 _, _ = pump.Engine().Close(Close{Physical: id})
1017 }
1018 _, tracked := pump.Engine().Status(1)
1019 require.False(t, tracked, "stream 1 record was evicted")
1020
1021 connection.syncFromEngine()
1022
1023 require.True(t, channelClosed(connection.Done()))
1024 var loss ports.BrokerStreamLost
1025 require.ErrorAs(t, connection.Err(), &loss)
1026 require.Equal(t, status.Ref.Client, loss.Stream)
1027 require.Equal(t, domain.RemoteFailureInvalidResponse, loss.Cause)
1028 require.NotEqual(t, domain.RemoteFailureNone, connection.FailureKind(),
1029 "a Reset never degrades into an orderly close")
1030}
1031
1032// TestLogicalWatcherCauseSurvivesEvictionStress retires a stream and then more
1033// streams than the engine retains before the watcher is scheduled, proving a
1034// late watcher still publishes the stream-local Reset as a loss instead of an
1035// orderly close.
1036func TestLogicalWatcherCauseSurvivesEvictionStress(t *testing.T) {
1037 const rounds = 12
1038 for round := 0; round < rounds; round++ {
1039 t.Run(fmt.Sprintf("round %d", round), func(t *testing.T) {
1040 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
1041 connector := mustConnector(t, brokerPump)
1042 deadline := time.Now().Add(15 * time.Second)
1043
1044 acceptor := newDaemonAcceptor(daemonPump, 1)
1045 acceptor.start(1, deadline)
1046 connection := openWithDeadline(t, connector, 1)
1047 require.NoError(t, acceptor.wait(1, deadline))
1048
1049 // Retire the stream with a stream-local Reset, then immediately
1050 // retire more stream records than the engine retains so a watcher
1051 // scheduled late observes an evicted record.
1052 _, err := brokerPump.Engine().Reset(Reset{
1053 Physical: 1,
1054 HasError: true,
1055 Error: ErrorDetail{Code: ports.BrokerErrorUnavailable, Text: "reset", FailureKind: domain.RemoteFailureInvalidResponse},
1056 })
1057 require.NoError(t, err)
1058 for i := 2; i <= MaxRetiredStreamRecords+2; i++ {
1059 id := PhysicalStreamID(i)
1060 require.NoError(t, brokerPump.Engine().Open(openFor(id)))
1061 _, _ = brokerPump.Engine().Close(Close{Physical: id})
1062 }
1063 _, tracked := brokerPump.Engine().Status(1)
1064 require.False(t, tracked, "stream 1 record was evicted")
1065
1066 require.Eventually(t, func() bool { return channelClosed(connection.Done()) }, 5*time.Second, time.Millisecond)
1067 var loss ports.BrokerStreamLost
1068 require.ErrorAs(t, connection.Err(), &loss,
1069 "a late watcher must not degrade a Reset into an orderly close")
1070 require.Equal(t, ports.BrokerStreamID(1), loss.Stream)
1071 require.Equal(t, domain.RemoteFailureInvalidResponse, loss.Cause)
1072 require.Zero(t, brokerPump.Engine().Live())
1073 acceptor.closePeers()
1074 })
1075 }
1076}
1077
1078// TestLogicalFreshOpenRefusedByDaemonCeiling proves the daemon's fresh-refusal
1079// path reaches the broker: the 129th physical identity is refused by the
1080// daemon's stream ceiling with one typed Mux Refused, and LogicalConnector
1081// classifies it as the admission limit without handing back a connection or
1082// recording a daemon-side stream.
1083func TestLogicalFreshOpenRefusedByDaemonCeiling(t *testing.T) {
1084 brokerPump, daemonPump := newPairedPumps(t, DefaultMuxCeilings())
1085 connector := mustConnector(t, brokerPump)
1086
1087 // Fill the daemon's stream ceiling directly so the broker's own admission
1088 // stays free: the connector's next allocated identity is the 129th.
1089 for i := 1; i <= int(MaxMuxStreams); i++ {
1090 require.NoError(t, daemonPump.Engine().Open(openFor(PhysicalStreamID(i))))
1091 }
1092 require.Equal(t, int(MaxMuxStreams), daemonPump.Engine().Live())
1093 connector.next = uint64(MaxMuxStreams)
1094
1095 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
1096 defer cancel()
1097 connection, err := connector.Open(ctx, controlRequest(1))
1098 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
1099 require.Nil(t, connection)
1100
1101 refusedID := PhysicalStreamID(MaxMuxStreams + 1)
1102 require.Eventually(t, func() bool {
1103 status, ok := brokerPump.Engine().Status(refusedID)
1104 return ok && status.State == StreamTerminal
1105 }, 15*time.Second, time.Millisecond)
1106 status := mustStatus(t, brokerPump.Engine(), refusedID)
1107 require.True(t, status.Refused, "the daemon's fresh refusal reached the broker")
1108 require.Equal(t, uint32(1), status.Refusal.AdmissionCode)
1109
1110 // The daemon recorded no stream for the refused identity, the broker
1111 // released its local reservation, and the physical connection stays open.
1112 _, tracked := daemonPump.Engine().Status(refusedID)
1113 require.False(t, tracked, "a refused fresh open records no daemon stream")
1114 require.Equal(t, int(MaxMuxStreams), daemonPump.Engine().Live())
1115 require.Zero(t, brokerPump.Engine().Live(), "the broker released its local reservation")
1116 require.False(t, channelClosed(brokerPump.Done()))
1117 require.False(t, channelClosed(daemonPump.Done()))
1118}