reproduce hang in test and fix

needed to cover the case where import routines have errors and exit early

Author: jaffee

client.go

@@ -1078,25 +1078,39 @@ func (c *Client) ExperimentalReplayImport(r io.Reader, concurrency int) error {
 		})
 	}
 
+	// wait on importers asynchronously in case they return early with errors
+	waitErrCh := make(chan error)
+	go func() {
+		waitErrCh <- eg.Wait()
+		close(waitErrCh)
+	}()
+
 	// populate work channel
 	dec := newImportLogDecoder(r)
 	var err error
 	for {
 		log := importLog{}
 		err = dec.Decode(&log)
 		if err != nil {
+			err = errors.Wrap(err, "decoding")
 			break
 		}
-		work <- &log
+		select {
+		case work <- &log:
+			continue
+		case err = <-waitErrCh:
+			err = errors.Wrap(err, "waiting")
+		}
+		break
 	}
 
 	// close work channel (now workers can exit)
 	close(work)
-	waitErr := eg.Wait()
-	if err != io.EOF {
-		return errors.Wrap(err, "decoding")
+	<-waitErrCh // make sure workers are complete
+	if errors.Cause(err) == io.EOF {
+		return nil
 	}
-	return errors.Wrap(waitErr, "waiting")
+	return err
 }
 
 func defaultProtobufHeaders() map[string]string {

client_internal_it_test.go

@@ -8,8 +8,10 @@ import (
 	"io/ioutil"
 	"net/http"
 	"reflect"
+	"strings"
 	"testing"
 	"testing/iotest"
+	"time"
 )
 
 func TestNewClientFromAddresses(t *testing.T) {
@@ -115,3 +117,57 @@ func TestImportWithReplay(t *testing.T) {
 		}
 	}
 }
+
+func TestImportWithReplayErrors(t *testing.T) {
+	buf := &bytes.Buffer{}
+	client := getClient(ExperimentalOptClientLogImports(buf))
+
+	// the first iterator for creating the target
+	iterator := &ColumnGenerator{numRows: 10, numColumns: 1000}
+	target := map[uint64][]uint64{}
+	for {
+		rec, err := iterator.NextRecord()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			t.Fatal(err)
+		}
+		if col, ok := rec.(Column); ok {
+			target[col.RowID] = append(target[col.RowID], col.ColumnID)
+		}
+	}
+	// the second iterator for the actual import
+	iterator = &ColumnGenerator{numRows: 10, numColumns: 1000}
+	field := index.Field("importfield-batchsize")
+	err := client.EnsureField(field)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	statusChan := make(chan ImportStatusUpdate, 1000)
+	err = client.ImportField(field, iterator, OptImportStatusChannel(statusChan), OptImportThreadCount(2), OptImportBatchSize(100))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// delete index, but don't recreate
+	Reset()
+
+	result := make(chan error)
+	go func() {
+		result <- client.ExperimentalReplayImport(buf, 2)
+	}()
+
+	select {
+	case err := <-result:
+		if err == nil {
+			t.Fatalf("import shouldn't have worked")
+		}
+		if !strings.Contains(err.Error(), "waiting: ") {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	case <-time.After(time.Second * 5):
+		t.Fatal("import replay hanging when no schema created")
+	}
+}