Potential reproduction of #20573

Evolve the code from Marek in https://github.com/serathius/etcd/blob/52673cee64e7da995f4bc700977a88a21a304066/tests/robustness/client/watch.go based on the work we did together on Friday,
and #20234 from Marek before (#20221 (comment)).

Please see discussion starting from #20349 (comment)

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

Author: henrybear327

tests/robustness/client/watch.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
@@ -43,14 +44,28 @@ func CollectClusterWatchEvents(ctx context.Context, lg *zap.Logger, endpoints []
 			return err
 		})
 	}
-
+	finish := make(chan struct{})
 	g.Go(func() error {
 		maxRevision := <-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, lg, &g, c, period, finish)
+		})
+	}
+
 	return g.Wait()
 }
 
@@ -122,3 +137,42 @@ resetWatch:
 		}
 	}
 }
+
+func openWatchPeriodically(ctx context.Context, lg *zap.Logger, 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 := int64(0) // no reproduction
+			// lg.Info("revisions", zap.Int64("rev", rev), zap.Int64("resp rev", resp.Header.Revision))
+
+			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/failpoint/failpoint.go

@@ -36,23 +36,27 @@ const (
 )
 
 var allFailpoints = []Failpoint{
-	KillFailpoint, BeforeCommitPanic, AfterCommitPanic, RaftBeforeSavePanic, RaftAfterSavePanic,
-	DefragBeforeCopyPanic, DefragBeforeRenamePanic, BackendBeforePreCommitHookPanic, BackendAfterPreCommitHookPanic,
-	BackendBeforeStartDBTxnPanic, BackendAfterStartDBTxnPanic, BackendBeforeWritebackBufPanic,
-	BackendAfterWritebackBufPanic, CompactBeforeCommitScheduledCompactPanic, CompactAfterCommitScheduledCompactPanic,
-	CompactBeforeSetFinishedCompactPanic, CompactAfterSetFinishedCompactPanic, CompactBeforeCommitBatchPanic,
-	CompactAfterCommitBatchPanic, RaftBeforeLeaderSendPanic, BlackholePeerNetwork, DelayPeerNetwork,
-	RaftBeforeFollowerSendPanic, RaftBeforeApplySnapPanic, RaftAfterApplySnapPanic, RaftAfterWALReleasePanic,
-	RaftBeforeSaveSnapPanic, RaftAfterSaveSnapPanic, BlackholeUntilSnapshot,
-	BeforeApplyOneConfChangeSleep,
-	MemberReplace,
-	MemberDowngrade,
-	MemberDowngradeUpgrade,
-	DropPeerNetwork,
-	RaftBeforeSaveSleep,
-	RaftAfterSaveSleep,
-	ApplyBeforeOpenSnapshot,
-	SleepBeforeSendWatchResponse,
+	// KillFailpoint, BeforeCommitPanic, AfterCommitPanic, RaftBeforeSavePanic, RaftAfterSavePanic,
+	// DefragBeforeCopyPanic, DefragBeforeRenamePanic, BackendBeforePreCommitHookPanic, BackendAfterPreCommitHookPanic,
+	// BackendBeforeStartDBTxnPanic, BackendAfterStartDBTxnPanic, BackendBeforeWritebackBufPanic,
+	// BackendAfterWritebackBufPanic, CompactBeforeCommitScheduledCompactPanic, CompactAfterCommitScheduledCompactPanic,
+	// CompactBeforeSetFinishedCompactPanic, CompactAfterSetFinishedCompactPanic, CompactBeforeCommitBatchPanic,
+	// CompactAfterCommitBatchPanic, RaftBeforeLeaderSendPanic, BlackholePeerNetwork, DelayPeerNetwork,
+	// RaftBeforeFollowerSendPanic, RaftBeforeApplySnapPanic, RaftAfterApplySnapPanic, RaftAfterWALReleasePanic,
+	// RaftBeforeSaveSnapPanic, RaftAfterSaveSnapPanic, BlackholeUntilSnapshot,
+	// BeforeApplyOneConfChangeSleep,
+	// MemberReplace,
+	// MemberDowngrade,
+	// MemberDowngradeUpgrade,
+	// DropPeerNetwork,
+	// RaftBeforeSaveSleep,
+	// RaftAfterSaveSleep,
+	// ApplyBeforeOpenSnapshot,
+	// SleepBeforeSendWatchResponse,
+	RaftAfterSaveSnapPanic,
+	RaftBeforeApplySnapPanic,
+	RaftAfterApplySnapPanic,
+	RaftAfterWALReleasePanic,
 }
 
 func PickRandom(clus *e2e.EtcdProcessCluster, profile traffic.Profile) (Failpoint, error) {

tests/robustness/scenarios/scenarios.go

@@ -24,7 +24,6 @@ import (
 
 	"go.etcd.io/etcd/api/v3/version"
 	"go.etcd.io/etcd/client/pkg/v3/fileutil"
-	"go.etcd.io/etcd/server/v3/etcdserver"
 	"go.etcd.io/etcd/tests/v3/framework/e2e"
 	"go.etcd.io/etcd/tests/v3/robustness/client"
 	"go.etcd.io/etcd/tests/v3/robustness/failpoint"
@@ -40,16 +39,16 @@ type TrafficProfile struct {
 }
 
 var trafficProfiles = []TrafficProfile{
-	{
-		Name:    "EtcdHighTraffic",
-		Traffic: traffic.EtcdPut,
-		Profile: traffic.HighTrafficProfile,
-	},
-	{
-		Name:    "EtcdTrafficDeleteLeases",
-		Traffic: traffic.EtcdPutDeleteLease,
-		Profile: traffic.LowTraffic,
-	},
+	// {
+	// 	Name:    "EtcdHighTraffic",
+	// 	Traffic: traffic.EtcdPut,
+	// 	Profile: traffic.HighTrafficProfile,
+	// },
+	// {
+	// 	Name:    "EtcdTrafficDeleteLeases",
+	// 	Traffic: traffic.EtcdPutDeleteLease,
+	// 	Profile: traffic.LowTraffic,
+	// },
 	{
 		Name:    "KubernetesHighTraffic",
 		Traffic: traffic.Kubernetes,
@@ -94,11 +93,11 @@ func Exploratory(_ *testing.T) []TestScenario {
 	mixedVersionOption := options.WithClusterOptionGroups(random.PickRandom[options.ClusterOptions](mixedVersionOptionChoices))
 
 	baseOptions := []e2e.EPClusterOption{
-		options.WithSnapshotCount(50, 100, 1000),
+		options.WithSnapshotCount(50),
 		options.WithSubsetOptions(randomizableOptions...),
 		e2e.WithGoFailEnabled(true),
 		// Set a low minimal compaction batch limit to allow for triggering multi batch compaction failpoints.
-		options.WithCompactionBatchLimit(10, 100, 1000),
+		options.WithCompactionBatchLimit(10),
 		e2e.WithWatchProcessNotifyInterval(100 * time.Millisecond),
 	}
 
@@ -107,20 +106,20 @@ func Exploratory(_ *testing.T) []TestScenario {
 	}
 
 	if e2e.CouldSetSnapshotCatchupEntries(e2e.BinPath.Etcd) {
-		baseOptions = append(baseOptions, options.WithSnapshotCatchUpEntries(100, etcdserver.DefaultSnapshotCatchUpEntries))
+		baseOptions = append(baseOptions, options.WithSnapshotCatchUpEntries(100))
 	}
 	scenarios := []TestScenario{}
-	for _, tp := range trafficProfiles {
-		name := filepath.Join(tp.Name, "ClusterOfSize1")
-		clusterOfSize1Options := baseOptions
-		clusterOfSize1Options = append(clusterOfSize1Options, e2e.WithClusterSize(1))
-		scenarios = append(scenarios, TestScenario{
-			Name:    name,
-			Traffic: tp.Traffic,
-			Profile: tp.Profile,
-			Cluster: *e2e.NewConfig(clusterOfSize1Options...),
-		})
-	}
+	// for _, tp := range trafficProfiles {
+	// 	name := filepath.Join(tp.Name, "ClusterOfSize1")
+	// 	clusterOfSize1Options := baseOptions
+	// 	clusterOfSize1Options = append(clusterOfSize1Options, e2e.WithClusterSize(1))
+	// 	scenarios = append(scenarios, TestScenario{
+	// 		Name:    name,
+	// 		Traffic: tp.Traffic,
+	// 		Profile: tp.Profile,
+	// 		Cluster: *e2e.NewConfig(clusterOfSize1Options...),
+	// 	})
+	// }
 
 	for _, tp := range trafficProfiles {
 		name := filepath.Join(tp.Name, "ClusterOfSize3")