streamio: Clarify why read errors are deferred

The implementation of `Read()` only returns read errors in case no data
is left in the buffer. This is required such that we correctly drain any
buffered data and only return the error when no data is left. This may
not be immediately clear, so this commit puts a comment on this and adds
a testcase which exercises this code path.

2 files changed, 31 insertions, 9 deletions

diff --git a/streamio/stream.go b/streamio/stream.go
index b82fb40fa..d3074d9f8 100644
--- a/streamio/stream.go
+++ b/streamio/stream.go
@@ -53,11 +53,17 @@ func (rr *receiveReader) Read(p []byte) (int, error) {
 	if len(rr.data) == 0 {
 		rr.data, rr.err = rr.receiver()
 	}
+
 	n := copy(p, rr.data)
 	rr.data = rr.data[n:]
+
+	// We want to return any potential error only in case we have no
+	// buffered data left. Otherwise, it can happen that we do not relay
+	// bytes when the reader returns both data and an error.
 	if len(rr.data) == 0 {
 		return n, rr.err
 	}
+
 	return n, nil
 }
 
diff --git a/streamio/stream_test.go b/streamio/stream_test.go
index fbc2944cf..ee45f7196 100644
--- a/streamio/stream_test.go
+++ b/streamio/stream_test.go
@@ -33,16 +33,32 @@ func TestReceiveSources(t *testing.T) {
 }
 
 func TestReadSizes(t *testing.T) {
-	testData := "Hello this is the test data that will be received. It goes on for a while bla bla bla."
-	for n := 1; n < 100; n *= 3 {
-		desc := fmt.Sprintf("reads of size %d", n)
-		result := &bytes.Buffer{}
-		reader := &opaqueReader{NewReader(receiverFromReader(strings.NewReader(testData)))}
-		_, err := io.CopyBuffer(&opaqueWriter{result}, reader, make([]byte, n))
-
-		require.NoError(t, err, desc)
-		require.Equal(t, testData, result.String())
+	readSizes := func(t *testing.T, newReader func(string) io.Reader) {
+		testData := "Hello this is the test data that will be received. It goes on for a while bla bla bla."
+
+		for n := 1; n < 100; n *= 3 {
+			desc := fmt.Sprintf("reads of size %d", n)
+			result := &bytes.Buffer{}
+			reader := &opaqueReader{NewReader(receiverFromReader(newReader(testData)))}
+			n, err := io.CopyBuffer(&opaqueWriter{result}, reader, make([]byte, n))
+
+			require.NoError(t, err, desc)
+			require.Equal(t, testData, result.String())
+			require.EqualValues(t, len(testData), n)
+		}
 	}
+
+	t.Run("normal reader", func(t *testing.T) {
+		readSizes(t, func(s string) io.Reader {
+			return strings.NewReader(s)
+		})
+	})
+
+	t.Run("err reader", func(t *testing.T) {
+		readSizes(t, func(s string) io.Reader {
+			return iotest.DataErrReader(strings.NewReader(s))
+		})
+	})
 }
 
 func TestWriterTo(t *testing.T) {