1package brokeripc
2
3import (
4 "bytes"
5 "errors"
6 "io"
7 "testing"
8
9 "github.com/stretchr/testify/require"
10)
11
12// TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose proves closeWith
13// keeps chunks that are already queued when its cause is nil, an orderly
14// close: Read drains every queued byte before it returns io.EOF. This is
15// exactly what a reply queued just ahead of an orderly StreamClosed needs -
16// the reply chunk must survive the close instead of being thrown away
17// underneath its reader, which is root cause 1 of the broker reply loss
18// ("vev ls" reporting a lost/unknown outcome on a successful call). An error
19// cause is the opposite: the queued content can no longer be trusted once the
20// pipe is torn down for cause, so it is discarded and Read observes the error
21// immediately instead of replaying stale data.
22func TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose(t *testing.T) {
23 errCause := errors.New("stream pipe test: cause")
24
25 tests := []struct {
26 name string
27 chunks [][]byte
28 cause error
29 readSize int
30 wantData []byte
31 wantErr error
32 }{
33 {
34 name: "orderly close keeps one queued chunk",
35 chunks: [][]byte{[]byte("hello")},
36 cause: nil,
37 readSize: 64,
38 wantData: []byte("hello"),
39 wantErr: io.EOF,
40 },
41 {
42 name: "orderly close keeps three chunks split across a frame boundary",
43 chunks: [][]byte{[]byte("AB"), []byte("CD"), []byte("EF")},
44 cause: nil,
45 // Smaller than the combined payload, so one logical frame's worth of
46 // queued data is drained across several Read calls, exactly like
47 // io.ReadFull reading a header then a body out of the same queue.
48 readSize: 3,
49 wantData: []byte("ABCDEF"),
50 wantErr: io.EOF,
51 },
52 {
53 name: "error cause discards the queue and Read returns the error",
54 chunks: [][]byte{[]byte("hello")},
55 cause: errCause,
56 readSize: 64,
57 wantData: nil,
58 wantErr: errCause,
59 },
60 {
61 name: "orderly close with no queued data is immediate EOF",
62 chunks: nil,
63 cause: nil,
64 readSize: 64,
65 wantData: nil,
66 wantErr: io.EOF,
67 },
68 }
69
70 for _, tt := range tests {
71 t.Run(tt.name, func(t *testing.T) {
72 p := newStreamPipe(0, 0, func([]byte) error { return nil })
73 for _, chunk := range tt.chunks {
74 require.NoError(t, p.deliver(chunk))
75 }
76 p.closeWith(tt.cause)
77
78 var got bytes.Buffer
79 buf := make([]byte, tt.readSize)
80 var readErr error
81 for {
82 n, err := p.Read(buf)
83 got.Write(buf[:n])
84 if err != nil {
85 readErr = err
86 break
87 }
88 }
89 require.Equal(t, tt.wantData, got.Bytes())
90 if errors.Is(tt.wantErr, io.EOF) {
91 require.ErrorIs(t, readErr, io.EOF)
92 } else {
93 require.Equal(t, tt.wantErr, readErr)
94 }
95
96 // closeWith is idempotent: a second call with a different cause never
97 // resurrects a discarded queue or changes the already-observed outcome.
98 p.closeWith(errors.New("stream pipe test: second cause"))
99 n, err := p.Read(buf)
100 require.Zero(t, n)
101 if errors.Is(tt.wantErr, io.EOF) {
102 require.ErrorIs(t, err, io.EOF)
103 } else {
104 require.Equal(t, tt.wantErr, err)
105 }
106 })
107 }
108}