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

Code before:
	CompactionSleepInterval := 50 * time.Millisecond

	clus, cerr := e2e.NewEtcdProcessCluster(ctx, t,
		e2e.WithClusterSize(1),
		e2e.WithGoFailEnabled(true),
		e2e.WithCompactionBatchLimit(compactionBatchLimit),
		e2e.WithCompactionSleepInterval(CompactionSleepInterval),
		e2e.WithWatchProcessNotifyInterval(100*time.Millisecond),
	)

	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
	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)

Review comment (performance):
Can reuse the v3 client in `generateTrafficAndVerifyPutLatency`?

Code after:

	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.