Add new target for issue20221 for the potential reproduction of #20573

Evolve the code from Marek (#20221 (comment)).

Please see discussion starting from #20349 (comment)

Signed-off-by: Chun-Hung Tseng <[email protected]>

Author: henrybear327

tests/robustness/Makefile

@@ -75,6 +75,11 @@ test-robustness-issue19179: /tmp/etcd-v3.5.17-failpoints/bin
 	GO_TEST_FLAGS='-v -run=TestRobustnessRegression/Issue19179 -count 200 -failfast --bin-dir=/tmp/etcd-v3.5.17-failpoints/bin' $(TOPLEVEL_MAKE) test-robustness && \
 	  echo "Failed to reproduce" || echo "Successful reproduction"
 
+.PHONY: test-robustness-issue20221
+test-robustness-issue20221: /tmp/etcd-v3.6.4-failpoints/bin
+	GO_TEST_FLAGS='-v -run=TestRobustnessRegression/issue20221 -count 200 -failfast --bin-dir=/tmp/etcd-v3.6.4-failpoints/bin' $(TOPLEVEL_MAKE) test-robustness && \
+	  echo "Failed to reproduce" || echo "Successful reproduction"
+
 # Etcd API usage by Kubernetes
 
 .PHONY: k8s-coverage
@@ -120,14 +125,6 @@ $(GOPATH)/bin/gofail: $(REPOSITORY_ROOT)/tools/mod/go.mod $(REPOSITORY_ROOT)/too
 	  $(MAKE) gofail-enable; \
 	  $(MAKE) build;
 
-/tmp/etcd-v3.6.0-failpoints/bin: $(GOPATH)/bin/gofail
-	rm -rf /tmp/etcd-v3.6.0-failpoints/
-	mkdir -p /tmp/etcd-v3.6.0-failpoints/
-	cd /tmp/etcd-v3.6.0-failpoints/; \
-	  git clone --depth 1 --branch main https://github.com/etcd-io/etcd.git .; \
-	  $(MAKE) gofail-enable; \
-	  $(MAKE) build;
-
 /tmp/etcd-release-3.6-failpoints/bin: $(GOPATH)/bin/gofail
 	rm -rf /tmp/etcd-release-3.6-failpoints/
 	mkdir -p /tmp/etcd-release-3.6-failpoints/
@@ -136,6 +133,14 @@ $(GOPATH)/bin/gofail: $(REPOSITORY_ROOT)/tools/mod/go.mod $(REPOSITORY_ROOT)/too
 	  $(MAKE) gofail-enable; \
 	  $(MAKE) build;
 
+/tmp/etcd-v3.6.4-failpoints/bin: $(GOPATH)/bin/gofail
+	rm -rf /tmp/etcd-v3.6.4-failpoints/
+	mkdir -p /tmp/etcd-v3.6.4-failpoints/
+	cd /tmp/etcd-v3.6.4-failpoints/; \
+	  git clone --depth 1 --branch v3.6.4 https://github.com/etcd-io/etcd.git .; \
+	  $(MAKE) gofail-enable; \
+	  $(MAKE) build;
+
 /tmp/etcd-v3.5.2-failpoints/bin:
 /tmp/etcd-v3.5.4-failpoints/bin:
 /tmp/etcd-v3.5.5-failpoints/bin:

tests/robustness/client/watch.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
@@ -51,14 +52,28 @@ func CollectClusterWatchEvents(ctx context.Context, param CollectClusterWatchEve
 			return err
 		})
 	}
-
+	finish := make(chan struct{})
 	g.Go(func() error {
 		maxRevision := <-param.MaxRevisionChan
 		for _, memberChan := range memberMaxRevisionChans {
 			memberChan <- maxRevision
 		}
+		close(finish)
 		return nil
 	})
+
+	for _, endpoint := range endpoints {
+		g.Go(func() error {
+			c, err := clientSet.NewClient([]string{endpoint})
+			if err != nil {
+				return err
+			}
+			defer c.Close()
+			period := 10 * time.Millisecond
+			return openWatchPeriodically(ctx, &g, c, period, finish)
+		})
+	}
+
 	return g.Wait()
 }
 
@@ -130,3 +145,42 @@ resetWatch:
 		}
 	}
 }
+
+func openWatchPeriodically(ctx context.Context, g *errgroup.Group, c *RecordingClient, period time.Duration, finish <-chan struct{}) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-finish:
+			return nil
+		case <-time.After(period):
+		}
+		g.Go(func() error {
+			// targeting commit 866bc0717
+			resp, err := c.Get(ctx, "/key")
+			if err != nil {
+				return err
+			}
+			// rev := resp.Header.Revision + (rand.Int64N(20) - 10) // reproduce OK (<2 min on my machine)
+			rev := resp.Header.Revision - 10 // reproduce OK (<1 min on my machine)
+			// rev := resp.Header.Revision // reproduce TBD
+			// rev := int64(0) // no reproduction
+
+			watchCtx, cancel := context.WithCancel(ctx)
+			defer cancel()
+			w := c.Watch(watchCtx, "", rev, true, true, true)
+			for {
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case <-finish:
+					return nil
+				case _, ok := <-w:
+					if !ok {
+						return nil
+					}
+				}
+			}
+		})
+	}
+}

tests/robustness/scenarios/scenarios.go

@@ -261,6 +261,7 @@ func Regression(t *testing.T) []TestScenario {
 			e2e.WithGoFailEnabled(true),
 		),
 	})
+
 	scenarios = append(scenarios, TestScenario{
 		Name:      "Issue18089",
 		Profile:   traffic.LowTraffic.WithCompactionPeriod(100 * time.Millisecond), // Use frequent compaction for high reproduce rate
@@ -271,6 +272,7 @@ func Regression(t *testing.T) []TestScenario {
 			e2e.WithGoFailEnabled(true),
 		),
 	})
+
 	if v.Compare(version.V3_5) >= 0 {
 		opts := []e2e.EPClusterOption{
 			e2e.WithSnapshotCount(100),
@@ -280,6 +282,7 @@ func Regression(t *testing.T) []TestScenario {
 		if e2e.CouldSetSnapshotCatchupEntries(e2e.BinPath.Etcd) {
 			opts = append(opts, e2e.WithSnapshotCatchUpEntries(100))
 		}
+
 		scenarios = append(scenarios, TestScenario{
 			Name:      "Issue15271",
 			Failpoint: failpoint.BlackholeUntilSnapshot,
@@ -288,5 +291,25 @@ func Regression(t *testing.T) []TestScenario {
 			Cluster:   *e2e.NewConfig(opts...),
 		})
 	}
+
+	scenarios = append(scenarios, TestScenario{
+		Name:      "issue20221",
+		Profile:   traffic.HighTrafficProfile.WithoutCompaction(),
+		Failpoint: failpoint.RaftAfterSaveSnapPanic,
+		// Failpoint: failpoint.RaftBeforeApplySnapPanic,
+		// Failpoint: failpoint.RaftAfterApplySnapPanic,
+		// Failpoint: failpoint.RaftAfterWALReleasePanic,
+		Traffic: traffic.Kubernetes,
+		Cluster: *e2e.NewConfig(
+			e2e.WithClusterSize(3),
+			e2e.WithCompactionBatchLimit(10),
+			e2e.WithSnapshotCount(50),
+			e2e.WithSnapshotCatchUpEntries(100),
+			e2e.WithGoFailEnabled(true),
+			e2e.WithPeerProxy(true),
+			e2e.WithIsPeerTLS(true),
+		),
+	})
+
 	return scenarios
 }