Do not fail decoding on first read error (#17)

* Do not fail decoding on first read error

Try decoding with the rest successfully read inputs.

* some small code improvements

* Allocate map memory upfront

pkg/eestream/decode.go

@@ -19,6 +19,11 @@ type decodedReader struct {
 	err    error
 }
 
+type readerError struct {
+	i   int // reader index in the map
+	err error
+}
+
 // DecodeReaders takes a map of readers and an ErasureScheme returning a
 // combined Reader. The map, 'rs', must be a mapping of erasure piece numbers
 // to erasure piece streams.
@@ -47,25 +52,29 @@ func (dr *decodedReader) Read(p []byte) (n int, err error) {
 		// the channel has a buffer size to contain all the errors
 		// even if we read none, so we can return without receiving
 		// every channel value
-		errs := make(chan error, len(dr.rs))
+		errs := make(chan readerError, len(dr.rs))
 		for i := range dr.rs {
 			go func(i int) {
 				// fill the buffer from the piece input
 				_, err := io.ReadFull(dr.rs[i], dr.inbufs[i])
-				errs <- err
+				errs <- readerError{i, err}
 			}(i)
 		}
 		// catch all the errors
+		inbufs := make(map[int][]byte, len(dr.inbufs))
 		for range dr.rs {
-			err := <-errs
-			if err != nil {
-				// return on the first failure
-				dr.err = err
-				return 0, err
+			re := <-errs
+			if re.err == nil {
+				// add inbuf for decoding only if no error
+				inbufs[re.i] = dr.inbufs[re.i]
+			} else if re.err == io.EOF {
+				// return on the first EOF
+				dr.err = re.err
+				return 0, re.err
 			}
 		}
 		// we have all the input buffers, fill the decoded output buffer
-		dr.outbuf, err = dr.es.Decode(dr.outbuf, dr.inbufs)
+		dr.outbuf, err = dr.es.Decode(dr.outbuf, inbufs)
 		if err != nil {
 			return 0, err
 		}

pkg/eestream/pad_test.go

@@ -25,7 +25,6 @@ func TestPad(t *testing.T) {
 		{"abcdef", 9, 12},
 		{"abcdef", 10, 4},
 		{"abcdef", 11, 5},
-		{"abcdef", 11, 5},
 		{"abcde", 7, 9},
 		{"abcdefg", 7, 7},
 		{"abcdef", 512, 506},

pkg/eestream/rs_test.go

@@ -5,11 +5,16 @@ package eestream
 
 import (
 	"bytes"
+	"errors"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/vivint/infectious"
+
+	"storj.io/storj/internal/pkg/readcloser"
 )
 
 func TestRS(t *testing.T) {
@@ -32,3 +37,83 @@ func TestRS(t *testing.T) {
 		t.Fatalf("rs encode/decode failed")
 	}
 }
+
+func TestRSErrors(t *testing.T) {
+	for i, tt := range []struct {
+		dataSize    int
+		blockSize   int
+		required    int
+		total       int
+		problematic int
+		fail        bool
+	}{
+		{4 * 1024, 1024, 1, 1, 0, false},
+		{4 * 1024, 1024, 1, 1, 1, true},
+		{4 * 1024, 1024, 1, 2, 0, false},
+		{4 * 1024, 1024, 1, 2, 1, false},
+		{4 * 1024, 1024, 1, 2, 2, true},
+		{4 * 1024, 1024, 2, 4, 0, false},
+		{4 * 1024, 1024, 2, 4, 1, false},
+		{4 * 1024, 1024, 2, 4, 2, false},
+		{4 * 1024, 1024, 2, 4, 3, true},
+		{4 * 1024, 1024, 2, 4, 4, true},
+		{6 * 1024, 1024, 3, 7, 0, false},
+		{6 * 1024, 1024, 3, 7, 1, false},
+		{6 * 1024, 1024, 3, 7, 2, false},
+		{6 * 1024, 1024, 3, 7, 3, false},
+		{6 * 1024, 1024, 3, 7, 4, false},
+		{6 * 1024, 1024, 3, 7, 5, true},
+		{6 * 1024, 1024, 3, 7, 6, true},
+		{6 * 1024, 1024, 3, 7, 7, true},
+	} {
+		errTag := fmt.Sprintf("Test case #%d", i)
+		data := randData(tt.dataSize)
+		fc, err := infectious.NewFEC(tt.required, tt.total)
+		if !assert.NoError(t, err, errTag) {
+			continue
+		}
+		rs := NewRSScheme(fc, tt.blockSize)
+		readers := EncodeReader(bytes.NewReader(data), rs)
+		// read all readers in []byte buffers to avoid deadlock if later
+		// we don't read in parallel from all of them
+		pieces, err := readAll(readers)
+		if !assert.NoError(t, err, errTag) {
+			continue
+		}
+		readerMap := make(map[int]io.ReadCloser, len(readers))
+		// some readers will return error on read
+		for i := 0; i < tt.problematic; i++ {
+			readerMap[i] = readcloser.FatalReadCloser(
+				errors.New("I am an error piece"))
+		}
+		// the rest will operate normally
+		for i := tt.problematic; i < tt.total; i++ {
+			readerMap[i] = ioutil.NopCloser(bytes.NewReader(pieces[i]))
+		}
+		data2, err := ioutil.ReadAll(DecodeReaders(readerMap, rs))
+		if tt.fail {
+			assert.Error(t, err, errTag)
+		} else if assert.NoError(t, err, errTag) {
+			assert.Equal(t, data, data2, errTag)
+		}
+	}
+}
+
+func readAll(readers []io.Reader) ([][]byte, error) {
+	pieces := make([][]byte, len(readers))
+	errs := make(chan error, len(readers))
+	var err error
+	for i := range readers {
+		go func(i int) {
+			pieces[i], err = ioutil.ReadAll(readers[i])
+			errs <- err
+		}(i)
+	}
+	for range readers {
+		err := <-errs
+		if err != nil {
+			return nil, err
+		}
+	}
+	return pieces, nil
+}