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) {