1package daemonmux
2
3import (
4 "context"
5 "io"
6 "testing"
7
8 "github.com/stretchr/testify/require"
9)
10
11// TestPumpInboundDataClosePipelineDrainsThenEOF proves a pipelined inbound
12// Data followed by an orderly Close stays readable: the consumer drains the
13// accepted bytes and only then observes io.EOF, at which point the stream is
14// terminal and its bytes are released.
15func TestPumpInboundDataClosePipelineDrainsThenEOF(t *testing.T) {
16 pump, carrier := newTestPump(t, DirectionClient)
17 pump.Start(context.Background())
18 require.NoError(t, pump.Engine().Open(openFor(1)))
19 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
20 status := mustStatus(t, pump.Engine(), 1)
21 pipe := newMuxStreamPipe(pump, 1, pump.Ceilings().StreamChunkLimit, status.Done)
22
23 // The peer pipelines one Data frame and its Close before the consumer reads.
24 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte("hello")})
25 deliverClientFrame(t, carrier, Close{Physical: 1})
26
27 buffer := make([]byte, len("hello"))
28 _, err := io.ReadFull(pipe, buffer)
29 require.NoError(t, err)
30 require.Equal(t, "hello", string(buffer))
31 // The stream ends only after the preserved bytes are drained.
32 _, err = pipe.Read(make([]byte, 1))
33 require.ErrorIs(t, err, io.EOF)
34
35 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 0 })
36 status = mustStatus(t, pump.Engine(), 1)
37 require.Equal(t, StreamTerminal, status.State)
38 require.True(t, channelClosed(status.Done))
39 require.Zero(t, status.QueuedChunks)
40 require.Zero(t, pump.Engine().AggregateBytes())
41 require.NoError(t, status.Err)
42 require.False(t, pump.Engine().Closed())
43 require.False(t, channelClosed(pump.Done()))
44}
45
46// TestPumpInboundCloseSiblingIsolation proves an orderly Close with unread data
47// settles only its own stream once drained: a sibling keeps its state, its
48// queued chunk, and its admission slot while the closing stream's data is
49// preserved. Draining the sibling never closes it.
50func TestPumpInboundCloseSiblingIsolation(t *testing.T) {
51 pump, carrier := newTestPump(t, DirectionClient)
52 pump.Start(context.Background())
53 require.NoError(t, pump.Engine().Open(openFor(1)))
54 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
55 require.NoError(t, pump.Engine().Open(openFor(2)))
56 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(2)}))
57
58 deliverClientFrame(t, carrier, Data{Physical: 1, Data: []byte("one")})
59 deliverClientFrame(t, carrier, Data{Physical: 2, Data: []byte("two")})
60 deliverClientFrame(t, carrier, Close{Physical: 1})
61
62 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
63 status, ok := e.Status(1)
64 return ok && status.State == StreamClosing
65 })
66 sibling := mustStatus(t, pump.Engine(), 2)
67 require.Equal(t, StreamOpen, sibling.State, "the sibling is untouched by the close")
68 require.Equal(t, 1, sibling.QueuedChunks)
69
70 chunk, ok := takeEventually(t, pump, 2)
71 require.True(t, ok)
72 require.Equal(t, []byte("two"), chunk)
73 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 2).State, "draining a sibling never closes it")
74
75 chunk, ok = takeEventually(t, pump, 1)
76 require.True(t, ok)
77 require.Equal(t, []byte("one"), chunk)
78 require.Equal(t, StreamTerminal, mustStatus(t, pump.Engine(), 1).State)
79 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 2).State)
80 require.Equal(t, 1, pump.Engine().Live())
81 require.False(t, pump.Engine().Closed())
82 require.False(t, channelClosed(pump.Done()))
83}
84
85// TestPumpInboundCloseChurnAccountingBounded proves the closing accounting is
86// balanced and bounded across many inbound Data-then-Close cycles: each closing
87// stream holds exactly its unread bytes until the draining Take releases them,
88// and the engine's retained record set never grows past its retention bound.
89func TestPumpInboundCloseChurnAccountingBounded(t *testing.T) {
90 pump, _ := newTestPump(t, DirectionClient)
91 pump.Start(context.Background())
92 const churn = int(MaxMuxStreams) * 8
93 for i := 1; i <= churn; i++ {
94 id := PhysicalStreamID(i)
95 require.NoError(t, pump.Engine().Open(openFor(id)))
96 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
97 require.NoError(t, pump.applyClient(Data{Physical: id, Data: []byte("payload")}))
98 require.NoError(t, pump.applyClient(Close{Physical: id}))
99
100 require.Equal(t, StreamClosing, mustStatus(t, pump.Engine(), id).State)
101 require.Equal(t, 1, pump.Engine().Live(), "the closing stream still holds its slot")
102 require.Equal(t, len("payload"), pump.Engine().AggregateBytes())
103
104 chunk, ok := pump.Take(id)
105 require.True(t, ok)
106 require.Equal(t, []byte("payload"), chunk)
107 require.Equal(t, StreamTerminal, mustStatus(t, pump.Engine(), id).State)
108 require.Zero(t, pump.Engine().Live())
109 require.Zero(t, pump.Engine().AggregateBytes())
110 }
111 require.Len(t, pump.Engine().retired, MaxRetiredStreamRecords)
112 require.LessOrEqual(t, len(pump.Engine().streams), MaxRetiredStreamRecords)
113 require.False(t, pump.Engine().Closed())
114 require.False(t, channelClosed(pump.Done()))
115}
116
117// TestPumpInboundResetDiscardsClosingQueue proves a Reset still abandons a
118// closing stream's preserved queue immediately, releasing the bytes and the
119// slot without waiting for a consumer.
120func TestPumpInboundResetDiscardsClosingQueue(t *testing.T) {
121 pump, _ := newTestPump(t, DirectionClient)
122 require.NoError(t, pump.Engine().Open(openFor(1)))
123 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(1)}))
124 require.NoError(t, pump.applyClient(Data{Physical: 1, Data: []byte("queued")}))
125 require.NoError(t, pump.applyClient(Close{Physical: 1}))
126 require.Equal(t, StreamClosing, mustStatus(t, pump.Engine(), 1).State)
127
128 require.NoError(t, pump.applyClient(Reset{Physical: 1}))
129
130 status := mustStatus(t, pump.Engine(), 1)
131 require.Equal(t, StreamTerminal, status.State)
132 require.True(t, channelClosed(status.Done))
133 require.Zero(t, status.QueuedChunks)
134 require.Zero(t, pump.Engine().AggregateBytes())
135 require.Zero(t, pump.Engine().Live())
136 _, ok := pump.Take(1)
137 require.False(t, ok, "the reset discarded the preserved queue")
138}