Catch errors creating new iteratos in rowReaders

repository_pool.go

@@ -276,7 +276,18 @@ func (i *rowRepoIter) rowReader(num int) {
 	defer i.wg.Done()
 
 	for repo := range i.repos {
-		iter, _ := i.iter.NewIterator(repo)
+		iter, err := i.iter.NewIterator(repo)
+		if err != nil {
+			// guard from possible previous error
+			select {
+			case <-i.done:
+				return
+			default:
+				i.setError(err)
+				close(i.done)
+				continue
+			}
+		}
 
 	loop:
 		for {

repository_pool_test.go

@@ -2,13 +2,15 @@ package gitbase
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strconv"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	"gopkg.in/src-d/go-git-fixtures.v3"
@@ -263,3 +265,49 @@ func TestRepositoryPoolAddDir(t *testing.T) {
 
 	require.ElementsMatch(arrayExpected, arrayID)
 }
+
+var errIter = fmt.Errorf("Error iter")
+
+type testErrorIter struct{}
+
+func (d *testErrorIter) NewIterator(
+	repo *Repository,
+) (RowRepoIter, error) {
+	return nil, errIter
+	// return &testErrorIter{}, nil
+}
+
+func (d *testErrorIter) Next() (sql.Row, error) {
+	return nil, io.EOF
+}
+
+func (d *testErrorIter) Close() error {
+	return nil
+}
+
+func TestRepositoryErrorIter(t *testing.T) {
+	require := require.New(t)
+
+	path := fixtures.Basic().ByTag("worktree").One().Worktree().Root()
+	pool := NewRepositoryPool()
+	pool.Add("one", path)
+
+	timeout, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+
+	ctx := sql.NewContext(timeout, sql.WithSession(NewSession(&pool)))
+	eIter := &testErrorIter{}
+
+	repoIter, err := NewRowRepoIter(ctx, eIter)
+	require.NoError(err)
+
+	go func() {
+		repoIter.Next()
+	}()
+
+	select {
+	case <-repoIter.done:
+		require.Equal(errIter, repoIter.err)
+	}
+
+	cancel()
+}