integration: add TestV3CompactInflightHashKV in v3_grpc_inflight_test.go

integration/v3_grpc_inflight_test.go

@@ -15,6 +15,7 @@
 package integration
 
 import (
+	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -54,6 +55,101 @@ func TestV3MaintenanceDefragmentInflightRange(t *testing.T) {
 	<-donec
 }
 
+// TestV3CompactInflightHashKV ensures that HashKV returns
+// correct hash while compacting.
+func TestV3CompactInflightHashKV(t *testing.T) {
+	defer testutil.AfterTest(t)
+	clus := NewClusterV3(t, &ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
+	cli1 := clus.Client(0)
+	mvcs := []pb.MaintenanceClient{
+		toGRPC(cli1).Maintenance,
+		toGRPC(clus.Client(1)).Maintenance,
+		toGRPC(clus.Client(2)).Maintenance,
+	}
+	cctx, cancel := context.WithCancel(context.Background())
+	var rev int64
+	for i := 0; i < 100; i++ {
+		resp, err := cli1.Put(cctx, "foo", "bar"+strconv.Itoa(i))
+		if err != nil && cctx.Err() == nil {
+			t.Fatal(err)
+		}
+		rev = resp.Header.Revision
+	}
+
+	// ensure appliedIdx is sync with committedIdx for each node.
+	_, err := clus.Client(0).Get(cctx, "foo")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = clus.Client(1).Get(cctx, "foo")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = clus.Client(2).Get(cctx, "foo")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	hashesc := make(chan *pb.HashKVResponse, 3)
+	donec := make(chan struct{})
+	var wg sync.WaitGroup
+	for i, _ := range mvcs {
+		wg.Add(1)
+		go func(cliIdx int) {
+			defer wg.Done()
+			for {
+				resp, err := mvcs[cliIdx].HashKV(cctx, &pb.HashKVRequest{rev})
+				if err != nil && cctx.Err() == nil {
+					t.Fatal(err)
+				}
+
+				select {
+				case hashesc <- resp:
+				case <-donec:
+					return
+				}
+			}
+		}(i)
+	}
+
+	compactRev := rev - 1
+	_, err = cli1.Compact(cctx, compactRev)
+	if err != nil {
+		t.Fatalf("compact error %v", err)
+	}
+
+	stopc := make(chan struct{})
+	go func() {
+		defer close(donec)
+		revHash := make(map[int64]uint32)
+		for i := 0; i < 12; i++ {
+			select {
+			case resp := <-hashesc:
+				if revHash[resp.CompactRevision] == 0 {
+					revHash[resp.CompactRevision] = resp.Hash
+					break
+				}
+				if resp.Hash != revHash[resp.CompactRevision] {
+					t.Fatalf("Hashes differ (current %v) != (saved %v)", resp.Hash, revHash[compactRev])
+				}
+			case <-stopc:
+				return
+			}
+		}
+	}()
+
+	select {
+	case <-donec:
+		cancel()
+	case <-time.After(10 * time.Second):
+		close(stopc)
+		cancel()
+		<-donec
+	}
+	wg.Wait()
+}
+
 // TestV3KVInflightRangeRequests ensures that inflight requests
 // (sent before server shutdown) are gracefully handled by server-side.
 // They are either finished or canceled, but never crash the backend.