1package brokeripc
2
3import (
4 "bytes"
5 "errors"
6 "io"
7 "testing"
8 "time"
9
10 "github.com/bnema/vev/internal/adapters/brokerwire"
11 "github.com/bnema/vev/internal/protocol"
12 "github.com/stretchr/testify/require"
13)
14
15// TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose proves closeWith
16// keeps chunks that are already queued when its cause is nil, an orderly
17// close: Read drains every queued byte before it returns io.EOF. This is
18// exactly what a reply queued just ahead of an orderly StreamClosed needs -
19// the reply chunk must survive the close instead of being thrown away
20// underneath its reader, which is root cause 1 of the broker reply loss
21// ("vev ls" reporting a lost/unknown outcome on a successful call). An error
22// cause is the opposite: the queued content can no longer be trusted once the
23// pipe is torn down for cause, so it is discarded and Read observes the error
24// immediately instead of replaying stale data.
25func TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose(t *testing.T) {
26 errCause := errors.New("stream pipe test: cause")
27
28 tests := []struct {
29 name string
30 chunks [][]byte
31 cause error
32 readSize int
33 wantData []byte
34 wantErr error
35 }{
36 {
37 name: "orderly close keeps one queued chunk",
38 chunks: [][]byte{[]byte("hello")},
39 cause: nil,
40 readSize: 64,
41 wantData: []byte("hello"),
42 wantErr: io.EOF,
43 },
44 {
45 name: "orderly close keeps three chunks split across a frame boundary",
46 chunks: [][]byte{[]byte("AB"), []byte("CD"), []byte("EF")},
47 cause: nil,
48 // Smaller than the combined payload, so one logical frame's worth of
49 // queued data is drained across several Read calls, exactly like
50 // io.ReadFull reading a header then a body out of the same queue.
51 readSize: 3,
52 wantData: []byte("ABCDEF"),
53 wantErr: io.EOF,
54 },
55 {
56 name: "error cause discards the queue and Read returns the error",
57 chunks: [][]byte{[]byte("hello")},
58 cause: errCause,
59 readSize: 64,
60 wantData: nil,
61 wantErr: errCause,
62 },
63 {
64 name: "orderly close with no queued data is immediate EOF",
65 chunks: nil,
66 cause: nil,
67 readSize: 64,
68 wantData: nil,
69 wantErr: io.EOF,
70 },
71 }
72
73 for _, tt := range tests {
74 t.Run(tt.name, func(t *testing.T) {
75 p := newTestPipe()
76 for _, chunk := range tt.chunks {
77 require.NoError(t, p.deliver(chunk))
78 }
79 p.closeWith(tt.cause)
80
81 var got bytes.Buffer
82 buf := make([]byte, tt.readSize)
83 var readErr error
84 for {
85 n, err := p.Read(buf)
86 got.Write(buf[:n])
87 if err != nil {
88 readErr = err
89 break
90 }
91 }
92 require.Equal(t, tt.wantData, got.Bytes())
93 if errors.Is(tt.wantErr, io.EOF) {
94 require.ErrorIs(t, readErr, io.EOF)
95 } else {
96 require.Equal(t, tt.wantErr, readErr)
97 }
98
99 // closeWith is idempotent: a second call with a different cause never
100 // resurrects a discarded queue or changes the already-observed outcome.
101 p.closeWith(errors.New("stream pipe test: second cause"))
102 n, err := p.Read(buf)
103 require.Zero(t, n)
104 if errors.Is(tt.wantErr, io.EOF) {
105 require.ErrorIs(t, err, io.EOF)
106 } else {
107 require.Equal(t, tt.wantErr, err)
108 }
109 })
110 }
111}
112
113// newTestPipe is a pipe whose outbound writes and credit grants go nowhere.
114func newTestPipe() *streamPipe {
115 return newStreamPipe(int(brokerwire.MaxStreamChunkBytes), func([]byte) error { return nil }, func(uint64) error { return nil })
116}
117
118// TestStreamWindowCoversTheSessionOutputWindow pins the sizing relation
119// between the two flow-control layers: one broker stream window holds the
120// daemon's unacknowledged byte budget, so the daemon is mostly paced by its own
121// ACK window and coalesces frames. Credit may still pace it at the margin: the
122// receiver batches up to half a window before returning it, chunks cost 64
123// bytes of overhead, and one frame may overrun the byte budget (a big image).
124// Credit stays the hard bound either way.
125func TestStreamWindowCoversTheSessionOutputWindow(t *testing.T) {
126 require.GreaterOrEqual(t, brokerwire.StreamWindowBytes, uint64(protocol.MaxOutputWindowBytes),
127 "one stream window must hold the daemon's whole unacknowledged byte budget")
128}
129
130// TestStreamPipeCredit covers the receive and send sides of stream credit:
131// a peer may send up to one window, the consumer's reads return credit in
132// batches, a sender waits instead of failing while it has none, and grants or
133// data beyond the window are flow-control violations.
134func TestStreamPipeCredit(t *testing.T) {
135 chunk := int(brokerwire.MaxStreamChunkBytes)
136 cost := brokerwire.StreamChunkCredit(chunk)
137 perWindow := int(brokerwire.StreamWindowBytes / cost)
138
139 t.Run("a full window is accepted and one more frame is a violation", func(t *testing.T) {
140 p := newTestPipe()
141 for range perWindow {
142 require.NoError(t, p.deliver(make([]byte, chunk)))
143 }
144 require.ErrorIs(t, p.deliver(make([]byte, chunk)), ErrStreamCredit)
145 })
146
147 t.Run("reads return credit in batches", func(t *testing.T) {
148 var grants []uint64
149 p := newStreamPipe(int(brokerwire.MaxStreamChunkBytes), func([]byte) error { return nil }, func(c uint64) error { grants = append(grants, c); return nil })
150 for range perWindow {
151 require.NoError(t, p.deliver(make([]byte, chunk)))
152 }
153 buf := make([]byte, chunk)
154 for range perWindow {
155 _, err := io.ReadFull(p, buf)
156 require.NoError(t, err)
157 }
158 require.NotEmpty(t, grants)
159 total := uint64(0)
160 for _, g := range grants {
161 require.GreaterOrEqual(t, g, streamCreditReturnAt(brokerwire.StreamWindowBytes))
162 total += g
163 }
164 require.LessOrEqual(t, total, uint64(perWindow)*cost)
165 // Once the returned credit reaches the peer, the window is open again.
166 require.NoError(t, p.deliver(make([]byte, chunk)))
167 })
168
169 t.Run("a sender waits for credit and resumes on a grant", func(t *testing.T) {
170 p := newTestPipe()
171 for range perWindow {
172 require.NoError(t, p.reserve(chunk))
173 }
174 reserved := make(chan error, 1)
175 go func() { reserved <- p.reserve(chunk) }()
176 select {
177 case err := <-reserved:
178 t.Fatalf("reserve returned %v without credit", err)
179 case <-time.After(20 * time.Millisecond):
180 }
181 require.NoError(t, p.granted(cost))
182 select {
183 case err := <-reserved:
184 require.NoError(t, err)
185 case <-time.After(5 * time.Second):
186 t.Fatal("a grant must release a sender waiting for credit")
187 }
188 })
189
190 t.Run("close releases a sender waiting for credit", func(t *testing.T) {
191 p := newTestPipe()
192 for range perWindow {
193 require.NoError(t, p.reserve(chunk))
194 }
195 reserved := make(chan error, 1)
196 go func() { reserved <- p.reserve(chunk) }()
197 select {
198 case err := <-reserved:
199 t.Fatalf("reserve returned %v without credit", err)
200 case <-time.After(20 * time.Millisecond):
201 }
202 p.closeWith(ErrStreamGone)
203 select {
204 case err := <-reserved:
205 require.ErrorIs(t, err, ErrStreamGone)
206 case <-time.After(5 * time.Second):
207 t.Fatal("close must release a sender waiting for credit")
208 }
209 })
210
211 t.Run("a grant beyond the window is a violation", func(t *testing.T) {
212 p := newTestPipe()
213 require.ErrorIs(t, p.granted(1), ErrStreamCredit)
214 require.NoError(t, p.reserve(chunk))
215 require.NoError(t, p.granted(cost))
216 require.ErrorIs(t, p.granted(1), ErrStreamCredit)
217 })
218}
219
220// TestClientStreamPrefersTheFullOutputWindow pins what a real broker-routed
221// attachment reports: its carriage is a reliable stream, not a datagram link,
222// so the client claims the full output window and frames pipeline.
223func TestClientStreamPrefersTheFullOutputWindow(t *testing.T) {
224 st, err := newClientStream(&client{}, 1)
225 require.NoError(t, err)
226 require.Equal(t, uint8(protocol.MaxOutputWindow), st.Capabilities().PreferredOutputWindow)
227}