Reproduce issue #17529

Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>

tests/robustness/failpoint/failpoint.go

@@ -50,6 +50,7 @@ var (
 		RaftBeforeSaveSleep,
 		RaftAfterSaveSleep,
 		ApplyBeforeOpenSnapshot,
+		SleepBeforeSendWatchResponse,
 	}
 )
 

tests/robustness/failpoint/gofail.go

@@ -59,6 +59,7 @@ var (
 	BeforeApplyOneConfChangeSleep            Failpoint = killAndGofailSleep{"beforeApplyOneConfChange", time.Second}
 	RaftBeforeSaveSleep                      Failpoint = gofailSleepAndDeactivate{"raftBeforeSave", time.Second}
 	RaftAfterSaveSleep                       Failpoint = gofailSleepAndDeactivate{"raftAfterSave", time.Second}
+	SleepBeforeSendWatchResponse             Failpoint = gofailSleepAndDeactivate{"beforeSendWatchResponse", time.Second}
 )
 
 type goPanicFailpoint struct {
@@ -238,9 +239,6 @@ func (f gofailSleepAndDeactivate) Name() string {
 }
 
 func (f gofailSleepAndDeactivate) Available(config e2e.EtcdProcessClusterConfig, member e2e.EtcdProcess) bool {
-	if config.ClusterSize == 1 {
-		return false
-	}
 	memberFailpoints := member.Failpoints()
 	if memberFailpoints == nil {
 		return false

tests/robustness/makefile.mk

@@ -39,6 +39,11 @@ test-robustness-issue15271: /tmp/etcd-v3.5.7-failpoints/bin
 	GO_TEST_FLAGS='-v --run=TestRobustnessRegression/Issue15271 --count 100 --failfast --bin-dir=/tmp/etcd-v3.5.7-failpoints/bin' make test-robustness && \
 	 echo "Failed to reproduce" || echo "Successful reproduction"
 
+.PHONY: test-robustness-issue17529
+test-robustness-issue17529: /tmp/etcd-v3.5.12-beforeSendWatchResponse/bin
+	GO_TEST_FLAGS='-v --run=TestRobustnessRegression/Issue17529 --count 100 --failfast --bin-dir=/tmp/etcd-v3.5.12-beforeSendWatchResponse/bin' make test-robustness && \
+	 echo "Failed to reproduce" || echo "Successful reproduction"
+
 # Failpoints
 
 GOPATH = $(shell go env GOPATH)
@@ -98,6 +103,21 @@ $(GOPATH)/bin/gofail: tools/mod/go.mod tools/mod/go.sum
 	  (cd etcdutl; go get go.etcd.io/gofail@${GOFAIL_VERSION}); \
 	  FAILPOINTS=true ./build;
 
+/tmp/etcd-v3.5.12-beforeSendWatchResponse/bin: $(GOPATH)/bin/gofail
+	rm -rf /tmp/etcd-v3.5.12-beforeSendWatchResponse/
+	mkdir -p /tmp/etcd-v3.5.12-beforeSendWatchResponse/
+	git clone --depth 1 --branch v3.5.12 https://github.com/etcd-io/etcd.git /tmp/etcd-v3.5.12-beforeSendWatchResponse/
+	cp -r ./tests/robustness/patches/beforeSendWatchResponse /tmp/etcd-v3.5.12-beforeSendWatchResponse/
+	cd /tmp/etcd-v3.5.12-beforeSendWatchResponse/; \
+	  patch -l server/etcdserver/api/v3rpc/watch.go ./beforeSendWatchResponse/watch.patch; \
+	  patch -l build.sh ./beforeSendWatchResponse/build.patch; \
+	  go get go.etcd.io/gofail@${GOFAIL_VERSION}; \
+	  (cd server; go get go.etcd.io/gofail@${GOFAIL_VERSION}); \
+	  (cd etcdctl; go get go.etcd.io/gofail@${GOFAIL_VERSION}); \
+	  (cd etcdutl; go get go.etcd.io/gofail@${GOFAIL_VERSION}); \
+	  (cd tools/mod; go get go.etcd.io/gofail@${GOFAIL_VERSION}); \
+	  FAILPOINTS=true ./build;
+
 /tmp/etcd-release-3.5-failpoints/bin: $(GOPATH)/bin/gofail
 	rm -rf /tmp/etcd-release-3.5-failpoints/
 	mkdir -p /tmp/etcd-release-3.5-failpoints/

tests/robustness/patches/beforeSendWatchResponse/build.patch

@@ -0,0 +1,9 @@
+@@ -25,7 +26,7 @@ GOFAIL_VERSION=$(cd tools/mod && go list -m -f {{.Version}} go.etcd.io/gofail)
+ toggle_failpoints() {
+   mode="$1"
+   if command -v gofail >/dev/null 2>&1; then
+-    run gofail "$mode" server/etcdserver/ server/mvcc/ server/wal/ server/mvcc/backend/
++    run gofail "$mode" server/etcdserver/ server/mvcc/ server/wal/ server/mvcc/backend/ server/etcdserver/api/v3rpc/
+     if [[ "$mode" == "enable" ]]; then
+       go get go.etcd.io/gofail@${GOFAIL_VERSION}
+       cd ./server && go get go.etcd.io/gofail@${GOFAIL_VERSION}

tests/robustness/patches/beforeSendWatchResponse/watch.patch

@@ -0,0 +1,12 @@
+diff --git a/server/etcdserver/api/v3rpc/watch.go b/server/etcdserver/api/v3rpc/watch.go
+index cd834aa..e6aaf2b 100644
+--- a/server/etcdserver/api/v3rpc/watch.go
++++ b/server/etcdserver/api/v3rpc/watch.go
+@@ -460,6 +460,7 @@ func (sws *serverWatchStream) sendLoop() {
+                        sws.mu.RUnlock()
+
+                        var serr error
++                       // gofail: var beforeSendWatchResponse struct{}
+                        if !fragmented && !ok {
+                                serr = sws.gRPCStream.Send(wr)
+                        } else {

tests/robustness/scenarios.go

@@ -191,6 +191,17 @@ func regressionScenarios(t *testing.T) []testScenario {
 			e2e.WithClusterSize(1),
 		),
 	})
+	scenarios = append(scenarios, testScenario{
+		name:      "Issue17529",
+		profile:   traffic.HighTrafficProfile,
+		traffic:   traffic.Kubernetes,
+		failpoint: failpoint.SleepBeforeSendWatchResponse,
+		cluster: *e2e.NewConfig(
+			e2e.WithClusterSize(1),
+			e2e.WithGoFailEnabled(true),
+			options.WithSnapshotCount(100),
+		),
+	})
 	if v.Compare(version.V3_5) >= 0 {
 		opts := []e2e.EPClusterOption{
 			e2e.WithSnapshotCount(100),

tests/robustness/traffic/kubernetes.go

@@ -37,9 +37,9 @@ var (
 		resource:        "pods",
 		namespace:       "default",
 		writeChoices: []choiceWeight[KubernetesRequestType]{
-			{choice: KubernetesUpdate, weight: 85},
-			{choice: KubernetesDelete, weight: 5},
-			{choice: KubernetesCreate, weight: 5},
+			{choice: KubernetesUpdate, weight: 75},
+			{choice: KubernetesDelete, weight: 10},
+			{choice: KubernetesCreate, weight: 10},
 			{choice: KubernetesCompact, weight: 5},
 		},
 	}

tests/robustness/traffic/traffic.go

@@ -34,7 +34,7 @@ import (
 var (
 	DefaultLeaseTTL   int64 = 7200
 	RequestTimeout          = 200 * time.Millisecond
-	WatchTimeout            = 400 * time.Millisecond
+	WatchTimeout            = time.Second
 	MultiOpTxnOpCount       = 4
 
 	LowTraffic = Profile{