[dbnode] Fix duplicate ID insertions causing transient error when flushing index block

src/m3ninx/index/segment/builder/builder.go

@@ -138,11 +138,19 @@ func (b *builder) Reset(offset postings.ID) {
 }
 
 func (b *builder) Insert(d doc.Document) ([]byte, error) {
+	b.status.RLock()
+	defer b.status.RUnlock()
+
 	// Use a preallocated slice to make insert able to avoid alloc
 	// a slice to call insert batch with.
 	b.batchSizeOne.Docs[0] = d
-	err := b.InsertBatch(b.batchSizeOne)
+	err := b.insertBatchWithRLock(b.batchSizeOne)
 	if err != nil {
+		if errs := err.Errs(); len(errs) == 1 {
+			// Return concrete error instead of the batch partial error.
+			return nil, errs[0].Err
+		}
+		// Fallback to returning batch partial error if not what we expect.
 		return nil, err
 	}
 	last := b.docs[len(b.docs)-1]
@@ -157,6 +165,16 @@ func (b *builder) InsertBatch(batch index.Batch) error {
 		return errClosed
 	}
 
+	// NB(r): This switch is required or else *index.BatchPartialError
+	// is returned as a non-nil wrapped "error" even though it is not
+	// an error and underlying error is nil.
+	if err := b.insertBatchWithRLock(batch); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (b *builder) insertBatchWithRLock(batch index.Batch) *index.BatchPartialError {
 	// NB(r): This is all kept in a single method to make the
 	// insertion path fast.
 	var wg sync.WaitGroup

src/m3ninx/index/segment/builder/builder_test.go

@@ -27,6 +27,7 @@ import (
 	"unsafe"
 
 	"github.com/m3db/m3/src/m3ninx/doc"
+	"github.com/m3db/m3/src/m3ninx/index"
 	"github.com/m3db/m3/src/m3ninx/index/segment"
 
 	"github.com/stretchr/testify/require"
@@ -142,6 +143,23 @@ func TestBuilderTerms(t *testing.T) {
 	}
 }
 
+// Test that calling Insert(...) API returns correct concrete errors
+// instead of partial batch error type.
+func TestBuilderInsertDuplicateReturnsErrDuplicateID(t *testing.T) {
+	builder, err := NewBuilderFromDocuments(testOptions)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, builder.Close())
+	}()
+
+	_, err = builder.Insert(testDocuments[2])
+	require.NoError(t, err)
+
+	_, err = builder.Insert(testDocuments[2])
+	require.Error(t, err)
+	require.Equal(t, index.ErrDuplicateID, err)
+}
+
 func toSlice(t *testing.T, iter segment.FieldsPostingsListIterator) [][]byte {
 	elems := [][]byte{}
 	for iter.Next() {