Repository: etcd-io/etcd
File: tests/e2e/reproduce_19406_test.go (Go)

Code before:
	require.NoError(t, clus.Procs[0].Failpoints().SetupHTTP(ctx, "compactAfterAcquiredBatchTxLock",
		fmt.Sprintf(`sleep("%s")`, CompactionSleepInterval)))

	// start compaction
	done := 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(done)
		t.Log("start compaction...")
		_, err := cli.Compact(context.Background(), latestRevision, clientv3.WithCompactPhysical())
		require.NoError(t, err)
		t.Log("finished compaction...")
	}()

	// The PUT latency during compaction should be less than CompactionSleepInterval * 2,
	// as the compaction workflow is expected to release the lock after
	// CompactionSleepInterval + per-batch execution time (which should be smaller than CompactionSleepInterval).
	numGoroutines := 20
	for i := 1; i <= numGoroutines; i++ {
		wg.Add(1) // Increment WaitGroup counter

		go func(gid int) {
			defer wg.Done() // Decrement counter when done
			generateAndVerifyPutLatency(ctx, t, clus.EndpointsGRPC(), done, CompactionSleepInterval*2)
		}(i)
	}
	wg.Wait()
}

func generateAndVerifyPutLatency(ctx context.Context, t *testing.T, gRPCURLs []string, stopChan <-chan struct{}, putLatencyLimit time.Duration) {
	cli, err := clientv3.New(clientv3.Config{
		DialTimeout:          dialTimeout,
		DialKeepAliveTime:    keepaliveTime,
		DialKeepAliveTimeout: keepaliveTimeout,
		Endpoints:            gRPCURLs,
	})
	require.NoError(t, err)
	t.Cleanup(func() { _ = cli.Close() })

	for {
		select {
		case <-stopChan: // compaction finished
			return
		default:
			start := time.Now()
			_, err := cli.Put(ctx, fmt.Sprintf("foo"), "far")
			require.NoError(t, err)

			if time.Since(start) > putLatencyLimit {

Review comment (performance):
Leave a comment like 

put is expected with interleaved with compaction per batch, so it should take `[compaction sleep interval, 2 * compaction sleep interval]` latency. 

would help clarify things a bit. 

Code after:
	// 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 := 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) {
	metricsURL, err := url.JoinPath(httpEndpoint, "metrics")
	require.NoError(t, err)

	// Fetch metrics from the endpoint
	metricFamilies, err := e2e.GetMetrics(metricsURL)
	require.NoError(t, err)

	// Extract sum from histogram metric