Merge pull request #905 from niaow/syncrace

Fix race condition when resetting translation

boltdb/translate_test.go

@@ -20,6 +20,7 @@ import (
 	"io/ioutil"
 	"os"
 	"reflect"
+	"strconv"
 	"testing"
 	"time"
 
@@ -216,6 +217,30 @@ func TestTranslateStore_TranslateIDs(t *testing.T) {
 	}
 }
 
+func TestTranslateStore_MaxID(t *testing.T) {
+	s := MustOpenNewTranslateStore()
+	defer MustCloseTranslateStore(s)
+
+	// Generate a bunch of keys.
+	var lastk uint64
+	for i := 0; i < 1026; i++ {
+		k, err := s.TranslateKey(strconv.Itoa(i), true)
+		if err != nil {
+			t.Fatalf("translating %d: %v", i, err)
+		}
+		lastk = k
+	}
+
+	// Verify the max ID.
+	max, err := s.MaxID()
+	if err != nil {
+		t.Fatalf("checking max ID: %v", err)
+	}
+	if max != lastk {
+		t.Fatalf("last key is %d but max is %d", lastk, max)
+	}
+}
+
 func TestTranslateStore_EntryReader(t *testing.T) {
 	t.Run("OK", func(t *testing.T) {
 		s := MustOpenNewTranslateStore()

holder.go

@@ -1275,6 +1275,8 @@ type holderSyncer struct {
 	// Translation sync handling.
 	readers []TranslateEntryReader
 
+	syncers errgroup.Group
+
 	// Stats
 	Stats stats.StatsClient
 
@@ -1569,6 +1571,7 @@ func (s *holderSyncer) stopTranslationSync() error {
 			return rd.Close()
 		})
 	}
+	g.Go(s.syncers.Wait)
 	return g.Wait()
 }
 
@@ -1658,7 +1661,11 @@ func (s *holderSyncer) initializeIndexTranslateReplication() error {
 		}
 		s.readers = append(s.readers, rd)
 
-		go func() { defer rd.Close(); s.readIndexTranslateReader(rd) }()
+		s.syncers.Go(func() error {
+			defer rd.Close()
+			s.readIndexTranslateReader(rd)
+			return nil
+		})
 	}
 
 	return nil
@@ -1697,8 +1704,11 @@ func (s *holderSyncer) initializeFieldTranslateReplication() error {
 	}
 	s.readers = append(s.readers, rd)
 
-	go func() { defer rd.Close(); s.readFieldTranslateReader(rd) }()
-
+	s.syncers.Go(func() error {
+		defer rd.Close()
+		s.readFieldTranslateReader(rd)
+		return nil
+	})
 	return nil
 }
 

server.go

@@ -534,10 +534,12 @@ func (s *Server) Close() error {
 	s.wg.Wait()
 
 	var errh error
+	var errhs error
 	var errc error
 	if s.cluster != nil {
 		errc = s.cluster.close()
 	}
+	errhs = s.syncer.stopTranslationSync()
 	if s.holder != nil {
 		errh = s.holder.Close()
 	}
@@ -553,6 +555,9 @@ func (s *Server) Close() error {
 	if errh != nil {
 		return errors.Wrap(errh, "closing holder")
 	}
+	if errhs != nil {
+		return errors.Wrap(errhs, "terminating holder translation sync")
+	}
 	if errc != nil {
 		return errors.Wrap(errc, "closing cluster")
 	}