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])
@@ -63,6 +64,7 @@ func (s *store) scheduleCompaction(compactMainRev, prevCompactRev int64) (KeyVal
 			// gofail: var compactBeforeSetFinishedCompact struct{}
 			UnsafeSetFinishedCompact(tx, compactMainRev)
 			tx.Unlock()
+			dbCompactionPauseMs.Observe(float64(time.Since(start) / time.Millisecond))
 			// gofail: var compactAfterSetFinishedCompact struct{}
 			hash := h.Hash()
 			size, sizeInUse := s.b.Size(), s.b.SizeInUse()

tests/e2e/reproduce_19406_test.go

@@ -0,0 +1,126 @@
+// Copyright 2025 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/pkg/v3/stringutil"
+	"go.etcd.io/etcd/tests/v3/framework/e2e"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestReproduce19406 reproduces the issue: https://github.com/etcd-io/etcd/issues/19406
+func TestReproduce19406(t *testing.T) {
+	e2e.BeforeTest(t)
+
+	compactionSleepInterval := 100 * time.Millisecond
+	ctx := context.TODO()
+
+	clus, cerr := e2e.NewEtcdProcessCluster(ctx, t,
+		e2e.WithClusterSize(1),
+		e2e.WithGoFailEnabled(true),
+		e2e.WithCompactionBatchLimit(1),
+		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
+
+	produceKeyNum := 20
+	for i := 0; i <= produceKeyNum; i++ {
+		resp, err := cli.Put(ctx, fmt.Sprintf("%d", i), stringutil.RandString(uint(valueSize)))
+		require.NoError(t, err)
+		latestRevision = resp.Header.Revision
+	}
+
+	// Sleep for PerCompactionInterationInterval to simulate a single iteration of compaction lasting at least this duration.
+	PerCompactionInterationInterval := compactionSleepInterval
+	require.NoError(t, clus.Procs[0].Failpoints().SetupHTTP(ctx, "compactAfterAcquiredBatchTxLock",
+		fmt.Sprintf(`sleep("%s")`, PerCompactionInterationInterval)))
+
+	// start compaction
+	t.Log("start compaction...")
+	_, err := cli.Compact(ctx, latestRevision, clientv3.WithCompactPhysical())
+	require.NoError(t, err)
+	t.Log("finished compaction...")
+
+	// Validate that total compaction sleep interval
+	// 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 total compaction sleep interval larger or equal to
+	// (compaction iteration number - 1) * compactionSleepInterval
+	httpEndpoint := clus.EndpointsHTTP()[0]
+	totalKeys := produceKeyNum + 1
+	pauseDuration, totalDuration, err := GetEtcdCompactionMetrics(t, httpEndpoint)
+	require.NoError(t, err)
+	actualSleepInterval := time.Duration(totalDuration-pauseDuration) * time.Millisecond
+	expectSleepInterval := compactionSleepInterval * time.Duration(totalKeys)
+	t.Logf("db_compaction_pause_duration: %.2f db_compaction_total_duration: %.2f, totalKeys: %d",
+		pauseDuration, totalDuration, totalKeys)
+	require.GreaterOrEqualf(t, actualSleepInterval, expectSleepInterval,
+		"expect total compact sleep interval larger than (%v) but got (%v)",
+		expectSleepInterval, actualSleepInterval)
+}
+
+func GetEtcdCompactionMetrics(t *testing.T, httpEndpoint string) (pauseDuration, totalDuration float64, err error) {
+	// make an http request to fetch all Prometheus metrics
+	url := httpEndpoint + "/metrics"
+	resp, err := http.Get(url)
+	if err != nil {
+		t.Fatalf("fetch error: %v", err)
+	}
+	b, err := io.ReadAll(resp.Body)
+	resp.Body.Close()
+	if err != nil {
+		t.Fatalf("fetch error: reading %s: %v", url, err)
+	}
+
+	for _, l := range strings.Split(string(b), "\n") {
+		if strings.Contains(l, `etcd_debugging_mvcc_db_compaction_pause_duration_milliseconds_sum`) {
+			parts := strings.Fields(l)
+			if len(parts) == 2 {
+				pauseDuration, err = strconv.ParseFloat(parts[1], 64)
+				if err != nil {
+					return 0, 0, fmt.Errorf("invalid pause sum value: %w", err)
+				}
+			}
+		}
+		if strings.Contains(l, `etcd_debugging_mvcc_db_compaction_total_duration_milliseconds_sum`) {
+			parts := strings.Fields(l)
+			if len(parts) == 2 {
+				totalDuration, err = strconv.ParseFloat(parts[1], 64)
+				if err != nil {
+					return 0, 0, fmt.Errorf("invalid pause sum value: %w", err)
+				}
+			}
+		}
+	}
+	return pauseDuration, totalDuration, nil
+}