Add e2e test to reproduce issue #19406

Signed-off-by: Miancheng Lin <[email protected]>

Author: miancheng7

server/storage/mvcc/kvstore_compaction.go

@@ -49,6 +49,7 @@ func (s *store) scheduleCompaction(compactMainRev, prevCompactRev int64) (KeyVal
 
 		tx := s.b.BatchTx()
 		tx.LockOutsideApply()
+		// gofail: var compactAfterAcquiredBatchTxLock struct{}
 		keys, values := tx.UnsafeRange(schema.Key, last, end, int64(batchNum))
 		for i := range keys {
 			rev = BytesToRev(keys[i])

tests/e2e/reproduce_19406_test.go

@@ -0,0 +1,97 @@
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"github.com/stretchr/testify/require"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/pkg/v3/stringutil"
+	"sync"
+	"testing"
+	"time"
+
+	"go.etcd.io/etcd/tests/v3/framework/e2e"
+)
+
+// TestReproduce19406 reproduces the issue: https://github.com/etcd-io/etcd/issues/19406
+func TestReproduce19406(t *testing.T) {
+	e2e.BeforeTest(t)
+
+	compactionBatchLimit := 1
+
+	ctx := context.TODO()
+	CompactionSleepInterval := 100 * time.Millisecond
+
+	clus, cerr := e2e.NewEtcdProcessCluster(ctx, t,
+		e2e.WithClusterSize(1),
+		e2e.WithGoFailEnabled(true),
+		e2e.WithCompactionBatchLimit(compactionBatchLimit),
+		e2e.WithCompactionSleepInterval(CompactionSleepInterval),
+	)
+
+	require.NoError(t, cerr)
+	t.Cleanup(func() { require.NoError(t, clus.Stop()) })
+
+	// produce some data
+	cli := newClient(t, clus.EndpointsGRPC(), e2e.ClientConfig{})
+	valueSize := 10
+	var latestRevision int64
+	for i := 0; i <= 100; i++ {
+		resp, err := cli.Put(ctx, fmt.Sprintf("%d", i), stringutil.RandString(uint(valueSize)))
+		require.NoError(t, err)
+		latestRevision = resp.Header.Revision
+	}
+
+	// sleep for CompactionSleepInterval
+	require.NoError(t, clus.Procs[0].Failpoints().SetupHTTP(ctx, "compactAfterAcquiredBatchTxLock",
+		fmt.Sprintf(`sleep("%s")`, CompactionSleepInterval)))
+
+	// start compaction
+	compactionFinished := make(chan struct{}) // Channel to signal completion of compaction
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done() // Decrement counter when done
+		defer close(compactionFinished)
+		t.Log("start compaction...")
+		_, err := cli.Compact(context.Background(), latestRevision, clientv3.WithCompactPhysical())
+		require.NoError(t, err)
+		t.Log("finished compaction...")
+	}()
+
+	// Compaction runs in batches. During each batch, it acquires a lock, releases it at the end,
+	// and then waits for a CompactionSleepInterval before starting the next batch. This pause
+	// allows PUT requests to be processed.
+	//
+	// Therefore, the expected PUT request latency should fall between
+	// [each batch execution time, each batch execution time + CompactionSleepInterval],
+	// which in this case is [CompactionSleepInterval, CompactionSleepInterval × 2].
+	numGoroutines := 10
+	for i := 1; i <= numGoroutines; i++ {
+		wg.Add(1) // Increment WaitGroup counter
+
+		go func(gid int) {
+			defer wg.Done() // Decrement counter when done
+			verifyPutLatencyDuringCompaction(ctx, t, cli, compactionFinished, CompactionSleepInterval*2)
+		}(i)
+	}
+	wg.Wait()
+}
+
+func verifyPutLatencyDuringCompaction(ctx context.Context, t *testing.T, cli *clientv3.Client, compactionFinished <-chan struct{}, putLatencyLimit time.Duration) {
+	for {
+		select {
+		case <-compactionFinished: // compaction finished
+			return
+		default:
+			start := time.Now()
+			_, err := cli.Put(ctx, fmt.Sprintf("foo"), "far")
+			require.NoError(t, err)
+
+			if time.Since(start) > putLatencyLimit {
+				t.Fatalf("Test failed: put latency is larger than %s", putLatencyLimit)
+			}
+
+		}
+	}
+}