fix batch persisted callback for upsidedown (#1209)

previous implementation would only execute the callback
if the control flow reached the end of the function

unfortunately there are other explicit returns earlier
in the function

users setting a callback, would reasonably expect it to
always be called (passing either nil or an err) and
due to this bug, certain error conditions resulted
in the callback not firing

this change checks for the callback, and if present
defers execution of the callback, ensuring it will
always be executed

index/upsidedown/upsidedown.go

@@ -801,6 +801,10 @@ func (udc *UpsideDownCouch) termFieldVectorsFromTermVectors(in []*TermVector) []
 }
 
 func (udc *UpsideDownCouch) Batch(batch *index.Batch) (err error) {
+	persistedCallback := batch.PersistedCallback()
+	if persistedCallback != nil {
+		defer persistedCallback(err)
+	}
 	analysisStart := time.Now()
 
 	resultChan := make(chan *index.AnalysisResult, len(batch.IndexOps))
@@ -965,10 +969,6 @@ func (udc *UpsideDownCouch) Batch(batch *index.Batch) (err error) {
 		atomic.AddUint64(&udc.stats.errors, 1)
 	}
 
-	persistedCallback := batch.PersistedCallback()
-	if persistedCallback != nil {
-		persistedCallback(err)
-	}
 	return
 }
 

index/upsidedown/upsidedown_test.go

@@ -19,6 +19,7 @@ import (
 	"reflect"
 	"regexp"
 	"strconv"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -1447,3 +1448,48 @@ func TestLargeField(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestIndexBatchPersistedCallbackWithErrorUpsideDown(t *testing.T) {
+	defer func() {
+		err := DestroyTest()
+		if err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	analysisQueue := index.NewAnalysisQueue(1)
+	idx, err := NewUpsideDownCouch(boltdb.Name, boltTestConfig, analysisQueue)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = idx.Open()
+	if err != nil {
+		t.Errorf("error opening index: %v", err)
+	}
+	defer func() {
+		err := idx.Close()
+		if err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	var callbackExecuted bool
+	batch := index.NewBatch()
+	batch.SetPersistedCallback(func(e error) {
+		callbackExecuted = true
+	})
+	// By using a really large ID, we ensure that the batch will fail,
+	// because the key generated by upside down will be too large for BoltDB
+	reallyBigId := strings.Repeat("x", 32768+1)
+	doc := document.NewDocument(reallyBigId)
+	doc.AddField(document.NewTextField("name", []uint64{}, []byte("test3")))
+	batch.Update(doc)
+
+	_ = idx.Batch(batch)
+	// don't fail on this error, that isn't what we're testing
+
+	if !callbackExecuted {
+		t.Fatal("expected callback to fire, it did not")
+	}
+
+}